成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

Swoole 源碼分析——Async 異步事件系統(tǒng) Swoole_Event

stefanieliang / 2471人閱讀

摘要:利用將傳入的轉(zhuǎn)為文件描述符新建對象,并對其設(shè)置文件描述符讀寫回調(diào)函數(shù)檢測是否存在,并對其進(jìn)行初始化。如果傳入在中不存在返回,用于修改事件監(jiān)聽的回調(diào)函數(shù)和掩碼。異常事件回調(diào)函數(shù)當(dāng)發(fā)現(xiàn)套接字發(fā)生錯誤后,就會自動刪除該套接字的監(jiān)聽。

前言

對于異步的任務(wù)來說,Server 端的 master 進(jìn)程與 worker 進(jìn)程會自動將異步的事件添加到 reactor 的事件循環(huán)中去,task_worker 進(jìn)程不允許存在異步任務(wù)。

對于異步的 Client 客戶端、swoole_process:: signalswoole_timer來說,PHP 代碼并不存在 reactor 事件循環(huán),這時候,swoole 就會為 PHP 代碼創(chuàng)建相應(yīng)的 swoole_eventreactor 事件循環(huán),來模擬異步事件。

除了異步 ServerClient 庫之外,Swoole 擴(kuò)展還提供了直接操作底層 epoll/kqueue 事件循環(huán)的接口。可將其他擴(kuò)展創(chuàng)建的 socket,PHP 代碼中 stream/socket 擴(kuò)展創(chuàng)建的 socket 等加入到 SwooleEventLoop 中。

只有了解了 swoole_event 的原理,才能更好的使用 swoole 中的定時器、信號、客戶端等等異步事件接口。

swoole_event_add 添加異步事件

函數(shù)首先利用 zend_parse_parameters 解析傳入的參數(shù)信息,并復(fù)制給 zfd、cb_read 讀回調(diào)函數(shù)、cb_write 寫回調(diào)函數(shù),event_flag 監(jiān)控事件。

利用 swoole_convert_to_fd 將傳入的 zfd 轉(zhuǎn)為文件描述符

新建 php_reactor_fd 對象,并對其設(shè)置文件描述符、讀寫回調(diào)函數(shù)

php_swoole_check_reactor 檢測是否存在 reactor,并對其進(jìn)行初始化。

設(shè)置套接字文件描述符為非阻塞,在 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->$sockswoole_process->$pipe 或者其他 fd

stream 資源,就是 stream_socket_client/fsockopen 創(chuàng)建的資源

sockets 資源,就是 sockets 擴(kuò)展中 socket_create 創(chuàng)建的資源,需要在編譯時加入 ./configure --enable-sockets

object,swoole_processswoole_client,底層自動轉(zhuǎn)換為管道或客戶端連接的 socket

swoole_convert_to_fd 中可以看到,

IS_LONGif 分支最為簡單,直接轉(zhuǎn)為 long 類型即可。

IS_RESOURCE 資源類型分為兩種

一種是 stream_socket_client/fsockopen,是標(biāo)準(zhǔn) PHP 創(chuàng)建 socket 的方式,這時會調(diào)用 SW_ZEND_FETCH_RESOURCE_NO_RETURNzfd 轉(zhuǎn)為 php_stream 類型,再將 php_stream 類型轉(zhuǎn)為 socket_fd

另一種是 PHP 提供的套接字,此時需要利用 SW_ZEND_FETCH_RESOURCE_NO_RETURNzfd 轉(zhuǎn)為 php_socket,socket_fd 就是 php_socketbsd_socket 屬性。

IS_OBJECT 對象類型也分為兩種:

程序通過 instanceof_function 函數(shù)判斷對象是 swoole_client,如果是則取出其 sock 屬性對象

如果對象是 swoole_process 對象,則取出 pipe 對象。

SW_ZEND_FETCH_RESOURCE_NO_RETURN 實際上是一個宏函數(shù),利用的是 zend_fetch_resource 函數(shù)。

#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 是否存在。

從函數(shù)中可以看到,異步事件只能在 CLI 模式下生效,不能用于 task_worker 進(jìn)程中。

如果當(dāng)前進(jìn)程不存在 main_reactor,那么就要創(chuàng)建 reactor,并且設(shè)置事件的回調(diào)函數(shù)

swoole_event_wait 注冊為 phpshutdown 函數(shù)。

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 函數(shù)

參數(shù)與 swoole_event_add 完全相同。如果傳入 $fdEventLoop 中不存在返回 false,用于修改事件監(jiān)聽的回調(diào)函數(shù)和掩碼。

最核心的是調(diào)用了 SwooleG.main_reactor->set 函數(shù)。

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 函數(shù)

用于PHP自帶 stream/sockets 擴(kuò)展創(chuàng)建的 socket,使用 fwrite/socket_send 等函數(shù)向?qū)Χ税l(fā)送數(shù)據(jù)。當(dāng)發(fā)送的數(shù)據(jù)量較大,socket 寫緩存區(qū)已滿,就會發(fā)送阻塞等待或者返回 EAGAIN 錯誤。

swoole_event_write 函數(shù)可以將 stream/sockets 資源的數(shù)據(jù)發(fā)送變成異步的,當(dāng)緩沖區(qū)滿了或者返回 EAGAIN,swoole 底層會將數(shù)據(jù)加入到發(fā)送隊列,并監(jiān)聽可寫。socket 可寫時 swoole 底層會自動寫入。

swoole_event_write 函數(shù)主要調(diào)用了 SwooleG.main_reactor->write 實現(xiàn)功能。

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 函數(shù)

swoole_event_wait 函數(shù)用于讓整個 PHP 程序進(jìn)入事件循環(huán),剛剛我們可以看到,swoole 把這個函數(shù)注冊為 shutdown 函數(shù),腳本在停止之前會自動調(diào)用這個函數(shù)。如果自己想要在程序中間進(jìn)行事件循環(huán)可以調(diào)用該函數(shù)。

該函數(shù)最重要的就是調(diào)用 SwooleG.main_reactor->wait 函數(shù),該函數(shù)會不斷 while 循環(huán)阻塞在 reactor->wait 上,直到有信號或者讀寫就緒事件發(fā)生。

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 延遲執(zhí)行回調(diào)函數(shù)

swoole_event_defer 函數(shù)會利用 SwooleG.main_reactor->deferreactor 注冊延遲執(zhí)行的函數(shù):

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 函數(shù)就是 swReactor_defer。從該函數(shù)可以看出,如果調(diào)用 defer 的時候 reactor 還沒有啟動,那么就用定時器來實現(xiàn)延遲執(zhí)行;如果此時 reactor 已經(jīng)啟動了,那么就添加到 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 無論是超時還是事件循環(huán)結(jié)束,都會調(diào)用 swReactor_onTimeout_and_Finish 函數(shù),該函數(shù)會調(diào)用 reactor->defer_tasks,執(zhí)行之后就會自動刪除延遲任務(wù)。

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);
    
    ...
}

延遲任務(wù)的執(zhí)行就調(diào)用回調(diào)函數(shù):

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 循環(huán)周期回調(diào)函數(shù)

swoole_event_cycle 函數(shù)中如果傳入的回調(diào)函數(shù)為 null,說明用戶想要清除周期回調(diào)函數(shù),swoole 將周期函數(shù)轉(zhuǎn)化為 defer 即可。

before 為 1,代表用戶想要在 EventLoop 之前調(diào)用該函數(shù),swoole 會將其放在 future_task 中;否則將會在 EventLoop 之后執(zhí)行,會放在 idle_task 中。

注意如果之前存在過周期循環(huán)函數(shù),此次是修改周期回調(diào)函數(shù),那么需要在此之前,要將之前的周期回調(diào)函數(shù)轉(zhuǎn)為 defer 執(zhí)行。

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);
}

在每次事件循環(huán)之前都要執(zhí)行 onBegin 函數(shù),也就是 swReactor_onBegin,此時會調(diào)用 future_task;當(dāng) reactor 超時(onTimeout)或者事件循環(huán)結(jié)束(onFinish),都會調(diào)用 swReactor_onTimeout_and_Finish ,此時會調(diào)用 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);
    }
    
    ...
}

真正執(zhí)行回調(diào)函數(shù)的是 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 讀就緒事件回調(diào)函數(shù)

讀就緒事件回調(diào)函數(shù)就是簡單的調(diào)用用戶的回調(diào)函數(shù)即可。

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 寫就緒事件回調(diào)函數(shù)

寫就緒事件回調(diào)函數(shù)就是調(diào)用 fd->cb_write 回調(diào)函數(shù),當(dāng)然如果用戶并沒有設(shè)置該回調(diào)函數(shù)的話,就會調(diào)用 swReactor_onWrite 發(fā)送 socket->out_buffer 的數(shù)據(jù)或者自動移除寫監(jiān)聽事件。

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 異常事件回調(diào)函數(shù)

當(dāng) reactor 發(fā)現(xiàn)套接字發(fā)生錯誤后,就會自動刪除該套接字的監(jiān)聽。

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 中刪除監(jiān)聽的文件描述符 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 退出事件循環(huán)

退出事件循環(huán)就是將 SwooleG.main_reactor->running 置為 0,使得 while 循環(huán)為 false。

PHP_FUNCTION(swoole_event_exit)
{
    if (SwooleWG.in_client == 1)
    {
        if (SwooleG.main_reactor)
        {
            SwooleG.main_reactor->running = 0;
        }
        SwooleG.running = 0;
    }
}

文章版權(quán)歸作者所有,未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址:http://systransis.cn/yun/29437.html

相關(guān)文章

  • Swoole筆記(五)

    摘要:超過此數(shù)量后,新進(jìn)入的連接將被拒絕。表示連接最大允許空閑的時間。當(dāng)出錯時底層會認(rèn)為是惡意連接,丟棄數(shù)據(jù)并強(qiáng)制關(guān)閉連接。在啟動時自動將進(jìn)程的寫入到文件,在關(guān)閉時自動刪除文件。 配置說明 $server->set(array( daemonize => true, log_file => /www/log/swoole.log, reactor_num => 2, ...

    zgbgx 評論0 收藏0
  • Swoole 源碼分析——進(jìn)程管理 Swoole_Process

    摘要:清空主進(jìn)程殘留的定時器與信號。設(shè)定為執(zhí)行回調(diào)函數(shù)如果在回調(diào)函數(shù)中調(diào)用了異步系統(tǒng),啟動函數(shù)進(jìn)行事件循環(huán)。因此為了區(qū)分兩者,規(guī)定并不允許兩者同時存在。 前言 swoole-1.7.2 增加了一個進(jìn)程管理模塊,用來替代 PHP 的 pcntl 擴(kuò)展。 PHP自帶的pcntl,存在很多不足,如 pcntl 沒有提供進(jìn)程間通信的功能 pcntl 不支持重定向標(biāo)準(zhǔn)輸入和輸出 pcntl 只...

    pepperwang 評論0 收藏0
  • PHP多進(jìn)程系列筆記(五)

    摘要:消息隊列更常見的用途是主進(jìn)程分配任務(wù),子進(jìn)程消費執(zhí)行。子進(jìn)程前面加了個,這是為了防止父進(jìn)程還未往消息隊列中加入內(nèi)容直接退出。 前面幾節(jié)都是講解pcntl擴(kuò)展實現(xiàn)的多進(jìn)程程序。本節(jié)給大家介紹swoole擴(kuò)展的swoole_process模塊。 swoole多進(jìn)程 swoole_process 是swoole提供的進(jìn)程管理模塊,用來替代PHP的pcntl擴(kuò)展。 首先,確保安裝的swoole...

    qianfeng 評論0 收藏0
  • Swoole筆記(四)

    摘要:配合模塊,創(chuàng)建的子進(jìn)程可以異步的事件驅(qū)動模式。默認(rèn)為阻塞讀取。函數(shù)用于將一個加入到的事件監(jiān)聽中。為事件類型的掩碼,可選擇關(guān)閉開啟可讀可寫事件,如,,或者。在程序中使用,可以理解為在進(jìn)程中將此注冊到事件中。 Process Process是swoole內(nèi)置的進(jìn)程管理模塊,用來替代PHP的pcntl擴(kuò)展。 swoole_process支持重定向標(biāo)準(zhǔn)輸入和輸出,在子進(jìn)程內(nèi)echo不會打印屏...

    yzd 評論0 收藏0
  • Swoole 源碼分析——Client模塊之Connect

    摘要:兩個函數(shù)是可選回調(diào)函數(shù)。附帶了一組可信任證書。應(yīng)該注意的是,驗證失敗并不意味著連接不能使用。在對證書進(jìn)行驗證時,有一些安全性檢查并沒有執(zhí)行,包括證書的失效檢查和對證書中通用名的有效性驗證。 前言 swoole_client 提供了 tcp/udp socket 的客戶端的封裝代碼,使用時僅需 new swoole_client 即可。 swoole 的 socket client 對比...

    Charles 評論0 收藏0

發(fā)表評論

0條評論

最新活動
閱讀需要支付1元查看
<