L11: ktcp

主講人: jserv / 課程討論區: 2024 年系統軟體課程

返回「Linux 核心設計/實作」課程進度表

比較 kthread 實作版本與 CMWQ 實作版本的執行差異

根據 kecho pull request #1 指出 CMWQ 版本的實作得益於 locality 及事先準備的執行緒使得 server 的執行時間可優於 kthread 的版本。

利用 commit 7038c2 並修正其問題改寫為 kthread-based kecho 並與目前使用 CMWQ 所實作的版本比較二者的差異。

• kthread-based 實作的測試結果
  
• CMWQ 實作的測試結果
  

可見二者在執行上的差異極大，主要的原因就是因為 kthread-based 的版本在收到 client 連線請求之後才會開始處理 kthread 的建立。而 CMWQ 因為讓 workqueue 可以依照目前任務執行的狀態來分配對應的執行緒，而且 workqueue 使用 thread pool 的概念，因此不用額外再重新建立新的 kthread，在 client 連線時可直接將 worker 函式分配給空閒的執行緒來進行。省去建立 kthread 所耗費的時間成本，在大量的連線湧入時，更能體現其好處。

測量 kthread 建立成本

kecho pull request #1 提到 kthread-based 的 kecho 在回應時間上有很大的比例是受到 kthread 建立的成本影響，為瞭解其實際造成影響的比例，利用 eBPF 來測量建立 kthread_run 的時間成本。

為了測量 kthread 建立的成本，使用 my_thread_run 來包裝 kthread_run 讓 eBPF 可以將測量的動作注入在該函式裡面：

from bcc import BPF
code = """
#include <uapi/linux/ptrace.h>

BPF_HASH(start, u64);

int probe_handler(struct pt_regs *ctx)
{
	u64 ts = bpf_ktime_get_ns();
	bpf_trace_printk("in %llu\\n",ts);
	return 0;
}

int end_function(struct pt_regs *ctx)
{
	u64 ts = bpf_ktime_get_ns();
	bpf_trace_printk("out %llu\\n",ts);
	return 0;
}
"""

b = BPF(text = code)
b.attach_kprobe(event = 'my_thread_run', fn_name = 'probe_handler')
b.attach_kretprobe(event = 'my_thread_run', fn_name = 'end_function')

while True:
	try:
		res = b.trace_fields()
	except ValueError:
		continue
	print(res[5].decode("UTF-8"))

在注入 kprobe 之後，執行 bench 進行測試：

可以發現 kthread_run 的執行成本大約為 200 us。

參考資料

1. workqueue: implement high priority workqueue
2. linux/workqueue.h - git commit log
3. Reimplementing printk()
4. kecho pull request #1

引入 CMWQ 到 khttpd

在 kecho 的實作中，為了有效管理 work ，所有的 work 都會被加到一個鏈結串列，可在 http_server.h 新增以下結構體:

struct httpd_service {
    bool is_stopped;
    struct list_head head;
};
extern struct httpd_service daemon_list;

該結構體的作用是充當鏈結串列的首個節點，成員 is_stopped 用以判斷是否有結束連線的訊號發生

接著修改原本的結構體 struct http_request ，新增鏈結串列節點及 work 結構體:

struct http_request {
    struct socket *socket;
    enum http_method method;
    char request_url[128];
    int complete;
+   struct list_head node;
+   struct work_struct khttpd_work;
};

整個程式的主要流程是建立 CMWQ

$\to$ 連線建立後建立 work

$\to$ workqueue 開始運作

$\to$ 釋放所有記憶體。

首先建立 CMWQ 的部份在掛載模組時執行，位於函式 khttpd_init ，以下為修改的部份

static int __init khttpd_init(void)
{
    int err = open_listen_socket(port, backlog, &listen_socket);
    if (err < 0) {
        pr_err("can't open listen socket\n");
        return err;
    }
    param.listen_socket = listen_socket;

+   // create CMWQ
+   khttpd_wq = alloc_workqueue(MODULE_NAME, 0, 0);
    http_server = kthread_run(http_server_daemon, &param, KBUILD_MODNAME);
    if (IS_ERR(http_server)) {
        pr_err("can't start http server daemon\n");
        close_listen_socket(listen_socket);
        return PTR_ERR(http_server);
    }
    return 0;
}

使用函式 alloc_workqueue 建立 CMWQ ，而這裡要注意參數 flag 的值會根據需求而不同，根據 kecho 的註解說明，如果是想要長時間連線，像是使用 telnet 連線，可將 flag 設成 WQ_UNBOUND ，否則指定為 0 即可

自己實際二個都設定過，的確使用 WQ_UNBOUND 的效率沒有來的非常好，主要原因可能是 work 可能會被 delay 導致，也有發生測試的時候電腦當機的情況

接著是建立 work 的部份，使用時機是在 server 和 client 建立連線後，以下新增函式 create_work 用來新增 work

static struct work_struct *create_work(struct socket *sk)
{
    struct http_request *work;

    // 分配 http_request 結構大小的空間
    // GFP_KERNEL: 正常配置記憶體
    if (!(work = kmalloc(sizeof(struct http_request), GFP_KERNEL)))
        return NULL;

    work->socket = sk;
    
    // 初始化已經建立的 work ，並運行函式 http_server_worker
    INIT_WORK(&work->khttpd_work, http_server_worker);

    list_add(&work->node, &daemon_list.head);

    return &work->khttpd_work;
}

函式 create_work 主要流程為建立 work 所需的空間

$\to$ 初始化 work

$\to$ 將 work 加進鏈結串列裡。

最後釋放記憶體的部份單純許多，就是走訪整個鏈結串列，並逐一釋放。

static void free_work(void)
{
    struct http_request *l, *tar;
    /* cppcheck-suppress uninitvar */

    list_for_each_entry_safe (tar, l, &daemon_list.head, node) {
        kernel_sock_shutdown(tar->socket, SHUT_RDWR);
        flush_work(&tar->khttpd_work);
        sock_release(tar->socket);
        kfree(tar);
    }
}

使用命令 ./htstress http://localhost:8081 -t 3 -c 20 -n 200000 測試，以下為執行結果

requests:      200000
good requests: 200000 [100%]
bad requests:  0 [0%]
socker errors: 0 [0%]
seconds:       3.861
requests/sec:  51801.192

可以發現整個伺服器的吞吐量 (throughput) 有大幅的成長

原本的實作	新增 CMWQ
30274.801	51801.192

HTTP keep-alive 模式

如下圖，可以簡單將 HTTP 分成二種傳輸模式，分別是 multiple connections 及 persistent connection ，前者會在伺服器回應請求之後中斷連線，後者則會持續保持連線，根據 HTTP 的敘述，可以得到幾件資訊

1. 在 HTTP 1.0 的版本中，預設的連線模式為 multiple connection ，如果要使用 persistent connection ，則需要在 header 添加以下資訊

   Connection: keep-alive

2. 在 HTTP 1.1 的版本則是預設使用 persistent connection ，允許在單一連線下處理多個請求

這邊可以利用 khttpd 做簡單的測試，使用命令 telnet localhost 8081 進行連線，在分別輸入 GET / HTTP/1.0 及 GET / HTTP/1.1 進行測試，並分別觀察伺服器回傳的資料

GET / HTTP/1.0

HTTP/1.1 200 OK
Server: khttpd
Content-Type: text/plain
Content-Length: 12
Connection: Close

GET / HTTP/1.1

HTTP/1.1 200 OK
Server: khttpd
Content-Type: text/plain
Content-Length: 12
Connection: Keep-Alive

根據回傳的 Connection: xxxxx 資訊可以得知，結果符合上述的敘述，因此可確認 kHTTPd 本身就有 keep-alive 的功能

實作 directory listing 功能

為了實作 directory listing 的功能，首先要做的第一件事就是讀取現行目錄的檔案名稱，新增函式 handle_directory 用來實踐該功能，完整的修改可以參考 Add the function of directory list

static bool handle_directory(struct http_request *request)
{
    struct file *fp;
    char buf[SEND_BUFFER_SIZE] = {0};

    request->dir_context.actor = tracedir;
    if (request->method != HTTP_GET) {
        snprintf(buf, SEND_BUFFER_SIZE,
                 "HTTP/1.1 501 Not Implemented\r\n%s%s%s%s",
                 "Content-Type: text/plain\r\n", "Content-Length: 19\r\n",
                 "Connection: Close\r\n", "501 Not Implemented\r\n");
        http_server_send(request->socket, buf, strlen(buf));
        return false;
    }

    snprintf(buf, SEND_BUFFER_SIZE, "HTTP/1.1 200 OK\r\n%s%s%s",
             "Connection: Keep-Alive\r\n", "Content-Type: text/html\r\n",
             "Keep-Alive: timeout=5, max=1000\r\n\r\n");
    http_server_send(request->socket, buf, strlen(buf));


    snprintf(buf, SEND_BUFFER_SIZE, "%s%s%s%s", "<html><head><style>\r\n",
             "body{font-family: monospace; font-size: 15px;}\r\n",
             "td {padding: 1.5px 6px;}\r\n",
             "</style></head><body><table>\r\n");
    http_server_send(request->socket, buf, strlen(buf));

    fp = filp_open("/home/benson/khttpd/", O_RDONLY | O_DIRECTORY, 0);
    if (IS_ERR(fp)) {
        pr_info("Open file failed");
        return false;
    }

    iterate_dir(fp, &request->dir_context);
    snprintf(buf, SEND_BUFFER_SIZE, "</table></body></html>\r\n");
    http_server_send(request->socket, buf, strlen(buf));
    filp_close(fp, NULL);
    return true;
}

函式 handle_directory 主要做以下幾件事

1. 判斷 clent 的請求是否為 GET ，並送出對應的 HTTP header (第 7 ~ 19 行)
2. 開啟現行目錄並透過函式 iterate_dir 走訪目錄內的所有資料夾 (第 28 ~ 34 行)
3. 結束連線

接著根據上述的第 6 行，將把函式 iterate_dir 導向到函式 tracedir ，換言之就是在執行函式 iterate_dir 的過程中會呼叫 tracedir ，以下為函式 tracedir 的實作

// callback for 'iterate_dir', trace entry.
static int tracedir(struct dir_context *dir_context,
                    const char *name,
                    int namelen,
                    loff_t offset,
                    u64 ino,
                    unsigned int d_type)
{
    if (strcmp(name, ".") && strcmp(name, "..")) {
        struct http_request *request =
            container_of(dir_context, struct http_request, dir_context);
        char buf[SEND_BUFFER_SIZE] = {0};

        snprintf(buf, SEND_BUFFER_SIZE,
                 "<tr><td><a href=\"%s\">%s</a></td></tr>\r\n", name, name);
        http_server_send(request->socket, buf, strlen(buf));
    }
    return 0;
}

函式 tracedir 的功能就是會走訪整個目錄的資料，並且每執行一次就會將資料送到 client

而這裡有個較特別之處，即使用到巨集 container_of ，由於函式 tracedir 的參數是固定的，又需要 socket 參數來送出資料，因此這邊將結構 dir_context 放進結構 http_request 裡，如此一來，透過巨集 container_of 就可以達到不用傳遞 socket 也可以使用的效果

struct http_request {
    struct socket *socket;
    enum http_method method;
    char request_url[128];
    int complete;
+   struct dir_context dir_context;
    struct list_head node;
    struct work_struct khttpd_work;
};

最後展現目前的結果 (節錄部份)

取得現行目錄

首先節錄主要測試的程式碼，使用到的函式位於 fs/d_path.c 及 fs/namei.c

struct path pwd;
char *cwd;
char current_path[100] = {0}, buf[SEND_BUFFER_SIZE] = {0};

pwd = current->fs->pwd;
path_get(&pwd);
cwd = d_path(&pwd, current_path, 100);
pr_info("path = %s\n", cwd);

輸入命令 sudo insmod khttpd.ko 並用 Chrome 網頁瀏覽器測試後，實際的結果如下所示，沒有顯示絕對路徑

path = /

接著嘗試另一種方法，在 fs/d_path.c 發現函式 d_absolute_path ，想嘗試執行試試，但函式 d_absolute_path 沒有使用巨集 EXPORT_SYMBOL ，因此無法直接在核心模組進行呼叫。

換另一種方式：新增核心模組參數 WWWROOT ，在掛載模組時直接指定要開啟的路徑。參考 The Linux Kernel Module Programming Guide ，使用巨集 module_param_string 新增參數 WWWROOT

#define PATH_SIZE   100
static char WWWROOT[PATH_SIZE] = {0};
module_param_string(WWWROOT, WWWROOT, PATH_SIZE, 0);

為了讓 WWWROOT 可以傳遞到其他檔案，在結構 httpd_service 新增成員 dir_path ，主要用來傳遞資料到不同檔案

struct httpd_service {
    bool is_stopped;
+   char *dir_path;
    struct list_head head;
};
extern struct httpd_service daemon_list;

接著在函式 khttpd_init 新增以下程式碼，主要功能是用來判斷參數 WWWROOT 是否為空字串，如果是則使用預設的路徑，這裡採用 "/"

// check WWWROOT is a empty string or not
if (!*WWWROOT)
    WWWROOT[0] = '/';
daemon_list.dir_path = WWWROOT;

分別在掛載核心模組時輸入 sudo insmod khttpd.ko 及 sudo insmod khttpd.ko WWWROOT='"home/user/khttpd"' ，並得到以下結果 (節錄部份結果)

sudo insmod khttpd.ko

sudo insmod khttpd.ko WWWROOT='"home/user/khttpd"'

目前可以藉由參數 WWWROOT 輸入伺服器開啟的目錄

讀取檔案資料

想要讀取檔案的資料，必需先知道檔案的屬性，如檔案大小以及檔案類型，在 Linux kernel 裡，檔案的屬性由結構 inode 所管理，位於 include/linux/fs.h ，而這裡主要使用到成員 i_mode 及 i_size ，前者主要表示檔案的類型，後者儲存檔案的大小

/*
 * Keep mostly read-only and often accessed (especially for
 * the RCU path lookup and 'stat' data) fields at the beginning
 * of the 'struct inode'
 */
struct inode {
	umode_t			i_mode;
	...
	loff_t			i_size;
	...
}

相同的，檔案類型一樣位於 include/linux/fs.h ，可以看到不同類型的檔案有不同的數值

/* these are defined by POSIX and also present in glibc's dirent.h */
#define DT_UNKNOWN  0
#define DT_FIFO     1
#define DT_CHR      2
#define DT_DIR      4
#define DT_BLK      6
#define DT_REG      8
#define DT_LNK      10
#define DT_SOCK     12
#define DT_WHT      14

接著如何判斷檔案類型，參考 include/uapi/linux/stat.h 的資料，發現可以判斷檔案類型的巨集，這裡主要使用巨集 S_ISDIR 及 S_ISREG ，前者用來判斷是否為目錄，後者則是判斷是否為一般文件

#define S_ISLNK(m)	(((m) & S_IFMT) == S_IFLNK)
#define S_ISREG(m)	(((m) & S_IFMT) == S_IFREG)
#define S_ISDIR(m)	(((m) & S_IFMT) == S_IFDIR)
#define S_ISCHR(m)	(((m) & S_IFMT) == S_IFCHR)
#define S_ISBLK(m)	(((m) & S_IFMT) == S_IFBLK)
#define S_ISFIFO(m)	(((m) & S_IFMT) == S_IFIFO)
#define S_ISSOCK(m)	(((m) & S_IFMT) == S_IFSOCK)

接著開始修改程式，完整修改位於 Add the function of read file 及 Fix bug on reading file in deeper directory ，主要修改函式 handle_directory

static bool handle_directory(struct http_request *request)
{
    struct file *fp;
    char pwd[BUFFER_SIZE] = {0};

    ...

    catstr(pwd, daemon_list.dir_path, request->request_url);
    fp = filp_open(pwd, O_RDONLY, 0);

    if (IS_ERR(fp)) {
        send_http_header(request->socket, HTTP_STATUS_NOT_FOUND,
                         http_status_str(HTTP_STATUS_NOT_FOUND), "text/plain",
                         13, "Close");
        send_http_content(request->socket, "404 Not Found");
        return false;
    }

    if (S_ISDIR(fp->f_inode->i_mode)) {
        char buf[SEND_BUFFER_SIZE] = {0};
        snprintf(buf, SEND_BUFFER_SIZE, "HTTP/1.1 200 OK\r\n%s%s%s",
                 "Connection: Keep-Alive\r\n", "Content-Type: text/html\r\n",
                 "Keep-Alive: timeout=5, max=1000\r\n\r\n");
        http_server_send(request->socket, buf, strlen(buf));

        snprintf(buf, SEND_BUFFER_SIZE, "%s%s%s%s", "<html><head><style>\r\n",
                 "body{font-family: monospace; font-size: 15px;}\r\n",
                 "td {padding: 1.5px 6px;}\r\n",
                 "</style></head><body><table>\r\n");
        http_server_send(request->socket, buf, strlen(buf));

        iterate_dir(fp, &request->dir_context);

        snprintf(buf, SEND_BUFFER_SIZE, "</table></body></html>\r\n");
        http_server_send(request->socket, buf, strlen(buf));
        kernel_sock_shutdown(request->socket, SHUT_RDWR);

    } else if (S_ISREG(fp->f_inode->i_mode)) {
        char *read_data = kmalloc(fp->f_inode->i_size, GFP_KERNEL);
        int ret = read_file(fp, read_data);

        send_http_header(request->socket, HTTP_STATUS_OK,
                         http_status_str(HTTP_STATUS_OK), "text/plain", ret,
                         "Close");
        http_server_send(request->socket, read_data, ret);
        kfree(read_data);
    }
    filp_close(fp, NULL);
    return true;
}

修改後的函式 handle_directory 做了以下幾件事

1. 第 8 行使用函式 catstr ，將 WWWROOT 的路徑及 client 的要求接在一起，並且輸出到 pwd，再由函式 filp_open 打開檔案
2. 第 11 行表示如果開檔失敗，則回傳 NOT FOUND 訊息給 client
3. 第 19 行表示如果為目錄，則將整個目錄擁有的檔案名稱傳送給 client
4. 第 38 行表示如果為一般文件，則直接讀取檔案資料並且送給 client

接著稍微修改前面的實作，讓伺服器可以處理 ".." 的要求，完整修改參考 Consider request ".." to go back previous page ，以下節錄主要的修改

// callback for 'iterate_dir', trace entry.
static int tracedir(struct dir_context *dir_context,
                    const char *name,
                    int namelen,
                    loff_t offset,
                    u64 ino,
                    unsigned int d_type)
{
-   if (strcmp(name, ".") && strcmp(name, "..")) {
+   if (strcmp(name, ".")) {
        struct http_request *request =
            container_of(dir_context, struct http_request, dir_context);
        char buf[SEND_BUFFER_SIZE] = {0};
-       char *url =
-           !strcmp(request->request_url, "/") ? "" : request->request_url;

        SEND_HTTP_MSG(request->socket, buf,
                      "%lx\r\n<tr><td><a href=\"%s/%s\">%s</a></td></tr>\r\n",
-                     34 + strlen(url) + (namelen << 1), url, name, name);
+                     34 + strlen(request->request_url) + (namelen << 1),
+                     request->request_url, name, name);
    }
    return 0;
}

static int http_parser_callback_request_url(http_parser *parser,
                                            const char *p,
                                            size_t len)
{
    struct http_request *request = parser->data;
+   // if requst is "..", remove last character
+   if (p[len - 1] == '/')
+       len--;
    strncat(request->request_url, p, len);
    return 0;
}

函式 tracedir 主要只是移除多餘的程式碼，而函式 http_parser_callback_request_url 因進入到多層目錄後會回不去原本的目錄而有的改動。

考慮以下：假設現行目錄為 /ab/cd 並且送出 .. ，原來的時候會產生的結果為 /ab/ ，接著再送出一次 .. 會產生的結果仍然為 /ab/ ，表示進到二層以上的目錄後會回不到更早的目錄。

為了解決這樣的問題才會有以上的更動，如果路徑的最後一個字元為 '/' ，只要將其移除即可，用一樣的例結果會變成 /ab/cd

$\to$ /ab

$\to$ (空字串)

使用 Chunked transfer encoding 送出目錄資料

之前的實作由於每次傳送目錄資料時，不知總資料大小，因此都是送完資料後直接關閉連線，而在 HTTP 1.1 中提供了 Chunked encoding 的方法，可以將資料分成一個個的 chunk 並且分批發送，如此一來可以避免要在 HTTP header 中傳送 Content-Length: xx

參考 Transfer-Encoding: Chunked encoding 並由以下的範例可以得到幾個資訊

1. 每次傳送資料前都要先送出資料的長度
2. 資料的長度是 16 進位表示
3. 資料長度和資料由 \r\n 隔開
4. 要中斷資料傳送只要送出長度為 0 的資料即可

HTTP/1.1 200 OK
Content-Type: text/plain
Transfer-Encoding: chunked

7\r\n
Mozilla\r\n
9\r\n
Developer\r\n
7\r\n
Network\r\n
0\r\n
\r\n

在正式修改程式之前，之前撰寫的函式 send_http_header 和 send_http_content 實在是太冗長，因此將二者重新修改並且寫的更有彈性，新增巨集函式 SEND_HTTP_MSG 如下

#define SEND_HTTP_MSG(socket, buf, format, ...)           \
    snprintf(buf, SEND_BUFFER_SIZE, format, __VA_ARGS__); \
    http_server_send(socket, buf, strlen(buf))

如此一來，輸入的資料可以讓使用者任意送出，程式碼也變得更簡潔

以下主要列出使用 chunked encoding 的部份，分別是函式 handle_directory 及 tracedir

// callback for 'iterate_dir', trace entry.
static int tracedir(struct dir_context *dir_context,
                    const char *name,
                    int namelen,
                    loff_t offset,
                    u64 ino,
                    unsigned int d_type)
{
    if (strcmp(name, ".") && strcmp(name, "..")) {
        struct http_request *request =
            container_of(dir_context, struct http_request, dir_context);
        char buf[SEND_BUFFER_SIZE] = {0};
        char *url =
            !strcmp(request->request_url, "/") ? "" : request->request_url;

        SEND_HTTP_MSG(request->socket, buf,
                      "%lx\r\n<tr><td><a href=\"%s/%s\">%s</a></td></tr>\r\n",
                      34 + strlen(url) + (namelen << 1), url, name, name);
    }
    return 0;
}

static bool handle_directory(struct http_request *request)
{
	...
	if (S_ISDIR(fp->f_inode->i_mode)) {
		SEND_HTTP_MSG(request->socket, buf, "%s%s%s", "HTTP/1.1 200 OK\r\n",
		              "Content-Type: text/html\r\n",
		              "Transfer-Encoding: chunked\r\n\r\n");
		SEND_HTTP_MSG(
		    request->socket, buf, "7B\r\n%s%s%s%s", "<html><head><style>\r\n",
		    "body{font-family: monospace; font-size: 15px;}\r\n",
		    "td {padding: 1.5px 6px;}\r\n", "</style></head><body><table>\r\n");

		iterate_dir(fp, &request->dir_context);

		SEND_HTTP_MSG(request->socket, buf, "%s",
		              "16\r\n</table></body></html>\r\n");
		SEND_HTTP_MSG(request->socket, buf, "%s", "0\r\n\r\n");
	}
	...
}

主要修改的部份在於發送 HTTP header 時，需要新增 Transfer-Encoding: chunked ，另外每次傳送資料時後要先送出該資料的長度，最後要記得送出長度為 0 的資料

經過這樣的修改後，目前的伺服器可以送出不固定大小的資料

最後展示執行結果

使用 MIME 處理不同類型的檔案

參考 MIME 類別 可以初步了解 MIME。MIME 是種表示檔案或各式位元組的標準，並定義於 RFC 6838 裡，如果要使用 MIME 的功能，則要在伺服器回應的 HTTP header 的項目 Content-Type 提供正確的類型

至於要回應什麼要的類型，可參考 Common MIME types ，裡頭提供了不同的副檔名應該要回應的型態

如此一來可以開始修改程式碼，完整修改參考 Add MIME to deal with different kind of files ，新增檔案 mime_type.h 裡面儲存常見的 MIME 類型

新增函式 get_mime_str ，功能為根據要求的檔案找到對應的回應訊息

// return mime type string
const char *get_mime_str(char *request_url)
{
    char *request_type = strchr(request_url, '.');
    int index = 0;
    if (!request_type)
        return "text/plain";

    while (mime_types[index].type) {
        if (!strcmp(mime_types[index].type, request_type))
            return mime_types[index].string;
        index++;
    }
    return "text/plain";
}

接著修改函式 handle_directory 裡處理一般檔案的部份，主要就是利用函式 get_mime_str 取得對應的回應訊息

static bool handle_directory(struct http_request *request)
{
	...
	else if (S_ISREG(fp->f_inode->i_mode)) {
		char *read_data = kmalloc(fp->f_inode->i_size, GFP_KERNEL);
		int ret = read_file(fp, read_data);

		SEND_HTTP_MSG(
			request->socket, buf, "%s%s%s%s%d%s", "HTTP/1.1 200 OK\r\n",
+			"Content-Type: ", get_mime_str(request->request_url),
			"\r\nContent-Length: ", ret, "\r\nConnection: Close\r\n\r\n");
		http_server_send(request->socket, read_data, ret);
		kfree(read_data);
	}
	...
}

卸載模組時產生錯誤

在目前的實作發現了一個問題，只要有對伺服器做請求後，在卸載模組時會產生以下的錯誤訊息

[ 3721.905941] ------------[ cut here ]------------
[ 3721.905958] kernel BUG at fs/inode.c:1676!
[ 3721.905974] invalid opcode: 0000 [#6] SMP PTI
[ 3721.905987] CPU: 0 PID: 10434 Comm: khttpd Tainted: G      D W  OE     5.13.0-41-generic #46~20.04.1-Ubuntu
[ 3721.905999] Hardware name: Acer Aspire F5-573G/Captain_SK  , BIOS V1.18 10/21/2016
[ 3721.906005] RIP: 0010:iput+0x1ac/0x200
[ 3721.906022] Code: 00 0f 1f 40 00 4c 89 e7 e8 01 fb ff ff
                     5b 41 5c 41 5d 5d c3 c3 85 d2 74 a4 49
                     83 bc 24 e0 00 00 00 00 0f 85 3a ff ff
                     ff eb 93 <0f> 0b 0f 0b e9 0e ff ff ff
                     a9 b7 08 00 00 75 17 41 8b 84 24 58 01
...

首先查了 fs/inode.c 的第 1676 行，參考 fs/inode.c 可以找到對應的函式 iput

void iput(struct inode *inode)
{
	if (!inode)
		return;
	BUG_ON(inode->i_state & I_CLEAR);
retry:
	if (atomic_dec_and_lock(&inode->i_count, &inode->i_lock)) {
		if (inode->i_nlink && (inode->i_state & I_DIRTY_TIME)) {
			atomic_inc(&inode->i_count);
			spin_unlock(&inode->i_lock);
			trace_writeback_lazytime_iput(inode);
			mark_inode_dirty_sync(inode);
			goto retry;
		}
		iput_final(inode);
	}
}

而程式錯誤就是發生在上述函式的第 5 行，從程式碼大致可以先猜這次的程式錯誤和檔案系統有關

最後發現，當我對伺服器送出請求後，伺服器會經過開啟檔案及讀取檔案的步驟，但是關閉檔案並沒有執行，程式會停留在函式 filp_close ，直到下一次的請求出現才會關閉，相關程式碼如下

static bool handle_directory(struct http_request *request, int keep_alive)
{
	...
	fp = filp_open(pwd, O_RDONLY, 0);
	...
	if (S_ISDIR(fp->f_inode->i_mode)) {
		...    
	} else if (S_ISREG(fp->f_inode->i_mode)) {
		...
		int ret = read_file(fp, read_data);
		...
	}
	filp_close(fp, NULL);
	return true;
}

因此當 client 從遠端關閉時，最後一次請求的檔案的 file descriptor 是沒有被關閉的，因此這時如果卸載模組就會產生上述的問題

為了解決這個問題，目前的想法是可以建立 timer 管理連線，讓伺服器可以主動關閉逾時的連線，詳細步驟在後面會有解釋

建立 timer 主動關閉連線

根據 高效 Web 伺服器開發 - 實作考量點 提到以下考量點

當 Web 伺服器和客戶端網頁瀏覽器保持著一個長期連線的狀況下，遇到客戶端網路離線，伺服器該如何處理？

$\to$ 通訊協定無法立刻知悉，所以僅能透過伺服器引入 timer 逾時事件來克服

目前的 kHTTPd 沒有使用 timer 來關閉閒置的連線，因此會導致部份資源被佔用。參考 sehttpd 裡 timer 的實作，主要使用 min heap 來做管理，相關資訊可以參考二元堆積。

為了方便解決這個問題，將問題分成以下幾個小問題並且逐一解決

1. 將 socket 設定為 non-blocking
2. 讀取目前的時間
3. 實作 prority queue 並且管理每個連線

將 socket 設定為 non-blocking

要將 socket 設定為 non-blocking 的原因在於，原本的實作中 socket 預設為 blocking ，因此執行緒會停滯在函式 kernel_accept ，但這樣的話沒有辦法去判斷是否已經有連線逾期，因此將 socket 設定為 non-blocking 可以避免執行緒停滯在函式 kernel_accept，完整修改參考 Set socket non-blocking and remove accept_err

參考 kernel_accept ，其中參數 flags 可以設定為 SOCK_NONBLOCK ，如下所示

int err = kernel_accept(param->listen_socket, &socket, SOCK_NONBLOCK);

如此一來 socket 就能被改成 non-blocking 模式。

讀取目前的時間

要讀取目前的時間，在 sehttpd 中使用系統呼叫 gettimeofday 實作，對應程式碼如下

static void time_update()
{
    struct timeval tv;
    int rc UNUSED = gettimeofday(&tv, NULL);
    assert(rc == 0 && "time_update: gettimeofday error");
    current_msec = tv.tv_sec * 1000 + tv.tv_usec / 1000;
}

而在 khttpd 裡無法使用系統呼叫，參考 include/linux/time64.h 裡的結構 timespec64 ，其定義如下，其中成員 tv_sec 表示秒而成員 tv_nsec 表示奈秒

struct timespec64 {
	time64_t	tv_sec;			/* seconds */
	long		tv_nsec;		/* nanoseconds */
};

接著參考 include/linux/timekeeping.h 裡的函式 ktime_get_ts64 可以將目前的時間轉換成上述提到的結構 timespec64 的形式，以下擷取部份程式碼

/**
 * ktime_get_ts64 - get the monotonic clock in timespec64 format
 * @ts:		pointer to timespec variable
 *
 * The function calculates the monotonic clock from the realtime
 * clock and the wall_to_monotonic offset and stores the result
 * in normalized timespec64 format in the variable pointed to by @ts.
 */
void ktime_get_ts64(struct timespec64 *ts)
{
    struct timekeeper *tk = &tk_core.timekeeper;
    struct timespec64 tomono;
    unsigned int seq;
    u64 nsec;
    ...
}

有了以上的背景知識，可開始在 kHTTPd 上進行實作，建立函式 time_update 如下:

static void time_update()
{
    struct timespec64 tv;
    ktime_get_ts64(&tv);
    current_msec = tv.tv_sec * 1000 + tv.tv_nsec / 1000000;
}

如此一來就可以得到當下的時間，單位為毫秒。

實作 prority queue 並且管理每個連線

先定義問題：首先只會有一個 consumer 移除資料，亦即執行在背景的執行緒，而 producer 則是由多個處理連線的執行緒組成，因此歸納為 MPSC 的問題。

以下是簡略但存在缺陷的 lock-free 實作。定義 timer 和 priority queue 的結構體:

typedef int (*timer_callback)(struct socket *, enum sock_shutdown_cmd);
typedef struct {
    size_t key;
    size_t pos; // the position of timer in queue
    timer_callback callback;
    struct socket *socket;
} timer_node_t;

typedef int (*prio_queue_comparator)(void *pi, void *pj);
typedef struct {
    void **priv;
    atomic_t nalloc;  // number of items in queue
    atomic_t size;
    prio_queue_comparator comp;
} prio_queue_t;

整個 priority queue 的流程如下所示

1. 建立 priority queue 並開始等待連線
2. 只要有新增連線，就使用函式 prio_queue_insert 新增新的 timer 並加到 priority queue
3. 使用函式 handle_expired_timers 偵測是否有 timer 逾期
4. 卸載模組時，使用函式 http_free_timer 釋放所有 timer 及 priority queue
5. 只要有連線再次送出請求，則需要更新其 key

插入 timer 到 priority queue

函式 prio_queue_insert 主要功能為插入 timer 到 priority queue 裡，如同前面所說，這次的實作可以解讀成 MPSC ，因此這裡需要解決多個 producer 要插入的問題

/* add a new item to the heap */
static bool prio_queue_insert(prio_queue_t *ptr, void *item)
{
    timer_node_t **slot;  // get the address we want to store item
    size_t old_nalloc, old_size;
    long long old;

restart:
    old_nalloc = atomic_read(&ptr->nalloc);
    old_size = atomic_read(&ptr->nalloc);

    // get the address want to store
    slot = (timer_node_t **) &ptr->priv[old_nalloc + 1];
    old = (long long) *slot;

    do {
        if (old_nalloc != atomic_read(&ptr->nalloc))
            goto restart;
    } while (!prio_queue_cmpxchg(slot, &old, (long long) item));

    atomic_inc(&ptr->nalloc);
    return true;
}

而這裡的解決方式是利用判斷新舊成員數決定資料是否被別人寫入，也就是上述程式碼第 17 行，接著使用函式 prio_queue_cmpxchg 執行 CAS 操作，程式碼如下。

參照 Semantics and Behavior of Atomic and Bitmask Operations，若要用 Linux kernel 提供的 atomic_cmpxchg 實作 CAS ，應當留意到 Linux 核心提供的 atomic API 只能對變數本身的值做讀寫，不能對變數指到的資料讀寫，因此改成以下 inline assembly 的方式實作，主要更動就是從原本的 128 位元改成了 64 位元:

static inline bool prio_queue_cmpxchg(timer_node_t **var,
                                      long long *old,
                                      long long neu)
{
    bool ret;
    union u64 {
        struct {
            int low, high;
        } s;
        long long ui;
    } cmp = {.ui = *old}, with = {.ui = neu};

    /**
     * 1. cmp.s.hi:cmp.s.lo compare with *var
     * 2. if equall, set ZF and copy with.s.hi:with.s.lo to *var
     * 3. if not equall， clear ZF and copy *var to cmp.s.hi:cmp.s.lo
     */
    __asm__ __volatile__("lock cmpxchg8b %1\n\tsetz %0"
                         : "=q"(ret), "+m"(*var), "+d"(cmp.s.high),
                           "+a"(cmp.s.low)
                         : "c"(with.s.high), "b"(with.s.low)
                         : "cc", "memory");
    if (!ret)
        *old = cmp.ui;
    return ret;
}

另外， min heap 在插入新的資料後都要經過 swim 的方式移動到正確的位置，而在這次的案例，資料 key 紀錄逾期的時間，且每個 timer 插入的時間一定都會比之前的 timer 大，因此不會出現後面的資料比前面的資料小的情況，也就可以省略 swim 的動作，如此一來，這樣就和 ring buffer 的操作相同。

從 priority queue 移除 timer

函式 prio_queue_delmin 主要功能為從 priority queue 移除最小的 timer，因是 MPSC ，這裡避免 root 和最後一個成員交換時會有 producer 加入新資料 (程式碼第 14 行)，也是依據 heap 的新舊成員數來判斷是否有受到其他 producer 的影響

接著就是更新新的成員數並且執行 sink 的動作，最後關閉該 timer 的連線以及釋放其記憶體

/* remove the item with minimum key value from the heap */
static bool prio_queue_delmin(prio_queue_t *ptr)
{
    size_t nalloc;
    timer_node_t *node;

    do {
        if (prio_queue_is_empty(ptr))
            return true;

        nalloc = atomic_read(&ptr->nalloc);
        prio_queue_swap(ptr, 1, nalloc);

        if (nalloc == atomic_read(&ptr->nalloc)) {
            node = ptr->priv[nalloc--];
            break;
        }
        // change again
        prio_queue_swap(ptr, 1, nalloc);
    } while (1);

    atomic_set(&ptr->nalloc, nalloc);
    prio_queue_sink(ptr, 1);

    if (node->callback)
        node->callback(node->socket, SHUT_RDWR);

    kfree(node);
    return true;
}

實測

使用命令 ./htstress localhost:8081 -n 20000 進行測試，以下節錄部份的程式運行過程，可觀察到多個執行緒執行的狀況符合預期。

remove node 00000000c8603c50 key 10712635 nalloc 17198
remove node 000000006d6d7424 key 10712635 nalloc 17197
remove node 00000000a7621098 key 10712635 nalloc 17196
add node 0000000047dc72c9 key 10720635 nalloc 17197
add node 00000000a80a0d9c key 10720635 nalloc 17198
add node 000000009c494072 key 10720635 nalloc 17199
add node 00000000972001db key 10720635 nalloc 17200
add node 00000000835c9d0f key 10720635 nalloc 17201
remove node 000000005c21a0ec key 10712636 nalloc 17200
remove node 000000004ed780e8 key 10712636 nalloc 17199
remove node 000000000be7bfbb key 10712636 nalloc 17198
remove node 00000000ad509208 key 10712636 nalloc 17197
remove node 000000001bf92ea9 key 10712636 nalloc 17196
remove node 00000000697caab7 key 10712636 nalloc 17195
add node 000000002fd797a7 key 10720636 nalloc 17196
add node 000000009502d850 key 10720636 nalloc 17197
add node 000000007d125979 key 10720636 nalloc 17198
add node 000000000b6728c7 key 10720636 nalloc 17199
add node 00000000a524b323 key 10720636 nalloc 17200
add node 00000000b827ea2c key 10720636 nalloc 17201
remove node 000000008791c3cb key 10712637 nalloc 17200

接著展示伺服器會更新每個連線的逾期時間，目前每個連線約等待 8 秒，可以看到第一次測試約等了 8 秒後自動關閉連線，且第二次的連線在送出請求後會再等待新的 8 秒

目前的成果，使用命令 ./htstress localhost:8081 -n 20000

requests:      20000
good requests: 20000 [100%]
bad requests:  0 [0%]
socket errors: 0 [0%]
seconds:       9.244
requests/sec:  2163.532

使用 Ftrace 觀察 kHTTPd

參考 Ftrace 及《Demystifying the Linux CPU Scheduler》第六章。

ftrace 是一個內建於 Linux 核心的動態追蹤工具，可用來追蹤函式、追蹤事件、計算 context switch 時間及中斷被關閉的時間點等等。

先確認目前的系統是否有 ftrace，輸入以下命令

cat /boot/config-`uname -r` | grep CONFIG_HAVE_FUNCTION_TRACER

期望輸出如下

CONFIG_HAVE_FUNCTION_TRACER=y

接著要怎麼使用 ftrace？可藉由寫入路徑 /sys/kernel/debug/tracing/ 內的檔案來設定 ftrace ，以下提供部份檔案，使用命令 sudo ls /sys/kernel/debug/tracing 查看

available_events            max_graph_depth   stack_max_size
available_filter_functions  options           stack_trace
available_tracers           per_cpu           stack_trace_filter
buffer_percent              printk_formats    synthetic_events
...

至於這些檔案負責什麼功能，以下列出實驗有使用到的設定，剩下可以從 Ftrace 找到說明

• current_tracer: 設定或顯示目前使用的 tracers ，像是 function 、 function_graph 等等
• tracing_on: 設定或顯示使用的 tracer 是否開啟寫入資料到 ring buffer 的功能，如果為 0 表示關閉，而 1 則表示開啟
• trace: 儲存 tracer 所輸出的資料，換言之，就是紀錄整個追蹤所輸出的訊息
• available_filter_functions: 列出 kernel 裡所有可以被追蹤的 kernel 函式
• set_ftrace_filter: 指定要追蹤的函式，該函式一定要出現在 available_filter_functions 裡
• set_graph_function: 指定要顯示呼叫關係的函數，顯示的資訊類似於程式碼的模樣，只是會將所有呼叫的函式都展開
• max_graph_depth: function graph tracer 追蹤函式的最大深度

有了以上的知識，可開始追蹤 kHTTPd，這裡嘗試追蹤其中每個連線都會執行的函式 http_server_worker ，第一步是要掛載核心模組，且透過檔案 available_filter_functions 確定是否可以追蹤 khttpd 的函式，輸入命令 cat available_filter_functions | grep khttpd 查看，可見 kHTTPd 裡可被追蹤的所有函式:

parse_url_char [khttpd]
http_message_needs_eof.part.0 [khttpd]
http_message_needs_eof [khttpd]
http_should_keep_alive [khttpd]
http_parser_execute [khttpd]
http_method_str [khttpd]
http_status_str [khttpd]
http_parser_init [khttpd]
http_parser_settings_init [khttpd]
http_errno_name [khttpd]
http_errno_description [khttpd]
http_parser_url_init [khttpd]
...

接著建立 shell script 來追蹤函式 http_server_worker ，如下所示

#!/bin/bash
TRACE_DIR=/sys/kernel/debug/tracing

# clear
echo 0 > $TRACE_DIR/tracing_on
echo > $TRACE_DIR/set_graph_function
echo > $TRACE_DIR/set_ftrace_filter
echo nop > $TRACE_DIR/current_tracer

# setting
echo function_graph > $TRACE_DIR/current_tracer
echo 3 > $TRACE_DIR/max_graph_depth
echo http_server_worker > $TRACE_DIR/set_graph_function

# execute
echo 1 > $TRACE_DIR/tracing_on
./htstress localhost:8081 -n 2000
echo 0 > $TRACE_DIR/tracing_on

主要邏輯就是先清空 ftrace 的設定，接著設定函式 http_server_worker 為要追蹤的函式，最後在測試時開啟 tracer

執行 shell script 後，從 ftrace 的檔案 trace 可以看到追蹤的輸出，以下節錄部份輸出

# tracer: function_graph
#
# CPU  DURATION                  FUNCTION CALLS
# |     |   |                     |   |   |   |
 0)               |  http_server_worker [khttpd]() {
 0)               |    kernel_sigaction() {
 0)   0.296 us    |      _raw_spin_lock_irq();
 0)   0.937 us    |    }
 0)   0.329 us    |    }
 0)               |    kmem_cache_alloc_trace() {
 0)   0.165 us    |      __cond_resched();
 0)   0.114 us    |      should_failslab();
 0)   0.913 us    |    }
 0)   0.111 us    |    http_parser_init [khttpd]();
 0)               |    http_add_timer [khttpd]() {
 0)   0.433 us    |      kmem_cache_alloc_trace();
 0)   0.174 us    |      ktime_get_ts64();
 0)   1.052 us    |    }
 0)               |    http_server_recv.constprop.0 [khttpd]() {
 0)   3.134 us    |      kernel_recvmsg();
 0)   3.367 us    |    }
 0)               |    kernel_sock_shutdown() {
 0) + 40.992 us   |      inet_shutdown();
 0) + 41.407 us   |    }
 0)   0.433 us    |    kfree();
 0) + 50.869 us   |  }

由上面的結果可以看到整個 http_server_worker 函式所花的時間以及內部函式所花的時間，有這樣的實驗可以開始分析造成 khttpd 效率低落的原因

找出 kHTTPd 的效能瓶頸

將可以追蹤函式的深度增加後，再次追蹤函式 http_server_worker 一次，以下為單次連線的追蹤結果

- echo 3 > $TRACE_DIR/max_graph_depth
+ echo 5 > $TRACE_DIR/max_graph_depth

# tracer: function_graph
#
# CPU  DURATION                  FUNCTION CALLS
# |     |   |                     |   |   |   |
 3)               |  http_server_worker [khttpd]() {
 3)               |    kernel_sigaction() {
 3)   0.082 us    |      _raw_spin_lock_irq();
 3)   0.238 us    |    }
 3)               |    kernel_sigaction() {
 3)   0.079 us    |      _raw_spin_lock_irq();
 3)   0.222 us    |    }
 3)               |    kmem_cache_alloc_trace() {
 3)               |      __cond_resched() {
 3)   0.070 us    |        rcu_all_qs();
 3)   0.209 us    |      }
 3)   0.068 us    |      should_failslab();
 3)   0.567 us    |    }
 3)   0.076 us    |    http_parser_init [khttpd]();
 3)               |    http_add_timer [khttpd]() {
 3)               |      kmem_cache_alloc_trace() {
 3)               |        __cond_resched() {
 3)   0.070 us    |          rcu_all_qs();
 3)   0.201 us    |        }
 3)   0.070 us    |        should_failslab();
 3)   0.490 us    |      }
 3)   0.084 us    |      ktime_get_ts64();
 3)   0.792 us    |    }
 3)               |    http_server_recv.constprop.0 [khttpd]() {
 3)               |      kernel_recvmsg() {
 3)               |        sock_recvmsg() {
 3)   0.260 us    |          security_socket_recvmsg();
 3) + 13.319 us   |          inet_recvmsg();
 3) + 13.813 us   |        }
 3) + 13.957 us   |      }
 3) + 14.118 us   |    }
 3)               |    http_parser_execute [khttpd]() {
 3)   0.085 us    |      http_parser_callback_message_begin [khttpd]();
 3)   0.350 us    |      parse_url_char [khttpd]();
 3)   0.114 us    |      http_parser_callback_request_url [khttpd]();
 3)   0.077 us    |      http_parser_callback_header_field [khttpd]();
 3)   0.068 us    |      http_parser_callback_header_value [khttpd]();
 3)   0.070 us    |      http_parser_callback_headers_complete [khttpd]();
 3)   0.073 us    |      http_should_keep_alive [khttpd]();
 3)               |      http_parser_callback_message_complete [khttpd]() {
 3)   0.069 us    |        http_should_keep_alive [khttpd]();
 3)               |        handle_directory [khttpd]() {
 3)   6.950 us    |          filp_open();
 3) + 16.874 us   |          http_server_send [khttpd]();
 3) + 12.320 us   |          http_server_send [khttpd]();
 3) ! 478.209 us  |          iterate_dir();
 3)   8.984 us    |          http_server_send [khttpd]();
 3)   9.507 us    |          http_server_send [khttpd]();
 3)   1.422 us    |          filp_close();
 3) ! 536.623 us  |        }
 3) ! 536.907 us  |      }
 3) ! 540.598 us  |    }
 3)   0.078 us    |    http_should_keep_alive [khttpd]();
 3)               |    kernel_sock_shutdown() {
 3)               |      inet_shutdown() {
 3)               |        lock_sock_nested() {
 3)   0.106 us    |          __cond_resched();
 3)   0.074 us    |          _raw_spin_lock_bh();
 3)   0.069 us    |          __local_bh_enable_ip();
 3)   0.524 us    |        }
 3)               |        tcp_shutdown() {
 3)   0.127 us    |          tcp_set_state();
 3)   5.040 us    |          tcp_send_fin();
 3)   5.393 us    |        }
 3)               |        sock_def_wakeup() {
 3)   0.077 us    |          rcu_read_unlock_strict();
 3)   0.230 us    |        }
 3)               |        release_sock() {
 3)   0.087 us    |          _raw_spin_lock_bh();
 3)   2.299 us    |          __release_sock();
 3)   0.078 us    |          tcp_release_cb();
 3)   0.106 us    |          _raw_spin_unlock_bh();
 3)   2.940 us    |        }
 3)   9.504 us    |      }
 3)   9.667 us    |    }
 3)   0.132 us    |    kfree();
 3) ! 567.482 us  |  }

由上面的結果可見，影響 kHTTPd 效能最大的部份在於走訪目錄的函式 iterate_dir ，其次為用來接受和送出資料的函式 kernel_recvmsg 及 http_server_send。