是否可以从父线程访问/更新子线程的资源？

Question

我正在用 C 进行套接字编程，而且我对多线程完全陌生。

这是我的场景，我需要一个父线程，它从套接字读取数据（比如说..）并将相同的数据排入其子线程的队列。这里的问题是，我如何更新队列来自父线程的子线程特定的。

Answer 1

支持多个生产者和消费者的线程安全队列。

MtQueue.h:

#ifndef MtQueue_H
#define MtQueue_H

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

// A fixed-size circular buffer.
typedef struct {
   pthread_mutex_t mutex;
   pthread_cond_t cond;
   int done;
   int empty;
   int full;
   size_t max;
   size_t next_insert;
   size_t next_read;
   void** buf;
} MtQueue;

// Returns NULL and sets errno on error.
// Free the queue with MtQueue_delete when done.
MtQueue* MtQueue_new(size_t max);

// Returns 0 and sets errno on error.
// Destroy the queue with MtQueue_destroy when done.
int MtQueue_init(MtQueue* q, size_t max);

// Inverse of MtQueue_new.
// Only call when the queue is no longer in use.
void MtQueue_delete(MtQueue* q);

// Inverse of MtQueue_init.
// Only call when the queue is no longer in use.
void MtQueue_destroy(MtQueue* q);

// Initiates shutdown of the queue.
// You must ensure that no there are no pending call to enqueue before this is called.
// You must ensure not to call enqueue once this is called.
void MtQueue_done(MtQueue* q);

// Returns the oldest item from the queue (via a parameter) and returns 1.
// If the queue is empty and done, returns 0.
// If the queue is empty and not done, waits until that changes.
int MtQueue_dequeue(MtQueue* q, void** pp);

// Adds the argument to the queue.
// If the queue is full, waits until that changes.
void MtQueue_enqueue(MtQueue* q, void* p);

#endif

MtQueue.c:

#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <stdlib.h>

#include "MtQueue.h"

MtQueue* MtQueue_new(size_t max) {
   MtQueue* q = malloc(sizeof(MtQueue));
   if (!q)
      goto Error1;

   if (!MtQueue_init(q, max))
      goto Error2;

   return q;

Error2:
   free(q);
Error1:
   return NULL;
}

int MtQueue_init(MtQueue* q, size_t max) {
   void** buf = malloc(sizeof(void*) * max);
   if (!buf)
      goto Error1;

   errno = pthread_mutex_init(&(q->mutex), NULL);
   if (errno)
      goto Error2;

   errno = pthread_cond_init(&(q->cond), NULL);
   if (errno)
      goto Error3;

   q->done = 0;
   q->empty = 1;
   q->full = 0;
   q->max = max;
   q->next_insert = 0;
   q->next_read = 0;
   q->buf = buf;
   return 1;

Error3:
   pthread_mutex_destroy(&(q->mutex));
Error2:
   free(buf);
Error1:
   return 0;
}

void MtQueue_delete(MtQueue* q) {
   MtQueue_destroy(q);
   free(q);
}

void MtQueue_destroy(MtQueue* q) {
   assert(q->empty);
   free(q->buf);
   pthread_cond_destroy(&(q->cond));
   pthread_mutex_destroy(&(q->mutex));
}

void MtQueue_done(MtQueue* q) {
   pthread_mutex_lock(&(q->mutex));
   q->done = 1;
   pthread_cond_signal(&(q->cond));
   pthread_mutex_unlock(&(q->mutex));
}

int MtQueue_dequeue(MtQueue* q, void** pp) {
   pthread_mutex_lock(&(q->mutex));
   while (q->empty && !q->done)
      pthread_cond_wait(&(q->cond), &(q->mutex));

   int dequeued;
   if (q->empty) {
      // q->done && q->empty is true.
      // We are completely done.
      dequeued = 0;
   } else {
      *pp = q->buf[ q->next_read ];
      q->next_read = ( q->next_read + 1 ) % q->max;
      q->empty = q->next_read == q->next_insert;
      q->full = 0;
      dequeued = 1;
   }

   pthread_cond_signal(&(q->cond));
   pthread_mutex_unlock(&(q->mutex));
   return dequeued;
}

void MtQueue_enqueue(MtQueue* q, void* p) {
   pthread_mutex_lock(&(q->mutex));
   while (q->full)
      pthread_cond_wait(&(q->cond), &(q->mutex));

   assert(!q->done);

   q->buf[q->next_insert] = p;
   q->next_insert = ( q->next_insert + 1 ) % q->max;
   q->empty = 0;
   q->full = q->next_insert == q->next_read;

   pthread_cond_signal(&(q->cond));
   pthread_mutex_unlock(&(q->mutex));
}

a.c（示例用户）：

#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

#include "MtQueue.h"

// Producers will block if there are this many items in the queue.
#ifndef QUEUE_SIZE
#define QUEUE_SIZE 10
#endif

// The number of consumers (worker threads) to create.
#ifndef NUM_WORKERS
#define NUM_WORKERS 4
#endif

// The amount of work to generate for this test.
#ifndef NUM_JOBS
#define NUM_JOBS 40
#endif

// Simulate work using a sleep.
#ifndef SIM_WORK
#define SIM_WORK 0
#endif

#if SIM_WORK
static int msleep(long msec) {
   struct timespec ts;
   int res;

   if (msec < 0) {
       errno = EINVAL;
       return -1;
   }

   ts.tv_sec = msec / 1000;
   ts.tv_nsec = (msec % 1000) * 1000000;

    do {
       res = nanosleep(&ts, &ts);
    } while (res && errno == EINTR);

    return res;
}
#endif

// Shared variables.
static MtQueue q;

static void* worker_func(void* worker_id_) {
   uintptr_t worker_id = (uintptr_t)worker_id_;

#if SIM_WORK
   unsigned int seed = worker_id;  // Whatever.
#endif

   uintptr_t j;
   while (MtQueue_dequeue(&q, (void**)&j)) {
      printf("[%" PRIuPTR "] Dequeued %" PRIuPTR "\n", worker_id, j);
#if SIM_WORK
      msleep( rand_r(&seed) % 1000 + 1000 );  // Simulate a 1 to 2s load.
#endif
      printf("[%" PRIuPTR "]    Finished processing %" PRIuPTR "\n", worker_id, j);
   }

   return NULL;
}

int main(void) {
   MtQueue_init(&q, QUEUE_SIZE);

   pthread_t workers[NUM_WORKERS];
   for (uintptr_t w=0; w<NUM_WORKERS; ++w) {
      if (errno = pthread_create(&(workers[w]), NULL, worker_func, (void*)w)) {
         perror(NULL);
         exit(1);
      }
   }

   for (uintptr_t j=0; j<NUM_JOBS; ++j) {
      printf("[x] Enqueuing %" PRIuPTR "...\n", j);
      MtQueue_enqueue(&q, (void*)j);
      printf("[x]    Enqueued %" PRIuPTR ".\n", j);
   }

   MtQueue_done(&q);
   printf("[x] Called done.\n");

   for (uintptr_t w=0; w<NUM_WORKERS; ++w)
      pthread_join(workers[w], NULL);

   MtQueue_destroy(&q);
   return 0;
}

如何运行示例用户：

gcc -Wall -Wextra -pedantic a.c MtQueue.c -o a -lpthread && ./a

gcc -D SIM_WORK=1 -D NUM_JOBS=20 -Wall -Wextra -pedantic a.c MtQueue.c -o a -pthread && ./a