src/vcl/vcl_locked.c

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