src/server/process.c
static int swFactoryProcess_dispatch(swFactory *factory, swSendData *task)
{
    swServer *serv = (swServer *) factory->ptr;
    int fd = task->info.fd;

    int target_worker_id = swServer_worker_schedule(serv, fd, task);

    //discard the data packet.
    if (target_worker_id < 0)
    {
        /**
         * -1 : discard packet
         * [TODO] -2 : close connection
         */
        return SW_ERR;
    }

    if (swEventData_is_stream(task->info.type))
    {
        swConnection *conn = swServer_connection_get(serv, fd);
        if (conn == NULL || conn->active == 0)
        {
            swWarn("dispatch[type=%d] failed, connection#%d is not active.", task->info.type, fd);
            return SW_ERR;
        }
        //server active close, discard data.
        if (conn->closed)
        {
            //Connection has been clsoed by server
            if (!(task->info.type == SW_EVENT_CLOSE && conn->close_force))
            {
                return SW_OK;
            }
        }
        //converted fd to session_id
        task->info.fd = conn->session_id;
        task->info.from_fd = conn->from_fd;
    }

    swWorker *worker = swServer_get_worker(serv, target_worker_id);

    //without data
    if (task->data == NULL)
    {
        task->info.flags = 0;
        return swReactorThread_send2worker(serv, worker, &task->info, sizeof(task->info));
    }

    switch (task->info.type)
    {
    case SW_EVENT_TCP6:
    case SW_EVENT_TCP:
    case SW_EVENT_UNIX_STREAM:
    case SW_EVENT_UDP:
    case SW_EVENT_UDP6:
    case SW_EVENT_UNIX_DGRAM:
        worker->dispatch_count++;
        break;
    }

    uint32_t send_n = task->length;
    uint32_t offset = 0;
    swEventData buf;
    char *data = task->data;

    buf.info = task->info;

    if (send_n <= SW_IPC_BUFFER_SIZE)
    {
        buf.info.flags = 0;
        buf.info.len = send_n;
        memcpy(buf.data, data, buf.info.len);
        return swReactorThread_send2worker(serv, worker, &buf, sizeof(buf.info) + buf.info.len);
    }

    buf.info.flags = SW_EVENT_DATA_CHUNK;

    while (send_n > 0)
    {
        if (send_n > SW_IPC_BUFFER_SIZE)
        {
            buf.info.len = SW_IPC_BUFFER_SIZE;
        }
        else
        {
            buf.info.flags |= SW_EVENT_DATA_END;
            buf.info.len = send_n;
        }

        memcpy(buf.data, data + offset, buf.info.len);

        send_n -= buf.info.len;
        offset += buf.info.len;

        swTrace("dispatch, type=%d|len=%d", buf.info.type, buf.info.len);

        if (swReactorThread_send2worker(serv, worker, &buf, sizeof(buf.info) + buf.info.len) < 0)
        {
            return SW_ERR;
        }
    }

    return SW_OK;
}
/**
 * [ReactorThread] dispatch request to worker
 */
[ReactorThread] dispatch request to worker
[ ReactorThread ] dispatch request to worker
static int swFactoryProcess_dispatch(swFactory *factory, swSendData *task)
{
    swServer *serv = (swServer *) factory->ptr;
    int fd = task->info.fd;
    int target_worker_id = swServer_worker_schedule(serv, fd, task);
    if (target_worker_id < 0)
    {
        return SW_ERR;
    }
    if (swEventData_is_stream(task->info.type))
    {
        swConnection *conn = swServer_connection_get(serv, fd);
        if (conn == NULL || conn->active == 0)
        {
            swWarn("dispatch[type=%d] failed, connection#%d is not active.", task->info.type, fd);
            return SW_ERR;
        }
        if (conn->closed)
        {
            if (!(task->info.type == SW_EVENT_CLOSE && conn->close_force))
            {
                return SW_OK;
            }
        }
        task->info.fd = conn->session_id;
        task->info.from_fd = conn->from_fd;
    }
    swWorker *worker = swServer_get_worker(serv, target_worker_id);
    if (task->data == NULL)
    {
        task->info.flags = 0;
        return swReactorThread_send2worker(serv, worker, &task->info, sizeof(task->info));
    }
    switch (task->info.type)
    {
    case SW_EVENT_TCP6:
    case SW_EVENT_TCP:
    case SW_EVENT_UNIX_STREAM:
    case SW_EVENT_UDP:
    case SW_EVENT_UDP6:
    case SW_EVENT_UNIX_DGRAM:
        worker->dispatch_count++;
        break;
    }
    uint32_t send_n = task->length;
    uint32_t offset = 0;
    swEventData buf;
    char *data = task->data;
    buf.info = task->info;
    if (send_n <= SW_IPC_BUFFER_SIZE)
    {
        buf.info.flags = 0;
        buf.info.len = send_n;
        memcpy(buf.data, data, buf.info.len);
        return swReactorThread_send2worker(serv, worker, &buf, sizeof(buf.info) + buf.info.len);
    }
    buf.info.flags = SW_EVENT_DATA_CHUNK;
    while (send_n > 0)
    {
        if (send_n > SW_IPC_BUFFER_SIZE)
        {
            buf.info.len = SW_IPC_BUFFER_SIZE;
        }
        else
        {
            buf.info.flags |= SW_EVENT_DATA_END;
            buf.info.len = send_n;
        }
        memcpy(buf.data, data + offset, buf.info.len);
        send_n -= buf.info.len;
        offset += buf.info.len;
        swTrace("dispatch, type=%d|len=%d", buf.info.type, buf.info.len);
        if (swReactorThread_send2worker(serv, worker, &buf, sizeof(buf.info) + buf.info.len) < 0)
        {
            return SW_ERR;
        }
    }
    return SW_OK;
}
static int swFactoryProcess_dispatch ( swFactory * factory , swSendData * task ) { swServer * serv = ( swServer * ) factory -> ptr ; int fd = task -> info . fd ; int target_worker_id = swServer_worker_schedule ( serv , fd , task ) ; if ( target_worker_id < 0 ) { return SW_ERR ; } if ( swEventData_is_stream ( task -> info . type ) ) { swConnection * conn = swServer_connection_get ( serv , fd ) ; if ( conn == NULL || conn -> active == 0 ) { swWarn ( " " , task -> info . type , fd ) ; return SW_ERR ; } if ( conn -> closed ) { if ( ! ( task -> info . type == SW_EVENT_CLOSE && conn -> close_force ) ) { return SW_OK ; } } task -> info . fd = conn -> session_id ; task -> info . from_fd = conn -> from_fd ; } swWorker * worker = swServer_get_worker ( serv , target_worker_id ) ; if ( task -> data == NULL ) { task -> info . flags = 0 ; return swReactorThread_send2worker ( serv , worker , & task -> info , sizeof ( task -> info ) ) ; } switch ( task -> info . type ) { case SW_EVENT_TCP6 : case SW_EVENT_TCP : case SW_EVENT_UNIX_STREAM : case SW_EVENT_UDP : case SW_EVENT_UDP6 : case SW_EVENT_UNIX_DGRAM : worker -> dispatch_count ++ ; break ; } uint32_t send_n = task -> length ; uint32_t offset = 0 ; swEventData buf ; char * data = task -> data ; buf . info = task -> info ; if ( send_n <= SW_IPC_BUFFER_SIZE ) { buf . info . flags = 0 ; buf . info . len = send_n ; memcpy ( buf . data , data , buf . info . len ) ; return swReactorThread_send2worker ( serv , worker , & buf , sizeof ( buf . info ) + buf . info . len ) ; } buf . info . flags = SW_EVENT_DATA_CHUNK ; while ( send_n > 0 ) { if ( send_n > SW_IPC_BUFFER_SIZE ) { buf . info . len = SW_IPC_BUFFER_SIZE ; } else { buf . info . flags |= SW_EVENT_DATA_END ; buf . info . len = send_n ; } memcpy ( buf . data , data + offset , buf . info . len ) ; send_n -= buf . info . len ; offset += buf . info . len ; swTrace ( " " , buf . info . type , buf . info . len ) ; if ( swReactorThread_send2worker ( serv , worker , & buf , sizeof ( buf . info ) + buf . info . len ) < 0 ) { return SW_ERR ; } } return SW_OK ; }
[ReactorThread] dispatch request to worker
[ ReactorThread ] dispatch request to worker
//discard the data packet. /**
         * -1 : discard packet
         * [TODO] -2 : close connection
         */ //server active close, discard data. //Connection has been clsoed by server //converted fd to session_id //without data