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