http2: set default concurrency, fix ConnectionExists for multiplex

lib/http.c

@@ -176,6 +176,8 @@ CURLcode Curl_http_setup_conn(struct connectdata *conn)
   http->len = BUFSIZE;
   http->memlen = 0;
 
+  Curl_http2_setup_conn(conn);
+
   return CURLE_OK;
 }
 

lib/http2.c

@@ -93,6 +93,13 @@ static CURLcode http2_disconnect(struct connectdata *conn,
   return CURLE_OK;
 }
 
+/* called from Curl_http_setup_conn */
+void Curl_http2_setup_conn(struct connectdata *conn)
+{
+  conn->proto.httpc.settings.max_concurrent_streams =
+    DEFAULT_MAX_CONCURRENT_STREAMS;
+}
+
 /*
  * HTTP2 handler interface. This isn't added to the general list of protocols
  * but will be used at run-time when the protocol is dynamically switched from
@@ -302,6 +309,9 @@ static int on_frame_recv(nghttp2_session *session, const nghttp2_frame *frame,
                  httpc->settings.max_concurrent_streams));
     DEBUGF(infof(conn->data, "ENABLE_PUSH == %s\n",
                  httpc->settings.enable_push?"TRUE":"false"));
+    infof(conn->data,
+          "Connection state changed (MAX_CONCURRENT_STREAMS updated)!\n");
+    Curl_multi_connchanged(conn->data->multi);
     break;
   default:
     DEBUGF(infof(conn->data, "Got frame type %x for stream %x!\n",
@@ -1198,6 +1208,9 @@ CURLcode Curl_http2_setup(struct connectdata *conn)
   conn->httpversion = 20;
   conn->bundle->multiuse = BUNDLE_MULTIPLEX;
 
+  infof(conn->data, "Connection state changed (HTTP/2 confirmed)\n");
+  Curl_multi_connchanged(conn->data->multi);
+
   return CURLE_OK;
 }
 

lib/http2.h

@@ -7,7 +7,7 @@
  *                            | (__| |_| |  _ <| |___
  *                             \___|\___/|_| \_\_____|
  *
- * Copyright (C) 1998 - 2014, Daniel Stenberg, <daniel@haxx.se>, et al.
+ * Copyright (C) 1998 - 2015, Daniel Stenberg, <daniel@haxx.se>, et al.
  *
  * This software is licensed as described in the file COPYING, which
  * you should have received as part of this distribution. The terms
@@ -26,6 +26,11 @@
 
 #ifdef USE_NGHTTP2
 #include "http.h"
+
+/* value for MAX_CONCURRENT_STREAMS we use until we get an updated setting
+   from the peer */
+#define DEFAULT_MAX_CONCURRENT_STREAMS 13
+
 /*
  * Store nghttp2 version info in this buffer, Prefix with a space.  Return
  * total length written.
@@ -39,12 +44,15 @@ CURLcode Curl_http2_request_upgrade(Curl_send_buffer *req,
 CURLcode Curl_http2_setup(struct connectdata *conn);
 CURLcode Curl_http2_switched(struct connectdata *conn,
                              const char *data, size_t nread);
+/* called from Curl_http_setup_conn */
+void Curl_http2_setup_conn(struct connectdata *conn);
 #else /* USE_NGHTTP2 */
 #define Curl_http2_init(x) CURLE_UNSUPPORTED_PROTOCOL
 #define Curl_http2_send_request(x) CURLE_UNSUPPORTED_PROTOCOL
 #define Curl_http2_request_upgrade(x,y) CURLE_UNSUPPORTED_PROTOCOL
 #define Curl_http2_setup(x) CURLE_UNSUPPORTED_PROTOCOL
 #define Curl_http2_switched(x,y,z) CURLE_UNSUPPORTED_PROTOCOL
+#define Curl_http2_setup_conn(x)
 #endif
 
 #endif /* HEADER_CURL_HTTP2_H */

lib/multi.c

@@ -928,6 +928,34 @@ CURLMcode curl_multi_wait(CURLM *multi_handle,
   return CURLM_OK;
 }
 
+/*
+ * Curl_multi_connchanged() is called to tell that there is a connection in
+ * this multi handle that has changed state (pipelining become possible, the
+ * number of allowed streams changed or similar), and a subsequent use of this
+ * multi handle should move CONNECT_PEND handles back to CONNECT to have them
+ * retry.
+ */
+void Curl_multi_connchanged(struct Curl_multi *multi)
+{
+  multi->recheckstate = TRUE;
+}
+
+/*
+ * multi_ischanged() is called
+ *
+ * Returns TRUE/FALSE whether the state is changed to trigger a CONNECT_PEND
+ * => CONNECT action.
+ *
+ * Set 'clear' to TRUE to have it also clear the state variable.
+ */
+static bool multi_ischanged(struct Curl_multi *multi, bool clear)
+{
+  bool retval = multi->recheckstate;
+  if(clear)
+    multi->recheckstate = FALSE;
+  return retval;
+}
+
 static CURLMcode multi_runsingle(struct Curl_multi *multi,
                                  struct timeval now,
                                  struct SessionHandle *data)
@@ -979,6 +1007,11 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
       return CURLM_INTERNAL_ERROR;
     }
 
+    if(multi_ischanged(multi, TRUE)) {
+      DEBUGF(infof(data, "multi changed, check CONNECT_PEND queue!\n"));
+      Curl_multi_process_pending_handles(multi);
+    }
+
     if(data->easy_conn && data->mstate > CURLM_STATE_CONNECT &&
        data->mstate < CURLM_STATE_COMPLETED)
       /* Make sure we set the connection's current owner */
@@ -1750,7 +1783,7 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
 
       multistate(data, CURLM_STATE_MSGSENT);
     }
-  } while(rc == CURLM_CALL_MULTI_PERFORM);
+  } while((rc == CURLM_CALL_MULTI_PERFORM) || multi_ischanged(multi, FALSE));
 
   data->result = result;
 

lib/multihandle.h

@@ -102,6 +102,8 @@ struct Curl_multi {
   /* pipelining wanted bits (CURLPIPE*) */
   long pipelining;
 
+  bool recheckstate; /* see Curl_multi_connchanged */
+
   /* Shared connection cache (bundles)*/
   struct conncache conn_cache;
 
@@ -144,4 +146,3 @@ struct Curl_multi {
 };
 
 #endif /* HEADER_CURL_MULTIHANDLE_H */
-

lib/multiif.h

@@ -74,6 +74,8 @@ struct curl_llist *Curl_multi_pipelining_server_bl(struct Curl_multi *multi);
 /* Return the value of the CURLMOPT_MAX_TOTAL_CONNECTIONS option */
 size_t Curl_multi_max_total_connections(struct Curl_multi *multi);
 
+void Curl_multi_connchanged(struct Curl_multi *multi);
+
 /*
  * Curl_multi_closed()
  *

lib/url.c

@@ -3127,7 +3127,9 @@ ConnectionExists(struct SessionHandle *data,
      particular host */
   bundle = Curl_conncache_find_bundle(needle, data->state.conn_cache);
   if(bundle) {
-    size_t max_pipe_len = max_pipeline_length(data->multi);
+    /* Max pipe length is zero (unlimited) for multiplexed connections */
+    size_t max_pipe_len = (bundle->multiuse != BUNDLE_MULTIPLEX)?
+      max_pipeline_length(data->multi):0;
     size_t best_pipe_len = max_pipe_len;
     struct curl_llist_element *curr;
 
@@ -3352,22 +3354,43 @@ ConnectionExists(struct SessionHandle *data,
           }
 
           /* We can't use the connection if the pipe is full */
-          if(pipeLen >= max_pipe_len) {
-            infof(data, "Pipe is full, skip (%d)\n", pipeLen);
+          if(max_pipe_len && (pipeLen >= max_pipe_len)) {
+            infof(data, "Pipe is full, skip (%zu)\n", pipeLen);
             continue;
           }
 
-          /* We can't use the connection if the pipe is penalized */
-          if(Curl_pipeline_penalized(data, check))
-            continue;
+          /* If multiplexed, make sure we don't go over concurrency limit */
+          if(check->bits.multiplex) {
+            /* Multiplexed connections can only be HTTP/2 for now */
+            struct http_conn *httpc = &check->proto.httpc;
+            if(pipeLen >= httpc->settings.max_concurrent_streams) {
+              infof(data, "MAX_CONCURRENT_STREAMS reached, skip (%zu)\n",
+                    pipeLen);
+              continue;
+            }
+          }
 
-          if(pipeLen < best_pipe_len) {
-            /* This connection has a shorter pipe so far. We'll pick this
-               and continue searching */
-            chosen = check;
-            best_pipe_len = pipeLen;
+          /* We can't use the connection if the pipe is penalized */
+          if(Curl_pipeline_penalized(data, check)) {
+            infof(data, "Penalized, skip\n");
             continue;
           }
+
+          if(max_pipe_len) {
+            if(pipeLen < best_pipe_len) {
+              /* This connection has a shorter pipe so far. We'll pick this
+                 and continue searching */
+              chosen = check;
+              best_pipe_len = pipeLen;
+              continue;
+            }
+          }
+          else {
+            /* When not pipelining (== multiplexed), we have a match here! */
+            chosen = check;
+            infof(data, "Multiplexed connection found!\n");
+            break;
+          }
         }
         else {
           /* We have found a connection. Let's stop searching. */