# 2021-04-06 Julian-Chu
contributed by < [Julian-Chu](https://github.com/Julian-Chu) >
###### tags: `linux2021`
> [GitHub](https://github.com/Julian-Chu/linux2021-quiz7)

## Part1: Unix like shell

/* Recursively run right-most part of the command line printing output to the    
 * file descriptor @t                                                                                                                                                                  
 */    
static void run(char *c, int t)    
{    
    char *redir_stdin = NULL, *redir_stdout = NULL;    
    int pipefds[2] = {0, 0}, outfd = 0;    
    char *v[99] = {0};    
    char **u = &v[98]; /* end of words */    
    for (;;) {    
        // move to last non-null char    
        c--;    
        if (is_delim(*c)) /* if NULL (start of string) or pipe: break */    
            break;    
        if (!is_special(*c)) {    
            // move back to null termination    
            c++;    /* Copy word of regular chars into previous u */    
            /* 在此提交你的程式碼 */    
            *c = '\0';    
            c--;    
            u--;    
            while(!is_special(*c)){    
                c--;    
            }    
            *u = (c+1);    
        }    
    
        if (is_redir(*c)) { /* If < or > */    
            if (*c == '<')    
                redir_stdin = *u;    
            else    
                redir_stdout = *u;    
            if ((u - v) != 98)    
                u++;    
        }    
    }    
    if ((u - v) == 98) /* empty input */    
        return;   

#### 解說：
原本 c 可能會是 `"0echo hi\0"`, 經過處理後變成 `"0echo\0hi\0"`
然後將v[index] 指向 c 裡面每個命令開頭 char 的位置,
`v[98] = "hi\0"`
`v[97] = "echo\0"`
這樣 v 就可以符合 `execvp` 的參數簽章

#### ref
`man pipe`
`man execvp`: 傳入命令字串陣列

## Part2: httpd

static void dequeue(queue_t *q, int *fd)    
{    
    node_t *old_head;    
    pthread_mutex_lock(q->head_lock);    
    /* Wait until signaled that queue is non_empty.    
     * Need while loop in case a new thread manages to steal the queue    
     * element after the waiting thread is signaled, but before it can    
     * re-acquire head_lock.    
     */    
    /* 在此提交你的程式碼 */;    
    while(q->head == q->tail) pthread_cond_wait(q->non_empty, q->head_lock);    
    old_head = q->head->next;    
    q->head = q->head->next;    
    q->size--;                                                                                                                                                                         
    *fd = old_head->fd;    
    pthread_mutex_unlock(q->head_lock);    
    free(old_head);     
}  

問題1: 執行緒函式 greeter_routine 和 worker_routine 在上述網頁伺服器中，分別負責什麼功能？
* greeter: 負責取得可用的 socket connection, 將 connection 轉換成 file descriptor 後放入 queue
* worker: 從 queue 之中取得可用的 fd, 從 fd 中讀取資料, 做 request 的解析, 並回傳 response

問題2: enqueue 和 dequeue 是單向鏈結串列 (singly-linked list) 所實作的佇列 (queue) 裡頭的基本操作，你如何驗證自己的實作能在並行環境正確無誤？

答: 利用 [Thread Sanitizer](https://clang.llvm.org/docs/ThreadSanitizer.html) 看在 benchmark 的情況下, 是否有發生 data race 的情況

---
## Part3: ringbuffer

/* 提供你的實作 */
#define ALIGN_FLOOR(val, align) (((typeof(val))(val + ((typeof(val))align-1)))&(~((typeof(val))align-1))) 



static inline int ringbuffer_sp_do_enqueue(ringbuf_t *r,    
                                           void *const *obj_table,    
                                           const unsigned n)    
{    
    uint32_t mask = r->prod.mask;    
    uint32_t prod_head = r->prod.head, cons_tail = r->cons.tail;    
    /* The subtraction is done between two unsigned 32-bits value (the result    
     * is always modulo 32 bits even if we have prod_head > cons_tail). So    
     * @free_entries is always between 0 and size(ring) - 1.    
     */                                                                           
    uint32_t free_entries = mask + cons_tail - prod_head;    
    
    /* check that we have enough room in ring */    
    if ((n > free_entries))                                                     
        return -ENOBUFS;                                                        
                                                                                
    uint32_t prod_next = prod_head + n;                                         
    r->prod.head = prod_next;                                                   
                                                                                
    /* write entries in ring */                                                 
    ENQUEUE_PTRS();                                                             
    __compiler_barrier();                                                       
                                                                                
    r->prod.tail = prod_next;                                                   
                                                                                
    /* if we exceed the watermark */    
    /* 請提交你的實作 */     
    if((cons_tail - prod_head + 1 + n) > r->prod.watermark){    
       return -EDQUOT;                                                          
    }    
    return 0;    
} 

:::warning
q1: 這邊有疑問的是`ALIGN_FLOOR`, 因為我的想法對齊應該是 round up/CEIL, 所以實作上是找到下一個 64byte 的倍數, 但就跟 macro 的命名不合?
:::
問題1: 上述程式碼的 watermark 有何作用？跟 lock-less 演算法有何關聯？
答：標記已處理資料的位置, 避免覆蓋掉尚未處理的資料。與 lock-less 演算法的關係, CAS的判斷條件(?)

問題2: 除了題目提到的 SPSC (single-producer and single-consumer)，該如何擴展為 single producer + multiple consumer (SPMC) 和 single producer + multiple consumer (MPMC) 呢？請簡述你的手段，都針對 fixed-size queue。
答: 利用 lock 或是 atomic operation 避免 data race

## Part4: mbus

static bool execute_client_callback(bus_client_t *client, void *msg)                                                                                                                   
{    
    /* Load the client with which we are attempting to communicate. */    
    bus_client_t local_client;    
    __atomic_load(client, &local_client, __ATOMIC_SEQ_CST);    
    
    /* Loop until reference count isupdated or client becomes unregistered */    
    while (local_client.registered) {    
        /* The expected reference count is the current one + 1 */    
        bus_client_t new_client = local_client;    
        ++(new_client.refcnt);    
      
        /* If CAS succeeds, the client had the expected reference count, and    
         * we updated it successfully. If CAS fails, the client was updated    
         * recently. The actual value is copied to local_client.    
         */    
    
        /* 提交你的實作 */ ;    
        if(CAS(&client->refcnt, &new_client.refcnt, &local_client.refcnt)){
            client->callback(client->ctx, msg);
            return true;    
        }    
        __atomic_load(&client->refcnt, &local_client.refcnt, __ATOMIC_SEQ_CST);    
                 
    }                         
     
    /* Client was not registered or got unregistered while we attempted to send    
     * a message    
     */    
    return false;    
} 

問題1: mbus.c 貌似彆扭的 callback function 設計，能夠解決什麼問題？
答：解耦執行緒之間的通訊跟資料傳送, e.g. thread 1不需要知道 thread 2 的具體函數即可以調用

問題2: 考慮到訊息處理的數量可能相當大，我們可事先建立 thread pool 並透過 Processor affinity 讓執行緒在執行 callback function 時，儘量不要相互干擾。請簡述你打算如何修改程式碼，以實現這目標，可斟酌列出關鍵程式碼。
答: 可考慮在 thread_func 引入`sched_setaffinity`指令, 並在 ctx 指定要綁定的 CPU ID