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