многопоточный потребитель, код производителя C, не работает?

Question

Я пытаюсь сделать простой многопоточный потребитель/производитель, где несколько потоков чтения и записи, чтение из файла в буфер, а затем из буфера обратно в файл. Он должен быть потокобезопасным. однако он не работает, как я ожидал. Он останавливается на полпути, но каждый раз на другой линии? Пожалуйста, помогите мне понять, что я делаю неправильно?!?

#include <pthread.h> 
#include <stdio.h> 
#include <stdlib.h> 
//TODO Define global data structures to be used 
#define BUF_SIZE 5 
FILE *fr; 
FILE *to;   /* declare the file pointer */ 

struct _data { 
    pthread_mutex_t mutex; 
    pthread_cond_t cond_read; 
    pthread_cond_t cond_write; 
    int condition; 
    char buffer[BUF_SIZE]; 
    int datainbuffer; 
}dc1 = { 
    PTHREAD_MUTEX_INITIALIZER,PTHREAD_COND_INITIALIZER,PTHREAD_COND_INITIALIZER,0,{0},0 
}; 


void *reader_thread(void *arg) { 
    //TODO: Define set-up required 
    struct _data *d = (struct _data *)arg; 
    int killreaders = 0; 
    while(1) { 
     //TODO: Define data extraction (queue) and processing 

     pthread_mutex_lock(&d->mutex); 

     while (d->condition == 0 || d->datainbuffer<=0){ 
      pthread_cond_wait(&d->cond_read, &d->mutex); 
      if(killreaders == 1){ 
       pthread_mutex_unlock(&d->mutex); 
       pthread_cond_signal(&d->cond_read); 
       pthread_cond_signal(&d->cond_write); 
       return NULL; 
      } 

     } 
     d->condition = 0; 

     int i; 
     char res; 
     //if the buffer is not full, that means the end of file is reached and it time to kill the threads remaining. 
     if(d->datainbuffer!=BUF_SIZE) 
      killreaders = 1; 

     for (i=0; i<(sizeof d->datainbuffer); i++) { 
      res = d->buffer[i]; 
      printf("to file:%c",res); 
      fputc(res, to); 
     } 
     d->datainbuffer = 0; 


     pthread_mutex_unlock(&d->mutex); 
     pthread_cond_signal(&d->cond_write); 


    } 

    return NULL; 
} 

void *writer_thread(void *arg) { 
    //TODO: Define set-up required 
    struct _data *d = (struct _data *)arg; 
    char * pChar; 
    int killwriters = 0; 

    while(1){ 
     pthread_mutex_lock(&d->mutex); 
     while(d->condition == 1 || d->datainbuffer>0){ 
      pthread_cond_wait(&d->cond_write, &d->mutex); 
      if(killwriters==1){ 
       pthread_mutex_unlock(&d->mutex); 
       pthread_cond_signal(&d->cond_write); 
       pthread_cond_signal(&d->cond_read); 
       return NULL; 

      } 
     } 
     d->condition = 1; 
     int i; 
     char rc; 
     for (i = 0; i < BUF_SIZE; i++){ 
      if((rc = getc(fr)) == EOF){ 
       killwriters = 1; 
       pthread_mutex_unlock(&d->mutex); 
       pthread_cond_signal(&d->cond_read); 

       return NULL; 
      } 
      d->datainbuffer = i+1; 
      d->buffer[i] = rc; 
      printf("%c",rc); 
     } 

     int m = 0; 

     pthread_mutex_unlock(&d->mutex); 
     pthread_cond_signal(&d->cond_read); 


    } 


    return NULL; 
} 


#define M 10 
#define N 20 
int main(int argc, char **argv) { 
    struct _data dc=dc1; 

    fr = fopen ("from.txt", "rt"); /* open the file for reading */ 
    if (fr == NULL) 
    { 
     printf("Could not open file!"); 
     return 1; 
    } 
    to = fopen("to.txt", "wt"); 


    int i; 
    pthread_t readers[N]; 
    pthread_t writers[M]; 


    for(i = 0; i < N; i++) { 
     pthread_create(&readers[i], NULL, reader_thread, (void*)&dc); 
    } 

    for(i = 0; i < M; i++) { 
     pthread_create(&writers[i], NULL, writer_thread, (void*)&dc); 
    } 
    fclose(fr); 
    fclose(to); 

    return 0; 
} 

любое предложение оценили!

Answer 1

Ваши темы прочитаны и пишут файлы, которые вы открываете & закрыть в main. Но главное не дожидаться завершения потоков до закрытия этих файлов.

Answer 2

В дополнение к проблеме, отмеченной Скоттом Хантером, ваши читатели и писатели выполняют всю свою «настоящую работу», удерживая мьютекс, в первую очередь побеждая точку с несколькими потоками.

Читатели должны работать следующим образом:

1) Acquire взаимную блокировку.
2) Блокируйте переменную условия, пока работа не будет доступна.
3) Удалите работу из очереди, возможно, переменной состояния сигнала.
4) Отпуск мьютекса.
5) Обработайте работу.
6) Перейдите к шагу 1.

Писатели должны работать следующим образом:

1) Получить информацию, необходимую для записи.
2) Приобретите мьютексы.
3) Блокируйте переменную условия, пока в очереди не будет места.
4) Поместите информацию в очередь, возможно переменную условия сигнала.
5) Отпустите мьютексы.
6) Перейдите к шагу 1.

Обратите внимание, что оба потока выполняют «настоящую работу» без удержания мьютекса? В противном случае, почему есть несколько потоков, если только один из них может работать одновременно?

Answer 3

Я не уверен, поможет ли мой ответ вам или нет. Но я собираюсь изо всех сил дать вам некоторый ссылочный код.

Я написал аналогичную программу (за исключением того, что она не записывает в файл, вместо этого отображает пункты очереди/производства/потребления в stdout). Его можно найти здесь - https://github.com/sangeeths/pc. Я отделил обработку командной строки и логику очереди в отдельные файлы.

Надеюсь, это поможет!