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