摘要:新建可以看到,自動采用包長檢測的方法該函數主要功能是設置各種回調函數值得注意的是第三個參數代表是否異步。發送數據函數并不是直接發送數據,而是將數據存儲在,等著寫事件就緒之后調用發送數據。
swReactorThread_dispatch 發送數據
reactor 線程會通過 swReactorThread_dispatch 發送數據,當采用 stream 發送數據的時候,會調用 swStream_new 新建 stream,利用 swStream_send 發送數據。
</>復制代碼
int swReactorThread_dispatch(swConnection *conn, char *data, uint32_t length)
{
...
if (serv->dispatch_mode == SW_DISPATCH_STREAM)
{
swStream *stream = swStream_new(serv->stream_socket, 0, SW_SOCK_UNIX_STREAM);
if (stream == NULL)
{
return SW_ERR;
}
stream->response = swReactorThread_onStreamResponse;
stream->session_id = conn->session_id;
swListenPort *port = swServer_get_port(serv, conn->fd);
swStream_set_max_length(stream, port->protocol.package_max_length);
task.data.info.fd = conn->session_id;
task.data.info.type = SW_EVENT_PACKAGE_END;
task.data.info.len = 0;
if (swStream_send(stream, (char*) &task.data.info, sizeof(task.data.info)) < 0)
{
return SW_ERR;
}
if (swStream_send(stream, data, length) < 0)
{
stream->cancel = 1;
return SW_ERR;
}
return SW_OK;
}
...
}
swStream_new 新建 stream
可以看到,stream 自動采用包長檢測的方法
該函數主要功能是設置各種回調函數
值得注意的是 swClient_create 第三個參數代表是否異步。在這里設置的是 1,也就是說,無論 connect 還是 send 都是異步。
</>復制代碼
typedef struct _swStream
{
swString *buffer;
uint32_t session_id;
uint8_t cancel;
void (*response)(struct _swStream *stream, char *data, uint32_t length);
swClient client;
} swStream;
swStream* swStream_new(char *dst_host, int dst_port, int type)
{
swStream *stream = (swStream*) sw_malloc(sizeof(swStream));
bzero(stream, sizeof(swStream));
swClient *cli = &stream->client;
if (swClient_create(cli, type, 1) < 0)
{
swStream_free(stream);
return NULL;
}
cli->onConnect = swStream_onConnect;
cli->onReceive = swStream_onReceive;
cli->onError = swStream_onError;
cli->onClose = swStream_onClose;
cli->object = stream;
cli->open_length_check = 1;
swStream_set_protocol(&cli->protocol);
if (cli->connect(cli, dst_host, dst_port, -1, 0) < 0)
{
swSysError("failed to connect to [%s:%d].", dst_host, dst_port);
swStream_free(stream);
return NULL;
}
else
{
return stream;
}
}
void swStream_set_protocol(swProtocol *protocol)
{
protocol->get_package_length = swProtocol_get_package_length;
protocol->package_length_size = 4;
protocol->package_length_type = "N";
protocol->package_body_offset = 4;
protocol->package_length_offset = 0;
}
swStream_onConnect 連接回調函數
swStream_onConnect 不僅是連接成功的回調函數,還是每次 onWrite 寫事件的回調函數,因此每次都需要調用 cli->send 函數,發送存儲在 stream->buffer 數據。值得注意的是,每次發送數據,都要將數據長度存放在 buffer 的頭部,否則包長檢測會失敗。
</>復制代碼
static void swStream_onConnect(swClient *cli)
{
swStream *stream = (swStream*) cli->object;
if (stream->cancel)
{
cli->close(cli);
}
*((uint32_t *) stream->buffer->str) = ntohl(stream->buffer->length - 4);
if (cli->send(cli, stream->buffer->str, stream->buffer->length, 0) < 0)
{
cli->close(cli);
}
else
{
swString_free(stream->buffer);
stream->buffer = NULL;
}
}
swStream_send 發送數據
swStream_send 函數并不是直接發送數據,而是將數據存儲在 stream->buffer,等著寫事件就緒之后調用 swStream_onConnect 發送數據。值得注意的是,每次新建 buffer 的時候,要預留 4 個字節來存儲 buffer 的數據長度
</>復制代碼
int swStream_send(swStream *stream, char *data, size_t length)
{
if (stream->buffer == NULL)
{
stream->buffer = swString_new(swoole_size_align(length + 4, SwooleG.pagesize));
if (stream->buffer == NULL)
{
return SW_ERR;
}
stream->buffer->length = 4;
}
if (swString_append_ptr(stream->buffer, data, length) < 0)
{
return SW_ERR;
}
return SW_OK;
}
swStream_onReceive 函數
swStream_onReceive 函數是 stream 讀事件就緒的回調函數,worker 進程發送給客戶端的數據將會發送到本函數。如果 length 為 4,說明 worker 只發送了一個 length 的空數據包,代表著 worker 進程已消費完畢,這時我們可以關閉 stream。
</>復制代碼
static void swStream_onReceive(swClient *cli, char *data, uint32_t length)
{
swStream *stream = (swStream*) cli->object;
if (length == 4)
{
cli->socket->close_wait = 1;
}
else
{
stream->response(stream, data + 4, length - 4);
}
}
static void swReactorThread_onStreamResponse(swStream *stream, char *data, uint32_t length)
{
swSendData response;
swConnection *conn = swServer_connection_verify(SwooleG.serv, stream->session_id);
if (!conn)
{
swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SESSION_NOT_EXIST, "connection[fd=%d] does not exists.", stream->session_id);
return;
}
response.info.fd = conn->session_id;
response.info.type = SW_EVENT_TCP;
response.info.len = 0;
response.length = length;
response.data = data;
swReactorThread_send(&response);
}
swWorker_onStreamAccept 接受連接請求
接受請求和主進程的 reactor 接受連接大致一致,略有不同的是 conn->socket_type 設置為了 SW_SOCK_UNIX_STREAM
</>復制代碼
static int swWorker_onStreamAccept(swReactor *reactor, swEvent *event)
{
int fd = 0;
swSocketAddress client_addr;
socklen_t client_addrlen = sizeof(client_addr);
#ifdef HAVE_ACCEPT4
fd = accept4(event->fd, (struct sockaddr *) &client_addr, &client_addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC);
#else
fd = accept(event->fd, (struct sockaddr *) &client_addr, &client_addrlen);
#endif
if (fd < 0)
{
switch (errno)
{
case EINTR:
case EAGAIN:
return SW_OK;
default:
swoole_error_log(SW_LOG_ERROR, SW_ERROR_SYSTEM_CALL_FAIL, "accept() failed. Error: %s[%d]", strerror(errno),
errno);
return SW_OK;
}
}
#ifndef HAVE_ACCEPT4
else
{
swoole_fcntl_set_option(fd, 1, 1);
}
#endif
swConnection *conn = swReactor_get(reactor, fd);
bzero(conn, sizeof(swConnection));
conn->fd = fd;
conn->active = 1;
conn->socket_type = SW_SOCK_UNIX_STREAM;
memcpy(&conn->info.addr, &client_addr, sizeof(client_addr));
if (reactor->add(reactor, fd, SW_FD_STREAM | SW_EVENT_READ) < 0)
{
return SW_ERR;
}
return SW_OK;
}
swWorker_onStreamRead 讀取數據
swWorker_onStreamRead 讀取數據核心是調用 swProtocol_recv_check_length 函數收取數據放入 serv->buffer_pool 單鏈表中,swProtocol_recv_check_length 函數我們在 reactor 線程的事件循環中已經了解了,我們這里不再重復,我們知道,該函數獲取數據之后,會調用 onPackage 函數,也就是 swWorker_onStreamPackage 函數
</>復制代碼
void swStream_set_protocol(swProtocol *protocol)
{
protocol->get_package_length = swProtocol_get_package_length;
protocol->package_length_size = 4;
protocol->package_length_type = "N";
protocol->package_body_offset = 4;
protocol->package_length_offset = 0;
}
static int swWorker_onStreamRead(swReactor *reactor, swEvent *event)
{
swConnection *conn = event->socket;
swServer *serv = SwooleG.serv;
swProtocol *protocol = &serv->stream_protocol;
swString *buffer;
if (!event->socket->recv_buffer)
{
buffer = swLinkedList_shift(serv->buffer_pool);
if (buffer == NULL)
{
buffer = swString_new(8192);
if (!buffer)
{
return SW_ERR;
}
}
event->socket->recv_buffer = buffer;
}
else
{
buffer = event->socket->recv_buffer;
}
if (swProtocol_recv_check_length(protocol, conn, buffer) < 0)
{
swWorker_onStreamClose(reactor, event);
}
return SW_OK;
}
swWorker_onStreamPackage 函數
swWorker_onStreamPackage 函數用于將數據包投送到 swWorker_onTask 函數進行消費。消費完畢會發送一個只含長度 0 的數據包,告知 reactor worker 已經結束。
</>復制代碼
static int swWorker_onStreamPackage(swConnection *conn, char *data, uint32_t length)
{
swServer *serv = SwooleG.serv;
swEventData *task = (swEventData *) (data + 4);
serv->last_stream_fd = conn->fd;
swString *package = swWorker_get_buffer(serv, task->info.from_id);
uint32_t data_length = length - sizeof(task->info) - 4;
//merge data to package buffer
swString_append_ptr(package, data + sizeof(task->info) + 4, data_length);
swWorker_onTask(&serv->factory, task);
int _end = htonl(0);
SwooleG.main_reactor->write(SwooleG.main_reactor, conn->fd, (void *) &_end, sizeof(_end));
return SW_OK;
}
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/29277.html
摘要:當此時的套接字不可寫的時候,會自動放入緩沖區中。當大于高水線時,會自動調用回調函數。寫就緒狀態當監控到套接字進入了寫就緒狀態時,就會調用函數。如果為,說明此時異步客戶端雖然建立了連接,但是還沒有調用回調函數,因此這時要調用函數。 前言 上一章我們說了客戶端的連接 connect,對于同步客戶端來說,連接已經建立成功;但是對于異步客戶端來說,此時可能還在進行 DNS 的解析,on...
摘要:函數事件循環在事件循環時,如果使用的是消息隊列,那么就不斷的調用從消息隊列中取出數據。獲取后的數據調用回調函數消費消息之后,向中發送空數據,告知進程已消費,并且關閉新連接。 swManager_start 創建進程流程 task_worker 進程的創建可以分為三個步驟:swServer_create_task_worker 申請所需的內存、swTaskWorker_init 初始化...
摘要:如果在調用之前我們設置了,但是不在第二個進程啟動前這個套接字,那么第二個進程仍然會在調用函數的時候出錯。 前言 本節主要介紹 server 模塊進行初始化的代碼,關于初始化過程中,各個屬性的意義,可以參考官方文檔: SERVER 配置選項 關于初始化過程中,用于監聽的 socket 綁定問題,可以參考: UNP 學習筆記——基本 TCP 套接字編程 UNP 學習筆記——套接字選項 構造...
前言 作為一個網絡框架,最為核心的就是消息的接受與發送。高效的 reactor 模式一直是眾多網絡框架的首要選擇,本節主要講解 swoole 中的 reactor 模塊。 UNP 學習筆記——IO 復用 Reactor 的數據結構 Reactor 的數據結構比較復雜,首先 object 是具體 Reactor 對象的首地址,ptr 是擁有 Reactor 對象的類的指針, event_nu...
摘要:兩個函數是可選回調函數。附帶了一組可信任證書。應該注意的是,驗證失敗并不意味著連接不能使用。在對證書進行驗證時,有一些安全性檢查并沒有執行,包括證書的失效檢查和對證書中通用名的有效性驗證。 前言 swoole_client 提供了 tcp/udp socket 的客戶端的封裝代碼,使用時僅需 new swoole_client 即可。 swoole 的 socket client 對比...
閱讀 2816·2021-10-26 09:48
閱讀 1681·2021-09-22 15:22
閱讀 4058·2021-09-22 15:05
閱讀 619·2021-09-06 15:02
閱讀 2610·2019-08-30 15:52
閱讀 2116·2019-08-29 18:38
閱讀 2760·2019-08-28 18:05
閱讀 2335·2019-08-26 13:55