Swoole 源码分析——Async 异步事件系统 Swoole_Event

liwenbocsu 2019-06-28

前言

对于异步的任务来说，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;
    }
}
liwenbocsu

liwenbocsu

相关推荐

php使用event扩展的io复用测试的示例

//sock_set_option; //复用还处于 TIME_WAIT

Nicolase / 5评论 2020-10-20

从Linux源码看Socket(TCP)的Listen及连接队列

从Linux源码看Socket的listen及连接队列前言笔者一直觉得如果能知道从应用到框架再到操作系统的每一处代码，是一件Exciting的事情。今天笔者就来从Linux源码的角度看下Server端的Socket在进行listen的时候到底做了哪些事情，

一叶梧桐 / 6评论 2020-10-14

从Linux源码看Socket(TCP)的Bind

笔者一直觉得如果能知道从应用到框架再到操作系统的每一处代码，是一件Exciting的事情。今天笔者就来从Linux源码的角度看下Server端的Socket在进行bind的时候到底做了哪些事情。众所周知，一个Server端Socket的建立，需要socke

lzzyok / 2评论 2020-10-10

基于python实现简单C/S模式代码实例

C/S模式就是指客bai户端/服务器模式，du是计算机软件协同工作的一种模式。由于Web浏览器的兴起，B/S模式逐步取代了daoC/S模式，被更广泛地应用。PC机的资源没有大型、中型甚至小型主机丰富，但将多台PC机联成网，必然会增加资源含量，各个用户都在网

digwtx / 0评论 2020-09-14

Python gevent协程切换实现详解

大家都知道gevent的机制是单线程+协程机制，当遇到可能会阻塞的操作时，就切换到可运行的协程中继续运行，以此来实现提交系统运行效率的目标，但是具体是怎么实现的呢？让我们直接从代码中看一下吧。这里会开启一个死循环，在循环中调用self._sock.recv

efeve / 0评论 2020-09-14

php 利用socket发送GET，POST请求的实例代码

　　作为php程序员一定会接触http协议，也只有深入了解http协议，编程水平才会更进一步。最近我一直在学习php的关于http的编程，许多东西恍然大悟，受益匪浅。希望分享给大家。本文需要有一定http基础的开发者阅读。　　今天给大家带来的是如何利用so

poplpsure / 0评论 2020-08-17

python socket了解使用

#服务器端import socketserver = socket.socket()server.bind#绑定要监听端口server.listen()#监听conn, addr = server.accept() # 等电话打进来# conn就是客户端

ITxiaobaibai / 0评论 2020-07-26

浅析Linux中的零拷贝技术

本文探讨Linux中 主要的几种零拷贝技术 以及零拷贝技术 适用的场景 。为了迅速建立起零拷贝的概念，我们拿一个常用的场景进行引入：。在写一个服务端程序时，文件下载是一个基本功能。这时候服务端的任务是：将服务端主机磁盘中的文件不做修改地从已连接的socke

libowenhit / 0评论 2020-07-23

浅析Linux中的零拷贝技术

在写一个服务端程序时，文件下载是一个基本功能。这时候服务端的任务是：将服务端主机磁盘中的文件不做修改地从已连接的socket发出去，我们通常用下面的代码完成：。基本操作就是循环的从磁盘读入文件内容到缓冲区，再将缓冲区的内容发送到socket。实际上在以上I

luckykapok / 0评论 2020-07-06

关于linux进程间的close-on-exec机制

大部分这种问题都能够解决，在文章的最后，提到了一种特殊情况，就是父子进程中的端口占用情况。父进程监听一个端口后，fork出一个子进程，然后kill掉父进程，再重启父进程，这个时候提示端口占用，用netstat查看，子进程占用了父进程监听的端口。刚刚fork

hongsheyoumo / 0评论 2020-06-27

【从BIO到Netty】1.BIO存在的问题

package org.scaventz.bio;import java.net.Socket;private ServerSocket serverSocket;private final int PORT = 8080;serverSocket = n

jannal / 0评论 2020-06-21

记录一次kattle抽取数据报错：socket read timeout

让统计个数据，但是数据跑的非常慢不多，还跑到一半报错。后来让同事帮忙看下。我之前是用的表输入-》插入/更新，后来同事改成了表输入-》表输出。

lanmantech / 0评论 2020-06-16

ubuntu 16.04快速搭建nginx + php7.1 + mysql 环境

安装Nginx使用 apt-get install nginx 就能自动安装 Nginx。为了确保获得最新的Nginx，可以先使用 apt-get update 命令更新源列表。安装好之后，使用 dpkg -S nginx 命令来搜索 nginx相关文件。

咻咻ing / 0评论 2020-06-16

Node.js API详解之 dgram模块用法实例分析

dgram模块提供了 UDP 数据包 socket 的实现。创建一个 dgram.Socket 对象. 一旦创建了套接字，调用 socket.bind() 会指示套接字开始监听数据报消息。如果 address 和 port 没传给 socket.bind(

weibingbingnet / 0评论 2020-06-14

从linux源码看socket的阻塞和非阻塞

笔者一直觉得如果能知道从应用到框架再到操作系统的每一处代码，是一件Exciting的事情。大部分高性能网络框架采用的是非阻塞模式。笔者这次就从linux源码的角度来阐述socket阻塞和非阻塞的区别。本文源码均来自采用Linux-2.6.24内核版本。很明

woyanyouxin / 0评论 2020-06-04

nginx优化之keepalive

一个解决方案是 TCP 堆栈的 TCP_NODELAY选项。这样就可以使缓冲区中的数据立即发送出去。这一机制是由Nagle算法保证。Nagle化后来成了一种标准并且立即在因特网上得以实现。现在假设某个应用程序发出了一个请求，希望发送小块数据。如果请求立即发

houjinkai / 0评论 2020-06-03

Redis单线程的正确理解

很多同学对Redis的单线程和I/O多路复用技术并不是很了解，所以我用简单易懂的语言让大家了解下Redis单线程和I/O多路复用技术的原理，对学好和运用好Redis打下基础。Redis客户端对服务端的每次调用都经历了发送命令，执行命令，返回结果三个过程。服

txj / 0评论 2020-06-02

Python3 socket 实现即时通讯脚本，threading 多线程

先执行 服务端代码，再执行客户端代码，可以实现基本通讯功能。

Winterto0 / 0评论 2020-05-29

Socket.IO入门原理

node.js提供了高效的服务端运行环境，但是由于浏览器端对HTML5的支持不一，为了兼容所有浏览器，提供卓越的实时的用户体验，并且为程序员提供客户端与服务端一致的编程体验，于是socket.io诞生。Socket.io将Websocket和轮询 机制以及

Chydar / 0评论 2020-05-15

RHEL 7.4 下yum安装配置nginx与php

我们可以这样增加两个资源库。安装完成后可以启动nginx，在浏览器里面访问，查看nginx是否安装成功。由于最新版本的php已经集成了php-fpm，故只需安装php即可。yum install php-gd php-mysql php-mbstring

小木兮子 / 0评论 2020-05-14

如何通过Python3和ssl实现加密通信功能

msg = "do i connect with server ?msg = f"yes , you have client_socketect with server.\r\n".encode

LczPtr / 0评论 2020-05-09

TCP通信程序练习3——服务器数据写入文件

public static void main(String[] args) throws IOException {. Socket s = new Socket("192.168.50.76",11111);BufferedRead

简单的快乐 / 0评论 2020-05-09

C语言基础-网络相关函数

位于：<sys/socket.h>;功能：把一个本地协议地址赋予一个套接字;返回：成功返回0，出错返回-1。位于：<sys/socket.h>;功能：关闭套接字;参数：sockfd为套接字标识；howto可能为: SHUT_RD关闭

bentocaffe / 0评论 2020-05-09

Python学习第二十二天

须知一个完整的计算机系统是由硬件、操作系统、应用软件三者组成,具备了这三个条件，一台计算机系统就可以自己跟自己玩了。如果你要跟别人一起玩，那你就需要上网了，什么是互联网？如果把计算机比作人，互联网协议就是计算机界的英语。所有的计算机都学会了互联网协议，那所

范群松 / 0评论 2020-05-06

TCP/IP协议和HTTP协议

　　看过谢希仁老师的《计算机网络》，但是当时编程能力弱，没有实践经验，现在遇到了关于网络连接方面的问题，还是有必要复习和梳理一些概念和过程。tcp协议存在于5层网络模型中的传输层，三次握手说的就是tcp。　　套接字是对TCP/IP协议的封装，是通信的基石，

zkwgpp / 0评论 2020-05-04

mysql基本使用

如何解决语言沟通的障碍？客户端连接服务端。update mysql.user set password=password where user=‘root‘ and host=‘localhost‘;字符编码是在默认的配置文件my_default.in

DAV数据库 / 0评论 2020-05-03

Unix域套接字-Unix Domain Socket（转）

add by zhj: Unix Domain Socket是网络socket的优化，适用于服务端与客户端在同一台机器上的情况。相比网络socket，它可以提高通信效率。最近在搭建Nginx+Gunicorn的时候，返现这两个进程可以通过一个后缀为.soc

书虫媛 / 0评论 2020-05-03

Netty消息传递

2、buffer里面数据未被读取完怎么办？

gzx0 / 0评论 2020-05-01

Netty解决粘包问题

2、buffer里面数据未被读取完怎么办？

追寻水中桥 / 0评论 2020-05-01

Python 第一个脚本

通过黑客python编程写的第一个脚本， 实现监听、连接、代码是写了上传功能、执行文件功能、但是测试是不成功的。通过这个脚本、了解几个模块 threading，getopt、subprocess、socket 、sys ;写代码过程中，发现boolean

cas的无名 / 0评论 2020-04-26

python学习40——并发编程之多线程

# 开启线程不需要在main下面执行代码 直接书写就可以。# 但是我们还是习惯性的将启动命令写在main下面。t.start() # 创建线程的开销非常小 几乎是代码一执行线程就已经创建了。"""针对刷个下划线开头双下滑线结

chongtianfeiyu / 0评论 2020-04-25

python入行036（socket模块）

　　socket是一个位于应用层和传输层之间的“抽象层”，他把复杂的tcp/ip协议族隐藏在socket接口后面。socket抽象层位于传输层上面，用户无需深入理解tcp/udp协议，只需要遵循socket的规定去编程，写出的程序自然就是遵循tcp/udp

CloudXli / 0评论 2020-04-20

python 36 套接字编程

phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM). phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM). sever=socket.soc

阿艾辣悟叩德 / 0评论 2020-04-20

Linux多线程

和多进程相比，多线程是一种比较节省资源的多任务操作方式。启动一个新的进程必须分配给它独立的地址空间，每个进程都有自己的堆栈段和数据段，系统开销比较高，进行数据的传递只能通过进行间通信的方式进行。在同一个进程中，可以运行多个线程，运行于同一个进程中的多个线程

insularisland / 0评论 2020-04-20

Linux多进程的应用

前面的章节介绍socket通信的时候，socket的服务端在同一时间只能和一个客户端通信，并不是服务端有多忙，而是因为单进程的程序在同一时间只能做一件事情，不可能一边等待客户端的新连接一边与其它的客户端进行通信。如果把socket服务端改为多进程，在每次a

iamplane / 0评论 2020-04-20

Cilium架构 (Cilium 2)

本文档描述了Cilium的架构。它通过记录BPF数据路径的钩子来实现Cilium数据路径，那么Cilium数据路径是如何与容器编排层继承，以及如何在各层之间更新对象的？该钩子可用于过滤程序丢弃恶意的或非期望的流量，以及其他常见的DDOS防护机制。此时，钩子

akcsdno / 0评论 2020-04-18

网络编程套接字，osi七层架构各层协议最全讲解

在计算机通信领域，socket 被翻译为“套接字”，它是计算机之间进行通信的一种约定或一种方式。通过 socket 这种约定，一台计算机可以接收其他计算机的数据，也可以向其他计算机发送数据。　　Socket()函数返回一个整型的Socket描述符，随后的连

架构技术交流 / 0评论 2020-04-14

有关多语言对WebSocket的支持情况和WebSocket和Socket的区别

WebSocket是应用层层协议，传输层采用TCP协议，但是在建立连接前需要特殊的HTPP请求进行连接。Apache 对 WebSocket 的支持： Apache Module mod_proxy_wstunnel. Nginx 对 WebSockets

woniyu / 0评论 2020-04-14

Redis（Remote Dictionary Server）

Redis是一个开源的使用C语言编写、支持网络、可基于内存亦可持久化的日志型、key-value数据库，并提供多种语言的API。与传统数据库不同的是redis的数据是存在内存中的，所以读写速度非常快，因此redis被广泛应用于缓存方向。另外，redis也经

憧憬 / 0评论 2020-04-11

Python网络通信编程总结

if data.decode()==‘exit‘: #如果客户端发来exit，就断开连接，判断语句要放在接收到客户端消息后，立刻执行，否则，不起作用。在使用线程实现了多个客户端和服务器之间的通信后。由于在客户端中input方法输入发送数据，而该

逍遥友 / 0评论 2020-04-07

单线程下TCP客户端循环发送数据给服务器

# 1.创建tcp套接字

hellobabygogo / 0评论 2020-04-07

Python3标准库：selectors I/O多路复用抽象

selectors模块在select中平台特定的I/O监视函数之上提供了一个平台独立的抽象层。选择器对象提供了一些方法，可以指定在一个套接字上查找哪些事件，然后以一种平台独立的方式让调用者等待事件。注册对事件的兴趣会创建一个SelectorKey，其中包含

laohyx / 0评论 2020-04-07

Python学习32天（socket、tcp协议）

　　守望了31天，终于等到了网络编程阶段，很多很多的基础原理知识，然后脑子懵了，进入新世界的感觉。　　腾讯作为服务端为你提供视频，你得下个腾讯视频客户端才能看它的视频） C/S架构与socket的关系：。　　Socket是应用层与TCP/IP协议族通信的中

Noseparte / 0评论 2020-03-26

Redis与Memcached区别？Redis线程模型？

　　redis 相比 memcached 来说，拥有更多的数据结构，能支持更丰富的数据操作。如果需要缓存能够支持更复杂的结构和操作， redis 会是不错的选择。　　由于 redis 只使用单核，而 memcached 可以使用多核，所以平均每一个核上 r

85590296 / 0评论 2020-03-25

Rabbitmq之socket descriptors

rabbitmq最大连接数,RabbitMQ的socket连接数是文件描述符的一个子集。也就是说，RabbitMQ能同时打开的最大连接数和最大文件句柄数都是受限于操作系统关于文件描述符数量的设置，两者是此消彼长的关系。初始时，可用socket描述符与可用f

lishijian / 0评论 2020-03-07

Linux UDP 单任务服务端和客户端实现

int socket_fd = socket; //创建UDP套接字，使用IPV4协议。struct sockaddr_in client_addr; //保存的客户端的ip协议和端口号信息。 , &client_addr_len); //接受客户

安得情怀似旧时 / 0评论 2020-03-04

python 网络编程-02 多进程的交互案例

server.py. import socket. from threading import Thread. def Server():. server = socket.socket(). server.bind(('127.0.0.1', 8000)

kikaylee / 0评论 2020-03-03

python tcp服务端-客户端

HOST = ‘0.0.0.0‘ # 主机ip. tcpS = socket # 创建socket对象。tcpS.setsockopt #加入socket配置，重用ip和端口。tcpS.bind # 绑定ip端口号。tcpS.listen # 设置最大链

x青年欢乐多 / 0评论 2020-03-02

python使用stuck 实现scoket编程实现文件传输

s = socket.socket # 定义socket类型。fileinfo_size = struct.calcsize#12s表示12个字符，l表示一个长整型数。filename_f = filename.decode.strip # C语言中’

chouliqingke / 0评论 2020-03-01

python基础 socketserver

socket只能处理一个client连接，如果需要并发处理多个链接则需要使用socketserver,下面是代码示例以及注释。#之前的socket只能同时处理一个客户端，当需要并发处理client发送的信息是需要使用到socketserver,下面创建一个

周小董 / 0评论 2020-02-26