Bug #2648: Reserved helpers not shut down after reconfigure/rotate

The race happens if the helpers are restarted(rotate/reconfigure) while
reserved. Those reserved are then not shut down when the reservation is
released.

This patch cleans this up and a couple of other related races.

diff --git a/src/helper.cc b/src/helper.cc
index 101f869409b6242fe3a39c35501a0fe51b23c931..2a233ea457605a6f7a4b4a92325ee08be2d5a9a9 100644 (file)
--- a/src/helper.cc
+++ b/src/helper.cc
@@ -481,15 +481,14 @@ helperStatefulReset(helper_stateful_server * srv)
 
     if (srv->queue.head) {
         srv->flags.reserved = S_HELPER_DEFERRED;
-        helperStatefulServerKickQueue(srv);
     } else {
         srv->flags.reserved = S_HELPER_FREE;
 
         if ((srv->parent->OnEmptyQueue != NULL) && (srv->data))
             srv->parent->OnEmptyQueue(srv->data);
-
-        helperStatefulKickQueue(hlp);
     }
+
+    helperStatefulServerKickQueue(srv);
 }
 
 /*
@@ -529,6 +528,8 @@ helperStatefulReleaseServer(helper_stateful_server * srv)
     srv->flags.reserved = S_HELPER_FREE;
     if (srv->parent->OnEmptyQueue != NULL && srv->data)
         srv->parent->OnEmptyQueue(srv->data);
+
+    helperStatefulServerKickQueue(srv);
 }
 
 void *
@@ -861,8 +862,6 @@ helperServerFree(int fd, void *data)
     if (!concurrency)
         concurrency = 1;
 
-    assert(srv->rfd == fd);
-
     if (srv->rbuf) {
         memFreeBuf(srv->rbuf_sz, srv->rbuf);
         srv->rbuf = NULL;
@@ -929,7 +928,6 @@ helperStatefulServerFree(int fd, void *data)
     helper_stateful_server *srv = (helper_stateful_server *)data;
     statefulhelper *hlp = srv->parent;
     helper_stateful_request *r;
-    assert(srv->rfd == fd);
 
     if (srv->rbuf) {
         memFreeBuf(srv->rbuf_sz, srv->rbuf);
@@ -996,7 +994,6 @@ helperHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xerrno, voi
     char *t = NULL;
     helper_server *srv = (helper_server *)data;
     helper *hlp = srv->parent;
-    assert(fd == srv->rfd);
     assert(cbdataReferenceValid(data));
 
     /* Bail out early on COMM_ERR_CLOSING - close handlers will tidy up for us */
@@ -1005,6 +1002,8 @@ helperHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xerrno, voi
         return;
     }
 
+    assert(fd == srv->rfd);
+
     debugs(84, 5, "helperHandleRead: " << len << " bytes from " << hlp->id_name << " #" << srv->index + 1);
 
     if (flag != COMM_OK || len <= 0) {
@@ -1086,17 +1085,21 @@ helperHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xerrno, voi
         srv->roffset -= (t - srv->rbuf);
         memmove(srv->rbuf, t, srv->roffset + 1);
 
-        if (srv->flags.shutdown) {
+        if (!srv->flags.shutdown) {
+            helperKickQueue(hlp);
+        } else if (!srv->flags.closing && !srv->stats.pending) {
             int wfd = srv->wfd;
             srv->wfd = -1;
+            if (srv->rfd == wfd)
+                srv->rfd = -1;
             srv->flags.closing=1;
             comm_close(wfd);
             return;
-        } else
-            helperKickQueue(hlp);
+        }
     }
 
-    comm_read(fd, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1, helperHandleRead, srv);
+    if (srv->rfd != -1)
+        comm_read(fd, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1, helperHandleRead, srv);
 }
 
 static void
@@ -1106,7 +1109,6 @@ helperStatefulHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xer
     helper_stateful_server *srv = (helper_stateful_server *)data;
     helper_stateful_request *r;
     statefulhelper *hlp = srv->parent;
-    assert(fd == srv->rfd);
     assert(cbdataReferenceValid(data));
 
     /* Bail out early on COMM_ERR_CLOSING - close handlers will tidy up for us */
@@ -1115,6 +1117,8 @@ helperStatefulHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xer
         return;
     }
 
+    assert(fd == srv->rfd);
+
     debugs(84, 5, "helperStatefulHandleRead: " << len << " bytes from " <<
            hlp->id_name << " #" << srv->index + 1);
 
@@ -1216,22 +1220,11 @@ helperStatefulHandleRead(int fd, char *buf, size_t len, comm_err_t flag, int xer
                        tvSubMsec(srv->dispatch_time, current_time),
                        hlp->stats.replies, REDIRECT_AV_FACTOR);
 
-        if (srv->flags.shutdown
-                && srv->flags.reserved == S_HELPER_FREE
-                && !srv->deferred_requests) {
-            int wfd = srv->wfd;
-            srv->wfd = -1;
-            srv->flags.closing=1;
-            comm_close(wfd);
-        } else {
-            if (srv->queue.head)
-                helperStatefulServerKickQueue(srv);
-            else
-                helperStatefulKickQueue(hlp);
-        }
+        helperStatefulServerKickQueue(srv);
     }
 
-    comm_read(srv->rfd, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1,
+    if (srv->rfd != -1)
+        comm_read(srv->rfd, srv->rbuf + srv->roffset, srv->rbuf_sz - srv->roffset - 1,
               helperStatefulHandleRead, srv);
 }
 
@@ -1549,21 +1542,8 @@ helperStatefulDispatch(helper_stateful_server * srv, helper_stateful_request * r
         /* and push the queue. Note that the callback may have submitted a new
          * request to the helper which is why we test for the request*/
 
-        if (srv->request == NULL) {
-            if (srv->flags.shutdown
-                    && srv->flags.reserved == S_HELPER_FREE
-                    && !srv->deferred_requests) {
-                int wfd = srv->wfd;
-                srv->wfd = -1;
-                srv->flags.closing=1;
-                comm_close(wfd);
-            } else {
-                if (srv->queue.head)
-                    helperStatefulServerKickQueue(srv);
-                else
-                    helperStatefulKickQueue(hlp);
-            }
-        }
+        if (srv->request == NULL)
+            helperStatefulServerKickQueue(srv);
 
         return;
     }
@@ -1610,8 +1590,22 @@ helperStatefulServerKickQueue(helper_stateful_server * srv)
 {
     helper_stateful_request *r;
 
-    if ((r = StatefulServerDequeue(srv)))
+    if ((r = StatefulServerDequeue(srv))) {
         helperStatefulDispatch(srv, r);
+        return;
+    }
+
+    if (!srv->flags.shutdown) {
+        helperStatefulKickQueue(srv->parent);
+    } else if (!srv->flags.closing && srv->flags.reserved == S_HELPER_FREE && !srv->flags.busy) {
+        int wfd = srv->wfd;
+        srv->wfd = -1;
+        if (srv->rfd == wfd)
+            srv->rfd = -1;
+        srv->flags.closing=1;
+        comm_close(wfd);
+        return;
+    }
 }
 
 static void