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