src/vnet/session/application_worker.c

   1 /*
   2  * Copyright (c) 2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <vnet/session/application.h>
  17 #include <vnet/session/application_interface.h>
  18 #include <vnet/session/session.h>
  19
  20 /**
  21  * Pool of workers associated to apps
  22  */
  23 static app_worker_t *app_workers;
  24
  25 app_worker_t *
  26 app_worker_alloc (application_t * app)
  27 {
  28   app_worker_t *app_wrk;
  29   pool_get (app_workers, app_wrk);
  30   clib_memset (app_wrk, 0, sizeof (*app_wrk));
  31   app_wrk->wrk_index = app_wrk - app_workers;
  32   app_wrk->app_index = app->app_index;
  33   app_wrk->wrk_map_index = ~0;
  34   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
  35   clib_spinlock_init (&app_wrk->detached_seg_managers_lock);
  36   APP_DBG ("New app %v worker %u", app->name, app_wrk->wrk_index);
  37   return app_wrk;
  38 }
  39
  40 app_worker_t *
  41 app_worker_get (u32 wrk_index)
  42 {
  43   return pool_elt_at_index (app_workers, wrk_index);
  44 }
  45
  46 app_worker_t *
  47 app_worker_get_if_valid (u32 wrk_index)
  48 {
  49   if (pool_is_free_index (app_workers, wrk_index))
  50     return 0;
  51   return pool_elt_at_index (app_workers, wrk_index);
  52 }
  53
  54 void
  55 app_worker_free (app_worker_t * app_wrk)
  56 {
  57   application_t *app = application_get (app_wrk->app_index);
  58   vnet_unlisten_args_t _a, *a = &_a;
  59   u64 handle, *handles = 0, *sm_indices = 0;
  60   segment_manager_t *sm;
  61   session_t *ls;
  62   u32 sm_index;
  63   int i, j;
  64
  65   /*
  66    *  Listener cleanup
  67    */
  68
  69   /* *INDENT-OFF* */
  70   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
  71     ls = listen_session_get_from_handle (handle);
  72     vec_add1 (handles, app_listen_session_handle (ls));
  73     vec_add1 (sm_indices, sm_index);
  74     sm = segment_manager_get (sm_index);
  75   }));
  76   /* *INDENT-ON* */
  77
  78   for (i = 0; i < vec_len (handles); i++)
  79     {
  80       /* Cleanup listener */
  81       a->app_index = app->app_index;
  82       a->wrk_map_index = app_wrk->wrk_map_index;
  83       a->handle = handles[i];
  84       (void) vnet_unlisten (a);
  85
  86       sm = segment_manager_get_if_valid (sm_indices[i]);
  87       if (sm && !segment_manager_app_detached (sm))
  88         {
  89           sm->first_is_protected = 0;
  90           segment_manager_init_free (sm);
  91         }
  92     }
  93   vec_reset_length (handles);
  94   vec_free (sm_indices);
  95   hash_free (app_wrk->listeners_table);
  96
  97   /*
  98    * Connects segment manager cleanup
  99    */
 100
 101   if (app_wrk->connects_seg_manager != APP_INVALID_SEGMENT_MANAGER_INDEX)
 102     {
 103       sm = segment_manager_get (app_wrk->connects_seg_manager);
 104       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 105       sm->first_is_protected = 0;
 106       segment_manager_init_free (sm);
 107     }
 108
 109   /*
 110    * Half-open cleanup
 111    */
 112
 113   for (i = 0; i < vec_len (app_wrk->half_open_table); i++)
 114     {
 115       if (!app_wrk->half_open_table[i])
 116         continue;
 117
 118       /* *INDENT-OFF* */
 119       hash_foreach (handle, sm_index, app_wrk->half_open_table[i], ({
 120         vec_add1 (handles, handle);
 121       }));
 122       /* *INDENT-ON* */
 123
 124       for (j = 0; j < vec_len (handles); j++)
 125         session_cleanup_half_open (i, handles[j]);
 126
 127       hash_free (app_wrk->half_open_table[i]);
 128       vec_reset_length (handles);
 129     }
 130
 131   vec_free (app_wrk->half_open_table);
 132   vec_free (handles);
 133
 134   /*
 135    * Detached listener segment managers cleanup
 136    */
 137   for (i = 0; i < vec_len (app_wrk->detached_seg_managers); i++)
 138     {
 139       sm = segment_manager_get (app_wrk->detached_seg_managers[i]);
 140       segment_manager_init_free (sm);
 141     }
 142   vec_free (app_wrk->detached_seg_managers);
 143   clib_spinlock_free (&app_wrk->detached_seg_managers_lock);
 144
 145   if (CLIB_DEBUG)
 146     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
 147   pool_put (app_workers, app_wrk);
 148 }
 149
 150 application_t *
 151 app_worker_get_app (u32 wrk_index)
 152 {
 153   app_worker_t *app_wrk;
 154   app_wrk = app_worker_get_if_valid (wrk_index);
 155   if (!app_wrk)
 156     return 0;
 157   return application_get_if_valid (app_wrk->app_index);
 158 }
 159
 160 static segment_manager_t *
 161 app_worker_alloc_segment_manager (app_worker_t * app_wrk)
 162 {
 163   segment_manager_t *sm;
 164
 165   sm = segment_manager_alloc ();
 166   sm->app_wrk_index = app_wrk->wrk_index;
 167   segment_manager_init (sm);
 168   return sm;
 169 }
 170
 171 static int
 172 app_worker_alloc_session_fifos (segment_manager_t * sm, session_t * s)
 173 {
 174   svm_fifo_t *rx_fifo = 0, *tx_fifo = 0;
 175   int rv;
 176
 177   if ((rv = segment_manager_alloc_session_fifos (sm, s->thread_index,
 178                                                  &rx_fifo, &tx_fifo)))
 179     return rv;
 180
 181   rx_fifo->shr->master_session_index = s->session_index;
 182   rx_fifo->master_thread_index = s->thread_index;
 183
 184   tx_fifo->shr->master_session_index = s->session_index;
 185   tx_fifo->master_thread_index = s->thread_index;
 186
 187   s->rx_fifo = rx_fifo;
 188   s->tx_fifo = tx_fifo;
 189   return 0;
 190 }
 191
 192 int
 193 app_worker_init_listener (app_worker_t * app_wrk, session_t * ls)
 194 {
 195   segment_manager_t *sm;
 196
 197   /* Allocate segment manager. All sessions derived out of a listen session
 198    * have fifos allocated by the same segment manager. */
 199   if (!(sm = app_worker_alloc_segment_manager (app_wrk)))
 200     return SESSION_E_ALLOC;
 201
 202   /* Keep track of the segment manager for the listener or this worker */
 203   hash_set (app_wrk->listeners_table, listen_session_get_handle (ls),
 204             segment_manager_index (sm));
 205
 206   if (transport_connection_is_cless (session_get_transport (ls)))
 207     {
 208       if (ls->rx_fifo)
 209         return SESSION_E_NOSUPPORT;
 210       return app_worker_alloc_session_fifos (sm, ls);
 211     }
 212   return 0;
 213 }
 214
 215 int
 216 app_worker_start_listen (app_worker_t * app_wrk,
 217                          app_listener_t * app_listener)
 218 {
 219   session_t *ls;
 220   int rv;
 221
 222   if (clib_bitmap_get (app_listener->workers, app_wrk->wrk_map_index))
 223     return SESSION_E_ALREADY_LISTENING;
 224
 225   app_listener->workers = clib_bitmap_set (app_listener->workers,
 226                                            app_wrk->wrk_map_index, 1);
 227
 228   if (app_listener->session_index != SESSION_INVALID_INDEX)
 229     {
 230       ls = session_get (app_listener->session_index, 0);
 231       if ((rv = app_worker_init_listener (app_wrk, ls)))
 232         return rv;
 233     }
 234
 235   if (app_listener->local_index != SESSION_INVALID_INDEX)
 236     {
 237       ls = session_get (app_listener->local_index, 0);
 238       if ((rv = app_worker_init_listener (app_wrk, ls)))
 239         return rv;
 240     }
 241
 242   return 0;
 243 }
 244
 245 static void
 246 app_worker_add_detached_sm (app_worker_t * app_wrk, u32 sm_index)
 247 {
 248   vec_add1 (app_wrk->detached_seg_managers, sm_index);
 249 }
 250
 251 void
 252 app_worker_del_detached_sm (app_worker_t * app_wrk, u32 sm_index)
 253 {
 254   u32 i;
 255
 256   clib_spinlock_lock (&app_wrk->detached_seg_managers_lock);
 257   for (i = 0; i < vec_len (app_wrk->detached_seg_managers); i++)
 258     {
 259       if (app_wrk->detached_seg_managers[i] == sm_index)
 260         {
 261           vec_del1 (app_wrk->detached_seg_managers, i);
 262           break;
 263         }
 264     }
 265   clib_spinlock_unlock (&app_wrk->detached_seg_managers_lock);
 266 }
 267
 268 static void
 269 app_worker_stop_listen_session (app_worker_t * app_wrk, session_t * ls)
 270 {
 271   session_handle_t handle;
 272   segment_manager_t *sm;
 273   uword *sm_indexp;
 274   session_state_t *states = 0;
 275
 276   handle = listen_session_get_handle (ls);
 277   sm_indexp = hash_get (app_wrk->listeners_table, handle);
 278   if (PREDICT_FALSE (!sm_indexp))
 279     return;
 280
 281   /* Dealloc fifos, if any (dgram listeners) */
 282   if (ls->rx_fifo)
 283     {
 284       segment_manager_dealloc_fifos (ls->rx_fifo, ls->tx_fifo);
 285       ls->tx_fifo = ls->rx_fifo = 0;
 286     }
 287
 288   /* Try to cleanup segment manager */
 289   sm = segment_manager_get (*sm_indexp);
 290   if (sm)
 291     {
 292       segment_manager_app_detach (sm);
 293       if (!segment_manager_has_fifos (sm))
 294         {
 295           /* Empty segment manager, cleanup it up */
 296           segment_manager_free (sm);
 297         }
 298       else
 299         {
 300           /* Delete sessions in CREATED state */
 301           vec_add1 (states, SESSION_STATE_CREATED);
 302           segment_manager_del_sessions_filter (sm, states);
 303           vec_free (states);
 304
 305           /* Track segment manager in case app detaches and all the
 306            * outstanding sessions need to be closed */
 307           app_worker_add_detached_sm (app_wrk, *sm_indexp);
 308           sm->flags |= SEG_MANAGER_F_DETACHED_LISTENER;
 309         }
 310     }
 311
 312   hash_unset (app_wrk->listeners_table, handle);
 313 }
 314
 315 int
 316 app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al)
 317 {
 318   session_t *ls;
 319
 320   if (!clib_bitmap_get (al->workers, app_wrk->wrk_map_index))
 321     return 0;
 322
 323   if (al->session_index != SESSION_INVALID_INDEX)
 324     {
 325       ls = listen_session_get (al->session_index);
 326       app_worker_stop_listen_session (app_wrk, ls);
 327     }
 328
 329   if (al->local_index != SESSION_INVALID_INDEX)
 330     {
 331       ls = listen_session_get (al->local_index);
 332       app_worker_stop_listen_session (app_wrk, ls);
 333     }
 334
 335   clib_bitmap_set_no_check (al->workers, app_wrk->wrk_map_index, 0);
 336   if (clib_bitmap_is_zero (al->workers))
 337     app_listener_cleanup (al);
 338
 339   return 0;
 340 }
 341
 342 int
 343 app_worker_init_accepted (session_t * s)
 344 {
 345   app_worker_t *app_wrk;
 346   segment_manager_t *sm;
 347   session_t *listener;
 348   application_t *app;
 349
 350   listener = listen_session_get_from_handle (s->listener_handle);
 351   app_wrk = application_listener_select_worker (listener);
 352   s->app_wrk_index = app_wrk->wrk_index;
 353
 354   app = application_get (app_wrk->app_index);
 355   if (app->cb_fns.fifo_tuning_callback)
 356     s->flags |= SESSION_F_CUSTOM_FIFO_TUNING;
 357
 358   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
 359   if (app_worker_alloc_session_fifos (sm, s))
 360     return -1;
 361
 362   return 0;
 363 }
 364
 365 int
 366 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
 367 {
 368   application_t *app = application_get (app_wrk->app_index);
 369   return app->cb_fns.session_accept_callback (s);
 370 }
 371
 372 int
 373 app_worker_init_connected (app_worker_t * app_wrk, session_t * s)
 374 {
 375   application_t *app = application_get (app_wrk->app_index);
 376   segment_manager_t *sm;
 377
 378   if (app->cb_fns.fifo_tuning_callback)
 379     s->flags |= SESSION_F_CUSTOM_FIFO_TUNING;
 380
 381   /* Allocate fifos for session, unless the app is a builtin proxy */
 382   if (application_is_builtin_proxy (app))
 383     return 0;
 384
 385   sm = app_worker_get_connect_segment_manager (app_wrk);
 386   return app_worker_alloc_session_fifos (sm, s);
 387 }
 388
 389 int
 390 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s,
 391                            session_error_t err, u32 opaque)
 392 {
 393   application_t *app = application_get (app_wrk->app_index);
 394   return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
 395                                                  s, err);
 396 }
 397
 398 int
 399 app_worker_add_half_open (app_worker_t * app_wrk, transport_proto_t tp,
 400                           session_handle_t ho_handle,
 401                           session_handle_t wrk_handle)
 402 {
 403   ASSERT (vlib_get_thread_index () == 0);
 404   vec_validate (app_wrk->half_open_table, tp);
 405   hash_set (app_wrk->half_open_table[tp], ho_handle, wrk_handle);
 406   return 0;
 407 }
 408
 409 int
 410 app_worker_del_half_open (app_worker_t * app_wrk, transport_proto_t tp,
 411                           session_handle_t ho_handle)
 412 {
 413   ASSERT (vlib_get_thread_index () == 0);
 414   hash_unset (app_wrk->half_open_table[tp], ho_handle);
 415   return 0;
 416 }
 417
 418 u64
 419 app_worker_lookup_half_open (app_worker_t * app_wrk, transport_proto_t tp,
 420                              session_handle_t ho_handle)
 421 {
 422   u64 *ho_wrk_handlep;
 423
 424   /* No locking because all updates are done from main thread */
 425   ho_wrk_handlep = hash_get (app_wrk->half_open_table[tp], ho_handle);
 426   if (!ho_wrk_handlep)
 427     return SESSION_INVALID_HANDLE;
 428
 429   return *ho_wrk_handlep;
 430 }
 431
 432 int
 433 app_worker_close_notify (app_worker_t * app_wrk, session_t * s)
 434 {
 435   application_t *app = application_get (app_wrk->app_index);
 436   app->cb_fns.session_disconnect_callback (s);
 437   return 0;
 438 }
 439
 440 int
 441 app_worker_transport_closed_notify (app_worker_t * app_wrk, session_t * s)
 442 {
 443   application_t *app = application_get (app_wrk->app_index);
 444   if (app->cb_fns.session_transport_closed_callback)
 445     app->cb_fns.session_transport_closed_callback (s);
 446   return 0;
 447 }
 448
 449 int
 450 app_worker_reset_notify (app_worker_t * app_wrk, session_t * s)
 451 {
 452   application_t *app = application_get (app_wrk->app_index);
 453   app->cb_fns.session_reset_callback (s);
 454   return 0;
 455 }
 456
 457 int
 458 app_worker_cleanup_notify (app_worker_t * app_wrk, session_t * s,
 459                            session_cleanup_ntf_t ntf)
 460 {
 461   application_t *app = application_get (app_wrk->app_index);
 462   if (app->cb_fns.session_cleanup_callback)
 463     app->cb_fns.session_cleanup_callback (s, ntf);
 464   return 0;
 465 }
 466
 467 int
 468 app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s)
 469 {
 470   application_t *app = application_get (app_wrk->app_index);
 471   app->cb_fns.builtin_app_rx_callback (s);
 472   return 0;
 473 }
 474
 475 int
 476 app_worker_builtin_tx (app_worker_t * app_wrk, session_t * s)
 477 {
 478   application_t *app = application_get (app_wrk->app_index);
 479
 480   if (!app->cb_fns.builtin_app_tx_callback)
 481     return 0;
 482
 483   app->cb_fns.builtin_app_tx_callback (s);
 484   return 0;
 485 }
 486
 487 int
 488 app_worker_migrate_notify (app_worker_t * app_wrk, session_t * s,
 489                            session_handle_t new_sh)
 490 {
 491   application_t *app = application_get (app_wrk->app_index);
 492   app->cb_fns.session_migrate_callback (s, new_sh);
 493   return 0;
 494 }
 495
 496 int
 497 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
 498 {
 499   segment_manager_t *sm;
 500   svm_fifo_t *rxf, *txf;
 501
 502   if (s->session_state == SESSION_STATE_LISTENING)
 503     return application_change_listener_owner (s, app_wrk);
 504
 505   s->app_wrk_index = app_wrk->wrk_index;
 506
 507   rxf = s->rx_fifo;
 508   txf = s->tx_fifo;
 509
 510   if (!rxf || !txf)
 511     return 0;
 512
 513   s->rx_fifo = 0;
 514   s->tx_fifo = 0;
 515
 516   sm = app_worker_get_connect_segment_manager (app_wrk);
 517   if (app_worker_alloc_session_fifos (sm, s))
 518     return -1;
 519
 520   if (!svm_fifo_is_empty_cons (rxf))
 521     svm_fifo_clone (s->rx_fifo, rxf);
 522
 523   if (!svm_fifo_is_empty_cons (txf))
 524     svm_fifo_clone (s->tx_fifo, txf);
 525
 526   segment_manager_dealloc_fifos (rxf, txf);
 527
 528   return 0;
 529 }
 530
 531 int
 532 app_worker_connect_session (app_worker_t * app_wrk, session_endpoint_t * sep,
 533                             u32 api_context)
 534 {
 535   int rv;
 536
 537   if ((rv = session_open (app_wrk->wrk_index, sep, api_context)))
 538     return rv;
 539
 540   return 0;
 541 }
 542
 543 int
 544 app_worker_session_fifo_tuning (app_worker_t * app_wrk, session_t * s,
 545                                 svm_fifo_t * f,
 546                                 session_ft_action_t act, u32 len)
 547 {
 548   application_t *app = application_get (app_wrk->app_index);
 549   return app->cb_fns.fifo_tuning_callback (s, f, act, len);
 550 }
 551
 552 segment_manager_t *
 553 app_worker_get_connect_segment_manager (app_worker_t * app)
 554 {
 555   ASSERT (app->connects_seg_manager != (u32) ~ 0);
 556   return segment_manager_get (app->connects_seg_manager);
 557 }
 558
 559 segment_manager_t *
 560 app_worker_get_listen_segment_manager (app_worker_t * app,
 561                                        session_t * listener)
 562 {
 563   uword *smp;
 564   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
 565   ALWAYS_ASSERT (smp != 0);
 566   return segment_manager_get (*smp);
 567 }
 568
 569 session_t *
 570 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
 571                            u8 transport_proto)
 572 {
 573   session_t *listener;
 574   u64 handle;
 575   u32 sm_index;
 576   u8 sst;
 577
 578   sst = session_type_from_proto_and_ip (transport_proto,
 579                                         fib_proto == FIB_PROTOCOL_IP4);
 580
 581   /* *INDENT-OFF* */
 582    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 583      listener = listen_session_get_from_handle (handle);
 584      if (listener->session_type == sst
 585          && !(listener->flags & SESSION_F_PROXY))
 586        return listener;
 587    }));
 588   /* *INDENT-ON* */
 589
 590   return 0;
 591 }
 592
 593 session_t *
 594 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
 595                            u8 transport_proto)
 596 {
 597   session_t *listener;
 598   u64 handle;
 599   u32 sm_index;
 600   u8 sst;
 601
 602   sst = session_type_from_proto_and_ip (transport_proto,
 603                                         fib_proto == FIB_PROTOCOL_IP4);
 604
 605   /* *INDENT-OFF* */
 606    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 607      listener = listen_session_get_from_handle (handle);
 608      if (listener->session_type == sst && (listener->flags & SESSION_F_PROXY))
 609        return listener;
 610    }));
 611   /* *INDENT-ON* */
 612
 613   return 0;
 614 }
 615
 616 /**
 617  * Send an API message to the external app, to map new segment
 618  */
 619 int
 620 app_worker_add_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 621 {
 622   application_t *app = application_get (app_wrk->app_index);
 623
 624   return app->cb_fns.add_segment_callback (app_wrk->wrk_index,
 625                                            segment_handle);
 626 }
 627
 628 int
 629 app_worker_del_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 630 {
 631   application_t *app = application_get (app_wrk->app_index);
 632   return app->cb_fns.del_segment_callback (app_wrk->wrk_index,
 633                                            segment_handle);
 634 }
 635
 636 static inline u8
 637 app_worker_application_is_builtin (app_worker_t * app_wrk)
 638 {
 639   return app_wrk->app_is_builtin;
 640 }
 641
 642 static inline int
 643 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s)
 644 {
 645   session_event_t *evt;
 646   svm_msg_q_msg_t msg;
 647   svm_msg_q_t *mq;
 648
 649   if (app_worker_application_is_builtin (app_wrk))
 650     return app_worker_builtin_rx (app_wrk, s);
 651
 652   if (svm_fifo_has_event (s->rx_fifo))
 653     return 0;
 654
 655   mq = app_wrk->event_queue;
 656   svm_msg_q_lock (mq);
 657
 658   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
 659     {
 660       clib_warning ("evt q full");
 661       svm_msg_q_unlock (mq);
 662       return -1;
 663     }
 664
 665   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 666     {
 667       clib_warning ("evt q rings full");
 668       svm_msg_q_unlock (mq);
 669       return -1;
 670     }
 671
 672   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 673   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 674   evt->session_index = s->rx_fifo->shr->client_session_index;
 675   evt->event_type = SESSION_IO_EVT_RX;
 676
 677   (void) svm_fifo_set_event (s->rx_fifo);
 678   svm_msg_q_add_and_unlock (mq, &msg);
 679
 680   return 0;
 681 }
 682
 683 static inline int
 684 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s)
 685 {
 686   svm_msg_q_t *mq;
 687   session_event_t *evt;
 688   svm_msg_q_msg_t msg;
 689
 690   if (app_worker_application_is_builtin (app_wrk))
 691     return app_worker_builtin_tx (app_wrk, s);
 692
 693   mq = app_wrk->event_queue;
 694   svm_msg_q_lock (mq);
 695
 696   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
 697     {
 698       clib_warning ("evt q full");
 699       svm_msg_q_unlock (mq);
 700       return -1;
 701     }
 702
 703   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 704     {
 705       clib_warning ("evt q rings full");
 706       svm_msg_q_unlock (mq);
 707       return -1;
 708     }
 709
 710   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 711   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 712   evt->event_type = SESSION_IO_EVT_TX;
 713   evt->session_index = s->tx_fifo->shr->client_session_index;
 714
 715   svm_msg_q_add_and_unlock (mq, &msg);
 716   return 0;
 717 }
 718
 719 /* *INDENT-OFF* */
 720 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
 721                                        session_t *s);
 722 static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
 723     app_send_io_evt_rx,
 724     app_send_io_evt_tx,
 725 };
 726 /* *INDENT-ON* */
 727
 728 /**
 729  * Send event to application
 730  *
 731  * Logic from queue perspective is blocking. However, if queue is full,
 732  * we return.
 733  */
 734 int
 735 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
 736                                 u8 evt_type)
 737 {
 738   return app_send_evt_handler_fns[evt_type] (app, s);
 739 }
 740
 741 u8 *
 742 format_app_worker_listener (u8 * s, va_list * args)
 743 {
 744   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
 745   u64 handle = va_arg (*args, u64);
 746   u32 sm_index = va_arg (*args, u32);
 747   int verbose = va_arg (*args, int);
 748   session_t *listener;
 749   const u8 *app_name;
 750   u8 *str;
 751
 752   if (!app_wrk)
 753     {
 754       if (verbose)
 755         s = format (s, "%-40s%-25s%-10s%-15s%-15s%-10s", "Connection", "App",
 756                     "Wrk", "API Client", "ListenerID", "SegManager");
 757       else
 758         s = format (s, "%-40s%-25s%-10s", "Connection", "App", "Wrk");
 759
 760       return s;
 761     }
 762
 763   app_name = application_name_from_index (app_wrk->app_index);
 764   listener = listen_session_get_from_handle (handle);
 765   str = format (0, "%U", format_session, listener, verbose);
 766
 767   if (verbose)
 768     {
 769       u8 *buf;
 770       buf = format (0, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
 771       s = format (s, "%-40v%-25v%-10v%-15u%-15u%-10u", str, app_name,
 772                   buf, app_wrk->api_client_index, handle, sm_index);
 773       vec_free (buf);
 774     }
 775   else
 776     s = format (s, "%-40v%-25v%=10u", str, app_name, app_wrk->wrk_map_index);
 777
 778   vec_free (str);
 779
 780   return s;
 781 }
 782
 783 u8 *
 784 format_app_worker (u8 * s, va_list * args)
 785 {
 786   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
 787   u32 indent = 1;
 788
 789   s = format (s, "%U wrk-index %u app-index %u map-index %u "
 790               "api-client-index %d\n", format_white_space, indent,
 791               app_wrk->wrk_index, app_wrk->app_index, app_wrk->wrk_map_index,
 792               app_wrk->api_client_index);
 793   return s;
 794 }
 795
 796 void
 797 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
 798 {
 799   segment_manager_t *sm;
 800
 801   /* Header */
 802   if (!app_wrk)
 803     {
 804       segment_manager_format_sessions (0, verbose);
 805       return;
 806     }
 807
 808   if (app_wrk->connects_seg_manager == (u32) ~ 0)
 809     return;
 810
 811   sm = segment_manager_get (app_wrk->connects_seg_manager);
 812   segment_manager_format_sessions (sm, verbose);
 813 }
 814
 815 /*
 816  * fd.io coding-style-patch-verification: ON
 817  *
 818  * Local Variables:
 819  * eval: (c-set-style "gnu")
 820  * End:
 821  */