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