owned this note
owned this note
Published
Linked with GitHub
# 2021q1 Homework4 (quiz4)
contributed by < `RZHuangJeff` >
> [Problem set](https://hackmd.io/@sysprog/linux2021-quiz4)
###### tags: `linux2021`
## Problem 1
### Structures
* **threadtask_t**
```cpp
typedef struct __threadtask {
void *(*func)(void *);
void *arg;
struct __tpool_future *future;
struct __threadtask *next;
} threadtask_t;
```
This is a structure that holds tasks that assigned to the thread pool, these tasks are collected by singly linked list.
- `func`: A pointer that points to the function that is asked to be done.
`arg`: Address of the argument that will be passed to `func`.
- `future`: Pointer that points to the corresponding future results of given task.
- `next`: Points to next task in job queue.
* **jobqueue_t**
```cpp
typedef struct __jobqueue {
threadtask_t *head, *tail;
pthread_cond_t cond_nonempty;
pthread_mutex_t rwlock;
} jobqueue_t;
```
This structure represents job queue that holds tasks waiting for a worker to deal with.
- `head`, `tail`: These pointers points to the head and the tail of list of assigned works, respectively.
- `cond_nonempty`: This is a condition variable used to inidcate that the queue is not empty.
- `rwlock`: This is a mutex lock that can be acquired in critical section while ones want to access the queue.
* **tpool_future_t**
```cpp
typedef struct __tpool_future *tpool_future_t;
struct __tpool_future {
int flag;
void *result;
pthread_mutex_t mutex;
pthread_cond_t cond_finished;
};
```
This structure represents the future of an assigned task.
- `flag`: This field holds flags set on this future.
- `result`: This is where the return value of corresponding task will be placed in.
- `mutex`: This mutex lock performs similar task as `rwlock` mentioned before, which avoids potential errors while different threads tries to access such structure at the same time.
- `cond_finished`: This condition variable allows a signal that indicates the given task has finished being broadcasted and waited for.
* **tpool_t**
```cpp
typedef struct __threadpool *tpool_t;
struct __threadpool {
size_t count;
pthread_t *workers;
jobqueue_t *jobqueue;
};
```
This is the main structure represents the pool of threads.
- `count`: This field indicates the amount of active workers running in the pool.
- `workers`: A list that holds worker threads.
- `jobqueue`: A pointer that points to the job queue.
### Code Description
* **tpool_create**
```cpp=
struct __threadpool *tpool_create(size_t count)
{
jobqueue_t *jobqueue = jobqueue_create();
struct __threadpool *pool = malloc(sizeof(struct __threadpool));
if (!jobqueue || !pool) {
if (jobqueue)
jobqueue_destroy(jobqueue);
free(pool);
return NULL;
}
pool->count = count, pool->jobqueue = jobqueue;
if ((pool->workers = malloc(count * sizeof(pthread_t)))) {
for (int i = 0; i < count; i++) {
if (pthread_create(&pool->workers[i], NULL, jobqueue_fetch,
(void *) jobqueue)) {
for (int j = 0; j < i; j++)
pthread_cancel(pool->workers[j]);
for (int j = 0; j < i; j++)
pthread_join(pool->workers[j], NULL);
free(pool->workers);
jobqueue_destroy(jobqueue);
free(pool);
return NULL;
}
}
return pool;
}
jobqueue_destroy(jobqueue);
free(pool);
return NULL;
}
```
This function aims to create a thread pool with `count` workers processing given work. At very begining, a `jobqueue` is created via involking function `jobqueue_create` where assigned works lined up waiting for processing (line 3), then, a structure `struct __threadpool` is created (line 4), after such creation, an error checking is performed and `NULL` is returned while there is something wrong when creating the queue and thread pool (line 5 - 10). Next, worker routes (function `jobqueue_fetch`) are created by involking function `pthread_create`, when facing any problem that causes failure on creating a worker route, workers that were already created is canceled and joined, moreover, queue and pool are also destroyed (line 15 - 25). If every thing works fine that workers are all ready for dealing with arriving of jobs, the thread pool created before is returned (line 27). After all, while it failed for allocating space for workers, the queue and pool is destroyed and `NULL` is returned (line 30 - 32).
* **jobqueue_create**
```cpp=
static jobqueue_t *jobqueue_create(void)
{
jobqueue_t *jobqueue = malloc(sizeof(jobqueue_t));
if (jobqueue) {
jobqueue->head = jobqueue->tail = NULL;
pthread_cond_init(&jobqueue->cond_nonempty, NULL);
pthread_mutex_init(&jobqueue->rwlock, NULL);
}
return jobqueue;
}
```
This function aims to allocate the the space for new `jobqueue_t` entity (line 3). `head` and `tail` is initialized to `NULL` at very begining, `cond_nonempty` and `rwlock` are also initialized by corresponding initializing function (line 4 - 8).
* **jobqueue_destroy**
```cpp=
static void jobqueue_destroy(jobqueue_t *jobqueue)
{
threadtask_t *tmp = jobqueue->head;
while (tmp) {
jobqueue->head = jobqueue->head->next;
pthread_mutex_lock(&tmp->future->mutex);
if (tmp->future->flag & __FUTURE_DESTROYED) {
pthread_mutex_unlock(&tmp->future->mutex);
pthread_mutex_destroy(&tmp->future->mutex);
pthread_cond_destroy(&tmp->future->cond_finished);
free(tmp->future);
} else {
tmp->future->flag |= __FUTURE_CANCELLED;
pthread_mutex_unlock(&tmp->future->mutex);
}
free(tmp);
tmp = jobqueue->head;
}
pthread_mutex_destroy(&jobqueue->rwlock);
pthread_cond_destroy(&jobqueue->cond_nonempty);
free(jobqueue);
}
```
This function aims to destroy a given job queue. All non-destroyed tasks in the queue are canceled by setting its cancel flag, and the task itself is freed (line 3 - 18), after all, the mutex lock and condition variable corresponding to given job queue is destroyed (line 20 - 21), and the queue itself is freed (line 22).
* **tpool_apply**
```cpp=
struct __tpool_future *tpool_apply(struct __threadpool *pool,
void *(*func)(void *),
void *arg)
{
jobqueue_t *jobqueue = pool->jobqueue;
threadtask_t *new_head = malloc(sizeof(threadtask_t));
struct __tpool_future *future = tpool_future_create();
if (new_head && future) {
new_head->func = func, new_head->arg = arg, new_head->future = future;
pthread_mutex_lock(&jobqueue->rwlock);
if (jobqueue->head) {
new_head->next = jobqueue->head;
jobqueue->head = new_head;
} else {
jobqueue->head = jobqueue->tail = new_head;
pthread_cond_broadcast(&jobqueue->cond_nonempty);
}
pthread_mutex_unlock(&jobqueue->rwlock);
} else if (new_head) {
free(new_head);
return NULL;
} else if (future) {
tpool_future_destroy(future);
return NULL;
}
return future;
}
```
This function will attach the given function to the end of job queue. At begining, an entity of `threadtask_t` and future that corresponding to the task are created (line 6 - 7). The job function and the argument are assigned to newly created task (line 9). Finally, the newly created task is attached to the head of job queue (line 11 - 17), when the queue is empty initially, a signal is broadcasted through `cond_nonempty` to tell workers that there is a job waiting for dealing with (line 16). It is important that asking for `rwlock` before accessing the job queue (line 10), since this queue is shared between workers and the caller, and that lock should be unlocked after modifying is done (line 18). Program from line 19 to line 25 is dealing with exceptional situations, that is, failed to allocate space for new task or future.
* **jobqueue_fetch**
```cpp=
static void *jobqueue_fetch(void *queue)
{
jobqueue_t *jobqueue = (jobqueue_t *) queue;
threadtask_t *task;
int old_state;
pthread_cleanup_push(__jobqueue_fetch_cleanup, (void *) &jobqueue->rwlock);
while (1) {
pthread_mutex_lock(&jobqueue->rwlock);
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_state);
pthread_testcancel();
while (!jobqueue->tail)
pthread_cond_wait(&jobqueue->cond_nonempty, &jobqueue->rwlock);
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_state);
if (jobqueue->head == jobqueue->tail) {
task = jobqueue->tail;
jobqueue->head = jobqueue->tail = NULL;
} else {
threadtask_t *tmp;
for (tmp = jobqueue->head; tmp->next != jobqueue->tail;
tmp = tmp->next)
;
task = tmp->next;
tmp->next = NULL;
jobqueue->tail = tmp;
}
pthread_mutex_unlock(&jobqueue->rwlock);
if (task->func) {
pthread_mutex_lock(&task->future->mutex);
if (task->future->flag & __FUTURE_CANCELLED) {
pthread_mutex_unlock(&task->future->mutex);
free(task);
continue;
} else {
task->future->flag |= __FUTURE_RUNNING;
pthread_mutex_unlock(&task->future->mutex);
}
void *ret_value = task->func(task->arg);
pthread_mutex_lock(&task->future->mutex);
if (task->future->flag & __FUTURE_DESTROYED) {
pthread_mutex_unlock(&task->future->mutex);
pthread_mutex_destroy(&task->future->mutex);
pthread_cond_destroy(&task->future->cond_finished);
free(task->future);
} else {
task->future->flag |= __FUTURE_FINISHED;
task->future->result = ret_value;
pthread_cond_broadcast(&task->future->cond_finished);
pthread_mutex_unlock(&task->future->mutex);
}
free(task);
} else {
pthread_mutex_destroy(&task->future->mutex);
pthread_cond_destroy(&task->future->cond_finished);
free(task->future);
free(task);
break;
}
}
pthread_cleanup_pop(0);
pthread_exit(NULL);
}
```
This is the most important function in this implementation of thread pool, since that this is a worker function which will keep fetching tasks from given job queue and dealing with it, this worker function will blocking wait a task arrives while the job queue is empty (line 14 - 15), which will not consume any calculating power comparing with polling wait.
Just mentioned before, since job queue is shared by lots of workers, a worker would need to ask for `rwlock` before accesses queue (line 10). Notice that the lock will be released while a thread is waiting for a condition happened via involking `pthread_cond_wait`, and it will be re-locked when the function returns.
The worker can only be canceled when it is waiting for a job, which is set by function `pthread_setcancelstate` (line 11, 17).
After a job was fetched from the job queue (line 18 - 29), worker will check whether pointer `func` points to a valid function (line 32), if so, the state of that task will be checked before it actually runs, when it finds that the task is cancelled, that task will simply be discarded (line 34 - 37), otherwise, the state of the task is set to `__FUTURE_RUNNING` and is involked with argument assigned before (line 39, 43).
After return from the job function, the state of the task will again be checked whether it was set to `__FUTURE_DESTROYED`, if so, the task will be freed (line 45 - 49), otherwise, the state of that task is set to `__FUTURE_FINISHED`, the result is assigned to `ret_value`, and finally, a signal is sent via condition variable `cond_finished`.
When a worker recieves a task whose `func` points to `NULL`, it will break out from its working loop and exit the thread that it stayed in (line 58 - 67).
After all, we can find that the entire working loop is wrapped by two macros : `pthread_cleanup_push` and `pthread_cleanup_pop`, which will be expended to program shown below.
```cpp=
/* pthread_clean_push(__jobqueue_fetch_cleanup, (void *) &jobqueue->rwlock) */
do {
__pthread_unwind_buf_t __cancel_buf;
void (*__cancel_routine) (void *) = (__jobqueue_fetch_cleanup);
void *__cancel_arg = ((void *) &jobqueue->rwlock);
int __not_first_call = __sigsetjmp ((struct __jmp_buf_tag *) (void *) __cancel_buf.__cancel_jmp_buf, 0);
if (__glibc_unlikely (__not_first_call)) {
__cancel_routine (__cancel_arg);
__pthread_unwind_next (&__cancel_buf);
}
__pthread_register_cancel (&__cancel_buf);
do {
// working loop stays here
/* pthread_cleanup_pop(0) */
do { } while (0);
} while (0);
__pthread_unregister_cancel (&__cancel_buf);
if (0)
__cancel_routine (__cancel_arg);
} while (0);
```
We can figure out that these macros acturally forms a `do ... while` loop. In this section, an entity of `__pthread_unwind_buf_t` is created (line 3), and the given function and argument are assigned to local variable `__cancel_route` and `__cancel_arg`, respectively. Next, an important statement is `__sigsetjmp` which is similar as `sigsetjmp`, both of them set a breakpoint for unconditional long jump, we can find out that the clean up function we set is involked in the `if` statement around line 7 - 10. The macro `__glibc_unlikely(__not_first_call)`, which is expended to `__builtin_expect((__not_first_call), 0)`, such macro provides branch prediction information to compiler to optimize such `if` statement. Such condition indicates that whether the return from `__sigsetjmp` is from long jump (from a call to `pthread_exit` or `pthread_cancel`) or not, if so, the clean up functions set before will be involked to perform clean up process.
:::warning
TODO: clarify the use of cancellation of POSIX threads. You should emphasize on its high-level motiviation and purpose at first glance.
:notes: jserv
:::
* **tpool_future_create**
```cpp
static struct __tpool_future *tpool_future_create(void)
{
struct __tpool_future *future = malloc(sizeof(struct __tpool_future));
if (future) {
future->flag = 0;
future->result = NULL;
pthread_mutex_init(&future->mutex, NULL);
pthread_condattr_t attr;
pthread_condattr_init(&attr);
pthread_cond_init(&future->cond_finished, &attr);
pthread_condattr_destroy(&attr);
}
return future;
}
```
This function aims to allocate space for a new future and initialize fields of it to their default value.
* **tpool_future_get**
```cpp=
void *tpool_future_get(struct __tpool_future *future, unsigned int seconds)
{
pthread_mutex_lock(&future->mutex);
/* turn off the timeout bit set previously */
future->flag &= ~__FUTURE_TIMEOUT;
while ((future->flag & __FUTURE_FINISHED) == 0) {
if (seconds) {
struct timespec expire_time;
clock_gettime(CLOCK_MONOTONIC, &expire_time);
expire_time.tv_sec += seconds;
int status = pthread_cond_timedwait(&future->cond_finished,
&future->mutex, &expire_time);
if (status == ETIMEDOUT) {
future->flag |= __FUTURE_TIMEOUT;
pthread_mutex_unlock(&future->mutex);
return NULL;
}
} else
pthread_cond_wait(&future->cond_finished, &future->mutex);
}
pthread_mutex_unlock(&future->mutex);
return future->result;
}
```
With involking this function, a caller can perform two kinds of waiting on given future. While `seconds` is set to 0, this function will blocking wait desired future until recieves its `cond_finished` signal (line 19). Otherwise, this function will wait given future upto `seconds` seconds. To perform timed wait, it waits `cond_finished` with function `pthread_cond_timedwait` which will return not only when it recieved desired condition, but also when ran out of time (line 7 - 17). For the situation that `pthread_cond_timedwait` returns due to running out of time, its return value will equal to `ETIMEDOUT`, with performing checking on that error code, a future can be marked as `__FUTURE_TIMEOUT` properly (line 13 - 17).
* **tpool_future_destroy**
```cpp=
int tpool_future_destroy(struct __tpool_future *future)
{
if (future) {
pthread_mutex_lock(&future->mutex);
if (future->flag & __FUTURE_FINISHED ||
future->flag & __FUTURE_CANCELLED) {
pthread_mutex_unlock(&future->mutex);
pthread_mutex_destroy(&future->mutex);
pthread_cond_destroy(&future->cond_finished);
free(future);
} else {
future->flag |= __FUTURE_DESTROYED;
pthread_mutex_unlock(&future->mutex);
}
}
return 0;
}
```
This function will destroy the given future. For futures that has already been marked as `__FUTURE_FINISHED` or `__FUTURE_CANCELLED`, they are destroyed directly (line 5 - 10). Otherwise, it is marked as `__FUTURE_DESTROYED` (line 11 14) and will be freed while a worker detect such flag.
* **tpool_join**
```cpp=
int tpool_join(struct __threadpool *pool)
{
size_t num_threads = pool->count;
for (int i = 0; i < num_threads; i++)
tpool_apply(pool, NULL, NULL);
for (int i = 0; i < num_threads; i++)
pthread_join(pool->workers[i], NULL);
free(pool->workers);
jobqueue_destroy(pool->jobqueue);
free(pool);
return 0;
}
```
This function will send `pool->count` null tasks to the job queue (line 3 - 5), which will make all of the workers stop their working loop, after that it waits for all workers to join (line 6 - 7), finally, the job queue an the pool itself is freed (8 - 10).
* **__jobqueue_fetch_cleanup**
```cpp
static void __jobqueue_fetch_cleanup(void *arg)
{
pthread_mutex_t *mutex = (pthread_mutex_t *) arg;
pthread_mutex_unlock(mutex);
}
```
This is the clean up function that will be involked while a worker is cancelled or exit normally, which will unlock the mutex lock while that lock is initially locked by cancelled worker. With this clean up function, we can ensure that cancelling a worker in any time will not cause a deadlock.
### Improvement
* **Time complexity taken on queue popping**
In the original version of thread pool, a worker will run through all tasks in waiting queue before it really targets a task and pops it from the queue, following is corresponding part of code (the `for` loop around line 6 - 8):
```cpp=
if (jobqueue->head == jobqueue->tail) {
task = jobqueue->tail;
jobqueue->head = jobqueue->tail = NULL;
} else {
threadtask_t *tmp;
for (tmp = jobqueue->head; tmp->next != jobqueue->tail;
tmp = tmp->next)
;
task = tmp->next;
tmp->next = NULL;
jobqueue->tail = tmp;
}
```
How serious will queue traversing affect the performace depends on the tasks assigned to the thread pool. While assigned tasks are simple that the speed of workers completeing a task is faster than producer assigning a task, the length of queue can hardly grouth, which makes traversing take little amount of time. But this becomes different while producer assigning heavy load tasks that workers need lots of time to complete a task, which makes the length of queue becoming large easily, and traversing through a large queue consumes lots of time.
Commit [dc80584](https://github.com/RZHuangJeff/tpool/commit/dc805842b6bcdbb144d51639c52ace0687be5382) fixes such problem, following figures present performace improved from one to another:
> $\uparrow$ heavy work load
> $\uparrow$ trivial work load
As shown in figure, there is a tremendous improvement for complex tasks, while there is only a slightly difference for trivial tasks.
* **Improve scalability**
Following figure shows different result of scalability of various implementation of queue.
* [original version](https://github.com/RZHuangJeff/tpool/commit/ee5e48514279f2f3ba61ad7e791b1a8b60100f89)
* [O(1) queue](https://github.com/RZHuangJeff/tpool/commit/dc805842b6bcdbb144d51639c52ace0687be5382)
* [atomic queue](https://github.com/RZHuangJeff/tpool/commit/f57bb7e86caef6808ff2c7d9b4b5868ef49ef6f3)
> $\uparrow$ Throughput of executing on different number of cores
As shown in figure, we can see that the original version has poor scalability, only a little increament on throughput when executed on more and more cores.
On the other hand, we can figure out that both atomic and O(1) have better scalabitiy than original one, since there is massive increament on throughput while number of executing core grows.
We can see that scalability of the O(1) version is almost the same as one of atomic version. The detail of O(1) queue is described [here](#Improvement), with that description, we can conclude that scalability of an implementation of thread pool can be improved by shrinking size of critical sections.
Sign in with Wallet
