X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fplugins%2Fhttp%2Fhttp.c;h=752ca47a6913689f29491601067e19202bb78768;hb=ee4172ef0a115b6d96cfb0d5cc6ef4fb211c01aa;hp=0868027982e505a2707568cab085440863721c7d;hpb=340bd8f1efdaaf40d87553a46314c51aba074eb3;p=vpp.git diff --git a/src/plugins/http/http.c b/src/plugins/http/http.c index 0868027982e..752ca47a691 100644 --- a/src/plugins/http/http.c +++ b/src/plugins/http/http.c @@ -19,6 +19,17 @@ static http_main_t http_main; +#define HTTP_FIFO_THRESH (16 << 10) +#define CONTENT_LEN_STR "Content-Length: " + +/* HTTP state machine result */ +typedef enum http_sm_result_t_ +{ + HTTP_SM_STOP = 0, + HTTP_SM_CONTINUE = 1, + HTTP_SM_ERROR = -1, +} http_sm_result_t; + const char *http_status_code_str[] = { #define _(c, s, str) str, foreach_http_status_code @@ -31,6 +42,11 @@ const char *http_content_type_str[] = { #undef _ }; +const http_buffer_type_t msg_to_buf_type[] = { + [HTTP_MSG_DATA_INLINE] = HTTP_BUFFER_FIFO, + [HTTP_MSG_DATA_PTR] = HTTP_BUFFER_PTR, +}; + static inline http_worker_t * http_worker_get (u32 thread_index) { @@ -69,17 +85,27 @@ static u32 http_listener_alloc (void) { http_main_t *hm = &http_main; - http_conn_t *ctx; + http_conn_t *lhc; - pool_get_zero (hm->listener_ctx_pool, ctx); - ctx->c_c_index = ctx - hm->listener_ctx_pool; - return ctx->c_c_index; + pool_get_zero (hm->listener_pool, lhc); + lhc->c_c_index = lhc - hm->listener_pool; + return lhc->c_c_index; } http_conn_t * -http_listener_get (u32 ctx_index) +http_listener_get (u32 lhc_index) { - return pool_elt_at_index (http_main.listener_ctx_pool, ctx_index); + return pool_elt_at_index (http_main.listener_pool, lhc_index); +} + +void +http_listener_free (http_conn_t *lhc) +{ + http_main_t *hm = &http_main; + + if (CLIB_DEBUG) + memset (lhc, 0xfc, sizeof (*lhc)); + pool_put (hm->listener_pool, lhc); } void @@ -96,59 +122,6 @@ http_disconnect_transport (http_conn_t *hc) clib_warning ("disconnect returned"); } -static void -http_buffer_init (http_buffer_t *hb, svm_fifo_t *f, u32 data_len) -{ - hb->len = data_len; - hb->offset = 0; - hb->cur_seg = 0; - hb->src = f; - hb->segs = 0; -} - -static void -http_buffer_free (http_buffer_t *hb) -{ - hb->src = 0; - vec_free (hb->segs); -} - -svm_fifo_seg_t * -http_buffer_get_segs (http_buffer_t *hb, u32 max_len, u32 *n_segs) -{ - u32 _n_segs = 5; - int len; - - max_len = clib_max (hb->len - hb->offset, max_len); - - vec_validate (hb->segs, _n_segs); - - len = svm_fifo_segments (hb->src, 0, hb->segs, &_n_segs, max_len); - if (len < 0) - return 0; - - *n_segs = _n_segs; - - HTTP_DBG (1, "available to send %u n_segs %u", len, *n_segs); - - return hb->segs; -} - -void -http_buffer_drain (http_buffer_t *hb, u32 len) -{ - hb->offset += len; - svm_fifo_dequeue_drop (hb->src, len); - HTTP_DBG (1, "drained %u len %u offset %u", len, hb->len, hb->offset); -} - -static inline u8 -http_buffer_is_drained (http_buffer_t *hb) -{ - ASSERT (hb->offset <= hb->len); - return (hb->offset == hb->len); -} - static void http_conn_timeout_cb (void *hc_handlep) { @@ -174,7 +147,7 @@ http_ts_accept_callback (session_t *ts) session_t *ts_listener, *as, *asl; app_worker_t *app_wrk; http_conn_t *lhc, *hc; - u32 hc_index; + u32 hc_index, thresh; int rv; ts_listener = listen_session_get_from_handle (ts->listener_handle); @@ -183,14 +156,14 @@ http_ts_accept_callback (session_t *ts) hc_index = http_conn_alloc_w_thread (ts->thread_index); hc = http_conn_get_w_thread (hc_index, ts->thread_index); clib_memcpy_fast (hc, lhc, sizeof (*lhc)); - hc->c_thread_index = vlib_get_thread_index (); + hc->c_thread_index = ts->thread_index; hc->h_hc_index = hc_index; hc->h_tc_session_handle = session_handle (ts); hc->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP; hc->state = HTTP_CONN_STATE_ESTABLISHED; - hc->req_state = HTTP_REQ_STATE_WAIT_METHOD; + hc->http_state = HTTP_STATE_WAIT_METHOD; ts->session_state = SESSION_STATE_READY; ts->opaque = hc_index; @@ -199,7 +172,6 @@ http_ts_accept_callback (session_t *ts) * Alloc session and initialize */ as = session_alloc (hc->c_thread_index); - as->session_state = SESSION_STATE_CREATED; hc->c_s_index = as->session_index; as->app_wrk_index = hc->h_pa_wrk_index; @@ -234,16 +206,80 @@ http_ts_accept_callback (session_t *ts) return rv; } + /* Avoid enqueuing small chunks of data on transport tx notifications. If + * the fifo is small (under 16K) we set the threshold to it's size, meaning + * a notification will be given when the fifo empties. + */ + ts = session_get_from_handle (hc->h_tc_session_handle); + thresh = clib_min (svm_fifo_size (ts->tx_fifo), HTTP_FIFO_THRESH); + svm_fifo_set_deq_thresh (ts->tx_fifo, thresh); + http_conn_timer_start (hc); return 0; } static int -http_ts_connected_callback (u32 http_app_index, u32 hc_index, session_t *ts, +http_ts_connected_callback (u32 http_app_index, u32 ho_hc_index, session_t *ts, session_error_t err) { - clib_warning ("not supported"); + u32 new_hc_index; + session_t *as; + http_conn_t *hc, *ho_hc; + app_worker_t *app_wrk; + int rv; + + if (err) + { + clib_warning ("ERROR: %d", err); + return 0; + } + + new_hc_index = http_conn_alloc_w_thread (ts->thread_index); + hc = http_conn_get_w_thread (new_hc_index, ts->thread_index); + ho_hc = http_conn_get_w_thread (ho_hc_index, 0); + + ASSERT (ho_hc->state == HTTP_CONN_STATE_CONNECTING); + + clib_memcpy_fast (hc, ho_hc, sizeof (*hc)); + + hc->c_thread_index = ts->thread_index; + hc->h_tc_session_handle = session_handle (ts); + hc->c_c_index = new_hc_index; + hc->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP; + hc->state = HTTP_CONN_STATE_ESTABLISHED; + hc->http_state = HTTP_STATE_WAIT_APP; + + ts->session_state = SESSION_STATE_READY; + ts->opaque = new_hc_index; + + /* allocate app session and initialize */ + + as = session_alloc (hc->c_thread_index); + hc->c_s_index = as->session_index; + as->connection_index = hc->c_c_index; + as->app_wrk_index = hc->h_pa_wrk_index; + as->session_state = SESSION_STATE_READY; + as->session_type = session_type_from_proto_and_ip ( + TRANSPORT_PROTO_HTTP, session_type_is_ip4 (ts->session_type)); + + app_wrk = app_worker_get (hc->h_pa_wrk_index); + if (!app_wrk) + { + clib_warning ("no app worker"); + return -1; + } + + if ((rv = app_worker_init_connected (app_wrk, as))) + { + HTTP_DBG (1, "failed to allocate fifos"); + session_free (as); + return rv; + } + app_worker_connect_notify (app_wrk, as, err, hc->h_pa_app_api_ctx); + hc->h_pa_session_handle = session_handle (as); + http_conn_timer_start (hc); + return 0; } @@ -257,6 +293,7 @@ http_ts_disconnect_callback (session_t *ts) if (hc->state < HTTP_CONN_STATE_TRANSPORT_CLOSED) hc->state = HTTP_CONN_STATE_TRANSPORT_CLOSED; + /* Nothing more to rx, propagate to app */ if (!svm_fifo_max_dequeue_cons (ts->rx_fifo)) session_transport_closing_notify (&hc->connection); } @@ -264,15 +301,16 @@ http_ts_disconnect_callback (session_t *ts) static void http_ts_reset_callback (session_t *ts) { - http_conn_t *ctx; + http_conn_t *hc; - ctx = http_conn_get_w_thread (ts->opaque, ts->thread_index); + hc = http_conn_get_w_thread (ts->opaque, ts->thread_index); - if (ctx->state < HTTP_CONN_STATE_TRANSPORT_CLOSED) - ctx->state = HTTP_CONN_STATE_TRANSPORT_CLOSED; + hc->state = HTTP_CONN_STATE_CLOSED; + http_buffer_free (&hc->tx_buf); + hc->http_state = HTTP_STATE_WAIT_METHOD; + session_transport_reset_notify (&hc->connection); - if (!svm_fifo_max_dequeue_cons (ts->rx_fifo)) - session_transport_reset_notify (&ctx->connection); + http_disconnect_transport (hc); } /** @@ -293,7 +331,11 @@ static const char *http_response_template = "HTTP/1.1 200 OK\r\n" "Expires: %U GMT\r\n" "Server: VPP Static\r\n" "Content-Type: %s\r\n" - "Content-Length: %d\r\n\r\n"; + "Content-Length: %lu\r\n\r\n"; + +static const char *http_request_template = "GET %s HTTP/1.1\r\n" + "User-Agent: VPP HTTP client\r\n" + "Accept: */*\r\n"; static u32 send_data (http_conn_t *hc, u8 *data, u32 length, u32 offset) @@ -335,7 +377,7 @@ send_error (http_conn_t *hc, http_status_code_t ec) } static int -read_request (http_conn_t *hc) +read_http_message (http_conn_t *hc) { u32 max_deq, cursize; session_t *ts; @@ -355,7 +397,7 @@ read_request (http_conn_t *hc) if (svm_fifo_is_empty (ts->rx_fifo)) svm_fifo_unset_event (ts->rx_fifo); - _vec_len (hc->rx_buf) = cursize + n_read; + vec_set_len (hc->rx_buf, cursize + n_read); return 0; } @@ -363,7 +405,7 @@ static int v_find_index (u8 *vec, u32 offset, char *str) { int start_index = offset; - u32 slen = (u32) strnlen_s_inline (str, 8); + u32 slen = (u32) strnlen_s_inline (str, 16); u32 vlen = vec_len (vec); ASSERT (slen > 0); @@ -383,8 +425,8 @@ v_find_index (u8 *vec, u32 offset, char *str) /** * waiting for request method from peer - parse request method and data */ -static int -state_wait_method (http_conn_t *hc, transport_send_params_t *sp) +static http_sm_result_t +state_srv_wait_method (http_conn_t *hc, transport_send_params_t *sp) { http_status_code_t ec; app_worker_t *app_wrk; @@ -394,11 +436,11 @@ state_wait_method (http_conn_t *hc, transport_send_params_t *sp) u32 len; u8 *buf; - rv = read_request (hc); + rv = read_http_message (hc); /* Nothing yet, wait for data or timer expire */ if (rv) - return 0; + return HTTP_SM_STOP; if (vec_len (hc->rx_buf) < 8) { @@ -418,13 +460,13 @@ state_wait_method (http_conn_t *hc, transport_send_params_t *sp) goto error; } - len = i - hc->rx_buf_offset; + len = i - hc->rx_buf_offset - 1; } else if ((i = v_find_index (hc->rx_buf, 0, "POST ")) >= 0) { hc->method = HTTP_REQ_POST; hc->rx_buf_offset = i + 6; - len = vec_len (hc->rx_buf) - hc->rx_buf_offset; + len = vec_len (hc->rx_buf) - hc->rx_buf_offset - 1; } else { @@ -437,9 +479,9 @@ state_wait_method (http_conn_t *hc, transport_send_params_t *sp) msg.type = HTTP_MSG_REQUEST; msg.method_type = hc->method; - msg.data.content_type = HTTP_CONTENT_TEXT_HTML; + msg.content_type = HTTP_CONTENT_TEXT_HTML; + msg.data.type = HTTP_MSG_DATA_INLINE; msg.data.len = len; - msg.data.offset = 0; svm_fifo_seg_t segs[2] = { { (u8 *) &msg, sizeof (msg) }, { buf, len } }; @@ -451,16 +493,16 @@ state_wait_method (http_conn_t *hc, transport_send_params_t *sp) /* This should not happen as we only handle 1 request per session, * and fifo is allocated, but going forward we should consider * rescheduling */ - return -1; + return HTTP_SM_ERROR; } vec_free (hc->rx_buf); - hc->req_state = HTTP_REQ_STATE_WAIT_APP; + hc->http_state = HTTP_STATE_WAIT_APP; app_wrk = app_worker_get_if_valid (as->app_wrk_index); app_worker_lock_and_send_event (app_wrk, as, SESSION_IO_EVT_RX); - return 0; + return HTTP_SM_STOP; error: @@ -468,14 +510,14 @@ error: session_transport_closing_notify (&hc->connection); http_disconnect_transport (hc); - return -1; + return HTTP_SM_ERROR; } /** * waiting for data from app */ -static int -state_wait_app (http_conn_t *hc, transport_send_params_t *sp) +static http_sm_result_t +state_srv_wait_app (http_conn_t *hc, transport_send_params_t *sp) { http_main_t *hm = &http_main; http_status_code_t ec; @@ -491,7 +533,7 @@ state_wait_app (http_conn_t *hc, transport_send_params_t *sp) rv = svm_fifo_dequeue (as->tx_fifo, sizeof (msg), (u8 *) &msg); ASSERT (rv == sizeof (msg)); - if (msg.type != HTTP_MSG_REPLY) + if (msg.type != HTTP_MSG_REPLY || msg.data.type > HTTP_MSG_DATA_PTR) { clib_warning ("unexpected msg type from app %u", msg.type); ec = HTTP_STATUS_INTERNAL_ERROR; @@ -504,7 +546,8 @@ state_wait_app (http_conn_t *hc, transport_send_params_t *sp) goto error; } - http_buffer_init (&hc->tx_buf, as->tx_fifo, msg.data.len); + http_buffer_init (&hc->tx_buf, msg_to_buf_type[msg.data.type], as->tx_fifo, + msg.data.len); /* * Add headers. For now: @@ -520,7 +563,7 @@ state_wait_app (http_conn_t *hc, transport_send_params_t *sp) /* Expires */ format_clib_timebase_time, now + 600.0, /* Content type */ - http_content_type_str[msg.data.content_type], + http_content_type_str[msg.content_type], /* Length */ msg.data.len); @@ -534,23 +577,25 @@ state_wait_app (http_conn_t *hc, transport_send_params_t *sp) vec_free (header); /* Start sending the actual data */ - hc->req_state = HTTP_REQ_STATE_SEND_MORE_DATA; + hc->http_state = HTTP_STATE_IO_MORE_DATA; - return 1; + ASSERT (sp->max_burst_size >= offset); + sp->max_burst_size -= offset; + + return HTTP_SM_CONTINUE; error: send_error (hc, ec); - hc->req_state = HTTP_REQ_STATE_WAIT_METHOD; + hc->http_state = HTTP_STATE_WAIT_METHOD; session_transport_closing_notify (&hc->connection); http_disconnect_transport (hc); - /* stop state machine processing */ - return 0; + return HTTP_SM_STOP; } -static int -state_send_more_data (http_conn_t *hc, transport_send_params_t *sp) +static http_sm_result_t +state_srv_send_more_data (http_conn_t *hc, transport_send_params_t *sp) { u32 max_send = 64 << 10, n_segs; http_buffer_t *hb = &hc->tx_buf; @@ -558,6 +603,7 @@ state_send_more_data (http_conn_t *hc, transport_send_params_t *sp) session_t *ts; int sent = 0; + max_send = clib_min (max_send, sp->max_burst_size); ts = session_get_from_handle (hc->h_tc_session_handle); if ((seg = http_buffer_get_segs (hb, max_send, &n_segs))) sent = svm_fifo_enqueue_segments (ts->tx_fifo, seg, n_segs, @@ -565,14 +611,9 @@ state_send_more_data (http_conn_t *hc, transport_send_params_t *sp) if (sent > 0) { - http_buffer_drain (hb, sent); - /* Ask scheduler to notify app of deq event if needed */ - sp->max_burst_size = sent; - } - else - { - sp->max_burst_size = 0; + sp->bytes_dequeued += http_buffer_drain (hb, sent); + sp->max_burst_size -= sent; } /* Not finished sending all data */ @@ -581,7 +622,7 @@ state_send_more_data (http_conn_t *hc, transport_send_params_t *sp) if (sent && svm_fifo_set_event (ts->tx_fifo)) session_send_io_evt_to_thread (ts->tx_fifo, SESSION_IO_EVT_TX); - if (svm_fifo_max_enqueue (ts->tx_fifo) < 16 << 10) + if (svm_fifo_max_enqueue (ts->tx_fifo) < HTTP_FIFO_THRESH) { /* Deschedule http session and wait for deq notification if * underlying ts tx fifo almost full */ @@ -595,52 +636,331 @@ state_send_more_data (http_conn_t *hc, transport_send_params_t *sp) if (sent && svm_fifo_set_event (ts->tx_fifo)) session_send_io_evt_to_thread (ts->tx_fifo, SESSION_IO_EVT_TX_FLUSH); - /* Finished transaction, back to HTTP_REQ_STATE_WAIT_METHOD */ - hc->req_state = HTTP_REQ_STATE_WAIT_METHOD; + /* Finished transaction, back to HTTP_STATE_WAIT_METHOD */ + hc->http_state = HTTP_STATE_WAIT_METHOD; http_buffer_free (&hc->tx_buf); } + return HTTP_SM_STOP; +} + +static int +parse_http_header (http_conn_t *hc, int *content_length) +{ + unformat_input_t input; + int i, len; + u8 *line; + + if ((i = v_find_index (hc->rx_buf, hc->rx_buf_offset, "200 OK") < 0)) + { + clib_warning ("bad response code"); + return -1; + } + + i = v_find_index (hc->rx_buf, hc->rx_buf_offset, CONTENT_LEN_STR); + if (i < 0) + { + clib_warning ("cannot find '%s' in the header!", CONTENT_LEN_STR); + return -1; + } + + hc->rx_buf_offset = i; + + i = v_find_index (hc->rx_buf, hc->rx_buf_offset, "\n"); + if (i < 0) + { + clib_warning ("end of line missing; incomplete data"); + return -1; + } + + len = i - hc->rx_buf_offset; + line = vec_new (u8, len); + clib_memcpy (line, hc->rx_buf + hc->rx_buf_offset, len); + + unformat_init_vector (&input, line); + if (!unformat (&input, CONTENT_LEN_STR "%d", content_length)) + { + clib_warning ("failed to unformat content length!"); + return -1; + } + unformat_free (&input); + + /* skip rest of the header */ + hc->rx_buf_offset += len; + i = v_find_index (hc->rx_buf, hc->rx_buf_offset, ""); + if (i < 0) + { + clib_warning (" tag not found"); + return -1; + } + hc->rx_buf_offset = i; + return 0; } -typedef int (*http_sm_handler) (http_conn_t *, transport_send_params_t *sp); +static int +state_cln_wait_method (http_conn_t *hc, transport_send_params_t *sp) +{ + session_t *as; + http_msg_t msg; + app_worker_t *app_wrk; + int rv, content_length; + + rv = read_http_message (hc); + if (rv) + return HTTP_SM_STOP; + + msg.type = HTTP_MSG_REPLY; + msg.content_type = HTTP_CONTENT_TEXT_HTML; + msg.code = HTTP_STATUS_OK; + msg.data.type = HTTP_MSG_DATA_INLINE; + msg.data.len = 0; + + rv = parse_http_header (hc, &content_length); + if (rv) + { + clib_warning ("failed to parse http reply"); + session_transport_closing_notify (&hc->connection); + http_disconnect_transport (hc); + return -1; + } + + msg.data.len = content_length; + u32 dlen = vec_len (hc->rx_buf) - hc->rx_buf_offset; + as = session_get_from_handle (hc->h_pa_session_handle); + svm_fifo_seg_t segs[2] = { { (u8 *) &msg, sizeof (msg) }, + { &hc->rx_buf[hc->rx_buf_offset], dlen } }; + + rv = svm_fifo_enqueue_segments (as->rx_fifo, segs, 2, 0 /* allow partial */); + if (rv < 0) + { + clib_warning ("error enqueue"); + return HTTP_SM_ERROR; + } + hc->rx_buf_offset += dlen; + hc->http_state = HTTP_STATE_IO_MORE_DATA; + hc->to_recv = content_length - dlen; + + if (hc->rx_buf_offset == vec_len (hc->rx_buf)) + { + vec_reset_length (hc->rx_buf); + hc->rx_buf_offset = 0; + } + + if (hc->to_recv == 0) + { + hc->rx_buf_offset = 0; + vec_reset_length (hc->rx_buf); + hc->http_state = HTTP_STATE_WAIT_APP; + } + + app_wrk = app_worker_get_if_valid (as->app_wrk_index); + app_worker_lock_and_send_event (app_wrk, as, SESSION_IO_EVT_RX); + return HTTP_SM_STOP; +} + +static int +cln_drain_rx_buf (http_conn_t *hc, session_t *ts, session_t *as) +{ + app_worker_t *app_wrk; + u32 max_enq, n_enq, dlen = vec_len (hc->rx_buf) - hc->rx_buf_offset; + int rv; + + 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) + { + clib_warning ("enqueue failed"); + return -1; + } + + hc->rx_buf_offset += rv; -static http_sm_handler req_state_funcs[HTTP_REQ_N_STATES] = { + 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_lock_and_send_event (app_wrk, as, SESSION_IO_EVT_RX); + return 1; +} + +static http_sm_result_t +state_cln_recv_more_data (http_conn_t *hc, transport_send_params_t *sp) +{ + session_t *as; + u32 max_deq; + session_t *ts; + int n_read, rv; + + as = session_get_from_handle (hc->h_pa_session_handle); + ts = session_get_from_handle (hc->h_tc_session_handle); + + u32 dlen = vec_len (hc->rx_buf) - hc->rx_buf_offset; + if (dlen) + { + rv = cln_drain_rx_buf (hc, ts, as); + if (rv < 0) + { + clib_warning ("drain rx error!"); + return HTTP_SM_ERROR; + } + goto maybe_reschedule; + } + + if (hc->to_recv == 0) + { + ASSERT (vec_len (hc->rx_buf) == 0); + ASSERT (hc->rx_buf_offset == 0); + hc->http_state = HTTP_STATE_WAIT_APP; + return HTTP_SM_STOP; + } + + max_deq = svm_fifo_max_dequeue (ts->rx_fifo); + if (max_deq == 0) + return HTTP_SM_STOP; + + ASSERT (vec_len (hc->rx_buf) == 0); + ASSERT (hc->rx_buf_offset == 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); + + if (svm_fifo_is_empty (ts->rx_fifo)) + svm_fifo_unset_event (ts->rx_fifo); + + hc->to_recv -= n_read; + vec_set_len (hc->rx_buf, max_deq); + +maybe_reschedule: + if (hc->rx_buf_offset < vec_len (hc->rx_buf) || + svm_fifo_max_dequeue_cons (ts->rx_fifo)) + { + if (svm_fifo_set_event (ts->rx_fifo)) + session_send_io_evt_to_thread (ts->rx_fifo, SESSION_IO_EVT_BUILTIN_RX); + } + return HTTP_SM_CONTINUE; +} + +static http_sm_result_t +state_cln_wait_app (http_conn_t *hc, transport_send_params_t *sp) +{ + session_t *as; + http_msg_t msg; + http_status_code_t ec; + u8 *buf = 0, *request; + u32 offset; + int rv; + + as = session_get_from_handle (hc->h_pa_session_handle); + rv = svm_fifo_dequeue (as->tx_fifo, sizeof (msg), (u8 *) &msg); + ASSERT (rv == sizeof (msg)); + if (msg.type != HTTP_MSG_REQUEST || msg.data.type > HTTP_MSG_DATA_PTR) + { + clib_warning ("unexpected msg type from app %u", msg.type); + ec = HTTP_STATUS_INTERNAL_ERROR; + goto error; + } + + vec_validate (buf, msg.data.len - 1); + rv = svm_fifo_dequeue (as->tx_fifo, msg.data.len, buf); + ASSERT (rv == msg.data.len); + + request = format (0, http_request_template, buf); + offset = send_data (hc, request, vec_len (request), 0); + if (offset != vec_len (request)) + { + clib_warning ("sending request failed!"); + ec = HTTP_STATUS_INTERNAL_ERROR; + goto error; + } + + hc->http_state = HTTP_STATE_WAIT_METHOD; + + vec_free (buf); + vec_free (request); + + return HTTP_SM_CONTINUE; + +error: + send_error (hc, ec); + session_transport_closing_notify (&hc->connection); + http_disconnect_transport (hc); + return HTTP_SM_STOP; +} + +typedef http_sm_result_t (*http_sm_handler) (http_conn_t *, + transport_send_params_t *sp); + +static http_sm_handler srv_state_funcs[HTTP_N_STATES] = { /* Waiting for GET, POST, etc. */ - state_wait_method, + state_srv_wait_method, /* Wait for data from app */ - state_wait_app, + state_srv_wait_app, /* Send more data */ - state_send_more_data, + state_srv_send_more_data, +}; + +static http_sm_handler cln_state_funcs[HTTP_N_STATES] = { + /* wait for reply */ + state_cln_wait_method, + /* wait for data from app */ + state_cln_wait_app, + /* receive more data */ + state_cln_recv_more_data, }; static void http_req_run_state_machine (http_conn_t *hc, transport_send_params_t *sp) { - int rv; - + http_sm_result_t res; + http_sm_handler *state_fn = + hc->is_client ? cln_state_funcs : srv_state_funcs; do { - rv = req_state_funcs[hc->req_state](hc, sp); - if (rv < 0) + res = state_fn[hc->http_state](hc, sp); + if (res == HTTP_SM_ERROR) return; } - while (rv); + while (res == HTTP_SM_CONTINUE); /* Reset the session expiration timer */ http_conn_timer_update (hc); } static int -http_ts_rx_callback (session_t *ts) +http_ts_server_rx_callback (session_t *ts, http_conn_t *hc) { - http_conn_t *hc; + if (hc->http_state != HTTP_STATE_WAIT_METHOD) + { + clib_warning ("tcp data in req state %u", hc->http_state); + return 0; + } - hc = http_conn_get_w_thread (ts->opaque, ts->thread_index); + http_req_run_state_machine (hc, 0); - if (hc->req_state != HTTP_REQ_STATE_WAIT_METHOD) + if (hc->state == HTTP_CONN_STATE_TRANSPORT_CLOSED) { - clib_warning ("tcp data in req state %u", hc->req_state); + if (!svm_fifo_max_dequeue_cons (ts->rx_fifo)) + session_transport_closing_notify (&hc->connection); + } + return 0; +} + +static int +http_ts_client_rx_callback (session_t *ts, http_conn_t *hc) +{ + if (hc->http_state != HTTP_STATE_WAIT_METHOD && + hc->http_state != HTTP_STATE_IO_MORE_DATA) + { + clib_warning ("http in unexpected state %d (ts %d)", hc->http_state, + ts->session_index); return 0; } @@ -654,6 +974,17 @@ http_ts_rx_callback (session_t *ts) return 0; } +static int +http_ts_rx_callback (session_t *ts) +{ + http_conn_t *hc; + + hc = http_conn_get_w_thread (ts->opaque, ts->thread_index); + if (hc->is_client) + return http_ts_client_rx_callback (ts, hc); + return http_ts_server_rx_callback (ts, hc); +} + int http_ts_builtin_tx_callback (session_t *ts) { @@ -717,12 +1048,10 @@ static session_cb_vft_t http_app_cb_vft = { static clib_error_t * http_transport_enable (vlib_main_t *vm, u8 is_en) { - u32 add_segment_size = 256 << 20, first_seg_size = 32 << 20; vnet_app_detach_args_t _da, *da = &_da; vnet_app_attach_args_t _a, *a = &_a; u64 options[APP_OPTIONS_N_OPTIONS]; http_main_t *hm = &http_main; - u32 fifo_size = 128 << 12; if (!is_en) { @@ -734,9 +1063,6 @@ http_transport_enable (vlib_main_t *vm, u8 is_en) vec_validate (hm->wrk, vlib_num_workers ()); - first_seg_size = hm->first_seg_size ? hm->first_seg_size : first_seg_size; - fifo_size = hm->fifo_size ? hm->fifo_size : fifo_size; - clib_memset (a, 0, sizeof (*a)); clib_memset (options, 0, sizeof (options)); @@ -744,10 +1070,10 @@ http_transport_enable (vlib_main_t *vm, u8 is_en) a->api_client_index = APP_INVALID_INDEX; a->options = options; a->name = format (0, "http"); - a->options[APP_OPTIONS_SEGMENT_SIZE] = first_seg_size; - a->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = add_segment_size; - a->options[APP_OPTIONS_RX_FIFO_SIZE] = fifo_size; - a->options[APP_OPTIONS_TX_FIFO_SIZE] = fifo_size; + a->options[APP_OPTIONS_SEGMENT_SIZE] = hm->first_seg_size; + a->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = hm->add_seg_size; + a->options[APP_OPTIONS_RX_FIFO_SIZE] = hm->fifo_size; + a->options[APP_OPTIONS_TX_FIFO_SIZE] = hm->fifo_size; a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN; a->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_USE_GLOBAL_SCOPE; a->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_IS_TRANSPORT_APP; @@ -769,14 +1095,41 @@ http_transport_enable (vlib_main_t *vm, u8 is_en) static int http_transport_connect (transport_endpoint_cfg_t *tep) { - return -1; + vnet_connect_args_t _cargs, *cargs = &_cargs; + http_main_t *hm = &http_main; + session_endpoint_cfg_t *sep = (session_endpoint_cfg_t *) tep; + application_t *app; + http_conn_t *hc; + int error; + u32 hc_index; + app_worker_t *app_wrk = app_worker_get (sep->app_wrk_index); + + clib_memset (cargs, 0, sizeof (*cargs)); + clib_memcpy (&cargs->sep_ext, sep, sizeof (session_endpoint_cfg_t)); + cargs->sep.transport_proto = TRANSPORT_PROTO_TCP; + cargs->app_index = hm->app_index; + app = application_get (app_wrk->app_index); + cargs->sep_ext.ns_index = app->ns_index; + + hc_index = http_conn_alloc_w_thread (0 /* ts->thread_index */); + hc = http_conn_get_w_thread (hc_index, 0); + hc->h_pa_wrk_index = sep->app_wrk_index; + hc->h_pa_app_api_ctx = sep->opaque; + hc->is_client = 1; + hc->state = HTTP_CONN_STATE_CONNECTING; + cargs->api_context = hc_index; + + if ((error = vnet_connect (cargs))) + return error; + + return 0; } static u32 -http_start_listen (u32 app_listener_index, transport_endpoint_t *tep) +http_start_listen (u32 app_listener_index, transport_endpoint_cfg_t *tep) { vnet_listen_args_t _args = {}, *args = &_args; - session_t *tc_listener, *app_listener; + session_t *ts_listener, *app_listener; http_main_t *hm = &http_main; session_endpoint_cfg_t *sep; app_worker_t *app_wrk; @@ -806,18 +1159,41 @@ http_start_listen (u32 app_listener_index, transport_endpoint_t *tep) /* Grab transport connection listener and link to http listener */ lhc->h_tc_session_handle = args->handle; al = app_listener_get_w_handle (lhc->h_tc_session_handle); - tc_listener = app_listener_get_session (al); - tc_listener->opaque = lhc_index; + ts_listener = app_listener_get_session (al); + ts_listener->opaque = lhc_index; /* Grab application listener and link to http listener */ app_listener = listen_session_get (app_listener_index); lhc->h_pa_wrk_index = sep->app_wrk_index; lhc->h_pa_session_handle = listen_session_get_handle (app_listener); + lhc->c_s_index = app_listener_index; lhc->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP; return lhc_index; } +static u32 +http_stop_listen (u32 listener_index) +{ + http_conn_t *lhc; + int rv; + + lhc = http_listener_get (listener_index); + + vnet_unlisten_args_t a = { + .handle = lhc->h_tc_session_handle, + .app_index = http_main.app_index, + .wrk_map_index = 0 /* default wrk */ + }; + + if ((rv = vnet_unlisten (&a))) + clib_warning ("unlisten returned %d", rv); + + http_listener_free (lhc); + + return 0; +} + static void http_transport_close (u32 hc_index, u32 thread_index) { @@ -827,6 +1203,13 @@ http_transport_close (u32 hc_index, u32 thread_index) HTTP_DBG (1, "App disconnecting %x", hc_index); hc = http_conn_get_w_thread (hc_index, thread_index); + if (hc->state == HTTP_CONN_STATE_CONNECTING) + { + hc->state = HTTP_CONN_STATE_APP_CLOSED; + http_disconnect_transport (hc); + return; + } + as = session_get_from_handle (hc->h_pa_session_handle); /* Nothing more to send, confirm close */ @@ -860,25 +1243,44 @@ static int http_app_tx_callback (void *session, transport_send_params_t *sp) { session_t *as = (session_t *) session; + u32 max_burst_sz, sent; http_conn_t *hc; - sp->flags = 0; - hc = http_conn_get_w_thread (as->connection_index, as->thread_index); - if (hc->req_state < HTTP_REQ_STATE_WAIT_APP) + if (hc->http_state < HTTP_STATE_WAIT_APP) { - clib_warning ("app data in req state %u", hc->req_state); + if (hc->state != HTTP_CONN_STATE_CLOSED) + clib_warning ("app data req state %u session state %u", hc->http_state, + hc->state); + svm_fifo_dequeue_drop_all (as->tx_fifo); return 0; } + max_burst_sz = sp->max_burst_size * TRANSPORT_PACER_MIN_MSS; + sp->max_burst_size = max_burst_sz; + http_req_run_state_machine (hc, sp); - if (hc->state == HTTP_CONN_STATE_CLOSED) + if (hc->state == HTTP_CONN_STATE_APP_CLOSED) { - if (!svm_fifo_max_dequeue_cons (as->rx_fifo)) + if (!svm_fifo_max_dequeue_cons (as->tx_fifo)) http_disconnect_transport (hc); } - return 0; + + sent = max_burst_sz - sp->max_burst_size; + + return sent > 0 ? clib_max (sent / TRANSPORT_PACER_MIN_MSS, 1) : 0; +} + +static void +http_transport_get_endpoint (u32 hc_index, u32 thread_index, + transport_endpoint_t *tep, u8 is_lcl) +{ + http_conn_t *hc = http_conn_get_w_thread (hc_index, thread_index); + session_t *ts; + + ts = session_get_from_handle (hc->h_tc_session_handle); + session_get_endpoint (ts, tep, is_lcl); } static u8 * @@ -982,10 +1384,12 @@ static const transport_proto_vft_t http_proto = { .enable = http_transport_enable, .connect = http_transport_connect, .start_listen = http_start_listen, + .stop_listen = http_stop_listen, .close = http_transport_close, .custom_tx = http_app_tx_callback, .get_connection = http_transport_get_connection, .get_listener = http_transport_get_listener, + .get_transport_endpoint = http_transport_get_endpoint, .format_connection = format_http_transport_connection, .format_listener = format_http_transport_listener, .transport_options = { @@ -999,15 +1403,60 @@ static const transport_proto_vft_t http_proto = { static clib_error_t * http_transport_init (vlib_main_t *vm) { + http_main_t *hm = &http_main; + transport_register_protocol (TRANSPORT_PROTO_HTTP, &http_proto, FIB_PROTOCOL_IP4, ~0); transport_register_protocol (TRANSPORT_PROTO_HTTP, &http_proto, FIB_PROTOCOL_IP6, ~0); + + /* Default values, configurable via startup conf */ + hm->add_seg_size = 256 << 20; + hm->first_seg_size = 32 << 20; + hm->fifo_size = 512 << 10; + return 0; } VLIB_INIT_FUNCTION (http_transport_init); +static clib_error_t * +http_config_fn (vlib_main_t *vm, unformat_input_t *input) +{ + http_main_t *hm = &http_main; + uword mem_sz; + + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + if (unformat (input, "first-segment-size %U", unformat_memory_size, + &mem_sz)) + { + hm->first_seg_size = clib_max (mem_sz, 1 << 20); + if (hm->first_seg_size != mem_sz) + clib_warning ("first seg size too small %u", mem_sz); + } + else if (unformat (input, "add-segment-size %U", unformat_memory_size, + &mem_sz)) + { + hm->add_seg_size = clib_max (mem_sz, 1 << 20); + if (hm->add_seg_size != mem_sz) + clib_warning ("add seg size too small %u", mem_sz); + } + else if (unformat (input, "fifo-size %U", unformat_memory_size, &mem_sz)) + { + hm->fifo_size = clib_clamp (mem_sz, 4 << 10, 2 << 30); + if (hm->fifo_size != mem_sz) + clib_warning ("invalid fifo size %lu", mem_sz); + } + else + return clib_error_return (0, "unknown input `%U'", + format_unformat_error, input); + } + return 0; +} + +VLIB_CONFIG_FUNCTION (http_config_fn, "http"); + VLIB_PLUGIN_REGISTER () = { .version = VPP_BUILD_VER, .description = "Hypertext Transfer Protocol (HTTP)",