src/vnet/session/transport.c

   1 /*
   2  * Copyright (c) 2017-2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   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 <vnet/session/transport.h>
  17 #include <vnet/session/session.h>
  18 #include <vnet/fib/fib.h>
  19
  20 /**
  21  * Per-type vector of transport protocol virtual function tables
  22  */
  23 transport_proto_vft_t *tp_vfts;
  24
  25 typedef struct local_endpoint_
  26 {
  27   transport_endpoint_t ep;
  28   int refcnt;
  29 } local_endpoint_t;
  30
  31 typedef struct transport_main_
  32 {
  33   transport_endpoint_table_t local_endpoints_table;
  34   local_endpoint_t *local_endpoints;
  35   u32 port_allocator_seed;
  36   clib_spinlock_t local_endpoints_lock;
  37 } transport_main_t;
  38
  39 static transport_main_t tp_main;
  40
  41 u8 *
  42 format_transport_proto (u8 * s, va_list * args)
  43 {
  44   u32 transport_proto = va_arg (*args, u32);
  45
  46   if (tp_vfts[transport_proto].transport_options.name)
  47     s = format (s, "%s", tp_vfts[transport_proto].transport_options.name);
  48   else
  49     s = format (s, "n/a");
  50
  51   return s;
  52 }
  53
  54 u8 *
  55 format_transport_proto_short (u8 * s, va_list * args)
  56 {
  57   u32 transport_proto = va_arg (*args, u32);
  58   char *short_name;
  59
  60   short_name = tp_vfts[transport_proto].transport_options.short_name;
  61   if (short_name)
  62     s = format (s, "%s", short_name);
  63   else
  64     s = format (s, "NA");
  65
  66   return s;
  67 }
  68
  69 u8 *
  70 format_transport_connection (u8 * s, va_list * args)
  71 {
  72   u32 transport_proto = va_arg (*args, u32);
  73   u32 conn_index = va_arg (*args, u32);
  74   u32 thread_index = va_arg (*args, u32);
  75   u32 verbose = va_arg (*args, u32);
  76   transport_proto_vft_t *tp_vft;
  77   transport_connection_t *tc;
  78   u32 indent;
  79
  80   tp_vft = transport_protocol_get_vft (transport_proto);
  81   if (!tp_vft)
  82     return s;
  83
  84   s = format (s, "%U", tp_vft->format_connection, conn_index, thread_index,
  85               verbose);
  86   tc = tp_vft->get_connection (conn_index, thread_index);
  87   if (tc && verbose > 1)
  88     {
  89       indent = format_get_indent (s) + 1;
  90       if (transport_connection_is_tx_paced (tc))
  91         s = format (s, "%Upacer: %U\n", format_white_space, indent,
  92                     format_transport_pacer, &tc->pacer, tc->thread_index);
  93       s = format (s, "%Utransport: flags 0x%x\n", format_white_space, indent,
  94                   tc->flags);
  95     }
  96   return s;
  97 }
  98
  99 u8 *
 100 format_transport_listen_connection (u8 * s, va_list * args)
 101 {
 102   u32 transport_proto = va_arg (*args, u32);
 103   transport_proto_vft_t *tp_vft;
 104
 105   tp_vft = transport_protocol_get_vft (transport_proto);
 106   if (!tp_vft)
 107     return s;
 108
 109   s = (tp_vft->format_listener) (s, args);
 110   return s;
 111 }
 112
 113 u8 *
 114 format_transport_half_open_connection (u8 * s, va_list * args)
 115 {
 116   u32 transport_proto = va_arg (*args, u32);
 117   u32 ho_index = va_arg (*args, u32);
 118   transport_proto_vft_t *tp_vft;
 119
 120   tp_vft = transport_protocol_get_vft (transport_proto);
 121   if (!tp_vft)
 122     return s;
 123
 124   s = format (s, "%U", tp_vft->format_half_open, ho_index);
 125   return s;
 126 }
 127
 128 static u8
 129 unformat_transport_str_match (unformat_input_t * input, const char *str)
 130 {
 131   int i;
 132
 133   if (strlen (str) > vec_len (input->buffer) - input->index)
 134     return 0;
 135
 136   for (i = 0; i < strlen (str); i++)
 137     {
 138       if (input->buffer[i + input->index] != str[i])
 139         return 0;
 140     }
 141   return 1;
 142 }
 143
 144 uword
 145 unformat_transport_proto (unformat_input_t * input, va_list * args)
 146 {
 147   u32 *proto = va_arg (*args, u32 *);
 148   transport_proto_vft_t *tp_vft;
 149   u8 longest_match = 0, match;
 150   char *str, *str_match = 0;
 151   transport_proto_t tp;
 152
 153   for (tp = 0; tp < vec_len (tp_vfts); tp++)
 154     {
 155       tp_vft = &tp_vfts[tp];
 156       str = tp_vft->transport_options.name;
 157       if (!str)
 158         continue;
 159       if (unformat_transport_str_match (input, str))
 160         {
 161           match = strlen (str);
 162           if (match > longest_match)
 163             {
 164               *proto = tp;
 165               longest_match = match;
 166               str_match = str;
 167             }
 168         }
 169     }
 170   if (longest_match)
 171     {
 172       (void) unformat (input, str_match);
 173       return 1;
 174     }
 175
 176   return 0;
 177 }
 178
 179 u8 *
 180 format_transport_protos (u8 * s, va_list * args)
 181 {
 182   transport_proto_vft_t *tp_vft;
 183
 184   vec_foreach (tp_vft, tp_vfts)
 185     s = format (s, "%s\n", tp_vft->transport_options.name);
 186
 187   return s;
 188 }
 189
 190 u32
 191 transport_endpoint_lookup (transport_endpoint_table_t * ht, u8 proto,
 192                            ip46_address_t * ip, u16 port)
 193 {
 194   clib_bihash_kv_24_8_t kv;
 195   int rv;
 196
 197   kv.key[0] = ip->as_u64[0];
 198   kv.key[1] = ip->as_u64[1];
 199   kv.key[2] = (u64) port << 8 | (u64) proto;
 200
 201   rv = clib_bihash_search_inline_24_8 (ht, &kv);
 202   if (rv == 0)
 203     return kv.value;
 204
 205   return ENDPOINT_INVALID_INDEX;
 206 }
 207
 208 void
 209 transport_endpoint_table_add (transport_endpoint_table_t * ht, u8 proto,
 210                               transport_endpoint_t * te, u32 value)
 211 {
 212   clib_bihash_kv_24_8_t kv;
 213
 214   kv.key[0] = te->ip.as_u64[0];
 215   kv.key[1] = te->ip.as_u64[1];
 216   kv.key[2] = (u64) te->port << 8 | (u64) proto;
 217   kv.value = value;
 218
 219   clib_bihash_add_del_24_8 (ht, &kv, 1);
 220 }
 221
 222 void
 223 transport_endpoint_table_del (transport_endpoint_table_t * ht, u8 proto,
 224                               transport_endpoint_t * te)
 225 {
 226   clib_bihash_kv_24_8_t kv;
 227
 228   kv.key[0] = te->ip.as_u64[0];
 229   kv.key[1] = te->ip.as_u64[1];
 230   kv.key[2] = (u64) te->port << 8 | (u64) proto;
 231
 232   clib_bihash_add_del_24_8 (ht, &kv, 0);
 233 }
 234
 235 void
 236 transport_register_protocol (transport_proto_t transport_proto,
 237                              const transport_proto_vft_t * vft,
 238                              fib_protocol_t fib_proto, u32 output_node)
 239 {
 240   u8 is_ip4 = fib_proto == FIB_PROTOCOL_IP4;
 241
 242   vec_validate (tp_vfts, transport_proto);
 243   tp_vfts[transport_proto] = *vft;
 244
 245   session_register_transport (transport_proto, vft, is_ip4, output_node);
 246 }
 247
 248 transport_proto_t
 249 transport_register_new_protocol (const transport_proto_vft_t * vft,
 250                                  fib_protocol_t fib_proto, u32 output_node)
 251 {
 252   transport_proto_t transport_proto;
 253   u8 is_ip4;
 254
 255   transport_proto = session_add_transport_proto ();
 256   is_ip4 = fib_proto == FIB_PROTOCOL_IP4;
 257
 258   vec_validate (tp_vfts, transport_proto);
 259   tp_vfts[transport_proto] = *vft;
 260
 261   session_register_transport (transport_proto, vft, is_ip4, output_node);
 262
 263   return transport_proto;
 264 }
 265
 266 /**
 267  * Get transport virtual function table
 268  *
 269  * @param type - session type (not protocol type)
 270  */
 271 transport_proto_vft_t *
 272 transport_protocol_get_vft (transport_proto_t transport_proto)
 273 {
 274   if (transport_proto >= vec_len (tp_vfts))
 275     return 0;
 276   return &tp_vfts[transport_proto];
 277 }
 278
 279 transport_service_type_t
 280 transport_protocol_service_type (transport_proto_t tp)
 281 {
 282   return tp_vfts[tp].transport_options.service_type;
 283 }
 284
 285 transport_tx_fn_type_t
 286 transport_protocol_tx_fn_type (transport_proto_t tp)
 287 {
 288   return tp_vfts[tp].transport_options.tx_type;
 289 }
 290
 291 void
 292 transport_cleanup (transport_proto_t tp, u32 conn_index, u8 thread_index)
 293 {
 294   tp_vfts[tp].cleanup (conn_index, thread_index);
 295 }
 296
 297 void
 298 transport_cleanup_half_open (transport_proto_t tp, u32 conn_index)
 299 {
 300   if (tp_vfts[tp].cleanup_ho)
 301     tp_vfts[tp].cleanup_ho (conn_index);
 302 }
 303
 304 int
 305 transport_connect (transport_proto_t tp, transport_endpoint_cfg_t * tep)
 306 {
 307   if (PREDICT_FALSE (!tp_vfts[tp].connect))
 308     return SESSION_E_TRANSPORT_NO_REG;
 309   return tp_vfts[tp].connect (tep);
 310 }
 311
 312 void
 313 transport_half_close (transport_proto_t tp, u32 conn_index, u8 thread_index)
 314 {
 315   if (tp_vfts[tp].half_close)
 316     tp_vfts[tp].half_close (conn_index, thread_index);
 317 }
 318
 319 void
 320 transport_close (transport_proto_t tp, u32 conn_index, u8 thread_index)
 321 {
 322   tp_vfts[tp].close (conn_index, thread_index);
 323 }
 324
 325 void
 326 transport_reset (transport_proto_t tp, u32 conn_index, u8 thread_index)
 327 {
 328   if (tp_vfts[tp].reset)
 329     tp_vfts[tp].reset (conn_index, thread_index);
 330   else
 331     tp_vfts[tp].close (conn_index, thread_index);
 332 }
 333
 334 u32
 335 transport_start_listen (transport_proto_t tp, u32 session_index,
 336                         transport_endpoint_cfg_t *tep)
 337 {
 338   if (PREDICT_FALSE (!tp_vfts[tp].start_listen))
 339     return SESSION_E_TRANSPORT_NO_REG;
 340   return tp_vfts[tp].start_listen (session_index, tep);
 341 }
 342
 343 u32
 344 transport_stop_listen (transport_proto_t tp, u32 conn_index)
 345 {
 346   return tp_vfts[tp].stop_listen (conn_index);
 347 }
 348
 349 u8
 350 transport_protocol_is_cl (transport_proto_t tp)
 351 {
 352   return (tp_vfts[tp].transport_options.service_type == TRANSPORT_SERVICE_CL);
 353 }
 354
 355 always_inline void
 356 default_get_transport_endpoint (transport_connection_t * tc,
 357                                 transport_endpoint_t * tep, u8 is_lcl)
 358 {
 359   if (is_lcl)
 360     {
 361       tep->port = tc->lcl_port;
 362       tep->is_ip4 = tc->is_ip4;
 363       clib_memcpy_fast (&tep->ip, &tc->lcl_ip, sizeof (tc->lcl_ip));
 364     }
 365   else
 366     {
 367       tep->port = tc->rmt_port;
 368       tep->is_ip4 = tc->is_ip4;
 369       clib_memcpy_fast (&tep->ip, &tc->rmt_ip, sizeof (tc->rmt_ip));
 370     }
 371 }
 372
 373 void
 374 transport_get_endpoint (transport_proto_t tp, u32 conn_index,
 375                         u32 thread_index, transport_endpoint_t * tep,
 376                         u8 is_lcl)
 377 {
 378   if (tp_vfts[tp].get_transport_endpoint)
 379     tp_vfts[tp].get_transport_endpoint (conn_index, thread_index, tep,
 380                                         is_lcl);
 381   else
 382     {
 383       transport_connection_t *tc;
 384       tc = transport_get_connection (tp, conn_index, thread_index);
 385       default_get_transport_endpoint (tc, tep, is_lcl);
 386     }
 387 }
 388
 389 void
 390 transport_get_listener_endpoint (transport_proto_t tp, u32 conn_index,
 391                                  transport_endpoint_t * tep, u8 is_lcl)
 392 {
 393   if (tp_vfts[tp].get_transport_listener_endpoint)
 394     tp_vfts[tp].get_transport_listener_endpoint (conn_index, tep, is_lcl);
 395   else
 396     {
 397       transport_connection_t *tc;
 398       tc = transport_get_listener (tp, conn_index);
 399       default_get_transport_endpoint (tc, tep, is_lcl);
 400     }
 401 }
 402
 403 int
 404 transport_connection_attribute (transport_proto_t tp, u32 conn_index,
 405                                 u8 thread_index, u8 is_get,
 406                                 transport_endpt_attr_t *attr)
 407 {
 408   if (!tp_vfts[tp].attribute)
 409     return -1;
 410
 411   return tp_vfts[tp].attribute (conn_index, thread_index, is_get, attr);
 412 }
 413
 414 #define PORT_MASK ((1 << 16)- 1)
 415
 416 void
 417 transport_endpoint_free (u32 tepi)
 418 {
 419   transport_main_t *tm = &tp_main;
 420   pool_put_index (tm->local_endpoints, tepi);
 421 }
 422
 423 always_inline local_endpoint_t *
 424 transport_endpoint_alloc (void)
 425 {
 426   transport_main_t *tm = &tp_main;
 427   local_endpoint_t *lep;
 428
 429   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
 430   pool_get_aligned_safe (tm->local_endpoints, lep, 0);
 431   return lep;
 432 }
 433
 434 void
 435 transport_endpoint_cleanup (u8 proto, ip46_address_t * lcl_ip, u16 port)
 436 {
 437   transport_main_t *tm = &tp_main;
 438   local_endpoint_t *lep;
 439   u32 lepi;
 440
 441   /* Cleanup local endpoint if this was an active connect */
 442   lepi = transport_endpoint_lookup (&tm->local_endpoints_table, proto, lcl_ip,
 443                                     clib_net_to_host_u16 (port));
 444   if (lepi == ENDPOINT_INVALID_INDEX)
 445     return;
 446
 447   lep = pool_elt_at_index (tm->local_endpoints, lepi);
 448   if (!clib_atomic_sub_fetch (&lep->refcnt, 1))
 449     {
 450       transport_endpoint_table_del (&tm->local_endpoints_table, proto,
 451                                     &lep->ep);
 452
 453       /* All workers can free connections. Synchronize access to pool */
 454       clib_spinlock_lock (&tm->local_endpoints_lock);
 455       transport_endpoint_free (lepi);
 456       clib_spinlock_unlock (&tm->local_endpoints_lock);
 457     }
 458 }
 459
 460 static int
 461 transport_endpoint_mark_used (u8 proto, ip46_address_t *ip, u16 port)
 462 {
 463   transport_main_t *tm = &tp_main;
 464   local_endpoint_t *lep;
 465   u32 tei;
 466
 467   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
 468
 469   tei =
 470     transport_endpoint_lookup (&tm->local_endpoints_table, proto, ip, port);
 471   if (tei != ENDPOINT_INVALID_INDEX)
 472     return SESSION_E_PORTINUSE;
 473
 474   /* Pool reallocs with worker barrier */
 475   lep = transport_endpoint_alloc ();
 476   clib_memcpy_fast (&lep->ep.ip, ip, sizeof (*ip));
 477   lep->ep.port = port;
 478   lep->refcnt = 1;
 479
 480   transport_endpoint_table_add (&tm->local_endpoints_table, proto, &lep->ep,
 481                                 lep - tm->local_endpoints);
 482
 483   return 0;
 484 }
 485
 486 void
 487 transport_share_local_endpoint (u8 proto, ip46_address_t * lcl_ip, u16 port)
 488 {
 489   transport_main_t *tm = &tp_main;
 490   local_endpoint_t *lep;
 491   u32 lepi;
 492
 493   lepi = transport_endpoint_lookup (&tm->local_endpoints_table, proto, lcl_ip,
 494                                     clib_net_to_host_u16 (port));
 495   if (lepi != ENDPOINT_INVALID_INDEX)
 496     {
 497       lep = pool_elt_at_index (tm->local_endpoints, lepi);
 498       clib_atomic_add_fetch (&lep->refcnt, 1);
 499     }
 500 }
 501
 502 /**
 503  * Allocate local port and add if successful add entry to local endpoint
 504  * table to mark the pair as used.
 505  */
 506 int
 507 transport_alloc_local_port (u8 proto, ip46_address_t * ip)
 508 {
 509   u16 min = 1024, max = 65535;  /* XXX configurable ? */
 510   int tries, limit;
 511
 512   limit = max - min;
 513
 514   /* Only support active opens from one of ctrl threads */
 515   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
 516
 517   /* Search for first free slot */
 518   for (tries = 0; tries < limit; tries++)
 519     {
 520       u16 port = 0;
 521
 522       /* Find a port in the specified range */
 523       while (1)
 524         {
 525           port = random_u32 (&tp_main.port_allocator_seed) & PORT_MASK;
 526           if (PREDICT_TRUE (port >= min && port < max))
 527             break;
 528         }
 529
 530       if (!transport_endpoint_mark_used (proto, ip, port))
 531         return port;
 532     }
 533   return -1;
 534 }
 535
 536 static session_error_t
 537 transport_get_interface_ip (u32 sw_if_index, u8 is_ip4, ip46_address_t * addr)
 538 {
 539   if (is_ip4)
 540     {
 541       ip4_address_t *ip4;
 542       ip4 = ip_interface_get_first_ip (sw_if_index, 1);
 543       if (!ip4)
 544         return SESSION_E_NOIP;
 545       addr->ip4.as_u32 = ip4->as_u32;
 546     }
 547   else
 548     {
 549       ip6_address_t *ip6;
 550       ip6 = ip_interface_get_first_ip (sw_if_index, 0);
 551       if (ip6 == 0)
 552         return SESSION_E_NOIP;
 553       clib_memcpy_fast (&addr->ip6, ip6, sizeof (*ip6));
 554     }
 555   return 0;
 556 }
 557
 558 static session_error_t
 559 transport_find_local_ip_for_remote (u32 *sw_if_index,
 560                                     transport_endpoint_t *rmt,
 561                                     ip46_address_t *lcl_addr)
 562 {
 563   fib_node_index_t fei;
 564   fib_prefix_t prefix;
 565
 566   if (*sw_if_index == ENDPOINT_INVALID_INDEX)
 567     {
 568       /* Find a FIB path to the destination */
 569       clib_memcpy_fast (&prefix.fp_addr, &rmt->ip, sizeof (rmt->ip));
 570       prefix.fp_proto = rmt->is_ip4 ? FIB_PROTOCOL_IP4 : FIB_PROTOCOL_IP6;
 571       prefix.fp_len = rmt->is_ip4 ? 32 : 128;
 572
 573       ASSERT (rmt->fib_index != ENDPOINT_INVALID_INDEX);
 574       fei = fib_table_lookup (rmt->fib_index, &prefix);
 575
 576       /* Couldn't find route to destination. Bail out. */
 577       if (fei == FIB_NODE_INDEX_INVALID)
 578         return SESSION_E_NOROUTE;
 579
 580       *sw_if_index = fib_entry_get_resolving_interface (fei);
 581       if (*sw_if_index == ENDPOINT_INVALID_INDEX)
 582         return SESSION_E_NOINTF;
 583     }
 584
 585   clib_memset (lcl_addr, 0, sizeof (*lcl_addr));
 586   return transport_get_interface_ip (*sw_if_index, rmt->is_ip4, lcl_addr);
 587 }
 588
 589 int
 590 transport_alloc_local_endpoint (u8 proto, transport_endpoint_cfg_t * rmt_cfg,
 591                                 ip46_address_t * lcl_addr, u16 * lcl_port)
 592 {
 593   transport_endpoint_t *rmt = (transport_endpoint_t *) rmt_cfg;
 594   session_error_t error;
 595   int port;
 596
 597   /*
 598    * Find the local address
 599    */
 600   if (ip_is_zero (&rmt_cfg->peer.ip, rmt_cfg->peer.is_ip4))
 601     {
 602       error = transport_find_local_ip_for_remote (&rmt_cfg->peer.sw_if_index,
 603                                                   rmt, lcl_addr);
 604       if (error)
 605         return error;
 606     }
 607   else
 608     {
 609       /* Assume session layer vetted this address */
 610       clib_memcpy_fast (lcl_addr, &rmt_cfg->peer.ip,
 611                         sizeof (rmt_cfg->peer.ip));
 612     }
 613
 614   /*
 615    * Allocate source port
 616    */
 617   if (rmt_cfg->peer.port == 0)
 618     {
 619       port = transport_alloc_local_port (proto, lcl_addr);
 620       if (port < 1)
 621         return SESSION_E_NOPORT;
 622       *lcl_port = port;
 623     }
 624   else
 625     {
 626       port = clib_net_to_host_u16 (rmt_cfg->peer.port);
 627       *lcl_port = port;
 628
 629       return transport_endpoint_mark_used (proto, lcl_addr, port);
 630     }
 631
 632   return 0;
 633 }
 634
 635 u8 *
 636 format_clib_us_time (u8 * s, va_list * args)
 637 {
 638   clib_us_time_t t = va_arg (*args, clib_us_time_t);
 639   if (t < 1e3)
 640     s = format (s, "%u us", t);
 641   else
 642     s = format (s, "%.3f s", (f64) t * CLIB_US_TIME_PERIOD);
 643   return s;
 644 }
 645
 646 u8 *
 647 format_transport_pacer (u8 * s, va_list * args)
 648 {
 649   spacer_t *pacer = va_arg (*args, spacer_t *);
 650   u32 thread_index = va_arg (*args, int);
 651   clib_us_time_t now, diff;
 652
 653   now = transport_us_time_now (thread_index);
 654   diff = now - pacer->last_update;
 655   s = format (s, "rate %lu bucket %ld t/p %.3f last_update %U burst %u",
 656               pacer->bytes_per_sec, pacer->bucket, pacer->tokens_per_period,
 657               format_clib_us_time, diff, pacer->max_burst);
 658   return s;
 659 }
 660
 661 static inline u32
 662 spacer_max_burst (spacer_t * pacer, clib_us_time_t time_now)
 663 {
 664   u64 n_periods = (time_now - pacer->last_update);
 665   i64 inc;
 666
 667   if ((inc = (f32) n_periods * pacer->tokens_per_period) > 10)
 668     {
 669       pacer->last_update = time_now;
 670       pacer->bucket = clib_min (pacer->bucket + inc, (i64) pacer->max_burst);
 671     }
 672
 673   return pacer->bucket >= 0 ? pacer->max_burst : 0;
 674 }
 675
 676 static inline void
 677 spacer_update_bucket (spacer_t * pacer, u32 bytes)
 678 {
 679   pacer->bucket -= bytes;
 680 }
 681
 682 static inline void
 683 spacer_set_pace_rate (spacer_t * pacer, u64 rate_bytes_per_sec,
 684                       clib_us_time_t rtt, clib_time_type_t sec_per_loop)
 685 {
 686   clib_us_time_t max_time;
 687
 688   ASSERT (rate_bytes_per_sec != 0);
 689   pacer->bytes_per_sec = rate_bytes_per_sec;
 690   pacer->tokens_per_period = rate_bytes_per_sec * CLIB_US_TIME_PERIOD;
 691
 692   /* Allow a min number of bursts per rtt, if their size is acceptable. Goal
 693    * is to spread the sending of data over the rtt but to also allow for some
 694    * coalescing that can potentially
 695    * 1) reduce load on session layer by reducing scheduling frequency for a
 696    *    session and
 697    * 2) optimize sending when tso if available
 698    *
 699    * Max "time-length" of a burst cannot be less than 1us or more than 1ms.
 700    */
 701   max_time = clib_max (rtt / TRANSPORT_PACER_BURSTS_PER_RTT,
 702                        (clib_us_time_t) (sec_per_loop * CLIB_US_TIME_FREQ));
 703   max_time = clib_clamp (max_time, 1 /* 1us */ , 1000 /* 1ms */ );
 704   pacer->max_burst = (rate_bytes_per_sec * max_time) * CLIB_US_TIME_PERIOD;
 705   pacer->max_burst = clib_clamp (pacer->max_burst, TRANSPORT_PACER_MIN_BURST,
 706                                  TRANSPORT_PACER_MAX_BURST);
 707 }
 708
 709 static inline u64
 710 spacer_pace_rate (spacer_t * pacer)
 711 {
 712   return pacer->bytes_per_sec;
 713 }
 714
 715 static inline void
 716 spacer_reset (spacer_t * pacer, clib_us_time_t time_now, u64 bucket)
 717 {
 718   pacer->last_update = time_now;
 719   pacer->bucket = bucket;
 720 }
 721
 722 void
 723 transport_connection_tx_pacer_reset (transport_connection_t * tc,
 724                                      u64 rate_bytes_per_sec, u32 start_bucket,
 725                                      clib_us_time_t rtt)
 726 {
 727   spacer_set_pace_rate (&tc->pacer, rate_bytes_per_sec, rtt,
 728                         transport_seconds_per_loop (tc->thread_index));
 729   spacer_reset (&tc->pacer, transport_us_time_now (tc->thread_index),
 730                 start_bucket);
 731 }
 732
 733 void
 734 transport_connection_tx_pacer_reset_bucket (transport_connection_t * tc,
 735                                             u32 bucket)
 736 {
 737   spacer_reset (&tc->pacer, transport_us_time_now (tc->thread_index), bucket);
 738 }
 739
 740 void
 741 transport_connection_tx_pacer_init (transport_connection_t * tc,
 742                                     u64 rate_bytes_per_sec,
 743                                     u32 initial_bucket)
 744 {
 745   tc->flags |= TRANSPORT_CONNECTION_F_IS_TX_PACED;
 746   transport_connection_tx_pacer_reset (tc, rate_bytes_per_sec,
 747                                        initial_bucket, 1e6);
 748 }
 749
 750 void
 751 transport_connection_tx_pacer_update (transport_connection_t * tc,
 752                                       u64 bytes_per_sec, clib_us_time_t rtt)
 753 {
 754   spacer_set_pace_rate (&tc->pacer, bytes_per_sec, rtt,
 755                         transport_seconds_per_loop (tc->thread_index));
 756 }
 757
 758 u32
 759 transport_connection_tx_pacer_burst (transport_connection_t * tc)
 760 {
 761   return spacer_max_burst (&tc->pacer,
 762                            transport_us_time_now (tc->thread_index));
 763 }
 764
 765 u64
 766 transport_connection_tx_pacer_rate (transport_connection_t * tc)
 767 {
 768   return spacer_pace_rate (&tc->pacer);
 769 }
 770
 771 void
 772 transport_connection_update_tx_bytes (transport_connection_t * tc, u32 bytes)
 773 {
 774   if (transport_connection_is_tx_paced (tc))
 775     spacer_update_bucket (&tc->pacer, bytes);
 776 }
 777
 778 void
 779 transport_connection_tx_pacer_update_bytes (transport_connection_t * tc,
 780                                             u32 bytes)
 781 {
 782   spacer_update_bucket (&tc->pacer, bytes);
 783 }
 784
 785 void
 786 transport_update_pacer_time (u32 thread_index, clib_time_type_t now)
 787 {
 788   session_wrk_update_time (session_main_get_worker (thread_index), now);
 789 }
 790
 791 void
 792 transport_connection_reschedule (transport_connection_t * tc)
 793 {
 794   tc->flags &= ~TRANSPORT_CONNECTION_F_DESCHED;
 795   transport_connection_tx_pacer_reset_bucket (tc, 0 /* bucket */);
 796   if (transport_max_tx_dequeue (tc))
 797     sesssion_reschedule_tx (tc);
 798   else
 799     {
 800       session_t *s = session_get (tc->s_index, tc->thread_index);
 801       svm_fifo_unset_event (s->tx_fifo);
 802       if (svm_fifo_max_dequeue_cons (s->tx_fifo))
 803         if (svm_fifo_set_event (s->tx_fifo))
 804           sesssion_reschedule_tx (tc);
 805     }
 806 }
 807
 808 void
 809 transport_fifos_init_ooo (transport_connection_t * tc)
 810 {
 811   session_t *s = session_get (tc->s_index, tc->thread_index);
 812   svm_fifo_init_ooo_lookup (s->rx_fifo, 0 /* ooo enq */ );
 813   svm_fifo_init_ooo_lookup (s->tx_fifo, 1 /* ooo deq */ );
 814 }
 815
 816 void
 817 transport_update_time (clib_time_type_t time_now, u8 thread_index)
 818 {
 819   transport_proto_vft_t *vft;
 820   vec_foreach (vft, tp_vfts)
 821   {
 822     if (vft->update_time)
 823       (vft->update_time) (time_now, thread_index);
 824   }
 825 }
 826
 827 void
 828 transport_enable_disable (vlib_main_t * vm, u8 is_en)
 829 {
 830   transport_proto_vft_t *vft;
 831   vec_foreach (vft, tp_vfts)
 832   {
 833     if (vft->enable)
 834       (vft->enable) (vm, is_en);
 835
 836     if (vft->update_time)
 837       session_register_update_time_fn (vft->update_time, is_en);
 838   }
 839 }
 840
 841 void
 842 transport_init (void)
 843 {
 844   vlib_thread_main_t *vtm = vlib_get_thread_main ();
 845   session_main_t *smm = vnet_get_session_main ();
 846   transport_main_t *tm = &tp_main;
 847   u32 num_threads;
 848
 849   if (smm->local_endpoints_table_buckets == 0)
 850     smm->local_endpoints_table_buckets = 250000;
 851   if (smm->local_endpoints_table_memory == 0)
 852     smm->local_endpoints_table_memory = 512 << 20;
 853
 854   /* Initialize [port-allocator] random number seed */
 855   tm->port_allocator_seed = (u32) clib_cpu_time_now ();
 856
 857   clib_bihash_init_24_8 (&tm->local_endpoints_table, "local endpoints table",
 858                          smm->local_endpoints_table_buckets,
 859                          smm->local_endpoints_table_memory);
 860   clib_spinlock_init (&tm->local_endpoints_lock);
 861
 862   num_threads = 1 /* main thread */  + vtm->n_threads;
 863   if (num_threads > 1)
 864     {
 865       /* Main not polled if there are workers */
 866       smm->transport_cl_thread = 1;
 867     }
 868 }
 869
 870 /*
 871  * fd.io coding-style-patch-verification: ON
 872  *
 873  * Local Variables:
 874  * eval: (c-set-style "gnu")
 875  * End:
 876  */