a->sep.port = mp->port;
a->sep.transport_proto = mp->proto;
a->sep.peer.fib_index = mp->vrf;
+ a->sep.dscp = mp->dscp;
clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
if (mp->is_ip4)
{
next = clib_llist_next (fwrk->event_elts, evt_list, elt);
clib_llist_remove (fwrk->event_elts, evt_list, elt);
session_mq_connect_one (session_evt_ctrl_data (fwrk, elt));
+ session_evt_ctrl_data_free (fwrk, elt);
clib_llist_put (fwrk->event_elts, elt);
elt = next;
n_connects += 1;
}
static void
-session_mq_unlisten_handler (void *data)
+session_mq_unlisten_rpc (session_unlisten_msg_t *mp)
{
- session_unlisten_msg_t *mp = (session_unlisten_msg_t *) data;
+ vlib_main_t *vm = vlib_get_main ();
vnet_unlisten_args_t _a, *a = &_a;
app_worker_t *app_wrk;
+ session_handle_t sh;
application_t *app;
+ u32 context;
int rv;
- app_check_thread_and_barrier (session_mq_unlisten_handler, mp);
+ sh = mp->handle;
+ context = mp->context;
app = application_lookup (mp->client_index);
if (!app)
clib_memset (a, 0, sizeof (*a));
a->app_index = app->app_index;
- a->handle = mp->handle;
+ a->handle = sh;
a->wrk_map_index = mp->wrk_index;
+
+ vlib_worker_thread_barrier_sync (vm);
+
if ((rv = vnet_unlisten (a)))
clib_warning ("unlisten returned: %d", rv);
+ vlib_worker_thread_barrier_release (vm);
+
app_wrk = application_get_worker (app, a->wrk_map_index);
if (!app_wrk)
return;
- mq_send_unlisten_reply (app_wrk, mp->handle, mp->context, rv);
+ mq_send_unlisten_reply (app_wrk, sh, context, rv);
+ clib_mem_free (mp);
+}
+
+static void
+session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ u32 thread_index = wrk - session_main.wrk;
+ session_unlisten_msg_t *mp, *arg;
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ arg = clib_mem_alloc (sizeof (session_unlisten_msg_t));
+ clib_memcpy_fast (arg, mp, sizeof (*arg));
+
+ if (PREDICT_FALSE (!thread_index))
+ {
+ session_mq_unlisten_rpc (arg);
+ return;
+ }
+
+ session_send_rpc_evt_to_thread_force (0, session_mq_unlisten_rpc, arg);
}
static void
if (!session_has_transport (s))
{
s->session_state = SESSION_STATE_READY;
- if (ct_session_connect_notify (s))
+ if (ct_session_connect_notify (s, SESSION_E_NONE))
return;
}
else
return s;
}
-#define foreach_session_queue_error \
-_(TX, "Packets transmitted") \
-_(TIMER, "Timer events") \
-_(NO_BUFFER, "Out of buffers")
+#define foreach_session_queue_error \
+ _ (TX, tx, INFO, "Packets transmitted") \
+ _ (TIMER, timer, INFO, "Timer events") \
+ _ (NO_BUFFER, no_buffer, ERROR, "Out of buffers")
typedef enum
{
-#define _(sym,str) SESSION_QUEUE_ERROR_##sym,
+#define _(f, n, s, d) SESSION_QUEUE_ERROR_##f,
foreach_session_queue_error
#undef _
SESSION_QUEUE_N_ERROR,
} session_queue_error_t;
-static char *session_queue_error_strings[] = {
-#define _(sym,string) string,
+static vlib_error_desc_t session_error_counters[] = {
+#define _(f, n, s, d) { #n, d, VL_COUNTER_SEVERITY_##s },
foreach_session_queue_error
#undef _
};
n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
ASSERT (n_bytes_read > 0);
- if (ctx->s->session_state == SESSION_STATE_LISTENING)
+ if (transport_connection_is_cless (ctx->tc))
{
ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
ctx->tc->rmt_port = hdr->rmt_port;
/* Can retransmit for closed sessions but can't send new data if
* session is not ready or closed */
else if (s->session_state < SESSION_STATE_READY)
- return 1;
+ {
+ /* Allow accepting session to send custom packets.
+ * For instance, tcp want to send acks in established, but
+ * the app has not called accept() yet */
+ if (s->session_state == SESSION_STATE_ACCEPTING &&
+ (s->flags & SESSION_F_CUSTOM_TX))
+ return 0;
+ return 1;
+ }
else if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
{
/* Allow closed transports to still send custom packets.
session_mq_listen_uri_handler (session_evt_ctrl_data (wrk, elt));
break;
case SESSION_CTRL_EVT_UNLISTEN:
- session_mq_unlisten_handler (session_evt_ctrl_data (wrk, elt));
+ session_mq_unlisten_handler (wrk, elt);
break;
case SESSION_CTRL_EVT_CONNECT:
session_mq_connect_handler (wrk, elt);
s = session_event_get_session (wrk, e);
if (PREDICT_FALSE (!s))
break;
- CLIB_PREFETCH (s->tx_fifo, 2 * CLIB_CACHE_LINE_BYTES, LOAD);
+ CLIB_PREFETCH (s->tx_fifo, sizeof (*(s->tx_fifo)), LOAD);
wrk->ctx.s = s;
/* Spray packets in per session type frames, since they go to
* different nodes */
session_flush_pending_tx_buffers (session_worker_t * wrk,
vlib_node_runtime_t * node)
{
- vlib_buffer_enqueue_to_next (wrk->vm, node, wrk->pending_tx_buffers,
- wrk->pending_tx_nexts,
- vec_len (wrk->pending_tx_nexts));
+ vlib_buffer_enqueue_to_next_vec (wrk->vm, node, &wrk->pending_tx_buffers,
+ &wrk->pending_tx_nexts,
+ vec_len (wrk->pending_tx_nexts));
vec_reset_length (wrk->pending_tx_buffers);
vec_reset_length (wrk->pending_tx_nexts);
}
}
/* *INDENT-OFF* */
-VLIB_REGISTER_NODE (session_queue_node) =
-{
+VLIB_REGISTER_NODE (session_queue_node) = {
.function = session_queue_node_fn,
.flags = VLIB_NODE_FLAG_TRACE_SUPPORTED,
.name = "session-queue",
.format_trace = format_session_queue_trace,
.type = VLIB_NODE_TYPE_INPUT,
- .n_errors = ARRAY_LEN (session_queue_error_strings),
- .error_strings = session_queue_error_strings,
+ .n_errors = SESSION_QUEUE_N_ERROR,
+ .error_counters = session_error_counters,
.state = VLIB_NODE_STATE_DISABLED,
};
/* *INDENT-ON* */
Code Review / vpp.git / blobdiff