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