src/vnet/session/application_worker.c

   1 /*
   2  * Copyright (c) 2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <vnet/session/application.h>
  17 #include <vnet/session/application_interface.h>
  18 #include <vnet/session/session.h>
  19
  20 /**
  21  * Pool of workers associated to apps
  22  */
  23 static app_worker_t *app_workers;
  24
  25 app_worker_t *
  26 app_worker_alloc (application_t * app)
  27 {
  28   app_worker_t *app_wrk;
  29   pool_get (app_workers, app_wrk);
  30   clib_memset (app_wrk, 0, sizeof (*app_wrk));
  31   app_wrk->wrk_index = app_wrk - app_workers;
  32   app_wrk->app_index = app->app_index;
  33   app_wrk->wrk_map_index = ~0;
  34   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
  35   clib_spinlock_init (&app_wrk->detached_seg_managers_lock);
  36   clib_spinlock_init (&app_wrk->postponed_mq_msgs_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_handle_t *sh;
  63   session_t *ls;
  64   u32 sm_index;
  65   int i;
  66
  67   /*
  68    *  Listener cleanup
  69    */
  70
  71   /* *INDENT-OFF* */
  72   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
  73     ls = listen_session_get_from_handle (handle);
  74     vec_add1 (handles, app_listen_session_handle (ls));
  75     vec_add1 (sm_indices, sm_index);
  76     sm = segment_manager_get (sm_index);
  77   }));
  78   /* *INDENT-ON* */
  79
  80   for (i = 0; i < vec_len (handles); i++)
  81     {
  82       /* Cleanup listener */
  83       a->app_index = app->app_index;
  84       a->wrk_map_index = app_wrk->wrk_map_index;
  85       a->handle = handles[i];
  86       (void) vnet_unlisten (a);
  87
  88       sm = segment_manager_get_if_valid (sm_indices[i]);
  89       if (sm && !segment_manager_app_detached (sm))
  90         {
  91           sm->first_is_protected = 0;
  92           segment_manager_init_free (sm);
  93         }
  94     }
  95   vec_reset_length (handles);
  96   vec_free (sm_indices);
  97   hash_free (app_wrk->listeners_table);
  98
  99   /*
 100    * Connects segment manager cleanup
 101    */
 102
 103   if (app_wrk->connects_seg_manager != APP_INVALID_SEGMENT_MANAGER_INDEX)
 104     {
 105       sm = segment_manager_get (app_wrk->connects_seg_manager);
 106       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 107       sm->first_is_protected = 0;
 108       segment_manager_init_free (sm);
 109     }
 110
 111   /*
 112    * Half-open cleanup
 113    */
 114
 115   pool_foreach (sh, app_wrk->half_open_table)
 116     session_cleanup_half_open (*sh);
 117
 118   pool_free (app_wrk->half_open_table);
 119
 120   /*
 121    * Detached listener segment managers cleanup
 122    */
 123   for (i = 0; i < vec_len (app_wrk->detached_seg_managers); i++)
 124     {
 125       sm = segment_manager_get (app_wrk->detached_seg_managers[i]);
 126       segment_manager_init_free (sm);
 127     }
 128   vec_free (app_wrk->detached_seg_managers);
 129   clib_spinlock_free (&app_wrk->detached_seg_managers_lock);
 130   clib_spinlock_free (&app_wrk->postponed_mq_msgs_lock);
 131
 132   if (CLIB_DEBUG)
 133     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
 134   pool_put (app_workers, app_wrk);
 135 }
 136
 137 application_t *
 138 app_worker_get_app (u32 wrk_index)
 139 {
 140   app_worker_t *app_wrk;
 141   app_wrk = app_worker_get_if_valid (wrk_index);
 142   if (!app_wrk)
 143     return 0;
 144   return application_get_if_valid (app_wrk->app_index);
 145 }
 146
 147 static segment_manager_t *
 148 app_worker_alloc_segment_manager (app_worker_t * app_wrk)
 149 {
 150   segment_manager_t *sm;
 151
 152   sm = segment_manager_alloc ();
 153   sm->app_wrk_index = app_wrk->wrk_index;
 154   segment_manager_init (sm);
 155   return sm;
 156 }
 157
 158 static int
 159 app_worker_alloc_session_fifos (segment_manager_t * sm, session_t * s)
 160 {
 161   svm_fifo_t *rx_fifo = 0, *tx_fifo = 0;
 162   int rv;
 163
 164   if ((rv = segment_manager_alloc_session_fifos (sm, s->thread_index,
 165                                                  &rx_fifo, &tx_fifo)))
 166     return rv;
 167
 168   rx_fifo->shr->master_session_index = s->session_index;
 169   rx_fifo->master_thread_index = s->thread_index;
 170
 171   tx_fifo->shr->master_session_index = s->session_index;
 172   tx_fifo->master_thread_index = s->thread_index;
 173
 174   s->rx_fifo = rx_fifo;
 175   s->tx_fifo = tx_fifo;
 176   return 0;
 177 }
 178
 179 int
 180 app_worker_init_listener (app_worker_t * app_wrk, session_t * ls)
 181 {
 182   segment_manager_t *sm;
 183
 184   /* Allocate segment manager. All sessions derived out of a listen session
 185    * have fifos allocated by the same segment manager. */
 186   if (!(sm = app_worker_alloc_segment_manager (app_wrk)))
 187     return SESSION_E_ALLOC;
 188
 189   /* Once the first segment is mapped, don't remove it until unlisten */
 190   sm->first_is_protected = 1;
 191
 192   /* Keep track of the segment manager for the listener or this worker */
 193   hash_set (app_wrk->listeners_table, listen_session_get_handle (ls),
 194             segment_manager_index (sm));
 195
 196   if (transport_connection_is_cless (session_get_transport (ls)))
 197     {
 198       if (ls->rx_fifo)
 199         return SESSION_E_NOSUPPORT;
 200       return app_worker_alloc_session_fifos (sm, ls);
 201     }
 202   return 0;
 203 }
 204
 205 int
 206 app_worker_start_listen (app_worker_t * app_wrk,
 207                          app_listener_t * app_listener)
 208 {
 209   session_t *ls;
 210   int rv;
 211
 212   if (clib_bitmap_get (app_listener->workers, app_wrk->wrk_map_index))
 213     return SESSION_E_ALREADY_LISTENING;
 214
 215   app_listener->workers = clib_bitmap_set (app_listener->workers,
 216                                            app_wrk->wrk_map_index, 1);
 217
 218   if (app_listener->session_index != SESSION_INVALID_INDEX)
 219     {
 220       ls = session_get (app_listener->session_index, 0);
 221       if ((rv = app_worker_init_listener (app_wrk, ls)))
 222         return rv;
 223     }
 224
 225   if (app_listener->local_index != SESSION_INVALID_INDEX)
 226     {
 227       ls = session_get (app_listener->local_index, 0);
 228       if ((rv = app_worker_init_listener (app_wrk, ls)))
 229         return rv;
 230     }
 231
 232   return 0;
 233 }
 234
 235 static void
 236 app_worker_add_detached_sm (app_worker_t * app_wrk, u32 sm_index)
 237 {
 238   vec_add1 (app_wrk->detached_seg_managers, sm_index);
 239 }
 240
 241 void
 242 app_worker_del_detached_sm (app_worker_t * app_wrk, u32 sm_index)
 243 {
 244   u32 i;
 245
 246   clib_spinlock_lock (&app_wrk->detached_seg_managers_lock);
 247   for (i = 0; i < vec_len (app_wrk->detached_seg_managers); i++)
 248     {
 249       if (app_wrk->detached_seg_managers[i] == sm_index)
 250         {
 251           vec_del1 (app_wrk->detached_seg_managers, i);
 252           break;
 253         }
 254     }
 255   clib_spinlock_unlock (&app_wrk->detached_seg_managers_lock);
 256 }
 257
 258 static void
 259 app_worker_stop_listen_session (app_worker_t * app_wrk, session_t * ls)
 260 {
 261   session_handle_t handle;
 262   segment_manager_t *sm;
 263   uword *sm_indexp;
 264   session_state_t *states = 0;
 265
 266   handle = listen_session_get_handle (ls);
 267   sm_indexp = hash_get (app_wrk->listeners_table, handle);
 268   if (PREDICT_FALSE (!sm_indexp))
 269     return;
 270
 271   /* Dealloc fifos, if any (dgram listeners) */
 272   if (ls->rx_fifo)
 273     {
 274       segment_manager_dealloc_fifos (ls->rx_fifo, ls->tx_fifo);
 275       ls->tx_fifo = ls->rx_fifo = 0;
 276     }
 277
 278   /* Try to cleanup segment manager */
 279   sm = segment_manager_get (*sm_indexp);
 280   if (sm)
 281     {
 282       sm->first_is_protected = 0;
 283       segment_manager_app_detach (sm);
 284       if (!segment_manager_has_fifos (sm))
 285         {
 286           /* Empty segment manager, cleanup it up */
 287           segment_manager_free (sm);
 288         }
 289       else
 290         {
 291           /* Delete sessions in CREATED state */
 292           vec_add1 (states, SESSION_STATE_CREATED);
 293           segment_manager_del_sessions_filter (sm, states);
 294           vec_free (states);
 295
 296           /* Track segment manager in case app detaches and all the
 297            * outstanding sessions need to be closed */
 298           app_worker_add_detached_sm (app_wrk, *sm_indexp);
 299           sm->flags |= SEG_MANAGER_F_DETACHED_LISTENER;
 300         }
 301     }
 302
 303   hash_unset (app_wrk->listeners_table, handle);
 304 }
 305
 306 int
 307 app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al)
 308 {
 309   session_t *ls;
 310
 311   if (!clib_bitmap_get (al->workers, app_wrk->wrk_map_index))
 312     return 0;
 313
 314   if (al->session_index != SESSION_INVALID_INDEX)
 315     {
 316       ls = listen_session_get (al->session_index);
 317       app_worker_stop_listen_session (app_wrk, ls);
 318     }
 319
 320   if (al->local_index != SESSION_INVALID_INDEX)
 321     {
 322       ls = listen_session_get (al->local_index);
 323       app_worker_stop_listen_session (app_wrk, ls);
 324     }
 325
 326   clib_bitmap_set_no_check (al->workers, app_wrk->wrk_map_index, 0);
 327   if (clib_bitmap_is_zero (al->workers))
 328     app_listener_cleanup (al);
 329
 330   return 0;
 331 }
 332
 333 int
 334 app_worker_init_accepted (session_t * s)
 335 {
 336   app_worker_t *app_wrk;
 337   segment_manager_t *sm;
 338   session_t *listener;
 339   application_t *app;
 340
 341   listener = listen_session_get_from_handle (s->listener_handle);
 342   app_wrk = application_listener_select_worker (listener);
 343   if (PREDICT_FALSE (app_wrk->mq_congested))
 344     return -1;
 345
 346   s->app_wrk_index = app_wrk->wrk_index;
 347   app = application_get (app_wrk->app_index);
 348   if (app->cb_fns.fifo_tuning_callback)
 349     s->flags |= SESSION_F_CUSTOM_FIFO_TUNING;
 350
 351   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
 352   if (app_worker_alloc_session_fifos (sm, s))
 353     return -1;
 354
 355   return 0;
 356 }
 357
 358 int
 359 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
 360 {
 361   application_t *app = application_get (app_wrk->app_index);
 362   return app->cb_fns.session_accept_callback (s);
 363 }
 364
 365 int
 366 app_worker_init_connected (app_worker_t * app_wrk, session_t * s)
 367 {
 368   application_t *app = application_get (app_wrk->app_index);
 369   segment_manager_t *sm;
 370
 371   if (app->cb_fns.fifo_tuning_callback)
 372     s->flags |= SESSION_F_CUSTOM_FIFO_TUNING;
 373
 374   /* Allocate fifos for session, unless the app is a builtin proxy */
 375   if (application_is_builtin_proxy (app))
 376     return 0;
 377
 378   sm = app_worker_get_connect_segment_manager (app_wrk);
 379   return app_worker_alloc_session_fifos (sm, s);
 380 }
 381
 382 int
 383 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s,
 384                            session_error_t err, u32 opaque)
 385 {
 386   application_t *app = application_get (app_wrk->app_index);
 387   return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
 388                                                  s, err);
 389 }
 390
 391 int
 392 app_worker_add_half_open (app_worker_t *app_wrk, session_handle_t sh)
 393 {
 394   session_handle_t *shp;
 395
 396   ASSERT (session_vlib_thread_is_cl_thread ());
 397   pool_get (app_wrk->half_open_table, shp);
 398   *shp = sh;
 399
 400   return (shp - app_wrk->half_open_table);
 401 }
 402
 403 int
 404 app_worker_del_half_open (app_worker_t *app_wrk, session_t *s)
 405 {
 406   application_t *app = application_get (app_wrk->app_index);
 407   ASSERT (session_vlib_thread_is_cl_thread ());
 408   pool_put_index (app_wrk->half_open_table, s->ho_index);
 409   if (app->cb_fns.half_open_cleanup_callback)
 410     app->cb_fns.half_open_cleanup_callback (s);
 411   return 0;
 412 }
 413
 414 int
 415 app_worker_close_notify (app_worker_t * app_wrk, session_t * s)
 416 {
 417   application_t *app = application_get (app_wrk->app_index);
 418   app->cb_fns.session_disconnect_callback (s);
 419   return 0;
 420 }
 421
 422 int
 423 app_worker_transport_closed_notify (app_worker_t * app_wrk, session_t * s)
 424 {
 425   application_t *app = application_get (app_wrk->app_index);
 426   if (app->cb_fns.session_transport_closed_callback)
 427     app->cb_fns.session_transport_closed_callback (s);
 428   return 0;
 429 }
 430
 431 int
 432 app_worker_reset_notify (app_worker_t * app_wrk, session_t * s)
 433 {
 434   application_t *app = application_get (app_wrk->app_index);
 435   app->cb_fns.session_reset_callback (s);
 436   return 0;
 437 }
 438
 439 int
 440 app_worker_cleanup_notify (app_worker_t * app_wrk, session_t * s,
 441                            session_cleanup_ntf_t ntf)
 442 {
 443   application_t *app = application_get (app_wrk->app_index);
 444   if (app->cb_fns.session_cleanup_callback)
 445     app->cb_fns.session_cleanup_callback (s, ntf);
 446   return 0;
 447 }
 448
 449 int
 450 app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s)
 451 {
 452   application_t *app = application_get (app_wrk->app_index);
 453   app->cb_fns.builtin_app_rx_callback (s);
 454   return 0;
 455 }
 456
 457 int
 458 app_worker_builtin_tx (app_worker_t * app_wrk, session_t * s)
 459 {
 460   application_t *app = application_get (app_wrk->app_index);
 461
 462   if (!app->cb_fns.builtin_app_tx_callback)
 463     return 0;
 464
 465   app->cb_fns.builtin_app_tx_callback (s);
 466   return 0;
 467 }
 468
 469 int
 470 app_worker_migrate_notify (app_worker_t * app_wrk, session_t * s,
 471                            session_handle_t new_sh)
 472 {
 473   application_t *app = application_get (app_wrk->app_index);
 474   app->cb_fns.session_migrate_callback (s, new_sh);
 475   return 0;
 476 }
 477
 478 int
 479 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
 480 {
 481   segment_manager_t *sm;
 482   svm_fifo_t *rxf, *txf;
 483   int rv;
 484
 485   if (s->session_state == SESSION_STATE_LISTENING)
 486     return application_change_listener_owner (s, app_wrk);
 487
 488   s->app_wrk_index = app_wrk->wrk_index;
 489
 490   rxf = s->rx_fifo;
 491   txf = s->tx_fifo;
 492
 493   if (!rxf || !txf)
 494     return 0;
 495
 496   s->rx_fifo = 0;
 497   s->tx_fifo = 0;
 498
 499   sm = app_worker_get_connect_segment_manager (app_wrk);
 500   if ((rv = app_worker_alloc_session_fifos (sm, s)))
 501     return rv;
 502
 503   if (!svm_fifo_is_empty_cons (rxf))
 504     svm_fifo_clone (s->rx_fifo, rxf);
 505
 506   if (!svm_fifo_is_empty_cons (txf))
 507     svm_fifo_clone (s->tx_fifo, txf);
 508
 509   segment_manager_dealloc_fifos (rxf, txf);
 510
 511   return 0;
 512 }
 513
 514 int
 515 app_worker_connect_session (app_worker_t *app_wrk, session_endpoint_cfg_t *sep,
 516                             session_handle_t *rsh)
 517 {
 518   if (PREDICT_FALSE (app_wrk->mq_congested))
 519     return SESSION_E_REFUSED;
 520
 521   sep->app_wrk_index = app_wrk->wrk_index;
 522
 523   return session_open (sep, rsh);
 524 }
 525
 526 int
 527 app_worker_session_fifo_tuning (app_worker_t * app_wrk, session_t * s,
 528                                 svm_fifo_t * f,
 529                                 session_ft_action_t act, u32 len)
 530 {
 531   application_t *app = application_get (app_wrk->app_index);
 532   return app->cb_fns.fifo_tuning_callback (s, f, act, len);
 533 }
 534
 535 segment_manager_t *
 536 app_worker_get_connect_segment_manager (app_worker_t * app)
 537 {
 538   ASSERT (app->connects_seg_manager != (u32) ~ 0);
 539   return segment_manager_get (app->connects_seg_manager);
 540 }
 541
 542 segment_manager_t *
 543 app_worker_get_listen_segment_manager (app_worker_t * app,
 544                                        session_t * listener)
 545 {
 546   uword *smp;
 547   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
 548   ALWAYS_ASSERT (smp != 0);
 549   return segment_manager_get (*smp);
 550 }
 551
 552 session_t *
 553 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
 554                            u8 transport_proto)
 555 {
 556   session_t *listener;
 557   u64 handle;
 558   u32 sm_index;
 559   u8 sst;
 560
 561   sst = session_type_from_proto_and_ip (transport_proto,
 562                                         fib_proto == FIB_PROTOCOL_IP4);
 563
 564   /* *INDENT-OFF* */
 565    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 566      listener = listen_session_get_from_handle (handle);
 567      if (listener->session_type == sst
 568          && !(listener->flags & SESSION_F_PROXY))
 569        return listener;
 570    }));
 571   /* *INDENT-ON* */
 572
 573   return 0;
 574 }
 575
 576 session_t *
 577 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
 578                            u8 transport_proto)
 579 {
 580   session_t *listener;
 581   u64 handle;
 582   u32 sm_index;
 583   u8 sst;
 584
 585   sst = session_type_from_proto_and_ip (transport_proto,
 586                                         fib_proto == FIB_PROTOCOL_IP4);
 587
 588   /* *INDENT-OFF* */
 589    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 590      listener = listen_session_get_from_handle (handle);
 591      if (listener->session_type == sst && (listener->flags & SESSION_F_PROXY))
 592        return listener;
 593    }));
 594   /* *INDENT-ON* */
 595
 596   return 0;
 597 }
 598
 599 /**
 600  * Send an API message to the external app, to map new segment
 601  */
 602 int
 603 app_worker_add_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 604 {
 605   application_t *app = application_get (app_wrk->app_index);
 606
 607   return app->cb_fns.add_segment_callback (app_wrk->wrk_index,
 608                                            segment_handle);
 609 }
 610
 611 int
 612 app_worker_del_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 613 {
 614   application_t *app = application_get (app_wrk->app_index);
 615   return app->cb_fns.del_segment_callback (app_wrk->wrk_index,
 616                                            segment_handle);
 617 }
 618
 619 static inline u8
 620 app_worker_application_is_builtin (app_worker_t * app_wrk)
 621 {
 622   return app_wrk->app_is_builtin;
 623 }
 624
 625 static int
 626 app_wrk_send_fd (app_worker_t *app_wrk, int fd)
 627 {
 628   if (!appns_sapi_enabled ())
 629     {
 630       vl_api_registration_t *reg;
 631       clib_error_t *error;
 632
 633       reg =
 634         vl_mem_api_client_index_to_registration (app_wrk->api_client_index);
 635       if (!reg)
 636         {
 637           clib_warning ("no api registration for client: %u",
 638                         app_wrk->api_client_index);
 639           return -1;
 640         }
 641
 642       if (vl_api_registration_file_index (reg) == VL_API_INVALID_FI)
 643         return -1;
 644
 645       error = vl_api_send_fd_msg (reg, &fd, 1);
 646       if (error)
 647         {
 648           clib_error_report (error);
 649           return -1;
 650         }
 651
 652       return 0;
 653     }
 654
 655   app_sapi_msg_t smsg = { 0 };
 656   app_namespace_t *app_ns;
 657   clib_error_t *error;
 658   application_t *app;
 659   clib_socket_t *cs;
 660   u32 cs_index;
 661
 662   app = application_get (app_wrk->app_index);
 663   app_ns = app_namespace_get (app->ns_index);
 664   cs_index = appns_sapi_handle_sock_index (app_wrk->api_client_index);
 665   cs = appns_sapi_get_socket (app_ns, cs_index);
 666   if (PREDICT_FALSE (!cs))
 667     return -1;
 668
 669   /* There's no payload for the message only the type */
 670   smsg.type = APP_SAPI_MSG_TYPE_SEND_FDS;
 671   error = clib_socket_sendmsg (cs, &smsg, sizeof (smsg), &fd, 1);
 672   if (error)
 673     {
 674       clib_error_report (error);
 675       return -1;
 676     }
 677
 678   return 0;
 679 }
 680
 681 static int
 682 mq_try_lock_and_alloc_msg (svm_msg_q_t *mq, session_mq_rings_e ring,
 683                            svm_msg_q_msg_t *msg)
 684 {
 685   int rv, n_try = 0;
 686
 687   while (n_try < 75)
 688     {
 689       rv = svm_msg_q_lock_and_alloc_msg_w_ring (mq, ring, SVM_Q_NOWAIT, msg);
 690       if (!rv)
 691         return 0;
 692       /*
 693        * Break the loop if mq is full, usually this is because the
 694        * app has crashed or is hanging on somewhere.
 695        */
 696       if (rv != -1)
 697         break;
 698       n_try += 1;
 699       usleep (1);
 700     }
 701
 702   return -1;
 703 }
 704
 705 typedef union app_wrk_mq_rpc_args_
 706 {
 707   struct
 708   {
 709     u32 thread_index;
 710     u32 app_wrk_index;
 711   };
 712   uword as_uword;
 713 } app_wrk_mq_rpc_ags_t;
 714
 715 static int
 716 app_wrk_handle_mq_postponed_msgs (void *arg)
 717 {
 718   svm_msg_q_msg_t _mq_msg, *mq_msg = &_mq_msg;
 719   app_wrk_postponed_msg_t *pm;
 720   app_wrk_mq_rpc_ags_t args;
 721   u32 max_msg, n_msg = 0;
 722   app_worker_t *app_wrk;
 723   session_event_t *evt;
 724   svm_msg_q_t *mq;
 725
 726   args.as_uword = pointer_to_uword (arg);
 727   app_wrk = app_worker_get_if_valid (args.app_wrk_index);
 728   if (!app_wrk)
 729     return 0;
 730
 731   mq = app_wrk->event_queue;
 732
 733   clib_spinlock_lock (&app_wrk->postponed_mq_msgs_lock);
 734
 735   max_msg = clib_min (32, clib_fifo_elts (app_wrk->postponed_mq_msgs));
 736
 737   while (n_msg < max_msg)
 738     {
 739       pm = clib_fifo_head (app_wrk->postponed_mq_msgs);
 740       if (mq_try_lock_and_alloc_msg (mq, pm->ring, mq_msg))
 741         break;
 742
 743       evt = svm_msg_q_msg_data (mq, mq_msg);
 744       clib_memset (evt, 0, sizeof (*evt));
 745       evt->event_type = pm->event_type;
 746       clib_memcpy_fast (evt->data, pm->data, pm->len);
 747
 748       if (pm->fd != -1)
 749         app_wrk_send_fd (app_wrk, pm->fd);
 750
 751       svm_msg_q_add_and_unlock (mq, mq_msg);
 752
 753       clib_fifo_advance_head (app_wrk->postponed_mq_msgs, 1);
 754       n_msg += 1;
 755     }
 756
 757   if (!clib_fifo_elts (app_wrk->postponed_mq_msgs))
 758     {
 759       app_wrk->mq_congested = 0;
 760     }
 761   else
 762     {
 763       session_send_rpc_evt_to_thread_force (
 764         args.thread_index, app_wrk_handle_mq_postponed_msgs,
 765         uword_to_pointer (args.as_uword, void *));
 766     }
 767
 768   clib_spinlock_unlock (&app_wrk->postponed_mq_msgs_lock);
 769
 770   return 0;
 771 }
 772
 773 static void
 774 app_wrk_add_mq_postponed_msg (app_worker_t *app_wrk, session_mq_rings_e ring,
 775                               u8 evt_type, void *msg, u32 msg_len, int fd)
 776 {
 777   app_wrk_postponed_msg_t *pm;
 778
 779   clib_spinlock_lock (&app_wrk->postponed_mq_msgs_lock);
 780
 781   app_wrk->mq_congested = 1;
 782
 783   clib_fifo_add2 (app_wrk->postponed_mq_msgs, pm);
 784   clib_memcpy_fast (pm->data, msg, msg_len);
 785   pm->event_type = evt_type;
 786   pm->ring = ring;
 787   pm->len = msg_len;
 788   pm->fd = fd;
 789
 790   if (clib_fifo_elts (app_wrk->postponed_mq_msgs) == 1)
 791     {
 792       app_wrk_mq_rpc_ags_t args = { .thread_index = vlib_get_thread_index (),
 793                                     .app_wrk_index = app_wrk->wrk_index };
 794
 795       session_send_rpc_evt_to_thread_force (
 796         args.thread_index, app_wrk_handle_mq_postponed_msgs,
 797         uword_to_pointer (args.as_uword, void *));
 798     }
 799
 800   clib_spinlock_unlock (&app_wrk->postponed_mq_msgs_lock);
 801 }
 802
 803 always_inline void
 804 app_wrk_send_ctrl_evt_inline (app_worker_t *app_wrk, u8 evt_type, void *msg,
 805                               u32 msg_len, int fd)
 806 {
 807   svm_msg_q_msg_t _mq_msg, *mq_msg = &_mq_msg;
 808   svm_msg_q_t *mq = app_wrk->event_queue;
 809   session_event_t *evt;
 810   int rv;
 811
 812   if (PREDICT_FALSE (app_wrk->mq_congested))
 813     goto handle_congestion;
 814
 815   rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_CTRL_EVT_RING, mq_msg);
 816   if (PREDICT_FALSE (rv))
 817     goto handle_congestion;
 818
 819   evt = svm_msg_q_msg_data (mq, mq_msg);
 820   clib_memset (evt, 0, sizeof (*evt));
 821   evt->event_type = evt_type;
 822   clib_memcpy_fast (evt->data, msg, msg_len);
 823
 824   if (fd != -1)
 825     app_wrk_send_fd (app_wrk, fd);
 826
 827   svm_msg_q_add_and_unlock (mq, mq_msg);
 828
 829   return;
 830
 831 handle_congestion:
 832
 833   app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_CTRL_EVT_RING, evt_type,
 834                                 msg, msg_len, fd);
 835 }
 836
 837 void
 838 app_wrk_send_ctrl_evt_fd (app_worker_t *app_wrk, u8 evt_type, void *msg,
 839                           u32 msg_len, int fd)
 840 {
 841   app_wrk_send_ctrl_evt_inline (app_wrk, evt_type, msg, msg_len, fd);
 842 }
 843
 844 void
 845 app_wrk_send_ctrl_evt (app_worker_t *app_wrk, u8 evt_type, void *msg,
 846                        u32 msg_len)
 847 {
 848   app_wrk_send_ctrl_evt_inline (app_wrk, evt_type, msg, msg_len, -1);
 849 }
 850
 851 static inline int
 852 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s)
 853 {
 854   svm_msg_q_msg_t _mq_msg = { 0 }, *mq_msg = &_mq_msg;
 855   session_event_t *evt;
 856   svm_msg_q_t *mq;
 857   u32 app_session;
 858   int rv;
 859
 860   if (app_worker_application_is_builtin (app_wrk))
 861     return app_worker_builtin_rx (app_wrk, s);
 862
 863   if (svm_fifo_has_event (s->rx_fifo))
 864     return 0;
 865
 866   app_session = s->rx_fifo->shr->client_session_index;
 867   mq = app_wrk->event_queue;
 868
 869   if (PREDICT_FALSE (app_wrk->mq_congested))
 870     goto handle_congestion;
 871
 872   rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_IO_EVT_RING, mq_msg);
 873
 874   if (PREDICT_FALSE (rv))
 875     goto handle_congestion;
 876
 877   evt = svm_msg_q_msg_data (mq, mq_msg);
 878   evt->event_type = SESSION_IO_EVT_RX;
 879   evt->session_index = app_session;
 880
 881   (void) svm_fifo_set_event (s->rx_fifo);
 882
 883   svm_msg_q_add_and_unlock (mq, mq_msg);
 884
 885   return 0;
 886
 887 handle_congestion:
 888
 889   app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_IO_EVT_RING,
 890                                 SESSION_IO_EVT_RX, &app_session,
 891                                 sizeof (app_session), -1);
 892   return -1;
 893 }
 894
 895 static inline int
 896 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s)
 897 {
 898   svm_msg_q_msg_t _mq_msg = { 0 }, *mq_msg = &_mq_msg;
 899   session_event_t *evt;
 900   svm_msg_q_t *mq;
 901   u32 app_session;
 902   int rv;
 903
 904   if (app_worker_application_is_builtin (app_wrk))
 905     return app_worker_builtin_tx (app_wrk, s);
 906
 907   app_session = s->tx_fifo->shr->client_session_index;
 908   mq = app_wrk->event_queue;
 909
 910   if (PREDICT_FALSE (app_wrk->mq_congested))
 911     goto handle_congestion;
 912
 913   rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_IO_EVT_RING, mq_msg);
 914
 915   if (PREDICT_FALSE (rv))
 916     goto handle_congestion;
 917
 918   evt = svm_msg_q_msg_data (mq, mq_msg);
 919   evt->event_type = SESSION_IO_EVT_TX;
 920   evt->session_index = app_session;
 921
 922   svm_msg_q_add_and_unlock (mq, mq_msg);
 923
 924   return 0;
 925
 926 handle_congestion:
 927
 928   app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_IO_EVT_RING,
 929                                 SESSION_IO_EVT_TX, &app_session,
 930                                 sizeof (app_session), -1);
 931   return -1;
 932 }
 933
 934 /* *INDENT-OFF* */
 935 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
 936                                        session_t *s);
 937 static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
 938     app_send_io_evt_rx,
 939     app_send_io_evt_tx,
 940 };
 941 /* *INDENT-ON* */
 942
 943 /**
 944  * Send event to application
 945  *
 946  * Logic from queue perspective is blocking. However, if queue is full,
 947  * we return.
 948  */
 949 int
 950 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
 951                                 u8 evt_type)
 952 {
 953   return app_send_evt_handler_fns[evt_type] (app, s);
 954 }
 955
 956 u8 *
 957 format_app_worker_listener (u8 * s, va_list * args)
 958 {
 959   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
 960   u64 handle = va_arg (*args, u64);
 961   u32 sm_index = va_arg (*args, u32);
 962   int verbose = va_arg (*args, int);
 963   session_t *listener;
 964   const u8 *app_name;
 965   u8 *str;
 966
 967   if (!app_wrk)
 968     {
 969       if (verbose)
 970         s = format (s, "%-" SESSION_CLI_ID_LEN "s%-25s%-10s%-15s%-15s%-10s",
 971                     "Connection", "App", "Wrk", "API Client", "ListenerID",
 972                     "SegManager");
 973       else
 974         s = format (s, "%-" SESSION_CLI_ID_LEN "s%-25s%-10s", "Connection",
 975                     "App", "Wrk");
 976
 977       return s;
 978     }
 979
 980   app_name = application_name_from_index (app_wrk->app_index);
 981   listener = listen_session_get_from_handle (handle);
 982   str = format (0, "%U", format_session, listener, verbose);
 983
 984   if (verbose)
 985     {
 986       u8 *buf;
 987       buf = format (0, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
 988       s = format (s, "%-" SESSION_CLI_ID_LEN "v%-25v%-10v%-15u%-15u%-10u", str,
 989                   app_name, buf, app_wrk->api_client_index, handle, sm_index);
 990       vec_free (buf);
 991     }
 992   else
 993     s = format (s, "%-" SESSION_CLI_ID_LEN "v%-25v%=10u", str, app_name,
 994                 app_wrk->wrk_map_index);
 995
 996   vec_free (str);
 997
 998   return s;
 999 }
1000
1001 u8 *
1002 format_app_worker (u8 * s, va_list * args)
1003 {
1004   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
1005   u32 indent = 1;
1006
1007   s = format (s,
1008               "%U wrk-index %u app-index %u map-index %u "
1009               "api-client-index %d mq-cong %u\n",
1010               format_white_space, indent, app_wrk->wrk_index,
1011               app_wrk->app_index, app_wrk->wrk_map_index,
1012               app_wrk->api_client_index, app_wrk->mq_congested);
1013   return s;
1014 }
1015
1016 void
1017 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
1018 {
1019   segment_manager_t *sm;
1020
1021   /* Header */
1022   if (!app_wrk)
1023     {
1024       segment_manager_format_sessions (0, verbose);
1025       return;
1026     }
1027
1028   if (app_wrk->connects_seg_manager == (u32) ~ 0)
1029     return;
1030
1031   sm = segment_manager_get (app_wrk->connects_seg_manager);
1032   segment_manager_format_sessions (sm, verbose);
1033 }
1034
1035 /*
1036  * fd.io coding-style-patch-verification: ON
1037  *
1038  * Local Variables:
1039  * eval: (c-set-style "gnu")
1040  * End:
1041  */