Swoole 源碼分析—— swoole_event(上)

來自專欄 Swoole

前言

對於非同步的任務來說，Server 端的 master 進程與 worker 進程會自動將非同步的事件添加到 reactor 的事件循環中去，task_worker 進程不允許存在非同步任務。

對於非同步的 Client 客戶端、swoole_process:: signal、swoole_timer來說，PHP 代碼並不存在 reactor 事件循環，這時候，swoole 就會為 PHP 代碼創建相應的 swoole_event 的 reactor 事件循環，來模擬非同步事件。

除了非同步 Server 和 Client 庫之外，Swoole 擴展還提供了直接操作底層 epoll/kqueue 事件循環的介面。可將其他擴展創建的 socket，PHP 代碼中 stream/socket 擴展創建的 socket 等加入到 Swoole 的EventLoop 中。

只有了解了 swoole_event 的原理，才能更好的使用 swoole 中的定時器、信號、客戶端等等非同步事件介面。

swoole_event_add 添加非同步事件

• 函數首先利用 zend_parse_parameters 解析傳入的參數信息，並複製給 zfd、cb_read 讀回調函數、cb_write 寫回調函數，event_flag 監控事件。
• 利用 swoole_convert_to_fd 將傳入的 zfd 轉為文件描述符
• 新建 php_reactor_fd 對象，並對其設置文件描述符、讀寫回調函數
• php_swoole_check_reactor 檢測是否存在 reactor，並對其進行初始化。
• 設置套接字文件描述符為非阻塞，在 reactor 中添加文件描述符

PHP_FUNCTION(swoole_event_add){ zval *cb_read = NULL; zval *cb_write = NULL; zval *zfd; char *func_name = NULL; long event_flag = 0; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z|zzl", &zfd, &cb_read, &cb_write, &event_flag) == FAILURE) { return; } int socket_fd = swoole_convert_to_fd(zfd TSRMLS_CC); php_reactor_fd *reactor_fd = emalloc(sizeof(php_reactor_fd)); reactor_fd->socket = zfd; sw_copy_to_stack(reactor_fd->socket, reactor_fd->stack.socket); sw_zval_add_ref(&reactor_fd->socket); if (cb_read!= NULL && !ZVAL_IS_NULL(cb_read)) { if (!sw_zend_is_callable(cb_read, 0, &func_name TSRMLS_CC)) { swoole_php_fatal_error(E_ERROR, "Function %s is not callable", func_name); efree(func_name); RETURN_FALSE; } efree(func_name); reactor_fd->cb_read = cb_read; sw_zval_add_ref(&cb_read); sw_copy_to_stack(reactor_fd->cb_read, reactor_fd->stack.cb_read); } else { reactor_fd->cb_read = NULL; } if (cb_write!= NULL && !ZVAL_IS_NULL(cb_write)) { if (!sw_zend_is_callable(cb_write, 0, &func_name TSRMLS_CC)) { swoole_php_fatal_error(E_ERROR, "Function %s is not callable", func_name); efree(func_name); RETURN_FALSE; } efree(func_name); reactor_fd->cb_write = cb_write; sw_zval_add_ref(&cb_write); sw_copy_to_stack(reactor_fd->cb_write, reactor_fd->stack.cb_write); } else { reactor_fd->cb_write = NULL; } php_swoole_check_reactor(); swSetNonBlock(socket_fd); //must be nonblock if (SwooleG.main_reactor->add(SwooleG.main_reactor, socket_fd, SW_FD_USER | event_flag) < 0) { swoole_php_fatal_error(E_WARNING, "swoole_event_add failed."); RETURN_FALSE; } swConnection *socket = swReactor_get(SwooleG.main_reactor, socket_fd); socket->object = reactor_fd; socket->active = 1; socket->nonblock = 1; RETURN_LONG(socket_fd);}

sock 可以為以下四種類型：

• int，就是文件描述符，包括 swoole_client->$sock、swoole_process->$pipe 或者其他 fd
• stream 資源，就是 stream_socket_client/fsockopen 創建的資源
• sockets 資源，就是 sockets 擴展中 socket_create 創建的資源，需要在編譯時加入 ./configure --enable-sockets
• object，swoole_process 或 swoole_client，底層自動轉換為管道或客戶端連接的 socket

從 swoole_convert_to_fd 中可以看到，

• IS_LONG 的 if 分支最為簡單，直接轉為 long 類型即可。
• IS_RESOURCE 資源類型分為兩種
• 一種是 stream_socket_client/fsockopen，是標準 PHP 創建 socket 的方式，這時會調用 SW_ZEND_FETCH_RESOURCE_NO_RETURN 將 zfd 轉為 php_stream 類型，再將 php_stream 類型轉為 socket_fd
• 另一種是 PHP 提供的套接字，此時需要利用 SW_ZEND_FETCH_RESOURCE_NO_RETURN 將 zfd 轉為 php_socket，socket_fd 就是 php_socket 的 bsd_socket 屬性。

• IS_OBJECT 對象類型也分為兩種：
• 程序通過 instanceof_function 函數判斷對象是 swoole_client，如果是則取出其 sock 屬性對象
• 如果對象是 swoole_process 對象，則取出 pipe 對象。

SW_ZEND_FETCH_RESOURCE_NO_RETURN 實際上是一個宏函數，利用的是 zend_fetch_resource 函數。

#define SW_ZEND_FETCH_RESOURCE_NO_RETURN(rsrc, rsrc_type, passed_id, default_id, resource_type_name, resource_type) (rsrc = (rsrc_type) zend_fetch_resource(Z_RES_P(*passed_id), resource_type_name, resource_type))int swoole_convert_to_fd(zval *zfd TSRMLS_DC){ php_stream *stream; int socket_fd;#ifdef SWOOLE_SOCKETS_SUPPORT php_socket *php_sock;#endif if (SW_Z_TYPE_P(zfd) == IS_RESOURCE) { if (SW_ZEND_FETCH_RESOURCE_NO_RETURN(stream, php_stream *, &zfd, -1, NULL, php_file_le_stream())) { if (php_stream_cast(stream, PHP_STREAM_AS_FD_FOR_SELECT | PHP_STREAM_CAST_INTERNAL, (void* )&socket_fd, 1) != SUCCESS || socket_fd < 0) { return SW_ERR; } } else {#ifdef SWOOLE_SOCKETS_SUPPORT if (SW_ZEND_FETCH_RESOURCE_NO_RETURN(php_sock, php_socket *, &zfd, -1, NULL, php_sockets_le_socket())) { socket_fd = php_sock->bsd_socket; } else { swoole_php_fatal_error(E_WARNING, "fd argument must be either valid PHP stream or valid PHP socket resource"); return SW_ERR; }#else swoole_php_fatal_error(E_WARNING, "fd argument must be valid PHP stream resource"); return SW_ERR;#endif } } else if (SW_Z_TYPE_P(zfd) == IS_LONG) { socket_fd = Z_LVAL_P(zfd); if (socket_fd < 0) { swoole_php_fatal_error(E_WARNING, "invalid file descriptor passed"); return SW_ERR; } } else if (SW_Z_TYPE_P(zfd) == IS_OBJECT) { zval *zsock = NULL; if (instanceof_function(Z_OBJCE_P(zfd), swoole_client_class_entry_ptr TSRMLS_CC)) { zsock = sw_zend_read_property(Z_OBJCE_P(zfd), zfd, SW_STRL("sock")-1, 0 TSRMLS_CC); } else if (instanceof_function(Z_OBJCE_P(zfd), swoole_process_class_entry_ptr TSRMLS_CC)) { zsock = sw_zend_read_property(Z_OBJCE_P(zfd), zfd, SW_STRL("pipe")-1, 0 TSRMLS_CC); } if (zsock == NULL || ZVAL_IS_NULL(zsock)) { swoole_php_fatal_error(E_WARNING, "object is not instanceof swoole_client or swoole_process."); return -1; } socket_fd = Z_LVAL_P(zsock); } else { return SW_ERR; } return socket_fd;}

php_swoole_check_reactor 用於檢測 reactor 是否存在。

• 從函數中可以看到，非同步事件只能在 CLI 模式下生效，不能用於 task_worker 進程中。
• 如果當前進程不存在 main_reactor，那麼就要創建 reactor，並且設置事件的回調函數
• 將 swoole_event_wait 註冊為 php 的 shutdown 函數。

void php_swoole_check_reactor(){ if (likely(SwooleWG.reactor_init)) { return; } if (!SWOOLE_G(cli)) { swoole_php_fatal_error(E_ERROR, "async-io must be used in PHP CLI mode."); return; } if (swIsTaskWorker()) { swoole_php_fatal_error(E_ERROR, "cant use async-io in task process."); return; } if (SwooleG.main_reactor == NULL) { swTraceLog(SW_TRACE_PHP, "init reactor"); SwooleG.main_reactor = (swReactor *) sw_malloc(sizeof(swReactor)); if (SwooleG.main_reactor == NULL) { swoole_php_fatal_error(E_ERROR, "malloc failed."); return; } if (swReactor_create(SwooleG.main_reactor, SW_REACTOR_MAXEVENTS) < 0) { swoole_php_fatal_error(E_ERROR, "failed to create reactor."); return; }#ifdef SW_COROUTINE SwooleG.main_reactor->can_exit = php_coroutine_reactor_can_exit;#endif //client, swoole_event_exit will set swoole_running = 0 SwooleWG.in_client = 1; SwooleWG.reactor_wait_onexit = 1; SwooleWG.reactor_ready = 0; //only client side php_swoole_at_shutdown("swoole_event_wait"); } SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_USER | SW_EVENT_READ, php_swoole_event_onRead); SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_USER | SW_EVENT_WRITE, php_swoole_event_onWrite); SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_USER | SW_EVENT_ERROR, php_swoole_event_onError); SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_WRITE, swReactor_onWrite); SwooleWG.reactor_init = 1;}

swoole_event_set 函數

參數與 swoole_event_add 完全相同。如果傳入 $fd 在 EventLoop 中不存在返回 false，用於修改事件監聽的回調函數和掩碼。

最核心的是調用了 SwooleG.main_reactor->set 函數。

PHP_FUNCTION(swoole_event_set){ zval *cb_read = NULL; zval *cb_write = NULL; zval *zfd; char *func_name = NULL; long event_flag = 0; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z|zzl", &zfd, &cb_read, &cb_write, &event_flag) == FAILURE) { return; } int socket_fd = swoole_convert_to_fd(zfd TSRMLS_CC); swConnection *socket = swReactor_get(SwooleG.main_reactor, socket_fd); php_reactor_fd *ev_set = socket->object; if (cb_read != NULL && !ZVAL_IS_NULL(cb_read)) { if (!sw_zend_is_callable(cb_read, 0, &func_name TSRMLS_CC)) { swoole_php_fatal_error(E_ERROR, "Function %s is not callable", func_name); efree(func_name); RETURN_FALSE; } else { if (ev_set->cb_read) { sw_zval_ptr_dtor(&ev_set->cb_read); } ev_set->cb_read = cb_read; sw_zval_add_ref(&cb_read); sw_copy_to_stack(ev_set->cb_read, ev_set->stack.cb_read); efree(func_name); } } if (cb_write != NULL && !ZVAL_IS_NULL(cb_write)) { if (socket_fd == 0 && (event_flag & SW_EVENT_WRITE)) { swoole_php_fatal_error(E_WARNING, "invalid socket fd [%d].", socket_fd); RETURN_FALSE; } if (!sw_zend_is_callable(cb_write, 0, &func_name TSRMLS_CC)) { swoole_php_fatal_error(E_ERROR, "Function %s is not callable", func_name); efree(func_name); RETURN_FALSE; } else { if (ev_set->cb_write) { sw_zval_ptr_dtor(&ev_set->cb_write); } ev_set->cb_write = cb_write; sw_zval_add_ref(&cb_write); sw_copy_to_stack(ev_set->cb_write, ev_set->stack.cb_write); efree(func_name); } } if ((event_flag & SW_EVENT_READ) && ev_set->cb_read == NULL) { swoole_php_fatal_error(E_WARNING, "swoole_event: no read callback."); RETURN_FALSE; } if ((event_flag & SW_EVENT_WRITE) && ev_set->cb_write == NULL) { swoole_php_fatal_error(E_WARNING, "swoole_event: no write callback."); RETURN_FALSE; } if (SwooleG.main_reactor->set(SwooleG.main_reactor, socket_fd, SW_FD_USER | event_flag) < 0) { swoole_php_fatal_error(E_WARNING, "swoole_event_set failed."); RETURN_FALSE; } RETURN_TRUE;}

swoole_event_write 函數

用於PHP自帶 stream/sockets 擴展創建的 socket，使用 fwrite/socket_send 等函數向對端發送數據。當發送的數據量較大，socket 寫緩存區已滿，就會發送阻塞等待或者返回 EAGAIN 錯誤。

swoole_event_write 函數可以將 stream/sockets 資源的數據發送變成非同步的，當緩衝區滿了或者返回 EAGAIN，swoole 底層會將數據加入到發送隊列，並監聽可寫。socket 可寫時 swoole 底層會自動寫入。

swoole_event_write 函數主要調用了 SwooleG.main_reactor->write 實現功能。

PHP_FUNCTION(swoole_event_write){ zval *zfd; char *data; zend_size_t len; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "zs", &zfd, &data, &len) == FAILURE) { return; } if (len <= 0) { swoole_php_fatal_error(E_WARNING, "data empty."); RETURN_FALSE; } int socket_fd = swoole_convert_to_fd(zfd TSRMLS_CC); if (socket_fd < 0) { swoole_php_fatal_error(E_WARNING, "unknow type."); RETURN_FALSE; } php_swoole_check_reactor(); if (SwooleG.main_reactor->write(SwooleG.main_reactor, socket_fd, data, len) < 0) { RETURN_FALSE; } else { RETURN_TRUE; }}

swoole_event_wait 函數

swoole_event_wait 函數用於讓整個 PHP 程序進入事件循環，剛剛我們可以看到，swoole 把這個函數註冊為 shutdown 函數，腳本在停止之前會自動調用這個函數。如果自己想要在程序中間進行事件循環可以調用該函數。

該函數最重要的就是調用 SwooleG.main_reactor->wait 函數，該函數會不斷 while 循環阻塞在 reactor->wait 上，直到有信號或者讀寫就緒事件發生。

PHP_FUNCTION(swoole_event_wait){ if (!SwooleG.main_reactor) { return; } php_swoole_event_wait();}void php_swoole_event_wait(){ if (SwooleWG.in_client == 1 && SwooleWG.reactor_ready == 0 && SwooleG.running) { if (PG(last_error_message)) { switch (PG(last_error_type)) { case E_ERROR: case E_CORE_ERROR: case E_USER_ERROR: case E_COMPILE_ERROR: return; default: break; } } SwooleWG.reactor_ready = 1;#ifdef HAVE_SIGNALFD if (SwooleG.main_reactor->check_signalfd) { swSignalfd_setup(SwooleG.main_reactor); }#endif#ifdef SW_COROUTINE if (COROG.active == 0) { coro_init(TSRMLS_C); }#endif if (!swReactor_empty(SwooleG.main_reactor)) { int ret = SwooleG.main_reactor->wait(SwooleG.main_reactor, NULL); if (ret < 0) { swoole_php_fatal_error(E_ERROR, "reactor wait failed. Error: %s [%d]", strerror(errno), errno); } } if (SwooleG.timer.map) { php_swoole_clear_all_timer(); } SwooleWG.reactor_exit = 1; }}

swoole_event_defer 延遲執行回調函數

swoole_event_defer 函數會利用 SwooleG.main_reactor->defer 向 reactor 註冊延遲執行的函數:

PHP_FUNCTION(swoole_event_defer){ zval *callback; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z", &callback) == FAILURE) { return; } char *func_name; if (!sw_zend_is_callable(callback, 0, &func_name TSRMLS_CC)) { swoole_php_fatal_error(E_ERROR, "Function %s is not callable", func_name); efree(func_name); RETURN_FALSE; } efree(func_name); php_swoole_check_reactor(); php_defer_callback *defer = emalloc(sizeof(php_defer_callback)); defer->callback = &defer->_callback; memcpy(defer->callback, callback, sizeof(zval)); sw_zval_add_ref(&callback); SW_CHECK_RETURN(SwooleG.main_reactor->defer(SwooleG.main_reactor, php_swoole_event_onDefer, defer));}

SwooleG.main_reactor->defer 函數就是 swReactor_defer。從該函數可以看出，如果調用 defer 的時候 reactor 還沒有啟動，那麼就用定時器來實現延遲執行；如果此時 reactor 已經啟動了，那麼就添加到 defer_tasks 屬性中。

static int swReactor_defer(swReactor *reactor, swCallback callback, void *data){ swDefer_callback *cb = sw_malloc(sizeof(swDefer_callback)); if (!cb) { swWarn("malloc(%ld) failed.", sizeof(swDefer_callback)); return SW_ERR; } cb->callback = callback; cb->data = data; if (unlikely(reactor->start == 0)) { if (unlikely(SwooleG.timer.fd == 0)) { swTimer_init(1); } SwooleG.timer.add(&SwooleG.timer, 1, 0, cb, swReactor_defer_timer_callback); } else { LL_APPEND(reactor->defer_tasks, cb); } return SW_OK;}static void swReactor_defer_timer_callback(swTimer *timer, swTimer_node *tnode){ swDefer_callback *cb = (swDefer_callback *) tnode->data; cb->callback(cb->data); sw_free(cb);}

reactor 無論是超時還是事件循環結束，都會調用 swReactor_onTimeout_and_Finish 函數，該函數會調用 reactor->defer_tasks，執行之後就會自動刪除延遲任務。

static void swReactor_onTimeout(swReactor *reactor){ swReactor_onTimeout_and_Finish(reactor); if (reactor->disable_accept) { reactor->enable_accept(reactor); reactor->disable_accept = 0; }}static void swReactor_onFinish(swReactor *reactor){ //check signal if (reactor->singal_no) { swSignal_callback(reactor->singal_no); reactor->singal_no = 0; } swReactor_onTimeout_and_Finish(reactor);}static void swReactor_onTimeout_and_Finish(swReactor *reactor){ if (reactor->check_timer) { swTimer_select(&SwooleG.timer); } do { swDefer_callback *defer_tasks = reactor->defer_tasks; swDefer_callback *cb, *tmp; reactor->defer_tasks = NULL; LL_FOREACH(defer_tasks, cb) { cb->callback(cb->data); } LL_FOREACH_SAFE(defer_tasks, cb, tmp) { sw_free(cb); } } while (reactor->defer_tasks); ...}

延遲任務的執行就調用回調函數：

static void php_swoole_event_onDefer(void *_cb){ php_defer_callback *defer = _cb; zval *retval; if (sw_call_user_function_ex(EG(function_table), NULL, defer->callback, &retval, 0, NULL, 0, NULL TSRMLS_CC) == FAILURE) { swoole_php_fatal_error(E_WARNING, "swoole_event: defer handler error"); return; } if (EG(exception)) { zend_exception_error(EG(exception), E_ERROR TSRMLS_CC); } if (retval != NULL) { sw_zval_ptr_dtor(&retval); } sw_zval_ptr_dtor(&defer->callback); efree(defer);}

推薦閱讀：