自学教程：C++ APR_STATUS_IS_EAGAIN函数代码示例

static apr_status_t pass_data_to_filter(ap_filter_t *f, const char *data,                                        apr_size_t len, apr_bucket_brigade *bb){    ef_ctx_t *ctx = f->ctx;    ef_dir_t *dc = ctx->dc;    apr_status_t rv;    apr_size_t bytes_written = 0;    apr_size_t tmplen;    do {        tmplen = len - bytes_written;        rv = apr_file_write(ctx->proc->in,                       (const char *)data + bytes_written,                       &tmplen);        bytes_written += tmplen;        if (rv != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(rv)) {            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, f->r,                          "apr_file_write(child input), len %" APR_SIZE_T_FMT,                          tmplen);            return rv;        }        if (APR_STATUS_IS_EAGAIN(rv)) {            /* XXX handle blocking conditions here...  if we block, we need             * to read data from the child process and pass it down to the             * next filter!             */            rv = drain_available_output(f, bb);            if (APR_STATUS_IS_EAGAIN(rv)) {#if APR_FILES_AS_SOCKETS                int num_events;                const apr_pollfd_t *pdesc;                rv = apr_pollset_poll(ctx->pollset, f->r->server->timeout,                                      &num_events, &pdesc);                if (rv || dc->debug >= DBGLVL_GORY) {                    ap_log_rerror(APLOG_MARK, APLOG_DEBUG,                                  rv, f->r, "apr_pollset_poll()");                }                if (rv != APR_SUCCESS && !APR_STATUS_IS_EINTR(rv)) {                    /* some error such as APR_TIMEUP */                    return rv;                }#else /* APR_FILES_AS_SOCKETS */                /* Yuck... I'd really like to wait until I can read                 * or write, but instead I have to sleep and try again                 */                apr_sleep(100000); /* 100 milliseconds */                if (dc->debug >= DBGLVL_GORY) {                    ap_log_rerror(APLOG_MARK, APLOG_DEBUG,                                  0, f->r, "apr_sleep()");                }#endif /* APR_FILES_AS_SOCKETS */            }            else if (rv != APR_SUCCESS) {                return rv;            }        }    } while (bytes_written < len);    return rv;}

static int proxy_wstunnel_transfer(request_rec *r, conn_rec *c_i, conn_rec *c_o,                                     apr_bucket_brigade *bb, char *name){    int rv;#ifdef DEBUGGING    apr_off_t len;#endif    do {        apr_brigade_cleanup(bb);        rv = ap_get_brigade(c_i->input_filters, bb, AP_MODE_READBYTES,                            APR_NONBLOCK_READ, AP_IOBUFSIZE);        if (rv == APR_SUCCESS) {            if (c_o->aborted) {                return APR_EPIPE;            }            if (APR_BRIGADE_EMPTY(bb)) {                break;            }#ifdef DEBUGGING            len = -1;            apr_brigade_length(bb, 0, &len);            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, r, APLOGNO(02440)                          "read %" APR_OFF_T_FMT                          " bytes from %s", len, name);#endif            rv = ap_pass_brigade(c_o->output_filters, bb);            if (rv == APR_SUCCESS) {                ap_fflush(c_o->output_filters, bb);            }            else {                ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, APLOGNO(02441)                              "error on %s - ap_pass_brigade",                              name);            }        } else if (!APR_STATUS_IS_EAGAIN(rv) && !APR_STATUS_IS_EOF(rv)) {            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, rv, r, APLOGNO(02442)                          "error on %s - ap_get_brigade",                          name);        }    } while (rv == APR_SUCCESS);    ap_log_rerror(APLOG_MARK, APLOG_TRACE2, rv, r, "wstunnel_transfer complete");    if (APR_STATUS_IS_EAGAIN(rv)) {        rv = APR_SUCCESS;    }       return rv;}

static apr_status_t send_brigade_blocking(apr_socket_t *s,                                          apr_bucket_brigade *bb,                                          apr_size_t *bytes_written,                                          conn_rec *c){    apr_status_t rv;    rv = APR_SUCCESS;    while (!APR_BRIGADE_EMPTY(bb)) {        rv = send_brigade_nonblocking(s, bb, bytes_written, c);        if (rv != APR_SUCCESS) {            if (APR_STATUS_IS_EAGAIN(rv)) {                /* Wait until we can send more data */                apr_int32_t nsds;                apr_interval_time_t timeout;                apr_pollfd_t pollset;                pollset.p = c->pool;                pollset.desc_type = APR_POLL_SOCKET;                pollset.reqevents = APR_POLLOUT;                pollset.desc.s = s;                apr_socket_timeout_get(s, &timeout);                rv = apr_poll(&pollset, 1, &nsds, timeout);                if (rv != APR_SUCCESS) {                    break;                }            }            else {                break;            }        }    }    return rv;}

apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,                             int stream_id, apr_bucket_brigade *bb,                             struct apr_thread_cond_t *iowait){    apr_status_t status;     AP_DEBUG_ASSERT(m);    if (m->aborted) {        return APR_ECONNABORTED;    }    status = apr_thread_mutex_lock(m->lock);    if (APR_SUCCESS == status) {        h2_io *io = h2_io_set_get(m->stream_ios, stream_id);        if (io && !io->orphaned) {            io->input_arrived = iowait;            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_pre");            status = h2_io_in_read(io, bb, -1);            while (APR_STATUS_IS_EAGAIN(status)                    && !is_aborted(m, &status)                   && block == APR_BLOCK_READ) {                apr_thread_cond_wait(io->input_arrived, m->lock);                status = h2_io_in_read(io, bb, -1);            }            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_post");            io->input_arrived = NULL;        }        else {            status = APR_EOF;        }        apr_thread_mutex_unlock(m->lock);    }    return status;}

static void do_read(void){    apr_file_t *file;    apr_status_t status;    if (apr_file_open(&file, testfile, APR_WRITE,                 APR_OS_DEFAULT, pool) != APR_SUCCESS)        errmsg("Could not open test file./n");    printf("Test file opened./n");    status = apr_file_lock(file, APR_FLOCK_EXCLUSIVE | APR_FLOCK_NONBLOCK);    if (!APR_STATUS_IS_EAGAIN(status)) {        char msg[200];        errmsg(apr_psprintf(pool, "Expected APR_EAGAIN. Got %d: %s./n",                            status, apr_strerror(status, msg, sizeof(msg))));    }    printf("First attempt: we were properly locked out./nWaiting for lock...");    fflush(stdout);    if (apr_file_lock(file, APR_FLOCK_EXCLUSIVE) != APR_SUCCESS)        errmsg("Could not establish lock on test file.");    printf(" got it./n");    (void) apr_file_close(file);    printf("Exiting./n");}

/* drain_available_output(): * * if any data is available from the filter, read it and append it * to the the bucket brigade */static apr_status_t drain_available_output(ap_filter_t *f,                                           apr_bucket_brigade *bb){    request_rec *r = f->r;    conn_rec *c = r->connection;    ef_ctx_t *ctx = f->ctx;    ef_dir_t *dc = ctx->dc;    apr_size_t len;    char buf[4096];    apr_status_t rv;    apr_bucket *b;    while (1) {        len = sizeof(buf);        rv = apr_file_read(ctx->proc->out,                      buf,                      &len);        if ((rv && !APR_STATUS_IS_EAGAIN(rv)) ||            dc->debug >= DBGLVL_GORY) {            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, rv, r,                          "apr_file_read(child output), len %" APR_SIZE_T_FMT,                          !rv ? len : -1);        }        if (rv != APR_SUCCESS) {            return rv;        }        b = apr_bucket_heap_create(buf, len, NULL, c->bucket_alloc);        APR_BRIGADE_INSERT_TAIL(bb, b);        return APR_SUCCESS;    }    /* we should never get here; if we do, a bogus error message would be     * the least of our problems     */    return APR_ANONYMOUS;}

static apr_status_t send_out(h2_task *task, apr_bucket_brigade* bb, int block){    apr_off_t written, left;    apr_status_t status;    apr_brigade_length(bb, 0, &written);    H2_TASK_OUT_LOG(APLOG_TRACE2, task, bb, "h2_task send_out");    h2_beam_log(task->output.beam, task->c, APLOG_TRACE2, "send_out(before)");    /* engines send unblocking */    status = h2_beam_send(task->output.beam, bb,                           block? APR_BLOCK_READ : APR_NONBLOCK_READ);    h2_beam_log(task->output.beam, task->c, APLOG_TRACE2, "send_out(after)");        if (APR_STATUS_IS_EAGAIN(status)) {        apr_brigade_length(bb, 0, &left);        written -= left;        status = APR_SUCCESS;    }    if (status == APR_SUCCESS) {        if (h2_task_logio_add_bytes_out) {            h2_task_logio_add_bytes_out(task->c, written);        }        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, task->c,                       "h2_task(%s): send_out done", task->id);    }    else {        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, task->c,                      "h2_task(%s): send_out (%ld bytes)",                       task->id, (long)written);    }    return status;}

static apr_status_t serf_response_read(serf_bucket_t *bucket,                                       apr_size_t requested,                                       const char **data, apr_size_t *len){    response_context_t *ctx = bucket->data;    apr_status_t rv;    rv = wait_for_body(bucket, ctx);    if (rv) {        /* It's not possible to have read anything yet! */        if (APR_STATUS_IS_EOF(rv) || APR_STATUS_IS_EAGAIN(rv)) {            *len = 0;        }        return rv;    }    rv = serf_bucket_read(ctx->body, requested, data, len);    if (APR_STATUS_IS_EOF(rv)) {        if (ctx->chunked) {            ctx->state = STATE_TRAILERS;            /* Mask the result. */            rv = APR_SUCCESS;        }        else {            ctx->state = STATE_DONE;        }    }    return rv;}

bool LLPluginProcessParent::accept(){    bool result = false;    apr_status_t status = APR_EGENERAL;    mSocket = LLSocket::create(status, mListenSocket);    if(status == APR_SUCCESS)    {//		llinfos << "SUCCESS" << llendl;        // Success.  Create a message pipe on the new socket        new LLPluginMessagePipe(this, mSocket);        result = true;    }    else    {        mSocket.reset();        // EAGAIN means "No incoming connections". This is not an error.        if (!APR_STATUS_IS_EAGAIN(status))        {            // Some other error.            ll_apr_warn_status(status);            errorState();        }    }    return result;}

static void im_exec_fill_buffer(nx_module_input_t *input){    apr_status_t rv;    apr_size_t len;    ASSERT(input != NULL);    ASSERT(input->buf != NULL);    ASSERT(input->module != NULL);    ASSERT(input->desc_type == APR_POLL_FILE);    ASSERT(input->desc.f != NULL);    if ( input->bufstart == input->bufsize )    {	input->bufstart = 0;	input->buflen = 0;    }    if ( input->buflen == 0 )    {	input->bufstart = 0;    }    ASSERT(input->bufstart + input->buflen <= input->bufsize);    len = (apr_size_t) (input->bufsize - (input->buflen + input->bufstart));    rv = apr_file_read(input->desc.f, input->buf + input->bufstart + input->buflen, &len);    if ( rv != APR_SUCCESS )    {	if ( APR_STATUS_IS_EOF(rv) )	{	    throw_msg("Module %s got EOF, process exited? ", input->module->name);	}	else if ( APR_STATUS_IS_EAGAIN(rv) )	{ #ifdef WIN32	    // on windows EAGAIN is normal because we are using NON-BLOCKING reads	    // so we try again after 500 ms	    im_exec_add_read_event(input->module, 500);	    log_debug("got EAGAIN");#else	    log_error("got EAGAIN for blocking read in module %s", input->module->name);#endif	    ASSERT(len == 0);	    return;	}	else	{	    throw(rv, "Module %s couldn't read from pipe", input->module->name);	}    }    else    {	log_debug("im_exec read %d bytes", (int) len);    }    input->buflen += (int) len;    ASSERT(input->buflen <= input->bufsize);}

static int h2_session_status_from_apr_status(apr_status_t rv){    if (rv == APR_SUCCESS) {        return NGHTTP2_NO_ERROR;    }    else if (APR_STATUS_IS_EAGAIN(rv)) {        return NGHTTP2_ERR_WOULDBLOCK;    }    else if (APR_STATUS_IS_EOF(rv)) {            return NGHTTP2_ERR_EOF;    }    return NGHTTP2_ERR_PROTO;}

bool LLPluginProcessParent::accept(){	bool result = false;		apr_status_t status = APR_EGENERAL;	apr_socket_t *new_socket = NULL;		status = apr_socket_accept(		&new_socket,		mListenSocket->getSocket(),		gAPRPoolp);		if(status == APR_SUCCESS)	{//		llinfos << "SUCCESS" << llendl;		// Success.  Create a message pipe on the new socket		// we MUST create a new pool for the LLSocket, since it will take ownership of it and delete it in its destructor!		apr_pool_t* new_pool = NULL;		status = apr_pool_create(&new_pool, gAPRPoolp);		mSocket = LLSocket::create(new_socket, new_pool);		new LLPluginMessagePipe(this, mSocket);		result = true;	}	else if(APR_STATUS_IS_EAGAIN(status))	{//		llinfos << "EAGAIN" << llendl;		// No incoming connections.  This is not an error.		status = APR_SUCCESS;	}	else	{//		llinfos << "Error:" << llendl;		ll_apr_warn_status(status);				// Some other error.		errorState();	}		return result;	}

int jxr_socket_sendfull(jaxer_connection *ac, const unsigned char *b, int len){    jaxer_dir_conf *config = (jaxer_dir_conf *)ap_get_module_config(ac->request->per_dir_config, &jaxer_module);    int sent = 0;    apr_socket_t *sock;	apr_status_t rv;	apr_size_t send_len;	int retry = 0;	apr_pool_t *p = (config) ? config->reqPool : ac->worker->pool;	sock = ac->sock;    if (g_jxr_network_trace)    {        jxr_trace("SEND", b, len, p);    }    while (sent < len)	{		send_len = len - sent;		rv = apr_socket_send(sock, (const char*)(b + sent), &send_len);		if (send_len == len - sent)		{			sent += (int) send_len;			break;		}		if (rv != APR_SUCCESS)		{            /*             * Let's hope this traps EWOULDBLOCK too !             */            if (APR_STATUS_IS_EAGAIN(rv) && retry<3)			{				retry++;            }else			{				ap_log_perror(APLOG_MARK, APLOG_WARNING, rv, p, "mod_jaxer: send data over socket error: total len=%d sent=%d", len, sent);				return -1;			}        }		sent += (int) send_len;	}	return sent;}

/* * Callback when nghttp2 wants to send bytes back to the client. */static ssize_t send_cb(nghttp2_session *ngh2,                       const uint8_t *data, size_t length,                       int flags, void *userp){    h2_session *session = (h2_session *)userp;    apr_status_t status;        (void)ngh2;    (void)flags;    status = h2_conn_io_write(&session->io, (const char *)data, length);    if (status == APR_SUCCESS) {        return length;    }    if (APR_STATUS_IS_EAGAIN(status)) {        return NGHTTP2_ERR_WOULDBLOCK;    }    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,                  "h2_session: send error");    return h2_session_status_from_apr_status(status);}

/** Receive at max howmany bytes on socket sock and store them in recvbuf. *  Parameter howmany allows to receive less than the available space in *  recvbuf. *  Setting howmany to the same value as recvdbuf->buf_len means receive as *  much as you can (offset is taken into consideration). */apr_status_tcpe_recv(apr_socket_t *sock, cpe_io_buf *recvbuf, apr_size_t *howmany){    apr_status_t  rv;    apr_size_t    len;    assert(recvbuf->buf_offset <= recvbuf->buf_capacity);    if (*howmany <= 0) {        cpe_log(CPE_DEB, "invalid howmany %d", *howmany);        return APR_EINVAL;    }    len = recvbuf->buf_capacity - recvbuf->buf_offset;    if (len == 0) {        cpe_log(CPE_DEB, "no buffer space to receive on buffer %p (socket %p)",            recvbuf, sock);        return APR_EGENERAL;    }    len = cpe_min(len, *howmany);    rv = apr_socket_recv(sock, &recvbuf->buf[recvbuf->buf_offset], &len);    if (rv != APR_SUCCESS) {        if (APR_STATUS_IS_EAGAIN(rv)) {            cpe_log(CPE_ERR, "CPE client programming error: socket %p marked "                "for non-blocking I/O and no data ready to be read.", sock);        }        cpe_log(CPE_DEB, "apr_socket_recv: %s (socket %p)", cpe_errmsg(rv),            sock);        /* Not clear what to do here; should look at the returned value of         * len? Should consider APR_EOF ?         */    }    *howmany = len;    recvbuf->buf_offset += len;    recvbuf->buf_len = recvbuf->buf_offset;    recvbuf->total += len;    cpe_log(CPE_DEB, "received %d bytes (offset %d, socket %p)", len,        recvbuf->buf_offset, sock);    return rv;}

int main(int argc, const char * const *argv){    apr_file_t *file;    apr_status_t status;    apr_pool_t *p;    apr_initialize();    apr_pool_create(&p, NULL);    if (apr_file_open(&file, TESTFILE, APR_WRITE, APR_OS_DEFAULT, p)         != APR_SUCCESS) {                exit(UNEXPECTED_ERROR);    }    status = apr_file_lock(file, APR_FLOCK_EXCLUSIVE | APR_FLOCK_NONBLOCK);    if (status == APR_SUCCESS) {        exit(SUCCESSFUL_READ);    }    if (APR_STATUS_IS_EAGAIN(status)) {        exit(FAILED_READ);    }    exit(UNEXPECTED_ERROR);}

int jxr_socket_recvfull(jaxer_connection *ac, unsigned char *b, int len){    jaxer_dir_conf *config = (jaxer_dir_conf *)ap_get_module_config(ac->request->per_dir_config, &jaxer_module);    int rdlen = 0;    apr_socket_t *sock;	apr_status_t rv;	apr_size_t read_len;	int retry = 0;	apr_pool_t *p = (config) ? config->reqPool : ac->worker->pool;	sock = ac->sock;    while (rdlen < len)	{		read_len = len - rdlen;		rv = apr_socket_recv(sock, (char *)b + rdlen, &read_len);		if (rv != APR_SUCCESS)		{			if (APR_STATUS_IS_EAGAIN(rv) && retry<3)			{				retry++;            }else			{				ap_log_perror(APLOG_MARK, APLOG_WARNING, rv, p, "mod_jaxer: receive data over socket error: total len=%d read=%d", len, rdlen);				return -1;			}		}        rdlen += (int) read_len;    }    if (g_jxr_network_trace)    {        jxr_trace("RECV", b, rdlen, p);    }    return rdlen;}

apr_status_t handle_response(serf_request_t *request,                             serf_bucket_t *response,                             void *handler_baton,                             apr_pool_t *pool){    handler_baton_t *ctx = handler_baton;    if (! response) {        serf_connection_request_create(ctx->tb->connection,                                       setup_request,                                       ctx);        return APR_SUCCESS;    }    while (1) {        apr_status_t status;        const char *data;        apr_size_t len;        status = serf_bucket_read(response, 2048, &data, &len);        if (SERF_BUCKET_READ_ERROR(status))            return status;        if (APR_STATUS_IS_EOF(status)) {            APR_ARRAY_PUSH(ctx->handled_requests, int) = ctx->req_id;            ctx->done = TRUE;            return APR_EOF;        }        if (APR_STATUS_IS_EAGAIN(status)) {            return status;        }    }    return APR_SUCCESS;}

apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,                             int stream_id, apr_bucket_brigade *bb,                              apr_table_t *trailers,                             struct apr_thread_cond_t *iowait){    apr_status_t status;     int acquired;        AP_DEBUG_ASSERT(m);    if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {        h2_io *io = h2_io_set_get(m->stream_ios, stream_id);        if (io && !io->orphaned) {            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_pre");                        h2_io_signal_init(io, H2_IO_READ, m->stream_timeout, iowait);            status = h2_io_in_read(io, bb, -1, trailers);            while (APR_STATUS_IS_EAGAIN(status)                    && !is_aborted(m, &status)                   && block == APR_BLOCK_READ) {                ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, m->c,                              "h2_mplx(%ld-%d): wait on in data (BLOCK_READ)",                               m->id, stream_id);                status = h2_io_signal_wait(m, io);                if (status == APR_SUCCESS) {                    status = h2_io_in_read(io, bb, -1, trailers);                }            }            H2_MPLX_IO_IN(APLOG_TRACE2, m, io, "h2_mplx_in_read_post");            h2_io_signal_exit(io);        }        else {            status = APR_EOF;        }        leave_mutex(m, acquired);    }    return status;}

static int connection_rx_cb(serv_ctx_t *serv_ctx, apr_pollset_t *pollset, apr_socket_t *sock){    //printf("connection_rx_cb()/n");    apr_size_t len = BUFSIZE - serv_ctx->down_buf_level;    apr_status_t rv = apr_socket_recv(sock, (char*)&serv_ctx->down_buf[serv_ctx->down_buf_level], &len); //Daten empfangen    if (APR_STATUS_IS_EAGAIN(rv))    {        /* we have no data to read. we should keep polling the socket */        return TRUE;    }    else if (APR_STATUS_IS_EOF(rv) || len == 0)    {        /* we lost TCP session.         * XXX On Windows, rv would equal to APR_SUCCESS and len==0 in this case. So, we should check @len in addition to APR_EOF check */        return FALSE;    }    else    {        /* we got data *///        printf("rx(%d): ", serv_ctx->channel_number);//        for(size_t i = 0; i < len ;++i)//            printf("0x%02x ", (int)serv_ctx->down_buf[serv_ctx->down_buf_level+i]);//        printf("/n");        serv_ctx->down_buf_level += len;        //ipdbgJtagWrite(serv_ctx->chain, (uint8_t*)buf, len, serv_ctx->valid_mask);    }    return TRUE;}

/* If a 200 OK was received for the CONNECT request, consider the connection   as ready for use. */static apr_status_t handle_response(serf_request_t *request,                                    serf_bucket_t *response,                                    void *handler_baton,                                    apr_pool_t *pool){    apr_status_t status;    serf_status_line sl;    req_ctx_t *ctx = handler_baton;    serf_connection_t *conn = request->conn;    /* CONNECT request was cancelled. Assuming that this is during connection       reset, we can safely discard the request as a new one will be created       when setting up the next connection. */    if (!response)        return APR_SUCCESS;    status = serf_bucket_response_status(response, &sl);    if (SERF_BUCKET_READ_ERROR(status)) {        return status;    }    if (!sl.version && (APR_STATUS_IS_EOF(status) ||                      APR_STATUS_IS_EAGAIN(status)))    {        return status;    }    status = serf_bucket_response_wait_for_headers(response);    if (status && !APR_STATUS_IS_EOF(status)) {        return status;    }    /* RFC 2817:  Any successful (2xx) response to a CONNECT request indicates       that the proxy has established a connection to the requested host and       port, and has switched to tunneling the current connection to that server       connection.    */    if (sl.code >= 200 && sl.code < 300) {        serf_bucket_t *hdrs;        const char *val;        conn->state = SERF_CONN_CONNECTED;        /* Body is supposed to be empty. */        apr_pool_destroy(ctx->pool);        serf_bucket_destroy(conn->ssltunnel_ostream);        serf_bucket_destroy(conn->stream);        conn->stream = NULL;        ctx = NULL;        serf__log_skt(CONN_VERBOSE, __FILE__, conn->skt,                      "successfully set up ssl tunnel./n");        /* Fix for issue #123: ignore the "Connection: close" header here,           leaving the header in place would make the serf's main context           loop close this connection immediately after reading the 200 OK           response. */        hdrs = serf_bucket_response_get_headers(response);        val = serf_bucket_headers_get(hdrs, "Connection");        if (val && strcasecmp("close", val) == 0) {            serf__log_skt(CONN_VERBOSE, __FILE__, conn->skt,                      "Ignore Connection: close header on this reponse, don't "                      "close the connection now that the tunnel is set up./n");            serf__bucket_headers_remove(hdrs, "Connection");        }        return APR_EOF;    }    /* Authentication failure and 2xx Ok are handled at this point,       the rest are errors. */    return SERF_ERROR_SSLTUNNEL_SETUP_FAILED;}

// virtualLLIOPipe::EStatus LLIOSocketReader::process_impl(	const LLChannelDescriptors& channels,	buffer_ptr_t& buffer,	bool& eos,	LLSD& context,	LLPumpIO* pump){	LL_RECORD_BLOCK_TIME(FTM_PROCESS_SOCKET_READER);	PUMP_DEBUG;	if(!mSource) return STATUS_PRECONDITION_NOT_MET;	if(!mInitialized)	{		PUMP_DEBUG;		// Since the read will not block, it's ok to initialize and		// attempt to read off the descriptor immediately.		mInitialized = true;		if(pump)		{			PUMP_DEBUG;			LL_DEBUGS() << "Initializing poll descriptor for LLIOSocketReader."					 << LL_ENDL;			apr_pollfd_t poll_fd;			poll_fd.p = NULL;			poll_fd.desc_type = APR_POLL_SOCKET;			poll_fd.reqevents = APR_POLLIN;			poll_fd.rtnevents = 0x0;			poll_fd.desc.s = mSource->getSocket();			poll_fd.client_data = NULL;			pump->setConditional(this, &poll_fd);		}	}	//if(!buffer)	//{	//	buffer = new LLBufferArray;	//}	PUMP_DEBUG;	const apr_size_t READ_BUFFER_SIZE = 1024;	char read_buf[READ_BUFFER_SIZE]; /*Flawfinder: ignore*/	apr_size_t len;	apr_status_t status = APR_SUCCESS;	do	{		PUMP_DEBUG;		len = READ_BUFFER_SIZE;		status = apr_socket_recv(mSource->getSocket(), read_buf, &len);		buffer->append(channels.out(), (U8*)read_buf, len);	} while((APR_SUCCESS == status) && (READ_BUFFER_SIZE == len));	LL_DEBUGS() << "socket read status: " << status << LL_ENDL;	LLIOPipe::EStatus rv = STATUS_OK;	PUMP_DEBUG;	// *FIX: Also need to check for broken pipe	if(APR_STATUS_IS_EOF(status))	{		// *FIX: Should we shut down the socket read?		if(pump)		{			pump->setConditional(this, NULL);		}		rv = STATUS_DONE;		eos = true;	}	else if(APR_STATUS_IS_EAGAIN(status))	{/*Commented out by Aura 9-9-8 for DEV-19961.		// everything is fine, but we can terminate this process pump.			rv = STATUS_BREAK;*/	}	else	{		if(ll_apr_warn_status(status))		{			rv = STATUS_ERROR;		}	}	PUMP_DEBUG;	return rv;}

static apr_status_t handle_response(serf_request_t *request,                                    serf_bucket_t *response,                                    void *handler_baton,                                    apr_pool_t *pool){    const char *data;    apr_size_t len;    serf_status_line sl;    apr_status_t status;    handler_baton_t *ctx = handler_baton;    if (!response) {        /* Oh no!  We've been cancelled! */        abort();    }    status = serf_bucket_response_status(response, &sl);    if (status) {        if (APR_STATUS_IS_EAGAIN(status)) {            return APR_SUCCESS;        }        abort();    }    while (1) {        status = serf_bucket_read(response, 2048, &data, &len);        if (SERF_BUCKET_READ_ERROR(status))            return status;        /*fwrite(data, 1, len, stdout);*/        if (!ctx->hdr_read) {            serf_bucket_t *hdrs;            const char *val;            printf("Processing %s/n", ctx->path);            hdrs = serf_bucket_response_get_headers(response);            val = serf_bucket_headers_get(hdrs, "Content-Type");            /* FIXME: This check isn't quite right because Content-Type could             * be decorated; ideally strcasestr would be correct.             */            if (val && strcasecmp(val, "text/html") == 0) {                ctx->is_html = 1;                apr_pool_create(&ctx->parser_pool, NULL);                ctx->parser = apr_xml_parser_create(ctx->parser_pool);            }            else {                ctx->is_html = 0;            }            ctx->hdr_read = 1;        }        if (ctx->is_html) {            apr_status_t xs;            xs = apr_xml_parser_feed(ctx->parser, data, len);            /* Uh-oh. */            if (xs) {#ifdef SERF_VERBOSE                printf("XML parser error (feed): %d/n", xs);#endif                ctx->is_html = 0;            }        }        /* are we done yet? */        if (APR_STATUS_IS_EOF(status)) {            if (ctx->is_html) {                apr_xml_doc *xmld;                apr_status_t xs;                doc_path_t *dup;                xs = apr_xml_parser_done(ctx->parser, &xmld);                if (xs) {#ifdef SERF_VERBOSE                    printf("XML parser error (done): %d/n", xs);#endif                    return xs;                }                dup = (doc_path_t*)                    serf_bucket_mem_alloc(ctx->doc_queue_alloc,                                          sizeof(doc_path_t));                dup->doc = xmld;                dup->path = (char*)serf_bucket_mem_alloc(ctx->doc_queue_alloc,                                                         ctx->path_len);                memcpy(dup->path, ctx->path, ctx->path_len);                dup->pool = ctx->parser_pool;                *(doc_path_t **)apr_array_push(ctx->doc_queue) = dup;                apr_thread_cond_signal(ctx->doc_queue_condvar);            }            apr_atomic_dec32(ctx->requests_outstanding);            serf_bucket_mem_free(ctx->allocator, ctx->path);            if (ctx->query) {                serf_bucket_mem_free(ctx->allocator, ctx->query);                serf_bucket_mem_free(ctx->allocator, ctx->full_path);//.........这里部分代码省略.........

void* thread_socket_pipe_receiver(apr_thread_t* thd, void* data){	frl_socket_pipe* pipe = (frl_socket_pipe*)data;	apr_status_t state;	apr_socket_t* listen_sock;	apr_socket_create(&listen_sock, pipe->sock_addr->family, SOCK_STREAM, APR_PROTO_TCP, pipe->sockpool);	apr_socket_opt_set(listen_sock, APR_SO_NONBLOCK, 1);	apr_socket_timeout_set(listen_sock, 0);	apr_socket_opt_set(listen_sock, APR_SO_REUSEADDR, 1);	pipe->recv_state = apr_socket_bind(listen_sock, pipe->sock_addr);	F_ERROR_IF_RUN(APR_SUCCESS != pipe->recv_state, return NULL, "[frl_socket_pipe::thread_socket_pipe_receiver]: Socket Binding Error: %d/n", pipe->recv_state);	pipe->recv_state = apr_socket_listen(listen_sock, SOMAXCONN);	F_ERROR_IF_RUN(APR_SUCCESS != pipe->recv_state, return NULL, "[frl_socket_pipe::thread_socket_pipe_receiver]: Socket Listen Error: %d/n", pipe->recv_state);	apr_uint32_t hash;	apr_pollset_t* pollset;	apr_pollset_create(&pollset, pipe->replicate+2, pipe->sockpool, 0);	apr_pollfd_t pfd = { pipe->sockpool, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, NULL };	pfd.desc.s = listen_sock;	apr_pollset_add(pollset, &pfd);	do {		// the fun loop		apr_int32_t total;		const apr_pollfd_t* ret_pfd;		pipe->recv_state = apr_pollset_poll(pollset, SOCKET_PIPE_POLL_TIMEOUT, &total, &ret_pfd);		if (APR_SUCCESS == pipe->recv_state)		{			for (int i = 0; i < total; i++)			{				if (ret_pfd[i].desc.s == listen_sock)				{					apr_socket_t* accept_sock;					state = apr_socket_accept(&accept_sock, listen_sock, pipe->sockpool);					F_ERROR_IF_RUN(APR_SUCCESS != state, continue, "[frl_socket_pipe::thread_socket_pipe_receiver]: Socket Accept Error: %d/n", state);					// accept connection, initiate recv					frl_pipe_state_t* pipestate = (frl_pipe_state_t*)frl_slab_palloc(pipe->statepool);					apr_pollfd_t pfd = { pipe->sockpool, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, pipestate };					pipestate->state = FRL_PIPE_READ_HEADER_START;					pipestate->reader = (char*)&pipestate->header;					pipestate->offset = 0;					pipestate->size = SIZEOF_FRL_PIPE_HEADER_T;					pfd.desc.s = accept_sock;					apr_socket_opt_set(accept_sock, APR_SO_NONBLOCK, 1);					apr_socket_timeout_set(accept_sock, 0);					apr_pollset_add(pollset, &pfd);				} else {					if (ret_pfd[i].rtnevents & APR_POLLIN)					{						frl_pipe_state_t* pipestate = (frl_pipe_state_t*)ret_pfd[i].client_data;						apr_size_t len_a = pipestate->size-pipestate->offset;						state = apr_socket_recv(ret_pfd[i].desc.s, pipestate->reader, &len_a);						pipestate->offset += len_a;						pipestate->reader += len_a;						// read buffer to reader						if ((pipestate->offset >= pipestate->size)||(APR_STATUS_IS_EAGAIN(state)))						{							pipestate->offset = pipestate->size;							PIPE_STATE_TO_COMPLETE(pipestate->state);							// read complete, move state to complete						} else if ((APR_STATUS_IS_EOF(state))||(len_a == 0)) {							apr_pollfd_t pfd = { pipe->sockpool, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, pipestate };							pfd.desc.s = ret_pfd[i].desc.s;							apr_pollset_remove(pollset, &pfd);							frl_slab_pfree(pipestate);							apr_socket_close(ret_pfd[i].desc.s);							// remote error, close connection							continue;						}						switch (pipestate->state)						{							case FRL_PIPE_READ_HEADER_COMPLETE:							{								// recv header (hash & size)								pipestate->data.offset = 0;								pipestate->data.size = pipestate->header.size;								state = pipe->recv_before(&pipestate->data.buf, &pipestate->data.size);								if (FRL_PROGRESS_IS_INTERRUPT(state))								{									apr_pollfd_t pfd = { pipe->sockpool, APR_POLL_SOCKET, APR_POLLIN, 0, { NULL }, pipestate };									pfd.desc.s = ret_pfd[i].desc.s;									apr_pollset_remove(pollset, &pfd);									frl_slab_pfree(pipestate);									apr_socket_close(ret_pfd[i].desc.s);									continue;								}								pipestate->state = FRL_PIPE_READ_BLOCK_START;								// start to read block (<= 4092 bytes each)								pipestate->reader = pipestate->buffer;								pipestate->offset = 0;								if (pipestate->data.size < SIZEOF_FRL_PIPE_BLOCK_BUFFER)									pipestate->size = pipestate->data.size+SIZEOF_FRL_PIPE_HEADER_T;								else									pipestate->size = SOCKET_PACKAGE_SIZE;								break;							}							case FRL_PIPE_READ_BLOCK_COMPLETE:							{								// a block complete, move to data								memcpy(pipestate->data.buf+pipestate->data.offset, &pipestate->block.start, pipestate->block.header.size);//.........这里部分代码省略.........

lt_http_status_t lt_http_server_run( lt_http_server_t * server ){    apr_pool_t * pool = NULL;    apr_socket_t * client = NULL;    char error[1025];    memset( error, 0, 1025 );    if( server == NULL ) return LT_HTTP_INVALID_ARG;    /* prepare connection pool */    apr_pool_create( &pool, server->pool );    /* make the socket non-blocking */    if( APR_SUCCESS != apr_socket_opt_set( server->socket,                    APR_SO_NONBLOCK, 1 ) ) {        my_perror( "ERROR: apr_socket_opt_set failed with: " );        return LT_HTTP_INVALID_ARG;    }    while( 1 ) {        apr_status_t rv;        /* bool reading should be atomic operation so no locking is needed */        if( server->stoprequested ) {            break;        }        /* clear pool memory */        apr_pool_clear( pool );        /* accept new connection */        rv = apr_socket_accept( &client, server->socket, pool );        if( APR_STATUS_IS_EAGAIN( rv ) || APR_STATUS_IS_EINTR( rv ) ) {            /* sleep for 100ms before accepting new client */            apr_sleep( 100 * 1000 );            continue;        }        if( APR_SUCCESS != rv ) {            my_perror( "ERROR: apr_socket_accept failed with: " );            continue;        }        /* determine client address */        {            apr_sockaddr_t * sa = NULL;            char * ip = NULL;            if( APR_SUCCESS != apr_socket_addr_get( &sa,                            APR_REMOTE, client ) ) {                my_perror( "ERROR: apr_socket_addr_get failed with: " );                apr_socket_close( client );                continue;            }            if( APR_SUCCESS != apr_sockaddr_ip_get( &ip, sa ) ) {                my_perror( "ERROR: apr_sockaddr_ip_get failed with: " );                apr_socket_close( client );                continue;            }        }        /* immediatelly start sending HTTP response headers */        {            char * headers = apr_pstrcat( pool,                    "HTTP/1.0 200 OK/r/n"                    "Content-Length: ",                    apr_ltoa( pool, server->finfo.size ),                    "/r/n",                    "Content-Type: application/octet-stream;"                    " charset=utf-8/r/n",                    "Connection: Close/r/n",                    "/r/n",                    NULL );            apr_size_t headers_size = strlen( headers );            if( APR_SUCCESS != apr_socket_send(                            client, headers, &headers_size ) ) {                my_perror( "ERROR: apr_socket_send failed with: " );                apr_socket_close( client );                continue;            }        }        /* send file contents */        {            apr_off_t offset = 0;            apr_size_t len = server->finfo.size;            if( APR_SUCCESS != apr_socket_sendfile(                            client, server->file, NULL,                            &offset, &len, 0 ) ) {                my_perror( "ERROR: apr_socket_sendfile failed with: " );                apr_socket_close( client );                continue;            }        }        /* read and discard all headers */        {            apr_status_t rv;//.........这里部分代码省略.........

void lfd_listen(apr_pool_t * mp){	apr_pool_t		* thd_pool = NULL;	apr_socket_t		* listen_sock;	apr_socket_t		* client_sock;	apr_thread_t *thd;	apr_threadattr_t	* thattr;	apr_pollfd_t		  pfd;	apr_interval_time_t	  timeout = lfd_config_max_acceptloop_timeout;	apr_int32_t		  nsds;	apr_status_t		  rc;	create_listen_socket(&listen_sock, mp);	if(NULL == listen_sock)	{		lfd_log(LFD_ERROR, "lfd_listen: could not create listen socket");		return;	}	rc = apr_threadattr_create(&thattr, mp);	if(APR_SUCCESS != rc)	{		lfd_log_apr_err(rc, "apr_threadattr_create failed");		return;	}	while(1)	{		//###: Should I allocate the pool as a subpool of the root pool?		//What is the amount allocated per pool and is it freed when the child pool is destroyed?		//rc = apr_pool_create(&thd_pool, mp);		if(NULL == thd_pool)		{			rc = apr_pool_create(&thd_pool, NULL);			if(APR_SUCCESS != rc)			{				lfd_log_apr_err(rc, "apr_pool_create of thd_pool failed");				continue;			}		}		create_pollfd_from_socket(&pfd, listen_sock, mp);		rc = apr_poll(&pfd, 1, &nsds, timeout);		if((APR_SUCCESS != rc) && (!APR_STATUS_IS_TIMEUP(rc)) && (!APR_STATUS_IS_EINTR(rc)))		{			//break - an unrecoverable error occured			lfd_log_apr_err(rc, "apr_poll failed");			break;		}		if(apr_atomic_read32(&ftp_must_exit))		{			//if the flag says we must exit, we comply, so bye bye!			return;		}		if(APR_STATUS_IS_TIMEUP(rc) || APR_STATUS_IS_EINTR(rc) || (APR_POLLIN != pfd.rtnevents))		{			continue;		}		rc = apr_socket_accept(&client_sock, listen_sock, thd_pool);		if(APR_SUCCESS != rc)		{			//###: For which errorcode must we break out of the loop?			lfd_log_apr_err(rc, "apr_socket_accept failed");			if(APR_STATUS_IS_EAGAIN(rc))			{				lfd_log(LFD_ERROR, "lfd_listen: APR_STATUS_IS_EAGAIN");			}			continue;		}		rc = apr_thread_create(&thd, thattr, &lfd_worker_protocol_main, (void*)client_sock, thd_pool);		if(APR_SUCCESS != rc)		{			lfd_log_apr_err(rc, "apr_thread_create failed");			apr_socket_close(client_sock);			continue;		}		thd_pool = NULL;	}}

static apr_status_t handle_response(serf_request_t *request,                                    serf_bucket_t *response,                                    void *handler_baton,                                    apr_pool_t *pool){    serf_status_line sl;    apr_status_t status;    handler_baton_t *ctx = handler_baton;    if (!response) {        /* A NULL response probably means that the connection was closed while           this request was already written. Just requeue it. */        serf_connection_t *conn = serf_request_get_conn(request);        serf_connection_request_create(conn, setup_request, handler_baton);        return APR_SUCCESS;    }    status = serf_bucket_response_status(response, &sl);    if (status) {        return status;    }    while (1) {        struct iovec vecs[64];        int vecs_read;        apr_size_t bytes_written;        status = serf_bucket_read_iovec(response, 8000, 64, vecs, &vecs_read);        if (SERF_BUCKET_READ_ERROR(status))            return status;        /* got some data. print it out. */        if (vecs_read) {            apr_file_writev(ctx->output_file, vecs, vecs_read, &bytes_written);        }        /* are we done yet? */        if (APR_STATUS_IS_EOF(status)) {            if (ctx->print_headers) {                serf_bucket_t *hdrs;                hdrs = serf_bucket_response_get_headers(response);                while (1) {                    status = serf_bucket_read_iovec(hdrs, 8000, 64, vecs,                                                    &vecs_read);                    if (SERF_BUCKET_READ_ERROR(status))                        return status;                    if (vecs_read) {                        apr_file_writev(ctx->output_file, vecs, vecs_read,                                        &bytes_written);                    }                    if (APR_STATUS_IS_EOF(status)) {                        break;                    }                }            }            apr_atomic_inc32(&ctx->completed_requests);            return APR_EOF;        }        /* have we drained the response so far? */        if (APR_STATUS_IS_EAGAIN(status))            return status;        /* loop to read some more. */    }    /* NOTREACHED */}

static void im_ssl_accept(nx_module_t *module){    nx_im_ssl_conf_t *imconf;    apr_socket_t *sock;    apr_sockaddr_t *sa;    char *ipstr;    nx_module_input_t *input;    SSL *ssl;    apr_pool_t *pool = NULL;    apr_status_t rv;    nx_exception_t e;    log_debug("im_ssl_accept");        imconf = (nx_im_ssl_conf_t *) module->config;    pool = nx_pool_create_child(module->pool);    try    {	if ( (rv = apr_socket_accept(&sock, imconf->listensock, pool)) != APR_SUCCESS )	{	    if ( APR_STATUS_IS_EAGAIN(rv) )	    {		nx_module_add_poll_event(module);		apr_pool_destroy(pool);	    }	    else	    {		throw(rv, "couldn't accept connection on %s:%u (statuscode: %d)",		      imconf->host, imconf->port, rv);	    }	}	if ( rv == APR_SUCCESS )	{	    CHECKERR_MSG(apr_socket_opt_set(sock, APR_SO_NONBLOCK, 1),			 "couldn't set SO_NONBLOCK on accepted socket");	    CHECKERR_MSG(apr_socket_timeout_set(sock, 0),			 "couldn't set socket timeout on accepted socket");	    CHECKERR_MSG(apr_socket_addr_get(&sa, APR_REMOTE, sock),			 "couldn't get info on accepted socket");	    CHECKERR_MSG(apr_sockaddr_ip_get(&ipstr, sa),			 "couldn't get IP of accepted socket");	    nx_module_pollset_add_socket(module, imconf->listensock, APR_POLLIN | APR_POLLHUP);	    	    ssl = nx_ssl_from_socket(&(imconf->ssl_ctx), sock);	    ASSERT(ssl != NULL);	    SSL_set_accept_state(ssl);	    //SSL_accept(ssl);    	    CHECKERR_MSG(apr_socket_opt_set(sock, APR_SO_NONBLOCK, 1),			 "couldn't set SO_NONBLOCK on accepted socket");	    CHECKERR_MSG(apr_socket_timeout_set(sock, 0),			 "couldn't set socket timeout on accepted socket");	    input = nx_module_input_new(module, pool);	    input->desc_type = APR_POLL_SOCKET;	    input->desc.s = sock;	    input->inputfunc = imconf->inputfunc;	    ASSERT(input->inputfunc != NULL);    	    nx_module_input_data_set(input, "ssl", ssl);			    CHECKERR_MSG(apr_socket_data_set(sock, input, "input", NULL),			 "couldn't set data on socket");	    NX_DLIST_INSERT_TAIL(imconf->connections, input, link);	    nx_module_input_data_set(input, "recv_from_str", ipstr);	    	    nx_module_pollset_add_socket(module, sock, APR_POLLIN | APR_POLLHUP);	    log_info("SSL connection accepted from %s:%u", ipstr, sa->port);	}    }    catch(e)    {	apr_pool_destroy(pool);	rethrow(e);    }}

static apr_status_t serf_deflate_read(serf_bucket_t *bucket,                                      apr_size_t requested,                                      const char **data, apr_size_t *len){    deflate_context_t *ctx = bucket->data;    apr_status_t status;    const char *private_data;    apr_size_t private_len;    int zRC;    while (1) {        switch (ctx->state) {        case STATE_READING_HEADER:        case STATE_READING_VERIFY:            status = serf_bucket_read(ctx->stream, ctx->stream_left,                                      &private_data, &private_len);            if (SERF_BUCKET_READ_ERROR(status)) {                return status;            }            memcpy(ctx->hdr_buffer + (ctx->stream_size - ctx->stream_left),                   private_data, private_len);            ctx->stream_left -= private_len;            if (ctx->stream_left == 0) {                ctx->state++;                if (APR_STATUS_IS_EAGAIN(status)) {                    *len = 0;                    return status;                }            }            else if (status) {                *len = 0;                return status;            }            break;        case STATE_HEADER:            if (ctx->hdr_buffer[0] != deflate_magic[0] ||                ctx->hdr_buffer[1] != deflate_magic[1]) {                return SERF_ERROR_DECOMPRESSION_FAILED;            }            if (ctx->hdr_buffer[3] != 0) {                return SERF_ERROR_DECOMPRESSION_FAILED;            }            ctx->state++;            break;        case STATE_VERIFY:        {            unsigned long compCRC, compLen, actualLen;            /* Do the checksum computation. */            compCRC = getLong((unsigned char*)ctx->hdr_buffer);            if (ctx->crc != compCRC) {                return SERF_ERROR_DECOMPRESSION_FAILED;            }            compLen = getLong((unsigned char*)ctx->hdr_buffer + 4);            /* The length in the trailer is module 2^32, so do the same for               the actual length. */            actualLen = ctx->zstream.total_out;            actualLen &= 0xFFFFFFFF;            if (actualLen != compLen) {                return SERF_ERROR_DECOMPRESSION_FAILED;            }            ctx->state++;            break;        }        case STATE_INIT:            zRC = inflateInit2(&ctx->zstream, ctx->windowSize);            if (zRC != Z_OK) {                return SERF_ERROR_DECOMPRESSION_FAILED;            }            ctx->zstream.next_out = ctx->buffer;            ctx->zstream.avail_out = ctx->bufferSize;            ctx->state++;            break;        case STATE_FINISH:            inflateEnd(&ctx->zstream);            serf_bucket_aggregate_prepend(ctx->stream, ctx->inflate_stream);            ctx->inflate_stream = 0;            ctx->state++;            break;        case STATE_INFLATE:            /* Do we have anything already uncompressed to read? */            status = serf_bucket_read(ctx->inflate_stream, requested, data,                                      len);            if (SERF_BUCKET_READ_ERROR(status)) {                return status;            }            /* Hide EOF. */            if (APR_STATUS_IS_EOF(status)) {                status = ctx->stream_status;                if (APR_STATUS_IS_EOF(status)) {                    /* We've read all of the data from our stream, but we                     * need to continue to iterate until we flush                     * out the zlib buffer.                     */                    status = APR_SUCCESS;                }//.........这里部分代码省略.........