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 static inline u64
  26 application_client_local_connect_key (local_session_t * ls)
  27 {
  28   return (((u64) ls->app_wrk_index) << 32 | (u64) ls->session_index);
  29 }
  30
  31 static inline void
  32 application_client_local_connect_key_parse (u64 key, u32 * app_wrk_index,
  33                                             u32 * session_index)
  34 {
  35   *app_wrk_index = key >> 32;
  36   *session_index = key & 0xFFFFFFFF;
  37 }
  38
  39 local_session_t *
  40 app_worker_local_session_alloc (app_worker_t * app_wrk)
  41 {
  42   local_session_t *s;
  43   pool_get (app_wrk->local_sessions, s);
  44   clib_memset (s, 0, sizeof (*s));
  45   s->app_wrk_index = app_wrk->wrk_index;
  46   s->session_index = s - app_wrk->local_sessions;
  47   s->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE, 0);
  48   return s;
  49 }
  50
  51 void
  52 app_worker_local_session_free (app_worker_t * app_wrk, local_session_t * s)
  53 {
  54   pool_put (app_wrk->local_sessions, s);
  55   if (CLIB_DEBUG)
  56     clib_memset (s, 0xfc, sizeof (*s));
  57 }
  58
  59 local_session_t *
  60 app_worker_get_local_session (app_worker_t * app_wrk, u32 session_index)
  61 {
  62   if (pool_is_free_index (app_wrk->local_sessions, session_index))
  63     return 0;
  64   return pool_elt_at_index (app_wrk->local_sessions, session_index);
  65 }
  66
  67 local_session_t *
  68 app_worker_get_local_session_from_handle (session_handle_t handle)
  69 {
  70   app_worker_t *server_wrk;
  71   u32 session_index, server_wrk_index;
  72   local_session_parse_handle (handle, &server_wrk_index, &session_index);
  73   server_wrk = app_worker_get_if_valid (server_wrk_index);
  74   if (!server_wrk)
  75     return 0;
  76   return app_worker_get_local_session (server_wrk, session_index);
  77 }
  78
  79 void
  80 app_worker_local_sessions_free (app_worker_t * app_wrk)
  81 {
  82   u32 index, server_wrk_index, session_index;
  83   u64 handle, *handles = 0;
  84   app_worker_t *server_wrk;
  85   segment_manager_t *sm;
  86   local_session_t *ls;
  87   int i;
  88
  89   /*
  90    * Local sessions
  91    */
  92   if (app_wrk->local_sessions)
  93     {
  94       /* *INDENT-OFF* */
  95       pool_foreach (ls, app_wrk->local_sessions, ({
  96         app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
  97       }));
  98       /* *INDENT-ON* */
  99     }
 100
 101   /*
 102    * Local connects
 103    */
 104   vec_reset_length (handles);
 105   /* *INDENT-OFF* */
 106   hash_foreach (handle, index, app_wrk->local_connects, ({
 107     vec_add1 (handles, handle);
 108   }));
 109   /* *INDENT-ON* */
 110
 111   for (i = 0; i < vec_len (handles); i++)
 112     {
 113       application_client_local_connect_key_parse (handles[i],
 114                                                   &server_wrk_index,
 115                                                   &session_index);
 116       server_wrk = app_worker_get_if_valid (server_wrk_index);
 117       if (server_wrk)
 118         {
 119           ls = app_worker_get_local_session (server_wrk, session_index);
 120           app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
 121         }
 122     }
 123
 124   sm = segment_manager_get (app_wrk->local_segment_manager);
 125   sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 126   segment_manager_del (sm);
 127 }
 128
 129 app_worker_t *
 130 app_worker_alloc (application_t * app)
 131 {
 132   app_worker_t *app_wrk;
 133   pool_get (app_workers, app_wrk);
 134   clib_memset (app_wrk, 0, sizeof (*app_wrk));
 135   app_wrk->wrk_index = app_wrk - app_workers;
 136   app_wrk->app_index = app->app_index;
 137   app_wrk->wrk_map_index = ~0;
 138   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
 139   app_wrk->first_segment_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
 140   app_wrk->local_segment_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
 141   APP_DBG ("New app %v worker %u", app_get_name (app), app_wrk->wrk_index);
 142   return app_wrk;
 143 }
 144
 145 app_worker_t *
 146 app_worker_get (u32 wrk_index)
 147 {
 148   return pool_elt_at_index (app_workers, wrk_index);
 149 }
 150
 151 app_worker_t *
 152 app_worker_get_if_valid (u32 wrk_index)
 153 {
 154   if (pool_is_free_index (app_workers, wrk_index))
 155     return 0;
 156   return pool_elt_at_index (app_workers, wrk_index);
 157 }
 158
 159 void
 160 app_worker_free (app_worker_t * app_wrk)
 161 {
 162   application_t *app = application_get (app_wrk->app_index);
 163   vnet_unlisten_args_t _a, *a = &_a;
 164   u64 handle, *handles = 0;
 165   segment_manager_t *sm;
 166   u32 sm_index;
 167   int i;
 168   app_listener_t *al;
 169   session_t *ls;
 170
 171   /*
 172    *  Listener cleanup
 173    */
 174
 175   /* *INDENT-OFF* */
 176   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 177     ls = listen_session_get_from_handle (handle);
 178     al = app_listener_get (app, ls->al_index);
 179     vec_add1 (handles, app_listener_handle (al));
 180     sm = segment_manager_get (sm_index);
 181     sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 182   }));
 183   /* *INDENT-ON* */
 184
 185   for (i = 0; i < vec_len (handles); i++)
 186     {
 187       a->app_index = app->app_index;
 188       a->wrk_map_index = app_wrk->wrk_map_index;
 189       a->handle = handles[i];
 190       /* seg manager is removed when unbind completes */
 191       (void) vnet_unlisten (a);
 192     }
 193
 194   /*
 195    * Connects segment manager cleanup
 196    */
 197
 198   if (app_wrk->connects_seg_manager != APP_INVALID_SEGMENT_MANAGER_INDEX)
 199     {
 200       sm = segment_manager_get (app_wrk->connects_seg_manager);
 201       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 202       sm->first_is_protected = 0;
 203       segment_manager_init_del (sm);
 204     }
 205
 206   /* If first segment manager is used by a listener */
 207   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
 208       && app_wrk->first_segment_manager != app_wrk->connects_seg_manager)
 209     {
 210       sm = segment_manager_get (app_wrk->first_segment_manager);
 211       sm->first_is_protected = 0;
 212       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 213       /* .. and has no fifos, e.g. it might be used for redirected sessions,
 214        * remove it */
 215       if (!segment_manager_has_fifos (sm))
 216         segment_manager_del (sm);
 217     }
 218
 219   /*
 220    * Local sessions
 221    */
 222   app_worker_local_sessions_free (app_wrk);
 223
 224   pool_put (app_workers, app_wrk);
 225   if (CLIB_DEBUG)
 226     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
 227 }
 228
 229 application_t *
 230 app_worker_get_app (u32 wrk_index)
 231 {
 232   app_worker_t *app_wrk;
 233   app_wrk = app_worker_get_if_valid (wrk_index);
 234   if (!app_wrk)
 235     return 0;
 236   return application_get_if_valid (app_wrk->app_index);
 237 }
 238
 239 static segment_manager_t *
 240 app_worker_alloc_segment_manager (app_worker_t * app_wrk)
 241 {
 242   segment_manager_t *sm = 0;
 243
 244   /* If the first segment manager is not in use, don't allocate a new one */
 245   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
 246       && app_wrk->first_segment_manager_in_use == 0)
 247     {
 248       sm = segment_manager_get (app_wrk->first_segment_manager);
 249       app_wrk->first_segment_manager_in_use = 1;
 250       return sm;
 251     }
 252
 253   sm = segment_manager_new ();
 254   sm->app_wrk_index = app_wrk->wrk_index;
 255
 256   return sm;
 257 }
 258
 259 static int
 260 app_worker_alloc_session_fifos (segment_manager_t * sm, session_t * s)
 261 {
 262   svm_fifo_t *rx_fifo = 0, *tx_fifo = 0;
 263   u32 fifo_segment_index;
 264   int rv;
 265
 266   if ((rv = segment_manager_alloc_session_fifos (sm, &rx_fifo, &tx_fifo,
 267                                                  &fifo_segment_index)))
 268     return rv;
 269
 270   rx_fifo->master_session_index = s->session_index;
 271   rx_fifo->master_thread_index = s->thread_index;
 272
 273   tx_fifo->master_session_index = s->session_index;
 274   tx_fifo->master_thread_index = s->thread_index;
 275
 276   s->rx_fifo = rx_fifo;
 277   s->tx_fifo = tx_fifo;
 278   s->svm_segment_index = fifo_segment_index;
 279   return 0;
 280 }
 281
 282 int
 283 app_worker_start_listen (app_worker_t * app_wrk,
 284                          app_listener_t * app_listener)
 285 {
 286   segment_manager_t *sm;
 287   session_t *ls;
 288
 289   if (clib_bitmap_get (app_listener->workers, app_wrk->wrk_map_index))
 290     return VNET_API_ERROR_ADDRESS_IN_USE;
 291
 292   app_listener->workers = clib_bitmap_set (app_listener->workers,
 293                                            app_wrk->wrk_map_index, 1);
 294
 295   if (app_listener->session_index == SESSION_INVALID_INDEX)
 296     return 0;
 297
 298   ls = session_get (app_listener->session_index, 0);
 299
 300   /* Allocate segment manager. All sessions derived out of a listen session
 301    * have fifos allocated by the same segment manager. */
 302   if (!(sm = app_worker_alloc_segment_manager (app_wrk)))
 303     return -1;
 304
 305   /* Keep track of the segment manager for the listener or this worker */
 306   hash_set (app_wrk->listeners_table, listen_session_get_handle (ls),
 307             segment_manager_index (sm));
 308
 309   if (session_transport_service_type (ls) == TRANSPORT_SERVICE_CL)
 310     {
 311       if (!ls->rx_fifo && app_worker_alloc_session_fifos (sm, ls))
 312         return -1;
 313     }
 314   return 0;
 315 }
 316
 317 int
 318 app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al)
 319 {
 320   session_handle_t handle;
 321   segment_manager_t *sm;
 322   uword *sm_indexp;
 323
 324   if (!clib_bitmap_get (al->workers, app_wrk->wrk_map_index))
 325     return 0;
 326
 327   if (al->session_index != SESSION_INVALID_INDEX)
 328     {
 329       session_t *ls;
 330
 331       ls = listen_session_get (al->session_index);
 332       handle = listen_session_get_handle (ls);
 333
 334       sm_indexp = hash_get (app_wrk->listeners_table, handle);
 335       if (PREDICT_FALSE (!sm_indexp))
 336         {
 337           clib_warning ("listener handle was removed %llu!", handle);
 338           return -1;
 339         }
 340
 341       sm = segment_manager_get (*sm_indexp);
 342       if (app_wrk->first_segment_manager == *sm_indexp)
 343         {
 344           /* Delete sessions but don't remove segment manager */
 345           app_wrk->first_segment_manager_in_use = 0;
 346           segment_manager_del_sessions (sm);
 347         }
 348       else
 349         {
 350           segment_manager_init_del (sm);
 351         }
 352       hash_unset (app_wrk->listeners_table, handle);
 353     }
 354
 355   if (al->local_index != SESSION_INVALID_INDEX)
 356     {
 357       local_session_t *ll, *ls;
 358       application_t *app;
 359
 360       app = application_get (app_wrk->app_index);
 361       ll = application_get_local_listen_session (app, al->local_index);
 362
 363       /* *INDENT-OFF* */
 364       pool_foreach (ls, app_wrk->local_sessions, ({
 365         if (ls->listener_index == ll->session_index)
 366           app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
 367       }));
 368       /* *INDENT-ON* */
 369     }
 370
 371   clib_bitmap_set_no_check (al->workers, app_wrk->wrk_map_index, 0);
 372   if (clib_bitmap_is_zero (al->workers))
 373     app_listener_cleanup (al);
 374
 375   return 0;
 376 }
 377
 378 int
 379 app_worker_init_accepted (session_t * s)
 380 {
 381   app_worker_t *app_wrk;
 382   segment_manager_t *sm;
 383   session_t *listener;
 384
 385   listener = listen_session_get (s->listener_index);
 386   app_wrk = application_listener_select_worker (listener);
 387   s->app_wrk_index = app_wrk->wrk_index;
 388   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
 389   return app_worker_alloc_session_fifos (sm, s);
 390 }
 391
 392 int
 393 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
 394 {
 395   application_t *app = application_get (app_wrk->app_index);
 396   return app->cb_fns.session_accept_callback (s);
 397 }
 398
 399 int
 400 app_worker_init_connected (app_worker_t * app_wrk, session_t * s)
 401 {
 402   application_t *app = application_get (app_wrk->app_index);
 403   segment_manager_t *sm;
 404
 405   /* Allocate fifos for session, unless the app is a builtin proxy */
 406   if (!application_is_builtin_proxy (app))
 407     {
 408       sm = app_worker_get_connect_segment_manager (app_wrk);
 409       if (app_worker_alloc_session_fifos (sm, s))
 410         return -1;
 411     }
 412   return 0;
 413 }
 414
 415 int
 416 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s, u32 opaque)
 417 {
 418   application_t *app = application_get (app_wrk->app_index);
 419   return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
 420                                                  s, s == 0 /* is_fail */ );
 421 }
 422
 423 int
 424 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
 425 {
 426   segment_manager_t *sm;
 427   svm_fifo_t *rxf, *txf;
 428
 429   if (s->session_state == SESSION_STATE_LISTENING)
 430     return application_change_listener_owner (s, app_wrk);
 431
 432   s->app_wrk_index = app_wrk->wrk_index;
 433
 434   rxf = s->rx_fifo;
 435   txf = s->tx_fifo;
 436
 437   if (!rxf || !txf)
 438     return 0;
 439
 440   s->rx_fifo = 0;
 441   s->tx_fifo = 0;
 442
 443   sm = app_worker_get_or_alloc_connect_segment_manager (app_wrk);
 444   if (app_worker_alloc_session_fifos (sm, s))
 445     return -1;
 446
 447   if (!svm_fifo_is_empty (rxf))
 448     {
 449       clib_memcpy_fast (s->rx_fifo->data, rxf->data, rxf->nitems);
 450       s->rx_fifo->head = rxf->head;
 451       s->rx_fifo->tail = rxf->tail;
 452       s->rx_fifo->cursize = rxf->cursize;
 453     }
 454
 455   if (!svm_fifo_is_empty (txf))
 456     {
 457       clib_memcpy_fast (s->tx_fifo->data, txf->data, txf->nitems);
 458       s->tx_fifo->head = txf->head;
 459       s->tx_fifo->tail = txf->tail;
 460       s->tx_fifo->cursize = txf->cursize;
 461     }
 462
 463   segment_manager_dealloc_fifos (rxf->segment_index, rxf, txf);
 464
 465   return 0;
 466 }
 467
 468 int
 469 app_worker_connect_session (app_worker_t * app, session_endpoint_t * sep,
 470                             u32 api_context)
 471 {
 472   int rv;
 473
 474   /* Make sure we have a segment manager for connects */
 475   app_worker_alloc_connects_segment_manager (app);
 476
 477   if ((rv = session_open (app->wrk_index, sep, api_context)))
 478     return rv;
 479
 480   return 0;
 481 }
 482
 483 int
 484 app_worker_alloc_connects_segment_manager (app_worker_t * app_wrk)
 485 {
 486   segment_manager_t *sm;
 487
 488   if (app_wrk->connects_seg_manager == APP_INVALID_SEGMENT_MANAGER_INDEX)
 489     {
 490       sm = app_worker_alloc_segment_manager (app_wrk);
 491       if (sm == 0)
 492         return -1;
 493       app_wrk->connects_seg_manager = segment_manager_index (sm);
 494     }
 495   return 0;
 496 }
 497
 498 segment_manager_t *
 499 app_worker_get_connect_segment_manager (app_worker_t * app)
 500 {
 501   ASSERT (app->connects_seg_manager != (u32) ~ 0);
 502   return segment_manager_get (app->connects_seg_manager);
 503 }
 504
 505 segment_manager_t *
 506 app_worker_get_or_alloc_connect_segment_manager (app_worker_t * app_wrk)
 507 {
 508   if (app_wrk->connects_seg_manager == (u32) ~ 0)
 509     app_worker_alloc_connects_segment_manager (app_wrk);
 510   return segment_manager_get (app_wrk->connects_seg_manager);
 511 }
 512
 513 segment_manager_t *
 514 app_worker_get_listen_segment_manager (app_worker_t * app,
 515                                        session_t * listener)
 516 {
 517   uword *smp;
 518   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
 519   ASSERT (smp != 0);
 520   return segment_manager_get (*smp);
 521 }
 522
 523 session_t *
 524 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
 525                            u8 transport_proto)
 526 {
 527   session_t *listener;
 528   u64 handle;
 529   u32 sm_index;
 530   u8 sst;
 531
 532   sst = session_type_from_proto_and_ip (transport_proto,
 533                                         fib_proto == FIB_PROTOCOL_IP4);
 534
 535   /* *INDENT-OFF* */
 536    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 537      listener = listen_session_get_from_handle (handle);
 538      if (listener->session_type == sst
 539          && listener->enqueue_epoch != SESSION_PROXY_LISTENER_INDEX)
 540        return listener;
 541    }));
 542   /* *INDENT-ON* */
 543
 544   return 0;
 545 }
 546
 547 session_t *
 548 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
 549                            u8 transport_proto)
 550 {
 551   session_t *listener;
 552   u64 handle;
 553   u32 sm_index;
 554   u8 sst;
 555
 556   sst = session_type_from_proto_and_ip (transport_proto,
 557                                         fib_proto == FIB_PROTOCOL_IP4);
 558
 559   /* *INDENT-OFF* */
 560    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
 561      listener = listen_session_get_from_handle (handle);
 562      if (listener->session_type == sst
 563          && listener->enqueue_epoch == SESSION_PROXY_LISTENER_INDEX)
 564        return listener;
 565    }));
 566   /* *INDENT-ON* */
 567
 568   return 0;
 569 }
 570
 571 /**
 572  * Send an API message to the external app, to map new segment
 573  */
 574 int
 575 app_worker_add_segment_notify (u32 app_wrk_index, u64 segment_handle)
 576 {
 577   app_worker_t *app_wrk = app_worker_get (app_wrk_index);
 578   application_t *app = application_get (app_wrk->app_index);
 579   return app->cb_fns.add_segment_callback (app_wrk->api_client_index,
 580                                            segment_handle);
 581 }
 582
 583 u8
 584 app_worker_application_is_builtin (app_worker_t * app_wrk)
 585 {
 586   return app_wrk->app_is_builtin;
 587 }
 588
 589 static inline int
 590 app_enqueue_evt (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, u8 lock)
 591 {
 592   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
 593     {
 594       clib_warning ("evt q full");
 595       svm_msg_q_free_msg (mq, msg);
 596       if (lock)
 597         svm_msg_q_unlock (mq);
 598       return -1;
 599     }
 600
 601   if (lock)
 602     {
 603       svm_msg_q_add_and_unlock (mq, msg);
 604       return 0;
 605     }
 606
 607   /* Even when not locking the ring, we must wait for queue mutex */
 608   if (svm_msg_q_add (mq, msg, SVM_Q_WAIT))
 609     {
 610       clib_warning ("msg q add returned");
 611       return -1;
 612     }
 613   return 0;
 614 }
 615
 616 static inline int
 617 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s, u8 lock)
 618 {
 619   session_event_t *evt;
 620   svm_msg_q_msg_t msg;
 621   svm_msg_q_t *mq;
 622
 623   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY
 624                      && s->session_state != SESSION_STATE_LISTENING))
 625     {
 626       /* Session is closed so app will never clean up. Flush rx fifo */
 627       if (s->session_state == SESSION_STATE_CLOSED)
 628         svm_fifo_dequeue_drop_all (s->rx_fifo);
 629       return 0;
 630     }
 631
 632   if (app_worker_application_is_builtin (app_wrk))
 633     {
 634       application_t *app = application_get (app_wrk->app_index);
 635       return app->cb_fns.builtin_app_rx_callback (s);
 636     }
 637
 638   if (svm_fifo_has_event (s->rx_fifo) || svm_fifo_is_empty (s->rx_fifo))
 639     return 0;
 640
 641   mq = app_wrk->event_queue;
 642   if (lock)
 643     svm_msg_q_lock (mq);
 644
 645   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 646     {
 647       clib_warning ("evt q rings full");
 648       if (lock)
 649         svm_msg_q_unlock (mq);
 650       return -1;
 651     }
 652
 653   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 654   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
 655
 656   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 657   evt->fifo = s->rx_fifo;
 658   evt->event_type = FIFO_EVENT_APP_RX;
 659
 660   (void) svm_fifo_set_event (s->rx_fifo);
 661
 662   if (app_enqueue_evt (mq, &msg, lock))
 663     return -1;
 664   return 0;
 665 }
 666
 667 static inline int
 668 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s, u8 lock)
 669 {
 670   svm_msg_q_t *mq;
 671   session_event_t *evt;
 672   svm_msg_q_msg_t msg;
 673
 674   if (app_worker_application_is_builtin (app_wrk))
 675     return 0;
 676
 677   mq = app_wrk->event_queue;
 678   if (lock)
 679     svm_msg_q_lock (mq);
 680
 681   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
 682     {
 683       clib_warning ("evt q rings full");
 684       if (lock)
 685         svm_msg_q_unlock (mq);
 686       return -1;
 687     }
 688
 689   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
 690   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
 691
 692   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
 693   evt->event_type = FIFO_EVENT_APP_TX;
 694   evt->fifo = s->tx_fifo;
 695
 696   return app_enqueue_evt (mq, &msg, lock);
 697 }
 698
 699 /* *INDENT-OFF* */
 700 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
 701                                        session_t *s,
 702                                        u8 lock);
 703 static app_send_evt_handler_fn * const app_send_evt_handler_fns[3] = {
 704     app_send_io_evt_rx,
 705     0,
 706     app_send_io_evt_tx,
 707 };
 708 /* *INDENT-ON* */
 709
 710 /**
 711  * Send event to application
 712  *
 713  * Logic from queue perspective is non-blocking. If there's
 714  * not enough space to enqueue a message, we return.
 715  */
 716 int
 717 app_worker_send_event (app_worker_t * app, session_t * s, u8 evt_type)
 718 {
 719   ASSERT (app && evt_type <= FIFO_EVENT_APP_TX);
 720   return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
 721 }
 722
 723 /**
 724  * Send event to application
 725  *
 726  * Logic from queue perspective is blocking. However, if queue is full,
 727  * we return.
 728  */
 729 int
 730 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
 731                                 u8 evt_type)
 732 {
 733   return app_send_evt_handler_fns[evt_type] (app, s, 1 /* lock */ );
 734 }
 735
 736 segment_manager_t *
 737 app_worker_get_local_segment_manager (app_worker_t * app_worker)
 738 {
 739   return segment_manager_get (app_worker->local_segment_manager);
 740 }
 741
 742 segment_manager_t *
 743 app_worker_get_local_segment_manager_w_session (app_worker_t * app_wrk,
 744                                                 local_session_t * ls)
 745 {
 746   session_t *listener;
 747   if (application_local_session_listener_has_transport (ls))
 748     {
 749       listener = listen_session_get (ls->listener_index);
 750       return app_worker_get_listen_segment_manager (app_wrk, listener);
 751     }
 752   return segment_manager_get (app_wrk->local_segment_manager);
 753 }
 754
 755 int
 756 app_worker_local_session_cleanup (app_worker_t * client_wrk,
 757                                   app_worker_t * server_wrk,
 758                                   local_session_t * ls)
 759 {
 760   svm_fifo_segment_private_t *seg;
 761   session_t *listener;
 762   segment_manager_t *sm;
 763   u64 client_key;
 764   u8 has_transport;
 765
 766   /* Retrieve listener transport type as it is the one that decides where
 767    * the fifos are allocated */
 768   has_transport = application_local_session_listener_has_transport (ls);
 769   if (!has_transport)
 770     sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
 771   else
 772     {
 773       listener = listen_session_get (ls->listener_index);
 774       sm = app_worker_get_listen_segment_manager (server_wrk, listener);
 775     }
 776
 777   seg = segment_manager_get_segment (sm, ls->svm_segment_index);
 778   if (client_wrk)
 779     {
 780       client_key = application_client_local_connect_key (ls);
 781       hash_unset (client_wrk->local_connects, client_key);
 782     }
 783
 784   if (!has_transport)
 785     {
 786       application_t *server = application_get (server_wrk->app_index);
 787       u64 segment_handle = segment_manager_segment_handle (sm, seg);
 788       server->cb_fns.del_segment_callback (server_wrk->api_client_index,
 789                                            segment_handle);
 790       if (client_wrk)
 791         {
 792           application_t *client = application_get (client_wrk->app_index);
 793           client->cb_fns.del_segment_callback (client_wrk->api_client_index,
 794                                                segment_handle);
 795         }
 796       segment_manager_del_segment (sm, seg);
 797     }
 798
 799   app_worker_local_session_free (server_wrk, ls);
 800
 801   return 0;
 802 }
 803
 804 static void
 805 application_local_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
 806 {
 807   int fd;
 808
 809   /*
 810    * segment manager initializes only the producer eventds, since vpp is
 811    * typically the producer. But for local sessions, we also pass to the
 812    * apps the mqs they listen on for events from peer apps, so they are also
 813    * consumer fds.
 814    */
 815   fd = svm_msg_q_get_producer_eventfd (sq);
 816   svm_msg_q_set_consumer_eventfd (sq, fd);
 817   fd = svm_msg_q_get_producer_eventfd (cq);
 818   svm_msg_q_set_consumer_eventfd (cq, fd);
 819 }
 820
 821 int
 822 app_worker_local_session_connect (app_worker_t * client_wrk,
 823                                   app_worker_t * server_wrk,
 824                                   local_session_t * ll, u32 opaque)
 825 {
 826   u32 seg_size, evt_q_sz, evt_q_elts, margin = 16 << 10;
 827   u32 round_rx_fifo_sz, round_tx_fifo_sz, sm_index;
 828   segment_manager_properties_t *props, *cprops;
 829   int rv, has_transport, seg_index;
 830   svm_fifo_segment_private_t *seg;
 831   application_t *server, *client;
 832   segment_manager_t *sm;
 833   local_session_t *ls;
 834   svm_msg_q_t *sq, *cq;
 835   u64 segment_handle;
 836
 837   ls = app_worker_local_session_alloc (server_wrk);
 838   server = application_get (server_wrk->app_index);
 839   client = application_get (client_wrk->app_index);
 840
 841   props = application_segment_manager_properties (server);
 842   cprops = application_segment_manager_properties (client);
 843   evt_q_elts = props->evt_q_size + cprops->evt_q_size;
 844   evt_q_sz = segment_manager_evt_q_expected_size (evt_q_elts);
 845   round_rx_fifo_sz = 1 << max_log2 (props->rx_fifo_size);
 846   round_tx_fifo_sz = 1 << max_log2 (props->tx_fifo_size);
 847   seg_size = round_rx_fifo_sz + round_tx_fifo_sz + evt_q_sz + margin;
 848
 849   has_transport = session_has_transport ((session_t *) ll);
 850   if (!has_transport)
 851     {
 852       /* Local sessions don't have backing transport */
 853       ls->port = ll->port;
 854       sm = app_worker_get_local_segment_manager (server_wrk);
 855     }
 856   else
 857     {
 858       session_t *sl = (session_t *) ll;
 859       transport_connection_t *tc;
 860       tc = listen_session_get_transport (sl);
 861       ls->port = tc->lcl_port;
 862       sm = app_worker_get_listen_segment_manager (server_wrk, sl);
 863     }
 864
 865   seg_index = segment_manager_add_segment (sm, seg_size);
 866   if (seg_index < 0)
 867     {
 868       clib_warning ("failed to add new cut-through segment");
 869       return seg_index;
 870     }
 871   seg = segment_manager_get_segment_w_lock (sm, seg_index);
 872   sq = segment_manager_alloc_queue (seg, props);
 873   cq = segment_manager_alloc_queue (seg, cprops);
 874
 875   if (props->use_mq_eventfd)
 876     application_local_session_fix_eventds (sq, cq);
 877
 878   ls->server_evt_q = pointer_to_uword (sq);
 879   ls->client_evt_q = pointer_to_uword (cq);
 880   rv = segment_manager_try_alloc_fifos (seg, props->rx_fifo_size,
 881                                         props->tx_fifo_size,
 882                                         &ls->rx_fifo, &ls->tx_fifo);
 883   if (rv)
 884     {
 885       clib_warning ("failed to add fifos in cut-through segment");
 886       segment_manager_segment_reader_unlock (sm);
 887       goto failed;
 888     }
 889   sm_index = segment_manager_index (sm);
 890   ls->rx_fifo->ct_session_index = ls->session_index;
 891   ls->tx_fifo->ct_session_index = ls->session_index;
 892   ls->rx_fifo->segment_manager = sm_index;
 893   ls->tx_fifo->segment_manager = sm_index;
 894   ls->rx_fifo->segment_index = seg_index;
 895   ls->tx_fifo->segment_index = seg_index;
 896   ls->svm_segment_index = seg_index;
 897   ls->listener_index = ll->session_index;
 898   ls->client_wrk_index = client_wrk->wrk_index;
 899   ls->client_opaque = opaque;
 900   ls->listener_session_type = ll->session_type;
 901   ls->session_state = SESSION_STATE_READY;
 902
 903   segment_handle = segment_manager_segment_handle (sm, seg);
 904   if ((rv = server->cb_fns.add_segment_callback (server_wrk->api_client_index,
 905                                                  segment_handle)))
 906     {
 907       clib_warning ("failed to notify server of new segment");
 908       segment_manager_segment_reader_unlock (sm);
 909       goto failed;
 910     }
 911   segment_manager_segment_reader_unlock (sm);
 912   if ((rv = server->cb_fns.session_accept_callback ((session_t *) ls)))
 913     {
 914       clib_warning ("failed to send accept cut-through notify to server");
 915       goto failed;
 916     }
 917   if (server->flags & APP_OPTIONS_FLAGS_IS_BUILTIN)
 918     app_worker_local_session_connect_notify (ls);
 919
 920   return 0;
 921
 922 failed:
 923   if (!has_transport)
 924     segment_manager_del_segment (sm, seg);
 925   return rv;
 926 }
 927
 928 int
 929 app_worker_local_session_connect_notify (local_session_t * ls)
 930 {
 931   svm_fifo_segment_private_t *seg;
 932   app_worker_t *client_wrk, *server_wrk;
 933   segment_manager_t *sm;
 934   application_t *client;
 935   int rv, is_fail = 0;
 936   u64 segment_handle;
 937   u64 client_key;
 938
 939   client_wrk = app_worker_get (ls->client_wrk_index);
 940   server_wrk = app_worker_get (ls->app_wrk_index);
 941   client = application_get (client_wrk->app_index);
 942
 943   sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
 944   seg = segment_manager_get_segment_w_lock (sm, ls->svm_segment_index);
 945   segment_handle = segment_manager_segment_handle (sm, seg);
 946   if ((rv = client->cb_fns.add_segment_callback (client_wrk->api_client_index,
 947                                                  segment_handle)))
 948     {
 949       clib_warning ("failed to notify client %u of new segment",
 950                     ls->client_wrk_index);
 951       segment_manager_segment_reader_unlock (sm);
 952       app_worker_local_session_disconnect (ls->client_wrk_index, ls);
 953       is_fail = 1;
 954     }
 955   else
 956     {
 957       segment_manager_segment_reader_unlock (sm);
 958     }
 959
 960   client->cb_fns.session_connected_callback (client_wrk->wrk_index,
 961                                              ls->client_opaque,
 962                                              (session_t *) ls, is_fail);
 963
 964   client_key = application_client_local_connect_key (ls);
 965   hash_set (client_wrk->local_connects, client_key, client_key);
 966   return 0;
 967 }
 968
 969 int
 970 app_worker_local_session_disconnect (u32 app_wrk_index, local_session_t * ls)
 971 {
 972   app_worker_t *client_wrk, *server_wrk;
 973
 974   client_wrk = app_worker_get_if_valid (ls->client_wrk_index);
 975   server_wrk = app_worker_get (ls->app_wrk_index);
 976
 977   if (ls->session_state == SESSION_STATE_CLOSED)
 978     return app_worker_local_session_cleanup (client_wrk, server_wrk, ls);
 979
 980   if (app_wrk_index == ls->client_wrk_index)
 981     {
 982       mq_send_local_session_disconnected_cb (ls->app_wrk_index, ls);
 983     }
 984   else
 985     {
 986       if (!client_wrk)
 987         {
 988           return app_worker_local_session_cleanup (client_wrk, server_wrk,
 989                                                    ls);
 990         }
 991       else if (ls->session_state < SESSION_STATE_READY)
 992         {
 993           application_t *client = application_get (client_wrk->app_index);
 994           client->cb_fns.session_connected_callback (client_wrk->wrk_index,
 995                                                      ls->client_opaque,
 996                                                      (session_t *) ls,
 997                                                      1 /* is_fail */ );
 998           ls->session_state = SESSION_STATE_CLOSED;
 999           return app_worker_local_session_cleanup (client_wrk, server_wrk,
1000                                                    ls);
1001         }
1002       else
1003         {
1004           mq_send_local_session_disconnected_cb (client_wrk->wrk_index, ls);
1005         }
1006     }
1007
1008   ls->session_state = SESSION_STATE_CLOSED;
1009
1010   return 0;
1011 }
1012
1013 int
1014 app_worker_local_session_disconnect_w_index (u32 app_wrk_index, u32 ls_index)
1015 {
1016   app_worker_t *app_wrk;
1017   local_session_t *ls;
1018   app_wrk = app_worker_get (app_wrk_index);
1019   ls = app_worker_get_local_session (app_wrk, ls_index);
1020   return app_worker_local_session_disconnect (app_wrk_index, ls);
1021 }
1022
1023 u8 *
1024 format_app_worker_listener (u8 * s, va_list * args)
1025 {
1026   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
1027   u64 handle = va_arg (*args, u64);
1028   u32 sm_index = va_arg (*args, u32);
1029   int verbose = va_arg (*args, int);
1030   session_t *listener;
1031   const u8 *app_name;
1032   u8 *str;
1033
1034   if (!app_wrk)
1035     {
1036       if (verbose)
1037         s = format (s, "%-40s%-25s%=10s%-15s%-15s%-10s", "Connection", "App",
1038                     "Wrk", "API Client", "ListenerID", "SegManager");
1039       else
1040         s = format (s, "%-40s%-25s%=10s", "Connection", "App", "Wrk");
1041
1042       return s;
1043     }
1044
1045   app_name = application_name_from_index (app_wrk->app_index);
1046   listener = listen_session_get_from_handle (handle);
1047   str = format (0, "%U", format_stream_session, listener, verbose);
1048
1049   if (verbose)
1050     {
1051       char buf[32];
1052       sprintf (buf, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
1053       s = format (s, "%-40s%-25s%=10s%-15u%-15u%-10u", str, app_name,
1054                   buf, app_wrk->api_client_index, handle, sm_index);
1055     }
1056   else
1057     s = format (s, "%-40s%-25s%=10u", str, app_name, app_wrk->wrk_map_index);
1058
1059   return s;
1060 }
1061
1062 u8 *
1063 format_app_worker (u8 * s, va_list * args)
1064 {
1065   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
1066   u32 indent = 1;
1067
1068   s = format (s, "%U wrk-index %u app-index %u map-index %u "
1069               "api-client-index %d\n", format_white_space, indent,
1070               app_wrk->wrk_index, app_wrk->app_index, app_wrk->wrk_map_index,
1071               app_wrk->api_client_index);
1072   return s;
1073 }
1074
1075 void
1076 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
1077 {
1078   svm_fifo_segment_private_t *fifo_segment;
1079   vlib_main_t *vm = vlib_get_main ();
1080   segment_manager_t *sm;
1081   const u8 *app_name;
1082   u8 *s = 0;
1083
1084   /* Header */
1085   if (!app_wrk)
1086     {
1087       if (verbose)
1088         vlib_cli_output (vm, "%-40s%-20s%-15s%-10s", "Connection", "App",
1089                          "API Client", "SegManager");
1090       else
1091         vlib_cli_output (vm, "%-40s%-20s", "Connection", "App");
1092       return;
1093     }
1094
1095   if (app_wrk->connects_seg_manager == (u32) ~ 0)
1096     return;
1097
1098   app_name = application_name_from_index (app_wrk->app_index);
1099
1100   /* Across all fifo segments */
1101   sm = segment_manager_get (app_wrk->connects_seg_manager);
1102
1103   /* *INDENT-OFF* */
1104   segment_manager_foreach_segment_w_lock (fifo_segment, sm, ({
1105     svm_fifo_t *fifo;
1106     u8 *str;
1107
1108     fifo = svm_fifo_segment_get_fifo_list (fifo_segment);
1109     while (fifo)
1110       {
1111         u32 session_index, thread_index;
1112         session_t *session;
1113
1114         session_index = fifo->master_session_index;
1115         thread_index = fifo->master_thread_index;
1116
1117         session = session_get (session_index, thread_index);
1118         str = format (0, "%U", format_stream_session, session, verbose);
1119
1120         if (verbose)
1121           s = format (s, "%-40s%-20s%-15u%-10u", str, app_name,
1122                       app_wrk->api_client_index, app_wrk->connects_seg_manager);
1123         else
1124           s = format (s, "%-40s%-20s", str, app_name);
1125
1126         vlib_cli_output (vm, "%v", s);
1127         vec_reset_length (s);
1128         vec_free (str);
1129
1130         fifo = fifo->next;
1131       }
1132     vec_free (s);
1133   }));
1134   /* *INDENT-ON* */
1135 }
1136
1137 void
1138 app_worker_format_local_sessions (app_worker_t * app_wrk, int verbose)
1139 {
1140   vlib_main_t *vm = vlib_get_main ();
1141   app_worker_t *client_wrk;
1142   local_session_t *ls;
1143   transport_proto_t tp;
1144   u8 *conn = 0;
1145
1146   /* Header */
1147   if (app_wrk == 0)
1148     {
1149       vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "ServerApp",
1150                        "ClientApp");
1151       return;
1152     }
1153
1154   if (!pool_elts (app_wrk->local_sessions)
1155       && !pool_elts (app_wrk->local_connects))
1156     return;
1157
1158   /* *INDENT-OFF* */
1159   pool_foreach (ls, app_wrk->local_sessions, ({
1160     tp = session_type_transport_proto(ls->listener_session_type);
1161     conn = format (0, "[L][%U] *:%u", format_transport_proto_short, tp,
1162                    ls->port);
1163     client_wrk = app_worker_get (ls->client_wrk_index);
1164     vlib_cli_output (vm, "%-40v%-15u%-20u", conn, ls->app_index,
1165                      client_wrk->app_index);
1166     vec_reset_length (conn);
1167   }));
1168   /* *INDENT-ON* */
1169
1170   vec_free (conn);
1171 }
1172
1173 void
1174 app_worker_format_local_connects (app_worker_t * app, int verbose)
1175 {
1176   vlib_main_t *vm = vlib_get_main ();
1177   u32 app_wrk_index, session_index;
1178   app_worker_t *server_wrk;
1179   local_session_t *ls;
1180   u64 client_key;
1181   u64 value;
1182
1183   /* Header */
1184   if (app == 0)
1185     {
1186       if (verbose)
1187         vlib_cli_output (vm, "%-40s%-15s%-20s%-10s", "Connection", "App",
1188                          "Peer App", "SegManager");
1189       else
1190         vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "App",
1191                          "Peer App");
1192       return;
1193     }
1194
1195   if (!app->local_connects)
1196     return;
1197
1198   /* *INDENT-OFF* */
1199   hash_foreach (client_key, value, app->local_connects, ({
1200     application_client_local_connect_key_parse (client_key, &app_wrk_index,
1201                                                 &session_index);
1202     server_wrk = app_worker_get (app_wrk_index);
1203     ls = app_worker_get_local_session (server_wrk, session_index);
1204     vlib_cli_output (vm, "%-40s%-15s%-20s", "TODO", ls->app_wrk_index,
1205                      ls->client_wrk_index);
1206   }));
1207   /* *INDENT-ON* */
1208 }
1209
1210 /*
1211  * fd.io coding-style-patch-verification: ON
1212  *
1213  * Local Variables:
1214  * eval: (c-set-style "gnu")
1215  * End:
1216  */