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