具有有限缓冲区的生产者/消费者

Question

有人可以检查我的代码并告诉我是否走在正确的轨道上。我似乎有点迷失了。如果您看到我的错误，请让我知道它们。

我想做的是解决有界问题使用我自己的信号量以及 GCD 进行缓冲区。

提前致谢..

sema.c

void procure( Semaphore *semaphore ) {

        pthread_mutex_lock(semaphore->mutex1);

        while(semaphore->value <= 0)
                pthread_cond_wait(&semaphore->condition, semaphore->mutex1);


        semaphore->value--;

        pthread_mutex_unlock(semaphore->mutex1);


}

void vacate( Semaphore *semaphore ) {

        pthread_mutex_lock(semaphore->mutex1);
        semaphore->value++;
        pthread_cond_signal(&semaphore->condition);
        pthread_mutex_unlock(semaphore->mutex1);        
}


void init ( Semaphore *semaphore ){

        semaphore->value = 1;

        pthread_mutex_t myMutex;
        semaphore->mutex1 = &myMutex;
        pthread_mutex_init( semaphore->mutex1, NULL);



}

void destroy ( Semaphore *semaphore ) {

        pthread_mutex_destroy(semaphore->mutex1);


}

和 main.c

struct variables {

        Semaphore *sem;

};

struct variables vars;

void constructer (int *buffer, int *in, int *out) {

        init(vars.sem);

}


void deconstructer () {

        destroy(vars.sem);

}



int rand_num_gen() {
        uint_fast16_t buffer;
        int file;
        int *rand;

        file = open("/dev/random", O_RDONLY);


        while( 1 ) {

                read(file, &buffer, sizeof(buffer));

                printf("16 bit number: %hu\n", buffer );

                *rand = (int) buffer;

                close(file);


                break;
        }

        return *rand;
}


void put_buffer( int* buffer, int* in, int* out ) {

        buffer[*in] = rand_num_gen();    // produce

        procure(vars.sem);                // wait here
                *in = (*in + 1) % BUF_SIZE;
        vacate(vars.sem);

}

void get_buffer( int* buffer, int* in, int* out ) {

        int value;  

        procure(vars.sem);
                value = buffer[*out];  
        vacate(vars.sem);

        *out = (*out + 1) % BUF_SIZE;

}

int main (void) {


        int *in, *out, *buffer;     

        constructer(buffer, in, out);        



        dispatch_queue_t producer, consumer;    

        producer = dispatch_queue_create("put_buffer", NULL);
        consumer = dispatch_queue_create("get_buffer", NULL);



        dispatch_async(producer,
                       ^{
                               int i;
                               do
                               {
                                        put_buffer( buffer, in, out );


                                       dispatch_async(consumer,
                                                      ^{
                                                              get_buffer( buffer, in, out );

                                                              if (i == RUN_LENGTH) exit(EXIT_SUCCESS);
                                                      });
                               }
                               while (i < RUN_LENGTH);             
                       });

        dispatch_main();                

        deconstructer();

        exit (0);
}

Answer 1

你的代码有一个错误。在init函数中，您将局部变量的地址分配给semaphore->mutex1，当函数返回时，该地址将无效。稍后您仍然使用该地址，因此这会导致未定义的行为。

您必须直接在信号量中（不使用指针）为互斥体分配内存，或者通过 malloc 分配内存。

更新：

您的程序有如此多的错误，您绝对应该选择一个更简单的主题来学习有关内存管理的基本概念，如何分配、使用和引用缓冲区，进行正确的错误处理等。在这里是您的代码的稍微编辑过的版本。它仍然行不通，但可能有一些你应该遵循的想法。

#include <limits.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

void procure(Semaphore *semaphore) {
  pthread_mutex_lock(semaphore->mutex1);

  while (semaphore->value <= 0)
    pthread_cond_wait(&semaphore->condition, semaphore->mutex1);

  semaphore->value--;
  pthread_mutex_unlock(semaphore->mutex1);
}

void vacate(Semaphore *semaphore) {
  pthread_mutex_lock(semaphore->mutex1);
  semaphore->value++;
  pthread_cond_signal(&semaphore->condition);
  pthread_mutex_unlock(semaphore->mutex1);  
}

struct variables {
  mutex_t sem_mutex;
  Semaphore sem;
};

struct variables vars;

void constructor(int *buffer, int *in, int *out) {
  vars.sem.value = 1;
  vars.sem.mutex1 = &vars.sem_mutex;
  pthread_mutex_init(vars.sem.mutex1, NULL);
}

void deconstructor() {
  pthread_mutex_destroy(&semaphore->mutex1);
}

int rand_num_gen() {
  const char *randomfile = "/dev/random";
  unsigned char buffer[2]; // Changed: always treat files as byte sequences.
  FILE *f = fopen(randomfile, "rb");
  // Changed: using stdio instead of raw POSIX file access,
  // since the API is much simpler; you don't have to care
  // about interrupting signals or partial reads.

  if (f == NULL) { // Added: error handling
    fprintf(stderr, "E: cannot open %s\n", randomfile);
    exit(EXIT_FAILURE);
  }
  if (fread(buffer, 1, 2, f) != 2) { // Added: error handling
    fprintf(stderr, "E: cannot read from %s\n", randomfile);
    exit(EXIT_FAILURE);
  }
  fclose(f);
  int number = (buffer[0] << CHAR_BIT) | buffer[1];
  // Changed: be independent of the endianness of the system.
  // This doesn't matter for random number generators but is
  // still an important coding style.
  printf("DEBUG: random number: %x\n", (unsigned int) number);
  return number;
}

void put_buffer( int* buffer, int* in, int* out ) {
  buffer[*in] = rand_num_gen();    // produce
  procure(&vars.sem);    // wait here
  *in = (*in + 1) % BUF_SIZE;
  vacate(&vars.sem);
}

void get_buffer( int* buffer, int* in, int* out ) {
  int value;  
  procure(&vars.sem);
  value = buffer[*out];  
  vacate(&vars.sem);
  *out = (*out + 1) % BUF_SIZE;
}

int main (void) {
  int inindex = 0, outindex = 0;
  int buffer[BUF_SIZE];

  constructor(buffer, &inindex, &outindex);
  // Changed: provided an actual buffer and actual variables
  // for the indices into the buffer.
  dispatch_queue_t producer, consumer;    
  producer = dispatch_queue_create("put_buffer", NULL);
  consumer = dispatch_queue_create("get_buffer", NULL);

  dispatch_async(producer, ^{
    int i;
    do {
      put_buffer(buffer, &inindex, &outindex);
      dispatch_async(consumer, ^{
        get_buffer(buffer, &inindex, &outindex);
        if (i == RUN_LENGTH) exit(EXIT_SUCCESS);
      });
    } while (i < RUN_LENGTH);
  });

  dispatch_main();    
  deconstructor();
  exit (0);
}

正如我所说，我没有发现所有错误。