term_t bif_read0_2(term_t Port, term_t Len, process_t *ctx)
{
	apr_status_t rs;
	apr_size_t size;
	apr_byte_t *buf;
	term_t bin;
	port_t *p;
	if (!is_port(Port) || !is_int(Len))
		return A_BADARG;
	p = port_lookup(prp_serial(Port));
	if (p == 0)
		return A_BADARG;
	size = (apr_size_t)int_value(Len);
	buf = xalloc(proc_gc_pool(ctx), size);
	rs = p->read(p, buf, &size);
	if (size == 0 && APR_STATUS_IS_EOF(rs))
		result(A_EOF);
	else if (size == 0 && rs != 0)
		return decipher_status(rs);
	else
	{
		bin = make_binary(intnum(size), buf, proc_gc_pool(ctx));
		result(bin);
	}
	return AI_OK;
}

// Called by parent process.
static std::string read_pipe(apr_file_t* in, bool timeout_ok = false)
{
	char buf[256];
	unsigned char bytes_to_read;
	apr_size_t bytes_read;
	apr_status_t status = apr_file_read_full(in, &bytes_to_read, 1, &bytes_read);
	if (status != APR_SUCCESS)
	{
		if (APR_STATUS_IS_TIMEUP(status) && timeout_ok)
		{
			return "TIMEOUT";
		}
		llwarns << "apr_file_read_full: " << apr_strerror(status, buf, sizeof(buf)) << llendl;
		assert(APR_STATUS_IS_EOF(status));
		return "END OF FILE";
	}
	assert(bytes_read == 1);
	status = apr_file_read_full(in, buf, bytes_to_read, &bytes_read);
	if (status != APR_SUCCESS)
	{
		llwarns << "apr_file_read_full: " << apr_strerror(status, buf, sizeof(buf)) << llendl;
		assert(status == APR_SUCCESS);	// Fail
	}
	assert(bytes_read == bytes_to_read);

	std::string received(buf, bytes_read);
	llinfos << "Received: \"" << received << "\" (bytes read: " << bytes_read << ")" << llendl;
	return received;
}

apr_status_t h2_conn_run(struct h2_ctx *ctx, conn_rec *c)
{
    apr_status_t status;
    int mpm_state = 0;
    
    do {
        if (c->cs) {
            c->cs->sense = CONN_SENSE_DEFAULT;
        }
        status = h2_session_process(h2_ctx_session_get(ctx), async_mpm);
        
        if (c->cs) {
            c->cs->state = CONN_STATE_WRITE_COMPLETION;
        }
        if (APR_STATUS_IS_EOF(status)) {
            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, c,
                          "h2_session(%ld): process, closing conn", c->id);
            c->keepalive = AP_CONN_CLOSE;
        }
        else {
            c->keepalive = AP_CONN_KEEPALIVE;
        }
        
        if (ap_mpm_query(AP_MPMQ_MPM_STATE, &mpm_state)) {
            break;
        }
    } while (!async_mpm
             && c->keepalive == AP_CONN_KEEPALIVE 
             && mpm_state != AP_MPMQ_STOPPING);
    
    return DONE;
}

/* Ensure the buffer is prepared for reading. Will return APR_SUCCESS,
 * APR_EOF, or some failure code. *len is only set for EOF. */
static apr_status_t common_databuf_prep(serf_databuf_t *databuf,
                                        apr_size_t *len)
{
    apr_size_t readlen;
    apr_status_t status;

    /* if there is data in the buffer, then we're happy. */
    if (databuf->remaining > 0)
        return APR_SUCCESS;

    /* if we already hit EOF, then keep returning that. */
    if (APR_STATUS_IS_EOF(databuf->status)) {
        /* *data = NULL;   ?? */
        *len = 0;
        return APR_EOF;
    }

    /* refill the buffer */
    status = (*databuf->read)(databuf->read_baton, sizeof(databuf->buf),
                              databuf->buf, &readlen);
    if (SERF_BUCKET_READ_ERROR(status)) {
        return status;
    }

    databuf->current = databuf->buf;
    databuf->remaining = readlen;
    databuf->status = status;

    return APR_SUCCESS;
}

static apr_status_t serf_chunk_read_iovec(serf_bucket_t *bucket,
                                          apr_size_t requested,
                                          int vecs_size,
                                          struct iovec *vecs,
                                          int *vecs_used)
{
    chunk_context_t *ctx = bucket->data;
    apr_status_t status;

    /* Before proceeding, we need to fetch some data from the stream. */
    if (ctx->state == STATE_FETCH) {
        status = create_chunk(bucket);
        if (status) {
            return status;
        }
    }

    status = serf_bucket_read_iovec(ctx->chunk, requested, vecs_size, vecs,
                                    vecs_used);

    /* Mask EOF from aggregate bucket. */
    if (APR_STATUS_IS_EOF(status) && ctx->state == STATE_CHUNK) {
        status = ctx->last_status;
        ctx->state = STATE_FETCH;
    }

    return status;
}

static apr_status_t talkTalk(apr_socket_t *socket, apr_pool_t *parent)
{
    apr_pool_t *pool;
    apr_size_t len;
    char *buf;
    apr_status_t rv;

    if (apr_pool_create(&pool, parent) != APR_SUCCESS)
        return APR_ENOPOOL;


    buf = apr_palloc(pool, BUF_SIZE);
    if (!buf)
        return ENOMEM;

    do {
        len = BUF_SIZE;
        rv = apr_socket_recv(socket, buf, &len);
        if (APR_STATUS_IS_EOF(rv) || len == 0 || rv != APR_SUCCESS)
            break;
        rv = apr_socket_send(socket, buf, &len);
        if (len == 0 || rv != APR_SUCCESS)
            break;
    } while (rv == APR_SUCCESS);

    apr_pool_clear(pool);
    return APR_SUCCESS;
}

/* Implements svn_read_fn_t */
static svn_error_t *
sock_read_cb(void *baton, char *buffer, apr_size_t *len)
{
  sock_baton_t *b = baton;
  apr_status_t status;
  apr_interval_time_t interval;

  status = apr_socket_timeout_get(b->sock, &interval);
  if (status)
    return svn_error_wrap_apr(status, _("Can't get socket timeout"));

  /* Always block on read.
   * During pipelining, we set the timeout to 0 for some write
   * operations so that we can try them without blocking. If APR had
   * separate timeouts for read and write, we would only set the
   * write timeout, but it doesn't. So here, we revert back to blocking.
   */
  apr_socket_timeout_set(b->sock, -1);
  status = apr_socket_recv(b->sock, buffer, len);
  apr_socket_timeout_set(b->sock, interval);

  if (status && !APR_STATUS_IS_EOF(status))
    return svn_error_wrap_apr(status, _("Can't read from connection"));
  if (*len == 0)
    return svn_error_create(SVN_ERR_RA_SVN_CONNECTION_CLOSED, NULL, NULL);
  return SVN_NO_ERROR;
}

static apr_status_t serf_chunk_read(serf_bucket_t *bucket,
                                    apr_size_t requested,
                                    const char **data, apr_size_t *len)
{
    chunk_context_t *ctx = bucket->data;
    apr_status_t status;

    /* Before proceeding, we need to fetch some data from the stream. */
    if (ctx->state == STATE_FETCH) {
        status = create_chunk(bucket);
        if (status) {
            return status;
        }
    }

    status = serf_bucket_read(ctx->chunk, requested, data, len);

    /* Mask EOF from aggregate bucket. */
    if (APR_STATUS_IS_EOF(status) && ctx->state == STATE_CHUNK) {
        status = ctx->last_status;
        ctx->state = STATE_FETCH;
    }

    return status;
}

static apr_status_t
read_complete_body(request_rec *r, apr_bucket_brigade *kept_body)
{
    apr_bucket_brigade *tmp_bb;
    apr_bucket *t_bucket1, *t_bucket2;
    unsigned short eos_seen = 0;
    apr_status_t status;

    tmp_bb = apr_brigade_create(r->pool, r->connection->bucket_alloc);

    while (!eos_seen) {
        status = ap_get_brigade(
                        r->input_filters,
                        tmp_bb,
                        AP_MODE_READBYTES,
                        APR_BLOCK_READ,
                        HUGE_STRING_LEN);

        /* This means the filter discovered an error.
         * Furthermore input-filter already handeld the error and sends
         * something to the output chain.
         * For example ap_http_filter does this if LimitRequestBody is reached
         */
        if (status == AP_FILTER_ERROR) {
            apr_brigade_destroy(tmp_bb);
            return AP_FILTER_ERROR;
        }

        /* Cool no need to search for the eos bucket */
        if (APR_STATUS_IS_EOF(status)) {
            apr_brigade_destroy(tmp_bb);
            return APR_SUCCESS;
        }

        if (status != APR_SUCCESS) {
            apr_brigade_destroy(tmp_bb);
            return status;
        }

        ITER_BRIGADE(t_bucket1, tmp_bb) {

            apr_bucket_copy(t_bucket1, &t_bucket2);

            /* If SSL is used TRANSIENT buckets are returned.
             * However we need this bucket for a longer period than
             * this function call, hence 'setaside' the bucket.
             */
            if APR_BUCKET_IS_TRANSIENT(t_bucket2) {
                apr_bucket_setaside(t_bucket2, r->pool);
            }

            APR_BRIGADE_INSERT_TAIL(kept_body, t_bucket2);

            if (!eos_seen && APR_BUCKET_IS_EOS(t_bucket1)) {
                eos_seen = 1;
            }
        }
        apr_brigade_cleanup(tmp_bb);
    }

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 apr_status_t fortune_process(conn_rec *c,
                                    apr_procattr_t *pattr,
                                    apr_bucket_brigade *bb)
{
    apr_status_t rv;
    int argc = 0;
    const char *argv[APP_MAX_ARGC];
    apr_proc_t proc;
    apr_bucket *b;
    apr_pool_t *p = c->pool;

    fortune_conf_t *fconf = ap_get_module_config(c->base_server->module_config,
						 &fortune_module);

    argv[argc++] = fconf->progname;
    argv[argc++] = NULL;	/* @argvs should be null-terminated */

    if ((rv = apr_proc_create(&proc, fconf->progname,
                              (const char *const *) argv,
                              NULL, (apr_procattr_t *) pattr,
                              p)) != APR_SUCCESS) {
        return rv;
    }

    while (TRUE) {
        char buf[BUFSIZE] = { 0, };

        /* read the command's output through the pipe */
        rv = apr_file_gets(buf, sizeof(buf), proc.out);
        if (APR_STATUS_IS_EOF(rv)) {
            break;
        }
        b = apr_bucket_pool_create(apr_pstrdup(p, buf),
                                   strlen(buf), p, c->bucket_alloc);
        APR_BRIGADE_INSERT_TAIL(bb, b);
    }
    apr_file_close(proc.out);

    {
        int st;
        apr_exit_why_e why;

        rv = apr_proc_wait(&proc, &st, &why, APR_WAIT);
        if (APR_STATUS_IS_CHILD_DONE(rv)) {
            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, c,
                          "child done: why = %d, exit status = %d", why, st);
        }
        else {
            ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, c, "child notdone");
            return APR_EGENERAL;
        }
    }

    return APR_SUCCESS;
}

/* Implements svn_read_fn_t */
static svn_error_t *
file_read_cb(void *baton, char *buffer, apr_size_t *len)
{
  file_baton_t *b = baton;
  apr_status_t status = apr_file_read(b->in_file, buffer, len);

  if (status && !APR_STATUS_IS_EOF(status))
    return svn_error_wrap_apr(status, _("Can't read from connection"));
  if (*len == 0)
    return svn_error_create(SVN_ERR_RA_SVN_CONNECTION_CLOSED, NULL, NULL);
  return SVN_NO_ERROR;
}

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;
}

apr_status_t h2_conn_run(conn_rec *c)
{
    apr_status_t status;
    int mpm_state = 0;
    h2_session *session = h2_ctx_get_session(c);
    
    ap_assert(session);
    do {
        if (c->cs) {
            c->cs->sense = CONN_SENSE_DEFAULT;
            c->cs->state = CONN_STATE_HANDLER;
        }
    
        status = h2_session_process(session, async_mpm);
        
        if (APR_STATUS_IS_EOF(status)) {
            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, c, 
                          H2_SSSN_LOG(APLOGNO(03045), session, 
                          "process, closing conn"));
            c->keepalive = AP_CONN_CLOSE;
        }
        else {
            c->keepalive = AP_CONN_KEEPALIVE;
        }
        
        if (ap_mpm_query(AP_MPMQ_MPM_STATE, &mpm_state)) {
            break;
        }
    } while (!async_mpm
             && c->keepalive == AP_CONN_KEEPALIVE 
             && mpm_state != AP_MPMQ_STOPPING);

    if (c->cs) {
        switch (session->state) {
            case H2_SESSION_ST_INIT:
            case H2_SESSION_ST_IDLE:
            case H2_SESSION_ST_BUSY:
            case H2_SESSION_ST_WAIT:
                c->cs->state = CONN_STATE_WRITE_COMPLETION;
                break;
            case H2_SESSION_ST_CLEANUP:
            case H2_SESSION_ST_DONE:
            default:
                c->cs->state = CONN_STATE_LINGER;
            break;
        }
    }

    return APR_SUCCESS;
}

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 serf_chunk_peek(serf_bucket_t *bucket,
                                     const char **data,
                                     apr_size_t *len)
{
    chunk_context_t *ctx = bucket->data;
    apr_status_t status;

    status = serf_bucket_peek(ctx->chunk, data, len);

    /* Mask EOF from aggregate bucket. */
    if (APR_STATUS_IS_EOF(status) && ctx->state == STATE_CHUNK) {
        status = APR_EAGAIN;
    }

    return status;
}

/* Read method of CGI bucket: polls on stderr and stdout of the child,
 * sending any stderr output immediately away to the error log. */
static apr_status_t aikido_bucket_read(apr_bucket *b, const char **str,
                                    apr_size_t *len, apr_read_type_e block)
{
    struct aikido_bucket_data *data = b->data;
    apr_interval_time_t timeout;
    apr_status_t rv;
    int gotdata = 0;

    timeout = block == APR_NONBLOCK_READ ? 0 : data->r->server->timeout;

    do {
        const apr_pollfd_t *results;
        apr_int32_t num;

        rv = apr_pollset_poll(data->pollset, timeout, &num, &results);
        if (APR_STATUS_IS_TIMEUP(rv)) {
            if (timeout) {
                ap_log_rerror(APLOG_MARK, APLOG_WARNING, 0, data->r, 
                              "Timeout waiting for output from CGI script %s",
                              data->r->filename);
                return rv;
            }
            else {
                return APR_EAGAIN;
            }
        }
        else if (APR_STATUS_IS_EINTR(rv)) {
            continue;
        }
        else if (rv != APR_SUCCESS) {
            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, data->r, 
                          "poll failed waiting for CGI child");
            return rv;
        }

        for (; num; num--, results++) {
            rv = aikido_read_stdout(b, results[0].desc.f, str, len);
            if (APR_STATUS_IS_EOF(rv)) {
                rv = APR_SUCCESS;
            }
            gotdata = 1;
        }

    } while (!gotdata);

    return rv;
}

static apr_status_t serf_chunk_readline(serf_bucket_t *bucket,
                                         int acceptable, int *found,
                                         const char **data, apr_size_t *len)
{
    chunk_context_t *ctx = bucket->data;
    apr_status_t status;

    status = serf_bucket_readline(ctx->chunk, acceptable, found, data, len);

    /* Mask EOF from aggregate bucket. */
    if (APR_STATUS_IS_EOF(status) && ctx->state == STATE_CHUNK) {
        status = APR_EAGAIN;
        ctx->state = STATE_FETCH;
    }

    return status;
}

static ssize_t ra_neon_body_provider(void *userdata,
                                     char *buffer,
                                     size_t buflen)
{
  body_provider_baton_t *b = userdata;
  svn_ra_neon__request_t *req = b->req;
  apr_file_t *body_file = b->body_file;

  if (req->sess->callbacks &&
      req->sess->callbacks->cancel_func)
    SVN_RA_NEON__REQ_ERR
      (req, (req->sess->callbacks->cancel_func)(req->sess->callback_baton));

  if (req->err)
    return -1;

  svn_pool_clear(req->iterpool);
  if (buflen == 0)
    {
      /* This is the beginning of a new body pull. Rewind the file. */
      apr_off_t offset = 0;
      SVN_RA_NEON__REQ_ERR
        (b->req,
         svn_io_file_seek(body_file, APR_SET, &offset, req->iterpool));
      return (req->err ? -1 : 0);
    }
  else
    {
      apr_size_t nbytes = buflen;
      svn_error_t *err = svn_io_file_read(body_file, buffer, &nbytes,
                                          req->iterpool);
      if (err)
        {
          if (APR_STATUS_IS_EOF(err->apr_err))
            {
              svn_error_clear(err);
              return 0;
            }

          SVN_RA_NEON__REQ_ERR(req, err);
          return -1;
        }
      else
        return nbytes;
    }
}

/* Returns the amount read. */
static int bio_bucket_read(BIO *bio, char *in, int inlen)
{
    serf_ssl_context_t *ctx = bio->ptr;
    const char *data;
    apr_status_t status;
    apr_size_t len;

#ifdef SSL_VERBOSE
    printf("bio_bucket_read called for %d bytes\n", inlen);
#endif

    if (ctx->encrypt.status == SERF_ERROR_WAIT_CONN
        && BIO_should_read(ctx->bio)) {
#ifdef SSL_VERBOSE
        printf("bio_bucket_read waiting: (%d %d %d)\n",
           BIO_should_retry(ctx->bio), BIO_should_read(ctx->bio),
           BIO_get_retry_flags(ctx->bio));
#endif
        /* Falling back... */
        ctx->encrypt.exhausted_reset = 1;
        BIO_clear_retry_flags(bio);
    }

    status = serf_bucket_read(ctx->decrypt.pending, inlen, &data, &len);

    ctx->decrypt.status = status;
#ifdef SSL_VERBOSE
    printf("bio_bucket_read received %d bytes (%d)\n", len, status);
#endif

    if (!SERF_BUCKET_READ_ERROR(status)) {
        /* Oh suck. */
        if (len) {
            memcpy(in, data, len);
            return len;
        }
        if (APR_STATUS_IS_EOF(status)) {
            BIO_set_retry_read(bio);
            return -1;
        }
    }

    return -1;
}