pipeline: switch some code over to functions

... to "compartmentalize" a bit and make it easier to change behavior
when multiplexing is used instead of good old pipelining.

5 files changed, 118 insertions, 59 deletions

diff --git a/lib/http.c b/lib/http.c
index 5469cc067..e4ff4010a 100644
--- a/lib/http.c
+++ b/lib/http.c
@@ -1211,7 +1211,7 @@ CURLcode Curl_add_buffer_send(Curl_send_buffer *in,
         */
         return CURLE_SEND_ERROR;
       else
-        conn->writechannel_inuse = FALSE;
+        Curl_pipeline_leave_write(conn);
     }
   }
   Curl_add_buffer_free(in);
diff --git a/lib/multi.c b/lib/multi.c
index cff3b8d55..718f65816 100644
--- a/lib/multi.c
+++ b/lib/multi.c
@@ -69,8 +69,6 @@ static void singlesocket(struct Curl_multi *multi,
                          struct SessionHandle *data);
 static int update_timer(struct Curl_multi *multi);
 
-static bool isHandleAtHead(struct SessionHandle *handle,
-                           struct curl_llist *pipeline);
 static CURLMcode add_next_timeout(struct timeval now,
                                   struct Curl_multi *multi,
                                   struct SessionHandle *d);
@@ -1269,19 +1267,8 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
 
     case CURLM_STATE_WAITDO:
       /* Wait for our turn to DO when we're pipelining requests */
-#ifdef DEBUGBUILD
-      infof(data, "WAITDO: Conn %ld send pipe %zu inuse %s athead %s\n",
-            data->easy_conn->connection_id,
-            data->easy_conn->send_pipe->size,
-            data->easy_conn->writechannel_inuse?"TRUE":"FALSE",
-            isHandleAtHead(data,
-                           data->easy_conn->send_pipe)?"TRUE":"FALSE");
-#endif
-      if(!data->easy_conn->writechannel_inuse &&
-         isHandleAtHead(data,
-                        data->easy_conn->send_pipe)) {
-        /* Grab the channel */
-        data->easy_conn->writechannel_inuse = TRUE;
+      if(Curl_pipeline_checkget_write(data, data->easy_conn)) {
+        /* Grabbed the channel */
         multistate(data, CURLM_STATE_DO);
         rc = CURLM_CALL_MULTI_PERFORM;
       }
@@ -1464,25 +1451,11 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
 
     case CURLM_STATE_WAITPERFORM:
       /* Wait for our turn to PERFORM */
-      if((!data->easy_conn->readchannel_inuse &&
-          isHandleAtHead(data,
-                         data->easy_conn->recv_pipe)) ||
-         data->easy_conn->bits.multiplex) {
-        /* Grab the channel */
-        data->easy_conn->readchannel_inuse = TRUE;
+      if(Curl_pipeline_checkget_read(data, data->easy_conn)) {
+        /* Grabbed the channel */
         multistate(data, CURLM_STATE_PERFORM);
         rc = CURLM_CALL_MULTI_PERFORM;
       }
-#ifdef DEBUGBUILD
-      else {
-        infof(data, "WAITPERFORM: Conn %ld recv pipe %zu inuse %s athead %s\n",
-              data->easy_conn->connection_id,
-              data->easy_conn->recv_pipe->size,
-              data->easy_conn->readchannel_inuse?"TRUE":"FALSE",
-              isHandleAtHead(data,
-                             data->easy_conn->recv_pipe)?"TRUE":"FALSE");
-      }
-#endif
       break;
 
     case CURLM_STATE_TOOFAST: /* limit-rate exceeded in either direction */
@@ -1543,15 +1516,13 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
 
       k = &data->req;
 
-      if(!(k->keepon & KEEP_RECV)) {
+      if(!(k->keepon & KEEP_RECV))
         /* We're done receiving */
-        data->easy_conn->readchannel_inuse = FALSE;
-      }
+        Curl_pipeline_leave_read(data->easy_conn);
 
-      if(!(k->keepon & KEEP_SEND)) {
+      if(!(k->keepon & KEEP_SEND))
         /* We're done sending */
-        data->easy_conn->writechannel_inuse = FALSE;
-      }
+        Curl_pipeline_leave_write(data->easy_conn);
 
       if(done || (result == CURLE_RECV_ERROR)) {
         /* If CURLE_RECV_ERROR happens early enough, we assume it was a race
@@ -1733,8 +1704,8 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
 
         if(data->easy_conn) {
           /* if this has a connection, unsubscribe from the pipelines */
-          data->easy_conn->writechannel_inuse = FALSE;
-          data->easy_conn->readchannel_inuse = FALSE;
+          Curl_pipeline_leave_write(data->easy_conn);
+          Curl_pipeline_leave_read(data->easy_conn);
           Curl_removeHandleFromPipeline(data, data->easy_conn->send_pipe);
           Curl_removeHandleFromPipeline(data, data->easy_conn->recv_pipe);
 
@@ -2081,7 +2052,7 @@ static void singlesocket(struct Curl_multi *multi,
                for the recv_pipe, or the first (in case this particular easy
                isn't already) */
             if(entry->easy == data) {
-              if(isHandleAtHead(data, easy_conn->recv_pipe))
+              if(Curl_recvpipe_head(data, easy_conn))
                 entry->easy = easy_conn->recv_pipe->head->next->ptr;
               else
                 entry->easy = easy_conn->recv_pipe->head->ptr;
@@ -2095,7 +2066,7 @@ static void singlesocket(struct Curl_multi *multi,
                for the send_pipe, or the first (in case this particular easy
                isn't already) */
             if(entry->easy == data) {
-              if(isHandleAtHead(data, easy_conn->send_pipe))
+              if(Curl_sendpipe_head(data, easy_conn))
                 entry->easy = easy_conn->send_pipe->head->next->ptr;
               else
                 entry->easy = easy_conn->send_pipe->head->ptr;
@@ -2533,16 +2504,6 @@ static int update_timer(struct Curl_multi *multi)
   return multi->timer_cb((CURLM*)multi, timeout_ms, multi->timer_userp);
 }
 
-static bool isHandleAtHead(struct SessionHandle *handle,
-                           struct curl_llist *pipeline)
-{
-  struct curl_llist_element *curr = pipeline->head;
-  if(curr)
-    return (curr->ptr == handle) ? TRUE : FALSE;
-
-  return FALSE;
-}
-
 /*
  * multi_freetimeout()
  *
diff --git a/lib/pipeline.c b/lib/pipeline.c
index 744151d0b..1b38836cb 100644
--- a/lib/pipeline.c
+++ b/lib/pipeline.c
@@ -113,7 +113,7 @@ CURLcode Curl_add_handle_to_pipeline(struct SessionHandle *handle,
 
   if(pipeline == conn->send_pipe && sendhead != conn->send_pipe->head) {
     /* this is a new one as head, expire it */
-    conn->writechannel_inuse = FALSE; /* not in use yet */
+    Curl_pipeline_leave_write(conn); /* not in use yet */
     Curl_expire(conn->send_pipe->head->ptr, 1);
   }
 
@@ -144,7 +144,7 @@ void Curl_move_handle_from_send_to_recv_pipe(struct SessionHandle *handle,
       if(conn->send_pipe->head) {
         /* Since there's a new easy handle at the start of the send pipeline,
            set its timeout value to 1ms to make it trigger instantly */
-        conn->writechannel_inuse = FALSE; /* not used now */
+        Curl_pipeline_leave_write(conn); /* not used now */
 #ifdef DEBUGBUILD
         infof(conn->data, "%p is at send pipe head B!\n",
               (void *)conn->send_pipe->head->ptr);
@@ -320,6 +320,93 @@ CURLMcode Curl_pipeline_set_server_blacklist(char **servers,
   return CURLM_OK;
 }
 
+static bool pipe_head(struct SessionHandle *data,
+                      struct curl_llist *pipeline)
+{
+  struct curl_llist_element *curr = pipeline->head;
+  if(curr)
+    return (curr->ptr == data) ? TRUE : FALSE;
+
+  return FALSE;
+}
+
+/* returns TRUE if the given handle is head of the recv pipe */
+bool Curl_recvpipe_head(struct SessionHandle *data,
+                        struct connectdata *conn)
+{
+  return pipe_head(data, conn->recv_pipe);
+}
+
+/* returns TRUE if the given handle is head of the send pipe */
+bool Curl_sendpipe_head(struct SessionHandle *data,
+                        struct connectdata *conn)
+{
+  return pipe_head(data, conn->send_pipe);
+}
+
+
+/*
+ * Check if the write channel is available and this handle as at the head,
+ * then grab the channel and return TRUE.
+ *
+ * If not available, return FALSE.
+ */
+
+bool Curl_pipeline_checkget_write(struct SessionHandle *data,
+                                  struct connectdata *conn)
+{
+  if(conn->bits.multiplex)
+    /* when multiplexing, we can use it at once */
+    return TRUE;
+
+  if(!conn->writechannel_inuse && Curl_sendpipe_head(data, conn)) {
+    /* Grab the channel */
+    conn->writechannel_inuse = TRUE;
+    return TRUE;
+  }
+  return FALSE;
+}
+
+
+/*
+ * Check if the read channel is available and this handle as at the head, then
+ * grab the channel and return TRUE.
+ *
+ * If not available, return FALSE.
+ */
+
+bool Curl_pipeline_checkget_read(struct SessionHandle *data,
+                                 struct connectdata *conn)
+{
+  if(conn->bits.multiplex)
+    /* when multiplexing, we can use it at once */
+    return TRUE;
+
+  if(!conn->readchannel_inuse && Curl_recvpipe_head(data, conn)) {
+    /* Grab the channel */
+    conn->readchannel_inuse = TRUE;
+    return TRUE;
+  }
+  return FALSE;
+}
+
+/*
+ * The current user of the pipeline write channel gives it up.
+ */
+void Curl_pipeline_leave_write(struct connectdata *conn)
+{
+  conn->writechannel_inuse = FALSE;
+}
+
+/*
+ * The current user of the pipeline read channel gives it up.
+ */
+void Curl_pipeline_leave_read(struct connectdata *conn)
+{
+  conn->readchannel_inuse = FALSE;
+}
+
+
 #if 0
 void print_pipeline(struct connectdata *conn)
 {
diff --git a/lib/pipeline.h b/lib/pipeline.h
index 96c4c33ec..bf229f199 100644
--- a/lib/pipeline.h
+++ b/lib/pipeline.h
@@ -7,6 +7,7 @@
  *                            | (__| |_| |  _ <| |___
  *                             \___|\___/|_| \_\_____|
  *
+ * Copyright (C) 2015, Daniel Stenberg, <daniel@haxx.se>, et al.
  * Copyright (C) 2013 - 2014, Linus Nielsen Feltzing, <linus@haxx.se>
  *
  * This software is licensed as described in the file COPYING, which
@@ -41,4 +42,15 @@ bool Curl_pipeline_server_blacklisted(struct SessionHandle *handle,
 CURLMcode Curl_pipeline_set_server_blacklist(char **servers,
                                              struct curl_llist **list_ptr);
 
+bool Curl_pipeline_checkget_write(struct SessionHandle *data,
+                                  struct connectdata *conn);
+bool Curl_pipeline_checkget_read(struct SessionHandle *data,
+                                 struct connectdata *conn);
+void Curl_pipeline_leave_write(struct connectdata *conn);
+void Curl_pipeline_leave_read(struct connectdata *conn);
+bool Curl_recvpipe_head(struct SessionHandle *data,
+                        struct connectdata *conn);
+bool Curl_sendpipe_head(struct SessionHandle *data,
+                        struct connectdata *conn);
+
 #endif /* HEADER_CURL_PIPELINE_H */
diff --git a/lib/url.c b/lib/url.c
index a07aaa837..3425039de 100644
--- a/lib/url.c
+++ b/lib/url.c
@@ -2885,15 +2885,14 @@ void Curl_getoff_all_pipelines(struct SessionHandle *data,
                                struct connectdata *conn)
 {
   bool recv_head = (conn->readchannel_inuse &&
-    (gethandleathead(conn->recv_pipe) == data)) ? TRUE : FALSE;
-
+                    Curl_recvpipe_head(data, conn));
   bool send_head = (conn->writechannel_inuse &&
-    (gethandleathead(conn->send_pipe) == data)) ? TRUE : FALSE;
+                    Curl_sendpipe_head(data, conn));
 
   if(Curl_removeHandleFromPipeline(data, conn->recv_pipe) && recv_head)
-    conn->readchannel_inuse = FALSE;
+    Curl_pipeline_leave_read(conn);
   if(Curl_removeHandleFromPipeline(data, conn->send_pipe) && send_head)
-    conn->writechannel_inuse = FALSE;
+    Curl_pipeline_leave_write(conn);
 }
 
 static void signalPipeClose(struct curl_llist *pipeline, bool pipe_broke)