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