src/vnet/session/application_local.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_local.h>
  17 #include <vnet/session/session.h>
  18
  19 local_session_t *
  20 application_local_listen_session_alloc (application_t * app)
  21 {
  22   local_session_t *ll;
  23   pool_get_zero (app->local_listen_sessions, ll);
  24   ll->session_index = ll - app->local_listen_sessions;
  25   ll->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE, 0);
  26   ll->app_index = app->app_index;
  27   ll->session_state = SESSION_STATE_LISTENING;
  28   return ll;
  29 }
  30
  31 void
  32 application_local_listen_session_free (application_t * app,
  33                                        local_session_t * ll)
  34 {
  35   pool_put (app->local_listen_sessions, ll);
  36   if (CLIB_DEBUG)
  37     clib_memset (ll, 0xfb, sizeof (*ll));
  38 }
  39
  40 void
  41 application_local_listener_session_endpoint (local_session_t * ll,
  42                                              session_endpoint_t * sep)
  43 {
  44   sep->transport_proto =
  45     session_type_transport_proto (ll->listener_session_type);
  46   sep->port = ll->port;
  47   sep->is_ip4 = ll->listener_session_type & 1;
  48 }
  49
  50 local_session_t *
  51 app_worker_local_session_alloc (app_worker_t * app_wrk)
  52 {
  53   local_session_t *s;
  54   pool_get (app_wrk->local_sessions, s);
  55   clib_memset (s, 0, sizeof (*s));
  56   s->app_wrk_index = app_wrk->wrk_index;
  57   s->session_index = s - app_wrk->local_sessions;
  58   s->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE, 0);
  59   return s;
  60 }
  61
  62 void
  63 app_worker_local_session_free (app_worker_t * app_wrk, local_session_t * s)
  64 {
  65   pool_put (app_wrk->local_sessions, s);
  66   if (CLIB_DEBUG)
  67     clib_memset (s, 0xfc, sizeof (*s));
  68 }
  69
  70 local_session_t *
  71 app_worker_get_local_session (app_worker_t * app_wrk, u32 session_index)
  72 {
  73   if (pool_is_free_index (app_wrk->local_sessions, session_index))
  74     return 0;
  75   return pool_elt_at_index (app_wrk->local_sessions, session_index);
  76 }
  77
  78 local_session_t *
  79 app_worker_get_local_session_from_handle (session_handle_t handle)
  80 {
  81   app_worker_t *server_wrk;
  82   u32 session_index, server_wrk_index;
  83   local_session_parse_handle (handle, &server_wrk_index, &session_index);
  84   server_wrk = app_worker_get_if_valid (server_wrk_index);
  85   if (!server_wrk)
  86     return 0;
  87   return app_worker_get_local_session (server_wrk, session_index);
  88 }
  89
  90 static inline u64
  91 application_client_local_connect_key (local_session_t * ls)
  92 {
  93   return (((u64) ls->app_wrk_index) << 32 | (u64) ls->session_index);
  94 }
  95
  96 static inline void
  97 application_client_local_connect_key_parse (u64 key, u32 * app_wrk_index,
  98                                             u32 * session_index)
  99 {
 100   *app_wrk_index = key >> 32;
 101   *session_index = key & 0xFFFFFFFF;
 102 }
 103
 104 void
 105 app_worker_local_sessions_free (app_worker_t * app_wrk)
 106 {
 107   u32 index, server_wrk_index, session_index;
 108   u64 handle, *handles = 0;
 109   app_worker_t *server_wrk;
 110   segment_manager_t *sm;
 111   local_session_t *ls;
 112   int i;
 113
 114   /*
 115    * Local sessions
 116    */
 117   if (app_wrk->local_sessions)
 118     {
 119       /* *INDENT-OFF* */
 120       pool_foreach (ls, app_wrk->local_sessions, ({
 121         app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
 122       }));
 123       /* *INDENT-ON* */
 124     }
 125
 126   /*
 127    * Local connects
 128    */
 129   vec_reset_length (handles);
 130   /* *INDENT-OFF* */
 131   hash_foreach (handle, index, app_wrk->local_connects, ({
 132     vec_add1 (handles, handle);
 133   }));
 134   /* *INDENT-ON* */
 135
 136   for (i = 0; i < vec_len (handles); i++)
 137     {
 138       application_client_local_connect_key_parse (handles[i],
 139                                                   &server_wrk_index,
 140                                                   &session_index);
 141       server_wrk = app_worker_get_if_valid (server_wrk_index);
 142       if (server_wrk)
 143         {
 144           ls = app_worker_get_local_session (server_wrk, session_index);
 145           app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
 146         }
 147     }
 148
 149   sm = segment_manager_get (app_wrk->local_segment_manager);
 150   sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
 151   segment_manager_del (sm);
 152 }
 153
 154 int
 155 app_worker_local_session_cleanup (app_worker_t * client_wrk,
 156                                   app_worker_t * server_wrk,
 157                                   local_session_t * ls)
 158 {
 159   svm_fifo_segment_private_t *seg;
 160   session_t *listener;
 161   segment_manager_t *sm;
 162   u64 client_key;
 163   u8 has_transport;
 164
 165   /* Retrieve listener transport type as it is the one that decides where
 166    * the fifos are allocated */
 167   has_transport = application_local_session_listener_has_transport (ls);
 168   if (!has_transport)
 169     sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
 170   else
 171     {
 172       listener = listen_session_get (ls->listener_index);
 173       sm = app_worker_get_listen_segment_manager (server_wrk, listener);
 174     }
 175
 176   seg = segment_manager_get_segment (sm, ls->svm_segment_index);
 177   if (client_wrk)
 178     {
 179       client_key = application_client_local_connect_key (ls);
 180       hash_unset (client_wrk->local_connects, client_key);
 181     }
 182
 183   if (!has_transport)
 184     {
 185       application_t *server = application_get (server_wrk->app_index);
 186       u64 segment_handle = segment_manager_segment_handle (sm, seg);
 187       server->cb_fns.del_segment_callback (server_wrk->api_client_index,
 188                                            segment_handle);
 189       if (client_wrk)
 190         {
 191           application_t *client = application_get (client_wrk->app_index);
 192           client->cb_fns.del_segment_callback (client_wrk->api_client_index,
 193                                                segment_handle);
 194         }
 195       segment_manager_del_segment (sm, seg);
 196     }
 197
 198   app_worker_local_session_free (server_wrk, ls);
 199
 200   return 0;
 201 }
 202
 203 int
 204 app_worker_local_session_connect_notify (local_session_t * ls)
 205 {
 206   svm_fifo_segment_private_t *seg;
 207   app_worker_t *client_wrk, *server_wrk;
 208   segment_manager_t *sm;
 209   application_t *client;
 210   int rv, is_fail = 0;
 211   u64 segment_handle;
 212   u64 client_key;
 213
 214   client_wrk = app_worker_get (ls->client_wrk_index);
 215   server_wrk = app_worker_get (ls->app_wrk_index);
 216   client = application_get (client_wrk->app_index);
 217
 218   sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
 219   seg = segment_manager_get_segment_w_lock (sm, ls->svm_segment_index);
 220   segment_handle = segment_manager_segment_handle (sm, seg);
 221   if ((rv = client->cb_fns.add_segment_callback (client_wrk->api_client_index,
 222                                                  segment_handle)))
 223     {
 224       clib_warning ("failed to notify client %u of new segment",
 225                     ls->client_wrk_index);
 226       segment_manager_segment_reader_unlock (sm);
 227       app_worker_local_session_disconnect (ls->client_wrk_index, ls);
 228       is_fail = 1;
 229     }
 230   else
 231     {
 232       segment_manager_segment_reader_unlock (sm);
 233     }
 234
 235   client->cb_fns.session_connected_callback (client_wrk->wrk_index,
 236                                              ls->client_opaque,
 237                                              (session_t *) ls, is_fail);
 238
 239   client_key = application_client_local_connect_key (ls);
 240   hash_set (client_wrk->local_connects, client_key, client_key);
 241   return 0;
 242 }
 243
 244 static void
 245 application_local_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
 246 {
 247   int fd;
 248
 249   /*
 250    * segment manager initializes only the producer eventds, since vpp is
 251    * typically the producer. But for local sessions, we also pass to the
 252    * apps the mqs they listen on for events from peer apps, so they are also
 253    * consumer fds.
 254    */
 255   fd = svm_msg_q_get_producer_eventfd (sq);
 256   svm_msg_q_set_consumer_eventfd (sq, fd);
 257   fd = svm_msg_q_get_producer_eventfd (cq);
 258   svm_msg_q_set_consumer_eventfd (cq, fd);
 259 }
 260
 261 int
 262 app_worker_local_session_connect (app_worker_t * client_wrk,
 263                                   app_worker_t * server_wrk,
 264                                   local_session_t * ll, u32 opaque)
 265 {
 266   u32 seg_size, evt_q_sz, evt_q_elts, margin = 16 << 10;
 267   u32 round_rx_fifo_sz, round_tx_fifo_sz, sm_index;
 268   segment_manager_properties_t *props, *cprops;
 269   int rv, has_transport, seg_index;
 270   svm_fifo_segment_private_t *seg;
 271   application_t *server, *client;
 272   segment_manager_t *sm;
 273   local_session_t *ls;
 274   svm_msg_q_t *sq, *cq;
 275   u64 segment_handle;
 276
 277   ls = app_worker_local_session_alloc (server_wrk);
 278   server = application_get (server_wrk->app_index);
 279   client = application_get (client_wrk->app_index);
 280
 281   props = application_segment_manager_properties (server);
 282   cprops = application_segment_manager_properties (client);
 283   evt_q_elts = props->evt_q_size + cprops->evt_q_size;
 284   evt_q_sz = segment_manager_evt_q_expected_size (evt_q_elts);
 285   round_rx_fifo_sz = 1 << max_log2 (props->rx_fifo_size);
 286   round_tx_fifo_sz = 1 << max_log2 (props->tx_fifo_size);
 287   seg_size = round_rx_fifo_sz + round_tx_fifo_sz + evt_q_sz + margin;
 288
 289   has_transport = session_has_transport ((session_t *) ll);
 290   if (!has_transport)
 291     {
 292       /* Local sessions don't have backing transport */
 293       ls->port = ll->port;
 294       sm = app_worker_get_local_segment_manager (server_wrk);
 295     }
 296   else
 297     {
 298       session_t *sl = (session_t *) ll;
 299       transport_connection_t *tc;
 300       tc = listen_session_get_transport (sl);
 301       ls->port = tc->lcl_port;
 302       sm = app_worker_get_listen_segment_manager (server_wrk, sl);
 303     }
 304
 305   seg_index = segment_manager_add_segment (sm, seg_size);
 306   if (seg_index < 0)
 307     {
 308       clib_warning ("failed to add new cut-through segment");
 309       return seg_index;
 310     }
 311   seg = segment_manager_get_segment_w_lock (sm, seg_index);
 312   sq = segment_manager_alloc_queue (seg, props);
 313   cq = segment_manager_alloc_queue (seg, cprops);
 314
 315   if (props->use_mq_eventfd)
 316     application_local_session_fix_eventds (sq, cq);
 317
 318   ls->server_evt_q = pointer_to_uword (sq);
 319   ls->client_evt_q = pointer_to_uword (cq);
 320   rv = segment_manager_try_alloc_fifos (seg, props->rx_fifo_size,
 321                                         props->tx_fifo_size,
 322                                         &ls->rx_fifo, &ls->tx_fifo);
 323   if (rv)
 324     {
 325       clib_warning ("failed to add fifos in cut-through segment");
 326       segment_manager_segment_reader_unlock (sm);
 327       goto failed;
 328     }
 329   sm_index = segment_manager_index (sm);
 330   ls->rx_fifo->ct_session_index = ls->session_index;
 331   ls->tx_fifo->ct_session_index = ls->session_index;
 332   ls->rx_fifo->segment_manager = sm_index;
 333   ls->tx_fifo->segment_manager = sm_index;
 334   ls->rx_fifo->segment_index = seg_index;
 335   ls->tx_fifo->segment_index = seg_index;
 336   ls->svm_segment_index = seg_index;
 337   ls->listener_index = ll->session_index;
 338   ls->client_wrk_index = client_wrk->wrk_index;
 339   ls->client_opaque = opaque;
 340   ls->listener_session_type = ll->session_type;
 341   ls->session_state = SESSION_STATE_READY;
 342
 343   segment_handle = segment_manager_segment_handle (sm, seg);
 344   if ((rv = server->cb_fns.add_segment_callback (server_wrk->api_client_index,
 345                                                  segment_handle)))
 346     {
 347       clib_warning ("failed to notify server of new segment");
 348       segment_manager_segment_reader_unlock (sm);
 349       goto failed;
 350     }
 351   segment_manager_segment_reader_unlock (sm);
 352   if ((rv = server->cb_fns.session_accept_callback ((session_t *) ls)))
 353     {
 354       clib_warning ("failed to send accept cut-through notify to server");
 355       goto failed;
 356     }
 357   if (server->flags & APP_OPTIONS_FLAGS_IS_BUILTIN)
 358     app_worker_local_session_connect_notify (ls);
 359
 360   return 0;
 361
 362 failed:
 363   if (!has_transport)
 364     segment_manager_del_segment (sm, seg);
 365   return rv;
 366 }
 367
 368 int
 369 app_worker_local_session_disconnect (u32 app_wrk_index, local_session_t * ls)
 370 {
 371   app_worker_t *client_wrk, *server_wrk;
 372
 373   client_wrk = app_worker_get_if_valid (ls->client_wrk_index);
 374   server_wrk = app_worker_get (ls->app_wrk_index);
 375
 376   if (ls->session_state == SESSION_STATE_CLOSED)
 377     return app_worker_local_session_cleanup (client_wrk, server_wrk, ls);
 378
 379   if (app_wrk_index == ls->client_wrk_index)
 380     {
 381       mq_send_local_session_disconnected_cb (ls->app_wrk_index, ls);
 382     }
 383   else
 384     {
 385       if (!client_wrk)
 386         {
 387           return app_worker_local_session_cleanup (client_wrk, server_wrk,
 388                                                    ls);
 389         }
 390       else if (ls->session_state < SESSION_STATE_READY)
 391         {
 392           application_t *client = application_get (client_wrk->app_index);
 393           client->cb_fns.session_connected_callback (client_wrk->wrk_index,
 394                                                      ls->client_opaque,
 395                                                      (session_t *) ls,
 396                                                      1 /* is_fail */ );
 397           ls->session_state = SESSION_STATE_CLOSED;
 398           return app_worker_local_session_cleanup (client_wrk, server_wrk,
 399                                                    ls);
 400         }
 401       else
 402         {
 403           mq_send_local_session_disconnected_cb (client_wrk->wrk_index, ls);
 404         }
 405     }
 406
 407   ls->session_state = SESSION_STATE_CLOSED;
 408
 409   return 0;
 410 }
 411
 412 int
 413 app_worker_local_session_disconnect_w_index (u32 app_wrk_index, u32 ls_index)
 414 {
 415   app_worker_t *app_wrk;
 416   local_session_t *ls;
 417   app_wrk = app_worker_get (app_wrk_index);
 418   ls = app_worker_get_local_session (app_wrk, ls_index);
 419   return app_worker_local_session_disconnect (app_wrk_index, ls);
 420 }
 421
 422 void
 423 app_worker_format_local_sessions (app_worker_t * app_wrk, int verbose)
 424 {
 425   vlib_main_t *vm = vlib_get_main ();
 426   app_worker_t *client_wrk;
 427   local_session_t *ls;
 428   transport_proto_t tp;
 429   u8 *conn = 0;
 430
 431   /* Header */
 432   if (app_wrk == 0)
 433     {
 434       vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "ServerApp",
 435                        "ClientApp");
 436       return;
 437     }
 438
 439   if (!pool_elts (app_wrk->local_sessions)
 440       && !pool_elts (app_wrk->local_connects))
 441     return;
 442
 443   /* *INDENT-OFF* */
 444   pool_foreach (ls, app_wrk->local_sessions, ({
 445     tp = session_type_transport_proto(ls->listener_session_type);
 446     conn = format (0, "[L][%U] *:%u", format_transport_proto_short, tp,
 447                    ls->port);
 448     client_wrk = app_worker_get (ls->client_wrk_index);
 449     vlib_cli_output (vm, "%-40v%-15u%-20u", conn, ls->app_index,
 450                      client_wrk->app_index);
 451     vec_reset_length (conn);
 452   }));
 453   /* *INDENT-ON* */
 454
 455   vec_free (conn);
 456 }
 457
 458 void
 459 app_worker_format_local_connects (app_worker_t * app, int verbose)
 460 {
 461   vlib_main_t *vm = vlib_get_main ();
 462   u32 app_wrk_index, session_index;
 463   app_worker_t *server_wrk;
 464   local_session_t *ls;
 465   u64 client_key;
 466   u64 value;
 467
 468   /* Header */
 469   if (app == 0)
 470     {
 471       if (verbose)
 472         vlib_cli_output (vm, "%-40s%-15s%-20s%-10s", "Connection", "App",
 473                          "Peer App", "SegManager");
 474       else
 475         vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "App",
 476                          "Peer App");
 477       return;
 478     }
 479
 480   if (!app->local_connects)
 481     return;
 482
 483   /* *INDENT-OFF* */
 484   hash_foreach (client_key, value, app->local_connects, ({
 485     application_client_local_connect_key_parse (client_key, &app_wrk_index,
 486                                                 &session_index);
 487     server_wrk = app_worker_get (app_wrk_index);
 488     ls = app_worker_get_local_session (server_wrk, session_index);
 489     vlib_cli_output (vm, "%-40s%-15s%-20s", "TODO", ls->app_wrk_index,
 490                      ls->client_wrk_index);
 491   }));
 492   /* *INDENT-ON* */
 493 }
 494
 495 /*
 496  * fd.io coding-style-patch-verification: ON
 497  *
 498  * Local Variables:
 499  * eval: (c-set-style "gnu")
 500  * End:
 501  */