zan-extension/src/workers/zanNetWorker.c
//receive data from worker process pipe
receive data from worker process pipe
receive data from worker process pipe
static int zanNetworker_onPipeReceive(swReactor *reactor, swEvent *ev)
{
    int n;
    swEventData resp;
    swSendData _send;
    swPackage_response pkg_resp;
    zanWorker *worker;
    while (1)
    {
        n = read(ev->fd, &resp, sizeof(resp));
        if (n > 0)
        {
            memcpy(&_send.info, &resp.info, sizeof(resp.info));
            if (_send.info.type == SW_EVENT_DENY_REQUEST) {
                int target_worker_id = _send.info.worker_id;
                ServerGS->event_workers.workers[target_worker_id].deny_request = 1;
                zanTrace("set worker deny_request, [dst_work_id=%d]", target_worker_id);
                return ZAN_OK;
            } else if(_send.info.type == SW_EVENT_DENY_EXIT) {
                int target_worker_id = _send.info.worker_id;
                ServerGS->event_workers.workers[target_worker_id].deny_request = 0;
                zanTrace("set worker accept request, [work_id=%d]", target_worker_id);
                return ZAN_OK;
            }
            if (_send.info.from_fd == ZAN_RESPONSE_SMALL)
            {
                zanTrace("small response, data_type=%d, session_id=%d, from_worker_id=%d, pipe_fd=%d",
                          _send.info.from_fd, _send.info.fd, _send.info.worker_id, ev->fd);
                _send.data = resp.data;
                _send.length = resp.info.len;
                zanNetworker_send(&_send);
                return ZAN_OK;
            }
            else
            {
                zanTrace("big response, data_type=%d, session_id=%d, from_worker_id=%d, pipe_fd=%d",
                          _send.info.from_fd, _send.info.fd, _send.info.worker_id, ev->fd);
                memcpy(&pkg_resp, resp.data, sizeof(pkg_resp));
                worker = zanServer_get_worker(ServerG.serv, pkg_resp.worker_id);
                _send.data = worker->send_shm;
                _send.length = pkg_resp.length;
                zanNetworker_send(&_send);
                worker->lock.unlock(&worker->lock);
                return ZAN_OK;
            }
        }
        else if (errno == EAGAIN)
        {
            return ZAN_OK;
        }
        else if (errno == EINTR)
        {
            zanWarn("read(worker_pipe) EINTR, n=%d, errno:%d:%s", n, errno, strerror(errno));
            continue;
        }
        else
        {
            zanWarn("read(worker_pipe) failed, n=%d, errno:%d:%s", n, errno, strerror(errno));
            return ZAN_ERR;
        }
    }
}
static int zanNetworker_onPipeReceive ( swReactor * reactor , swEvent * ev ) { int n ; swEventData resp ; swSendData _send ; swPackage_response pkg_resp ; zanWorker * worker ; while ( 1 ) { n = read ( ev -> fd , & resp , sizeof ( resp ) ) ; if ( n > 0 ) { memcpy ( & _send . info , & resp . info , sizeof ( resp . info ) ) ; if ( _send . info . type == SW_EVENT_DENY_REQUEST ) { int target_worker_id = _send . info . worker_id ; ServerGS -> event_workers . workers [ target_worker_id ] . deny_request = 1 ; zanTrace ( " " , target_worker_id ) ; return ZAN_OK ; } else if ( _send . info . type == SW_EVENT_DENY_EXIT ) { int target_worker_id = _send . info . worker_id ; ServerGS -> event_workers . workers [ target_worker_id ] . deny_request = 0 ; zanTrace ( " " , target_worker_id ) ; return ZAN_OK ; } if ( _send . info . from_fd == ZAN_RESPONSE_SMALL ) { zanTrace ( " " , _send . info . from_fd , _send . info . fd , _send . info . worker_id , ev -> fd ) ; _send . data = resp . data ; _send . length = resp . info . len ; zanNetworker_send ( & _send ) ; return ZAN_OK ; } else { zanTrace ( " " , _send . info . from_fd , _send . info . fd , _send . info . worker_id , ev -> fd ) ; memcpy ( & pkg_resp , resp . data , sizeof ( pkg_resp ) ) ; worker = zanServer_get_worker ( ServerG . serv , pkg_resp . worker_id ) ; _send . data = worker -> send_shm ; _send . length = pkg_resp . length ; zanNetworker_send ( & _send ) ; worker -> lock . unlock ( & worker -> lock ) ; return ZAN_OK ; } } else if ( errno == EAGAIN ) { return ZAN_OK ; } else if ( errno == EINTR ) { zanWarn ( " " , n , errno , strerror ( errno ) ) ; continue ; } else { zanWarn ( " " , n , errno , strerror ( errno ) ) ; return ZAN_ERR ; } } }
receive data from worker process pipe
receive data from worker process pipe