我有一个代表汽车制造厂的程序。队列中有要制造的汽车零件，每个数字代表汽车的不同部分，例如。 4是窗口，我们需要7个窗口：
queue(^: front, *: rear): [0^, 1, 2, 3, 4, 4, 4, 4, 4, 4, 4, 5, 5, 5, 5, 6, 7*]
num_workers代表可以并行工作（线程）的工人数量。这是我调用线程函数的主要部分。
int num_work = num_cars * NUM_PARTS;
int k=0;
while(k<num_work){
for(int i=0;i<num_workers; i++){
pthread_mutex_lock(&lock);
if(k>=num_work)
break;
++k;
pthread_mutex_unlock(&lock);
if((rc = pthread_create(&thr[i], NULL ,workerThreadFuc, &wpack[i]))){
fprintf(stderr, "error: pthread_create, rc: %d\n", rc);
return EXIT_FAILURE;
}
}
这是线程函数，每个线程（工作者）在其中使任务出队并完成任务。
void* workerThreadFuc(void *arg){
// Extract work_pack
work_pack *wpack = (work_pack *)arg;
int tid = wpack->tid;
queue *jobQ = wpack->jobQ;
int jobID;
//TODO finish workerThreadFuc
queueDequeueFront(jobQ, &jobID);
work(jobID, wpack->resource, tid);
pthread_exit(NULL);
}
这是工作函数中有关汽车零件制造方式的示例：
switch(jobID) {
case SKELETON :
makeSkeleton(pack->sem_space, pack->sem_skeleton);
#if DEBUG
printf("Worker[%d]: One car skeleton made!\n", tid);
fflush(stdout);
#endif
break;
case ENGINE :
makeEngine(pack->sem_space, pack->sem_engine);
#if DEBUG
printf("Worker[%d]: One engine made!\n", tid);
fflush(stdout);
#endif
break;
case CHASSIS :
依此类推。我的问题是，无论我有多少辆汽车，出队和获得所有工作的时间都需要一位工人花费相同的时间。例如，一位工人花15秒完成一辆汽车，但花5到10辆汽车花费相同的时间。
同样，无论我有多少工人，一辆车和五辆车的完成率总是相同的。
这是工作功能调用的功能：
void makeItem(sem_t *space, int makeTime, sem_t* item) {
requestSpace(space);
printf("About to sleep for %d\n",makeTime );
sleep(makeTime);
printf("Woke up after %d\n",makeTime );
sem_post(item);
}
输出：
Job defined, 1 workers will build 1 cars with 100 storage spaces
About to sleep for 5
About to sleep for 4
About to sleep for 3
About to sleep for 1
About to sleep for 1
About to sleep for 1
About to sleep for 1
About to sleep for 1
About to sleep for 1
About to sleep for 1
About to sleep for 2
About to sleep for 2
About to sleep for 2
About to sleep for 2
About to sleep for 3
Woke up after 1
Woke up after 1
Woke up after 1
Woke up after 1
Woke up after 1
Woke up after 1
Woke up after 1
Woke up after 2
Woke up after 2
Woke up after 2
Woke up after 2
Woke up after 3
Woke up after 3
WILL BE REPORTING RESULTS
=====Production report=====
Number of workers: 1, Storage space size: 100
Unused Skeleton: 0
Unused Engine: 0
Unused Chassis: 1
Unused Body: 0
Unused Window: 0
Unused Tire: 0
Unused Battery: 0
There are waste car parts!
Production of 0 car done, production time: 3.000766 sec
该程序应该出队，入睡并等待，直到时间结束并出队新工作并入睡。但是由于某种原因，所有内容都会立即出队，并且似乎该程序甚至无法为每个任务休眠。
您尝试为每个作业创建一个新线程。但是您忘记了按需要限制并发工作程序的数量。如果退出的工作线程数已达到上限，则需要等待一个完成，然后再创建另一个。
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#define MAX_WORKERS 4
#define NUM_JOBS 40
typedef struct {
int created;
int joinable;
pthread_t thread;
size_t job;
} Worker;
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;
}
// Shared variables.
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
static Worker workers[MAX_WORKERS];
Worker *get_free_worker(void) {
pthread_mutex_lock(&mutex);
while (1) {
for (size_t w=MAX_WORKERS; w--; ) {
Worker *worker = workers + w;
if (worker->joinable) {
pthread_join(worker->thread, NULL);
worker->created = 0;
worker->joinable = 0;
}
if (!worker->created) {
worker->created = 1;
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
return worker;
}
}
pthread_cond_wait(&cond, &mutex);
}
}
static void *worker_func(void *worker_) {
Worker *worker = (Worker*)worker_;
unsigned int seed = (uintptr_t)worker; // Whatever.
printf("Processing job %zu\n", worker->job);
// msleep( rand_r(&seed) % 1000 + 1000 ); // Simulate a 1 to 2s load.
printf(" Finished processing %zu\n", worker->job);
{
pthread_mutex_lock(&mutex);
worker->joinable = 1;
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
}
return NULL;
}
int main(void) {
// For all w, workers[w].created and workers[w].joinable are
// already initialized to 0 by virtue of being in static storage.
for (size_t j=0; j<NUM_JOBS; ++j) {
Worker *worker = get_free_worker();
printf("Creating worker for job %zu\n", j);
worker->job = j;
if (errno = pthread_create(&(worker->thread), NULL, worker_func, worker)) {
perror("pthread_create");
worker->created = 0;
break;
}
}
for (size_t w=MAX_WORKERS; w--; ) {
Worker *worker = workers + w;
if (worker->created) {
pthread_join(worker->thread, NULL);
}
}
return 0;
}
一种不同的方法是创建由队列提供的静态工作线程池。通过这种方法，可以将工人重复用于多个工作。
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#define NUM_WORKERS 4
#define QUEUE_SIZE 10
#define NUM_JOBS 40
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;
uintptr_t *buf;
} Queue;
static void Queue_init(Queue* q, size_t max) {
pthread_mutex_init(&(q->mutex), NULL);
pthread_cond_init(&(q->cond), NULL);
q->done = 0;
q->empty = 1;
q->full = 0;
q->max = max;
q->next_insert = 0;
q->next_read = 0;
q->buf = malloc(sizeof(uintptr_t)*max);
}
static void Queue_destroy(Queue *q) {
free(q->buf);
pthread_cond_destroy(&(q->cond));
pthread_mutex_destroy(&(q->mutex));
}
static void Queue_done(Queue *q) {
pthread_mutex_lock(&(q->mutex));
q->done = 1;
pthread_cond_signal(&(q->cond));
pthread_mutex_unlock(&(q->mutex));
}
// 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.
static int Queue_dequeue(Queue *q, uintptr_t *i) {
pthread_mutex_lock(&(q->mutex));
while (q->empty && !q->done)
pthread_cond_wait(&(q->cond), &(q->mutex));
int dequeued;
if (q->empty) {
// We are completely done.
dequeued = 0;
} else {
*i = 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;
}
// Adds the argument to the queue.
// If the queue is full, waits until that changes.
static void Queue_enqueue(Queue *q, uintptr_t i) {
pthread_mutex_lock(&(q->mutex));
while (q->full && !q->done)
pthread_cond_wait(&(q->cond), &(q->mutex));
if (q->done) {
fprintf(stderr, "Error: Attempted to add item to \"done\" queue.\n");
} else {
q->buf[q->next_insert] = i;
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));
}
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;
}
// Shared variables.
static Queue q;
static void *worker_func(void *worker_id_) {
uintptr_t worker_id = (uintptr_t)worker_id_;
unsigned int seed = worker_id; // Whatever.
uintptr_t j;
while (Queue_dequeue(&q, &j)) {
printf("[%" PRIuPTR "] Dequeued %" PRIuPTR "\n", worker_id, j);
// msleep( rand_r(&seed) % 1000 + 1000 ); // Simulate a 1 to 2s load.
printf("[%" PRIuPTR "] Finished processing %" PRIuPTR "\n", worker_id, j);
}
return NULL;
}
int main(void) {
Queue_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);
Queue_enqueue(&q, j);
printf("[x] Enqueued %" PRIuPTR ".\n", j);
}
Queue_done(&q);
printf("[x] Called done.\n");
for (uintptr_t w=0; w<NUM_WORKERS; ++w)
pthread_join(workers[w], NULL);
Queue_destroy(&q);
return 0;
}