Swoole 源碼分析——Client模塊之Send

caozhijian 發(fā)布于2019-07-01 10:48 / 1956人閱讀

摘要：當此時的套接字不可寫的時候，會自動放入緩沖區(qū)中。當大于高水線時，會自動調(diào)用回調(diào)函數(shù)。寫就緒狀態(tài)當監(jiān)控到套接字進入了寫就緒狀態(tài)時，就會調(diào)用函數(shù)。如果為，說明此時異步客戶端雖然建立了連接，但是還沒有調(diào)用回調(diào)函數(shù)，因此這時要調(diào)用函數(shù)。

前言

上一章我們說了客戶端的連接 connect，對于同步客戶端來說，連接已經(jīng)建立成功；但是對于異步客戶端來說，此時可能還在進行 DNS 的解析，onConnect 回調(diào)函數(shù)還未執(zhí)行。

本節(jié)中，我們將繼續(xù)說明客戶端發(fā)送數(shù)據(jù)的流程，同時我們可以看到 TCP 異步客戶端執(zhí)行 onConnect 回調(diào)函數(shù)的過程。

send 入口

本入口函數(shù)邏輯非常簡單，從 PHP 函數(shù)中獲取數(shù)據(jù) data，然后調(diào)用 connect 函數(shù)。

static PHP_METHOD(swoole_client, send)
{
    char *data;
    zend_size_t data_len;
    zend_long flags = 0;

#ifdef FAST_ZPP
    ZEND_PARSE_PARAMETERS_START(1, 2)
        Z_PARAM_STRING(data, data_len)
        Z_PARAM_OPTIONAL
        Z_PARAM_LONG(flags)
    ZEND_PARSE_PARAMETERS_END();
#else
    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|l", &data, &data_len, &flags) == FAILURE)
    {
        return;
    }
#endif

    swClient *cli = client_get_ptr(getThis() TSRMLS_CC);

    //clear errno
    SwooleG.error = 0;
    int ret = cli->send(cli, data, data_len, flags);
    if (ret < 0)
    {
        swoole_php_sys_error(E_WARNING, "failed to send(%d) %zd bytes.", cli->socket->fd, data_len);
        zend_update_property_long(swoole_client_class_entry_ptr, getThis(), SW_STRL("errCode")-1, SwooleG.error TSRMLS_CC);
        RETVAL_FALSE;
    }
    else
    {
        RETURN_LONG(ret);
    }
}

swClient_tcp_send_sync 同步 TCP 客戶端

對于同步客戶端來說，發(fā)送數(shù)據(jù)是通過 swConnection_send 函數(shù)來進行阻塞式調(diào)用 send，當返回的錯誤是 EAGAIN 的時候調(diào)用 swSocket_wait 等待 1s。

static int swClient_tcp_send_sync(swClient *cli, char *data, int length, int flags)
{
    int written = 0;
    int n;

    assert(length > 0);
    assert(data != NULL);

    while (written < length)
    {
        n = swConnection_send(cli->socket, data, length - written, flags);
        if (n < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            else if (errno == EAGAIN)
            {
                swSocket_wait(cli->socket->fd, 1000, SW_EVENT_WRITE);
                continue;
            }
            else
            {
                SwooleG.error = errno;
                return SW_ERR;
            }
        }
        written += n;
        data += n;
    }
    return written;
}

swClient_tcp_send_async 異步 TCP 客戶端

由于異步客戶端已經(jīng)設置為非阻塞，并且加入了 reactor 的監(jiān)控，因此發(fā)送數(shù)據(jù)只需要 reactor->write 函數(shù)即可。當此時的套接字不可寫的時候，會自動放入 out_buff 緩沖區(qū)中。

當 out_buffer 大于高水線時，會自動調(diào)用 onBufferFull 回調(diào)函數(shù)。

static int swClient_tcp_send_async(swClient *cli, char *data, int length, int flags)
{
    int n = length;
    if (cli->reactor->write(cli->reactor, cli->socket->fd, data, length) < 0)
    {
        if (SwooleG.error == SW_ERROR_OUTPUT_BUFFER_OVERFLOW)
        {
            n = -1;
            cli->socket->high_watermark = 1;
        }
        else
        {
            return SW_ERR;
        }
    }
    if (cli->onBufferFull && cli->socket->out_buffer && cli->socket->high_watermark == 0
            && cli->socket->out_buffer->length >= cli->buffer_high_watermark)
    {
        cli->socket->high_watermark = 1;
        cli->onBufferFull(cli);
    }
    return n;
}

swClient_udp_send UDP 客戶端

對于 UDP 客戶端來說，如果 Socket 緩存區(qū)塞滿，并不會像 TCP 進行等待 reactor 可寫狀態(tài)，而是直接返回了結(jié)果。對于異步客戶端來說，僅僅是非阻塞調(diào)用 sendto 函數(shù)。

static int swClient_udp_send(swClient *cli, char *data, int len, int flags)
{
    int n;
    n = sendto(cli->socket->fd, data, len, 0, (struct sockaddr *) &cli->server_addr.addr, cli->server_addr.len);
    if (n < 0 || n < len)
    {
        return SW_ERR;
    }
    else
    {
        return n;
    }
}

sendto UDP 客戶端

類似于 send 函數(shù)，sendto 函數(shù)專門針對 UDP 客戶端，與 send 函數(shù)不同的是，sendto 函數(shù)在底層套接字緩沖區(qū)塞滿的時候，會調(diào)用 swSocket_wait 進行阻塞等待。

static PHP_METHOD(swoole_client, sendto)
{
    char* ip;
    zend_size_t ip_len;
    long port;
    char *data;
    zend_size_t len;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sls", &ip, &ip_len, &port, &data, &len) == FAILURE)
    {
        return;
    }

    swClient *cli = (swClient *) swoole_get_object(getThis());

    int ret;
    if (cli->type == SW_SOCK_UDP)
    {
        ret = swSocket_udp_sendto(cli->socket->fd, ip, port, data, len);
    }
    else if (cli->type == SW_SOCK_UDP6)
    {
        ret = swSocket_udp_sendto6(cli->socket->fd, ip, port, data, len);
    }
    else
    {
        swoole_php_fatal_error(E_WARNING, "only supports SWOOLE_SOCK_UDP or SWOOLE_SOCK_UDP6.");
        RETURN_FALSE;
    }
    SW_CHECK_RETURN(ret);
}

int swSocket_udp_sendto(int server_sock, char *dst_ip, int dst_port, char *data, uint32_t len)
{
    struct sockaddr_in addr;
    if (inet_aton(dst_ip, &addr.sin_addr) == 0)
    {
        swWarn("ip[%s] is invalid.", dst_ip);
        return SW_ERR;
    }
    addr.sin_family = AF_INET;
    addr.sin_port = htons(dst_port);
    return swSocket_sendto_blocking(server_sock, data, len, 0, (struct sockaddr *) &addr, sizeof(addr));
}

int swSocket_udp_sendto6(int server_sock, char *dst_ip, int dst_port, char *data, uint32_t len)
{
    struct sockaddr_in6 addr;
    bzero(&addr, sizeof(addr));
    if (inet_pton(AF_INET6, dst_ip, &addr.sin6_addr) < 0)
    {
        swWarn("ip[%s] is invalid.", dst_ip);
        return SW_ERR;
    }
    addr.sin6_port = (uint16_t) htons(dst_port);
    addr.sin6_family = AF_INET6;
    return swSocket_sendto_blocking(server_sock, data, len, 0, (struct sockaddr *) &addr, sizeof(addr));
}

int swSocket_sendto_blocking(int fd, void *__buf, size_t __n, int flag, struct sockaddr *__addr, socklen_t __addr_len)
{
    int n = 0;

    while (1)
    {
        n = sendto(fd, __buf, __n, flag, __addr, __addr_len);
        if (n >= 0)
        {
            break;
        }
        else
        {
            if (errno == EINTR)
            {
                continue;
            }
            else if (swConnection_error(errno) == SW_WAIT)
            {
                swSocket_wait(fd, 1000, SW_EVENT_WRITE);
                continue;
            }
            else
            {
                break;
            }
        }
    }

    return n;
}

swClient_onWrite 寫就緒狀態(tài)

當 reactor 監(jiān)控到套接字進入了寫就緒狀態(tài)時，就會調(diào)用 swClient_onWrite 函數(shù)。

從上一章我們知道，異步客戶端建立連接過程中 swoole 調(diào)用了 connect 函數(shù)，該返回 0，或者返回錯誤碼 EINPROGRESS 都會對寫就緒進行監(jiān)控。無論哪種情況，swClient_onWrite 被調(diào)用就說明此時連接已經(jīng)建立成功，三次握手已經(jīng)完成，但是 cli->socket->active 還是 0。

如果 cli->socket->active 為 0，說明此時異步客戶端雖然建立了連接，但是還沒有調(diào)用 onConnect 回調(diào)函數(shù)，因此這時要調(diào)用 execute_onConnect 函數(shù)。如果使用了 SSL 隧道加密，還要進行 SSL 握手，并且設置 _socket->ssl_state = SW_SSL_STATE_WAIT_STREAM。

當 active 為 1 的時候，就可以調(diào)用 swReactor_onWrite 來發(fā)送數(shù)據(jù)。

static int swClient_onWrite(swReactor *reactor, swEvent *event)
{
    swClient *cli = event->socket->object;
    swConnection *_socket = cli->socket;

    if (cli->socket->active)
    {
#ifdef SW_USE_OPENSSL
        if (cli->open_ssl && _socket->ssl_state == SW_SSL_STATE_WAIT_STREAM)
        {
            if (swClient_ssl_handshake(cli) < 0)
            {
                goto connect_fail;
            }
            else if (_socket->ssl_state == SW_SSL_STATE_READY)
            {
                goto connect_success;
            }
            else
            {
                if (_socket->ssl_want_read)
                {
                    cli->reactor->set(cli->reactor, event->fd, SW_FD_STREAM_CLIENT | SW_EVENT_READ);
                }
                return SW_OK;
            }
        }
#endif
        if (swReactor_onWrite(cli->reactor, event) < 0)
        {
            return SW_ERR;
        }
        if (cli->onBufferEmpty && _socket->high_watermark && _socket->out_buffer->length <= cli->buffer_low_watermark)
        {
            _socket->high_watermark = 0;
            cli->onBufferEmpty(cli);
        }
        return SW_OK;
    }

    socklen_t len = sizeof(SwooleG.error);
    if (getsockopt(event->fd, SOL_SOCKET, SO_ERROR, &SwooleG.error, &len) < 0)
    {
        swWarn("getsockopt(%d) failed. Error: %s[%d]", event->fd, strerror(errno), errno);
        return SW_ERR;
    }

    //success
    if (SwooleG.error == 0)
    {
        //listen read event
        cli->reactor->set(cli->reactor, event->fd, SW_FD_STREAM_CLIENT | SW_EVENT_READ);
        //connected
        _socket->active = 1;
        
#ifdef SW_USE_OPENSSL
        if (cli->open_ssl)
        {
            if (swClient_enable_ssl_encrypt(cli) < 0)
            {
                goto connect_fail;
            }
            if (swClient_ssl_handshake(cli) < 0)
            {
                goto connect_fail;
            }
            else
            {
                _socket->ssl_state = SW_SSL_STATE_WAIT_STREAM;
            }
            return SW_OK;
        }
        connect_success:
#endif
        if (cli->onConnect)
        {
            execute_onConnect(cli);
        }
    }
    else
    {
#ifdef SW_USE_OPENSSL
        connect_fail:
#endif
        _socket->active = 0;
        cli->close(cli);
        if (cli->onError)
        {
            cli->onError(cli);
        }
    }

    return SW_OK;
}

static sw_inline void execute_onConnect(swClient *cli)
{
    if (cli->timer)
    {
        swTimer_del(&SwooleG.timer, cli->timer);
        cli->timer = NULL;
    }
    cli->onConnect(cli);
}

client_onConnect

static void client_onConnect(swClient *cli)
{
    zval *zobject = (zval *) cli->object;
#ifdef SW_USE_OPENSSL
    if (cli->ssl_wait_handshake)
    {
        client_execute_callback(zobject, SW_CLIENT_CB_onSSLReady);
    }
    else
#endif
    if (!cli->redirect)
    {
        client_execute_callback(zobject, SW_CLIENT_CB_onConnect);
    }
    else
    {
        client_callback *cb = (client_callback *) swoole_get_property(zobject, 0);
        if (!cb || !cb->onReceive)
        {
            swoole_php_fatal_error(E_ERROR, "has no "onReceive" callback function.");
        }
    }
}

云服務器 GPU云服務器 POP3Client模塊源碼中文說明之源碼編譯模塊 linux 模塊源碼

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

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

發(fā)表評論

登陸后可評論

0條評論

caozhijian

男|高級講師

我要關(guān)注我要私信

TA的文章

UAT測試的帶隊經(jīng)驗

閱讀 3057·2021-10-13 09:39
軟件開發(fā)與生命周期

閱讀 2720·2021-09-27 13:34
給自己的博客網(wǎng)站加上酷炫的初音未來音樂游戲？

閱讀 2086·2019-08-30 15:55
前端面試題

閱讀 3281·2019-08-30 15:43
CSS、Sass、Scss，以及sass和scss的區(qū)別

閱讀 3675·2019-08-30 11:16
npm run dev 和 npm run build 解析（轉(zhuǎn)載）

閱讀 1787·2019-08-26 18:28
5、TypeScript 之不可以重復聲明類型

閱讀 1327·2019-08-26 13:56
新上課程推薦：TypeScript完全解讀(總26課時）

閱讀 948·2019-08-26 13:35

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

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

Swoole 源碼分析——Client模塊之Send

相關(guān)文章

Swoole 源碼分析——Server模塊之Stream 模式

Swoole 源碼分析——Reactor模塊之ReactorBase

Swoole 源碼分析——Client模塊之Connect

Swoole 源碼分析——Server模塊之OpenSSL(下)

Swoole 源碼分析——Server模塊之ReactorThread事件循環(huán)（下）

發(fā)表評論

0條評論

caozhijian

男|高級講師

TA的文章

UAT測試的帶隊經(jīng)驗

軟件開發(fā)與生命周期

給自己的博客網(wǎng)站加上酷炫的初音未來音樂游戲？

前端面試題

CSS、Sass、Scss，以及sass和scss的區(qū)別

npm run dev 和 npm run build 解析（轉(zhuǎn)載）

5、TypeScript 之不可以重復聲明類型

新上課程推薦：TypeScript完全解讀(總26課時）

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

Swoole 源碼分析——Client模塊之Send

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！