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   if (s->session_state != VCL_STATE_LISTEN_NO_MQ)
 714     vcl_send_session_worker_update (wrk, s, vls_shd->owner_wrk_index);
 715
 716   /* XXX is this still needed? */
 717   if (vec_len (vls_shd->workers_subscribed) > 1)
 718     clib_warning ("more workers need to be updated");
 719
 720 done:
 721
 722   clib_spinlock_unlock (&vls_shd->lock);
 723   vls_shared_data_pool_runlock ();
 724
 725   return 0;
 726 }
 727
 728 void
 729 vls_init_share_session (vls_worker_t * vls_wrk, vcl_locked_session_t * vls)
 730 {
 731   vls_shared_data_t *vls_shd;
 732
 733   u32 vls_shd_index = vls_shared_data_alloc ();
 734
 735   vls_shared_data_pool_rlock ();
 736
 737   vls_shd = vls_shared_data_get (vls_shd_index);
 738   vls_shd->owner_wrk_index = vls_wrk->wrk_index;
 739   vls->shared_data_index = vls_shd_index;
 740   vec_add1 (vls_shd->workers_subscribed, vls_wrk->wrk_index);
 741
 742   vls_shared_data_pool_runlock ();
 743 }
 744
 745 void
 746 vls_share_session (vls_worker_t * vls_wrk, vcl_locked_session_t * vls)
 747 {
 748   vcl_worker_t *vcl_wrk = vcl_worker_get (vls_wrk->wrk_index);
 749   vls_shared_data_t *vls_shd;
 750   vcl_session_t *s;
 751
 752   s = vcl_session_get (vcl_wrk, vls->session_index);
 753   if (!s)
 754     {
 755       clib_warning ("wrk %u session %u vls %u NOT AVAILABLE",
 756                     vcl_wrk->wrk_index, vls->session_index, vls->vls_index);
 757       return;
 758     }
 759
 760   ASSERT (vls->shared_data_index != ~0);
 761
 762   /* Reinit session lock */
 763   clib_spinlock_init (&vls->lock);
 764
 765   vls_shared_data_pool_rlock ();
 766
 767   vls_shd = vls_shared_data_get (vls->shared_data_index);
 768
 769   clib_spinlock_lock (&vls_shd->lock);
 770   vec_add1 (vls_shd->workers_subscribed, vls_wrk->wrk_index);
 771   clib_spinlock_unlock (&vls_shd->lock);
 772
 773   vls_shared_data_pool_runlock ();
 774
 775   if (s->rx_fifo)
 776     {
 777       vcl_session_share_fifos (s, s->rx_fifo, s->tx_fifo);
 778     }
 779   else if (s->session_state == VCL_STATE_LISTEN)
 780     {
 781       s->session_state = VCL_STATE_LISTEN_NO_MQ;
 782     }
 783 }
 784
 785 static void
 786 vls_share_sessions (vls_worker_t * vls_parent_wrk, vls_worker_t * vls_wrk)
 787 {
 788   vcl_locked_session_t *vls, *parent_vls;
 789
 790   /* *INDENT-OFF* */
 791   pool_foreach (vls, vls_wrk->vls_pool)  {
 792     /* Initialize sharing on parent session */
 793     if (vls->shared_data_index == ~0)
 794       {
 795         parent_vls = vls_session_get (vls_parent_wrk, vls->vls_index);
 796         vls_init_share_session (vls_parent_wrk, parent_vls);
 797         vls->shared_data_index = parent_vls->shared_data_index;
 798       }
 799     vls_share_session (vls_wrk, vls);
 800   }
 801   /* *INDENT-ON* */
 802 }
 803
 804 static void
 805 vls_validate_veps (vcl_worker_t *wrk)
 806 {
 807   vcl_session_t *s;
 808   u32 session_index, wrk_index;
 809
 810   pool_foreach (s, wrk->sessions)
 811     {
 812       if (s->vep.vep_sh != ~0)
 813         {
 814           vcl_session_handle_parse (s->vep.vep_sh, &wrk_index, &session_index);
 815           s->vep.vep_sh = vcl_session_handle_from_index (session_index);
 816         }
 817       if (s->vep.next_sh != ~0)
 818         {
 819           vcl_session_handle_parse (s->vep.next_sh, &wrk_index,
 820                                     &session_index);
 821           s->vep.next_sh = vcl_session_handle_from_index (session_index);
 822         }
 823       if (s->vep.prev_sh != ~0)
 824         {
 825           vcl_session_handle_parse (s->vep.prev_sh, &wrk_index,
 826                                     &session_index);
 827           s->vep.prev_sh = vcl_session_handle_from_index (session_index);
 828         }
 829     }
 830 }
 831
 832 void
 833 vls_worker_copy_on_fork (vcl_worker_t * parent_wrk)
 834 {
 835   vls_worker_t *vls_wrk = vls_worker_get_current (), *vls_parent_wrk;
 836   vcl_worker_t *wrk = vcl_worker_get_current ();
 837   u32 vls_index, session_index, wrk_index;
 838   vcl_session_handle_t sh;
 839
 840   /*
 841    * init vcl worker
 842    */
 843   wrk->sessions = pool_dup (parent_wrk->sessions);
 844   wrk->session_index_by_vpp_handles =
 845     hash_dup (parent_wrk->session_index_by_vpp_handles);
 846
 847   /*
 848    * init vls worker
 849    */
 850   vls_parent_wrk = vls_worker_get (parent_wrk->wrk_index);
 851   /* *INDENT-OFF* */
 852   hash_foreach (sh, vls_index, vls_parent_wrk->session_handle_to_vlsh_table,
 853     ({
 854       vcl_session_handle_parse (sh, &wrk_index, &session_index);
 855       hash_set (vls_wrk->session_handle_to_vlsh_table,
 856                 vcl_session_handle_from_index (session_index), vls_index);
 857     }));
 858   /* *INDENT-ON* */
 859   vls_wrk->vls_pool = pool_dup (vls_parent_wrk->vls_pool);
 860
 861   /* Validate vep's handle */
 862   vls_validate_veps (wrk);
 863
 864   vls_share_sessions (vls_parent_wrk, vls_wrk);
 865 }
 866
 867 static void
 868 vls_mt_acq_locks (vcl_locked_session_t * vls, vls_mt_ops_t op, int *locks_acq)
 869 {
 870   vcl_worker_t *wrk = vcl_worker_get_current ();
 871   vcl_session_t *s = 0;
 872   int is_nonblk = 0;
 873
 874   if (vls)
 875     {
 876       s = vcl_session_get (wrk, vls->session_index);
 877       if (PREDICT_FALSE (!s))
 878         return;
 879       is_nonblk = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
 880     }
 881
 882   switch (op)
 883     {
 884     case VLS_MT_OP_READ:
 885       if (!is_nonblk)
 886         is_nonblk = vcl_session_read_ready (s) != 0;
 887       if (!is_nonblk)
 888         {
 889           vls_mt_mq_lock ();
 890           *locks_acq |= VLS_MT_LOCK_MQ;
 891         }
 892       break;
 893     case VLS_MT_OP_WRITE:
 894       ASSERT (s);
 895       if (!is_nonblk)
 896         is_nonblk = vcl_session_write_ready (s) != 0;
 897       if (!is_nonblk)
 898         {
 899           vls_mt_mq_lock ();
 900           *locks_acq |= VLS_MT_LOCK_MQ;
 901         }
 902       break;
 903     case VLS_MT_OP_XPOLL:
 904       vls_mt_mq_lock ();
 905       *locks_acq |= VLS_MT_LOCK_MQ;
 906       break;
 907     case VLS_MT_OP_SPOOL:
 908       vls_mt_spool_lock ();
 909       *locks_acq |= VLS_MT_LOCK_SPOOL;
 910       break;
 911     default:
 912       break;
 913     }
 914 }
 915
 916 static void
 917 vls_mt_rel_locks (int locks_acq)
 918 {
 919   if (locks_acq & VLS_MT_LOCK_MQ)
 920     vls_mt_mq_unlock ();
 921   if (locks_acq & VLS_MT_LOCK_SPOOL)
 922     vls_mt_create_unlock ();
 923 }
 924
 925 static inline u8
 926 vls_mt_session_should_migrate (vcl_locked_session_t * vls)
 927 {
 928   return (vls_mt_wrk_supported ()
 929           && vls->worker_index != vcl_get_worker_index ());
 930 }
 931
 932 static vcl_locked_session_t *
 933 vls_mt_session_migrate (vcl_locked_session_t *vls)
 934 {
 935   u32 wrk_index = vcl_get_worker_index ();
 936   vcl_worker_t *wrk;
 937   vls_worker_t *vls_wrk = vls_worker_get_current ();
 938   u32 src_sid, sid, vls_index, own_vcl_wrk_index;
 939   vcl_session_t *session;
 940   uword *p;
 941
 942   ASSERT (vls_mt_wrk_supported () && vls->worker_index != wrk_index);
 943
 944   /*
 945    * VCL session on current vcl worker already allocated. Update current
 946    * owner worker and index and return
 947    */
 948   if ((p = hash_get (vls->vcl_wrk_index_to_session_index, wrk_index)))
 949     {
 950       vls->worker_index = wrk_index;
 951       vls->session_index = (u32) p[0];
 952       return vls;
 953     }
 954
 955   /*
 956    * Ask vcl worker that owns the original vcl session to clone it into
 957    * current vcl worker session pool
 958    */
 959
 960   if (!(p = hash_get (vls->vcl_wrk_index_to_session_index,
 961                       vls->owner_vcl_wrk_index)))
 962     {
 963       VERR ("session in owner worker(%u) is free", vls->owner_vcl_wrk_index);
 964       ASSERT (0);
 965       vls_unlock (vls);
 966       vls_mt_table_runlock ();
 967       return 0;
 968     }
 969
 970   src_sid = (u32) p[0];
 971   wrk = vcl_worker_get_current ();
 972   session = vcl_session_alloc (wrk);
 973   sid = session->session_index;
 974   VDBG (1, "migrate session of worker (session): %u (%u) -> %u (%u)",
 975         vls->owner_vcl_wrk_index, src_sid, wrk_index, sid);
 976
 977   /* Drop lock to prevent dead lock when dst wrk trying to get lock. */
 978   vls_index = vls->vls_index;
 979   own_vcl_wrk_index = vls->owner_vcl_wrk_index;
 980   vls_unlock (vls);
 981   vls_mt_table_runlock ();
 982   vls_send_clone_and_share_rpc (wrk, vls_index, sid, vls_get_worker_index (),
 983                                 own_vcl_wrk_index, vls_index, src_sid);
 984
 985   if (PREDICT_FALSE (wrk->rpc_done == VLS_RPC_STATE_SESSION_NOT_EXIST))
 986     {
 987       VWRN ("session %u not exist", src_sid);
 988       goto err;
 989     }
 990   else if (PREDICT_FALSE (wrk->rpc_done == VLS_RPC_STATE_INIT))
 991     {
 992       VWRN ("failed to wait rpc response");
 993       goto err;
 994     }
 995   else if (PREDICT_FALSE ((session->flags & VCL_SESSION_F_IS_VEP) &&
 996                           session->vep.next_sh != ~0))
 997     {
 998       VERR ("can't migrate nonempty epoll session");
 999       ASSERT (0);
1000       goto err;
1001     }
1002   else if (PREDICT_FALSE (!(session->flags & VCL_SESSION_F_IS_VEP) &&
1003                           session->session_state != VCL_STATE_CLOSED))
1004     {
1005       VERR ("migrate NOT supported, session_status (%u)",
1006             session->session_state);
1007       ASSERT (0);
1008       goto err;
1009     }
1010
1011   vls = vls_get_w_dlock (vls_index);
1012   if (PREDICT_FALSE (!vls))
1013     {
1014       VWRN ("failed to get vls %u", vls_index);
1015       goto err;
1016     }
1017
1018   session->session_index = sid;
1019   vls->worker_index = wrk_index;
1020   vls->session_index = sid;
1021   hash_set (vls->vcl_wrk_index_to_session_index, wrk_index, sid);
1022   vls_sh_to_vlsh_table_add (vls_wrk, vcl_session_handle (session),
1023                             vls->vls_index);
1024   return vls;
1025
1026 err:
1027   vcl_session_free (wrk, session);
1028   return 0;
1029 }
1030
1031 static inline void
1032 vls_mt_detect (void)
1033 {
1034   if (PREDICT_FALSE (vcl_get_worker_index () == ~0))
1035     vls_mt_add ();
1036 }
1037
1038 #define vls_mt_guard(_vls, _op)                                               \
1039   int _locks_acq = 0;                                                         \
1040   if (vls_mt_wrk_supported ())                                                \
1041     {                                                                         \
1042       if (PREDICT_FALSE (_vls &&                                              \
1043                          ((vcl_locked_session_t *) _vls)->worker_index !=     \
1044                            vcl_get_worker_index ()))                          \
1045         {                                                                     \
1046           _vls = vls_mt_session_migrate (_vls);                               \
1047           if (PREDICT_FALSE (!_vls))                                          \
1048             return VPPCOM_EBADFD;                                             \
1049         }                                                                     \
1050     }                                                                         \
1051   else                                                                        \
1052     {                                                                         \
1053       if (PREDICT_FALSE (vlsl->vls_mt_n_threads > 1))                         \
1054         vls_mt_acq_locks (_vls, _op, &_locks_acq);                            \
1055     }
1056
1057 #define vls_mt_unguard()                                                \
1058   if (PREDICT_FALSE (_locks_acq))                                       \
1059     vls_mt_rel_locks (_locks_acq)
1060
1061 int
1062 vls_write (vls_handle_t vlsh, void *buf, size_t nbytes)
1063 {
1064   vcl_locked_session_t *vls;
1065   int rv;
1066
1067   vls_mt_detect ();
1068   if (!(vls = vls_get_w_dlock (vlsh)))
1069     return VPPCOM_EBADFD;
1070
1071   vls_mt_guard (vls, VLS_MT_OP_WRITE);
1072   rv = vppcom_session_write (vls_to_sh_tu (vls), buf, nbytes);
1073   vls_mt_unguard ();
1074   vls_get_and_unlock (vlsh);
1075   return rv;
1076 }
1077
1078 int
1079 vls_write_msg (vls_handle_t vlsh, void *buf, size_t nbytes)
1080 {
1081   vcl_locked_session_t *vls;
1082   int rv;
1083
1084   vls_mt_detect ();
1085   if (!(vls = vls_get_w_dlock (vlsh)))
1086     return VPPCOM_EBADFD;
1087   vls_mt_guard (vls, VLS_MT_OP_WRITE);
1088   rv = vppcom_session_write_msg (vls_to_sh_tu (vls), buf, nbytes);
1089   vls_mt_unguard ();
1090   vls_get_and_unlock (vlsh);
1091   return rv;
1092 }
1093
1094 int
1095 vls_sendto (vls_handle_t vlsh, void *buf, int buflen, int flags,
1096             vppcom_endpt_t * ep)
1097 {
1098   vcl_locked_session_t *vls;
1099   int rv;
1100
1101   vls_mt_detect ();
1102   if (!(vls = vls_get_w_dlock (vlsh)))
1103     return VPPCOM_EBADFD;
1104   vls_mt_guard (vls, VLS_MT_OP_WRITE);
1105   rv = vppcom_session_sendto (vls_to_sh_tu (vls), buf, buflen, flags, ep);
1106   vls_mt_unguard ();
1107   vls_get_and_unlock (vlsh);
1108   return rv;
1109 }
1110
1111 ssize_t
1112 vls_read (vls_handle_t vlsh, void *buf, size_t nbytes)
1113 {
1114   vcl_locked_session_t *vls;
1115   int rv;
1116
1117   vls_mt_detect ();
1118   if (!(vls = vls_get_w_dlock (vlsh)))
1119     return VPPCOM_EBADFD;
1120   vls_mt_guard (vls, VLS_MT_OP_READ);
1121   rv = vppcom_session_read (vls_to_sh_tu (vls), buf, nbytes);
1122   vls_mt_unguard ();
1123   vls_get_and_unlock (vlsh);
1124   return rv;
1125 }
1126
1127 ssize_t
1128 vls_recvfrom (vls_handle_t vlsh, void *buffer, uint32_t buflen, int flags,
1129               vppcom_endpt_t * ep)
1130 {
1131   vcl_locked_session_t *vls;
1132   int rv;
1133
1134   vls_mt_detect ();
1135   if (!(vls = vls_get_w_dlock (vlsh)))
1136     return VPPCOM_EBADFD;
1137   vls_mt_guard (vls, VLS_MT_OP_READ);
1138   rv = vppcom_session_recvfrom (vls_to_sh_tu (vls), buffer, buflen, flags,
1139                                 ep);
1140   vls_mt_unguard ();
1141   vls_get_and_unlock (vlsh);
1142   return rv;
1143 }
1144
1145 int
1146 vls_attr (vls_handle_t vlsh, uint32_t op, void *buffer, uint32_t * buflen)
1147 {
1148   vcl_locked_session_t *vls;
1149   int rv;
1150
1151   vls_mt_detect ();
1152   if (!(vls = vls_get_w_dlock (vlsh)))
1153     return VPPCOM_EBADFD;
1154   if (vls_mt_session_should_migrate (vls))
1155     {
1156       vls = vls_mt_session_migrate (vls);
1157       if (PREDICT_FALSE (!vls))
1158         return VPPCOM_EBADFD;
1159     }
1160   rv = vppcom_session_attr (vls_to_sh_tu (vls), op, buffer, buflen);
1161   vls_get_and_unlock (vlsh);
1162   return rv;
1163 }
1164
1165 int
1166 vls_bind (vls_handle_t vlsh, vppcom_endpt_t * ep)
1167 {
1168   vcl_locked_session_t *vls;
1169   int rv;
1170
1171   vls_mt_detect ();
1172   if (!(vls = vls_get_w_dlock (vlsh)))
1173     return VPPCOM_EBADFD;
1174   rv = vppcom_session_bind (vls_to_sh_tu (vls), ep);
1175   vls_get_and_unlock (vlsh);
1176   return rv;
1177 }
1178
1179 int
1180 vls_listen (vls_handle_t vlsh, int q_len)
1181 {
1182   vcl_locked_session_t *vls;
1183   int rv;
1184
1185   vls_mt_detect ();
1186   if (!(vls = vls_get_w_dlock (vlsh)))
1187     return VPPCOM_EBADFD;
1188   vls_mt_guard (vls, VLS_MT_OP_XPOLL);
1189   rv = vppcom_session_listen (vls_to_sh_tu (vls), q_len);
1190   vls_mt_unguard ();
1191   vls_get_and_unlock (vlsh);
1192   return rv;
1193 }
1194
1195 int
1196 vls_connect (vls_handle_t vlsh, vppcom_endpt_t * server_ep)
1197 {
1198   vcl_locked_session_t *vls;
1199   int rv;
1200
1201   vls_mt_detect ();
1202   if (!(vls = vls_get_w_dlock (vlsh)))
1203     return VPPCOM_EBADFD;
1204   vls_mt_guard (vls, VLS_MT_OP_XPOLL);
1205   rv = vppcom_session_connect (vls_to_sh_tu (vls), server_ep);
1206   vls_mt_unguard ();
1207   vls_get_and_unlock (vlsh);
1208   return rv;
1209 }
1210
1211 static inline void
1212 vls_mp_checks (vcl_locked_session_t * vls, int is_add)
1213 {
1214   vcl_worker_t *wrk = vcl_worker_get_current ();
1215   vcl_session_t *s;
1216   u32 owner_wrk;
1217
1218   if (vls_mt_wrk_supported ())
1219     return;
1220
1221   s = vcl_session_get (wrk, vls->session_index);
1222   switch (s->session_state)
1223     {
1224     case VCL_STATE_LISTEN:
1225       if (is_add)
1226         {
1227           vls_listener_wrk_set (vls, vls->worker_index, 1 /* is_active */ );
1228           break;
1229         }
1230       vls_listener_wrk_stop_listen (vls, vls->worker_index);
1231       break;
1232     case VCL_STATE_LISTEN_NO_MQ:
1233       if (!is_add)
1234         break;
1235
1236       /* Register worker as listener */
1237       vls_listener_wrk_start_listen (vls, wrk->wrk_index);
1238
1239       /* If owner worker did not attempt to accept/xpoll on the session,
1240        * force a listen stop for it, since it may not be interested in
1241        * accepting new sessions.
1242        * This is pretty much a hack done to give app workers the illusion
1243        * that it is fine to listen and not accept new sessions for a
1244        * given listener. Without it, we would accumulate unhandled
1245        * accepts on the passive worker message queue. */
1246       owner_wrk = vls_shared_get_owner (vls);
1247       if (!vls_listener_wrk_is_active (vls, owner_wrk))
1248         vls_listener_wrk_stop_listen (vls, owner_wrk);
1249       break;
1250     default:
1251       break;
1252     }
1253 }
1254
1255 vls_handle_t
1256 vls_accept (vls_handle_t listener_vlsh, vppcom_endpt_t * ep, int flags)
1257 {
1258   vls_handle_t accepted_vlsh;
1259   vcl_locked_session_t *vls;
1260   int sh;
1261
1262   vls_mt_detect ();
1263   if (!(vls = vls_get_w_dlock (listener_vlsh)))
1264     return VPPCOM_EBADFD;
1265   if (vcl_n_workers () > 1)
1266     vls_mp_checks (vls, 1 /* is_add */ );
1267   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1268   sh = vppcom_session_accept (vls_to_sh_tu (vls), ep, flags);
1269   vls_mt_unguard ();
1270   vls_get_and_unlock (listener_vlsh);
1271   if (sh < 0)
1272     return sh;
1273   accepted_vlsh = vls_alloc (sh);
1274   if (PREDICT_FALSE (accepted_vlsh == VLS_INVALID_HANDLE))
1275     vppcom_session_close (sh);
1276   return accepted_vlsh;
1277 }
1278
1279 vls_handle_t
1280 vls_create (uint8_t proto, uint8_t is_nonblocking)
1281 {
1282   vcl_session_handle_t sh;
1283   vls_handle_t vlsh;
1284   vcl_locked_session_t *vls = NULL;
1285
1286   vls_mt_detect ();
1287   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1288   sh = vppcom_session_create (proto, is_nonblocking);
1289   vls_mt_unguard ();
1290   if (sh == INVALID_SESSION_ID)
1291     return VLS_INVALID_HANDLE;
1292
1293   vlsh = vls_alloc (sh);
1294   if (PREDICT_FALSE (vlsh == VLS_INVALID_HANDLE))
1295     vppcom_session_close (sh);
1296
1297   return vlsh;
1298 }
1299
1300 static void
1301 vls_mt_session_cleanup (vcl_locked_session_t * vls)
1302 {
1303   u32 session_index, wrk_index, current_vcl_wrk;
1304   vcl_worker_t *wrk = vcl_worker_get_current ();
1305
1306   ASSERT (vls_mt_wrk_supported ());
1307
1308   current_vcl_wrk = vcl_get_worker_index ();
1309
1310   /* *INDENT-OFF* */
1311   hash_foreach (wrk_index, session_index, vls->vcl_wrk_index_to_session_index,
1312     ({
1313       if (current_vcl_wrk != wrk_index)
1314         vls_send_session_cleanup_rpc (wrk, wrk_index, session_index);
1315     }));
1316   /* *INDENT-ON* */
1317   hash_free (vls->vcl_wrk_index_to_session_index);
1318 }
1319
1320 int
1321 vls_close (vls_handle_t vlsh)
1322 {
1323   vcl_locked_session_t *vls;
1324   int rv;
1325
1326   vls_mt_detect ();
1327   vls_mt_table_wlock ();
1328
1329   vls = vls_get_and_lock (vlsh);
1330   if (!vls)
1331     {
1332       vls_mt_table_wunlock ();
1333       return VPPCOM_EBADFD;
1334     }
1335
1336   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1337
1338   if (vls_is_shared (vls))
1339     rv = vls_unshare_session (vls, vcl_worker_get_current ());
1340   else
1341     rv = vppcom_session_close (vls_to_sh (vls));
1342
1343   if (vls_mt_wrk_supported ())
1344     vls_mt_session_cleanup (vls);
1345
1346   vls_free (vls);
1347   vls_mt_unguard ();
1348
1349   vls_mt_table_wunlock ();
1350
1351   return rv;
1352 }
1353
1354 int
1355 vls_shutdown (vls_handle_t vlsh, int how)
1356 {
1357   vcl_locked_session_t *vls;
1358   int rv;
1359
1360   vls_mt_detect ();
1361   if (!(vls = vls_get_w_dlock (vlsh)))
1362     return VPPCOM_EBADFD;
1363
1364   vls_mt_guard (vls, VLS_MT_OP_SPOOL);
1365   rv = vppcom_session_shutdown (vls_to_sh_tu (vls), how);
1366   vls_mt_unguard ();
1367   vls_get_and_unlock (vlsh);
1368
1369   return rv;
1370 }
1371
1372 vls_handle_t
1373 vls_epoll_create (void)
1374 {
1375   vcl_session_handle_t sh;
1376   vls_handle_t vlsh;
1377
1378   vls_mt_detect ();
1379
1380   sh = vppcom_epoll_create ();
1381   if (sh == INVALID_SESSION_ID)
1382     return VLS_INVALID_HANDLE;
1383
1384   vlsh = vls_alloc (sh);
1385   if (vlsh == VLS_INVALID_HANDLE)
1386     vppcom_session_close (sh);
1387
1388   return vlsh;
1389 }
1390
1391 static void
1392 vls_epoll_ctl_mp_checks (vcl_locked_session_t * vls, int op)
1393 {
1394   if (vcl_n_workers () <= 1 || op == EPOLL_CTL_MOD)
1395     return;
1396
1397   vls_mp_checks (vls, op == EPOLL_CTL_ADD);
1398 }
1399
1400 int
1401 vls_epoll_ctl (vls_handle_t ep_vlsh, int op, vls_handle_t vlsh,
1402                struct epoll_event *event)
1403 {
1404   vcl_locked_session_t *ep_vls, *vls;
1405   vcl_session_handle_t ep_sh, sh;
1406   int rv;
1407
1408   vls_mt_detect ();
1409   vls_mt_table_rlock ();
1410   ep_vls = vls_get_and_lock (ep_vlsh);
1411
1412   if (vls_mt_session_should_migrate (ep_vls))
1413     {
1414       ep_vls = vls_mt_session_migrate (ep_vls);
1415       if (PREDICT_FALSE (!ep_vls))
1416         return VPPCOM_EBADFD;
1417     }
1418
1419   ep_sh = vls_to_sh (ep_vls);
1420   vls = vls_get_and_lock (vlsh);
1421   sh = vls_to_sh (vls);
1422
1423   vls_epoll_ctl_mp_checks (vls, op);
1424   vls_mt_table_runlock ();
1425   rv = vppcom_epoll_ctl (ep_sh, op, sh, event);
1426
1427   vls_mt_table_rlock ();
1428   ep_vls = vls_get (ep_vlsh);
1429   vls = vls_get (vlsh);
1430   vls_unlock (vls);
1431   vls_unlock (ep_vls);
1432   vls_mt_table_runlock ();
1433   return rv;
1434 }
1435
1436 int
1437 vls_epoll_wait (vls_handle_t ep_vlsh, struct epoll_event *events,
1438                 int maxevents, double wait_for_time)
1439 {
1440   vcl_locked_session_t *vls, *vls_tmp = NULL;
1441   int rv;
1442
1443   vls_mt_detect ();
1444   if (!(vls = vls_get_w_dlock (ep_vlsh)))
1445     return VPPCOM_EBADFD;
1446   vls_mt_guard (vls_tmp, VLS_MT_OP_XPOLL);
1447   rv = vppcom_epoll_wait (vls_to_sh_tu (vls), events, maxevents,
1448                           wait_for_time);
1449   vls_mt_unguard ();
1450   vls_get_and_unlock (ep_vlsh);
1451   vls_handle_pending_wrk_cleanup ();
1452   return rv;
1453 }
1454
1455 static void
1456 vls_select_mp_checks (vcl_si_set * read_map)
1457 {
1458   vcl_locked_session_t *vls;
1459   vcl_worker_t *wrk;
1460   vcl_session_t *s;
1461   u32 si;
1462
1463   if (vcl_n_workers () <= 1)
1464     {
1465       vlsl->select_mp_check = 1;
1466       return;
1467     }
1468
1469   if (!read_map)
1470     return;
1471
1472   vlsl->select_mp_check = 1;
1473   wrk = vcl_worker_get_current ();
1474
1475   /* *INDENT-OFF* */
1476   clib_bitmap_foreach (si, read_map)  {
1477     s = vcl_session_get (wrk, si);
1478     if (s->session_state == VCL_STATE_LISTEN)
1479       {
1480         vls = vls_get (vls_session_index_to_vlsh (si));
1481         vls_mp_checks (vls, 1 /* is_add */);
1482       }
1483   }
1484   /* *INDENT-ON* */
1485 }
1486
1487 int
1488 vls_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
1489             vcl_si_set * except_map, double wait_for_time)
1490 {
1491   int rv;
1492   vcl_locked_session_t *vls = NULL;
1493
1494   vls_mt_detect ();
1495   vls_mt_guard (vls, VLS_MT_OP_XPOLL);
1496   if (PREDICT_FALSE (!vlsl->select_mp_check))
1497     vls_select_mp_checks (read_map);
1498   rv = vppcom_select (n_bits, read_map, write_map, except_map, wait_for_time);
1499   vls_mt_unguard ();
1500   vls_handle_pending_wrk_cleanup ();
1501   return rv;
1502 }
1503
1504 static void
1505 vls_unshare_vcl_worker_sessions (vcl_worker_t * wrk)
1506 {
1507   u32 current_wrk, is_current;
1508   vcl_locked_session_t *vls;
1509   vcl_session_t *s;
1510
1511   if (pool_elts (vcm->workers) <= 1)
1512     return;
1513
1514   current_wrk = vcl_get_worker_index ();
1515   is_current = current_wrk == wrk->wrk_index;
1516
1517   /* *INDENT-OFF* */
1518   pool_foreach (s, wrk->sessions)  {
1519     vls = vls_get (vls_si_wi_to_vlsh (s->session_index, wrk->wrk_index));
1520     if (vls && (is_current || vls_is_shared_by_wrk (vls, current_wrk)))
1521       vls_unshare_session (vls, wrk);
1522   }
1523   /* *INDENT-ON* */
1524 }
1525
1526 static void
1527 vls_cleanup_vcl_worker (vcl_worker_t * wrk)
1528 {
1529   vls_worker_t *vls_wrk = vls_worker_get (wrk->wrk_index);
1530
1531   /* Unshare sessions and also cleanup worker since child may have
1532    * called _exit () and therefore vcl may not catch the event */
1533   vls_unshare_vcl_worker_sessions (wrk);
1534
1535   /* Since child may have exited and thereforce fd of vpp_app_socket_api
1536    * may have been closed, so DONOT notify VPP.
1537    */
1538   vcl_worker_cleanup (wrk, vcm->cfg.vpp_app_socket_api ? 0 : 1);
1539
1540   vls_worker_free (vls_wrk);
1541 }
1542
1543 static void
1544 vls_cleanup_forked_child (vcl_worker_t * wrk, vcl_worker_t * child_wrk)
1545 {
1546   vcl_worker_t *sub_child;
1547   int tries = 0;
1548
1549   if (child_wrk->forked_child != ~0)
1550     {
1551       sub_child = vcl_worker_get_if_valid (child_wrk->forked_child);
1552       if (sub_child)
1553         {
1554           /* Wait a bit, maybe the process is going away */
1555           while (kill (sub_child->current_pid, 0) >= 0 && tries++ < 50)
1556             usleep (1e3);
1557           if (kill (sub_child->current_pid, 0) < 0)
1558             vls_cleanup_forked_child (child_wrk, sub_child);
1559         }
1560     }
1561   vls_cleanup_vcl_worker (child_wrk);
1562   VDBG (0, "Cleaned up forked child wrk %u", child_wrk->wrk_index);
1563   wrk->forked_child = ~0;
1564 }
1565
1566 static void
1567 vls_handle_pending_wrk_cleanup (void)
1568 {
1569   u32 *wip;
1570   vcl_worker_t *child_wrk, *wrk;
1571   vls_worker_t *vls_wrk = vls_worker_get_current ();
1572
1573   if (PREDICT_TRUE (vec_len (vls_wrk->pending_wrk_cleanup) == 0))
1574     return;
1575
1576   wrk = vcl_worker_get_current ();
1577   vec_foreach (wip, vls_wrk->pending_wrk_cleanup)
1578     {
1579       child_wrk = vcl_worker_get_if_valid (*wip);
1580       if (!child_wrk)
1581         continue;
1582       vls_cleanup_forked_child (wrk, child_wrk);
1583     }
1584   vec_reset_length (vls_wrk->pending_wrk_cleanup);
1585 }
1586
1587 static struct sigaction old_sa;
1588
1589 static void
1590 vls_intercept_sigchld_handler (int signum, siginfo_t * si, void *uc)
1591 {
1592   vcl_worker_t *wrk, *child_wrk;
1593   vls_worker_t *vls_wrk;
1594
1595   if (vcl_get_worker_index () == ~0)
1596     return;
1597
1598   if (sigaction (SIGCHLD, &old_sa, 0))
1599     {
1600       VERR ("couldn't restore sigchld");
1601       exit (-1);
1602     }
1603
1604   wrk = vcl_worker_get_current ();
1605   if (wrk->forked_child == ~0)
1606     return;
1607
1608   child_wrk = vcl_worker_get_if_valid (wrk->forked_child);
1609   if (!child_wrk)
1610     goto done;
1611
1612   if (si && si->si_pid != child_wrk->current_pid)
1613     {
1614       VDBG (0, "unexpected child pid %u", si->si_pid);
1615       goto done;
1616     }
1617
1618   /* Parent process may enter sighandler with a lock, such as lock in localtime
1619    * or in mspace_free, and child wrk cleanup may try to get such locks and
1620    * cause deadlock.
1621    * So move child wrk cleanup from sighandler to vls_epoll_wait/vls_select.
1622    */
1623   vls_wrk = vls_worker_get_current ();
1624   vec_add1 (vls_wrk->pending_wrk_cleanup, child_wrk->wrk_index);
1625
1626 done:
1627   if (old_sa.sa_flags & SA_SIGINFO)
1628     {
1629       void (*fn) (int, siginfo_t *, void *) = old_sa.sa_sigaction;
1630       fn (signum, si, uc);
1631     }
1632   else
1633     {
1634       void (*fn) (int) = old_sa.sa_handler;
1635       if (fn)
1636         fn (signum);
1637     }
1638 }
1639
1640 static void
1641 vls_incercept_sigchld ()
1642 {
1643   struct sigaction sa;
1644   if (old_sa.sa_sigaction)
1645     {
1646       VDBG (0, "have intercepted sigchld");
1647       return;
1648     }
1649   clib_memset (&sa, 0, sizeof (sa));
1650   sa.sa_sigaction = vls_intercept_sigchld_handler;
1651   sa.sa_flags = SA_SIGINFO;
1652   if (sigaction (SIGCHLD, &sa, &old_sa))
1653     {
1654       VERR ("couldn't intercept sigchld");
1655       exit (-1);
1656     }
1657 }
1658
1659 static void
1660 vls_app_pre_fork (void)
1661 {
1662   vls_incercept_sigchld ();
1663   vcl_flush_mq_events ();
1664 }
1665
1666 static void
1667 vls_app_fork_child_handler (void)
1668 {
1669   vcl_worker_t *parent_wrk;
1670   int parent_wrk_index;
1671
1672   parent_wrk_index = vcl_get_worker_index ();
1673   VDBG (0, "initializing forked child %u with parent wrk %u", getpid (),
1674         parent_wrk_index);
1675
1676   /*
1677    * Clear old state
1678    */
1679   vcl_set_worker_index (~0);
1680
1681   /*
1682    * Allocate and register vcl worker with vpp
1683    */
1684   if (vppcom_worker_register ())
1685     {
1686       VERR ("couldn't register new worker!");
1687       return;
1688     }
1689
1690   /*
1691    * Allocate/initialize vls worker and share sessions
1692    */
1693   vls_worker_alloc ();
1694   parent_wrk = vcl_worker_get (parent_wrk_index);
1695   vls_worker_copy_on_fork (parent_wrk);
1696   parent_wrk->forked_child = vcl_get_worker_index ();
1697
1698   /* Reset number of threads and set wrk index */
1699   vlsl->vls_mt_n_threads = 0;
1700   vlsl->vls_wrk_index = vcl_get_worker_index ();
1701   vlsl->select_mp_check = 0;
1702   vls_mt_locks_init ();
1703
1704   VDBG (0, "forked child main worker initialized");
1705   vcm->forking = 0;
1706 }
1707
1708 static void
1709 vls_app_fork_parent_handler (void)
1710 {
1711   vcm->forking = 1;
1712   while (vcm->forking)
1713     ;
1714 }
1715
1716 void
1717 vls_app_exit (void)
1718 {
1719   vls_worker_t *wrk = vls_worker_get_current ();
1720
1721   /* Handle pending wrk cleanup */
1722   vls_handle_pending_wrk_cleanup ();
1723
1724   /* Unshare the sessions. VCL will clean up the worker */
1725   vls_unshare_vcl_worker_sessions (vcl_worker_get_current ());
1726   vls_worker_free (wrk);
1727 }
1728
1729 static void
1730 vls_clone_and_share_rpc_handler (void *args)
1731 {
1732   vls_clone_and_share_msg_t *msg = (vls_clone_and_share_msg_t *) args;
1733   vls_worker_t *wrk = vls_worker_get_current (), *dst_wrk;
1734   vcl_locked_session_t *vls, *dst_vls;
1735   vcl_worker_t *vcl_wrk = vcl_worker_get_current (), *dst_vcl_wrk;
1736   vcl_session_t *s, *dst_s;
1737
1738   VDBG (1, "process session clone of worker (session): %u (%u) -> %u (%u)",
1739         vcl_wrk->wrk_index, msg->session_index, msg->origin_vcl_wrk,
1740         msg->origin_session_index);
1741
1742   /* VCL locked session can't been protected, so DONT touch it.
1743    * VCL session may been free, check it.
1744    */
1745   dst_vcl_wrk = vcl_worker_get (msg->origin_vcl_wrk);
1746   s = vcl_session_get (vcl_wrk, msg->session_index);
1747   if (PREDICT_FALSE (!s))
1748     {
1749       dst_vcl_wrk->rpc_done = VLS_RPC_STATE_SESSION_NOT_EXIST;
1750       return;
1751     }
1752
1753   if (!vls_mt_wrk_supported ())
1754     {
1755       vls = vls_session_get (wrk, msg->vls_index);
1756       vls_init_share_session (wrk, vls);
1757       dst_wrk = vls_worker_get (msg->origin_vls_wrk);
1758       dst_vls = vls_session_get (dst_wrk, msg->origin_vls_index);
1759       dst_vls->shared_data_index = vls->shared_data_index;
1760     }
1761   dst_s = vcl_session_get (dst_vcl_wrk, msg->origin_session_index);
1762   clib_memcpy (dst_s, s, sizeof (*s));
1763
1764   dst_vcl_wrk->rpc_done = VLS_RPC_STATE_SUCCESS;
1765 }
1766
1767 static void
1768 vls_session_cleanup_rpc_handler (void *args)
1769 {
1770   vls_sess_cleanup_msg_t *msg = (vls_sess_cleanup_msg_t *) args;
1771   vcl_worker_t *wrk = vcl_worker_get_current ();
1772   vls_worker_t *vls_wrk = vls_worker_get_current ();
1773   vcl_session_handle_t sh = vcl_session_handle_from_index (msg->session_index);
1774
1775   VDBG (1, "process session cleanup of worker (session): %u (%u) from %u ()",
1776         wrk->wrk_index, msg->session_index, msg->origin_vcl_wrk);
1777
1778   vppcom_session_close (sh);
1779   vls_sh_to_vlsh_table_del (vls_wrk, sh);
1780 }
1781
1782 static void
1783 vls_rpc_handler (void *args)
1784 {
1785   vls_rpc_msg_t *msg = (vls_rpc_msg_t *) args;
1786   switch (msg->type)
1787     {
1788     case VLS_RPC_CLONE_AND_SHARE:
1789       vls_clone_and_share_rpc_handler (msg->data);
1790       break;
1791     case VLS_RPC_SESS_CLEANUP:
1792       vls_session_cleanup_rpc_handler (msg->data);
1793       break;
1794     default:
1795       break;
1796     }
1797 }
1798
1799 void
1800 vls_send_clone_and_share_rpc (vcl_worker_t *wrk, u32 origin_vls_index,
1801                               u32 session_index, u32 vls_wrk_index,
1802                               u32 dst_wrk_index, u32 dst_vls_index,
1803                               u32 dst_session_index)
1804 {
1805   u8 data[sizeof (u8) + sizeof (vls_clone_and_share_msg_t)];
1806   vls_clone_and_share_msg_t *msg;
1807   vls_rpc_msg_t *rpc;
1808   int ret;
1809   f64 timeout = clib_time_now (&wrk->clib_time) + VLS_WORKER_RPC_TIMEOUT;
1810
1811   rpc = (vls_rpc_msg_t *) & data;
1812   rpc->type = VLS_RPC_CLONE_AND_SHARE;
1813   msg = (vls_clone_and_share_msg_t *) & rpc->data;
1814   msg->origin_vls_wrk = vls_wrk_index;
1815   msg->origin_vls_index = origin_vls_index;
1816   msg->origin_vcl_wrk = wrk->wrk_index;
1817   msg->origin_session_index = session_index;
1818   msg->vls_index = dst_vls_index;
1819   msg->session_index = dst_session_index;
1820
1821   /* Try lock and handle rpcs if two threads send each other
1822    * clone requests at the same time.
1823    */
1824   wrk->rpc_done = VLS_RPC_STATE_INIT;
1825   while (!clib_spinlock_trylock (&vlsm->worker_rpc_lock))
1826     vcl_flush_mq_events ();
1827   ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
1828
1829   VDBG (1, "send session clone to wrk (session): %u (%u) -> %u (%u), ret=%d",
1830         dst_wrk_index, msg->session_index, msg->origin_vcl_wrk,
1831         msg->origin_session_index, ret);
1832   while (!ret && wrk->rpc_done == VLS_RPC_STATE_INIT &&
1833          clib_time_now (&wrk->clib_time) < timeout)
1834     ;
1835   clib_spinlock_unlock (&vlsm->worker_rpc_lock);
1836 }
1837
1838 void
1839 vls_send_session_cleanup_rpc (vcl_worker_t * wrk,
1840                               u32 dst_wrk_index, u32 dst_session_index)
1841 {
1842   u8 data[sizeof (u8) + sizeof (vls_sess_cleanup_msg_t)];
1843   vls_sess_cleanup_msg_t *msg;
1844   vls_rpc_msg_t *rpc;
1845   int ret;
1846
1847   rpc = (vls_rpc_msg_t *) & data;
1848   rpc->type = VLS_RPC_SESS_CLEANUP;
1849   msg = (vls_sess_cleanup_msg_t *) & rpc->data;
1850   msg->origin_vcl_wrk = wrk->wrk_index;
1851   msg->session_index = dst_session_index;
1852
1853   ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
1854
1855   VDBG (1, "send session cleanup to wrk (session): %u (%u) from %u, ret=%d",
1856         dst_wrk_index, msg->session_index, msg->origin_vcl_wrk, ret);
1857 }
1858
1859 int
1860 vls_app_create (char *app_name)
1861 {
1862   int rv;
1863
1864   if ((rv = vppcom_app_create (app_name)))
1865     return rv;
1866
1867   vlsm = clib_mem_alloc (sizeof (vls_main_t));
1868   clib_memset (vlsm, 0, sizeof (*vlsm));
1869   clib_rwlock_init (&vlsm->vls_table_lock);
1870   clib_rwlock_init (&vlsm->shared_data_lock);
1871   clib_spinlock_init (&vlsm->worker_rpc_lock);
1872   pool_alloc (vlsm->workers, vcm->cfg.max_workers);
1873
1874   pthread_atfork (vls_app_pre_fork, vls_app_fork_parent_handler,
1875                   vls_app_fork_child_handler);
1876   atexit (vls_app_exit);
1877   vls_worker_alloc ();
1878   vlsl->vls_wrk_index = vcl_get_worker_index ();
1879   vls_mt_locks_init ();
1880   vcm->wrk_rpc_fn = vls_rpc_handler;
1881   return VPPCOM_OK;
1882 }
1883
1884 unsigned char
1885 vls_use_eventfd (void)
1886 {
1887   return vcm->cfg.use_mq_eventfd;
1888 }
1889
1890 unsigned char
1891 vls_mt_wrk_supported (void)
1892 {
1893   return vcm->cfg.mt_wrk_supported;
1894 }
1895
1896 int
1897 vls_use_real_epoll (void)
1898 {
1899   if (vcl_get_worker_index () == ~0)
1900     return 0;
1901
1902   return vcl_worker_get_current ()->vcl_needs_real_epoll;
1903 }
1904
1905 void
1906 vls_register_vcl_worker (void)
1907 {
1908   vls_mt_add ();
1909 }
1910
1911 /*
1912  * fd.io coding-style-patch-verification: ON
1913  *
1914  * Local Variables:
1915  * eval: (c-set-style "gnu")
1916  * End:
1917  */