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/application_local.h>
  19 #include <vnet/session/session.h>
  20
  21 /**
  22  * Pool of workers associated to apps
  23  */
  24 static app_worker_t *app_workers;
  25
  26 app_worker_t *
  27 app_worker_alloc (application_t * app)
  28 {
  29   app_worker_t *app_wrk;
  30   pool_get (app_workers, app_wrk);
  31   clib_memset (app_wrk, 0, sizeof (*app_wrk));
  32   app_wrk->wrk_index = app_wrk - app_workers;
  33   app_wrk->app_index = app->app_index;
  34   app_wrk->wrk_map_index = ~0;
  35   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
  36   app_wrk->first_segment_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
  37   app_wrk->local_segment_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
  38   APP_DBG ("New app %v worker %u", app_get_name (app), app_wrk->wrk_index);
  39   return app_wrk;
  40 }
  41
  42 app_worker_t *
  43 app_worker_get (u32 wrk_index)
  44 {
  45   return pool_elt_at_index (app_workers, wrk_index);
  46 }
  47
  48 app_worker_t *
  49 app_worker_get_if_valid (u32 wrk_index)
  50 {
  51   if (pool_is_free_index (app_workers, wrk_index))
  52     return 0;
  53   return pool_elt_at_index (app_workers, wrk_index);
  54 }
  55
  56 void
  57 app_worker_free (app_worker_t * app_wrk)
  58 {
  59   application_t *app = application_get (app_wrk->app_index);
  60   vnet_unlisten_args_t _a, *a = &_a;
  61   u64 handle, *handles = 0;
  62   segment_manager_t *sm;
  63   u32 sm_index;
  64   int i;
  65   app_listener_t *al;
  66   session_t *ls;
  67
  68   /*
  69    *  Listener cleanup
  70    */
  71
  72   /* *INDENT-OFF* */
  73   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
  74     ls = listen_session_get_from_handle (handle);
  75     al = app_listener_get (app, ls->al_index);
  76     vec_add1 (handles, app_listener_handle (al));
  77     sm = segment_manager_get (sm_index);
  78     sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
  79   }));
  80   /* *INDENT-ON* */
  81
  82   for (i = 0; i < vec_len (handles); i++)
  83     {
  84       a->app_index = app->app_index;
  85       a->wrk_map_index = app_wrk->wrk_map_index;
  86       a->handle = handles[i];
  87       /* seg manager is removed when unbind completes */
  88       (void) vnet_unlisten (a);
  89     }
  90
  91   /*
  92    * Connects segment manager cleanup
  93    */
  94
  95   if (app_wrk->connects_seg_manager != APP_INVALID_SEGMENT_MANAGER_INDEX)
  96     {
  97       sm = segment_manager_get (app_wrk->connects_seg_manager);
  98       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
  99       sm->first_is_protected = 0;
 100       segment_manager_init_del (sm);
 101     }
 102
 103   /* If first segment manager is used by a listener */
 104   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
 105       && app_wrk->first_segment_manager != app_wrk->connects_seg_manager)
 106     {
 107       sm = segment_manager_get (app_wrk->first_segment_manager);
 108       sm->first_is_protected = 0;
 109       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 110       /* .. and has no fifos, e.g. it might be used for redirected sessions,
 111        * remove it */
 112       if (!segment_manager_has_fifos (sm))
 113         segment_manager_del (sm);
 114     }
 115
 116   /*
 117    * Local sessions
 118    */
 119   app_worker_local_sessions_free (app_wrk);
 120
 121   pool_put (app_workers, app_wrk);
 122   if (CLIB_DEBUG)
 123     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
 124 }
 125
 126 application_t *
 127 app_worker_get_app (u32 wrk_index)
 128 {
 129   app_worker_t *app_wrk;
 130   app_wrk = app_worker_get_if_valid (wrk_index);
 131   if (!app_wrk)
 132     return 0;
 133   return application_get_if_valid (app_wrk->app_index);
 134 }
 135
 136 static segment_manager_t *
 137 app_worker_alloc_segment_manager (app_worker_t * app_wrk)
 138 {
 139   segment_manager_t *sm = 0;
 140
 141   /* If the first segment manager is not in use, don't allocate a new one */
 142   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
 143       && app_wrk->first_segment_manager_in_use == 0)
 144     {
 145       sm = segment_manager_get (app_wrk->first_segment_manager);
 146       app_wrk->first_segment_manager_in_use = 1;
 147       return sm;
 148     }
 149
 150   sm = segment_manager_new ();
 151   sm->app_wrk_index = app_wrk->wrk_index;
 152
 153   return sm;
 154 }
 155
 156 static int
 157 app_worker_alloc_session_fifos (segment_manager_t * sm, session_t * s)
 158 {
 159   svm_fifo_t *rx_fifo = 0, *tx_fifo = 0;
 160   u32 fifo_segment_index;
 161   int rv;
 162
 163   if ((rv = segment_manager_alloc_session_fifos (sm, &rx_fifo, &tx_fifo,
 164                                                  &fifo_segment_index)))
 165     return rv;
 166
 167   rx_fifo->master_session_index = s->session_index;
 168   rx_fifo->master_thread_index = s->thread_index;
 169
 170   tx_fifo->master_session_index = s->session_index;
 171   tx_fifo->master_thread_index = s->thread_index;
 172
 173   s->rx_fifo = rx_fifo;
 174   s->tx_fifo = tx_fifo;
 175   s->svm_segment_index = fifo_segment_index;
 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 -1;
 188
 189   /* Keep track of the segment manager for the listener or this worker */
 190   hash_set (app_wrk->listeners_table, listen_session_get_handle (ls),
 191             segment_manager_index (sm));
 192
 193   if (session_transport_service_type (ls) == TRANSPORT_SERVICE_CL)
 194     {
 195       if (!ls->rx_fifo && app_worker_alloc_session_fifos (sm, ls))
 196         return -1;
 197     }
 198   return 0;
 199 }
 200
 201 int
 202 app_worker_start_listen (app_worker_t * app_wrk,
 203                          app_listener_t * app_listener)
 204 {
 205   session_t *ls;
 206
 207   if (clib_bitmap_get (app_listener->workers, app_wrk->wrk_map_index))
 208     return VNET_API_ERROR_ADDRESS_IN_USE;
 209
 210   app_listener->workers = clib_bitmap_set (app_listener->workers,
 211                                            app_wrk->wrk_map_index, 1);
 212
 213   if (app_listener->session_index != SESSION_INVALID_INDEX)
 214     {
 215       ls = session_get (app_listener->session_index, 0);
 216       if (app_worker_init_listener (app_wrk, ls))
 217         return -1;
 218     }
 219
 220   if (app_listener->local_index != SESSION_INVALID_INDEX)
 221     {
 222       ls = session_get (app_listener->local_index, 0);
 223       if (app_worker_init_listener (app_wrk, ls))
 224         return -1;
 225     }
 226
 227   return 0;
 228 }
 229
 230 int
 231 app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al)
 232 {
 233   session_handle_t handle;
 234   segment_manager_t *sm;
 235   uword *sm_indexp;
 236   session_t *ls;
 237
 238   if (!clib_bitmap_get (al->workers, app_wrk->wrk_map_index))
 239     return 0;
 240
 241   if (al->session_index != SESSION_INVALID_INDEX)
 242     {
 243       ls = listen_session_get (al->session_index);
 244       handle = listen_session_get_handle (ls);
 245
 246       sm_indexp = hash_get (app_wrk->listeners_table, handle);
 247       if (PREDICT_FALSE (!sm_indexp))
 248         {
 249           clib_warning ("listener handle was removed %llu!", handle);
 250           return -1;
 251         }
 252
 253       sm = segment_manager_get (*sm_indexp);
 254       if (app_wrk->first_segment_manager == *sm_indexp)
 255         {
 256           /* Delete sessions but don't remove segment manager */
 257           app_wrk->first_segment_manager_in_use = 0;
 258           segment_manager_del_sessions (sm);
 259         }
 260       else
 261         {
 262           segment_manager_init_del (sm);
 263         }
 264       hash_unset (app_wrk->listeners_table, handle);
 265     }
 266
 267   if (al->local_index != SESSION_INVALID_INDEX)
 268     {
 269       local_session_t *local;
 270       ls = listen_session_get (al->local_index);
 271       handle = listen_session_get_handle (ls);
 272
 273       /* *INDENT-OFF* */
 274       pool_foreach (local, app_wrk->local_sessions, ({
 275         if (local->listener_index == ls->session_index)
 276           app_worker_local_session_disconnect (app_wrk->wrk_index, local);
 277       }));
 278       /* *INDENT-ON* */
 279
 280       sm_indexp = hash_get (app_wrk->listeners_table, handle);
 281       if (PREDICT_FALSE (!sm_indexp))
 282         return -1;
 283
 284       sm = segment_manager_get (*sm_indexp);
 285       if (app_wrk->first_segment_manager == *sm_indexp)
 286         {
 287           /* Delete sessions but don't remove segment manager */
 288           app_wrk->first_segment_manager_in_use = 0;
 289           segment_manager_del_sessions (sm);
 290         }
 291       else
 292         {
 293           segment_manager_init_del (sm);
 294         }
 295       hash_unset (app_wrk->listeners_table, handle);
 296     }
 297
 298   clib_bitmap_set_no_check (al->workers, app_wrk->wrk_map_index, 0);
 299   if (clib_bitmap_is_zero (al->workers))
 300     app_listener_cleanup (al);
 301
 302   return 0;
 303 }
 304
 305 int
 306 app_worker_init_accepted (session_t * s)
 307 {
 308   app_worker_t *app_wrk;
 309   segment_manager_t *sm;
 310   session_t *listener;
 311
 312   listener = listen_session_get (s->listener_index);
 313   app_wrk = application_listener_select_worker (listener);
 314   s->app_wrk_index = app_wrk->wrk_index;
 315   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
 316   return app_worker_alloc_session_fifos (sm, s);
 317 }
 318
 319 int
 320 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
 321 {
 322   application_t *app = application_get (app_wrk->app_index);
 323   return app->cb_fns.session_accept_callback (s);
 324 }
 325
 326 int
 327 app_worker_init_connected (app_worker_t * app_wrk, session_t * s)
 328 {
 329   application_t *app = application_get (app_wrk->app_index);
 330   segment_manager_t *sm;
 331
 332   /* Allocate fifos for session, unless the app is a builtin proxy */
 333   if (!application_is_builtin_proxy (app))
 334     {
 335       sm = app_worker_get_connect_segment_manager (app_wrk);
 336       if (app_worker_alloc_session_fifos (sm, s))
 337         return -1;
 338     }
 339   return 0;
 340 }
 341
 342 int
 343 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s, u32 opaque)
 344 {
 345   application_t *app = application_get (app_wrk->app_index);
 346   return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
 347                                                  s, s == 0 /* is_fail */ );
 348 }
 349
 350 int
 351 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
 352 {
 353   segment_manager_t *sm;
 354   svm_fifo_t *rxf, *txf;
 355
 356   if (s->session_state == SESSION_STATE_LISTENING)
 357     return application_change_listener_owner (s, app_wrk);
 358
 359   s->app_wrk_index = app_wrk->wrk_index;
 360
 361   rxf = s->rx_fifo;
 362   txf = s->tx_fifo;
 363
 364   if (!rxf || !txf)
 365     return 0;
 366
 367   s->rx_fifo = 0;
 368   s->tx_fifo = 0;
 369
 370   sm = app_worker_get_or_alloc_connect_segment_manager (app_wrk);
 371   if (app_worker_alloc_session_fifos (sm, s))
 372     return -1;
 373
 374   if (!svm_fifo_is_empty (rxf))
 375     {
 376       clib_memcpy_fast (s->rx_fifo->data, rxf->data, rxf->nitems);
 377       s->rx_fifo->head = rxf->head;
 378       s->rx_fifo->tail = rxf->tail;
 379       s->rx_fifo->cursize = rxf->cursize;
 380     }
 381
 382   if (!svm_fifo_is_empty (txf))
 383     {
 384       clib_memcpy_fast (s->tx_fifo->data, txf->data, txf->nitems);
 385       s->tx_fifo->head = txf->head;
 386       s->tx_fifo->tail = txf->tail;
 387       s->tx_fifo->cursize = txf->cursize;
 388     }
 389
 390   segment_manager_dealloc_fifos (rxf->segment_index, rxf, txf);
 391
 392   return 0;
 393 }
 394
 395 int
 396 app_worker_connect_session (app_worker_t * app, session_endpoint_t * sep,
 397                             u32 api_context)
 398 {
 399   int rv;
 400
 401   /* Make sure we have a segment manager for connects */
 402   app_worker_alloc_connects_segment_manager (app);
 403
 404   if ((rv = session_open (app->wrk_index, sep, api_context)))
 405     return rv;
 406
 407   return 0;
 408 }
 409
 410 int
 411 app_worker_alloc_connects_segment_manager (app_worker_t * app_wrk)
 412 {
 413   segment_manager_t *sm;
 414
 415   if (app_wrk->connects_seg_manager == APP_INVALID_SEGMENT_MANAGER_INDEX)
 416     {
 417       sm = app_worker_alloc_segment_manager (app_wrk);
 418       if (sm == 0)
 419         return -1;
 420       app_wrk->connects_seg_manager = segment_manager_index (sm);
 421     }
 422   return 0;
 423 }
 424
 425 segment_manager_t *
 426 app_worker_get_connect_segment_manager (app_worker_t * app)
 427 {
 428   ASSERT (app->connects_seg_manager != (u32) ~ 0);
 429   return segment_manager_get (app->connects_seg_manager);
 430 }
 431
 432 segment_manager_t *
 433 app_worker_get_or_alloc_connect_segment_manager (app_worker_t * app_wrk)
 434 {
 435   if (app_wrk->connects_seg_manager == (u32) ~ 0)
 436     app_worker_alloc_connects_segment_manager (app_wrk);
 437   return segment_manager_get (app_wrk->connects_seg_manager);
 438 }
 439
 440 segment_manager_t *
 441 app_worker_get_listen_segment_manager (app_worker_t * app,
 442                                        session_t * listener)
 443 {
 444   uword *smp;
 445   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
 446   ASSERT (smp != 0);
 447   return segment_manager_get (*smp);
 448 }
 449
 450 session_t *
 451 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
 452                            u8 transport_proto)
 453 {
 454   session_t *listener;
 455   u64 handle;
 456   u32 sm_index;
 457   u8 sst;
 458
 459   sst = session_type_from_proto_and_ip (transport_proto,
 460                                         fib_proto == FIB_PROTOCOL_IP4);
 461
 462   /* *INDENT-OFF* */
 463    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 464      listener = listen_session_get_from_handle (handle);
 465      if (listener->session_type == sst
 466          && listener->enqueue_epoch != SESSION_PROXY_LISTENER_INDEX)
 467        return listener;
 468    }));
 469   /* *INDENT-ON* */
 470
 471   return 0;
 472 }
 473
 474 session_t *
 475 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
 476                            u8 transport_proto)
 477 {
 478   session_t *listener;
 479   u64 handle;
 480   u32 sm_index;
 481   u8 sst;
 482
 483   sst = session_type_from_proto_and_ip (transport_proto,
 484                                         fib_proto == FIB_PROTOCOL_IP4);
 485
 486   /* *INDENT-OFF* */
 487    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 488      listener = listen_session_get_from_handle (handle);
 489      if (listener->session_type == sst
 490          && listener->enqueue_epoch == SESSION_PROXY_LISTENER_INDEX)
 491        return listener;
 492    }));
 493   /* *INDENT-ON* */
 494
 495   return 0;
 496 }
 497
 498 /**
 499  * Send an API message to the external app, to map new segment
 500  */
 501 int
 502 app_worker_add_segment_notify (u32 app_wrk_index, u64 segment_handle)
 503 {
 504   app_worker_t *app_wrk = app_worker_get (app_wrk_index);
 505   application_t *app = application_get (app_wrk->app_index);
 506   return app->cb_fns.add_segment_callback (app_wrk->api_client_index,
 507                                            segment_handle);
 508 }
 509
 510 u8
 511 app_worker_application_is_builtin (app_worker_t * app_wrk)
 512 {
 513   return app_wrk->app_is_builtin;
 514 }
 515
 516 static inline int
 517 app_enqueue_evt (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, u8 lock)
 518 {
 519   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
 520     {
 521       clib_warning ("evt q full");
 522       svm_msg_q_free_msg (mq, msg);
 523       if (lock)
 524         svm_msg_q_unlock (mq);
 525       return -1;
 526     }
 527
 528   if (lock)
 529     {
 530       svm_msg_q_add_and_unlock (mq, msg);
 531       return 0;
 532     }
 533
 534   /* Even when not locking the ring, we must wait for queue mutex */
 535   if (svm_msg_q_add (mq, msg, SVM_Q_WAIT))
 536     {
 537       clib_warning ("msg q add returned");
 538       return -1;
 539     }
 540   return 0;
 541 }
 542
 543 static inline int
 544 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s, u8 lock)
 545 {
 546   session_event_t *evt;
 547   svm_msg_q_msg_t msg;
 548   svm_msg_q_t *mq;
 549
 550   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY
 551                      && s->session_state != SESSION_STATE_LISTENING))
 552     {
 553       /* Session is closed so app will never clean up. Flush rx fifo */
 554       if (s->session_state == SESSION_STATE_CLOSED)
 555         svm_fifo_dequeue_drop_all (s->rx_fifo);
 556       return 0;
 557     }
 558
 559   if (app_worker_application_is_builtin (app_wrk))
 560     {
 561       application_t *app = application_get (app_wrk->app_index);
 562       return app->cb_fns.builtin_app_rx_callback (s);
 563     }
 564
 565   if (svm_fifo_has_event (s->rx_fifo) || svm_fifo_is_empty (s->rx_fifo))
 566     return 0;
 567
 568   mq = app_wrk->event_queue;
 569   if (lock)
 570     svm_msg_q_lock (mq);
 571
 572   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 573     {
 574       clib_warning ("evt q rings full");
 575       if (lock)
 576         svm_msg_q_unlock (mq);
 577       return -1;
 578     }
 579
 580   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 581   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
 582
 583   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 584   evt->fifo = s->rx_fifo;
 585   evt->event_type = FIFO_EVENT_APP_RX;
 586
 587   (void) svm_fifo_set_event (s->rx_fifo);
 588
 589   if (app_enqueue_evt (mq, &msg, lock))
 590     return -1;
 591   return 0;
 592 }
 593
 594 static inline int
 595 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s, u8 lock)
 596 {
 597   svm_msg_q_t *mq;
 598   session_event_t *evt;
 599   svm_msg_q_msg_t msg;
 600
 601   if (app_worker_application_is_builtin (app_wrk))
 602     return 0;
 603
 604   mq = app_wrk->event_queue;
 605   if (lock)
 606     svm_msg_q_lock (mq);
 607
 608   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 609     {
 610       clib_warning ("evt q rings full");
 611       if (lock)
 612         svm_msg_q_unlock (mq);
 613       return -1;
 614     }
 615
 616   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 617   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
 618
 619   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 620   evt->event_type = FIFO_EVENT_APP_TX;
 621   evt->fifo = s->tx_fifo;
 622
 623   return app_enqueue_evt (mq, &msg, lock);
 624 }
 625
 626 /* *INDENT-OFF* */
 627 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
 628                                        session_t *s,
 629                                        u8 lock);
 630 static app_send_evt_handler_fn * const app_send_evt_handler_fns[3] = {
 631     app_send_io_evt_rx,
 632     0,
 633     app_send_io_evt_tx,
 634 };
 635 /* *INDENT-ON* */
 636
 637 /**
 638  * Send event to application
 639  *
 640  * Logic from queue perspective is non-blocking. If there's
 641  * not enough space to enqueue a message, we return.
 642  */
 643 int
 644 app_worker_send_event (app_worker_t * app, session_t * s, u8 evt_type)
 645 {
 646   ASSERT (app && evt_type <= FIFO_EVENT_APP_TX);
 647   return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
 648 }
 649
 650 /**
 651  * Send event to application
 652  *
 653  * Logic from queue perspective is blocking. However, if queue is full,
 654  * we return.
 655  */
 656 int
 657 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
 658                                 u8 evt_type)
 659 {
 660   return app_send_evt_handler_fns[evt_type] (app, s, 1 /* lock */ );
 661 }
 662
 663 segment_manager_t *
 664 app_worker_get_local_segment_manager (app_worker_t * app_worker)
 665 {
 666   return segment_manager_get (app_worker->local_segment_manager);
 667 }
 668
 669 segment_manager_t *
 670 app_worker_get_local_segment_manager_w_session (app_worker_t * app_wrk,
 671                                                 local_session_t * ls)
 672 {
 673   session_t *listener;
 674   if (application_local_session_listener_has_transport (ls))
 675     {
 676       listener = listen_session_get (ls->listener_index);
 677       return app_worker_get_listen_segment_manager (app_wrk, listener);
 678     }
 679   return segment_manager_get (app_wrk->local_segment_manager);
 680 }
 681
 682 u8 *
 683 format_app_worker_listener (u8 * s, va_list * args)
 684 {
 685   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
 686   u64 handle = va_arg (*args, u64);
 687   u32 sm_index = va_arg (*args, u32);
 688   int verbose = va_arg (*args, int);
 689   session_t *listener;
 690   const u8 *app_name;
 691   u8 *str;
 692
 693   if (!app_wrk)
 694     {
 695       if (verbose)
 696         s = format (s, "%-40s%-25s%=10s%-15s%-15s%-10s", "Connection", "App",
 697                     "Wrk", "API Client", "ListenerID", "SegManager");
 698       else
 699         s = format (s, "%-40s%-25s%=10s", "Connection", "App", "Wrk");
 700
 701       return s;
 702     }
 703
 704   app_name = application_name_from_index (app_wrk->app_index);
 705   listener = listen_session_get_from_handle (handle);
 706   str = format (0, "%U", format_stream_session, listener, verbose);
 707
 708   if (verbose)
 709     {
 710       char buf[32];
 711       sprintf (buf, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
 712       s = format (s, "%-40s%-25s%=10s%-15u%-15u%-10u", str, app_name,
 713                   buf, app_wrk->api_client_index, handle, sm_index);
 714     }
 715   else
 716     s = format (s, "%-40s%-25s%=10u", str, app_name, app_wrk->wrk_map_index);
 717
 718   return s;
 719 }
 720
 721 u8 *
 722 format_app_worker (u8 * s, va_list * args)
 723 {
 724   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
 725   u32 indent = 1;
 726
 727   s = format (s, "%U wrk-index %u app-index %u map-index %u "
 728               "api-client-index %d\n", format_white_space, indent,
 729               app_wrk->wrk_index, app_wrk->app_index, app_wrk->wrk_map_index,
 730               app_wrk->api_client_index);
 731   return s;
 732 }
 733
 734 void
 735 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
 736 {
 737   svm_fifo_segment_private_t *fifo_segment;
 738   vlib_main_t *vm = vlib_get_main ();
 739   segment_manager_t *sm;
 740   const u8 *app_name;
 741   u8 *s = 0;
 742
 743   /* Header */
 744   if (!app_wrk)
 745     {
 746       if (verbose)
 747         vlib_cli_output (vm, "%-40s%-20s%-15s%-10s", "Connection", "App",
 748                          "API Client", "SegManager");
 749       else
 750         vlib_cli_output (vm, "%-40s%-20s", "Connection", "App");
 751       return;
 752     }
 753
 754   if (app_wrk->connects_seg_manager == (u32) ~ 0)
 755     return;
 756
 757   app_name = application_name_from_index (app_wrk->app_index);
 758
 759   /* Across all fifo segments */
 760   sm = segment_manager_get (app_wrk->connects_seg_manager);
 761
 762   /* *INDENT-OFF* */
 763   segment_manager_foreach_segment_w_lock (fifo_segment, sm, ({
 764     svm_fifo_t *fifo;
 765     u8 *str;
 766
 767     fifo = svm_fifo_segment_get_fifo_list (fifo_segment);
 768     while (fifo)
 769       {
 770         u32 session_index, thread_index;
 771         session_t *session;
 772
 773         session_index = fifo->master_session_index;
 774         thread_index = fifo->master_thread_index;
 775
 776         session = session_get (session_index, thread_index);
 777         str = format (0, "%U", format_stream_session, session, verbose);
 778
 779         if (verbose)
 780           s = format (s, "%-40s%-20s%-15u%-10u", str, app_name,
 781                       app_wrk->api_client_index, app_wrk->connects_seg_manager);
 782         else
 783           s = format (s, "%-40s%-20s", str, app_name);
 784
 785         vlib_cli_output (vm, "%v", s);
 786         vec_reset_length (s);
 787         vec_free (str);
 788
 789         fifo = fifo->next;
 790       }
 791     vec_free (s);
 792   }));
 793   /* *INDENT-ON* */
 794 }
 795
 796 /*
 797  * fd.io coding-style-patch-verification: ON
 798  *
 799  * Local Variables:
 800  * eval: (c-set-style "gnu")
 801  * End:
 802  */