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