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