1
// threadpool - a concurrent work queue
2

3
#include <stdio.h>
4
#include <stdlib.h>
5
#include <string.h>
6
#include <unistd.h>
7
#include <pthread.h>
8

9
typedef struct task_node {
10
    char task[1024];
11
    struct task_node *next;
12
} task_node_t;
13

14
typedef struct {
15
    task_node_t *head;
16
    task_node_t *tail;
17
    int done;
18
    pthread_mutex_t mutex;
19
    pthread_cond_t cond;
20
} task_queue_t;
21

22
task_queue_t queue = { .head = NULL, .tail = NULL, .done = 0 };
23

24
// This hash function takes a string and returns a long unsigned integer
25
unsigned long djb2_hash(const char *str) {
26
    unsigned long hash = 5381;
27
    int c;
28
    while ((c = *str++))
29
    hash = ((hash << 5) + hash) + c;
30
    return hash;
31
}
32

33
// This function takes a task, sleeps to stimulate work, computes a hash, then prints the result.
34
void process_task(const char *task) {
35
    usleep(100000);
36
    unsigned long hash = djb2_hash(task);
37
    printf("%lu %s\n", hash, task);
38
    fflush(stdout);
39
}
40

41
// This function creates a new task node and adds it to the tail of the linked list of all the task nodes.
42
void push_task(task_queue_t *q, const char *task) {
43
    task_node_t *node = malloc(sizeof(task_node_t));
44
    strcpy(node->task, task);
45
    node->next = NULL;
46

47
    if (q->tail) {
48
        q->tail->next = node;
49
        q->tail = node;
50
    } else {
51
        q->head = node;
52
        q->tail = node;
53
    }
54
}
55

56
// This function takes the task node at the head of the linked list of all the task nodes.
57
int pop_task(task_queue_t *q, char *buf) {
58
    if (q->head == NULL) return 0;
59
    task_node_t *node = q->head;
60
    strcpy(buf, node->task);
61
    q->head = node->next;
62
    if (q->head == NULL) q->tail = NULL;
63
    free(node);
64
    return 1;
65
}
66

67
// This function takes a void* argument and returns void*.
68
void *worker(void *arg) {
69
    char buffer[1024];
70
    while(1) {
71
        pthread_mutex_lock(&queue.mutex);
72

73
        while(queue.head == NULL && !queue.done) {
74
            pthread_cond_wait(&queue.cond, &queue.mutex);
75
        }
76

77
        int got_task = pop_task(&queue, buffer);
78
        int is_done = queue.done;
79
        pthread_mutex_unlock(&queue.mutex);
80

81
        if(got_task == 1) {
82
            process_task(buffer);
83
        } else if(is_done == 1) {
84
            return NULL;
85
        }
86
    }
87
}
88

89
// This main function is where the code for the tasks exists.
90
int main(int argc, char *argv[]) {
91
    if(argc < 2 || argv[1] == NULL) { return 1; }
92

93
    int thread_count = atoi(argv[1]);
94
    char buffer[1024];
95

96
    pthread_t thread_handles[thread_count];
97
    int worker_ids[thread_count];
98

99
    pthread_mutex_init(&queue.mutex, NULL);
100
    pthread_cond_init(&queue.cond, NULL);
101

102
    for(size_t i = 0; i < thread_count; i++) {
103
        pthread_create(&thread_handles[i], NULL, worker, &worker_ids[i]);
104
    }
105

106
    // This while loop takes the string input and makes a task out of each line in the string.
107
    while (fgets(buffer, sizeof(buffer), stdin) != NULL) {
108
        char *newline = strchr(buffer, '\n');
109
        if(newline) { *newline = '\0'; }
110
        if(buffer[0] == '\0') { continue; }
111

112
        push_task(&queue, buffer);
113
        pthread_cond_signal(&queue.cond);
114
    }
115
    
116
    pthread_mutex_lock(&queue.mutex);
117
    queue.done = 1;
118
    pthread_cond_broadcast(&queue.cond);
119
    pthread_mutex_unlock(&queue.mutex);
120

121
    for(size_t i = 0; i < thread_count; i++) {
122
        pthread_join(thread_handles[i], NULL);
123
    }
124
    return 0;
125
}