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