src/network/ReactorThread.c
/**
 * send to client or append to out_buffer
 */
send to client or append to out_buffer
send to client or append to out_buffer
int swReactorThread_send(swSendData *_send)
{
    swServer *serv = SwooleG.serv;
    uint32_t session_id = _send->info.fd;
    void *_send_data = _send->data;
    uint32_t _send_length = _send->length;
    swConnection *conn = swServer_connection_verify(serv, session_id);
    if (!conn)
    {
        if (_send->info.type == SW_EVENT_TCP)
        {
            swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SESSION_NOT_EXIST, "send %d byte failed, session#%d does not exist.", _send_length, session_id);
        }
        else
        {
            swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SESSION_NOT_EXIST, "send event$[%d] failed, session#%d does not exist.", _send->info.type, session_id);
        }
        return SW_ERR;
    }
    int fd = conn->fd;
    swReactor *reactor;
    if (serv->factory_mode == SW_MODE_SINGLE)
    {
        reactor = &(serv->reactor_threads[0].reactor);
    }
    else
    {
        reactor = &(serv->reactor_threads[conn->from_id].reactor);
    }
    if (_send->info.type == SW_EVENT_CLOSE && conn->close_reset)
    {
        goto close_fd;
    }
    if (swBuffer_empty(conn->out_buffer))
    {
        if (_send->info.type == SW_EVENT_CLOSE)
        {
            close_fd:
            reactor->close(reactor, fd);
            return SW_OK;
        }
#ifdef SW_REACTOR_SYNC_SEND
        if (_send->info.type != SW_EVENT_SENDFILE)
        {
            if (!conn->direct_send)
            {
                goto buffer_send;
            }
            int n;
            direct_send:
            n = swConnection_send(conn, _send_data, _send_length, 0);
            if (n == _send_length)
            {
                return SW_OK;
            }
            else if (n > 0)
            {
                _send_data += n;
                _send_length -= n;
                goto buffer_send;
            }
            else if (errno == EINTR)
            {
                goto direct_send;
            }
            else
            {
                goto buffer_send;
            }
        }
#endif
        else
        {
#ifdef SW_REACTOR_SYNC_SEND
            buffer_send:
#endif
            if (!conn->out_buffer)
            {
                conn->out_buffer = swBuffer_new(SW_BUFFER_SIZE);
                if (conn->out_buffer == NULL)
                {
                    return SW_ERR;
                }
            }
        }
    }
    swBuffer_trunk *trunk;
    if (_send->info.type == SW_EVENT_CLOSE)
    {
        trunk = swBuffer_new_trunk(conn->out_buffer, SW_CHUNK_CLOSE, 0);
        trunk->store.data.val1 = _send->info.type;
    }
    else if (_send->info.type == SW_EVENT_SENDFILE)
    {
        swConnection_sendfile(conn, _send_data);
    }
    else
    {
        if (conn->removed)
        {
            swWarn("connection#%d is closed by client.", fd);
            return SW_ERR;
        }
        if (conn->out_buffer->length >= serv->buffer_output_size)
        {
            swoole_error_log(SW_LOG_WARNING, SW_ERROR_OUTPUT_BUFFER_OVERFLOW, "connection#%d output buffer overflow.", fd);
            conn->overflow = 1;
        }
        int _length = _send_length;
        void* _pos = _send_data;
        int _n;
        while (_length > 0)
        {
            _n = _length >= SW_BUFFER_SIZE_BIG ? SW_BUFFER_SIZE_BIG : _length;
            swBuffer_append(conn->out_buffer, _pos, _n);
            _pos += _n;
            _length -= _n;
        }
    }
    if (reactor->set(reactor, fd, SW_EVENT_TCP | SW_EVENT_WRITE | SW_EVENT_READ) < 0
            && (errno == EBADF || errno == ENOENT))
    {
        goto close_fd;
    }
    return SW_OK;
}
int swReactorThread_send ( swSendData * _send ) { swServer * serv = SwooleG . serv ; uint32_t session_id = _send -> info . fd ; void * _send_data = _send -> data ; uint32_t _send_length = _send -> length ; swConnection * conn = swServer_connection_verify ( serv , session_id ) ; if ( ! conn ) { if ( _send -> info . type == SW_EVENT_TCP ) { swoole_error_log ( SW_LOG_NOTICE , SW_ERROR_SESSION_NOT_EXIST , " " , _send_length , session_id ) ; } else { swoole_error_log ( SW_LOG_NOTICE , SW_ERROR_SESSION_NOT_EXIST , " " , _send -> info . type , session_id ) ; } return SW_ERR ; } int fd = conn -> fd ; swReactor * reactor ; if ( serv -> factory_mode == SW_MODE_SINGLE ) { reactor = & ( serv -> reactor_threads [ 0 ] . reactor ) ; } else { reactor = & ( serv -> reactor_threads [ conn -> from_id ] . reactor ) ; } if ( _send -> info . type == SW_EVENT_CLOSE && conn -> close_reset ) { goto close_fd ; } if ( swBuffer_empty ( conn -> out_buffer ) ) { if ( _send -> info . type == SW_EVENT_CLOSE ) { close_fd : reactor -> close ( reactor , fd ) ; return SW_OK ; } #ifdef SW_REACTOR_SYNC_SEND if ( _send -> info . type != SW_EVENT_SENDFILE ) { if ( ! conn -> direct_send ) { goto buffer_send ; } int n ; direct_send : n = swConnection_send ( conn , _send_data , _send_length , 0 ) ; if ( n == _send_length ) { return SW_OK ; } else if ( n > 0 ) { _send_data += n ; _send_length -= n ; goto buffer_send ; } else if ( errno == EINTR ) { goto direct_send ; } else { goto buffer_send ; } } #endif else { #ifdef SW_REACTOR_SYNC_SEND buffer_send : #endif if ( ! conn -> out_buffer ) { conn -> out_buffer = swBuffer_new ( SW_BUFFER_SIZE ) ; if ( conn -> out_buffer == NULL ) { return SW_ERR ; } } } } swBuffer_trunk * trunk ; if ( _send -> info . type == SW_EVENT_CLOSE ) { trunk = swBuffer_new_trunk ( conn -> out_buffer , SW_CHUNK_CLOSE , 0 ) ; trunk -> store . data . val1 = _send -> info . type ; } else if ( _send -> info . type == SW_EVENT_SENDFILE ) { swConnection_sendfile ( conn , _send_data ) ; } else { if ( conn -> removed ) { swWarn ( " " , fd ) ; return SW_ERR ; } if ( conn -> out_buffer -> length >= serv -> buffer_output_size ) { swoole_error_log ( SW_LOG_WARNING , SW_ERROR_OUTPUT_BUFFER_OVERFLOW , " " , fd ) ; conn -> overflow = 1 ; } int _length = _send_length ; void * _pos = _send_data ; int _n ; while ( _length > 0 ) { _n = _length >= SW_BUFFER_SIZE_BIG ? SW_BUFFER_SIZE_BIG : _length ; swBuffer_append ( conn -> out_buffer , _pos , _n ) ; _pos += _n ; _length -= _n ; } } if ( reactor -> set ( reactor , fd , SW_EVENT_TCP | SW_EVENT_WRITE | SW_EVENT_READ ) < 0 && ( errno == EBADF || errno == ENOENT ) ) { goto close_fd ; } return SW_OK ; }
send to client or append to out_buffer
send to client or append to out_buffer
/**
     * Reset send buffer, Immediately close the connection.
     */ /**
         * close connection.
         */ //Direct send //buffer send //close connection //sendfile to client //send data //connection is closed //connection output buffer overflow //buffer enQueue //listen EPOLLOUT event