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