linux进程间的通信(C): 使用信号量进行同步的共享内存机制

一、简介

共享内存为在多个进程之间共享和传递数据提供了一种有效的方式。

但它本身并未提供同步机制。

在实际编程中，可以使用

  信号量，

  传递消息(使用管道或IPC消息)，

  生成信号，

  条件变量，

等方法来提供读写之间的有效的同步机制。

本例程序使用信号量进行同步，

主要是因为它方便，使用广泛，且独立于进程。

本例程序实现了，

生产者进程:

  每次读取YUV输入文件的一帧，

  然后将其写到共享内存中。

消费者进程:

  每次从共享内存中读到一帧，

  处理后，

  将其写到输出文件。

两个进程间使用信号量来保证同步处理每一帧。

本例程序很好地示范了共享内存和信号量的机制，

对于实际程序的开发很有意义。

二、生产者进程

common.h

用来设置一些测试用的基本参数。

/*
 * \File
 *   common.h
 */
#ifndef __COMMON_H__
#define __COMMON_H__

#define TEST_FILE "coastguard_cif.yuv"
#define OUTPUT_FILE "coastguard_cif_trsd.yuv"

#define FYUV_WIDTH  352
#define FYUV_HEIGHT 288

#endif

共享内存相关的头文件。

shm_com.h

/*
 * \File
 *   shm_com.h
 * \Brief
 */
#ifndef __SHM_COM_H__
#define __SHM_COM_H__

#define SHM_SEED 1001
#define MAX_SHM_SIZE 2048*2048*3

typedef struct shared_use_st
{
  int  end_flag;              //用来标记进程间的内存共享是否结束: 0, 未结束； 1， 结束
  char shm_sp[MAX_SHM_SIZE]; //共享内存的空间
}shared_use_st;

#endif

信号量的头文件

semaphore.h

/*
 * \File
 *   semaphore.h
 * \Brief
 *   semaphore operation
 */
#ifndef __SEMAPHORE_H__
#define __SEMAPHORE_H__

#define SEM_SEED 1000

union semun
{
  int val;
  struct semid_ds *buf;
  unsigned short *array;
};

int  set_semvalue(int sem_id, int value);
void del_semvalue(int sem_id);
int  semaphore_p(int sem_id);
int  semaphore_v(int sem_id);

#endif

帧处理的头文件

frame.h

/*
 * \File
 *   frame.h
 * \Brief
 *
 */
#ifndef __FRAME_H__
#define __FRAME_H__

#include "shm_com.h"

#define PIX_VALUE 1

typedef enum
{
  YUV420,
  YUV422,
  YUV444,
  RGB
}frame_type_e;

typedef struct
{
  int frm_w;     // width
  int frm_h;     // height
  frame_type_e frm_type;
  int frm_size;

  char *frm_comps;
}frame_t, *frame_p;


int init_frame(frame_p frame, int width, int height, frame_type_e type);
int free_frame(frame_p frame);

int read_frame_from_file(frame_p frame, FILE* fp);
int write_frame_into_file(FILE* fp, frame_p frame);

int read_frame_from_shm(frame_p frame, shared_use_st* shm);
int write_frame_into_shm(shared_use_st* shm, frame_p frame);

int crop_frame(frame_p frame, int l_offset, int t_offset, int c_w, int c_h);

#endif

生产者进程的基本流程:

打开输入文件并初始化帧;

创建并初始化共享内存和信号量;

然后每次读取一帧，

  用信号量获取共享内存的权限后，

  将读取的帧写入共享内存，

  再释放共享内存的权限。

当处理完所有的帧后，

设置结束标志，

并释放相关的资源。

producer.c

/*
 * \File
 *   producer.c
 * \Brief
 *   Test shared-memory and message-queue
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>

#include <sys/shm.h>
#include <sys/sem.h>

#include "common.h"
#include "semaphore.h"
#include "shm_com.h"
#include "frame.h"


int main(char argc, char* argv[])
{
  FILE* fp_in = NULL;
  frame_t frame;
  int frame_cnt = 0;

  int sem_id;   // semaphore id

  int shm_id;   // shared-memory id
  void* shared_memory = (void*)0;
  shared_use_st* shared_stuff;

  /* Open input file */
  if ((fp_in = fopen(TEST_FILE, "rb")) < 0 )
  {
    printf("Open input file failed: %s\n", TEST_FILE);
    exit(EXIT_FAILURE);
  }

  /* Init frame */
  init_frame(&frame, FYUV_WIDTH, FYUV_HEIGHT, YUV420);
  printf("FRAME: w = %d, h = %d\n", frame.frm_w, frame.frm_h);

  /* Create and init semaphore */
  sem_id = semget((key_t)SEM_SEED, 1, 0666 | IPC_CREAT);
  if (sem_id == -1)
  {
    fprintf(stderr, "semget failed.\n");
    exit(EXIT_FAILURE);
  }

  /* Init shared-memory */
  shm_id = shmget((key_t)SHM_SEED, sizeof(struct shared_use_st), 0666 | IPC_CREAT);
  if (shm_id == -1)
  {
    fprintf(stderr, "shmget failed.\n");
    exit(EXIT_FAILURE);
  }

  shared_memory = shmat(shm_id, (void*)0, 0);
  if (shared_memory == (void*)-1)
  {
    fprintf(stderr, "shmat failed.\n");
    exit(EXIT_FAILURE);
  }

  shared_stuff = (struct shared_use_st*)shared_memory;
  shared_stuff->end_flag = 0;

  printf("FRAME_CNT: %d\n", frame_cnt);
  set_semvalue(sem_id, 1);

  while (read_frame_from_file(&frame, fp_in) == 0)
  {
    semaphore_p(sem_id);

    /* Write it into shared memory */
    write_frame_into_shm(shared_stuff, &frame);
    shared_stuff->end_flag = 0;

    semaphore_v(sem_id);

    frame_cnt++;
    printf("FRAME_CNT: %d\n", frame_cnt);
  }
  semaphore_p(sem_id);
  shared_stuff->end_flag = 1;
  semaphore_v(sem_id);

  /* over */
  printf("\nProducer over!\n");
  fclose(fp_in);
  free_frame(&frame);
  del_semvalue(sem_id);
  if (shmdt(shared_memory) == -1)
  {
    fprintf(stderr, "shmdt failed.\n");
    exit(EXIT_FAILURE);
  }

  exit(EXIT_SUCCESS);
}

三、消费者进程

消费者进程的基本流程:

打开输出文件并初始化帧;

获取共享内存和信号量;

每次

  得到共享内存的权限后，

  从共享内存中读取一帧并获得结束标志

  进行帧处理，

  释放共享内存的权限。

直到结束标志为真。

最后释放相关的资源。

consumer.c

/*
 * \File
 *    consumer.c
 * \Brief
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>

#include <sys/shm.h>
#include <sys/sem.h>

#include "common.h"
#include "semaphore.h"
#include "shm_com.h"
#include "frame.h"


int main(char argc, char *argv[])
{
  FILE* fp_out = NULL;
  frame_t frame;
  int frame_cnt = 0;

  int sem_id;   // semaphore id

  int shm_id;   // shared-memory id
  void* shared_memory = (void*)0;
  shared_use_st* shared_stuff;
  int end_flag = 0;

  /* Open output file */
  if ((fp_out = fopen(OUTPUT_FILE, "wb")) < 0 )
  {
    printf("Open output file failed: %s\n", OUTPUT_FILE);
    exit(EXIT_FAILURE);
  }

  /* Init frame */
  init_frame(&frame, FYUV_WIDTH, FYUV_HEIGHT, YUV420);
  printf("FRAME: w = %d, h = %d\n", frame.frm_w, frame.frm_h);

  /* Create semaphore */
  sem_id = semget((key_t)SEM_SEED, 1, 0666 | IPC_CREAT);
  if (sem_id == -1)
  {
    fprintf(stderr, "semget failed.\n");
    exit(EXIT_FAILURE);
  }

  /* Init shared-memory */
  shm_id = shmget((key_t)SHM_SEED, sizeof(struct shared_use_st), 0666 | IPC_CREAT);
  if (shm_id == -1)
  {
    fprintf(stderr, "shmget failed.\n");
    exit(EXIT_FAILURE);
  }

  shared_memory = shmat(shm_id, (void*)0, 0);
  if (shared_memory == (void*)-1)
  {
    fprintf(stderr, "shmat failed.\n");
    exit(EXIT_FAILURE);
  }
  shared_stuff = (struct shared_use_st*)shared_memory;

  printf("FRAME_CNT: %d\n", frame_cnt);
  /*
   * 必须先置0，
   * 否则会因生产者进程的异常退出未释放信号量而导致程序出错
   */
  set_semvalue(sem_id, 0);

  do
  {
    semaphore_p(sem_id);

    /* Read frame from shared-memory */
    read_frame_from_shm(&frame, shared_stuff);
    end_flag = shared_stuff->end_flag;

    crop_frame(&frame, 10, 10, 40, 40);
    write_frame_into_file(fp_out, &frame);

    semaphore_v(sem_id);

    frame_cnt++;
    printf("FRAME_CNT: %d\n", frame_cnt);
  } while(!end_flag);

  /* Over */
  printf("\nConsumer over!\n");
  fclose(fp_out);
  free_frame(&frame);
  if (shmdt(shared_memory) == -1)
  {
    fprintf(stderr, "shmdt failed.\n");
    exit(EXIT_FAILURE);
  }

  exit(EXIT_SUCCESS);
}

四、信号量函数和帧处理函数

信号量函数

semaphore.c

/*
 * \File
 *   semaphore.c
 * \Breif
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/sem.h>

#include "semaphore.h"

/* init semaphore by semctl */
int set_semvalue(int sem_id, int value)
{
  union semun sem_union;

  sem_union.val = value;
  if (semctl(sem_id, 0, SETVAL, sem_union) == -1)
    return 1;

  return 0;
}


/* delete semaphore by sectl */
void del_semvalue(int sem_id)
{
  union semun sem_union;

  if (semctl(sem_id, 0, IPC_RMID, sem_union) == -1)
    fprintf(stderr, "Failed to delete semaphore\n");
}

/* P(v) */
int semaphore_p(int sem_id)
{
  struct sembuf sem_b;
  sem_b.sem_num = 0;
  sem_b.sem_op  = -1; /* P(v) */
  sem_b.sem_flg = SEM_UNDO;

  if (semop(sem_id, &sem_b, 1) == -1)
  {
    fprintf(stderr, "semaphore_p failed\n");
    return 1;
  }

  return 0;
}

/* V(v) */
int semaphore_v(int sem_id)
{
  struct sembuf sem_b;

  sem_b.sem_num = 0;
  sem_b.sem_op  = 1; // V(v)
  sem_b.sem_flg = SEM_UNDO;

  if (semop(sem_id, &sem_b, 1) == -1)
  {
    fprintf(stderr, "semaphore_v failed\n");
    return 1;
  }

  return 0;
}

帧处理函数

frame.c

/*
 * \File
 *   frame.c
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <memory.h>
#include <assert.h>
#include <errno.h>

#include "frame.h"

/*
 * \Brief
 *   init frame
 */
int init_frame(frame_p frame, int width, int height, frame_type_e type)
{
  assert(frame != NULL);

  frame->frm_w = width;
  frame->frm_h = height;
  frame->frm_type = type;

  switch(frame->frm_type)
  {
  case YUV420:
    frame->frm_size = (frame->frm_w * frame->frm_h * 3) / 2;
    break;
  case YUV422:
    frame->frm_size = frame->frm_w * frame->frm_h * 2;
    break;
  case YUV444:
    frame->frm_size = frame->frm_w * frame->frm_h * 3;
    break;
  case RGB:
    frame->frm_size = frame->frm_w * frame->frm_h * 3;
    break;
  default:
    fprintf(stderr, "frame type is invalid.");
    return 1;
  }

  if ((frame->frm_comps = (char*)calloc(frame->frm_size, sizeof(char))) == NULL)
  {
    fprintf(stderr, "calloc failed.");
    return 1;
  }

  return 0;
}

/*
 * \Brief
 *   init frame
 */
int free_frame(frame_p frame)
{
  assert(frame != NULL);

  free(frame->frm_comps);
  return 0;
}


/*
 * \Brief
 *   read a frame from file
 */
int read_frame_from_file(frame_p frame, FILE* fp)
{
  int ret;
  assert(frame != NULL && fp != NULL);

  if (ret = (fread(frame->frm_comps, sizeof(char), frame->frm_size, fp))
        != frame->frm_size)
  {
    fprintf(stderr, "read_frame_from_file failed. %d\n", ret);
    return 1;
  }

  return 0;
}


/*
 * \Brief
 *   write a frame into file
 */
int write_frame_into_file(FILE* fp, frame_p frame)
{
  assert(frame != NULL && fp != NULL);

  if (fwrite(frame->frm_comps, sizeof(char), frame->frm_size, fp)
      != frame->frm_size)
  {
    fprintf(stderr, "write_frame_into_file failed.\n");
    return 1;
  }

  return 0;
}

/*
 * \Brief
 *   read a frame from shared-memory
 */
int read_frame_from_shm(frame_p frame, shared_use_st* shm)
{
  assert(frame != NULL && shm != NULL);

  memcpy((char*)frame->frm_comps, (char*)shm->shm_sp, frame->frm_size);

  return 0;
}

/*
 * \Brief
 *   write a frame into shared-memory
 */
int write_frame_into_shm(shared_use_st* shm, frame_p frame)
{
  assert(frame != NULL && shm != NULL);

  memcpy((char*)shm->shm_sp, (char*)frame->frm_comps, frame->frm_size);

  return 0;
}



/*
 * \Brief
 *   crop frame
 */
int crop_frame(frame_p frame, int l_offset, int t_offset, int c_w, int c_h)
{
  char* crop_loc;
  int   index_h;

  assert(frame != NULL);

  if ((l_offset + c_w) > frame->frm_w)
  {
    printf("Crop width is out of range.\n");
    return 1;
  }
  if ((t_offset + c_h) > frame->frm_h)
  {
    printf("Crop height is out of range.\n");
    return 1;
  }

  crop_loc = frame->frm_comps + (t_offset * frame->frm_w) + l_offset;
  for (index_h = 0; index_h < c_h; index_h++)
  {
    memset(crop_loc, PIX_VALUE, c_w);

    crop_loc += frame->frm_w;
  }

  return 0;
}

五、编译与运行

makefile.producer

OBJECTS = producer.o semaphore.o frame.o
CC = gcc
CFLAG = -g -Wa

producer : $(OBJECTS)
  $(CC) $(CFLAG) -o producer $(OBJECTS)

producer.o: common.h semaphore.h frame.h shm_com.h
semaphore.o:semaphore.h
frame.o: frame.h

.PHONY:clean
clean:
  rm producer $(OBJECTS)

makefile.consumer

OBJECTS = consumer.o semaphore.o frame.o
CC = gcc
CFLAG = -g -Wa

consumer : $(OBJECTS)
  $(CC) $(CFLAG) -o consumer $(OBJECTS)

consumer.o: common.h semaphore.h frame.h shm_com.h
semaphore.o:semaphore.h
frame.o:frame.h

.PHONY:clean
clean:
  rm consumer $(OBJECTS)

编译与运行:

$make -f makefile.producer
$make -f makefile.consumer
$producer &
$consumer