Question

Я изучаю pthreads, когда натолкнулся на проблему чтения / записи в C. Проблема очень проста, когда потоки «писателя» будут получать доступ к данным из внешнего источника, а потоки «читателя» будут читать эти данные из общий буфер, а затем псевдообработка данных. Потоки чтения и записи должны выполняться непрерывно в цикле while.

Я пытаюсь реализовать и скомпилировать это в стандартной системе Unix с интерфейсом POSIX.

Я просмотрел некоторые вопросы о переполнении стека:

Программа Reader Writer на C, использующая мьютексы и потоки

блокировка чтения / записи в pthread

и я никуда не попал с этими.

Вот что у меня есть:

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <assert.h>

#define BUFF_SIZE 50
#define M 10
#define N 20

int get_external_data(char *buffer, int bufferSizeInBytes);
void process_data(char *buffer, int bufferSizeInBytes);

int get_external_data(char *buffer, int bufferSizeInBytes){
    int status;
    int val;
    char srcString[] = "0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIJKLMNOPQRSTUVWXYZ";

    val = (int)(rand() % 62);
    if (bufferSizeInBytes < val){
        return (-1);
    }

    strncpy(buffer, srcString, val);
    return val;
}

void process_data(char *buffer, int bufferSizeInBytes){
    int i;
    if(buffer) {
        printf("thread %li - ", pthread_self());

        for(i = 0; i < bufferSizeInBytes; i++) {
            printf("%c", buffer[i]);
        }

        printf("\n");
        memset(buffer, 0, bufferSizeInBytes);
    } else {
        printf("error in process data - %li\n", pthread_self());
    }

    return;
}


pthread_mutex_t data_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t print_lock = PTHREAD_MUTEX_INITIALIZER;
sem_t data_count;

typedef struct node {
    struct node *next;
    char *data;
    int length;
} node_t;

node_t *head, *tail;

/**
 * This thread is responsible for pulling data off of the shared data 
 * area and processing it using the process_data() API.
 */

void *reader_thread(void *arg) {
    int rc;
    node_t *removed_node;

    while(1) {
        rc = sem_wait(&data_count);
        if (0 != rc) {
            return NULL;
        }

        pthread_mutex_lock(&data_lock);

        assert(NULL != head);
        removed_node = head;
        head = head->next;

        pthread_mutex_unlock(&data_lock);

        //Adding this lock for sake of readability at the cost of reduced consumption rate...will run out of memory eventually.

        pthread_mutex_lock(&print_lock);

        process_data(removed_node->data, removed_node->length);

        pthread_mutex_unlock(&print_lock);

        free(removed_node->data);
        free(removed_node);
    }

    return NULL;
}


/**
 * This thread is responsible for pulling data from a device using
 * the get_external_data() API and placing it into a shared area
 * for later processing by one of the reader threads.
 */

void *writer_thread(void *arg) {
    int length;
    char *buffer;
    node_t *new_node;

    new_node = (node_t*) malloc(sizeof(*new_node));
    buffer = (char*) malloc(sizeof(*buffer) * BUFF_SIZE);

    while(1) {

        length = get_external_data(buffer, BUFF_SIZE);

        if (length == -1) {
            //data too big, discard it and try again;
            continue;
        }

        new_node->next = NULL;
        new_node->length = length;
        new_node->data = buffer;

        pthread_mutex_lock(&data_lock);


        if (head == NULL) { //The linked list is completely empty   
            head = new_node;
            tail = new_node;        
        } else { //There are items in the list and we're appending  
            tail->next = new_node;
            tail = new_node;
        }

        pthread_mutex_unlock(&data_lock);

        pthread_mutex_lock(&print_lock);

        printf("thread %ld wrote - %s \n", pthread_self(), buffer);

        pthread_mutex_unlock(&print_lock);

        sem_post(&data_count);

        buffer = (char*) malloc(sizeof(*buffer) * BUFF_SIZE);
        new_node = (node_t*) malloc(sizeof(*new_node));
    }

    return NULL;
}

int main(int argc, char **argv) {
    int i = sem_init(&data_count, 0, 0);
    pthread_t dummy; //creating a dummy thread

    for(i = 0; i < N; i++) { 
        pthread_create(&dummy, NULL, reader_thread, NULL);
    }

    for(i = 0; i < M; i++) { 
        pthread_create(&dummy, NULL, writer_thread, NULL);
    }

    sleep(100);
    return 0;   
}

Синхронизация потока выполняется без ошибок компиляции, но моя программа останавливается после того, как поток записи записывает в буфер несколько раз.

Потоки чтения и записи должны выполняться непрерывно в цикле while, но в моем случае это не так.

Есть идеи, как решить эту проблему?

bruno · Answer 1 · 17 апреля 2019

С помощью gcc добавьте параметр -lpthread для связи с библиотекой pthread , я также призываю вас добавить параметры -pedantic -Wextra -Wall

Есть несколькопроблемы в вашей программе, некоторые из них указаны компилятором

In:

printf("thread %i - ", pthread_self());
printf("error in process data - %i\n", pthread_self());

В стандарте POSIX не требуется, чтобы pthread_t был арифметическим типомтак что это может быть структура и т. д., и вы не можете записать ее как int .В вашем случае это может быть int (более вероятно, unsigned long), но это не переносимо, и лучше самостоятельно управлять идентификатором, связанным с каждым потоком.

В reader_thread:

return;

, но функция возвращает void*, замените его на return NULL;, например

В writer_thread:

printf("thread %d wrote - %s", buffer);

имеет три проблемы:

отсутствует аргумент типа int (возможно, вы хотели pthread_self())
буфер устанавливается get_external_data с использованием strncpy, поэтому нулевой символ окончания отсутствует присутствует, но требуется printf% s (когдапропущенный аргумент будет добавлен или % d удален)
буфер может быть освобожден с помощью reader_thread, возможно, вы предполагали, что он защищен семафором, но это не таккорпус

В get_external_data статус не используется

В основной

 usleep(100);

- очень короткое время, которое дается темамработать, и на самом деле вы просто хотите быть заблокированными, поэтому вы можете заменить его на pthread_join(dummy, NULL);

Предложение иметь идентификаторы потоков: выделите int в куче,присвойте ему уникальный номер и задайте его в параметрах созданным потокам:

int main() {
    ...
    for(i = 0; i < N; i++) { 
      int * m = malloc(sizeof(int));

      *m = i;
      pthread_create(&dummy, NULL, reader_thread, m);
    }

    for(i = 0; i < M; i++) { 
      int * m = malloc(sizeof(int));

      *m = 100 + i;
      pthread_create(&dummy, NULL, writer_thread, m);
    }
    ...
}

и

void *writer_thread(void *arg) {
    int id = *((int*) arg);
    ...
    free((int*) arg);
    ...
    printf("thread %d wrote - %s", id, buffer);
    ...
}

и

void *reader_thread(void *arg) {
    int id = *((int*) arg);
    ...
    free((int*) arg);
    ...
    process_data(removed_node->data, removed_node->length, id);
    ...
}

и

void process_data(char *buffer, int bufferSizeInBytes, int id){
    ...
        printf("thread %i - ", id);
...
        printf("error in process data - %i\n", id);
...
}

После предыдущих изменений выполнение в valgrind с ошибками из-за других проблем:

pi@raspberrypi:/tmp $ gcc -g -pedantic -Wextra -Wall t.c -lpthread
pi@raspberrypi:/tmp $ valgrind ./a.out
==3847== Memcheck, a memory error detector
==3847== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==3847== Using Valgrind-3.13.0 and LibVEX; rerun with -h for copyright info
==3847== Command: ./a.out
==3847== 
==3847== Thread 22:
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x484B20C: strlen (vg_replace_strmem.c:458)
==3847==    by 0x48FD68F: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
thread 100 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEthread 103 wrote - 0123456789abcdefghijklmnopqrstuvwxyxthread 1 - 0123456789abcdefghijklmnopqrstuvwxyxABCDE
thread 0 - 0123456789abcdefghijklmnopqrstuvwxyx
thread 102 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABthread 103 wrote - 0123456thread 2 - 012
thread 101 wrote - 0123456789abcdefghithread 105 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIJKLMthread 3 - 0123456
thread 109 wrote - 0123456789abcdefghijklmnopqrstuvwxyxAthread 3 - 0123456789abcdefghi
==3847== Invalid read of size 1
==3847==    at 0x484B1EC: strlen (vg_replace_strmem.c:458)
==3847==    by 0x48FD68F: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847==  Address 0x49fcd18 is 0 bytes inside a block of size 50 free'd
==3847==    at 0x4848B8C: free (vg_replace_malloc.c:530)
==3847==  Block was alloc'd at
==3847==    at 0x4847568: malloc (vg_replace_malloc.c:299)
==3847== 
thread 107 wrote - 0123456789abcdefgthread 100 wrote - thread 100 wrote - 0123456789athread 101 wrote - 0123456789abcthread 7 - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIJKLM
thread 7 - 0123456789abcdefghijklmnopqrstuvwxyxAB
thread 9 - 0123456789abc
thread 9 - 0123456789abcdefghijklmnopqrstuvwxyxA
thread 108 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABthread 105 wrote - 01thread 8 - 0123456789ab
thread 108 wrote - 0123456789abcdefghithread 12 - 0123
thread 103 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGthread 101 wrote - 0123456789abcdefghijklmnopqrstuvthread 103 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGthread 101 wrote - 0123456789abcdefghijklmnopqrstuvthread 14 - 0123456789a
thread 101 wrote - 0123456789abcdefghijklthread 101 wrote - 0123456789abcdefghijklmnopqrstuvwxyxthread 101 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHthread 101 wrote - 0123456789abcdefghijklmnopqthread 101 wrote - 0123456789abcdefghijklmnopqrstuthread 101 wrote - 0123thread 102 wrote - thread 2 - 0123456789abcdefghijklmnopqrstuv
thread 2 - 0123456789abcdefghijklmnopqr
thread 2 - 0123456789abcdefghi
thread 2 - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFG
thread 100 wrote - 0123456789abcthread 100 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIJKthread 4 - 0123456789abcdefghijkl
thread 109 wrote - thread 13 - 0123456789abcdefghijklmnopqrstuvwxyxAB
thread 108 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIthread 19 - 0123456789abc
thread 101 wrote - 0thread 18 - 01
thread 108 wrote - 0123456thread 108 wrote - 01thread 108 wrote - 0123456789abcdefghijthread 10 - 0123456789abcdefghijklmnopq
thread 8 - 0123456789abcdefghijklmnopqrstu
==3847== Thread 30:
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x484B1F4: strlen (vg_replace_strmem.c:458)
==3847==    by 0x48FD68F: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBEEC: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBF0C: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x49245B8: _IO_file_xsputn@@GLIBC_2.4 (fileops.c:1294)
==3847==    by 0x48FBF7B: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBF80: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBF90: vfprintf (vfprintf.c:1637)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBE24: vfprintf (vfprintf.c:1668)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
==3847== Conditional jump or move depends on uninitialised value(s)
==3847==    at 0x48FBE6C: vfprintf (vfprintf.c:1668)
==3847==    by 0x4902ADF: printf (printf.c:33)
==3847==    by 0x10B83: writer_thread (t.c:144)
==3847==    by 0x4898FC3: start_thread (pthread_create.c:458)
==3847==    by 0x498D037: ??? (clone.S:76)
==3847== 
thread 108 wrote - thread 108 wrote - 0123456789abcdefghijkthread 12 - 0123456789abcdefghijklmnopqrstuvwxyxABCDEFGHIJKL
thread 108 wrote - 0123456789abcdefghijthread 106 wrote - thread 5 - 0123456789abcdefg
thread 106 wrote - thread 106 wrote - 0123456789abcdefghijklmnopqrstuvwxyxABCDthread 16 - 0123456789abcdefghijklmnopqrstuvwx
thread 16 - 0123456789abcdefghijklmnopqrstuv
==3847== 
==3847== HEAP SUMMARY:
==3847==     in use at exit: 5,444 bytes in 74 blocks
==3847==   total heap usage: 155 allocs, 81 frees, 8,138 bytes allocated
==3847== 
==3847== LEAK SUMMARY:
==3847==    definitely lost: 0 bytes in 0 blocks
==3847==    indirectly lost: 0 bytes in 0 blocks
==3847==      possibly lost: 4,080 bytes in 30 blocks
==3847==    still reachable: 1,364 bytes in 44 blocks
==3847==         suppressed: 0 bytes in 0 blocks
==3847== Rerun with --leak-check=full to see details of leaked memory
==3847== 
==3847== For counts of detected and suppressed errors, rerun with: -v
==3847== Use --track-origins=yes to see where uninitialised values come from
==3847== ERROR SUMMARY: 51 errors from 10 contexts (suppressed: 6 from 3)

Проблема чтения / записи в C с использованием pthreads

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

1 Ответ

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Проблема чтения / записи в C с использованием pthreads

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

1 Ответ

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Похожие темы