http: fix client receiving large data

HTTP client was relying on synchronous rx notifications to the client
app when moving lage data from underlying transport proto.
Recent change in session layer made such notifications asynchronous
making http client not working. This patch fixes the issue.

Type: fix

Change-Id: I4b24c6185a594a0fe8d5d87c149c53d3b40d7110
Signed-off-by: Filip Tehlar <ftehlar@cisco.com>
Signed-off-by: Matus Fabian <matfabia@cisco.com>

1 files changed, 63 insertions, 69 deletions

diff --git a/src/plugins/http/http.c b/src/plugins/http/http.c
index 036e6929987..0fa113c8155 100644
--- a/src/plugins/http/http.c
+++ b/src/plugins/http/http.c
@@ -74,14 +74,14 @@ format_http_state (u8 *s, va_list *va)
   return format (s, "unknown");
 }
 
-static inline void
-http_state_change (http_conn_t *hc, http_state_t state)
-{
-  HTTP_DBG (1, "changing http state %U -> %U", format_http_state,
-	    hc->http_state, format_http_state, state);
-  ASSERT (hc->http_state != state);
-  hc->http_state = state;
-}
+#define http_state_change(_hc, _state)                                        \
+  do                                                                          \
+    {                                                                         \
+      HTTP_DBG (1, "changing http state %U -> %U", format_http_state,         \
+		(_hc)->http_state, format_http_state, _state);                \
+      (_hc)->http_state = _state;                                             \
+    }                                                                         \
+  while (0)
 
 static inline http_worker_t *
 http_worker_get (u32 thread_index)
@@ -577,7 +577,7 @@ http_state_wait_server_reply (http_conn_t *hc, transport_send_params_t *sp)
 	{
 	  hc->rx_buf_offset = 0;
 	  vec_reset_length (hc->rx_buf);
-	  http_state_change (hc, HTTP_STATE_WAIT_CLIENT_METHOD);
+	  http_state_change (hc, HTTP_STATE_WAIT_APP_METHOD);
 	}
       else
 	{
@@ -585,7 +585,8 @@ http_state_wait_server_reply (http_conn_t *hc, transport_send_params_t *sp)
 	}
 
       app_wrk = app_worker_get_if_valid (as->app_wrk_index);
-      app_worker_rx_notify (app_wrk, as);
+      if (app_wrk)
+	app_worker_rx_notify (app_wrk, as);
       return HTTP_SM_STOP;
     }
   else
@@ -594,7 +595,6 @@ http_state_wait_server_reply (http_conn_t *hc, transport_send_params_t *sp)
       ec = HTTP_STATUS_METHOD_NOT_ALLOWED;
       goto error;
     }
-  return HTTP_SM_STOP;
 
 error:
 
@@ -791,7 +791,6 @@ error:
 static http_sm_result_t
 http_state_wait_app_method (http_conn_t *hc, transport_send_params_t *sp)
 {
-  http_status_code_t sc;
   http_msg_t msg;
   session_t *as;
   u8 *buf = 0, *request;
@@ -806,19 +805,15 @@ http_state_wait_app_method (http_conn_t *hc, transport_send_params_t *sp)
   if (msg.data.type > HTTP_MSG_DATA_PTR)
     {
       clib_warning ("no data");
-      sc = HTTP_STATUS_INTERNAL_ERROR;
       goto error;
     }
 
   if (msg.type != HTTP_MSG_REQUEST)
     {
       clib_warning ("unexpected message type %d", msg.type);
-      sc = HTTP_STATUS_INTERNAL_ERROR;
       goto error;
     }
 
-  sc = msg.code;
-
   vec_validate (buf, msg.data.len - 1);
   rv = svm_fifo_dequeue (as->tx_fifo, msg.data.len, buf);
   ASSERT (rv == msg.data.len);
@@ -828,7 +823,6 @@ http_state_wait_app_method (http_conn_t *hc, transport_send_params_t *sp)
   if (offset != vec_len (request))
     {
       clib_warning ("sending request failed!");
-      sc = HTTP_STATUS_INTERNAL_ERROR;
       goto error;
     }
 
@@ -837,83 +831,76 @@ http_state_wait_app_method (http_conn_t *hc, transport_send_params_t *sp)
   vec_free (buf);
   vec_free (request);
 
-  return HTTP_SM_CONTINUE;
+  return HTTP_SM_STOP;
 
 error:
-  clib_warning ("unexpected msg type from app %u", msg.type);
-  http_send_error (hc, sc);
   session_transport_closing_notify (&hc->connection);
   http_disconnect_transport (hc);
-  return HTTP_SM_STOP;
-}
-
-static void
-http_app_enqueue (http_conn_t *hc, session_t *as)
-{
-  app_worker_t *app_wrk;
-  u32 dlen, max_enq, n_enq;
-  int rv;
-
-  dlen = vec_len (hc->rx_buf) - hc->rx_buf_offset;
-  if (!dlen)
-    return;
-
-  max_enq = svm_fifo_max_enqueue (as->rx_fifo);
-  n_enq = clib_min (max_enq, dlen);
-  rv = svm_fifo_enqueue (as->rx_fifo, n_enq, &hc->rx_buf[hc->rx_buf_offset]);
-  if (rv < 0)
-    return;
-
-  hc->rx_buf_offset += rv;
-  if (hc->rx_buf_offset >= vec_len (hc->rx_buf))
-    {
-      vec_reset_length (hc->rx_buf);
-      hc->rx_buf_offset = 0;
-    }
-
-  app_wrk = app_worker_get_if_valid (as->app_wrk_index);
-  ASSERT (app_wrk);
-  app_worker_rx_notify (app_wrk, as);
+  return HTTP_SM_ERROR;
 }
 
 static http_sm_result_t
 http_state_client_io_more_data (http_conn_t *hc, transport_send_params_t *sp)
 {
   session_t *as, *ts;
-  u32 max_deq;
-  int n_read;
+  app_worker_t *app_wrk;
+  svm_fifo_seg_t _seg, *seg = &_seg;
+  u32 max_len, max_deq, max_enq, n_segs = 1;
+  int rv, len;
 
   as = session_get_from_handle (hc->h_pa_session_handle);
   ts = session_get_from_handle (hc->h_tc_session_handle);
 
-  http_app_enqueue (hc, as);
-
-  if (hc->to_recv == 0)
+  max_deq = svm_fifo_max_dequeue (ts->rx_fifo);
+  if (max_deq == 0)
     {
-      http_state_change (hc, HTTP_STATE_WAIT_CLIENT_METHOD);
+      HTTP_DBG (1, "no data to deq");
       return HTTP_SM_STOP;
     }
 
-  max_deq = svm_fifo_max_dequeue (ts->rx_fifo);
-  if (max_deq > 0)
+  max_enq = svm_fifo_max_enqueue (as->rx_fifo);
+  if (max_enq == 0)
     {
-      vec_validate (hc->rx_buf, max_deq - 1);
-      n_read = svm_fifo_dequeue (ts->rx_fifo, max_deq, hc->rx_buf);
-      ASSERT (n_read == max_deq);
+      HTTP_DBG (1, "app's rx fifo full");
+      svm_fifo_add_want_deq_ntf (as->rx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
+      return HTTP_SM_STOP;
+    }
 
-      if (svm_fifo_is_empty (ts->rx_fifo))
-	svm_fifo_unset_event (ts->rx_fifo);
+  max_len = clib_min (max_enq, max_deq);
+  len = svm_fifo_segments (ts->rx_fifo, 0, seg, &n_segs, max_len);
+  if (len < 0)
+    {
+      HTTP_DBG (1, "svm_fifo_segments() len %d", len);
+      return HTTP_SM_STOP;
+    }
 
-      hc->to_recv -= n_read;
-      vec_set_len (hc->rx_buf, n_read);
+  rv = svm_fifo_enqueue_segments (as->rx_fifo, seg, 1, 0 /* allow partial */);
+  if (rv < 0)
+    {
+      clib_warning ("data enqueue failed, rv: %d", rv);
+      return HTTP_SM_ERROR;
     }
 
-  if (hc->rx_buf_offset < vec_len (hc->rx_buf) ||
-      svm_fifo_max_dequeue_cons (ts->rx_fifo))
+  svm_fifo_dequeue_drop (ts->rx_fifo, rv);
+  if (rv > hc->to_recv)
     {
-	session_enqueue_notify (ts);
+      clib_warning ("http protocol error: received more data than expected");
+      session_transport_closing_notify (&hc->connection);
+      http_disconnect_transport (hc);
+      http_state_change (hc, HTTP_STATE_WAIT_APP_METHOD);
+      return HTTP_SM_ERROR;
     }
-  return HTTP_SM_CONTINUE;
+  hc->to_recv -= rv;
+  HTTP_DBG (1, "drained %d from ts; remains %d", rv, hc->to_recv);
+
+  app_wrk = app_worker_get_if_valid (as->app_wrk_index);
+  if (app_wrk)
+    app_worker_rx_notify (app_wrk, as);
+
+  if (svm_fifo_max_dequeue_cons (ts->rx_fifo))
+    session_enqueue_notify (ts);
+
+  return HTTP_SM_STOP;
 }
 
 static http_sm_result_t
@@ -983,6 +970,7 @@ static void
 http_req_run_state_machine (http_conn_t *hc, transport_send_params_t *sp)
 {
   http_sm_result_t res;
+
   do
     {
       res = state_funcs[hc->http_state](hc, sp);
@@ -1010,6 +998,12 @@ http_ts_rx_callback (session_t *ts)
       return -1;
     }
 
+  if (hc->state == HTTP_CONN_STATE_CLOSED)
+    {
+      svm_fifo_dequeue_drop_all (ts->tx_fifo);
+      return 0;
+    }
+
   http_req_run_state_machine (hc, 0);
 
   if (hc->state == HTTP_CONN_STATE_TRANSPORT_CLOSED)