owned this note
owned this note
Published
Linked with GitHub
# 2021q1 Homework4 (quiz4)
contributed by < `yellow-hank` >
###### tags: `LinuxKernel`
## 測驗`1`
### 解釋程式碼
```cpp
enum __future_flags {
__FUTURE_RUNNING = 01,
__FUTURE_FINISHED = 02,
__FUTURE_TIMEOUT = 04,
__FUTURE_CANCELLED = 010,
__FUTURE_DESTROYED = 020,
};
```
> C99 6.7.2.2 Enumeration specifiers
> 2. The expression that defines the value of an enumeration constant shall be an integer constant expression that has a value representable as an int.
這裡是以八進位進行參數的設定，需要了解的是 enum 只接受整數，任何整數變數是不允許的。
---
:::warning
延伸思考：為何 C 語言原生支援八進位表示法和運算？什麼場合用得到？
:notes: jserv
:::
### 延伸問題回答
對於記憶位址空間只有 12 位元 24 位元等位元為 3 的倍數，在運算時使用八進位可以方便做運算，如果使用 16 進位運算會有空間上的浪費。在這種記憶體位址空間非常小的硬體上使用八進位，會比十六進位來得更恰當。在嵌入式系統軟體比較常見，因為需要節省記憶體的空間，多放入會增加成本。現在使用八進位的時機越來越少，因為記憶體空間取得容易。Linux 目前有使用到的地方為檔案權限管理，在設定讀( r )、寫( w )、執行( x )的權限時，使用八進位非常方便。只需要紀錄有無的狀態，3 個位元能紀錄權限的狀態。在 Linux 會透過 `chmod` 來變更檔案權限，例如：
```
chmod 744 test.c
```
其中的 744 轉會為八進位分別為
```graphviz
digraph struct{
node [shape = record];
struct1[label="{{7|4|4}|{1|1|1|1|0|0|1|0|0}|{r|w|x|r|w|x|r|w|x}}"];
}
```
這樣能清楚的表示檔案權限，如果使用二進位會顯得的太過冗長。
---
#### 資料結構
```cpp
typedef struct __threadtask {
void *(*func)(void *);
void *arg;
struct __tpool_future *future;
struct __threadtask *next;
} threadtask_t;
typedef struct __jobqueue {
threadtask_t *head, *tail;
pthread_cond_t cond_nonempty;
pthread_mutex_t rwlock;
} jobqueue_t;
struct __tpool_future {
int flag;
void *result;
pthread_mutex_t mutex;
pthread_cond_t cond_finished;
};
struct __threadpool {
size_t count;
pthread_t *workers;
jobqueue_t *jobqueue;
};
```
每個執行緒的任務是採用鏈結串列的資料結構進行儲存，每個任務中要包含需要執行的函式、參數、執行結果、執行狀態。對於並行程式，需要針對 critical section 進行鎖定，保證不會有兩個執行緒同時更改重要的資料，導致結果不正確。
以下是 jobqueue 的示意圖：
```graphviz
digraph {
node[shape = record]
struct1[label ="head"];
struct2[label = "task1"];
struct3[label = "task2"];
struct4[label = "task3"];
struct5[label = "tail"];
struct1 -> struct2;
struct2 -> struct3;
struct3 -> struct4;
struct5 -> struct4;
}
```
```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;
}
```
`tpool_future_create` 創造 future 資料結構，並且初始化，不論成功與否都會回傳值，所以需要額外檢查 future 是否建立成功
```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;
}
```
`tpool_future_destroy` 釋放 future 資料結構
在檢測任務的執行狀態，這裡使用 bitwise and 取代等於運算子，等於運算子是透過減法之後跟 0 比較，所以速度會較慢。針對任務完成或是被取消進行釋放。這裡需要進行鎖定，因為同時有可能會有其他的執行緒進行更動造成狀態的更動，進而導致執行結果的錯誤。
```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;
}
```
`tpool_future_get` 取得函式計算完的結果
設計兩種等待方式，第一種方式除了等待條件和另外設定一個計時器，如果時間到，則會回傳 ETIMEDOUT，脫離等待特定條件。第二種方式會一直等待到條件成立時，才離開。第一種設計優點是，可以有良好的回應時間，針對運算量大的的函式，可以之後再來取得運算結果，讓程式不會在此空等結果。
#### timeout 機制
CLOCK_MONOTONIC 的參數用意為，此時間不能經由設定更改(包括系統管理員更動系統時間，此時間不受影響)，時間的起點是在先前的某個隨意的點開始紀錄。在 Linux 上，時間的起始點為系統開機的時刻。
:::danger
依據 man page ([clock_gettime](https://linux.die.net/man/3/clock_gettime))
> `CLOCK_MONOTONIC`: Clock that cannot be set and represents monotonic time since—as described by POSIX—"some unspecified point in the past". On Linux, that point corresponds to the number of seconds that the system has been running since it was booted.
用語和你上述表達不同。
:notes: jserv
:::
利用上述的優點，在 `pthread_cond_timedwait` 裡面會持續檢查現在的 CLOCK_MONOTONIC 時間和 abstime 的差值，當檢測到值為負時，說明時間已經到了，會離開函式並且回傳 ETIMEDOUT ，以下節錄自 pthread_ond_wait.c
```cpp
while(1) {
...
/* Block, but with a timeout.
Work around the fact that the kernel rejects negative timeout
values despite them being valid. */
if (__glibc_unlikely (abstime->tv_sec < 0))
err = ETIMEDOUT;
else if ((flags & __PTHREAD_COND_CLOCK_MONOTONIC_MASK) != 0)
{
/* CLOCK_MONOTONIC is requested. */
struct timespec rt;
if (__clock_gettime (CLOCK_MONOTONIC, &rt) != 0)
__libc_fatal ("clock_gettime does not support "
"CLOCK_MONOTONIC\n");
/* Convert the absolute timeout value to a relative
timeout. */
rt.tv_sec = abstime->tv_sec - rt.tv_sec;
rt.tv_nsec = abstime->tv_nsec - rt.tv_nsec;
if (rt.tv_nsec < 0)
{
rt.tv_nsec += 1000000000;
--rt.tv_sec;
}
/* Did we already time out? */
if (__glibc_unlikely (rt.tv_sec < 0))
err = ETIMEDOUT;
else
err = futex_reltimed_wait_cancelable
(cond->__data.__g_signals + g, 0, &rt, private);
}
else
{
/* Use CLOCK_REALTIME. */
err = futex_abstimed_wait_cancelable
(cond->__data.__g_signals + g, 0, abstime, private);
}
...
}
```
```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;
}
```
`jobqueue_create` 建立一個 queue 用來紀錄任務，呼叫函式需要額外檢查函式的回傳值，避免存取到 NULL 導致程式無預期終止
```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);
}
```
`jobqueue_destroy` 釋放所有 jobqueue 內的所有資源，針對移除的順序是由貯列的前面開始釋放資源，針對被標定為 `__FUTURE_DESTROYED` 進行額外 future 資料結構的釋放。
```cpp
static void __jobqueue_fetch_cleanup(void *arg)
{
pthread_mutex_t *mutex = (pthread_mutex_t *) arg;
pthread_mutex_unlock(mutex);
}
```
`__jobqueue_fetch_cleanup` 是一個會在執行緒被取消或是特殊情況下被執行的函式，用意是釋放參數的鎖定，不然其他執行緒無法使用這些參數，造成 deadlock 的情況發生。
> These functions manipulate the calling thread's stack of thread-
cancellation clean-up handlers. A clean-up handler is a function
that is automatically executed when a thread is canceled (or in
various other circumstances described below); it might, for
example, unlock a mutex so that it becomes available to other
threads in the process.
> from [pthread_cleanup_push man page](https://man7.org/linux/man-pages/man3/pthread_cleanup_push.3.html)
```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);
}
```
`jobqueue_fetch` 挑選任務的執行順序，此函式採取的方式是取 jobqueue 的最後一個當做此次的執行任務，挑選策略為先進先出。針對完成計算的任務，需要告訴其他執行緒他完成計算，接續之後的動作。針對被被破壞的任務，需進行資源的釋放。針對取消狀態的任務，需要釋放資源，並且重新挑選一個任務。
```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;
}
```
`tpool_create` 建立一個有 count 個執行緒的執行緒池子，在建立執行緒有失敗的情況發生時，需要將之前建立的執行緒，進行資源的釋放，不然會有記憶體洩漏。
:::info
```cpp
if (!jobqueue || !pool) {
if (jobqueue)
jobqueue_destroy(jobqueue);
free(pool);
return NULL;
}
```
在 jobqueue 配置空間成功，但是 pool 配置失敗時，會有 free NULL 的情況發生，修正程式碼如下:
```cpp
if (!jobqueue || !pool) {
if (jobqueue)
jobqueue_destroy(jobqueue);
if (pool)
free(pool);
return NULL;
}
```
:::
```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;
}
```
`tpool_apply` 加入一個任務到 pool，加入的優先次序為先進先出，加入失敗時，回傳 NULL ，成功時回傳 future 的資料結構。針對 jobqueue 的狀態為空時，需要額外告訴其他執行緒任務佇列現在不為空
```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;
}
```
`tpool_join` 等待所有的執行緒終止，終止後釋放所有資源
Sign in with Wallet
