src/vcl/vcl_locked.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
   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 <vcl/vcl_locked.h>
  17 #include <vcl/vcl_private.h>
  18
  19 typedef struct vls_shared_data_
  20 {
  21   clib_spinlock_t lock;
  22   u32 owner_wrk_index;
  23   u32 *workers_subscribed;
  24   clib_bitmap_t *listeners;
  25 } vls_shared_data_t;
  26
  27 typedef struct vcl_locked_session_
  28 {
  29   clib_spinlock_t lock;
  30   u32 session_index;
  31   u32 worker_index;
  32   u32 vls_index;
  33   u32 shared_data_index;
  34   /** VCL session owned by different workers because of migration */
  35   u32 owner_vcl_wrk_index;
  36   uword *vcl_wrk_index_to_session_index;
  37 } vcl_locked_session_t;
  38
  39 typedef struct vls_worker_
  40 {
  41   vcl_locked_session_t *vls_pool;
  42   uword *session_handle_to_vlsh_table;
  43   u32 wrk_index;
  44 } vls_worker_t;
  45
  46 typedef struct vls_local_
  47 {
  48   int vls_wrk_index;
  49   volatile int vls_mt_n_threads;
  50   pthread_mutex_t vls_mt_mq_mlock;
  51   pthread_mutex_t vls_mt_spool_mlock;
  52   volatile u8 select_mp_check;
  53   volatile u8 epoll_mp_check;
  54 } vls_process_local_t;
  55
  56 static vls_process_local_t vls_local;
  57 static vls_process_local_t *vlsl = &vls_local;
  58
  59 typedef struct vls_main_
  60 {
  61   vls_worker_t *workers;
  62   clib_rwlock_t vls_table_lock;
  63   /** Pool of data shared by sessions owned by different workers */
  64   vls_shared_data_t *shared_data_pool;
  65   clib_rwlock_t shared_data_lock;
  66 } vls_main_t;
  67
  68 vls_main_t *vlsm;
  69
  70 typedef enum vls_rpc_msg_type_
  71 {
  72   VLS_RPC_CLONE_AND_SHARE,
  73   VLS_RPC_SESS_CLEANUP,
  74 } vls_rpc_msg_type_e;
  75
  76 typedef struct vls_rpc_msg_
  77 {
  78   u8 type;
  79   u8 data[0];
  80 } vls_rpc_msg_t;
  81
  82 typedef struct vls_clone_and_share_msg_
  83 {
  84   u32 vls_index;                /**< vls to be shared */
  85   u32 session_index;            /**< vcl session to be shared */
  86   u32 origin_vls_wrk;           /**< vls worker that initiated the rpc */
  87   u32 origin_vls_index;         /**< vls session of the originator */
  88   u32 origin_vcl_wrk;           /**< vcl worker that initiated the rpc */
  89   u32 origin_session_index;     /**< vcl session of the originator */
  90 } vls_clone_and_share_msg_t;
  91
  92 typedef struct vls_sess_cleanup_msg_
  93 {
  94   u32 session_index;            /**< vcl session to be cleaned */
  95   u32 origin_vcl_wrk;           /**< worker that initiated the rpc */
  96 } vls_sess_cleanup_msg_t;
  97
  98 void vls_send_session_cleanup_rpc (vcl_worker_t * wrk,
  99                                    u32 dst_wrk_index, u32 dst_session_index);
 100 void vls_send_clone_and_share_rpc (vcl_worker_t * wrk,
 101                                    vcl_locked_session_t * vls,
 102                                    u32 session_index, u32 vls_wrk_index,
 103                                    u32 dst_wrk_index, u32 dst_vls_index,
 104                                    u32 dst_session_index);
 105
 106
 107 static inline u32
 108 vls_get_worker_index (void)
 109 {
 110   if (vls_mt_wrk_supported ())
 111     return vlsl->vls_wrk_index;
 112   else
 113     return vcl_get_worker_index ();
 114 }
 115
 116 static u32
 117 vls_shared_data_alloc (void)
 118 {
 119   vls_shared_data_t *vls_shd;
 120   u32 shd_index;
 121
 122   clib_rwlock_writer_lock (&vlsm->shared_data_lock);
 123   pool_get_zero (vlsm->shared_data_pool, vls_shd);
 124   clib_spinlock_init (&vls_shd->lock);
 125   shd_index = vls_shd - vlsm->shared_data_pool;
 126   clib_rwlock_writer_unlock (&vlsm->shared_data_lock);
 127
 128   return shd_index;
 129 }
 130
 131 static u32
 132 vls_shared_data_index (vls_shared_data_t * vls_shd)
 133 {
 134   return vls_shd - vlsm->shared_data_pool;
 135 }
 136
 137 vls_shared_data_t *
 138 vls_shared_data_get (u32 shd_index)
 139 {
 140   if (pool_is_free_index (vlsm->shared_data_pool, shd_index))
 141     return 0;
 142   return pool_elt_at_index (vlsm->shared_data_pool, shd_index);
 143 }
 144
 145 static void
 146 vls_shared_data_free (u32 shd_index)
 147 {
 148   vls_shared_data_t *vls_shd;
 149
 150   clib_rwlock_writer_lock (&vlsm->shared_data_lock);
 151   vls_shd = vls_shared_data_get (shd_index);
 152   clib_spinlock_free (&vls_shd->lock);
 153   clib_bitmap_free (vls_shd->listeners);
 154   vec_free (vls_shd->workers_subscribed);
 155   pool_put (vlsm->shared_data_pool, vls_shd);
 156   clib_rwlock_writer_unlock (&vlsm->shared_data_lock);
 157 }
 158
 159 static inline void
 160 vls_shared_data_pool_rlock (void)
 161 {
 162   clib_rwlock_reader_lock (&vlsm->shared_data_lock);
 163 }
 164
 165 static inline void
 166 vls_shared_data_pool_runlock (void)
 167 {
 168   clib_rwlock_reader_unlock (&vlsm->shared_data_lock);
 169 }
 170
 171 static inline void
 172 vls_mt_table_rlock (void)
 173 {
 174   if (vlsl->vls_mt_n_threads > 1)
 175     clib_rwlock_reader_lock (&vlsm->vls_table_lock);
 176 }
 177
 178 static inline void
 179 vls_mt_table_runlock (void)
 180 {
 181   if (vlsl->vls_mt_n_threads > 1)
 182     clib_rwlock_reader_unlock (&vlsm->vls_table_lock);
 183 }
 184
 185 static inline void
 186 vls_mt_table_wlock (void)
 187 {
 188   if (vlsl->vls_mt_n_threads > 1)
 189     clib_rwlock_writer_lock (&vlsm->vls_table_lock);
 190 }
 191
 192 static inline void
 193 vls_mt_table_wunlock (void)
 194 {
 195   if (vlsl->vls_mt_n_threads > 1)
 196     clib_rwlock_writer_unlock (&vlsm->vls_table_lock);
 197 }
 198
 199 typedef enum
 200 {
 201   VLS_MT_OP_READ,
 202   VLS_MT_OP_WRITE,
 203   VLS_MT_OP_SPOOL,
 204   VLS_MT_OP_XPOLL,
 205 } vls_mt_ops_t;
 206
 207 typedef enum
 208 {
 209   VLS_MT_LOCK_MQ = 1 << 0,
 210   VLS_MT_LOCK_SPOOL = 1 << 1
 211 } vls_mt_lock_type_t;
 212
 213 static void
 214 vls_mt_add (void)
 215 {
 216   vlsl->vls_mt_n_threads += 1;
 217
 218   /* If multi-thread workers are supported, for each new thread register a new
 219    * vcl worker with vpp. Otherwise, all threads use the same vcl worker, so
 220    * update the vcl worker's thread local worker index variable */
 221   if (vls_mt_wrk_supported ())
 222     {
 223       if (vppcom_worker_register () != VPPCOM_OK)
 224         VERR ("failed to register worker");
 225     }
 226   else
 227     vcl_set_worker_index (vlsl->vls_wrk_index);
 228 }
 229
 230 static inline void
 231 vls_mt_mq_lock (void)
 232 {
 233   pthread_mutex_lock (&vlsl->vls_mt_mq_mlock);
 234 }
 235
 236 static inline void
 237 vls_mt_mq_unlock (void)
 238 {
 239   pthread_mutex_unlock (&vlsl->vls_mt_mq_mlock);
 240 }
 241
 242 static inline void
 243 vls_mt_spool_lock (void)
 244 {
 245   pthread_mutex_lock (&vlsl->vls_mt_spool_mlock);
 246 }
 247
 248 static inline void
 249 vls_mt_create_unlock (void)
 250 {
 251   pthread_mutex_unlock (&vlsl->vls_mt_spool_mlock);
 252 }
 253
 254 static void
 255 vls_mt_locks_init (void)
 256 {
 257   pthread_mutex_init (&vlsl->vls_mt_mq_mlock, NULL);
 258   pthread_mutex_init (&vlsl->vls_mt_spool_mlock, NULL);
 259 }
 260
 261 u8
 262 vls_is_shared (vcl_locked_session_t * vls)
 263 {
 264   return (vls->shared_data_index != ~0);
 265 }
 266
 267 static inline void
 268 vls_lock (vcl_locked_session_t * vls)
 269 {
 270   if ((vlsl->vls_mt_n_threads > 1) || vls_is_shared (vls))
 271     clib_spinlock_lock (&vls->lock);
 272 }
 273
 274 static inline void
 275 vls_unlock (vcl_locked_session_t * vls)
 276 {
 277   if ((vlsl->vls_mt_n_threads > 1) || vls_is_shared (vls))
 278     clib_spinlock_unlock (&vls->lock);
 279 }
 280
 281 static inline vcl_session_handle_t
 282 vls_to_sh (vcl_locked_session_t * vls)
 283 {
 284   return vcl_session_handle_from_index (vls->session_index);
 285 }
 286
 287 static inline vcl_session_handle_t
 288 vls_to_sh_tu (vcl_locked_session_t * vls)
 289 {
 290   vcl_session_handle_t sh;
 291   sh = vls_to_sh (vls);
 292   vls_mt_table_runlock ();
 293   return sh;
 294 }
 295
 296 static vls_worker_t *
 297 vls_worker_get_current (void)
 298 {
 299   return pool_elt_at_index (vlsm->workers, vls_get_worker_index ());
 300 }
 301
 302 static void
 303 vls_worker_alloc (void)
 304 {
 305   vls_worker_t *wrk;
 306
 307   pool_get_zero (vlsm->workers, wrk);
 308   wrk->wrk_index = vcl_get_worker_index ();
 309 }
 310
 311 static void
 312 vls_worker_free (vls_worker_t * wrk)
 313 {
 314   hash_free (wrk->session_handle_to_vlsh_table);
 315   pool_free (wrk->vls_pool);
 316   pool_put (vlsm->workers, wrk);
 317 }
 318
 319 static vls_worker_t *
 320 vls_worker_get (u32 wrk_index)
 321 {
 322   if (pool_is_free_index (vlsm->workers, wrk_index))
 323     return 0;
 324   return pool_elt_at_index (vlsm->workers, wrk_index);
 325 }
 326
 327 static vls_handle_t
 328 vls_alloc (vcl_session_handle_t sh)
 329 {
 330   vls_worker_t *wrk = vls_worker_get_current ();
 331   vcl_locked_session_t *vls;
 332
 333   vls_mt_table_wlock ();
 334
 335   pool_get_zero (wrk->vls_pool, vls);
 336   vls->session_index = vppcom_session_index (sh);
 337   vls->worker_index = vppcom_session_worker (sh);
 338   vls->vls_index = vls - wrk->vls_pool;
 339   vls->shared_data_index = ~0;
 340   hash_set (wrk->session_handle_to_vlsh_table, sh, vls->vls_index);
 341   if (vls_mt_wrk_supported ())
 342     {
 343       hash_set (vls->vcl_wrk_index_to_session_index, vls->worker_index,
 344                 vls->session_index);
 345       vls->owner_vcl_wrk_index = vls->worker_index;
 346     }
 347   clib_spinlock_init (&vls->lock);
 348
 349   vls_mt_table_wunlock ();
 350   return vls->vls_index;
 351 }
 352
 353 static vcl_locked_session_t *
 354 vls_get (vls_handle_t vlsh)
 355 {
 356   vls_worker_t *wrk = vls_worker_get_current ();
 357   if (pool_is_free_index (wrk->vls_pool, vlsh))
 358     return 0;
 359   return pool_elt_at_index (wrk->vls_pool, vlsh);
 360 }
 361
 362 static void
 363 vls_free (vcl_locked_session_t * vls)
 364 {
 365   vls_worker_t *wrk = vls_worker_get_current ();
 366
 367   ASSERT (vls != 0);
 368   hash_unset (wrk->session_handle_to_vlsh_table,
 369               vcl_session_handle_from_index (vls->session_index));
 370   clib_spinlock_free (&vls->lock);
 371   pool_put (wrk->vls_pool, vls);
 372 }
 373
 374 static vcl_locked_session_t *
 375 vls_get_and_lock (vls_handle_t vlsh)
 376 {
 377   vls_worker_t *wrk = vls_worker_get_current ();
 378   vcl_locked_session_t *vls;
 379   if (pool_is_free_index (wrk->vls_pool, vlsh))
 380     return 0;
 381   vls = pool_elt_at_index (wrk->vls_pool, vlsh);
 382   vls_lock (vls);
 383   return vls;
 384 }
 385
 386 static vcl_locked_session_t *
 387 vls_get_w_dlock (vls_handle_t vlsh)
 388 {
 389   vcl_locked_session_t *vls;
 390   vls_mt_table_rlock ();
 391   vls = vls_get_and_lock (vlsh);
 392   if (!vls)
 393     vls_mt_table_runlock ();
 394   return vls;
 395 }
 396
 397 static inline void
 398 vls_get_and_unlock (vls_handle_t vlsh)
 399 {
 400   vcl_locked_session_t *vls;
 401   vls_mt_table_rlock ();
 402   vls = vls_get (vlsh);
 403   vls_unlock (vls);
 404   vls_mt_table_runlock ();
 405 }
 406
 407 static inline void
 408 vls_dunlock (vcl_locked_session_t * vls)
 409 {
 410   vls_unlock (vls);
 411   vls_mt_table_runlock ();
 412 }
 413
 414 static vcl_locked_session_t *
 415 vls_session_get (vls_worker_t * wrk, u32 vls_index)
 416 {
 417   if (pool_is_free_index (wrk->vls_pool, vls_index))
 418     return 0;
 419   return pool_elt_at_index (wrk->vls_pool, vls_index);
 420 }
 421
 422 vcl_session_handle_t
 423 vlsh_to_sh (vls_handle_t vlsh)
 424 {
 425   vcl_locked_session_t *vls;
 426   int rv;
 427
 428   vls = vls_get_w_dlock (vlsh);
 429   if (!vls)
 430     return INVALID_SESSION_ID;
 431   rv = vls_to_sh (vls);
 432   vls_dunlock (vls);
 433   return rv;
 434 }
 435
 436 vcl_session_handle_t
 437 vlsh_to_session_index (vls_handle_t vlsh)
 438 {
 439   vcl_session_handle_t sh;
 440   sh = vlsh_to_sh (vlsh);
 441   return vppcom_session_index (sh);
 442 }
 443
 444 vls_handle_t
 445 vls_si_wi_to_vlsh (u32 session_index, u32 vcl_wrk_index)
 446 {
 447   vls_worker_t *wrk = vls_worker_get_current ();
 448   uword *vlshp;
 449   vlshp = hash_get (wrk->session_handle_to_vlsh_table,
 450                     vcl_session_handle_from_wrk_session_index (session_index,
 451                                                                vcl_wrk_index));
 452   return vlshp ? *vlshp : VLS_INVALID_HANDLE;
 453 }
 454
 455 vls_handle_t
 456 vls_session_index_to_vlsh (uint32_t session_index)
 457 {
 458   vls_handle_t vlsh;
 459
 460   vls_mt_table_rlock ();
 461   vlsh = vls_si_wi_to_vlsh (session_index, vcl_get_worker_index ());
 462   vls_mt_table_runlock ();
 463
 464   return vlsh;
 465 }
 466
 467 u8
 468 vls_is_shared_by_wrk (vcl_locked_session_t * vls, u32 wrk_index)
 469 {
 470   vls_shared_data_t *vls_shd;
 471   int i;
 472
 473   if (vls->shared_data_index == ~0)
 474     return 0;
 475
 476   vls_shared_data_pool_rlock ();
 477
 478   vls_shd = vls_shared_data_get (vls->shared_data_index);
 479   clib_spinlock_lock (&vls_shd->lock);
 480
 481   for (i = 0; i < vec_len (vls_shd->workers_subscribed); i++)
 482     if (vls_shd->workers_subscribed[i] == wrk_index)
 483       {
 484         clib_spinlock_unlock (&vls_shd->lock);
 485         vls_shared_data_pool_runlock ();
 486         return 1;
 487       }
 488   clib_spinlock_unlock (&vls_shd->lock);
 489
 490   vls_shared_data_pool_runlock ();
 491   return 0;
 492 }
 493
 494 static void
 495 vls_listener_wrk_set (vcl_locked_session_t * vls, u32 wrk_index, u8 is_active)
 496 {
 497   vls_shared_data_t *vls_shd;
 498
 499   if (vls->shared_data_index == ~0)
 500     {
 501       clib_warning ("not a shared session");
 502       return;
 503     }
 504
 505   vls_shared_data_pool_rlock ();
 506
 507   vls_shd = vls_shared_data_get (vls->shared_data_index);
 508
 509   clib_spinlock_lock (&vls_shd->lock);
 510   clib_bitmap_set (vls_shd->listeners, wrk_index, is_active);
 511   clib_spinlock_unlock (&vls_shd->lock);
 512
 513   vls_shared_data_pool_runlock ();
 514 }
 515
 516 static u32
 517 vls_shared_get_owner (vcl_locked_session_t * vls)
 518 {
 519   vls_shared_data_t *vls_shd;
 520   u32 owner_wrk;
 521
 522   vls_shared_data_pool_rlock ();
 523
 524   vls_shd = vls_shared_data_get (vls->shared_data_index);
 525   owner_wrk = vls_shd->owner_wrk_index;
 526
 527   vls_shared_data_pool_runlock ();
 528
 529   return owner_wrk;
 530 }
 531
 532 static u8
 533 vls_listener_wrk_is_active (vcl_locked_session_t * vls, u32 wrk_index)
 534 {
 535   vls_shared_data_t *vls_shd;
 536   u8 is_set;
 537
 538   if (vls->shared_data_index == ~0)
 539     {
 540       clib_warning ("not a shared session");
 541       return 0;
 542     }
 543
 544   vls_shared_data_pool_rlock ();
 545
 546   vls_shd = vls_shared_data_get (vls->shared_data_index);
 547
 548   clib_spinlock_lock (&vls_shd->lock);
 549   is_set = clib_bitmap_get (vls_shd->listeners, wrk_index);
 550   clib_spinlock_unlock (&vls_shd->lock);
 551
 552   vls_shared_data_pool_runlock ();
 553
 554   return (is_set == 1);
 555 }
 556
 557 static void
 558 vls_listener_wrk_start_listen (vcl_locked_session_t * vls, u32 wrk_index)
 559 {
 560   vppcom_session_listen (vls_to_sh (vls), ~0);
 561   vls_listener_wrk_set (vls, wrk_index, 1 /* is_active */ );
 562 }
 563
 564 static void
 565 vls_listener_wrk_stop_listen (vcl_locked_session_t * vls, u32 wrk_index)
 566 {
 567   vcl_worker_t *wrk;
 568   vcl_session_t *s;
 569
 570   wrk = vcl_worker_get (wrk_index);
 571   s = vcl_session_get (wrk, vls->session_index);
 572   if (s->session_state != VCL_STATE_LISTEN)
 573     return;
 574   vcl_send_session_unlisten (wrk, s);
 575   s->session_state = VCL_STATE_LISTEN_NO_MQ;
 576   vls_listener_wrk_set (vls, wrk_index, 0 /* is_active */ );
 577 }
 578
 579 static int
 580 vls_shared_data_subscriber_position (vls_shared_data_t * vls_shd,
 581                                      u32 wrk_index)
 582 {
 583   int i;
 584
 585   for (i = 0; i < vec_len (vls_shd->workers_subscribed); i++)
 586     {
 587       if (vls_shd->workers_subscribed[i] == wrk_index)
 588         return i;
 589     }
 590   return -1;
 591 }
 592
 593 int
 594 vls_unshare_session (vcl_locked_session_t * vls, vcl_worker_t * wrk)
 595 {
 596   vls_shared_data_t *vls_shd;
 597   int do_disconnect, pos;
 598   u32 n_subscribers;
 599   vcl_session_t *s;
 600
 601   if (vls->shared_data_index == ~0)
 602     return 0;
 603
 604   s = vcl_session_get (wrk, vls->session_index);
 605   if (s->session_state == VCL_STATE_LISTEN)
 606     vls_listener_wrk_set (vls, wrk->wrk_index, 0 /* is_active */ );
 607
 608   vls_shared_data_pool_rlock ();
 609
 610   vls_shd = vls_shared_data_get (vls->shared_data_index);
 611   clib_spinlock_lock (&vls_shd->lock);
 612
 613   pos = vls_shared_data_subscriber_position (vls_shd, wrk->wrk_index);
 614   if (pos < 0)
 615     {
 616       clib_warning ("worker %u not subscribed for vls %u", wrk->wrk_index,
 617                     vls->worker_index);
 618       goto done;
 619     }
 620
 621   /*
 622    * Unsubscribe from share data and fifos
 623    */
 624   if (s->rx_fifo)
 625     {
 626       svm_fifo_del_subscriber (s->rx_fifo, wrk->vpp_wrk_index);
 627       svm_fifo_del_subscriber (s->tx_fifo, wrk->vpp_wrk_index);
 628     }
 629   vec_del1 (vls_shd->workers_subscribed, pos);
 630
 631   /*
 632    * Cleanup vcl state
 633    */
 634   n_subscribers = vec_len (vls_shd->workers_subscribed);
 635   do_disconnect = s->session_state == VCL_STATE_LISTEN || !n_subscribers;
 636   vcl_session_cleanup (wrk, s, vcl_session_handle (s), do_disconnect);
 637
 638   /*
 639    * No subscriber left, cleanup shared data
 640    */
 641   if (!n_subscribers)
 642     {
 643       u32 shd_index = vls_shared_data_index (vls_shd);
 644
 645       clib_spinlock_unlock (&vls_shd->lock);
 646       vls_shared_data_pool_runlock ();
 647
 648       vls_shared_data_free (shd_index);
 649
 650       /* All locks have been dropped */
 651       return 0;
 652     }
 653
 654   /* Return, if this is not the owning worker */
 655   if (vls_shd->owner_wrk_index != wrk->wrk_index)
 656     goto done;
 657
 658   ASSERT (vec_len (vls_shd->workers_subscribed));
 659
 660   /*
 661    *  Check if we can change owner or close
 662    */
 663   vls_shd->owner_wrk_index = vls_shd->workers_subscribed[0];
 664   vcl_send_session_worker_update (wrk, s, vls_shd->owner_wrk_index);
 665
 666   /* XXX is this still needed? */
 667   if (vec_len (vls_shd->workers_subscribed) > 1)
 668     clib_warning ("more workers need to be updated");
 669
 670 done:
 671
 672   clib_spinlock_unlock (&vls_shd->lock);
 673   vls_shared_data_pool_runlock ();
 674
 675   return 0;
 676 }
 677
 678 void
 679 vls_init_share_session (vls_worker_t * vls_wrk, vcl_locked_session_t * vls)
 680 {
 681   vls_shared_data_t *vls_shd;
 682
 683   u32 vls_shd_index = vls_shared_data_alloc ();
 684
 685   vls_shared_data_pool_rlock ();
 686
 687   vls_shd = vls_shared_data_get (vls_shd_index);
 688   vls_shd->owner_wrk_index = vls_wrk->wrk_index;
 689   vls->shared_data_index = vls_shd_index;
 690   vec_add1 (vls_shd->workers_subscribed, vls_wrk->wrk_index);
 691
 692   vls_shared_data_pool_runlock ();
 693 }
 694
 695 void
 696 vls_share_session (vls_worker_t * vls_wrk, vcl_locked_session_t * vls)
 697 {
 698   vcl_worker_t *vcl_wrk = vcl_worker_get (vls_wrk->wrk_index);
 699   vls_shared_data_t *vls_shd;
 700   vcl_session_t *s;
 701
 702   s = vcl_session_get (vcl_wrk, vls->session_index);
 703   if (!s)
 704     {
 705       clib_warning ("wrk %u session %u vls %u NOT AVAILABLE",
 706                     vcl_wrk->wrk_index, vls->session_index, vls->vls_index);
 707       return;
 708     }
 709
 710   ASSERT (vls->shared_data_index != ~0);
 711
 712   /* Reinit session lock */
 713   clib_spinlock_init (&vls->lock);
 714
 715   vls_shared_data_pool_rlock ();
 716
 717   vls_shd = vls_shared_data_get (vls->shared_data_index);
 718
 719   clib_spinlock_lock (&vls_shd->lock);
 720   vec_add1 (vls_shd->workers_subscribed, vls_wrk->wrk_index);
 721   clib_spinlock_unlock (&vls_shd->lock);
 722
 723   vls_shared_data_pool_runlock ();
 724
 725   if (s->rx_fifo)
 726     {
 727       svm_fifo_add_subscriber (s->rx_fifo, vcl_wrk->vpp_wrk_index);
 728       svm_fifo_add_subscriber (s->tx_fifo, vcl_wrk->vpp_wrk_index);
 729     }
 730   else if (s->session_state == VCL_STATE_LISTEN)
 731     {
 732       s->session_state = VCL_STATE_LISTEN_NO_MQ;
 733     }
 734 }
 735
 736 static void
 737 vls_share_sessions (vls_worker_t * vls_parent_wrk, vls_worker_t * vls_wrk)
 738 {
 739   vcl_locked_session_t *vls, *parent_vls;
 740
 741   /* *INDENT-OFF* */
 742   pool_foreach (vls, vls_wrk->vls_pool)  {
 743     /* Initialize sharing on parent session */
 744     if (vls->shared_data_index == ~0)
 745       {
 746         parent_vls = vls_session_get (vls_parent_wrk, vls->vls_index);
 747         vls_init_share_session (vls_parent_wrk, parent_vls);
 748         vls->shared_data_index = parent_vls->shared_data_index;
 749       }
 750     vls_share_session (vls_wrk, vls);
 751   }
 752   /* *INDENT-ON* */
 753 }
 754
 755 void
 756 vls_worker_copy_on_fork (vcl_worker_t * parent_wrk)
 757 {
 758   vls_worker_t *vls_wrk = vls_worker_get_current (), *vls_parent_wrk;
 759   vcl_worker_t *wrk = vcl_worker_get_current ();
 760   u32 vls_index, session_index, wrk_index;
 761   vcl_session_handle_t sh;
 762
 763   /*
 764    * init vcl worker
 765    */
 766   wrk->sessions = pool_dup (parent_wrk->sessions);
 767   wrk->session_index_by_vpp_handles =
 768     hash_dup (parent_wrk->session_index_by_vpp_handles);
 769
 770   /*
 771    * init vls worker
 772    */
 773   vls_parent_wrk = vls_worker_get (parent_wrk->wrk_index);
 774   /* *INDENT-OFF* */
 775   hash_foreach (sh, vls_index, vls_parent_wrk->session_handle_to_vlsh_table,
 776     ({
 777       vcl_session_handle_parse (sh, &wrk_index, &session_index);
 778       hash_set (vls_wrk->session_handle_to_vlsh_table,
 779                 vcl_session_handle_from_index (session_index), vls_index);
 780     }));
 781   /* *INDENT-ON* */
 782   vls_wrk->vls_pool = pool_dup (vls_parent_wrk->vls_pool);
 783
 784   vls_share_sessions (vls_parent_wrk, vls_wrk);
 785 }
 786
 787 static void
 788 vls_mt_acq_locks (vcl_locked_session_t * vls, vls_mt_ops_t op, int *locks_acq)
 789 {
 790   vcl_worker_t *wrk = vcl_worker_get_current ();
 791   vcl_session_t *s = 0;
 792   int is_nonblk = 0;
 793
 794   if (vls)
 795     {
 796       s = vcl_session_get (wrk, vls->session_index);
 797       if (PREDICT_FALSE (!s))
 798         return;
 799       is_nonblk = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
 800     }
 801
 802   switch (op)
 803     {
 804     case VLS_MT_OP_READ:
 805       if (!is_nonblk)
 806         is_nonblk = vcl_session_read_ready (s) != 0;
 807       if (!is_nonblk)
 808         {
 809           vls_mt_mq_lock ();
 810           *locks_acq |= VLS_MT_LOCK_MQ;
 811         }
 812       break;
 813     case VLS_MT_OP_WRITE:
 814       ASSERT (s);
 815       if (!is_nonblk)
 816         is_nonblk = vcl_session_write_ready (s) != 0;
 817       if (!is_nonblk)
 818         {
 819           vls_mt_mq_lock ();
 820           *locks_acq |= VLS_MT_LOCK_MQ;
 821         }
 822       break;
 823     case VLS_MT_OP_XPOLL:
 824       vls_mt_mq_lock ();
 825       *locks_acq |= VLS_MT_LOCK_MQ;
 826       break;
 827     case VLS_MT_OP_SPOOL:
 828       vls_mt_spool_lock ();
 829       *locks_acq |= VLS_MT_LOCK_SPOOL;
 830       break;
 831     default:
 832       break;
 833     }
 834 }
 835
 836 static void
 837 vls_mt_rel_locks (int locks_acq)
 838 {
 839   if (locks_acq & VLS_MT_LOCK_MQ)
 840     vls_mt_mq_unlock ();
 841   if (locks_acq & VLS_MT_LOCK_SPOOL)
 842     vls_mt_create_unlock ();
 843 }
 844
 845 static inline u8
 846 vls_mt_session_should_migrate (vcl_locked_session_t * vls)
 847 {
 848   return (vls_mt_wrk_supported ()
 849           && vls->worker_index != vcl_get_worker_index ());
 850 }
 851
 852 static void
 853 vls_mt_session_migrate (vcl_locked_session_t * vls)
 854 {
 855   u32 wrk_index = vcl_get_worker_index ();
 856   vcl_worker_t *wrk;
 857   u32 src_sid, sid;
 858   vcl_session_t *session;
 859   uword *p;
 860
 861   ASSERT (vls_mt_wrk_supported () && vls->worker_index != wrk_index);
 862
 863   /*
 864    * VCL session on current vcl worker already allocated. Update current
 865    * owner worker and index and return
 866    */
 867   if ((p = hash_get (vls->vcl_wrk_index_to_session_index, wrk_index)))
 868     {
 869       vls->worker_index = wrk_index;
 870       vls->session_index = (u32) p[0];
 871       return;
 872     }
 873
 874   /*
 875    * Ask vcl worker that owns the original vcl session to clone it into
 876    * current vcl worker session pool
 877    */
 878
 879   if (!(p = hash_get (vls->vcl_wrk_index_to_session_index,
 880                       vls->owner_vcl_wrk_index)))
 881     {
 882       VERR ("session in owner worker(%u) is free", vls->owner_vcl_wrk_index);
 883       ASSERT (0);
 884       return;
 885     }
 886
 887   src_sid = (u32) p[0];
 888   wrk = vcl_worker_get_current ();
 889   session = vcl_session_alloc (wrk);
 890   sid = session->session_index;
 891   vls_send_clone_and_share_rpc (wrk, vls, sid, vls_get_worker_index (),
 892                                 vls->owner_vcl_wrk_index, vls->vls_index,
 893                                 src_sid);
 894   session->session_index = sid;
 895   vls->worker_index = wrk_index;
 896   vls->session_index = sid;
 897   hash_set (vls->vcl_wrk_index_to_session_index, wrk_index, sid);
 898   VDBG (1, "migrate session of worker (session): %u (%u) -> %u (%u)",
 899         vls->owner_vcl_wrk_index, src_sid, wrk_index, sid);
 900
 901   if (PREDICT_FALSE ((session->flags & VCL_SESSION_F_IS_VEP)
 902                      && session->vep.next_sh != ~0))
 903     {
 904       /* TODO: rollback? */
 905       VERR ("can't migrate nonempty epoll session");
 906       ASSERT (0);
 907       return;
 908     }
 909   else if (PREDICT_FALSE (!(session->flags & VCL_SESSION_F_IS_VEP) &&
 910                           session->session_state != VCL_STATE_CLOSED))
 911     {
 912       /* TODO: rollback? */
 913       VERR ("migrate NOT supported, session_status (%u)",
 914             session->session_state);
 915       ASSERT (0);
 916       return;
 917     }
 918 }
 919
 920 static inline void
 921 vls_mt_detect (void)
 922 {
 923   if (PREDICT_FALSE (vcl_get_worker_index () == ~0))
 924     vls_mt_add ();
 925 }
 926
 927 #define vls_mt_guard(_vls, _op)                                         \
 928   int _locks_acq = 0;                                                   \
 929   if (vls_mt_wrk_supported ())                                          \
 930     {                                                                   \
 931       if (PREDICT_FALSE (_vls                                           \
 932             && ((vcl_locked_session_t *)_vls)->worker_index !=          \
 933                 vcl_get_worker_index ()))                               \
 934           vls_mt_session_migrate (_vls);                                \
 935     }                                                                   \
 936   else                                                                  \
 937     {                                                                   \
 938       if (PREDICT_FALSE (vlsl->vls_mt_n_threads > 1))                   \
 939         vls_mt_acq_locks (_vls, _op, &_locks_acq);                      \
 940     }                                                                   \
 941
 942 #define vls_mt_unguard()                                                \
 943   if (PREDICT_FALSE (_locks_acq))                                       \
 944     vls_mt_rel_locks (_locks_acq)
 945
 946 int
 947 vls_write (vls_handle_t vlsh, void *buf, size_t nbytes)
 948 {
 949   vcl_locked_session_t *vls;
 950   int rv;
 951
 952   vls_mt_detect ();
 953   if (!(vls = vls_get_w_dlock (vlsh)))
 954     return VPPCOM_EBADFD;
 955
 956   vls_mt_guard (vls, VLS_MT_OP_WRITE);
 957   rv = vppcom_session_write (vls_to_sh_tu (vls), buf, nbytes);
 958   vls_mt_unguard ();
 959   vls_get_and_unlock (vlsh);
 960   return rv;
 961 }
 962
 963 int
 964 vls_write_msg (vls_handle_t vlsh, void *buf, size_t nbytes)
 965 {
 966   vcl_locked_session_t *vls;
 967   int rv;
 968
 969   vls_mt_detect ();
 970   if (!(vls = vls_get_w_dlock (vlsh)))
 971     return VPPCOM_EBADFD;
 972   vls_mt_guard (vls, VLS_MT_OP_WRITE);
 973   rv = vppcom_session_write_msg (vls_to_sh_tu (vls), buf, nbytes);
 974   vls_mt_unguard ();
 975   vls_get_and_unlock (vlsh);
 976   return rv;
 977 }
 978
 979 int
 980 vls_sendto (vls_handle_t vlsh, void *buf, int buflen, int flags,
 981             vppcom_endpt_t * ep)
 982 {
 983   vcl_locked_session_t *vls;
 984   int rv;
 985
 986   vls_mt_detect ();
 987   if (!(vls = vls_get_w_dlock (vlsh)))
 988     return VPPCOM_EBADFD;
 989   vls_mt_guard (vls, VLS_MT_OP_WRITE);
 990   rv = vppcom_session_sendto (vls_to_sh_tu (vls), buf, buflen, flags, ep);
 991   vls_mt_unguard ();
 992   vls_get_and_unlock (vlsh);
 993   return rv;
 994 }
 995
 996 ssize_t
 997 vls_read (vls_handle_t vlsh, void *buf, size_t nbytes)
 998 {
 999   vcl_locked_session_t *vls;
1000   int rv;
1001
1002   vls_mt_detect ();
1003   if (!(vls = vls_get_w_dlock (vlsh)))
1004     return VPPCOM_EBADFD;
1005   vls_mt_guard (vls, VLS_MT_OP_READ);
1006   rv = vppcom_session_read (vls_to_sh_tu (vls), buf, nbytes);
1007   vls_mt_unguard ();
1008   vls_get_and_unlock (vlsh);
1009   return rv;
1010 }
1011
1012 ssize_t
1013 vls_recvfrom (vls_handle_t vlsh, void *buffer, uint32_t buflen, int flags,
1014               vppcom_endpt_t * ep)
1015 {
1016   vcl_locked_session_t *vls;
1017   int rv;
1018
1019   vls_mt_detect ();
1020   if (!(vls = vls_get_w_dlock (vlsh)))
1021     return VPPCOM_EBADFD;
1022   vls_mt_guard (vls, VLS_MT_OP_READ);
1023   rv = vppcom_session_recvfrom (vls_to_sh_tu (vls), buffer, buflen, flags,
1024                                 ep);
1025   vls_mt_unguard ();
1026   vls_get_and_unlock (vlsh);
1027   return rv;
1028 }
1029
1030 int
1031 vls_attr (vls_handle_t vlsh, uint32_t op, void *buffer, uint32_t * buflen)
1032 {
1033   vcl_locked_session_t *vls;
1034   int rv;
1035
1036   vls_mt_detect ();
1037   if (!(vls = vls_get_w_dlock (vlsh)))
1038     return VPPCOM_EBADFD;
1039   if (vls_mt_session_should_migrate (vls))
1040     vls_mt_session_migrate (vls);
1041   rv = vppcom_session_attr (vls_to_sh_tu (vls), op, buffer, buflen);
1042   vls_get_and_unlock (vlsh);
1043   return rv;
1044 }
1045
1046 int
1047 vls_bind (vls_handle_t vlsh, vppcom_endpt_t * ep)
1048 {
1049   vcl_locked_session_t *vls;
1050   int rv;
1051
1052   vls_mt_detect ();
1053   if (!(vls = vls_get_w_dlock (vlsh)))
1054     return VPPCOM_EBADFD;
1055   rv = vppcom_session_bind (vls_to_sh_tu (vls), ep);
1056   vls_get_and_unlock (vlsh);
1057   return rv;
1058 }
1059
1060 int
1061 vls_listen (vls_handle_t vlsh, int q_len)
1062 {
1063   vcl_locked_session_t *vls;
1064   int rv;
1065
1066   vls_mt_detect ();
1067   if (!(vls = vls_get_w_dlock (vlsh)))
1068     return VPPCOM_EBADFD;
1069   vls_mt_guard (vls, VLS_MT_OP_XPOLL);
1070   rv = vppcom_session_listen (vls_to_sh_tu (vls), q_len);
1071   vls_mt_unguard ();
1072   vls_get_and_unlock (vlsh);
1073   return rv;
1074 }
1075
1076 int
1077 vls_connect (vls_handle_t vlsh, vppcom_endpt_t * server_ep)
1078 {
1079   vcl_locked_session_t *vls;
1080   int rv;
1081
1082   vls_mt_detect ();
1083   if (!(vls = vls_get_w_dlock (vlsh)))
1084     return VPPCOM_EBADFD;
1085   vls_mt_guard (vls, VLS_MT_OP_XPOLL);
1086   rv = vppcom_session_connect (vls_to_sh_tu (vls), server_ep);
1087   vls_mt_unguard ();
1088   vls_get_and_unlock (vlsh);
1089   return rv;
1090 }
1091
1092 static inline void
1093 vls_mp_checks (vcl_locked_session_t * vls, int is_add)
1094 {
1095   vcl_worker_t *wrk = vcl_worker_get_current ();
1096   vcl_session_t *s;
1097   u32 owner_wrk;
1098
1099   if (vls_mt_wrk_supported ())
1100     return;
1101
1102   s = vcl_session_get (wrk, vls->session_index);
1103   switch (s->session_state)
1104     {
1105     case VCL_STATE_LISTEN:
1106       if (is_add)
1107         {
1108           vls_listener_wrk_set (vls, vls->worker_index, 1 /* is_active */ );
1109           break;
1110         }
1111       vls_listener_wrk_stop_listen (vls, vls->worker_index);
1112       break;
1113     case VCL_STATE_LISTEN_NO_MQ:
1114       if (!is_add)
1115         break;
1116
1117       /* Register worker as listener */
1118       vls_listener_wrk_start_listen (vls, wrk->wrk_index);
1119
1120       /* If owner worker did not attempt to accept/xpoll on the session,
1121        * force a listen stop for it, since it may not be interested in
1122        * accepting new sessions.
1123        * This is pretty much a hack done to give app workers the illusion
1124        * that it is fine to listen and not accept new sessions for a
1125        * given listener. Without it, we would accumulate unhandled
1126        * accepts on the passive worker message queue. */
1127       owner_wrk = vls_shared_get_owner (vls);
1128       if (!vls_listener_wrk_is_active (vls, owner_wrk))
1129         vls_listener_wrk_stop_listen (vls, owner_wrk);
1130       break;
1131     default:
1132       break;
1133     }
1134 }
1135
1136 vls_handle_t
1137 vls_accept (vls_handle_t listener_vlsh, vppcom_endpt_t * ep, int flags)
1138 {
1139   vls_handle_t accepted_vlsh;
1140   vcl_locked_session_t *vls;
1141   int sh;
1142
1143   vls_mt_detect ();
1144   if (!(vls = vls_get_w_dlock (listener_vlsh)))
1145     return VPPCOM_EBADFD;
1146   if (vcl_n_workers () > 1)
1147     vls_mp_checks (vls, 1 /* is_add */ );
1148   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1149   sh = vppcom_session_accept (vls_to_sh_tu (vls), ep, flags);
1150   vls_mt_unguard ();
1151   vls_get_and_unlock (listener_vlsh);
1152   if (sh < 0)
1153     return sh;
1154   accepted_vlsh = vls_alloc (sh);
1155   if (PREDICT_FALSE (accepted_vlsh == VLS_INVALID_HANDLE))
1156     vppcom_session_close (sh);
1157   return accepted_vlsh;
1158 }
1159
1160 vls_handle_t
1161 vls_create (uint8_t proto, uint8_t is_nonblocking)
1162 {
1163   vcl_session_handle_t sh;
1164   vls_handle_t vlsh;
1165
1166   vls_mt_detect ();
1167   vls_mt_guard (0, VLS_MT_OP_SPOOL);
1168   sh = vppcom_session_create (proto, is_nonblocking);
1169   vls_mt_unguard ();
1170   if (sh == INVALID_SESSION_ID)
1171     return VLS_INVALID_HANDLE;
1172
1173   vlsh = vls_alloc (sh);
1174   if (PREDICT_FALSE (vlsh == VLS_INVALID_HANDLE))
1175     vppcom_session_close (sh);
1176
1177   return vlsh;
1178 }
1179
1180 static void
1181 vls_mt_session_cleanup (vcl_locked_session_t * vls)
1182 {
1183   u32 session_index, wrk_index, current_vcl_wrk;
1184   vcl_worker_t *wrk = vcl_worker_get_current ();
1185
1186   ASSERT (vls_mt_wrk_supported ());
1187
1188   current_vcl_wrk = vcl_get_worker_index ();
1189
1190   /* *INDENT-OFF* */
1191   hash_foreach (wrk_index, session_index, vls->vcl_wrk_index_to_session_index,
1192     ({
1193       if (current_vcl_wrk != wrk_index)
1194         vls_send_session_cleanup_rpc (wrk, wrk_index, session_index);
1195     }));
1196   /* *INDENT-ON* */
1197   hash_free (vls->vcl_wrk_index_to_session_index);
1198 }
1199
1200 int
1201 vls_close (vls_handle_t vlsh)
1202 {
1203   vcl_locked_session_t *vls;
1204   int rv;
1205
1206   vls_mt_detect ();
1207   vls_mt_table_wlock ();
1208
1209   vls = vls_get_and_lock (vlsh);
1210   if (!vls)
1211     {
1212       vls_mt_table_wunlock ();
1213       return VPPCOM_EBADFD;
1214     }
1215
1216   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1217
1218   if (vls_is_shared (vls))
1219     rv = vls_unshare_session (vls, vcl_worker_get_current ());
1220   else
1221     rv = vppcom_session_close (vls_to_sh (vls));
1222
1223   if (vls_mt_wrk_supported ())
1224     vls_mt_session_cleanup (vls);
1225
1226   vls_free (vls);
1227   vls_mt_unguard ();
1228
1229   vls_mt_table_wunlock ();
1230
1231   return rv;
1232 }
1233
1234 vls_handle_t
1235 vls_epoll_create (void)
1236 {
1237   vcl_session_handle_t sh;
1238   vls_handle_t vlsh;
1239
1240   vls_mt_detect ();
1241
1242   sh = vppcom_epoll_create ();
1243   if (sh == INVALID_SESSION_ID)
1244     return VLS_INVALID_HANDLE;
1245
1246   vlsh = vls_alloc (sh);
1247   if (vlsh == VLS_INVALID_HANDLE)
1248     vppcom_session_close (sh);
1249
1250   return vlsh;
1251 }
1252
1253 static void
1254 vls_epoll_ctl_mp_checks (vcl_locked_session_t * vls, int op)
1255 {
1256   if (vcl_n_workers () <= 1)
1257     {
1258       vlsl->epoll_mp_check = 1;
1259       return;
1260     }
1261
1262   if (op == EPOLL_CTL_MOD)
1263     return;
1264
1265   vlsl->epoll_mp_check = 1;
1266   vls_mp_checks (vls, op == EPOLL_CTL_ADD);
1267 }
1268
1269 int
1270 vls_epoll_ctl (vls_handle_t ep_vlsh, int op, vls_handle_t vlsh,
1271                struct epoll_event *event)
1272 {
1273   vcl_locked_session_t *ep_vls, *vls;
1274   vcl_session_handle_t ep_sh, sh;
1275   int rv;
1276
1277   vls_mt_detect ();
1278   vls_mt_table_rlock ();
1279   ep_vls = vls_get_and_lock (ep_vlsh);
1280   vls = vls_get_and_lock (vlsh);
1281
1282   if (vls_mt_session_should_migrate (ep_vls))
1283     vls_mt_session_migrate (ep_vls);
1284
1285   ep_sh = vls_to_sh (ep_vls);
1286   sh = vls_to_sh (vls);
1287
1288   if (PREDICT_FALSE (!vlsl->epoll_mp_check))
1289     vls_epoll_ctl_mp_checks (vls, op);
1290
1291   vls_mt_table_runlock ();
1292
1293   rv = vppcom_epoll_ctl (ep_sh, op, sh, event);
1294
1295   vls_mt_table_rlock ();
1296   ep_vls = vls_get (ep_vlsh);
1297   vls = vls_get (vlsh);
1298   vls_unlock (vls);
1299   vls_unlock (ep_vls);
1300   vls_mt_table_runlock ();
1301   return rv;
1302 }
1303
1304 int
1305 vls_epoll_wait (vls_handle_t ep_vlsh, struct epoll_event *events,
1306                 int maxevents, double wait_for_time)
1307 {
1308   vcl_locked_session_t *vls;
1309   int rv;
1310
1311   vls_mt_detect ();
1312   if (!(vls = vls_get_w_dlock (ep_vlsh)))
1313     return VPPCOM_EBADFD;
1314   vls_mt_guard (0, VLS_MT_OP_XPOLL);
1315   rv = vppcom_epoll_wait (vls_to_sh_tu (vls), events, maxevents,
1316                           wait_for_time);
1317   vls_mt_unguard ();
1318   vls_get_and_unlock (ep_vlsh);
1319   return rv;
1320 }
1321
1322 static void
1323 vls_select_mp_checks (vcl_si_set * read_map)
1324 {
1325   vcl_locked_session_t *vls;
1326   vcl_worker_t *wrk;
1327   vcl_session_t *s;
1328   u32 si;
1329
1330   if (vcl_n_workers () <= 1)
1331     {
1332       vlsl->select_mp_check = 1;
1333       return;
1334     }
1335
1336   if (!read_map)
1337     return;
1338
1339   vlsl->select_mp_check = 1;
1340   wrk = vcl_worker_get_current ();
1341
1342   /* *INDENT-OFF* */
1343   clib_bitmap_foreach (si, read_map, ({
1344     s = vcl_session_get (wrk, si);
1345     if (s->session_state == VCL_STATE_LISTEN)
1346       {
1347         vls = vls_get (vls_session_index_to_vlsh (si));
1348         vls_mp_checks (vls, 1 /* is_add */);
1349       }
1350   }));
1351   /* *INDENT-ON* */
1352 }
1353
1354 int
1355 vls_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
1356             vcl_si_set * except_map, double wait_for_time)
1357 {
1358   int rv;
1359
1360   vls_mt_detect ();
1361   vls_mt_guard (0, VLS_MT_OP_XPOLL);
1362   if (PREDICT_FALSE (!vlsl->select_mp_check))
1363     vls_select_mp_checks (read_map);
1364   rv = vppcom_select (n_bits, read_map, write_map, except_map, wait_for_time);
1365   vls_mt_unguard ();
1366   return rv;
1367 }
1368
1369 static void
1370 vls_unshare_vcl_worker_sessions (vcl_worker_t * wrk)
1371 {
1372   u32 current_wrk, is_current;
1373   vcl_locked_session_t *vls;
1374   vcl_session_t *s;
1375
1376   if (pool_elts (vcm->workers) <= 1)
1377     return;
1378
1379   current_wrk = vcl_get_worker_index ();
1380   is_current = current_wrk == wrk->wrk_index;
1381
1382   /* *INDENT-OFF* */
1383   pool_foreach (s, wrk->sessions)  {
1384     vls = vls_get (vls_si_wi_to_vlsh (s->session_index, wrk->wrk_index));
1385     if (vls && (is_current || vls_is_shared_by_wrk (vls, current_wrk)))
1386       vls_unshare_session (vls, wrk);
1387   }
1388   /* *INDENT-ON* */
1389 }
1390
1391 static void
1392 vls_cleanup_vcl_worker (vcl_worker_t * wrk)
1393 {
1394   vls_worker_t *vls_wrk = vls_worker_get (wrk->wrk_index);
1395
1396   /* Unshare sessions and also cleanup worker since child may have
1397    * called _exit () and therefore vcl may not catch the event */
1398   vls_unshare_vcl_worker_sessions (wrk);
1399   vcl_worker_cleanup (wrk, 1 /* notify vpp */ );
1400
1401   vls_worker_free (vls_wrk);
1402 }
1403
1404 static void
1405 vls_cleanup_forked_child (vcl_worker_t * wrk, vcl_worker_t * child_wrk)
1406 {
1407   vcl_worker_t *sub_child;
1408   int tries = 0;
1409
1410   if (child_wrk->forked_child != ~0)
1411     {
1412       sub_child = vcl_worker_get_if_valid (child_wrk->forked_child);
1413       if (sub_child)
1414         {
1415           /* Wait a bit, maybe the process is going away */
1416           while (kill (sub_child->current_pid, 0) >= 0 && tries++ < 50)
1417             usleep (1e3);
1418           if (kill (sub_child->current_pid, 0) < 0)
1419             vls_cleanup_forked_child (child_wrk, sub_child);
1420         }
1421     }
1422   vls_cleanup_vcl_worker (child_wrk);
1423   VDBG (0, "Cleaned up forked child wrk %u", child_wrk->wrk_index);
1424   wrk->forked_child = ~0;
1425 }
1426
1427 static struct sigaction old_sa;
1428
1429 static void
1430 vls_intercept_sigchld_handler (int signum, siginfo_t * si, void *uc)
1431 {
1432   vcl_worker_t *wrk, *child_wrk;
1433
1434   if (vcl_get_worker_index () == ~0)
1435     return;
1436
1437   if (sigaction (SIGCHLD, &old_sa, 0))
1438     {
1439       VERR ("couldn't restore sigchld");
1440       exit (-1);
1441     }
1442
1443   wrk = vcl_worker_get_current ();
1444   if (wrk->forked_child == ~0)
1445     return;
1446
1447   child_wrk = vcl_worker_get_if_valid (wrk->forked_child);
1448   if (!child_wrk)
1449     goto done;
1450
1451   if (si && si->si_pid != child_wrk->current_pid)
1452     {
1453       VDBG (0, "unexpected child pid %u", si->si_pid);
1454       goto done;
1455     }
1456   vls_cleanup_forked_child (wrk, child_wrk);
1457
1458 done:
1459   if (old_sa.sa_flags & SA_SIGINFO)
1460     {
1461       void (*fn) (int, siginfo_t *, void *) = old_sa.sa_sigaction;
1462       fn (signum, si, uc);
1463     }
1464   else
1465     {
1466       void (*fn) (int) = old_sa.sa_handler;
1467       if (fn)
1468         fn (signum);
1469     }
1470 }
1471
1472 static void
1473 vls_incercept_sigchld ()
1474 {
1475   struct sigaction sa;
1476   clib_memset (&sa, 0, sizeof (sa));
1477   sa.sa_sigaction = vls_intercept_sigchld_handler;
1478   sa.sa_flags = SA_SIGINFO;
1479   if (sigaction (SIGCHLD, &sa, &old_sa))
1480     {
1481       VERR ("couldn't intercept sigchld");
1482       exit (-1);
1483     }
1484 }
1485
1486 static void
1487 vls_app_pre_fork (void)
1488 {
1489   vls_incercept_sigchld ();
1490   vcl_flush_mq_events ();
1491 }
1492
1493 static void
1494 vls_app_fork_child_handler (void)
1495 {
1496   vcl_worker_t *parent_wrk;
1497   int parent_wrk_index;
1498
1499   parent_wrk_index = vcl_get_worker_index ();
1500   VDBG (0, "initializing forked child %u with parent wrk %u", getpid (),
1501         parent_wrk_index);
1502
1503   /*
1504    * Clear old state
1505    */
1506   vcl_set_worker_index (~0);
1507
1508   /*
1509    * Allocate and register vcl worker with vpp
1510    */
1511   if (vppcom_worker_register ())
1512     {
1513       VERR ("couldn't register new worker!");
1514       return;
1515     }
1516
1517   /*
1518    * Allocate/initialize vls worker and share sessions
1519    */
1520   vls_worker_alloc ();
1521   parent_wrk = vcl_worker_get (parent_wrk_index);
1522   vls_worker_copy_on_fork (parent_wrk);
1523   parent_wrk->forked_child = vcl_get_worker_index ();
1524
1525   /* Reset number of threads and set wrk index */
1526   vlsl->vls_mt_n_threads = 0;
1527   vlsl->vls_wrk_index = vcl_get_worker_index ();
1528   vlsl->select_mp_check = 0;
1529   vlsl->epoll_mp_check = 0;
1530   vls_mt_locks_init ();
1531
1532   VDBG (0, "forked child main worker initialized");
1533   vcm->forking = 0;
1534 }
1535
1536 static void
1537 vls_app_fork_parent_handler (void)
1538 {
1539   vcm->forking = 1;
1540   while (vcm->forking)
1541     ;
1542 }
1543
1544 void
1545 vls_app_exit (void)
1546 {
1547   vls_worker_t *wrk = vls_worker_get_current ();
1548
1549   /* Unshare the sessions. VCL will clean up the worker */
1550   vls_unshare_vcl_worker_sessions (vcl_worker_get_current ());
1551   vls_worker_free (wrk);
1552 }
1553
1554 static void
1555 vls_clone_and_share_rpc_handler (void *args)
1556 {
1557   vls_clone_and_share_msg_t *msg = (vls_clone_and_share_msg_t *) args;
1558   vls_worker_t *wrk = vls_worker_get_current (), *dst_wrk;
1559   vcl_locked_session_t *vls, *dst_vls;
1560   vcl_worker_t *vcl_wrk = vcl_worker_get_current (), *dst_vcl_wrk;
1561   vcl_session_t *s, *dst_s;
1562
1563   vls = vls_session_get (wrk, msg->vls_index);
1564
1565   if (!vls_mt_wrk_supported ())
1566     vls_init_share_session (wrk, vls);
1567
1568   s = vcl_session_get (vcl_wrk, msg->session_index);
1569   dst_wrk = vls_worker_get (msg->origin_vls_wrk);
1570   dst_vcl_wrk = vcl_worker_get (msg->origin_vcl_wrk);
1571   dst_vls = vls_session_get (dst_wrk, msg->origin_vls_index);
1572   dst_vls->shared_data_index = vls->shared_data_index;
1573   dst_s = vcl_session_get (dst_vcl_wrk, msg->origin_session_index);
1574   clib_memcpy (dst_s, s, sizeof (*s));
1575
1576   dst_vcl_wrk->rpc_done = 1;
1577
1578   VDBG (1, "proces session clone of worker (session): %u (%u) -> %u (%u)",
1579         vcl_wrk->wrk_index, msg->session_index, dst_vcl_wrk->wrk_index,
1580         msg->origin_session_index);
1581 }
1582
1583 static void
1584 vls_session_cleanup_rpc_handler (void *args)
1585 {
1586   vls_sess_cleanup_msg_t *msg = (vls_sess_cleanup_msg_t *) args;
1587   vcl_worker_t *wrk = vcl_worker_get_current ();
1588   vcl_worker_t *dst_wrk = vcl_worker_get (msg->origin_vcl_wrk);
1589
1590   vppcom_session_close (vcl_session_handle_from_index (msg->session_index));
1591
1592   dst_wrk->rpc_done = 1;
1593
1594   VDBG (1, "proces session cleanup of worker (session): %u (%u) from %u ()",
1595         wrk->wrk_index, msg->session_index, dst_wrk->wrk_index);
1596 }
1597
1598 static void
1599 vls_rpc_handler (void *args)
1600 {
1601   vls_rpc_msg_t *msg = (vls_rpc_msg_t *) args;
1602   switch (msg->type)
1603     {
1604     case VLS_RPC_CLONE_AND_SHARE:
1605       vls_clone_and_share_rpc_handler (msg->data);
1606       break;
1607     case VLS_RPC_SESS_CLEANUP:
1608       vls_session_cleanup_rpc_handler (msg->data);
1609       break;
1610     default:
1611       break;
1612     }
1613 }
1614
1615 void
1616 vls_send_clone_and_share_rpc (vcl_worker_t * wrk,
1617                               vcl_locked_session_t * vls, u32 session_index,
1618                               u32 vls_wrk_index, u32 dst_wrk_index,
1619                               u32 dst_vls_index, u32 dst_session_index)
1620 {
1621   u8 data[sizeof (u8) + sizeof (vls_clone_and_share_msg_t)];
1622   vls_clone_and_share_msg_t *msg;
1623   vls_rpc_msg_t *rpc;
1624   int ret;
1625
1626   rpc = (vls_rpc_msg_t *) & data;
1627   rpc->type = VLS_RPC_CLONE_AND_SHARE;
1628   msg = (vls_clone_and_share_msg_t *) & rpc->data;
1629   msg->origin_vls_wrk = vls_wrk_index;
1630   msg->origin_vls_index = vls->vls_index;
1631   msg->origin_vcl_wrk = wrk->wrk_index;
1632   msg->origin_session_index = session_index;
1633   msg->vls_index = dst_vls_index;
1634   msg->session_index = dst_session_index;
1635
1636   wrk->rpc_done = 0;
1637   ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
1638
1639   VDBG (1, "send session clone to wrk (session): %u (%u) -> %u (%u), ret=%d",
1640         dst_wrk_index, msg->session_index, msg->origin_vcl_wrk,
1641         msg->origin_session_index, ret);
1642   while (!ret && !wrk->rpc_done)
1643     ;
1644 }
1645
1646 void
1647 vls_send_session_cleanup_rpc (vcl_worker_t * wrk,
1648                               u32 dst_wrk_index, u32 dst_session_index)
1649 {
1650   u8 data[sizeof (u8) + sizeof (vls_sess_cleanup_msg_t)];
1651   vls_sess_cleanup_msg_t *msg;
1652   vls_rpc_msg_t *rpc;
1653   int ret;
1654
1655   rpc = (vls_rpc_msg_t *) & data;
1656   rpc->type = VLS_RPC_SESS_CLEANUP;
1657   msg = (vls_sess_cleanup_msg_t *) & rpc->data;
1658   msg->origin_vcl_wrk = wrk->wrk_index;
1659   msg->session_index = dst_session_index;
1660
1661   wrk->rpc_done = 0;
1662   ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
1663
1664   VDBG (1, "send session cleanup to wrk (session): %u (%u) from %u, ret=%d",
1665         dst_wrk_index, msg->session_index, msg->origin_vcl_wrk, ret);
1666   while (!ret && !wrk->rpc_done)
1667     ;
1668 }
1669
1670 int
1671 vls_app_create (char *app_name)
1672 {
1673   int rv;
1674
1675   if ((rv = vppcom_app_create (app_name)))
1676     return rv;
1677
1678   vlsm = clib_mem_alloc (sizeof (vls_main_t));
1679   clib_memset (vlsm, 0, sizeof (*vlsm));
1680   clib_rwlock_init (&vlsm->vls_table_lock);
1681   clib_rwlock_init (&vlsm->shared_data_lock);
1682   pool_alloc (vlsm->workers, vcm->cfg.max_workers);
1683
1684   pthread_atfork (vls_app_pre_fork, vls_app_fork_parent_handler,
1685                   vls_app_fork_child_handler);
1686   atexit (vls_app_exit);
1687   vls_worker_alloc ();
1688   vlsl->vls_wrk_index = vcl_get_worker_index ();
1689   vls_mt_locks_init ();
1690   vcm->wrk_rpc_fn = vls_rpc_handler;
1691   return VPPCOM_OK;
1692 }
1693
1694 unsigned char
1695 vls_use_eventfd (void)
1696 {
1697   return vcm->cfg.use_mq_eventfd;
1698 }
1699
1700 unsigned char
1701 vls_mt_wrk_supported (void)
1702 {
1703   return vcm->cfg.mt_wrk_supported;
1704 }
1705
1706 int
1707 vls_use_real_epoll (void)
1708 {
1709   if (vcl_get_worker_index () == ~0)
1710     return 0;
1711
1712   return vcl_worker_get_current ()->vcl_needs_real_epoll;
1713 }
1714
1715 void
1716 vls_register_vcl_worker (void)
1717 {
1718   vls_mt_add ();
1719 }
1720
1721 /*
1722  * fd.io coding-style-patch-verification: ON
1723  *
1724  * Local Variables:
1725  * eval: (c-set-style "gnu")
1726  * End:
1727  */