libuv 中文編程指南(三)文件系統

文件系統

簡單的文件讀寫是通過 uv_fs_* 函數族和與之相關的 uv_fs_t 結構體完成的.

libuv 提供的文件操作和 socket operations 並不相同. 套接字操作使用了操作系統本身提供了非阻塞操作, 而文件操作內部使用了阻塞函數, 但是 libuv 是在線程池中調用這些函數, 並在應用程序需要交互時通知在事件循環中注冊的監視器.

所有的文件操作函數都有兩種形式 - 同步 synchronous 和 asynchronous.

同步 synchronous 形式如果沒有指定回調函數則會被自動調用( 阻塞的 ), 函數的返回值和 Unix 系統的函數調用返回值相同(調用成功通常返回 0, 若出現錯誤則返回 -1).

而異步 asynchronous 形式則會在傳入回調函數時被調用, 並且返回 0.

讀寫文件

文件描述符可以采用如下方式獲得:

int uv_fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path, int flags, int mode, uv_fs_cb cb)

 

參數 flags 與 mode 和標准的 Unix flags 相同. libuv 會小心地處理 Windows 環境下的相關標志位(flags)的轉換, 所以編寫跨平台程序時你不用擔心不同平台上文件打開的標志位不同。

關閉文件描述符可以使用:

int uv_fs_close(uv_loop_t* loop, uv_fs_t* req, uv_file file, uv_fs_cb cb)

與文件系統相關的操作的回調函數具有如下簽名:

void callback(uv_fs_t* req);

讓我們來看看 cat 命令的一個簡單實現吧: 我們首先注冊一個在文件打開時的回調函數 (顧名思義, 該函數將在文件打開時被調用).

void on_open(uv_fs_t *req) {
    if (req->result != -1) {
        uv_fs_read(uv_default_loop(), &read_req, req->result,
                   buffer, sizeof(buffer), -1, on_read);
    }
    else {
        fprintf(stderr, "error opening file: %d\n", req->errorno);
    }
    uv_fs_req_cleanup(req);
}

uv_fs_t 的 result 字段在執行 us_fs_open 時代表一個文件描述符, 如果文件成功被打開, 我們開始讀取文件.

必須調用 uv_fs_req_cleanup() 來釋放 libuv 內部使用的內存空間.

void on_read(uv_fs_t *req) {
    uv_fs_req_cleanup(req);
    if (req->result < 0) {
        fprintf(stderr, "Read error: %s\n", uv_strerror(uv_last_error(uv_default_loop())));
    }
    else if (req->result == 0) {
        uv_fs_t close_req;
        // synchronous
        uv_fs_close(uv_default_loop(), &close_req, open_req.result, NULL);
    }
    else {
        uv_fs_write(uv_default_loop(), &write_req, 1, buffer, req->result, -1, on_write);
    }
}

在調用 read 時, 你應該傳遞一個初始化的緩沖區, 在 read 回調函數被觸發(調用之前), 該緩沖區將會被填滿數據.

在 read 的回調函數中 result 如果是 0, 則讀取文件時遇到了文件尾(EOF), -1 則代表出現了錯誤, 而正整數則是表示成功讀取的字節數.

此處給你展示了編寫異步程序的通用模式, uv_fs_close() 是異步調用的. 通常如果任務是一次性的, 或者只在程序啟動和關閉時被執行的話都可以采用同步方式執行, 因為我們期望提高 I/O 效率, 采用異步編程時程序也可以做一些基本的任務並處理多路 I/O.. 對於單個任務而言性能差異可以忽略, 但是代碼卻能大大簡化.

我們可以總結出真正的系統調用返回值一般是存放在 uv_fs_t.result.

寫入文件與上述過程類似, 使用 uv_fs_write 即可. write 的回調函數在寫入完成時被調用.. 在我們的程序中回調函數只是只是簡單地發起了下一次讀操作, 因此, 讀寫操作會通過回調函數連續進行下去.

void on_write(uv_fs_t *req) {
    uv_fs_req_cleanup(req);
    if (req->result < 0) {
        fprintf(stderr, "Write error: %s\n", uv_strerror(uv_last_error(uv_default_loop())));
    }
    else {
        uv_fs_read(uv_default_loop(), &read_req, open_req.result, buffer, sizeof(buffer), -1, on_read);
    }
}

錯誤值通常保存在 errno 並可以通過 uv_fs_t.errorno 獲取, 但是被轉換成了標准的 UV_* 錯誤碼. 目前還沒有方法直接從 errorno 解析得到錯誤消息的字符串表示.

由於文件系統和磁盤通常為了提高性能吞吐率而配置了緩沖區, libuv 中一次 ‘成功’ 的寫操作可能不會被立刻提交到磁盤上, 你可以通過 uv_fs_fsync 來保證一致性.

我們再來看看 main 函數中設置的多米諾骨牌吧(原作者意指在 main 中設置回調函數后會觸發整個程序開始執行):

int main(int argc, char **argv) {
    uv_fs_open(uv_default_loop(), &open_req, argv[1], O_RDONLY, 0, on_open);
    uv_run(uv_default_loop(), UV_RUN_DEFAULT);
    return 0;
}

 

文件系統相關操作(Filesystem operations)

所有的標准文件系統操作, 例如 unlink, rmdir, stat 都支持異步操作, 並且各個函數的參數非常直觀. 他們和 read/write/open 的調用模式一致, 返回值都存放在 uv_fs_t.result 域. 完整的列表如下:

UV_EXTERN int uv_fs_close(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path,
    int flags, int mode, uv_fs_cb cb);

UV_EXTERN int uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    void* buf, size_t length, int64_t offset, uv_fs_cb cb);

UV_EXTERN int uv_fs_unlink(uv_loop_t* loop, uv_fs_t* req, const char* path,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_write(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    void* buf, size_t length, int64_t offset, uv_fs_cb cb);

UV_EXTERN int uv_fs_mkdir(uv_loop_t* loop, uv_fs_t* req, const char* path,
    int mode, uv_fs_cb cb);

UV_EXTERN int uv_fs_rmdir(uv_loop_t* loop, uv_fs_t* req, const char* path,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_readdir(uv_loop_t* loop, uv_fs_t* req,
    const char* path, int flags, uv_fs_cb cb);

UV_EXTERN int uv_fs_stat(uv_loop_t* loop, uv_fs_t* req, const char* path,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_fstat(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_rename(uv_loop_t* loop, uv_fs_t* req, const char* path,
    const char* new_path, uv_fs_cb cb);

UV_EXTERN int uv_fs_fsync(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_fdatasync(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_ftruncate(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    int64_t offset, uv_fs_cb cb);

UV_EXTERN int uv_fs_sendfile(uv_loop_t* loop, uv_fs_t* req, uv_file out_fd,
    uv_file in_fd, int64_t in_offset, size_t length, uv_fs_cb cb);

UV_EXTERN int uv_fs_chmod(uv_loop_t* loop, uv_fs_t* req, const char* path,
    int mode, uv_fs_cb cb);

UV_EXTERN int uv_fs_utime(uv_loop_t* loop, uv_fs_t* req, const char* path,
    double atime, double mtime, uv_fs_cb cb);

UV_EXTERN int uv_fs_futime(uv_loop_t* loop, uv_fs_t* req, uv_file file,
    double atime, double mtime, uv_fs_cb cb);

UV_EXTERN int uv_fs_lstat(uv_loop_t* loop, uv_fs_t* req, const char* path,
    uv_fs_cb cb);

UV_EXTERN int uv_fs_link(uv_loop_t* loop, uv_fs_t* req, const char* path,
    const char* new_path, uv_fs_cb cb);

 

回調函數中應該調用 uv_fs_req_cleanup() 函數來釋放 uv_fs_t 參數占用的內存.

緩沖區與流(Buffers and Streams)

libuv 中基本的 I/O 工具是流(uv_stream_t). TCP 套接字, UDP 套接字, 文件, 管道, 和進程間通信都可以作為 流 的子類.

流 (Streams) 通過每個子類特定的函數來初始化, 然后可以通過如下函數進行操作:

int uv_read_start(uv_stream_t*, uv_alloc_cb alloc_cb, uv_read_cb read_cb);
int uv_read_stop(uv_stream_t*);
int uv_write(uv_write_t* req, uv_stream_t* handle,
            uv_buf_t bufs[], int bufcnt, uv_write_cb cb);

 

基於流的函數比上面介紹的文件系統相關的函數更容易使用, libuv 在調用 uv_read_start 后會自動從流中讀取數據, 直到調用了 uv_read_stop.

用於保存數據的單元被抽象成了 buffer 結構 – uv_buf_t. 它其實只保存了指向真實數據的指針(uv_buf_t.base) 以及真實數據的長度 (uv_buf_t.len). uv_buf_t 本身是輕量級的, 通常作為值被傳遞給函數, 真正需要進行內存管理的是 buffer 結構中的指針所指向的真實數據, 通常由應用程序申請分配並釋放.

為了示范流的用法, 我們借助了(管道) uv_pipe_t , 這使得我們把本地文件變成了流[#]_. 下面是利用 libuv 實現的一個簡單的 tee . 將所有的操作變成了異步方式后, 事件 I/O 的強大能力便展現出來. 兩個寫操作並不會阻塞對方, 但是我們必須小心地拷貝數據至緩沖區, 並確保在寫入數據之前緩沖區不被釋放.

該程序按照如下方式執行:

./uvtee <output_file>

我們在指定的文件上打開了一個管道, libuv 的文件管道默認是雙向打開的.

int main(int argc, char **argv) {
    loop = uv_default_loop();

    uv_pipe_init(loop, &stdin_pipe, 0);
    uv_pipe_open(&stdin_pipe, 0);

    uv_pipe_init(loop, &stdout_pipe, 0);
    uv_pipe_open(&stdout_pipe, 1);
    
    uv_fs_t file_req;
    int fd = uv_fs_open(loop, &file_req, argv[1], O_CREAT | O_RDWR, 0644, NULL);
    uv_pipe_init(loop, &file_pipe, 0);
    uv_pipe_open(&file_pipe, fd);

    uv_read_start((uv_stream_t*)&stdin_pipe, alloc_buffer, read_stdin);

    uv_run(loop, UV_RUN_DEFAULT);
    return 0;
}

若是 IPC 或命名管道, uv_pipe_init() 的第三個參數應該設置為 1, 我們會在 進程 一節對此作出詳細解釋. 調用 uv_pipe_open() 將文件描述符和文件關聯在了一起.

我們開始監控標准輸入 stdin. 回調函數 alloc_buffer 為程序開辟了一個新的緩沖區來容納新到來的數據. read_stdin 也會被調用, 並且 uv_buf_t 作為調用參數.

uv_buf_t alloc_buffer(uv_handle_t *handle, size_t suggested_size) {
    return uv_buf_init((char*) malloc(suggested_size), suggested_size);
}

void read_stdin(uv_stream_t *stream, ssize_t nread, uv_buf_t buf) {
    if (nread == -1) {
        if (uv_last_error(loop).code == UV_EOF) {
            uv_close((uv_handle_t*)&stdin_pipe, NULL);
            uv_close((uv_handle_t*)&stdout_pipe, NULL);
            uv_close((uv_handle_t*)&file_pipe, NULL);
        }
    }
    else {
        if (nread > 0) {
            write_data((uv_stream_t*)&stdout_pipe, nread, buf, on_stdout_write);
            write_data((uv_stream_t*)&file_pipe, nread, buf, on_file_write);
        }
    }
    if (buf.base)
        free(buf.base);
}

 

 

此處使用標准的 malloc 已經可以足夠, 但是你也可以指定其他的內存分配策略. 例如, node.js 使用自己特定的 slab 分配器.

在任何情況下出錯, read 回調函數 nread 參數都為 -1. 出錯原因可能是 EOF(遇到文件尾), 在此種情況下我們使用 ‘’uv_close()’’ 函數關閉所有的流, uv_close() 會根據所傳遞進來句柄的內部類型來自動處理. 如果沒有出現錯誤, nread 是一個非負數, 意味着我們可以向輸出流中寫入 nread 字節的數據. 最后記住一點, 緩沖區 buffer 的分配和釋放是由應用程序負責的, 所以記得釋放不再使用的內存空間.

typedef struct {
    uv_write_t req;
    uv_buf_t buf;
} write_req_t;

void free_write_req(uv_write_t *req) {
    write_req_t *wr = (write_req_t*) req;
    free(wr->buf.base);
    free(wr);
}

void on_stdout_write(uv_write_t *req, int status) {
    free_write_req(req);
}

void on_file_write(uv_write_t *req, int status) {
    free_write_req(req);
}

void write_data(uv_stream_t *dest, size_t size, uv_buf_t buf, uv_write_cb callback) {
    write_req_t *req = (write_req_t*) malloc(sizeof(write_req_t));
    req->buf = uv_buf_init((char*) malloc(size), size);
    memcpy(req->buf.base, buf.base, size);
    uv_write((uv_write_t*) req, (uv_stream_t*)dest, &req->buf, 1, callback);
}

write_data() 將讀取的數據拷貝一份至緩沖區 req->buf.base, 同樣地, 當 write 完成后回調函數被調用時, 該緩沖區也並不會被傳遞到回調函數中, 所以, 為了繞過這一缺點, 我們將寫請求和緩沖區封裝在 write_req_t 結構體中, 然后在回調函數中解封該結構體來獲取相關參數.

文件變更事件(File change events)

現代操作系統都提供了 API 用來在單獨的文件或文件夾上設置監視器, 當文件被修改時應用程序會得到通知, libuv 也封裝了常用的文件變更通知程序庫 [1]. 這是 libuv 中最不一致的部分了, 文件變更通知系統本身在不同的系統中實現起來差別非常大, 因此讓所有的事情在每個平台上都完美地工作將變得異常困難, 為了給出一個示例，我寫了一個簡單的工具, 該函數按照如下命令行運行， 並監視指定的文件.

./onchange <command> <file1> [file2] ...

文件變更通知通過 uv_fs_event_init() 啟動:

while (argc-- > 2) {
        fprintf(stderr, "Adding watch on %s\n", argv[argc]);
        uv_fs_event_init(loop, (uv_fs_event_t*) malloc(sizeof(uv_fs_event_t)), argv[argc], run_command, 0);
    }

 

第三個參數是實際監控的文件或者文件夾, 最后一個參數 flags 可取值如下:

 UV_FS_EVENT_WATCH_ENTRY = 1,
 UV_FS_EVENT_STAT = 2,
 UV_FS_EVENT_RECURSIVE = 3

 

若設置 UV_FS_EVENT_WATCH_ENTRY 和 UV_FS_EVENT_STAT 不做任何事情(目前). 設置了 UV_FS_EVENT_RECURSIVE 將會監視子文件夾(需 libuv 支持).

回調函數將接受以下參數:

1. uv_fs_event_t *handle - 監視器. filename

   字段是該監視器需要監視的文件.

2. const char *filename - 如果監視目錄, 則該參數指明該目錄中發生了變更的文件,

   在 Linux 和 Windows 平台上可以是非 null.

3. int flags - UV_RENAME 或 UV_CHANGE.

4. int status - 目前為 0.

我們的例子只是簡單地打印出參數, 並通過 system 函數運行指定命令.

void run_command(uv_fs_event_t *handle, const char *filename, int events, int status) {
    fprintf(stderr, "Change detected in %s: ", handle->filename);
    if (events == UV_RENAME)
        fprintf(stderr, "renamed");
    if (events == UV_CHANGE)
        fprintf(stderr, "changed");

    fprintf(stderr, " %s\n", filename ? filename : "");
    system(command);
}