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