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