src/plugins/nat/nat_ha.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 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 "nat_ha.h"
  17 #include <vnet/udp/udp.h>
  18 #include <nat/nat.h>
  19 #include <vppinfra/atomics.h>
  20
  21 /* number of retries */
  22 #define NAT_HA_RETRIES 3
  23
  24 #define foreach_nat_ha_counter           \
  25 _(RECV_ADD, "add-event-recv", 0)         \
  26 _(RECV_DEL, "del-event-recv", 1)         \
  27 _(RECV_REFRESH, "refresh-event-recv", 2) \
  28 _(SEND_ADD, "add-event-send", 3)         \
  29 _(SEND_DEL, "del-event-send", 4)         \
  30 _(SEND_REFRESH, "refresh-event-send", 5) \
  31 _(RECV_ACK, "ack-recv", 6)               \
  32 _(SEND_ACK, "ack-send", 7)               \
  33 _(RETRY_COUNT, "retry-count", 8)         \
  34 _(MISSED_COUNT, "missed-count", 9)
  35
  36 /* NAT HA protocol version */
  37 #define NAT_HA_VERSION 0x01
  38
  39 /* NAT HA protocol flags */
  40 #define NAT_HA_FLAG_ACK 0x01
  41
  42 /* NAT HA event types */
  43 typedef enum
  44 {
  45   NAT_HA_ADD = 1,
  46   NAT_HA_DEL,
  47   NAT_HA_REFRESH,
  48 } nat_ha_event_type_t;
  49
  50 /* NAT HA protocol header */
  51 typedef struct
  52 {
  53   /* version */
  54   u8 version;
  55   /* flags */
  56   u8 flags;
  57   /* event count */
  58   u16 count;
  59   /* sequence number */
  60   u32 sequence_number;
  61   /* thread index where events originated */
  62   u32 thread_index;
  63 } __attribute__ ((packed)) nat_ha_message_header_t;
  64
  65 /* NAT HA protocol event data */
  66 typedef struct
  67 {
  68   /* event type */
  69   u8 event_type;
  70   /* session data */
  71   u8 protocol;
  72   u16 flags;
  73   u32 in_addr;
  74   u32 out_addr;
  75   u16 in_port;
  76   u16 out_port;
  77   u32 eh_addr;
  78   u32 ehn_addr;
  79   u16 eh_port;
  80   u16 ehn_port;
  81   u32 fib_index;
  82   u32 total_pkts;
  83   u64 total_bytes;
  84 } __attribute__ ((packed)) nat_ha_event_t;
  85
  86 typedef enum
  87 {
  88 #define _(N, s, v) NAT_HA_COUNTER_##N = v,
  89   foreach_nat_ha_counter
  90 #undef _
  91   NAT_HA_N_COUNTERS
  92 } nat_ha_counter_t;
  93
  94 /* data waiting for ACK */
  95 typedef struct
  96 {
  97   /* sequence number */
  98   u32 seq;
  99   /* retry count */
 100   u32 retry_count;
 101   /* next retry time */
 102   f64 retry_timer;
 103   /* 1 if HA resync */
 104   u8 is_resync;
 105   /* packet data */
 106   u8 *data;
 107 } nat_ha_resend_entry_t;
 108
 109 /* per thread data */
 110 typedef struct
 111 {
 112   /* buffer under construction */
 113   vlib_buffer_t *state_sync_buffer;
 114   /* frame containing NAT HA buffers */
 115   vlib_frame_t *state_sync_frame;
 116   /* number of events */
 117   u16 state_sync_count;
 118   /* next event offset */
 119   u32 state_sync_next_event_offset;
 120   /* data waiting for ACK */
 121   nat_ha_resend_entry_t *resend_queue;
 122 } nat_ha_per_thread_data_t;
 123
 124 /* NAT HA settings */
 125 typedef struct nat_ha_main_s
 126 {
 127   /* local IP address and UDP port */
 128   ip4_address_t src_ip_address;
 129   u16 src_port;
 130   /* failvoer IP address and UDP port */
 131   ip4_address_t dst_ip_address;
 132   u16 dst_port;
 133   /* path MTU between local and failover */
 134   u32 state_sync_path_mtu;
 135   /* number of seconds after which to send session counters refresh */
 136   u32 session_refresh_interval;
 137   /* counters */
 138   vlib_simple_counter_main_t counters[NAT_HA_N_COUNTERS];
 139   vlib_main_t *vlib_main;
 140   /* sequence number counter */
 141   u32 sequence_number;
 142   /* 1 if resync in progress */
 143   u8 in_resync;
 144   /* number of remaing ACK for resync */
 145   u32 resync_ack_count;
 146   /* number of missed ACK for resync */
 147   u32 resync_ack_missed;
 148   /* resync data */
 149   nat_ha_resync_event_cb_t event_callback;
 150   u32 client_index;
 151   u32 pid;
 152   /* call back functions for received HA events on failover */
 153   nat_ha_sadd_cb_t sadd_cb;
 154   nat_ha_sdel_cb_t sdel_cb;
 155   nat_ha_sref_cb_t sref_cb;
 156   /* per thread data */
 157   u32 num_workers;
 158   nat_ha_per_thread_data_t *per_thread_data;
 159   /* worker handoff frame-queue index */
 160   u32 fq_index;
 161 } nat_ha_main_t;
 162
 163 nat_ha_main_t nat_ha_main;
 164 vlib_node_registration_t nat_ha_process_node;
 165 vlib_node_registration_t nat_ha_worker_node;
 166 vlib_node_registration_t nat_ha_node;
 167 vlib_node_registration_t nat_ha_handoff_node;
 168
 169 static void
 170 nat_ha_resync_fin (void)
 171 {
 172   nat_ha_main_t *ha = &nat_ha_main;
 173
 174   /* if no more resync ACK remainig we are done */
 175   if (ha->resync_ack_count)
 176     return;
 177
 178   ha->in_resync = 0;
 179   if (ha->resync_ack_missed)
 180     {
 181       nat_elog_info ("resync completed with result FAILED");
 182     }
 183   else
 184     {
 185       nat_elog_info ("resync completed with result SUCCESS");
 186     }
 187   if (ha->event_callback)
 188     ha->event_callback (ha->client_index, ha->pid, ha->resync_ack_missed);
 189 }
 190
 191 /* cache HA NAT data waiting for ACK */
 192 static int
 193 nat_ha_resend_queue_add (u32 seq, u8 * data, u8 data_len, u8 is_resync,
 194                          u32 thread_index)
 195 {
 196   nat_ha_main_t *ha = &nat_ha_main;
 197   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 198   nat_ha_resend_entry_t *entry;
 199   f64 now = vlib_time_now (ha->vlib_main);
 200
 201   vec_add2 (td->resend_queue, entry, 1);
 202   clib_memset (entry, 0, sizeof (*entry));
 203   entry->retry_timer = now + 2.0;
 204   entry->seq = seq;
 205   entry->is_resync = is_resync;
 206   vec_add (entry->data, data, data_len);
 207
 208   return 0;
 209 }
 210
 211 static_always_inline void
 212 nat_ha_ack_recv (u32 seq, u32 thread_index)
 213 {
 214   nat_ha_main_t *ha = &nat_ha_main;
 215   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 216   u32 i;
 217
 218   vec_foreach_index (i, td->resend_queue)
 219   {
 220     if (td->resend_queue[i].seq != seq)
 221       continue;
 222
 223     vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_ACK],
 224                                    thread_index, 0, 1);
 225     /* ACK received remove cached data */
 226     if (td->resend_queue[i].is_resync)
 227       {
 228         clib_atomic_fetch_sub (&ha->resync_ack_count, 1);
 229         nat_ha_resync_fin ();
 230       }
 231     vec_free (td->resend_queue[i].data);
 232     vec_del1 (td->resend_queue, i);
 233     nat_elog_debug_X1 ("ACK for seq %d received", "i4",
 234                        clib_net_to_host_u32 (seq));
 235
 236     return;
 237   }
 238 }
 239
 240 /* scan non-ACKed HA NAT for retry */
 241 static void
 242 nat_ha_resend_scan (f64 now, u32 thread_index)
 243 {
 244   nat_ha_main_t *ha = &nat_ha_main;
 245   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 246   u32 i, *del, *to_delete = 0;
 247   vlib_main_t *vm = ha->vlib_main;
 248   vlib_buffer_t *b = 0;
 249   vlib_frame_t *f;
 250   u32 bi, *to_next;
 251   ip4_header_t *ip;
 252
 253   vec_foreach_index (i, td->resend_queue)
 254   {
 255     if (td->resend_queue[i].retry_timer > now)
 256       continue;
 257
 258     /* maximum retry reached delete cached data */
 259     if (td->resend_queue[i].retry_count >= NAT_HA_RETRIES)
 260       {
 261         nat_elog_notice_X1 ("seq %d missed", "i4",
 262                             clib_net_to_host_u32 (td->resend_queue[i].seq));
 263         if (td->resend_queue[i].is_resync)
 264           {
 265             clib_atomic_fetch_add (&ha->resync_ack_missed, 1);
 266             clib_atomic_fetch_sub (&ha->resync_ack_count, 1);
 267             nat_ha_resync_fin ();
 268           }
 269         vec_add1 (to_delete, i);
 270         vlib_increment_simple_counter (&ha->counters
 271                                        [NAT_HA_COUNTER_MISSED_COUNT],
 272                                        thread_index, 0, 1);
 273         continue;
 274       }
 275
 276     /* retry to send non-ACKed data */
 277     nat_elog_debug_X1 ("state sync seq %d resend", "i4",
 278                        clib_net_to_host_u32 (td->resend_queue[i].seq));
 279     td->resend_queue[i].retry_count++;
 280     vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RETRY_COUNT],
 281                                    thread_index, 0, 1);
 282     if (vlib_buffer_alloc (vm, &bi, 1) != 1)
 283       {
 284         nat_elog_warn ("HA NAT state sync can't allocate buffer");
 285         return;
 286       }
 287     b = vlib_get_buffer (vm, bi);
 288     b->current_length = vec_len (td->resend_queue[i].data);
 289     b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 290     b->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
 291     vnet_buffer (b)->sw_if_index[VLIB_RX] = 0;
 292     vnet_buffer (b)->sw_if_index[VLIB_TX] = 0;
 293     ip = vlib_buffer_get_current (b);
 294     clib_memcpy (ip, td->resend_queue[i].data,
 295                  vec_len (td->resend_queue[i].data));
 296     f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
 297     to_next = vlib_frame_vector_args (f);
 298     to_next[0] = bi;
 299     f->n_vectors = 1;
 300     vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
 301     td->resend_queue[i].retry_timer = now + 2.0;
 302   }
 303
 304   vec_foreach (del, to_delete)
 305   {
 306     vec_free (td->resend_queue[*del].data);
 307     vec_del1 (td->resend_queue, *del);
 308   }
 309   vec_free (to_delete);
 310 }
 311
 312 void
 313 nat_ha_init (vlib_main_t * vm, nat_ha_sadd_cb_t sadd_cb,
 314              nat_ha_sdel_cb_t sdel_cb, nat_ha_sref_cb_t sref_cb)
 315 {
 316   nat_ha_main_t *ha = &nat_ha_main;
 317   vlib_thread_main_t *tm = vlib_get_thread_main ();
 318   vlib_thread_registration_t *tr;
 319   uword *p;
 320
 321   ha->src_ip_address.as_u32 = 0;
 322   ha->src_port = 0;
 323   ha->dst_ip_address.as_u32 = 0;
 324   ha->dst_port = 0;
 325   ha->in_resync = 0;
 326   ha->resync_ack_count = 0;
 327   ha->resync_ack_missed = 0;
 328   ha->vlib_main = vm;
 329   ha->sadd_cb = sadd_cb;
 330   ha->sdel_cb = sdel_cb;
 331   ha->sref_cb = sref_cb;
 332   ha->num_workers = 0;
 333   vec_validate (ha->per_thread_data, tm->n_vlib_mains - 1);
 334   ha->fq_index = ~0;
 335   p = hash_get_mem (tm->thread_registrations_by_name, "workers");
 336   if (p)
 337     {
 338       tr = (vlib_thread_registration_t *) p[0];
 339       if (tr)
 340         ha->num_workers = tr->count;
 341     }
 342
 343 #define _(N, s, v) ha->counters[v].name = s;          \
 344   ha->counters[v].stat_segment_name = "/nat44/ha/" s; \
 345   vlib_validate_simple_counter(&ha->counters[v], 0);  \
 346   vlib_zero_simple_counter(&ha->counters[v], 0);
 347   foreach_nat_ha_counter
 348 #undef _
 349 }
 350
 351 int
 352 nat_ha_set_listener (ip4_address_t * addr, u16 port, u32 path_mtu)
 353 {
 354   nat_ha_main_t *ha = &nat_ha_main;
 355
 356   /* unregister previously set UDP port */
 357   if (ha->src_port)
 358     udp_unregister_dst_port (ha->vlib_main, ha->src_port, 1);
 359
 360   ha->src_ip_address.as_u32 = addr->as_u32;
 361   ha->src_port = port;
 362   ha->state_sync_path_mtu = path_mtu;
 363
 364   if (port)
 365     {
 366       /* if multiple worker threads first go to handoff node */
 367       if (ha->num_workers > 1)
 368         {
 369           if (ha->fq_index == ~0)
 370             ha->fq_index = vlib_frame_queue_main_init (nat_ha_node.index, 0);
 371           udp_register_dst_port (ha->vlib_main, port,
 372                                  nat_ha_handoff_node.index, 1);
 373         }
 374       else
 375         {
 376           udp_register_dst_port (ha->vlib_main, port, nat_ha_node.index, 1);
 377         }
 378       nat_elog_info_X1 ("HA listening on port %d for state sync", "i4", port);
 379     }
 380
 381   return 0;
 382 }
 383
 384 void
 385 nat_ha_get_listener (ip4_address_t * addr, u16 * port, u32 * path_mtu)
 386 {
 387   nat_ha_main_t *ha = &nat_ha_main;
 388
 389   addr->as_u32 = ha->src_ip_address.as_u32;
 390   *port = ha->src_port;
 391   *path_mtu = ha->state_sync_path_mtu;
 392 }
 393
 394 int
 395 nat_ha_set_failover (ip4_address_t * addr, u16 port,
 396                      u32 session_refresh_interval)
 397 {
 398   nat_ha_main_t *ha = &nat_ha_main;
 399
 400   ha->dst_ip_address.as_u32 = addr->as_u32;
 401   ha->dst_port = port;
 402   ha->session_refresh_interval = session_refresh_interval;
 403
 404   vlib_process_signal_event (ha->vlib_main, nat_ha_process_node.index, 1, 0);
 405
 406   return 0;
 407 }
 408
 409 void
 410 nat_ha_get_failover (ip4_address_t * addr, u16 * port,
 411                      u32 * session_refresh_interval)
 412 {
 413   nat_ha_main_t *ha = &nat_ha_main;
 414
 415   addr->as_u32 = ha->dst_ip_address.as_u32;
 416   *port = ha->dst_port;
 417   *session_refresh_interval = ha->session_refresh_interval;
 418 }
 419
 420 static_always_inline void
 421 nat_ha_recv_add (nat_ha_event_t * event, f64 now, u32 thread_index)
 422 {
 423   nat_ha_main_t *ha = &nat_ha_main;
 424   ip4_address_t in_addr, out_addr, eh_addr, ehn_addr;
 425   u32 fib_index;
 426   u16 flags;
 427
 428   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_ADD],
 429                                  thread_index, 0, 1);
 430
 431   in_addr.as_u32 = event->in_addr;
 432   out_addr.as_u32 = event->out_addr;
 433   eh_addr.as_u32 = event->eh_addr;
 434   ehn_addr.as_u32 = event->ehn_addr;
 435   fib_index = clib_net_to_host_u32 (event->fib_index);
 436   flags = clib_net_to_host_u16 (event->flags);
 437
 438   ha->sadd_cb (&in_addr, event->in_port, &out_addr, event->out_port, &eh_addr,
 439                event->eh_port, &ehn_addr, event->ehn_port, event->protocol,
 440                fib_index, flags, thread_index);
 441 }
 442
 443 static_always_inline void
 444 nat_ha_recv_del (nat_ha_event_t * event, u32 thread_index)
 445 {
 446   nat_ha_main_t *ha = &nat_ha_main;
 447   ip4_address_t out_addr, eh_addr;
 448   u32 fib_index;
 449
 450   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_DEL],
 451                                  thread_index, 0, 1);
 452
 453   out_addr.as_u32 = event->out_addr;
 454   eh_addr.as_u32 = event->eh_addr;
 455   fib_index = clib_net_to_host_u32 (event->fib_index);
 456
 457   ha->sdel_cb (&out_addr, event->out_port, &eh_addr, event->eh_port,
 458                event->protocol, fib_index, thread_index);
 459 }
 460
 461 static_always_inline void
 462 nat_ha_recv_refresh (nat_ha_event_t * event, f64 now, u32 thread_index)
 463 {
 464   nat_ha_main_t *ha = &nat_ha_main;
 465   ip4_address_t out_addr, eh_addr;
 466   u32 fib_index, total_pkts;
 467   u64 total_bytes;
 468
 469   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_REFRESH],
 470                                  thread_index, 0, 1);
 471
 472   out_addr.as_u32 = event->out_addr;
 473   eh_addr.as_u32 = event->eh_addr;
 474   fib_index = clib_net_to_host_u32 (event->fib_index);
 475   total_pkts = clib_net_to_host_u32 (event->total_pkts);
 476   total_bytes = clib_net_to_host_u64 (event->total_bytes);
 477
 478   ha->sref_cb (&out_addr, event->out_port, &eh_addr, event->eh_port,
 479                event->protocol, fib_index, total_pkts, total_bytes,
 480                thread_index);
 481 }
 482
 483 /* process received NAT HA event */
 484 static_always_inline void
 485 nat_ha_event_process (nat_ha_event_t * event, f64 now, u32 thread_index)
 486 {
 487   switch (event->event_type)
 488     {
 489     case NAT_HA_ADD:
 490       nat_ha_recv_add (event, now, thread_index);
 491       break;
 492     case NAT_HA_DEL:
 493       nat_ha_recv_del (event, thread_index);
 494       break;
 495     case NAT_HA_REFRESH:
 496       nat_ha_recv_refresh (event, now, thread_index);
 497       break;
 498     default:
 499       nat_elog_notice_X1 ("Unsupported HA event type %d", "i4",
 500                           event->event_type);
 501       break;
 502     }
 503 }
 504
 505 static inline void
 506 nat_ha_header_create (vlib_buffer_t * b, u32 * offset, u32 thread_index)
 507 {
 508   nat_ha_main_t *ha = &nat_ha_main;
 509   nat_ha_message_header_t *h;
 510   ip4_header_t *ip;
 511   udp_header_t *udp;
 512   u32 sequence_number;
 513
 514   b->current_data = 0;
 515   b->current_length = sizeof (*ip) + sizeof (*udp) + sizeof (*h);
 516   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 517   b->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
 518   vnet_buffer (b)->sw_if_index[VLIB_RX] = 0;
 519   vnet_buffer (b)->sw_if_index[VLIB_TX] = 0;
 520   ip = vlib_buffer_get_current (b);
 521   udp = (udp_header_t *) (ip + 1);
 522   h = (nat_ha_message_header_t *) (udp + 1);
 523
 524   /* IP header */
 525   ip->ip_version_and_header_length = 0x45;
 526   ip->ttl = 254;
 527   ip->protocol = IP_PROTOCOL_UDP;
 528   ip->flags_and_fragment_offset =
 529     clib_host_to_net_u16 (IP4_HEADER_FLAG_DONT_FRAGMENT);
 530   ip->src_address.as_u32 = ha->src_ip_address.as_u32;
 531   ip->dst_address.as_u32 = ha->dst_ip_address.as_u32;
 532   /* UDP header */
 533   udp->src_port = clib_host_to_net_u16 (ha->src_port);
 534   udp->dst_port = clib_host_to_net_u16 (ha->dst_port);
 535   udp->checksum = 0;
 536
 537   /* NAT HA protocol header */
 538   h->version = NAT_HA_VERSION;
 539   h->flags = 0;
 540   h->count = 0;
 541   h->thread_index = clib_host_to_net_u32 (thread_index);
 542   sequence_number = clib_atomic_fetch_add (&ha->sequence_number, 1);
 543   h->sequence_number = clib_host_to_net_u32 (sequence_number);
 544
 545   *offset =
 546     sizeof (ip4_header_t) + sizeof (udp_header_t) +
 547     sizeof (nat_ha_message_header_t);
 548 }
 549
 550 static inline void
 551 nat_ha_send (vlib_frame_t * f, vlib_buffer_t * b, u8 is_resync,
 552              u32 thread_index)
 553 {
 554   nat_ha_main_t *ha = &nat_ha_main;
 555   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 556   nat_ha_message_header_t *h;
 557   ip4_header_t *ip;
 558   udp_header_t *udp;
 559   vlib_main_t *vm = vlib_mains[thread_index];
 560
 561   ip = vlib_buffer_get_current (b);
 562   udp = ip4_next_header (ip);
 563   h = (nat_ha_message_header_t *) (udp + 1);
 564
 565   h->count = clib_host_to_net_u16 (td->state_sync_count);
 566
 567   ip->length = clib_host_to_net_u16 (b->current_length);
 568   ip->checksum = ip4_header_checksum (ip);
 569   udp->length = clib_host_to_net_u16 (b->current_length - sizeof (*ip));
 570
 571   nat_ha_resend_queue_add (h->sequence_number, (u8 *) ip, b->current_length,
 572                            is_resync, thread_index);
 573
 574   vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
 575 }
 576
 577 /* add NAT HA protocol event */
 578 static_always_inline void
 579 nat_ha_event_add (nat_ha_event_t * event, u8 do_flush, u32 thread_index,
 580                   u8 is_resync)
 581 {
 582   nat_ha_main_t *ha = &nat_ha_main;
 583   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 584   vlib_main_t *vm = vlib_mains[thread_index];
 585   vlib_buffer_t *b = 0;
 586   vlib_frame_t *f;
 587   u32 bi = ~0, offset;
 588
 589   b = td->state_sync_buffer;
 590
 591   if (PREDICT_FALSE (b == 0))
 592     {
 593       if (do_flush)
 594         return;
 595
 596       if (vlib_buffer_alloc (vm, &bi, 1) != 1)
 597         {
 598           nat_elog_warn ("HA NAT state sync can't allocate buffer");
 599           return;
 600         }
 601
 602       b = td->state_sync_buffer = vlib_get_buffer (vm, bi);
 603       clib_memset (vnet_buffer (b), 0, sizeof (*vnet_buffer (b)));
 604       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b);
 605       offset = 0;
 606     }
 607   else
 608     {
 609       bi = vlib_get_buffer_index (vm, b);
 610       offset = td->state_sync_next_event_offset;
 611     }
 612
 613   f = td->state_sync_frame;
 614   if (PREDICT_FALSE (f == 0))
 615     {
 616       u32 *to_next;
 617       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
 618       td->state_sync_frame = f;
 619       to_next = vlib_frame_vector_args (f);
 620       to_next[0] = bi;
 621       f->n_vectors = 1;
 622     }
 623
 624   if (PREDICT_FALSE (td->state_sync_count == 0))
 625     nat_ha_header_create (b, &offset, thread_index);
 626
 627   if (PREDICT_TRUE (do_flush == 0))
 628     {
 629       clib_memcpy_fast (b->data + offset, event, sizeof (*event));
 630       offset += sizeof (*event);
 631       td->state_sync_count++;
 632       b->current_length += sizeof (*event);
 633
 634       switch (event->event_type)
 635         {
 636         case NAT_HA_ADD:
 637           vlib_increment_simple_counter (&ha->counters
 638                                          [NAT_HA_COUNTER_SEND_ADD],
 639                                          thread_index, 0, 1);
 640           break;
 641         case NAT_HA_DEL:
 642           vlib_increment_simple_counter (&ha->counters
 643                                          [NAT_HA_COUNTER_SEND_DEL],
 644                                          thread_index, 0, 1);
 645           break;
 646         case NAT_HA_REFRESH:
 647           vlib_increment_simple_counter (&ha->counters
 648                                          [NAT_HA_COUNTER_SEND_REFRESH],
 649                                          thread_index, 0, 1);
 650           break;
 651         default:
 652           break;
 653         }
 654     }
 655
 656   if (PREDICT_FALSE
 657       (do_flush || offset + (sizeof (*event)) > ha->state_sync_path_mtu))
 658     {
 659       nat_ha_send (f, b, is_resync, thread_index);
 660       td->state_sync_buffer = 0;
 661       td->state_sync_frame = 0;
 662       td->state_sync_count = 0;
 663       offset = 0;
 664       if (is_resync)
 665         {
 666           clib_atomic_fetch_add (&ha->resync_ack_count, 1);
 667           nat_ha_resync_fin ();
 668         }
 669     }
 670
 671   td->state_sync_next_event_offset = offset;
 672 }
 673
 674 #define skip_if_disabled()          \
 675 do {                                \
 676   nat_ha_main_t *ha = &nat_ha_main; \
 677   if (PREDICT_TRUE (!ha->dst_port)) \
 678     return;                         \
 679 } while (0)
 680
 681 void
 682 nat_ha_flush (u8 is_resync)
 683 {
 684   skip_if_disabled ();
 685   nat_ha_event_add (0, 1, 0, is_resync);
 686 }
 687
 688 void
 689 nat_ha_sadd (ip4_address_t * in_addr, u16 in_port, ip4_address_t * out_addr,
 690              u16 out_port, ip4_address_t * eh_addr, u16 eh_port,
 691              ip4_address_t * ehn_addr, u16 ehn_port, u8 proto, u32 fib_index,
 692              u16 flags, u32 thread_index, u8 is_resync)
 693 {
 694   nat_ha_event_t event;
 695
 696   skip_if_disabled ();
 697
 698   clib_memset (&event, 0, sizeof (event));
 699   event.event_type = NAT_HA_ADD;
 700   event.flags = clib_host_to_net_u16 (flags);
 701   event.in_addr = in_addr->as_u32;
 702   event.in_port = in_port;
 703   event.out_addr = out_addr->as_u32;
 704   event.out_port = out_port;
 705   event.eh_addr = eh_addr->as_u32;
 706   event.eh_port = eh_port;
 707   event.ehn_addr = ehn_addr->as_u32;
 708   event.ehn_port = ehn_port;
 709   event.fib_index = clib_host_to_net_u32 (fib_index);
 710   event.protocol = proto;
 711   nat_ha_event_add (&event, 0, thread_index, is_resync);
 712 }
 713
 714 void
 715 nat_ha_sdel (ip4_address_t * out_addr, u16 out_port, ip4_address_t * eh_addr,
 716              u16 eh_port, u8 proto, u32 fib_index, u32 thread_index)
 717 {
 718   nat_ha_event_t event;
 719
 720   skip_if_disabled ();
 721
 722   clib_memset (&event, 0, sizeof (event));
 723   event.event_type = NAT_HA_DEL;
 724   event.out_addr = out_addr->as_u32;
 725   event.out_port = out_port;
 726   event.eh_addr = eh_addr->as_u32;
 727   event.eh_port = eh_port;
 728   event.fib_index = clib_host_to_net_u32 (fib_index);
 729   event.protocol = proto;
 730   nat_ha_event_add (&event, 0, thread_index, 0);
 731 }
 732
 733 void
 734 nat_ha_sref (ip4_address_t * out_addr, u16 out_port, ip4_address_t * eh_addr,
 735              u16 eh_port, u8 proto, u32 fib_index, u32 total_pkts,
 736              u64 total_bytes, u32 thread_index, f64 * last_refreshed, f64 now)
 737 {
 738   nat_ha_main_t *ha = &nat_ha_main;
 739   nat_ha_event_t event;
 740
 741   skip_if_disabled ();
 742
 743   if ((*last_refreshed + ha->session_refresh_interval) > now)
 744     return;
 745
 746   *last_refreshed = now;
 747   clib_memset (&event, 0, sizeof (event));
 748   event.event_type = NAT_HA_REFRESH;
 749   event.out_addr = out_addr->as_u32;
 750   event.out_port = out_port;
 751   event.eh_addr = eh_addr->as_u32;
 752   event.eh_port = eh_port;
 753   event.fib_index = clib_host_to_net_u32 (fib_index);
 754   event.protocol = proto;
 755   event.total_pkts = clib_host_to_net_u32 (total_pkts);
 756   event.total_bytes = clib_host_to_net_u64 (total_bytes);
 757   nat_ha_event_add (&event, 0, thread_index, 0);
 758 }
 759
 760 /* per thread process waiting for interrupt */
 761 static uword
 762 nat_ha_worker_fn (vlib_main_t * vm, vlib_node_runtime_t * rt,
 763                   vlib_frame_t * f)
 764 {
 765   u32 thread_index = vm->thread_index;
 766   /* flush HA NAT data under construction */
 767   nat_ha_event_add (0, 1, thread_index, 0);
 768   /* scan if we need to resend some non-ACKed data */
 769   nat_ha_resend_scan (vlib_time_now (vm), thread_index);
 770   return 0;
 771 }
 772
 773 /* *INDENT-OFF* */
 774 VLIB_REGISTER_NODE (nat_ha_worker_node) = {
 775     .function = nat_ha_worker_fn,
 776     .type = VLIB_NODE_TYPE_INPUT,
 777     .state = VLIB_NODE_STATE_INTERRUPT,
 778     .name = "nat-ha-worker",
 779 };
 780 /* *INDENT-ON* */
 781
 782 /* periodically send interrupt to each thread */
 783 static uword
 784 nat_ha_process (vlib_main_t * vm, vlib_node_runtime_t * rt, vlib_frame_t * f)
 785 {
 786   nat_ha_main_t *ha = &nat_ha_main;
 787   uword event_type;
 788   uword *event_data = 0;
 789   u32 ti;
 790
 791   vlib_process_wait_for_event (vm);
 792   event_type = vlib_process_get_events (vm, &event_data);
 793   if (event_type)
 794     nat_elog_info ("nat-ha-process: bogus kickoff event received");
 795   vec_reset_length (event_data);
 796
 797   while (1)
 798     {
 799       vlib_process_wait_for_event_or_clock (vm, 1.0);
 800       event_type = vlib_process_get_events (vm, &event_data);
 801       vec_reset_length (event_data);
 802       for (ti = 0; ti < vec_len (vlib_mains); ti++)
 803         {
 804           if (ti >= vec_len (ha->per_thread_data))
 805             continue;
 806
 807           vlib_node_set_interrupt_pending (vlib_mains[ti],
 808                                            nat_ha_worker_node.index);
 809         }
 810     }
 811
 812   return 0;
 813 }
 814
 815 /* *INDENT-OFF* */
 816 VLIB_REGISTER_NODE (nat_ha_process_node) = {
 817     .function = nat_ha_process,
 818     .type = VLIB_NODE_TYPE_PROCESS,
 819     .name = "nat-ha-process",
 820 };
 821 /* *INDENT-ON* */
 822
 823 void
 824 nat_ha_get_resync_status (u8 * in_resync, u32 * resync_ack_missed)
 825 {
 826   nat_ha_main_t *ha = &nat_ha_main;
 827
 828   *in_resync = ha->in_resync;
 829   *resync_ack_missed = ha->resync_ack_missed;
 830 }
 831
 832 typedef struct
 833 {
 834   ip4_address_t addr;
 835   u32 event_count;
 836 } nat_ha_trace_t;
 837
 838 static u8 *
 839 format_nat_ha_trace (u8 * s, va_list * args)
 840 {
 841   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
 842   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
 843   nat_ha_trace_t *t = va_arg (*args, nat_ha_trace_t *);
 844
 845   s =
 846     format (s, "nat-ha: %u events from %U", t->event_count,
 847             format_ip4_address, &t->addr);
 848
 849   return s;
 850 }
 851
 852 typedef enum
 853 {
 854   NAT_HA_NEXT_IP4_LOOKUP,
 855   NAT_HA_NEXT_DROP,
 856   NAT_HA_N_NEXT,
 857 } nat_ha_next_t;
 858
 859 #define foreach_nat_ha_error   \
 860 _(PROCESSED, "pkts-processed") \
 861 _(BAD_VERSION, "bad-version")
 862
 863 typedef enum
 864 {
 865 #define _(sym, str) NAT_HA_ERROR_##sym,
 866   foreach_nat_ha_error
 867 #undef _
 868     NAT_HA_N_ERROR,
 869 } nat_ha_error_t;
 870
 871 static char *nat_ha_error_strings[] = {
 872 #define _(sym, str) str,
 873   foreach_nat_ha_error
 874 #undef _
 875 };
 876
 877 /* process received HA NAT protocol messages */
 878 static uword
 879 nat_ha_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 880                 vlib_frame_t * frame)
 881 {
 882   u32 n_left_from, *from, next_index, *to_next;
 883   f64 now = vlib_time_now (vm);
 884   u32 thread_index = vm->thread_index;
 885   u32 pkts_processed = 0;
 886   ip4_main_t *i4m = &ip4_main;
 887   u8 host_config_ttl = i4m->host_config.ttl;
 888   nat_ha_main_t *ha = &nat_ha_main;
 889
 890   from = vlib_frame_vector_args (frame);
 891   n_left_from = frame->n_vectors;
 892   next_index = node->cached_next_index;
 893
 894   while (n_left_from > 0)
 895     {
 896       u32 n_left_to_next;
 897
 898       vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
 899
 900       while (n_left_from > 0 && n_left_to_next > 0)
 901         {
 902           u32 bi0, next0, src_addr0, dst_addr0;;
 903           vlib_buffer_t *b0;
 904           nat_ha_message_header_t *h0;
 905           nat_ha_event_t *e0;
 906           u16 event_count0, src_port0, dst_port0, old_len0;
 907           ip4_header_t *ip0;
 908           udp_header_t *udp0;
 909           ip_csum_t sum0;
 910
 911           bi0 = from[0];
 912           to_next[0] = bi0;
 913           from += 1;
 914           to_next += 1;
 915           n_left_from -= 1;
 916           n_left_to_next -= 1;
 917
 918           b0 = vlib_get_buffer (vm, bi0);
 919           h0 = vlib_buffer_get_current (b0);
 920           vlib_buffer_advance (b0, -sizeof (*udp0));
 921           udp0 = vlib_buffer_get_current (b0);
 922           vlib_buffer_advance (b0, -sizeof (*ip0));
 923           ip0 = vlib_buffer_get_current (b0);
 924
 925           next0 = NAT_HA_NEXT_DROP;
 926
 927           if (h0->version != NAT_HA_VERSION)
 928             {
 929               b0->error = node->errors[NAT_HA_ERROR_BAD_VERSION];
 930               goto done0;
 931             }
 932
 933           event_count0 = clib_net_to_host_u16 (h0->count);
 934           /* ACK for previously send data */
 935           if (!event_count0 && (h0->flags & NAT_HA_FLAG_ACK))
 936             {
 937               nat_ha_ack_recv (h0->sequence_number, thread_index);
 938               b0->error = node->errors[NAT_HA_ERROR_PROCESSED];
 939               goto done0;
 940             }
 941
 942           e0 = (nat_ha_event_t *) (h0 + 1);
 943
 944           /* process each event */
 945           while (event_count0)
 946             {
 947               nat_ha_event_process (e0, now, thread_index);
 948               event_count0--;
 949               e0 = (nat_ha_event_t *) ((u8 *) e0 + sizeof (nat_ha_event_t));
 950             }
 951
 952           next0 = NAT_HA_NEXT_IP4_LOOKUP;
 953           pkts_processed++;
 954
 955           /* reply with ACK */
 956           b0->current_length = sizeof (*ip0) + sizeof (*udp0) + sizeof (*h0);
 957
 958           src_addr0 = ip0->src_address.data_u32;
 959           dst_addr0 = ip0->dst_address.data_u32;
 960           ip0->src_address.data_u32 = dst_addr0;
 961           ip0->dst_address.data_u32 = src_addr0;
 962           old_len0 = ip0->length;
 963           ip0->length = clib_host_to_net_u16 (b0->current_length);
 964
 965           sum0 = ip0->checksum;
 966           sum0 = ip_csum_update (sum0, ip0->ttl, host_config_ttl,
 967                                  ip4_header_t, ttl);
 968           ip0->ttl = host_config_ttl;
 969           sum0 =
 970             ip_csum_update (sum0, old_len0, ip0->length, ip4_header_t,
 971                             length);
 972           ip0->checksum = ip_csum_fold (sum0);
 973
 974           udp0->checksum = 0;
 975           src_port0 = udp0->src_port;
 976           dst_port0 = udp0->dst_port;
 977           udp0->src_port = dst_port0;
 978           udp0->dst_port = src_port0;
 979           udp0->length =
 980             clib_host_to_net_u16 (b0->current_length - sizeof (*ip0));
 981
 982           h0->flags = NAT_HA_FLAG_ACK;
 983           h0->count = 0;
 984           vlib_increment_simple_counter (&ha->counters
 985                                          [NAT_HA_COUNTER_SEND_ACK],
 986                                          thread_index, 0, 1);
 987
 988         done0:
 989           if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE)
 990                              && (b0->flags & VLIB_BUFFER_IS_TRACED)))
 991             {
 992               nat_ha_trace_t *t = vlib_add_trace (vm, node, b0, sizeof (*t));
 993               ip4_header_t *ip =
 994                 (void *) (b0->data + vnet_buffer (b0)->l3_hdr_offset);
 995               t->event_count = clib_net_to_host_u16 (h0->count);
 996               t->addr.as_u32 = ip->src_address.data_u32;
 997             }
 998
 999           vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
1000                                            to_next, n_left_to_next,
1001                                            bi0, next0);
1002         }
1003
1004       vlib_put_next_frame (vm, node, next_index, n_left_to_next);
1005     }
1006
1007   vlib_node_increment_counter (vm, nat_ha_node.index,
1008                                NAT_HA_ERROR_PROCESSED, pkts_processed);
1009
1010   return frame->n_vectors;
1011 }
1012
1013 /* *INDENT-OFF* */
1014 VLIB_REGISTER_NODE (nat_ha_node) = {
1015   .function = nat_ha_node_fn,
1016   .name = "nat-ha",
1017   .vector_size = sizeof (u32),
1018   .format_trace = format_nat_ha_trace,
1019   .type = VLIB_NODE_TYPE_INTERNAL,
1020   .n_errors = ARRAY_LEN (nat_ha_error_strings),
1021   .error_strings = nat_ha_error_strings,
1022   .n_next_nodes = NAT_HA_N_NEXT,
1023   .next_nodes = {
1024      [NAT_HA_NEXT_IP4_LOOKUP] = "ip4-lookup",
1025      [NAT_HA_NEXT_DROP] = "error-drop",
1026   },
1027 };
1028 /* *INDENT-ON* */
1029
1030 typedef struct
1031 {
1032   u32 next_worker_index;
1033   u8 in2out;
1034 } nat_ha_handoff_trace_t;
1035
1036 #define foreach_nat_ha_handoff_error  \
1037 _(CONGESTION_DROP, "congestion drop") \
1038 _(SAME_WORKER, "same worker")         \
1039 _(DO_HANDOFF, "do handoff")
1040
1041 typedef enum
1042 {
1043 #define _(sym,str) NAT_HA_HANDOFF_ERROR_##sym,
1044   foreach_nat_ha_handoff_error
1045 #undef _
1046     NAT44_HANDOFF_N_ERROR,
1047 } nat_ha_handoff_error_t;
1048
1049 static char *nat_ha_handoff_error_strings[] = {
1050 #define _(sym,string) string,
1051   foreach_nat_ha_handoff_error
1052 #undef _
1053 };
1054
1055 static u8 *
1056 format_nat_ha_handoff_trace (u8 * s, va_list * args)
1057 {
1058   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
1059   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
1060   nat_ha_handoff_trace_t *t = va_arg (*args, nat_ha_handoff_trace_t *);
1061
1062   s =
1063     format (s, "NAT_HA_WORKER_HANDOFF: next-worker %d", t->next_worker_index);
1064
1065   return s;
1066 }
1067
1068 /* do worker handoff based on thread_index in NAT HA protcol header */
1069 static uword
1070 nat_ha_handoff_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
1071                         vlib_frame_t * frame)
1072 {
1073   nat_ha_main_t *ha = &nat_ha_main;
1074   vlib_buffer_t *bufs[VLIB_FRAME_SIZE], **b;
1075   u32 n_enq, n_left_from, *from;
1076   u16 thread_indices[VLIB_FRAME_SIZE], *ti;
1077   u32 thread_index = vm->thread_index;
1078   u32 do_handoff = 0, same_worker = 0;
1079
1080   from = vlib_frame_vector_args (frame);
1081   n_left_from = frame->n_vectors;
1082   vlib_get_buffers (vm, from, bufs, n_left_from);
1083
1084   b = bufs;
1085   ti = thread_indices;
1086
1087   while (n_left_from > 0)
1088     {
1089       nat_ha_message_header_t *h0;
1090
1091       h0 = vlib_buffer_get_current (b[0]);
1092       ti[0] = clib_net_to_host_u32 (h0->thread_index);
1093
1094       if (ti[0] != thread_index)
1095         do_handoff++;
1096       else
1097         same_worker++;
1098
1099       if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE)
1100                          && (b[0]->flags & VLIB_BUFFER_IS_TRACED)))
1101         {
1102           nat_ha_handoff_trace_t *t =
1103             vlib_add_trace (vm, node, b[0], sizeof (*t));
1104           t->next_worker_index = ti[0];
1105         }
1106
1107       n_left_from -= 1;
1108       ti += 1;
1109       b += 1;
1110     }
1111
1112   n_enq =
1113     vlib_buffer_enqueue_to_thread (vm, ha->fq_index, from, thread_indices,
1114                                    frame->n_vectors, 1);
1115
1116   if (n_enq < frame->n_vectors)
1117     vlib_node_increment_counter (vm, node->node_index,
1118                                  NAT_HA_HANDOFF_ERROR_CONGESTION_DROP,
1119                                  frame->n_vectors - n_enq);
1120   vlib_node_increment_counter (vm, node->node_index,
1121                                NAT_HA_HANDOFF_ERROR_SAME_WORKER, same_worker);
1122   vlib_node_increment_counter (vm, node->node_index,
1123                                NAT_HA_HANDOFF_ERROR_DO_HANDOFF, do_handoff);
1124   return frame->n_vectors;
1125 }
1126
1127 /* *INDENT-OFF* */
1128 VLIB_REGISTER_NODE (nat_ha_handoff_node) = {
1129   .function = nat_ha_handoff_node_fn,
1130   .name = "nat-ha-handoff",
1131   .vector_size = sizeof (u32),
1132   .format_trace = format_nat_ha_handoff_trace,
1133   .type = VLIB_NODE_TYPE_INTERNAL,
1134   .n_errors = ARRAY_LEN(nat_ha_handoff_error_strings),
1135   .error_strings = nat_ha_handoff_error_strings,
1136   .n_next_nodes = 1,
1137   .next_nodes = {
1138     [0] = "error-drop",
1139   },
1140 };
1141 /* *INDENT-ON* */
1142
1143 /*
1144  * fd.io coding-style-patch-verification: ON
1145  *
1146  * Local Variables:
1147  * eval: (c-set-style "gnu")
1148  * End:
1149  */