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