NAT: VPP-1537 IPFIX per worker processing

Change-Id: I428bd25a513eb9fe65bea56572fea8cab7c51681
Signed-off-by: Filip Varga <fivarga@cisco.com>

1 files changed, 339 insertions, 405 deletions

diff --git a/src/plugins/nat/nat_ipfix_logging.c b/src/plugins/nat/nat_ipfix_logging.c
index 8bee8e444dc..4bb96bcf4a0 100644
--- a/src/plugins/nat/nat_ipfix_logging.c
+++ b/src/plugins/nat/nat_ipfix_logging.c
@@ -19,7 +19,9 @@
 #include <vlibmemory/api.h>
 #include <nat/nat_inlines.h>
 #include <nat/nat_ipfix_logging.h>
+#include <vppinfra/atomics.h>
 
+vlib_node_registration_t snat_ipfix_flush_node;
 snat_ipfix_logging_main_t snat_ipfix_logging_main;
 
 #define NAT44_SESSION_CREATE_LEN 26
@@ -111,11 +113,17 @@ typedef struct
   u32 vrf_id;
 } nat_ipfix_logging_nat64_bib_args_t;
 
-#define skip_if_disabled()                                    \
-do {                                                          \
-  snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main; \
-  if (PREDICT_TRUE (!silm->enabled))                          \
-    return;                                                   \
+#define skip_if_disabled()                                        \
+do {                                                              \
+  snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;     \
+  if (PREDICT_TRUE (!clib_atomic_fetch_or(&silm->enabled, 0)))    \
+    return;                                                       \
+} while (0)
+
+#define update_template_id(old_id, new_id)                \
+do {                                                      \
+  u16 template_id = clib_atomic_fetch_or(old_id, 0);      \
+  clib_atomic_cmp_and_swap(old_id, template_id, new_id);  \
 } while (0)
 
 /**
@@ -151,56 +159,79 @@ snat_template_rewrite (flow_report_main_t * frm,
   ip4_ipfix_template_packet_t *tp;
   u32 field_count = 0;
   flow_report_stream_t *stream;
+  u32 stream_index;
 
   stream = &frm->streams[fr->stream_index];
-  silm->stream_index = fr->stream_index;
+
+  stream_index = clib_atomic_fetch_or(&silm->stream_index, 0);
+  clib_atomic_cmp_and_swap (&silm->stream_index,
+                            stream_index, fr->stream_index);
 
   if (event == NAT_ADDRESSES_EXHAUTED)
     {
       field_count = NAT_ADDRESSES_EXHAUTED_FIELD_COUNT;
-      silm->addr_exhausted_template_id = fr->template_id;
+
+      update_template_id(&silm->addr_exhausted_template_id,
+                         fr->template_id);
     }
   else if (event == NAT44_SESSION_CREATE)
     {
       field_count = NAT44_SESSION_CREATE_FIELD_COUNT;
-      silm->nat44_session_template_id = fr->template_id;
+
+      update_template_id(&silm->nat44_session_template_id,
+                         fr->template_id);
     }
   else if (event == NAT64_BIB_CREATE)
     {
       field_count = NAT64_BIB_FIELD_COUNT;
-      silm->nat64_bib_template_id = fr->template_id;
+
+      update_template_id(&silm->nat64_bib_template_id,
+                         fr->template_id);
     }
   else if (event == NAT64_SESSION_CREATE)
     {
       field_count = NAT64_SES_FIELD_COUNT;
-      silm->nat64_ses_template_id = fr->template_id;
+
+      update_template_id(&silm->nat64_ses_template_id,
+                         fr->template_id);
     }
   else if (event == QUOTA_EXCEEDED)
     {
       if (quota_event == MAX_ENTRIES_PER_USER)
 	{
 	  field_count = MAX_ENTRIES_PER_USER_FIELD_COUNT;
-	  silm->max_entries_per_user_template_id = fr->template_id;
+
+          update_template_id(&silm->max_entries_per_user_template_id,
+                             fr->template_id);
+
 	}
       else if (quota_event == MAX_SESSION_ENTRIES)
 	{
 	  field_count = MAX_SESSIONS_FIELD_COUNT;
-	  silm->max_sessions_template_id = fr->template_id;
+
+          update_template_id(&silm->max_sessions_template_id,
+                             fr->template_id);
 	}
       else if (quota_event == MAX_BIB_ENTRIES)
 	{
 	  field_count = MAX_BIBS_FIELD_COUNT;
-	  silm->max_bibs_template_id = fr->template_id;
+
+          update_template_id(&silm->max_bibs_template_id,
+                             fr->template_id);
 	}
       else if (quota_event == MAX_FRAGMENTS_PENDING_REASSEMBLY)
 	{
 	  field_count = MAX_FRAGMENTS_FIELD_COUNT;
-	  silm->max_frags_ip4_template_id = fr->template_id;
+
+          update_template_id(&silm->max_frags_ip4_template_id,
+                             fr->template_id);
 	}
       else if (quota_event == MAX_FRAGMENTS_PENDING_REASSEMBLY_IP6)
         {
           field_count = MAX_FRAGMENTS_FIELD_COUNT;
-          silm->max_frags_ip6_template_id = fr->template_id;
+
+          update_template_id(&silm->max_frags_ip6_template_id,
+                             fr->template_id);
         }
     }
 
@@ -530,10 +561,13 @@ snat_ipfix_header_create (flow_report_main_t * frm,
   ip4_ipfix_template_packet_t *tp;
   ipfix_message_header_t *h = 0;
   ipfix_set_header_t *s = 0;
+  u32 sequence_number;
+  u32 stream_index;
   ip4_header_t *ip;
   udp_header_t *udp;
-
-  stream = &frm->streams[silm->stream_index];
+  
+  stream_index = clib_atomic_fetch_or(&silm->stream_index, 0);
+  stream = &frm->streams[stream_index];
 
   b0->current_data = 0;
   b0->current_length = sizeof (*ip) + sizeof (*udp) + sizeof (*h) +
@@ -561,7 +595,9 @@ snat_ipfix_header_create (flow_report_main_t * frm,
 					 (((f64) frm->unix_time_0) +
 					  (vlib_time_now (frm->vlib_main) -
 					   frm->vlib_time_0)));
-  h->sequence_number = clib_host_to_net_u32 (stream->sequence_number++);
+
+  sequence_number = clib_atomic_fetch_add (&stream->sequence_number, 1);
+  h->sequence_number = clib_host_to_net_u32 (sequence_number);
   h->domain_id = clib_host_to_net_u32 (stream->domain_id);
 
   *offset = (u32) (((u8 *) (s + 1)) - (u8 *) tp);
@@ -608,11 +644,13 @@ snat_ipfix_send (flow_report_main_t * frm,
 }
 
 static void
-snat_ipfix_logging_nat44_ses (u8 nat_event, u32 src_ip, u32 nat_src_ip,
-			      snat_protocol_t snat_proto, u16 src_port,
-			      u16 nat_src_port, u32 vrf_id, int do_flush)
+snat_ipfix_logging_nat44_ses (u32 thread_index, u8 nat_event, u32 src_ip,
+                              u32 nat_src_ip, snat_protocol_t snat_proto,
+                              u16 src_port, u16 nat_src_port, u32 vrf_id,
+                              int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -621,16 +659,14 @@ snat_ipfix_logging_nat44_ses (u8 nat_event, u32 src_ip, u32 nat_src_ip,
   vlib_main_t *vm = frm->vlib_main;
   u64 now;
   u8 proto = ~0;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   proto = snat_proto_to_ip_proto (snat_proto);
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->nat44_session_buffer;
+  b0 = sitd->nat44_session_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -643,22 +679,22 @@ snat_ipfix_logging_nat44_ses (u8 nat_event, u32 src_ip, u32 nat_src_ip,
 	  return;
 	}
 
-      b0 = silm->nat44_session_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->nat44_session_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->nat44_session_next_record_offset;
+      offset = sitd->nat44_session_next_record_offset;
     }
 
-  f = silm->nat44_session_frame;
+  f = sitd->nat44_session_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->nat44_session_frame = f;
+      sitd->nat44_session_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -700,18 +736,22 @@ snat_ipfix_logging_nat44_ses (u8 nat_event, u32 src_ip, u32 nat_src_ip,
   if (PREDICT_FALSE
       (do_flush || (offset + NAT44_SESSION_CREATE_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->nat44_session_template_id);
-      silm->nat44_session_frame = 0;
-      silm->nat44_session_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->nat44_session_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->nat44_session_frame = 0;
+      sitd->nat44_session_buffer = 0;
       offset = 0;
     }
-  silm->nat44_session_next_record_offset = offset;
+  sitd->nat44_session_next_record_offset = offset;
 }
 
 static void
-snat_ipfix_logging_addr_exhausted (u32 pool_id, int do_flush)
+snat_ipfix_logging_addr_exhausted (u32 thread_index, u32 pool_id, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -720,14 +760,12 @@ snat_ipfix_logging_addr_exhausted (u32 pool_id, int do_flush)
   vlib_main_t *vm = frm->vlib_main;
   u64 now;
   u8 nat_event = NAT_ADDRESSES_EXHAUTED;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->addr_exhausted_buffer;
+  b0 = sitd->addr_exhausted_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -740,22 +778,22 @@ snat_ipfix_logging_addr_exhausted (u32 pool_id, int do_flush)
 	  return;
 	}
 
-      b0 = silm->addr_exhausted_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->addr_exhausted_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->addr_exhausted_next_record_offset;
+      offset = sitd->addr_exhausted_next_record_offset;
     }
 
-  f = silm->addr_exhausted_frame;
+  f = sitd->addr_exhausted_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->addr_exhausted_frame = f;
+      sitd->addr_exhausted_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -782,18 +820,23 @@ snat_ipfix_logging_addr_exhausted (u32 pool_id, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + NAT_ADDRESSES_EXHAUTED_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->addr_exhausted_template_id);
-      silm->addr_exhausted_frame = 0;
-      silm->addr_exhausted_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+          &silm->addr_exhausted_template_id,
+          0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->addr_exhausted_frame = 0;
+      sitd->addr_exhausted_buffer = 0;
       offset = 0;
     }
-  silm->addr_exhausted_next_record_offset = offset;
+  sitd->addr_exhausted_next_record_offset = offset;
 }
 
 static void
-snat_ipfix_logging_max_entries_per_usr (u32 limit, u32 src_ip, int do_flush)
+snat_ipfix_logging_max_entries_per_usr (u32 thread_index,
+                                        u32 limit, u32 src_ip, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -803,14 +846,12 @@ snat_ipfix_logging_max_entries_per_usr (u32 limit, u32 src_ip, int do_flush)
   u64 now;
   u8 nat_event = QUOTA_EXCEEDED;
   u32 quota_event = MAX_ENTRIES_PER_USER;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->max_entries_per_user_buffer;
+  b0 = sitd->max_entries_per_user_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -823,22 +864,22 @@ snat_ipfix_logging_max_entries_per_usr (u32 limit, u32 src_ip, int do_flush)
 	  return;
 	}
 
-      b0 = silm->max_entries_per_user_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->max_entries_per_user_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->max_entries_per_user_next_record_offset;
+      offset = sitd->max_entries_per_user_next_record_offset;
     }
 
-  f = silm->max_entries_per_user_frame;
+  f = sitd->max_entries_per_user_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->max_entries_per_user_frame = f;
+      sitd->max_entries_per_user_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -871,18 +912,22 @@ snat_ipfix_logging_max_entries_per_usr (u32 limit, u32 src_ip, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + MAX_ENTRIES_PER_USER_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->max_entries_per_user_template_id);
-      silm->max_entries_per_user_frame = 0;
-      silm->max_entries_per_user_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+          &silm->max_entries_per_user_template_id,
+          0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->max_entries_per_user_frame = 0;
+      sitd->max_entries_per_user_buffer = 0;
       offset = 0;
     }
-  silm->max_entries_per_user_next_record_offset = offset;
+  sitd->max_entries_per_user_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_max_ses (u32 limit, int do_flush)
+nat_ipfix_logging_max_ses (u32 thread_index, u32 limit, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -892,14 +937,12 @@ nat_ipfix_logging_max_ses (u32 limit, int do_flush)
   u64 now;
   u8 nat_event = QUOTA_EXCEEDED;
   u32 quota_event = MAX_SESSION_ENTRIES;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->max_sessions_buffer;
+  b0 = sitd->max_sessions_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -912,22 +955,22 @@ nat_ipfix_logging_max_ses (u32 limit, int do_flush)
 	  return;
 	}
 
-      b0 = silm->max_sessions_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->max_sessions_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->max_sessions_next_record_offset;
+      offset = sitd->max_sessions_next_record_offset;
     }
 
-  f = silm->max_sessions_frame;
+  f = sitd->max_sessions_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->max_sessions_frame = f;
+      sitd->max_sessions_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -957,18 +1000,22 @@ nat_ipfix_logging_max_ses (u32 limit, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + MAX_SESSIONS_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->max_sessions_template_id);
-      silm->max_sessions_frame = 0;
-      silm->max_sessions_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->max_sessions_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->max_sessions_frame = 0;
+      sitd->max_sessions_buffer = 0;
       offset = 0;
     }
-  silm->max_sessions_next_record_offset = offset;
+  sitd->max_sessions_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_max_bib (u32 limit, int do_flush)
+nat_ipfix_logging_max_bib (u32 thread_index, u32 limit, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -978,14 +1025,12 @@ nat_ipfix_logging_max_bib (u32 limit, int do_flush)
   u64 now;
   u8 nat_event = QUOTA_EXCEEDED;
   u32 quota_event = MAX_BIB_ENTRIES;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->max_bibs_buffer;
+  b0 = sitd->max_bibs_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -998,22 +1043,22 @@ nat_ipfix_logging_max_bib (u32 limit, int do_flush)
 	  return;
 	}
 
-      b0 = silm->max_bibs_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->max_bibs_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->max_bibs_next_record_offset;
+      offset = sitd->max_bibs_next_record_offset;
     }
 
-  f = silm->max_bibs_frame;
+  f = sitd->max_bibs_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->max_bibs_frame = f;
+      sitd->max_bibs_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -1043,18 +1088,23 @@ nat_ipfix_logging_max_bib (u32 limit, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + MAX_BIBS_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->max_bibs_template_id);
-      silm->max_bibs_frame = 0;
-      silm->max_bibs_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->max_bibs_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->max_bibs_frame = 0;
+      sitd->max_bibs_buffer = 0;
       offset = 0;
     }
-  silm->max_bibs_next_record_offset = offset;
+  sitd->max_bibs_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_max_frag_ip4 (u32 limit, u32 src, int do_flush)
+nat_ipfix_logging_max_frag_ip4 (u32 thread_index,
+                                u32 limit, u32 src, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -1064,14 +1114,12 @@ nat_ipfix_logging_max_frag_ip4 (u32 limit, u32 src, int do_flush)
   u64 now;
   u8 nat_event = QUOTA_EXCEEDED;
   u32 quota_event = MAX_FRAGMENTS_PENDING_REASSEMBLY;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->max_frags_ip4_buffer;
+  b0 = sitd->max_frags_ip4_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -1084,22 +1132,22 @@ nat_ipfix_logging_max_frag_ip4 (u32 limit, u32 src, int do_flush)
 	  return;
 	}
 
-      b0 = silm->max_frags_ip4_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->max_frags_ip4_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->max_frags_ip4_next_record_offset;
+      offset = sitd->max_frags_ip4_next_record_offset;
     }
 
-  f = silm->max_frags_ip4_frame;
+  f = sitd->max_frags_ip4_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->max_frags_ip4_frame = f;
+      sitd->max_frags_ip4_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -1132,18 +1180,23 @@ nat_ipfix_logging_max_frag_ip4 (u32 limit, u32 src, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + MAX_BIBS_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->max_frags_ip4_template_id);
-      silm->max_frags_ip4_frame = 0;
-      silm->max_frags_ip4_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->max_frags_ip4_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->max_frags_ip4_frame = 0;
+      sitd->max_frags_ip4_buffer = 0;
       offset = 0;
     }
-  silm->max_frags_ip4_next_record_offset = offset;
+  sitd->max_frags_ip4_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_max_frag_ip6 (u32 limit, ip6_address_t * src, int do_flush)
+nat_ipfix_logging_max_frag_ip6 (u32 thread_index,
+                                u32 limit, ip6_address_t * src, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -1153,14 +1206,12 @@ nat_ipfix_logging_max_frag_ip6 (u32 limit, ip6_address_t * src, int do_flush)
   u64 now;
   u8 nat_event = QUOTA_EXCEEDED;
   u32 quota_event = MAX_FRAGMENTS_PENDING_REASSEMBLY;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->max_frags_ip6_buffer;
+  b0 = sitd->max_frags_ip6_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -1173,22 +1224,22 @@ nat_ipfix_logging_max_frag_ip6 (u32 limit, ip6_address_t * src, int do_flush)
 	  return;
 	}
 
-      b0 = silm->max_frags_ip6_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->max_frags_ip6_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->max_frags_ip6_next_record_offset;
+      offset = sitd->max_frags_ip6_next_record_offset;
     }
 
-  f = silm->max_frags_ip6_frame;
+  f = sitd->max_frags_ip6_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->max_frags_ip6_frame = f;
+      sitd->max_frags_ip6_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -1221,20 +1272,25 @@ nat_ipfix_logging_max_frag_ip6 (u32 limit, ip6_address_t * src, int do_flush)
   if (PREDICT_FALSE
       (do_flush || (offset + MAX_BIBS_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->max_frags_ip6_template_id);
-      silm->max_frags_ip6_frame = 0;
-      silm->max_frags_ip6_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->max_frags_ip6_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->max_frags_ip6_frame = 0;
+      sitd->max_frags_ip6_buffer = 0;
       offset = 0;
     }
-  silm->max_frags_ip6_next_record_offset = offset;
+  sitd->max_frags_ip6_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_nat64_bibe (u8 nat_event, ip6_address_t * src_ip,
-                              u32 nat_src_ip, u8 proto, u16 src_port,
-                              u16 nat_src_port, u32 vrf_id, int do_flush)
+nat_ipfix_logging_nat64_bibe (u32 thread_index, u8 nat_event,
+                              ip6_address_t * src_ip, u32 nat_src_ip,
+                              u8 proto, u16 src_port, u16 nat_src_port,
+                              u32 vrf_id, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -1242,14 +1298,12 @@ nat_ipfix_logging_nat64_bibe (u8 nat_event, ip6_address_t * src_ip,
   u32 offset;
   vlib_main_t *vm = frm->vlib_main;
   u64 now;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->nat64_bib_buffer;
+  b0 = sitd->nat64_bib_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -1262,22 +1316,22 @@ nat_ipfix_logging_nat64_bibe (u8 nat_event, ip6_address_t * src_ip,
 	  return;
 	}
 
-      b0 = silm->nat64_bib_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->nat64_bib_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->nat64_bib_next_record_offset;
+      offset = sitd->nat64_bib_next_record_offset;
     }
 
-  f = silm->nat64_bib_frame;
+  f = sitd->nat64_bib_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->nat64_bib_frame = f;
+      sitd->nat64_bib_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -1319,22 +1373,27 @@ nat_ipfix_logging_nat64_bibe (u8 nat_event, ip6_address_t * src_ip,
   if (PREDICT_FALSE
       (do_flush || (offset + NAT64_BIB_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->nat64_bib_template_id);
-      silm->nat64_bib_frame = 0;
-      silm->nat64_bib_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->nat64_bib_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->nat64_bib_frame = 0;
+      sitd->nat64_bib_buffer = 0;
       offset = 0;
     }
-  silm->nat64_bib_next_record_offset = offset;
+  sitd->nat64_bib_next_record_offset = offset;
 }
 
 static void
-nat_ipfix_logging_nat64_ses (u8 nat_event, ip6_address_t * src_ip,
-                             u32 nat_src_ip, u8 proto, u16 src_port,
-                             u16 nat_src_port, ip6_address_t * dst_ip,
-                             u32 nat_dst_ip, u16 dst_port, u16 nat_dst_port,
+nat_ipfix_logging_nat64_ses (u32 thread_index, u8 nat_event,
+                             ip6_address_t * src_ip, u32 nat_src_ip,
+                             u8 proto, u16 src_port, u16 nat_src_port,
+                             ip6_address_t * dst_ip, u32 nat_dst_ip,
+                             u16 dst_port, u16 nat_dst_port,
                              u32 vrf_id, int do_flush)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  snat_ipfix_per_thread_data_t *sitd = &silm->per_thread_data[thread_index];
   flow_report_main_t *frm = &flow_report_main;
   vlib_frame_t *f;
   vlib_buffer_t *b0 = 0;
@@ -1342,14 +1401,12 @@ nat_ipfix_logging_nat64_ses (u8 nat_event, ip6_address_t * src_ip,
   u32 offset;
   vlib_main_t *vm = frm->vlib_main;
   u64 now;
-
-  if (!silm->enabled)
-    return;
+  u16 template_id;
 
   now = (u64) ((vlib_time_now (vm) - silm->vlib_time_0) * 1e3);
   now += silm->milisecond_time_0;
 
-  b0 = silm->nat64_ses_buffer;
+  b0 = sitd->nat64_ses_buffer;
 
   if (PREDICT_FALSE (b0 == 0))
     {
@@ -1362,22 +1419,22 @@ nat_ipfix_logging_nat64_ses (u8 nat_event, ip6_address_t * src_ip,
 	  return;
 	}
 
-      b0 = silm->nat64_ses_buffer = vlib_get_buffer (vm, bi0);
+      b0 = sitd->nat64_ses_buffer = vlib_get_buffer (vm, bi0);
       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
       offset = 0;
     }
   else
     {
       bi0 = vlib_get_buffer_index (vm, b0);
-      offset = silm->nat64_ses_next_record_offset;
+      offset = sitd->nat64_ses_next_record_offset;
     }
 
-  f = silm->nat64_ses_frame;
+  f = sitd->nat64_ses_frame;
   if (PREDICT_FALSE (f == 0))
     {
       u32 *to_next;
       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
-      silm->nat64_ses_frame = f;
+      sitd->nat64_ses_frame = f;
       to_next = vlib_frame_vector_args (f);
       to_next[0] = bi0;
       f->n_vectors = 1;
@@ -1431,25 +1488,73 @@ nat_ipfix_logging_nat64_ses (u8 nat_event, ip6_address_t * src_ip,
   if (PREDICT_FALSE
       (do_flush || (offset + NAT64_SES_LEN) > frm->path_mtu))
     {
-      snat_ipfix_send (frm, f, b0, silm->nat64_ses_template_id);
-      silm->nat64_ses_frame = 0;
-      silm->nat64_ses_buffer = 0;
+      template_id = clib_atomic_fetch_or (
+        &silm->nat64_ses_template_id,
+        0);
+      snat_ipfix_send (frm, f, b0, template_id);
+      sitd->nat64_ses_frame = 0;
+      sitd->nat64_ses_buffer = 0;
       offset = 0;
     }
-  silm->nat64_ses_next_record_offset = offset;
+  sitd->nat64_ses_next_record_offset = offset;
 }
 
-static void
-snat_ipfix_logging_nat44_ses_rpc_cb (snat_ipfix_logging_nat44_ses_args_t * a)
+void
+snat_ipfix_flush (u32 thread_index)
 {
-  snat_ipfix_logging_nat44_ses (a->nat_event, a->src_ip, a->nat_src_ip,
-				a->snat_proto, a->src_port, a->nat_src_port,
-				a->vrf_id, 0);
+  int do_flush = 1;
+
+  snat_ipfix_logging_nat44_ses (thread_index,
+                                0, 0, 0, 0, 0, 0, 0, do_flush);
+  snat_ipfix_logging_addr_exhausted (thread_index, 0, do_flush);
+  snat_ipfix_logging_max_entries_per_usr (thread_index, 0, 0, do_flush);
+  nat_ipfix_logging_max_ses (thread_index, 0, do_flush);
+  nat_ipfix_logging_max_bib (thread_index, 0, do_flush);
+  nat_ipfix_logging_max_frag_ip4 (thread_index, 0, 0, do_flush);
+  nat_ipfix_logging_max_frag_ip6 (thread_index, 0, 0, do_flush);
+  nat_ipfix_logging_nat64_bibe (thread_index,
+                                0, 0, 0, 0, 0, 0, 0, do_flush);
+  nat_ipfix_logging_nat64_ses (thread_index,
+                               0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, do_flush);
+}
+
+void
+snat_ipfix_flush_from_main (void)
+{
+  snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  vlib_main_t *worker_vm;
+  int i;
+
+  if (PREDICT_TRUE (!clib_atomic_fetch_or(&silm->enabled, 0)))
+    return;
+
+  if (PREDICT_FALSE (!silm->worker_vms))
+    {
+      for (i = 1; i < vec_len (vlib_mains); i++)
+        {
+          worker_vm = vlib_mains[i];
+          if (worker_vm)
+            vec_add1 (silm->worker_vms, worker_vm);
+        }
+    }
+
+  /* Trigger flush for each worker thread */
+  for (i = 1; i < vec_len (silm->worker_vms); i++)
+    {
+      worker_vm = silm->worker_vms[i];
+      if (worker_vm)
+	vlib_node_set_interrupt_pending (worker_vm,
+					 snat_ipfix_flush_node.index);
+    }
+
+  /* Finally flush main thread */
+  snat_ipfix_flush (0);
 }
 
 /**
  * @brief Generate NAT44 session create event
  *
+ * @param thread_index thread index
  * @param src_ip       source IPv4 address
  * @param nat_src_ip   transaltes source IPv4 address
  * @param snat_proto   NAT transport protocol
@@ -1458,31 +1563,24 @@ snat_ipfix_logging_nat44_ses_rpc_cb (snat_ipfix_logging_nat44_ses_args_t * a)
  * @param vrf_id       VRF ID
  */
 void
-snat_ipfix_logging_nat44_ses_create (u32 src_ip,
+snat_ipfix_logging_nat44_ses_create (u32 thread_index,
+                                     u32 src_ip,
 				     u32 nat_src_ip,
 				     snat_protocol_t snat_proto,
 				     u16 src_port,
 				     u16 nat_src_port, u32 vrf_id)
 {
-  snat_ipfix_logging_nat44_ses_args_t a;
-
   skip_if_disabled ();
 
-  a.nat_event = NAT44_SESSION_CREATE;
-  a.src_ip = src_ip;
-  a.nat_src_ip = nat_src_ip;
-  a.snat_proto = snat_proto;
-  a.src_port = src_port;
-  a.nat_src_port = nat_src_port;
-  a.vrf_id = vrf_id;
-
-  vl_api_rpc_call_main_thread (snat_ipfix_logging_nat44_ses_rpc_cb,
-			       (u8 *) & a, sizeof (a));
+  snat_ipfix_logging_nat44_ses (thread_index, NAT44_SESSION_CREATE, src_ip,
+                                nat_src_ip, snat_proto, src_port, nat_src_port,
+				vrf_id, 0);
 }
 
 /**
  * @brief Generate NAT44 session delete event
  *
+ * @param thread_index thread index
  * @param src_ip       source IPv4 address
  * @param nat_src_ip   transaltes source IPv4 address
  * @param snat_proto   NAT transport protocol
@@ -1491,274 +1589,131 @@ snat_ipfix_logging_nat44_ses_create (u32 src_ip,
  * @param vrf_id       VRF ID
  */
 void
-snat_ipfix_logging_nat44_ses_delete (u32 src_ip,
+snat_ipfix_logging_nat44_ses_delete (u32 thread_index,
+                                     u32 src_ip,
 				     u32 nat_src_ip,
 				     snat_protocol_t snat_proto,
 				     u16 src_port,
 				     u16 nat_src_port, u32 vrf_id)
 {
-  snat_ipfix_logging_nat44_ses_args_t a;
-
   skip_if_disabled ();
 
-  a.nat_event = NAT44_SESSION_DELETE;
-  a.src_ip = src_ip;
-  a.nat_src_ip = nat_src_ip;
-  a.snat_proto = snat_proto;
-  a.src_port = src_port;
-  a.nat_src_port = nat_src_port;
-  a.vrf_id = vrf_id;
-
-  vl_api_rpc_call_main_thread (snat_ipfix_logging_nat44_ses_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-snat_data_callback_nat44_session (flow_report_main_t * frm,
-				  flow_report_t * fr,
-				  vlib_frame_t * f,
-				  u32 * to_next, u32 node_index)
-{
-  snat_ipfix_logging_nat44_ses (0, 0, 0, 0, 0, 0, 0, 1);
-  return f;
-}
-
-static void
-  snat_ipfix_logging_addr_exhausted_rpc_cb
-  (snat_ipfix_logging_addr_exhausted_args_t * a)
-{
-  snat_ipfix_logging_addr_exhausted (a->pool_id, 0);
+  snat_ipfix_logging_nat44_ses (thread_index, NAT44_SESSION_DELETE, src_ip,
+                                nat_src_ip, snat_proto, src_port, nat_src_port,
+				vrf_id, 0);
 }
 
 /**
  * @brief Generate NAT addresses exhausted event
  *
+ * @param thread_index thread index
  * @param pool_id NAT pool ID
  */
 void
-snat_ipfix_logging_addresses_exhausted (u32 pool_id)
+snat_ipfix_logging_addresses_exhausted (u32 thread_index, u32 pool_id)
 {
   //TODO: This event SHOULD be rate limited
-  snat_ipfix_logging_addr_exhausted_args_t a;
-
   skip_if_disabled ();
 
-  a.pool_id = pool_id;
-
-  vl_api_rpc_call_main_thread (snat_ipfix_logging_addr_exhausted_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-snat_data_callback_addr_exhausted (flow_report_main_t * frm,
-				   flow_report_t * fr,
-				   vlib_frame_t * f,
-				   u32 * to_next, u32 node_index)
-{
-  snat_ipfix_logging_addr_exhausted (0, 1);
-  return f;
-}
-
-static void
-  snat_ipfix_logging_max_entries_per_usr_rpc_cb
-  (snat_ipfix_logging_max_entries_per_user_args_t * a)
-{
-  snat_ipfix_logging_max_entries_per_usr (a->limit, a->src_ip, 0);
+  snat_ipfix_logging_addr_exhausted (thread_index, pool_id, 0);
 }
 
 /**
  * @brief Generate maximum entries per user exceeded event
  *
+ * @param thread_index thread index
  * @param limit maximum NAT entries that can be created per user
  * @param src_ip source IPv4 address
  */
 void
-snat_ipfix_logging_max_entries_per_user (u32 limit, u32 src_ip)
+snat_ipfix_logging_max_entries_per_user (u32 thread_index, u32 limit, u32 src_ip)
 {
   //TODO: This event SHOULD be rate limited
-  snat_ipfix_logging_max_entries_per_user_args_t a;
-
   skip_if_disabled ();
 
-  a.limit = limit;
-  a.src_ip = src_ip;
-
-  vl_api_rpc_call_main_thread (snat_ipfix_logging_max_entries_per_usr_rpc_cb,
-			       (u8 *) & a, sizeof (a));
+  snat_ipfix_logging_max_entries_per_usr (thread_index, limit, src_ip, 0);
 }
 
 vlib_frame_t *
-snat_data_callback_max_entries_per_usr (flow_report_main_t * frm,
+deterministic_nat_data_callback
+(flow_report_main_t * frm,
 					flow_report_t * fr,
 					vlib_frame_t * f,
 					u32 * to_next, u32 node_index)
 {
-  snat_ipfix_logging_max_entries_per_usr (0, 0, 1);
-  return f;
-}
+  snat_ipfix_flush_from_main();
 
-static void
-nat_ipfix_logging_max_ses_rpc_cb (nat_ipfix_logging_max_sessions_args_t * a)
-{
-  nat_ipfix_logging_max_ses (a->limit, 0);
+  return f;
 }
 
 /**
  * @brief Generate maximum session entries exceeded event
  *
+ * @param thread_index thread index
  * @param limit configured limit
  */
 void
-nat_ipfix_logging_max_sessions (u32 limit)
+nat_ipfix_logging_max_sessions (u32 thread_index, u32 limit)
 {
   //TODO: This event SHOULD be rate limited
-  nat_ipfix_logging_max_sessions_args_t a;
-
   skip_if_disabled ();
 
-  a.limit = limit;
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_max_ses_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-nat_data_callback_max_sessions (flow_report_main_t * frm,
-				flow_report_t * fr,
-				vlib_frame_t * f,
-				u32 * to_next, u32 node_index)
-{
-  nat_ipfix_logging_max_ses (0, 1);
-  return f;
-}
-
-static void
-nat_ipfix_logging_max_bib_rpc_cb (nat_ipfix_logging_max_bibs_args_t * a)
-{
-  nat_ipfix_logging_max_bib (a->limit, 0);
+  nat_ipfix_logging_max_ses (thread_index, limit, 0);
 }
 
 /**
  * @brief Generate maximum BIB entries exceeded event
  *
+ * @param thread_index thread index
  * @param limit configured limit
  */
 void
-nat_ipfix_logging_max_bibs (u32 limit)
+nat_ipfix_logging_max_bibs (u32 thread_index, u32 limit)
 {
   //TODO: This event SHOULD be rate limited
-  nat_ipfix_logging_max_bibs_args_t a;
-
   skip_if_disabled ();
 
-  a.limit = limit;
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_max_bib_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-nat_data_callback_max_bibs (flow_report_main_t * frm,
-			    flow_report_t * fr,
-			    vlib_frame_t * f,
-			    u32 * to_next, u32 node_index)
-{
-  nat_ipfix_logging_max_bib (0, 1);
-  return f;
-}
-
-static void
-nat_ipfix_logging_max_frag_ip4_rpc_cb (nat_ipfix_logging_max_frags_ip4_args_t * a)
-{
-  nat_ipfix_logging_max_frag_ip4 (a->limit, a->src, 0);
+  nat_ipfix_logging_max_bib (thread_index, limit, 0);
 }
 
 /**
  * @brief Generate maximum IPv4 fragments pending reassembly exceeded event
  *
+ * @param thread_index thread index
  * @param limit configured limit
  * @param src source IPv4 address
  */
 void
-nat_ipfix_logging_max_fragments_ip4 (u32 limit, ip4_address_t * src)
+nat_ipfix_logging_max_fragments_ip4 (u32 thread_index,
+                                     u32 limit, ip4_address_t * src)
 {
   //TODO: This event SHOULD be rate limited
-  nat_ipfix_logging_max_frags_ip4_args_t a;
-
   skip_if_disabled ();
 
-  a.limit = limit;
-  a.src = src->as_u32;
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_max_frag_ip4_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-nat_data_callback_max_frags_ip4 (flow_report_main_t * frm,
-			         flow_report_t * fr,
-			         vlib_frame_t * f,
-			         u32 * to_next, u32 node_index)
-{
-  nat_ipfix_logging_max_frag_ip4 (0, 0, 1);
-  return f;
-}
-
-static void
-nat_ipfix_logging_max_frag_ip6_rpc_cb (nat_ipfix_logging_max_frags_ip6_args_t * a)
-{
-  ip6_address_t src;
-  src.as_u64[0] = a->src[0];
-  src.as_u64[1] = a->src[1];
-  nat_ipfix_logging_max_frag_ip6 (a->limit, &src, 0);
+  nat_ipfix_logging_max_frag_ip4 (thread_index, limit, src->as_u32, 0);
 }
 
 /**
  * @brief Generate maximum IPv6 fragments pending reassembly exceeded event
  *
+ * @param thread_index thread index
  * @param limit configured limit
  * @param src source IPv6 address
  */
 void
-nat_ipfix_logging_max_fragments_ip6 (u32 limit, ip6_address_t * src)
+nat_ipfix_logging_max_fragments_ip6 (u32 thread_index,
+                                     u32 limit, ip6_address_t * src)
 {
   //TODO: This event SHOULD be rate limited
-  nat_ipfix_logging_max_frags_ip6_args_t a;
-
   skip_if_disabled ();
 
-  a.limit = limit;
-  a.src[0] = src->as_u64[0];
-  a.src[1] = src->as_u64[1];
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_max_frag_ip6_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
-
-vlib_frame_t *
-nat_data_callback_max_frags_ip6 (flow_report_main_t * frm,
-			         flow_report_t * fr,
-			         vlib_frame_t * f,
-			         u32 * to_next, u32 node_index)
-{
-  nat_ipfix_logging_max_frag_ip6 (0, 0, 1);
-  return f;
-}
-
-static void
-nat_ipfix_logging_nat64_bib_rpc_cb (nat_ipfix_logging_nat64_bib_args_t * a)
-{
-  ip6_address_t src_ip;
-  src_ip.as_u64[0] = a->src_ip[0];
-  src_ip.as_u64[1] = a->src_ip[1];
-  nat_ipfix_logging_nat64_bibe (a->nat_event, &src_ip, a->nat_src_ip,
-                                a->proto, a->src_port, a->nat_src_port,
-                                a->vrf_id, 0);
+  nat_ipfix_logging_max_frag_ip6 (thread_index, limit, src, 0);
 }
 
 /**
  * @brief Generate NAT64 BIB create and delete events
  *
+ * @param thread_index thread index
  * @param src_ip       source IPv6 address
  * @param nat_src_ip   transaltes source IPv4 address
  * @param proto        L4 protocol
@@ -1768,55 +1723,26 @@ nat_ipfix_logging_nat64_bib_rpc_cb (nat_ipfix_logging_nat64_bib_args_t * a)
  * @param is_create    non-zero value if create event otherwise delete event
  */
 void
-nat_ipfix_logging_nat64_bib (ip6_address_t * src_ip,
+nat_ipfix_logging_nat64_bib (u32 thread_index, ip6_address_t * src_ip,
                              ip4_address_t * nat_src_ip, u8 proto,
                              u16 src_port, u16 nat_src_port, u32 vrf_id,
                              u8 is_create)
 {
-  nat_ipfix_logging_nat64_bib_args_t a;
+  u8 nat_event;
 
   skip_if_disabled ();
 
-  a.src_ip[0] = src_ip->as_u64[0];
-  a.src_ip[1] = src_ip->as_u64[1];
-  a.nat_src_ip = nat_src_ip->as_u32;
-  a.proto = proto;
-  a.src_port = src_port;
-  a.nat_src_port = nat_src_port;
-  a.vrf_id = vrf_id;
-  a.nat_event = is_create ? NAT64_BIB_CREATE : NAT64_BIB_DELETE;
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_nat64_bib_rpc_cb,
-			       (u8 *) & a, sizeof (a));
-}
+  nat_event = is_create ? NAT64_BIB_CREATE : NAT64_BIB_DELETE;
 
-vlib_frame_t *
-nat_data_callback_nat64_bib (flow_report_main_t * frm,
-			     flow_report_t * fr,
-			     vlib_frame_t * f,
-			     u32 * to_next, u32 node_index)
-{
-  nat_ipfix_logging_nat64_bibe (0, 0, 0, 0, 0, 0, 0, 1);
-  return f;
-}
-
-static void
-nat_ipfix_logging_nat64_ses_rpc_cb (nat_ipfix_logging_nat64_ses_args_t * a)
-{
-  ip6_address_t src_ip, dst_ip;
-  src_ip.as_u64[0] = a->src_ip[0];
-  src_ip.as_u64[1] = a->src_ip[1];
-  dst_ip.as_u64[0] = a->dst_ip[0];
-  dst_ip.as_u64[1] = a->dst_ip[1];
-  nat_ipfix_logging_nat64_ses (a->nat_event, &src_ip, a->nat_src_ip,
-                               a->proto, a->src_port, a->nat_src_port,
-                               &dst_ip, a->nat_dst_ip, a->dst_port,
-                               a->nat_dst_port, a->vrf_id, 0);
+  nat_ipfix_logging_nat64_bibe (thread_index, nat_event, src_ip,
+                                nat_src_ip->as_u32, proto, src_port,
+                                nat_src_port, vrf_id, 0);
 }
 
 /**
  * @brief Generate NAT64 session create and delete events
  *
+ * @param thread_index thread index
  * @param src_ip       source IPv6 address
  * @param nat_src_ip   transaltes source IPv4 address
  * @param proto        L4 protocol
@@ -1830,42 +1756,38 @@ nat_ipfix_logging_nat64_ses_rpc_cb (nat_ipfix_logging_nat64_ses_args_t * a)
  * @param is_create    non-zero value if create event otherwise delete event
  */
 void
-nat_ipfix_logging_nat64_session (ip6_address_t * src_ip,
+nat_ipfix_logging_nat64_session (u32 thread_index,
+                                 ip6_address_t * src_ip,
                                  ip4_address_t * nat_src_ip, u8 proto,
                                  u16 src_port, u16 nat_src_port,
                                  ip6_address_t * dst_ip,
                                  ip4_address_t * nat_dst_ip, u16 dst_port,
                                  u16 nat_dst_port, u32 vrf_id, u8 is_create)
 {
-  nat_ipfix_logging_nat64_ses_args_t a;
+  u8 nat_event;
 
   skip_if_disabled ();
 
-  a.src_ip[0] = src_ip->as_u64[0];
-  a.src_ip[1] = src_ip->as_u64[1];
-  a.nat_src_ip = nat_src_ip->as_u32;
-  a.proto = proto;
-  a.src_port = src_port;
-  a.nat_src_port = nat_src_port;
-  a.dst_ip[0] = dst_ip->as_u64[0];
-  a.dst_ip[1] = dst_ip->as_u64[1];
-  a.nat_dst_ip = nat_dst_ip->as_u32;
-  a.dst_port = dst_port;
-  a.nat_dst_port = nat_dst_port;
-  a.vrf_id = vrf_id;
-  a.nat_event = is_create ? NAT64_SESSION_CREATE : NAT64_SESSION_DELETE;
-
-  vl_api_rpc_call_main_thread (nat_ipfix_logging_nat64_ses_rpc_cb,
-			       (u8 *) & a, sizeof (a));
+  nat_event = is_create ? NAT64_SESSION_CREATE : NAT64_SESSION_DELETE;
+
+  nat_ipfix_logging_nat64_ses (thread_index, nat_event, src_ip,
+                               nat_src_ip->as_u32, proto, src_port,
+                               nat_src_port, dst_ip, nat_dst_ip->as_u32,
+                               dst_port, nat_dst_port, vrf_id, 0);
 }
 
 vlib_frame_t *
-nat_data_callback_nat64_session (flow_report_main_t * frm,
-			         flow_report_t * fr,
-			         vlib_frame_t * f,
-			         u32 * to_next, u32 node_index)
+data_callback (flow_report_main_t * frm, flow_report_t * fr,
+               vlib_frame_t * f, u32 * to_next, u32 node_index)
 {
-  nat_ipfix_logging_nat64_ses (0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1);
+  snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+
+  if (PREDICT_FALSE (++silm->call_counter >= vec_len (frm->reports)))
+    {
+      snat_ipfix_flush_from_main();
+      silm->call_counter = 0;
+    }
+
   return f;
 }
 
@@ -1888,20 +1810,18 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
   int rv;
   u8 e = enable ? 1 : 0;
 
-  if (silm->enabled == e)
+  if (clib_atomic_cmp_and_swap (&silm->enabled, e ^ 1, e) == e)
     return 0;
 
-  silm->enabled = e;
-
   clib_memset (&a, 0, sizeof (a));
   a.is_add = enable;
   a.domain_id = domain_id ? domain_id : 1;
   a.src_port = src_port ? src_port : UDP_DST_PORT_ipfix;
+  a.flow_data_callback = data_callback;
 
   if (sm->deterministic)
     {
       a.rewrite_callback = snat_template_rewrite_max_entries_per_usr;
-      a.flow_data_callback = snat_data_callback_max_entries_per_usr;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1913,7 +1833,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
   else
     {
       a.rewrite_callback = snat_template_rewrite_nat44_session;
-      a.flow_data_callback = snat_data_callback_nat44_session;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1923,7 +1842,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = snat_template_rewrite_addr_exhausted;
-      a.flow_data_callback = snat_data_callback_addr_exhausted;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1933,7 +1851,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_max_sessions;
-      a.flow_data_callback = nat_data_callback_max_sessions;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1943,7 +1860,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_max_bibs;
-      a.flow_data_callback = nat_data_callback_max_bibs;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1953,7 +1869,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_max_frags_ip4;
-      a.flow_data_callback = nat_data_callback_max_frags_ip4;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1963,7 +1878,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_max_frags_ip6;
-      a.flow_data_callback = nat_data_callback_max_frags_ip6;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1973,7 +1887,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_nat64_bib;
-      a.flow_data_callback = nat_data_callback_nat64_bib;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1983,7 +1896,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
 	}
 
       a.rewrite_callback = nat_template_rewrite_nat64_session;
-      a.flow_data_callback = nat_data_callback_nat64_session;
 
       rv = vnet_flow_report_add_del (frm, &a, NULL);
       if (rv)
@@ -1995,7 +1907,6 @@ snat_ipfix_logging_enable_disable (int enable, u32 domain_id, u16 src_port)
       if (sm->endpoint_dependent)
         {
           a.rewrite_callback = snat_template_rewrite_max_entries_per_usr;
-          a.flow_data_callback = snat_data_callback_max_entries_per_usr;
 
           rv = vnet_flow_report_add_del (frm, &a, NULL);
           if (rv)
@@ -2018,10 +1929,33 @@ void
 snat_ipfix_logging_init (vlib_main_t * vm)
 {
   snat_ipfix_logging_main_t *silm = &snat_ipfix_logging_main;
+  vlib_thread_main_t *tm = vlib_get_thread_main ();
 
   silm->enabled = 0;
+  silm->worker_vms = 0;
+  silm->call_counter = 0;
 
   /* Set up time reference pair */
   silm->vlib_time_0 = vlib_time_now (vm);
   silm->milisecond_time_0 = unix_time_now_nsec () * 1e-6;
+
+  vec_validate (silm->per_thread_data, tm->n_vlib_mains - 1);
 }
+
+static uword
+ipfix_flush_process (vlib_main_t *vm,
+		     vlib_node_runtime_t *rt,
+                     vlib_frame_t *f)
+{
+  snat_ipfix_flush(vm->thread_index);
+  return 0;
+}
+
+/* *INDENT-OFF* */
+VLIB_REGISTER_NODE (snat_ipfix_flush_node) = {
+  .function = ipfix_flush_process,
+  .name = "snat-ipfix-flush",
+  .type = VLIB_NODE_TYPE_INPUT,
+  .state = VLIB_NODE_STATE_INTERRUPT,
+};
+/* *INDENT-ON* */