WIP

Author: ylavic

include/scoreboard.h

@@ -149,6 +149,7 @@ struct process_score {
     apr_uint32_t keep_alive;        /* async connections in keep alive */
     apr_uint32_t suspended;         /* connections suspended by some module */
     apr_uint32_t read_line;         /* async connections doing read line */
+    apr_uint32_t pending;           /* connections waiting a worker */
 };
 
 /* Scoreboard is now in 'local' memory, since it isn't updated once created,

modules/filters/mod_reqtimeout.c

@@ -97,8 +97,8 @@ static void extend_timeout(reqtimeout_con_cfg *ccfg, apr_bucket_brigade *bb)
     }
 }
 
-static apr_status_t check_time_left(reqtimeout_con_cfg *ccfg,
-                                    apr_time_t now)
+static apr_status_t check_and_update_time_left(reqtimeout_con_cfg *ccfg,
+                                               apr_time_t now)
 {
     if (!now)
         now = apr_time_now();
@@ -209,11 +209,11 @@ static apr_status_t reqtimeout_filter(ap_filter_t *f,
         /* set new timeout */
         now = apr_time_now();
         ccfg->timeout_at = now + apr_time_from_sec(ccfg->cur_stage.timeout);
-        ccfg->cur_stage.timeout = 0;
         if (ccfg->cur_stage.max_timeout > 0) {
             ccfg->max_timeout_at = now + apr_time_from_sec(ccfg->cur_stage.max_timeout);
             ccfg->cur_stage.max_timeout = 0;
         }
+        ccfg->cur_stage.timeout = 0;
     }
     else if (ccfg->timeout_at == 0) {
         /* no timeout set, or in between requests */
@@ -227,25 +227,27 @@ static apr_status_t reqtimeout_filter(ap_filter_t *f,
     rv = apr_socket_timeout_get(ccfg->socket, &saved_sock_timeout);
     AP_DEBUG_ASSERT(rv == APR_SUCCESS);
 
-    rv = check_time_left(ccfg, now);
+    rv = check_and_update_time_left(ccfg, now);
     if (rv != APR_SUCCESS)
         goto cleanup;
 
     if (mode == AP_MODE_GETLINE && block == APR_BLOCK_READ) {
+        apr_off_t remaining = HUGE_STRING_LEN;
+#if APR_MAJOR_VERSION < 2
+        apr_int32_t nsds;
+        apr_interval_time_t poll_timeout;
+        apr_pollfd_t pollset;
+        pollset.p = NULL;
+#endif
+
         /*
          * For a blocking AP_MODE_GETLINE read, apr_brigade_split_line()
          * would loop until a whole line has been read. As this would make it
          * impossible to enforce a total timeout, we only do non-blocking
          * reads.
          */
-        apr_off_t remaining = HUGE_STRING_LEN;
         do {
             apr_off_t bblen;
-#if APR_MAJOR_VERSION < 2
-            apr_int32_t nsds;
-            apr_interval_time_t poll_timeout;
-            apr_pollfd_t pollset;
-#endif
 
             rv = ap_get_brigade(f->next, bb, AP_MODE_GETLINE, APR_NONBLOCK_READ, remaining);
             if (rv != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(rv)) {
@@ -282,10 +284,12 @@ static apr_status_t reqtimeout_filter(ap_filter_t *f,
 
             /* ... and wait for more */
 #if APR_MAJOR_VERSION < 2
-            pollset.p = f->c->pool;
-            pollset.desc_type = APR_POLL_SOCKET;
-            pollset.reqevents = APR_POLLIN|APR_POLLHUP;
-            pollset.desc.s = ccfg->socket;
+            if (pollset.p == NULL) {
+                pollset.p = f->c->pool;
+                pollset.desc_type = APR_POLL_SOCKET;
+                pollset.reqevents = APR_POLLIN | APR_POLLHUP | APR_POLLERR;
+                pollset.desc.s = ccfg->socket;
+            }
             apr_socket_timeout_get(ccfg->socket, &poll_timeout);
             rv = apr_poll(&pollset, 1, &nsds, poll_timeout);
 #else
@@ -294,7 +298,7 @@ static apr_status_t reqtimeout_filter(ap_filter_t *f,
             if (rv != APR_SUCCESS)
                 break;
 
-            rv = check_time_left(ccfg, 0);
+            rv = check_and_update_time_left(ccfg, 0);
             if (rv != APR_SUCCESS)
                 break;
 
@@ -306,12 +310,14 @@ static apr_status_t reqtimeout_filter(ap_filter_t *f,
     }
     else { /* mode != AP_MODE_GETLINE */
         rv = ap_get_brigade(f->next, bb, mode, block, readbytes);
+        
         /* Don't extend the timeout in speculative mode, wait for
          * the real (relevant) bytes to be asked later, within the
          * currently allotted time.
          */
-        if (ccfg->cur_stage.rate_factor && rv == APR_SUCCESS
-                && mode != AP_MODE_SPECULATIVE) {
+        if (rv == APR_SUCCESS
+                && mode != AP_MODE_SPECULATIVE
+                && ccfg->cur_stage.rate_factor) {
             extend_timeout(ccfg, bb);
         }
     }

modules/generators/mod_status.c

@@ -557,7 +557,8 @@ static int status_handler(request_rec *r)
         ap_rputs("</dl>", r);
 
     if (is_async) {
-        int read_line = 0, write_completion = 0, lingering_close = 0, keep_alive = 0,
+        int read_line = 0, write_completion = 0,
+            pending = 0, keep_alive = 0, lingering_close = 0,
             connections = 0, stopping = 0, procs = 0;
         /*
          * These differ from 'busy' and 'ready' in how gracefully finishing
@@ -574,13 +575,15 @@ static int status_handler(request_rec *r)
                          "<th colspan=\"3\">Async connections</th></tr>\n"
                      "<tr><th>total</th><th>accepting</th>"
                          "<th>busy</th><th>idle</th>"
-                         "<th>reading</th><th>writing</th><th>keep-alive</th><th>closing</th></tr>\n", r);
+                         "<th>reading</th><th>writing</th>"
+                         "<th>pending</th><th>keep-alive</th><th>closing</th></tr>\n", r);
         for (i = 0; i < server_limit; ++i) {
             ps_record = ap_get_scoreboard_process(i);
             if (ps_record->pid) {
                 connections      += ps_record->connections;
                 read_line        += ps_record->read_line;
                 write_completion += ps_record->write_completion;
+                pending          += ps_record->pending;
                 keep_alive       += ps_record->keep_alive;
                 lingering_close  += ps_record->lingering_close;
                 busy_workers     += thread_busy_buffer[i];
@@ -601,7 +604,8 @@ static int status_handler(request_rec *r)
                                       "<td>%s%s</td>"
                                       "<td>%u</td><td>%s</td>"
                                       "<td>%u</td><td>%u</td>"
-                                      "<td>%u</td><td>%u</td><td>%u</td><td>%u</td>"
+                                      "<td>%u</td><td>%u</td>"
+                                      "<td>%u</td><td>%u</td><td>%u</td>"
                                       "</tr>\n",
                                i, ps_record->pid,
                                dying, old,
@@ -611,6 +615,7 @@ static int status_handler(request_rec *r)
                                thread_idle_buffer[i],
                                ps_record->read_line,
                                ps_record->write_completion,
+                               ps_record->pending,
                                ps_record->keep_alive,
                                ps_record->lingering_close);
                 }
@@ -621,12 +626,14 @@ static int status_handler(request_rec *r)
                           "<td>%d</td><td>%d</td>"
                           "<td>%d</td><td>&nbsp;</td>"
                           "<td>%d</td><td>%d</td>"
-                          "<td>%d</td><td>%d</td><td>%d</td><td>%d</td>"
+                          "<td>%d</td><td>%d</td>"
+                          "<td>%d</td><td>%d</td><td>%d</td>"
                           "</tr>\n</table>\n",
                           procs, stopping,
                           connections,
                           busy_workers, idle_workers,
-                          read_line, write_completion, keep_alive, lingering_close);
+                          read_line, write_completion,
+                          pending, keep_alive, lingering_close);
         }
         else {
             ap_rprintf(r, "Processes: %d\n"
@@ -636,12 +643,14 @@ static int status_handler(request_rec *r)
                           "ConnsTotal: %d\n"
                           "ConnsAsyncReading: %d\n"
                           "ConnsAsyncWriting: %d\n"
+                          "ConnsAsyncPending: %d\n"
                           "ConnsAsyncKeepAlive: %d\n"
                           "ConnsAsyncClosing: %d\n",
                           procs, stopping,
                           busy_workers, idle_workers,
                           connections,
-                          read_line, write_completion, keep_alive, lingering_close);
+                          read_line, write_completion,
+                          pending, keep_alive, lingering_close);
         }
     }
 

modules/lua/lua_request.c

@@ -1264,10 +1264,18 @@ static int lua_ap_scoreboard_process(lua_State *L)
         lua_pushnumber(L, ps_record->suspended);
         lua_settable(L, -3);
 
+        lua_pushstring(L, "read_line");
+        lua_pushnumber(L, ps_record->read_line);
+        lua_settable(L, -3);
+
         lua_pushstring(L, "write_completion");
         lua_pushnumber(L, ps_record->write_completion);
         lua_settable(L, -3);
 
+        lua_pushstring(L, "pending");
+        lua_pushnumber(L, ps_record->pending);
+        lua_settable(L, -3);
+
         lua_pushstring(L, "not_accepting");
         lua_pushnumber(L, ps_record->not_accepting);
         lua_settable(L, -3);

server/mpm/event/event.c

@@ -690,7 +690,7 @@ static void * APR_THREAD_FUNC listener_thread(apr_thread_t *thd, void * dummy)
                 accept_mutex_error("unlock", rv, process_slot);
             }
             if (csd != NULL) {
-                rv = ap_queue_push_socket(worker_queue, csd, NULL, ptrans);
+                rv = ap_queue_push_socket(worker_queue, csd, ptrans);
                 if (rv) {
                     /* trash the connection; we couldn't queue the connected
                      * socket to a worker

server/mpm/worker/worker.c

@@ -40,8 +40,10 @@
 #include <apr_thread_cond.h>
 #include <apr_network_io.h>
 
+struct fd_queue_t;      /* opaque */
 struct fd_queue_info_t; /* opaque */
 struct fd_queue_elem_t; /* opaque */
+typedef struct fd_queue_t fd_queue_t;
 typedef struct fd_queue_info_t fd_queue_info_t;
 typedef struct fd_queue_elem_t fd_queue_elem_t;
 
@@ -50,9 +52,11 @@ AP_DECLARE(apr_status_t) ap_queue_info_create(fd_queue_info_t **queue_info,
                                               int max_recycled_pools);
 AP_DECLARE(apr_status_t) ap_queue_info_set_idle(fd_queue_info_t *queue_info,
                                                 apr_pool_t *pool_to_recycle);
+AP_DECLARE(apr_status_t) ap_queue_info_get_idler(fd_queue_info_t *queue_info);
 AP_DECLARE(apr_status_t) ap_queue_info_try_get_idler(fd_queue_info_t *queue_info);
 AP_DECLARE(apr_status_t) ap_queue_info_wait_for_idler(fd_queue_info_t *queue_info,
                                                       int *had_to_block);
+AP_DECLARE(apr_int32_t) ap_queue_info_avail(fd_queue_info_t *queue_info);
 AP_DECLARE(apr_uint32_t) ap_queue_info_num_idlers(fd_queue_info_t *queue_info);
 AP_DECLARE(apr_status_t) ap_queue_info_term(fd_queue_info_t *queue_info);
 
@@ -62,6 +66,43 @@ AP_DECLARE(void) ap_queue_info_push_pool(fd_queue_info_t *queue_info,
                                          apr_pool_t *pool_to_recycle);
 AP_DECLARE(void) ap_queue_info_free_idle_pools(fd_queue_info_t *queue_info);
 
+enum fd_queue_event_type_e
+{
+    FD_QUEUE_EVENT_SOCKET,
+    FD_QUEUE_EVENT_TIMER,
+};
+typedef enum fd_queue_event_type_e fd_queue_event_type_e;
+
+struct sock_event_t;
+struct timer_event_t;
+struct fd_queue_event_t
+{
+    /* queue container (used internally) */
+    fd_queue_elem_t *elem;
+
+    /* called back when (de)queuing (under the queue lock) */
+    void (*cb)(void *baton, int add);
+    void *baton;
+
+    /* event data */
+    fd_queue_event_type_e type;
+    union {
+        struct sock_event_t *se;
+        struct timer_event_t *te;
+    } data;
+};
+typedef struct fd_queue_event_t fd_queue_event_t;
+
+struct sock_event_t
+{
+    apr_socket_t *sd;
+    void *sd_baton;
+    apr_pool_t *p;
+
+    fd_queue_event_t qe; /* in-queue */
+};
+typedef struct sock_event_t sock_event_t;
+
 struct timer_event_t
 {
     APR_RING_ENTRY(timer_event_t) link;
@@ -71,36 +112,37 @@ struct timer_event_t
     int canceled;
     apr_array_header_t *pfds;
     apr_interval_time_t timeout;
+
+    fd_queue_event_t qe; /* in-queue */
 };
 typedef struct timer_event_t timer_event_t;
 
-struct fd_queue_t
-{
-    APR_RING_HEAD(timers_t, timer_event_t) timers;
-    fd_queue_elem_t *data;
-    unsigned int nelts;
-    unsigned int bounds;
-    unsigned int in;
-    unsigned int out;
-    apr_thread_mutex_t *one_big_mutex;
-    apr_thread_cond_t *not_empty;
-    volatile int terminated;
-};
-typedef struct fd_queue_t fd_queue_t;
+apr_status_t ap_queue_create(fd_queue_t **pqueue,
+                             int capacity, apr_pool_t *p);
+
+/* mpm_event API */
+AP_DECLARE(apr_status_t) ap_queue_push_event(fd_queue_t *queue,
+                                             fd_queue_event_t *event);
+AP_DECLARE(apr_status_t) ap_queue_pop_event(fd_queue_t *queue,
+                                            fd_queue_event_t **event);
+AP_DECLARE(apr_status_t) ap_queue_lock(fd_queue_t *queue);
+AP_DECLARE(apr_status_t) ap_queue_unlock(fd_queue_t *queue);
+AP_DECLARE(void) ap_queue_kill_event_locked(fd_queue_t *queue,
+                                            fd_queue_event_t *event);
 
-AP_DECLARE(apr_status_t) ap_queue_create(fd_queue_t **pqueue,
-                                         int capacity, apr_pool_t *p);
-AP_DECLARE(apr_status_t) ap_queue_push_socket(fd_queue_t *queue,
-                                              apr_socket_t *sd, void *sd_baton,
-                                              apr_pool_t *p);
-AP_DECLARE(apr_status_t) ap_queue_push_timer(fd_queue_t *queue,
-                                             timer_event_t *te);
+/* mpm_worker API */
+AP_DECLARE(apr_status_t) ap_queue_push_something(fd_queue_t *queue,
+                                                 apr_socket_t *sd, void *sd_baton,
+                                                 apr_pool_t *p, timer_event_t *te);
 AP_DECLARE(apr_status_t) ap_queue_pop_something(fd_queue_t *queue,
                                                 apr_socket_t **sd, void **sd_baton,
                                                 apr_pool_t **p, timer_event_t **te);
-#define                  ap_queue_pop_socket(q_, s_, p_) \
-                            ap_queue_pop_something((q_), (s_), NULL, (p_), NULL)
+#define ap_queue_push_socket(q_, s_, p_) \
+    ap_queue_push_something((q_), (s_), NULL, (p_), NULL)
+#define ap_queue_pop_socket(q_, s_, p_) \
+    ap_queue_pop_something((q_), (s_), NULL, (p_), NULL)
 
+/* common API */
 AP_DECLARE(apr_status_t) ap_queue_interrupt_all(fd_queue_t *queue);
 AP_DECLARE(apr_status_t) ap_queue_interrupt_one(fd_queue_t *queue);
 AP_DECLARE(apr_status_t) ap_queue_term(fd_queue_t *queue);

server/mpm_fdqueue.c

@@ -1819,7 +1819,7 @@ static void read_connection(struct connection * c)
     }
 
     /* are we done? */
-    if (started >= requests && (c->bread >= c->length)) {
+    if (done >= requests && (c->bread >= c->length)) {
         close_connection(c);
     }