@@ -24,7 +24,7 @@ */ #include "Bucket.h" -using namespace Passenger; +namespace Passenger { static void bucket_destroy(void *data); static apr_status_t bucket_read(apr_bucket *a, const char **str, apr_size_t *len, apr_read_type_e block); @@ -42,7 +42,8 @@ static const apr_bucket_type_t apr_bucket_type_passenger_pipe = { struct BucketData { Application::SessionPtr session; - apr_file_t *pipe; + PassengerBucketStatePtr state; + int stream; ~BucketData() { /* The session here is an ApplicationPoolServer::RemoteSession. @@ -71,15 +72,13 @@ bucket_destroy(void *data) { static apr_status_t bucket_read(apr_bucket *bucket, const char **str, apr_size_t *len, apr_read_type_e block) { - apr_file_t *pipe; char *buf; - apr_status_t ret; - - BucketData *data = (BucketData *) bucket->data; - pipe = data->pipe; - + ssize_t ret; + BucketData *data; + + data = (BucketData *) bucket->data; *str = NULL; - *len = APR_BUCKET_BUFF_SIZE; + *len = 0; if (block == APR_NONBLOCK_READ) { /* @@ -101,55 +100,72 @@ bucket_read(apr_bucket *bucket, const char **str, apr_size_t *len, apr_read_type return APR_EAGAIN; } - buf = (char *) apr_bucket_alloc(*len, bucket->list); // TODO: check for failure? - - do { - ret = apr_file_read(pipe, buf, len); - } while (APR_STATUS_IS_EAGAIN(ret)); - - if (ret != APR_SUCCESS && ret != APR_EOF) { - // ... we might want to set an error flag here ... - delete data; - bucket->data = NULL; - apr_bucket_free(buf); - return ret; + buf = (char *) apr_bucket_alloc(APR_BUCKET_BUFF_SIZE, bucket->list); + if (buf == NULL) { + return APR_ENOMEM; } - /* - * If there's more to read we have to keep the rest of the pipe - * for later. - */ - if (*len > 0) { + + do { + ret = read(data->stream, buf, APR_BUCKET_BUFF_SIZE); + } while (ret == -1 && errno == EINTR); + + if (ret > 0) { apr_bucket_heap *h; + data->state->bytesRead += ret; + *str = buf; + *len = ret; bucket->data = NULL; - /* Change the current bucket to refer to what we read */ + /* Change the current bucket (which is a Passenger Bucket) into a heap bucket + * that contains the data that we just read. This newly created heap bucket + * will be the first in the bucket list. + */ bucket = apr_bucket_heap_make(bucket, buf, *len, apr_bucket_free); h = (apr_bucket_heap *) bucket->data; h->alloc_len = APR_BUCKET_BUFF_SIZE; /* note the real buffer size */ + + /* And after this newly created bucket we insert a new Passenger Bucket + * which can read the next chunk from the stream. + */ APR_BUCKET_INSERT_AFTER(bucket, passenger_bucket_create( - data->session, pipe, bucket->list)); + data->session, data->state, bucket->list)); + + /* The newly created Passenger Bucket has a reference to the session + * object, so we can delete data here. + */ delete data; - } else { + + return APR_SUCCESS; + + } else if (ret == 0) { + data->state->completed = true; delete data; bucket->data = NULL; - apr_bucket_free(buf); bucket = apr_bucket_immortal_make(bucket, "", 0); *str = (const char *) bucket->data; - // if (rv != APR_EOF) { - // ... we might want to set an error flag here ... - // } + *len = 0; + return APR_SUCCESS; + + } else /* ret == -1 */ { + int e = errno; + data->state->completed = true; + data->state->errorCode = e; + delete data; + bucket->data = NULL; + apr_bucket_free(buf); + return APR_FROM_OS_ERROR(e); } - return APR_SUCCESS; } static apr_bucket * -passenger_bucket_make(apr_bucket *bucket, Application::SessionPtr session, apr_file_t *pipe) { +passenger_bucket_make(apr_bucket *bucket, Application::SessionPtr session, PassengerBucketStatePtr state) { BucketData *data = new BucketData(); data->session = session; - data->pipe = pipe; + data->stream = session->getStream(); + data->state = state; bucket->type = &apr_bucket_type_passenger_pipe; bucket->length = (apr_size_t)(-1); @@ -159,13 +175,14 @@ passenger_bucket_make(apr_bucket *bucket, Application::SessionPtr session, apr_f } apr_bucket * -passenger_bucket_create(Application::SessionPtr session, apr_file_t *pipe, apr_bucket_alloc_t *list) { +passenger_bucket_create(Application::SessionPtr session, PassengerBucketStatePtr state, apr_bucket_alloc_t *list) { apr_bucket *bucket; bucket = (apr_bucket *) apr_bucket_alloc(sizeof(*bucket), list); APR_BUCKET_INIT(bucket); bucket->free = apr_bucket_free; bucket->list = list; - return passenger_bucket_make(bucket, session, pipe); + return passenger_bucket_make(bucket, session, state); } +} // namespace Passenger ext/apache2/Bucket.cpp @@ -25,26 +25,63 @@ #ifndef _PASSENGER_BUCKET_H_ #define _PASSENGER_BUCKET_H_ -/** - * apr_bucket_pipe closes a pipe's file descriptor when it has reached - * end-of-stream, but not when an error has occurred. This behavior is - * undesirable because it can easily cause file descriptor leaks. - * - * passenger_bucket is like apr_bucket_pipe, but it also holds a reference to - * a Session. When a read error has occured or when end-of-stream has been - * reached, the Session will be dereferenced, so that the underlying file - * descriptor is closed. - * - * passenger_bucket also ignores the APR_NONBLOCK_READ flag because that's - * known to cause strange I/O problems. - */ - +#include <boost/shared_ptr.hpp> #include <apr_buckets.h> #include "Application.h" +namespace Passenger { + +using namespace boost; + +struct PassengerBucketState { + /** The number of bytes that this PassengerBucket has read so far. */ + unsigned long bytesRead; + + /** Whether this PassengerBucket is completed, i.e. no more data + * can be read from the underlying file descriptor. When true, + * this can either mean that EOF has been reached, or that an I/O + * error occured. Use errorCode to check whether an error occurred. + */ + bool completed; + + /** When completed is true, errorCode contains the errno value of + * the last read() call. + * + * A value of 0 means that no error occured. + */ + int errorCode; + + PassengerBucketState() { + bytesRead = 0; + completed = false; + errorCode = 0; + } +}; + +typedef shared_ptr<PassengerBucketState> PassengerBucketStatePtr; + +/** + * We used to use an apr_bucket_pipe for forwarding the backend process's + * response to the HTTP client. However, apr_bucket_pipe has a number of + * issues: + * - It closes the pipe's file descriptor when it has reached + * end-of-stream, but not when an error has occurred. This behavior is + * undesirable because it can easily cause file descriptor leaks. + * - It does weird non-blocking-I/O related things which can cause it + * to read less data than can actually be read. + * + * PassengerBucket is like apr_bucket_pipe, but: + * - It also holds a reference to a Session. When a read error has occured + * or when end-of-stream has been reached, the Session will be + * dereferenced, so that the underlying file descriptor is closed. + * - It ignores the APR_NONBLOCK_READ flag because that's known to cause + * strange I/O problems. + * - It can store its current state in a PassengerBucketState data structure. + */ apr_bucket *passenger_bucket_create(Passenger::Application::SessionPtr session, - apr_file_t *pipe, + PassengerBucketStatePtr state, apr_bucket_alloc_t *list); -#endif /* _PASSENGER_BUCKET_H_ */ +} // namespace Passenger +#endif /* _PASSENGER_BUCKET_H_ */ ext/apache2/Bucket.h @@ -447,8 +447,6 @@ private: try { this_thread::disable_interruption di; this_thread::disable_syscall_interruption dsi; - apr_bucket_brigade *bb; - apr_bucket *b; Application::SessionPtr session; bool expectingUploadData; shared_ptr<BufferedUpload> uploadData; @@ -554,21 +552,21 @@ private: /********** Step 4: forwarding the response from the backend process back to the HTTP client **********/ - /* Setup all the streams. */ UPDATE_TRACE_POINT(); - apr_file_t *readerPipe = NULL; - int reader = session->getStream(); - pid_t backendPid = session->getPid(); - apr_os_pipe_put(&readerPipe, &reader, r->pool); - apr_file_pipe_timeout_set(readerPipe, r->server->timeout); - + apr_bucket_brigade *bb; + apr_bucket *b; + PassengerBucketStatePtr bucketState; + pid_t backendPid; + /* Setup the bucket brigade. */ + bucketState = ptr(new PassengerBucketState()); bb = apr_brigade_create(r->connection->pool, r->connection->bucket_alloc); - b = passenger_bucket_create(session, readerPipe, r->connection->bucket_alloc); + b = passenger_bucket_create(session, bucketState, r->connection->bucket_alloc); /* The bucket (b) still has a reference to the session, so the reset() * call here is guaranteed not to throw any exceptions. */ + backendPid = session->getPid(); session.reset(); APR_BRIGADE_INSERT_TAIL(bb, b); @@ -610,6 +608,19 @@ private: UPDATE_TRACE_POINT(); ap_pass_brigade(r->output_filters, bb); + + if (r->connection->aborted) { + P_WARN("The HTTP client closed the connection before " + "the response could be completely sent. As a " + "result, you will probably see a 'Broken Pipe' " + "error in this log file. Please ignore it, " + "this is normal."); + } else if (!bucketState->completed) { + P_WARN("Apache stopped forwarding the backend's response, " + "even though the HTTP client did not close the " + "connection. Is this an Apache bug?"); + } + return OK; } else if (backendData[0] == '\0') { if ((long long) timer.elapsed() >= r->server->timeout / 1000) { ext/apache2/Hooks.cpp a702b5c4e30a3790a63a041e0bfa068b28f61fcf Hongli Lai (Phusion) hongli@phusion.nl http://github.com/FooBarWidget/passenger/commit/3fbcdffd5db60ecdd137c384480853c9debbe1aa 3fbcdffd5db60ecdd137c384480853c9debbe1aa 2009-05-31T07:37:17-07:00 2009-05-22T11:23:29-07:00 Do not use apr_file_t for reading the data from the backend process. Instead, use our own I/O functions because APR's are buggy. 9964fd30d338ad780e12c173433375f0d309850b Hongli Lai (Phusion) hongli@phusion.nl