摘要:利用將傳入的轉為文件描述符新建對象,并對其設置文件描述符讀寫回調函數檢測是否存在,并對其進行初始化。如果傳入在中不存在返回,用于修改事件監聽的回調函數和掩碼。異常事件回調函數當發現套接字發生錯誤后,就會自動刪除該套接字的監聽。
前言
對于異步的任務來說,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, "can"t 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); }swoole_event_cycle 循環周期回調函數
swoole_event_cycle 函數中如果傳入的回調函數為 null,說明用戶想要清除周期回調函數,swoole 將周期函數轉化為 defer 即可。
before 為 1,代表用戶想要在 EventLoop 之前調用該函數,swoole 會將其放在 future_task 中;否則將會在 EventLoop 之后執行,會放在 idle_task 中。
注意如果之前存在過周期循環函數,此次是修改周期回調函數,那么需要在此之前,要將之前的周期回調函數轉為 defer 執行。
PHP_FUNCTION(swoole_event_cycle) { if (!SwooleG.main_reactor) { swoole_php_fatal_error(E_WARNING, "reactor no ready, cannot swoole_event_defer."); RETURN_FALSE; } zval *callback; zend_bool before = 0; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z|b", &callback, &before) == FAILURE) { return; } if (ZVAL_IS_NULL(callback)) { if (SwooleG.main_reactor->idle_task.callback == NULL) { RETURN_FALSE; } else { SwooleG.main_reactor->defer(SwooleG.main_reactor, free_callback, SwooleG.main_reactor->idle_task.data); SwooleG.main_reactor->idle_task.callback = NULL; SwooleG.main_reactor->idle_task.data = NULL; RETURN_TRUE; } } 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_defer_callback *cb = emalloc(sizeof(php_defer_callback)); cb->callback = &cb->_callback; memcpy(cb->callback, callback, sizeof(zval)); sw_zval_add_ref(&callback); if (before == 0) { if (SwooleG.main_reactor->idle_task.data != NULL) { SwooleG.main_reactor->defer(SwooleG.main_reactor, free_callback, SwooleG.main_reactor->idle_task.data); } SwooleG.main_reactor->idle_task.callback = php_swoole_event_onEndCallback; SwooleG.main_reactor->idle_task.data = cb; } else { if (SwooleG.main_reactor->future_task.data != NULL) { SwooleG.main_reactor->defer(SwooleG.main_reactor, free_callback, SwooleG.main_reactor->future_task.data); } SwooleG.main_reactor->future_task.callback = php_swoole_event_onEndCallback; SwooleG.main_reactor->future_task.data = cb; //Registration onBegin callback function swReactor_activate_future_task(SwooleG.main_reactor); } RETURN_TRUE; } static void free_callback(void* data) { php_defer_callback *cb = (php_defer_callback *) data; sw_zval_ptr_dtor(&cb->callback); efree(cb); }
在每次事件循環之前都要執行 onBegin 函數,也就是 swReactor_onBegin,此時會調用 future_task;當 reactor 超時(onTimeout)或者事件循環結束(onFinish),都會調用 swReactor_onTimeout_and_Finish ,此時會調用 idle_task:
static int swReactorEpoll_wait(swReactor *reactor, struct timeval *timeo) { ... while (reactor->running > 0) { if (reactor->onBegin != NULL) { reactor->onBegin(reactor); } n = epoll_wait(epoll_fd, events, max_event_num, msec); ... else if (n == 0) { if (reactor->onTimeout != NULL) { reactor->onTimeout(reactor); } continue; } ... if (reactor->onFinish != NULL) { reactor->onFinish(reactor); } if (reactor->once) { break; } } return 0; } static void swReactor_onBegin(swReactor *reactor) { if (reactor->future_task.callback) { reactor->future_task.callback(reactor->future_task.data); } } static void swReactor_onTimeout_and_Finish(swReactor *reactor) { ... if (reactor->idle_task.callback) { reactor->idle_task.callback(reactor->idle_task.data); } ... }
真正執行回調函數的是 php_swoole_event_onEndCallback:
static void php_swoole_event_onEndCallback(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); } }php_swoole_event_onRead 讀就緒事件回調函數
讀就緒事件回調函數就是簡單的調用用戶的回調函數即可。
static int php_swoole_event_onRead(swReactor *reactor, swEvent *event) { zval *retval; zval **args[1]; php_reactor_fd *fd = event->socket->object; args[0] = &fd->socket; if (sw_call_user_function_ex(EG(function_table), NULL, fd->cb_read, &retval, 1, args, 0, NULL TSRMLS_CC) == FAILURE) { swoole_php_fatal_error(E_WARNING, "swoole_event: onRead handler error."); SwooleG.main_reactor->del(SwooleG.main_reactor, event->fd); return SW_ERR; } if (EG(exception)) { zend_exception_error(EG(exception), E_ERROR TSRMLS_CC); } if (retval != NULL) { sw_zval_ptr_dtor(&retval); } return SW_OK; }php_swoole_event_onWrite 寫就緒事件回調函數
寫就緒事件回調函數就是調用 fd->cb_write 回調函數,當然如果用戶并沒有設置該回調函數的話,就會調用 swReactor_onWrite 發送 socket->out_buffer 的數據或者自動移除寫監聽事件。
static int php_swoole_event_onWrite(swReactor *reactor, swEvent *event) { zval *retval; zval **args[1]; php_reactor_fd *fd = event->socket->object; if (!fd->cb_write) { return swReactor_onWrite(reactor, event); } args[0] = &fd->socket; if (sw_call_user_function_ex(EG(function_table), NULL, fd->cb_write, &retval, 1, args, 0, NULL TSRMLS_CC) == FAILURE) { swoole_php_fatal_error(E_WARNING, "swoole_event: onWrite handler error"); SwooleG.main_reactor->del(SwooleG.main_reactor, event->fd); return SW_ERR; } if (EG(exception)) { zend_exception_error(EG(exception), E_ERROR TSRMLS_CC); } if (retval != NULL) { sw_zval_ptr_dtor(&retval); } return SW_OK; }php_swoole_event_onError 異常事件回調函數
當 reactor 發現套接字發生錯誤后,就會自動刪除該套接字的監聽。
static int php_swoole_event_onError(swReactor *reactor, swEvent *event) { int error; socklen_t len = sizeof(error); if (getsockopt(event->fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0) { swoole_php_fatal_error(E_WARNING, "swoole_event->onError[1]: getsockopt[sock=%d] failed. Error: %s[%d]", event->fd, strerror(errno), errno); } if (error != 0) { swoole_php_fatal_error(E_WARNING, "swoole_event->onError[1]: socket error. Error: %s [%d]", strerror(error), error); } efree(event->socket->object); event->socket->active = 0; SwooleG.main_reactor->del(SwooleG.main_reactor, event->fd); return SW_OK; }swoole_event_del 刪除套接字
刪除套接字就是從 reactor 中刪除監聽的文件描述符 SwooleG.main_reactor->del
PHP_FUNCTION(swoole_event_del) { zval *zfd; if (!SwooleG.main_reactor) { swoole_php_fatal_error(E_WARNING, "reactor no ready, cannot swoole_event_del."); RETURN_FALSE; } if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z", &zfd) == FAILURE) { return; } int socket_fd = swoole_convert_to_fd(zfd TSRMLS_CC); if (socket_fd < 0) { swoole_php_fatal_error(E_WARNING, "unknow type."); RETURN_FALSE; } swConnection *socket = swReactor_get(SwooleG.main_reactor, socket_fd); if (socket->object) { SwooleG.main_reactor->defer(SwooleG.main_reactor, free_event_callback, socket->object); socket->object = NULL; } int ret = SwooleG.main_reactor->del(SwooleG.main_reactor, socket_fd); socket->active = 0; SW_CHECK_RETURN(ret); }swoole_event_exit 退出事件循環
退出事件循環就是將 SwooleG.main_reactor->running 置為 0,使得 while 循環為 false。
PHP_FUNCTION(swoole_event_exit) { if (SwooleWG.in_client == 1) { if (SwooleG.main_reactor) { SwooleG.main_reactor->running = 0; } SwooleG.running = 0; } }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/29437.html
摘要:超過此數量后,新進入的連接將被拒絕。表示連接最大允許空閑的時間。當出錯時底層會認為是惡意連接,丟棄數據并強制關閉連接。在啟動時自動將進程的寫入到文件,在關閉時自動刪除文件。 配置說明 $server->set(array( daemonize => true, log_file => /www/log/swoole.log, reactor_num => 2, ...
摘要:清空主進程殘留的定時器與信號。設定為執行回調函數如果在回調函數中調用了異步系統,啟動函數進行事件循環。因此為了區分兩者,規定并不允許兩者同時存在。 前言 swoole-1.7.2 增加了一個進程管理模塊,用來替代 PHP 的 pcntl 擴展。 PHP自帶的pcntl,存在很多不足,如 pcntl 沒有提供進程間通信的功能 pcntl 不支持重定向標準輸入和輸出 pcntl 只...
摘要:消息隊列更常見的用途是主進程分配任務,子進程消費執行。子進程前面加了個,這是為了防止父進程還未往消息隊列中加入內容直接退出。 前面幾節都是講解pcntl擴展實現的多進程程序。本節給大家介紹swoole擴展的swoole_process模塊。 swoole多進程 swoole_process 是swoole提供的進程管理模塊,用來替代PHP的pcntl擴展。 首先,確保安裝的swoole...
摘要:配合模塊,創建的子進程可以異步的事件驅動模式。默認為阻塞讀取。函數用于將一個加入到的事件監聽中。為事件類型的掩碼,可選擇關閉開啟可讀可寫事件,如,,或者。在程序中使用,可以理解為在進程中將此注冊到事件中。 Process Process是swoole內置的進程管理模塊,用來替代PHP的pcntl擴展。 swoole_process支持重定向標準輸入和輸出,在子進程內echo不會打印屏...
摘要:兩個函數是可選回調函數。附帶了一組可信任證書。應該注意的是,驗證失敗并不意味著連接不能使用。在對證書進行驗證時,有一些安全性檢查并沒有執行,包括證書的失效檢查和對證書中通用名的有效性驗證。 前言 swoole_client 提供了 tcp/udp socket 的客戶端的封裝代碼,使用時僅需 new swoole_client 即可。 swoole 的 socket client 對比...
閱讀 1751·2021-09-27 14:02
閱讀 3100·2021-09-27 13:36
閱讀 1046·2019-08-30 12:46
閱讀 1834·2019-08-30 10:51
閱讀 3571·2019-08-29 17:02
閱讀 941·2019-08-29 16:38
閱讀 1846·2019-08-29 16:37
閱讀 3004·2019-08-26 10:32