fix(cpworker): count every output packet once, with its sliced length

cpworker/src/output.h

@@ -27,6 +27,16 @@ typedef struct OutputEntry
     OutputFactory factory;
 } output_entry_t;
 
+// Length of the frame an output carries and counts: the captured length after slice, capped at
+// the largest frame the output can carry. Every output counter and rate limiter uses this length.
+static inline size_t output_frame_len(uint32_t caplen, int slice, size_t max_len)
+{
+    size_t frame_len = caplen;
+    if (slice > 0 && (uint32_t)slice < caplen)
+        frame_len = (size_t)slice;
+    return frame_len < max_len ? frame_len : max_len;
+}
+
 static inline int output_send_packet(output_base_t *output, const struct pcap_pkthdr *header, const uint8_t *pkt_data,
                                      int direct)
 {

cpworker/src/output_gre.c

@@ -42,34 +42,28 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const
 {
     gre_output_t *output = (gre_output_t *)self;
 
-    int32_t caplen = header->caplen;
-    if (output->slice > 0 && output->slice < caplen)
-    {
-        caplen = output->slice;
-    }
-
-    size_t length = (size_t)(caplen <= 65535 ? caplen : 65535);
+    const size_t frame_len = output_frame_len(header->caplen, output->slice, GRE_MAX_FRAME_LEN);
 
     if (direct == PKT_DIR_UNKNOWN)
     {
-        bytes_stats_add(&output->base.stats->direction_drop_bytes, length);
+        bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len);
         packets_stats_add(&output->base.stats->direction_drop_packets, 1);
         return -1;
     }
 
     if (output->rate_limit_mbps > 0)
     {
-        if (!token_bucket_consume(&output->throttle, GRE_HEADER_LEN + length, header->ts))
+        if (!token_bucket_consume(&output->throttle, frame_len, header->ts))
         {
-            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, GRE_HEADER_LEN + length);
+            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len);
             packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1);
             return -1;
         }
     }
 
     struct gre_header *gre_hdr = (struct gre_header *)output->buf;
     gre_hdr->keybit = htonl(output->service_tag | (direct << 28));
-    memcpy(&(output->buf[GRE_HEADER_LEN]), pkt_data, length);
+    memcpy(&(output->buf[GRE_HEADER_LEN]), pkt_data, frame_len);
 
     // check error_info
     if (output->error_info.first_pktsec == 0)
@@ -84,7 +78,7 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const
     const int max_retries = 10;
     do
     {
-        ssize_t send_bytes = sendto(output->socket_fd, output->buf, GRE_HEADER_LEN + length, 0,
+        ssize_t send_bytes = sendto(output->socket_fd, output->buf, GRE_HEADER_LEN + frame_len, 0,
                                     (struct sockaddr *)&output->remote_addr, sizeof(struct sockaddr_in));
 
         if (send_bytes == -1)
@@ -112,22 +106,21 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const
                 output->error_info.nb_other_send_error_drops++;
             }
 
-            bytes_stats_add(&output->base.stats->error_drop_bytes, GRE_HEADER_LEN + length);
+            bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len);
             packets_stats_add(&output->base.stats->error_drop_packets, 1);
             return -1;
         }
 
-        if (send_bytes < GRE_HEADER_LEN + length)
+        if (send_bytes < GRE_HEADER_LEN + frame_len)
         {
             output->error_info.nb_partial_sends++;
 
-            bytes_stats_add(&output->base.stats->error_drop_bytes, GRE_HEADER_LEN + length - send_bytes);
-            bytes_stats_add(&output->base.stats->fwd_bytes, send_bytes);
-            packets_stats_add(&output->base.stats->fwd_packets, 1);
+            bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len);
+            packets_stats_add(&output->base.stats->error_drop_packets, 1);
             return -1;
         }
 
-        bytes_stats_add(&output->base.stats->fwd_bytes, GRE_HEADER_LEN + length);
+        bytes_stats_add(&output->base.stats->fwd_bytes, frame_len);
         packets_stats_add(&output->base.stats->fwd_packets, 1);
         return 0;
     } while (true);

cpworker/src/output_gre.h

@@ -9,7 +9,8 @@
 #include "output.h"
 #include "ratelimit.h"
 
-#define GRE_OUTPUT_BUFSIZE 65551 // 8(GRE_HEADER_LEN) + 65535
+#define GRE_MAX_FRAME_LEN 65535
+#define GRE_OUTPUT_BUFSIZE 65551 // 8(GRE_HEADER_LEN) + 65535(GRE_MAX_FRAME_LEN)
 
 typedef struct GreOptions
 {

cpworker/src/output_null.c

@@ -1,3 +1,4 @@
+#include <stdint.h>
 #include <stdlib.h>
 
 #include "errorf.h"
@@ -10,29 +11,25 @@ int null_send_packet(output_base_t *self, const struct pcap_pkthdr *header, cons
 {
     null_output_t *output = (null_output_t *)self;
 
-    int32_t length = header->caplen;
-    if (output->slice > 0 && output->slice < length)
-    {
-        length = output->slice;
-    }
+    const size_t frame_len = output_frame_len(header->caplen, output->slice, SIZE_MAX);
 
     if (direct == PKT_DIR_UNKNOWN)
     {
-        bytes_stats_add(&output->base.stats->direction_drop_bytes, length);
+        bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len);
         packets_stats_add(&output->base.stats->direction_drop_packets, 1);
         return -1;
     }
 
     if (output->rate_limit_mbps > 0)
     {
-        if (!token_bucket_consume(&output->throttle, length, header->ts))
+        if (!token_bucket_consume(&output->throttle, frame_len, header->ts))
         {
-            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length);
+            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len);
             packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1);
             return -1;
         }
     }
-    bytes_stats_add(&output->base.stats->fwd_bytes, length);
+    bytes_stats_add(&output->base.stats->fwd_bytes, frame_len);
     packets_stats_add(&output->base.stats->fwd_packets, 1);
     return 0;
 }

cpworker/src/output_vxlan.c

@@ -93,21 +93,25 @@ static void flush_error_info(vxlan_output_t *output)
     }
 }
 
-static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const uint8_t *pkt_data, size_t length,
+// Sends one VXLAN datagram. Returns 0 when it was sent in full, -1 otherwise. The caller does the counting.
+static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const uint8_t *pkt_data, size_t frame_len,
                           int direct)
 {
     struct vxlan_header *vxlan_hdr = (struct vxlan_header *)output->buf;
-    memcpy(&(output->buf[VXLAN_HEADER_LEN]), pkt_data, length);
+    memcpy(&(output->buf[VXLAN_HEADER_LEN]), pkt_data, frame_len);
+
+    // VXLAN payload: the frame, followed by the capture time when enabled
+    size_t payload_len = frame_len;
 
     if (output->capture_time)
     {
         uint32_t tv_sec = htonl(ts.tv_sec);
         // 注意：通过libpcap获取的捕获时间精度为微秒，而数据包中附加的时间为纳秒，所以需要*1000
         uint32_t tv_nsec = htonl(ts.tv_usec * 1000);
-        memcpy(&(output->buf[VXLAN_HEADER_LEN + length]), &tv_sec, 4);
-        length += 4;
-        memcpy(&(output->buf[VXLAN_HEADER_LEN + length]), &tv_nsec, 4);
-        length += 4;
+        memcpy(&(output->buf[VXLAN_HEADER_LEN + payload_len]), &tv_sec, 4);
+        payload_len += 4;
+        memcpy(&(output->buf[VXLAN_HEADER_LEN + payload_len]), &tv_nsec, 4);
+        payload_len += 4;
     }
     if (output->vni_version == 1)
     {
@@ -132,7 +136,7 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const
     const int max_retries = 10;
     do
     {
-        ssize_t send_bytes = sendto(output->socket_fd, output->buf, VXLAN_HEADER_LEN + length, 0,
+        ssize_t send_bytes = sendto(output->socket_fd, output->buf, VXLAN_HEADER_LEN + payload_len, 0,
                                     (struct sockaddr *)&output->remote_addr, sizeof(struct sockaddr_in));
 
         if (send_bytes == -1)
@@ -159,52 +163,66 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const
 
                 output->error_info.nb_other_send_error_drops++;
             }
-
-            bytes_stats_add(&output->base.stats->error_drop_bytes, VXLAN_HEADER_LEN + length);
-            packets_stats_add(&output->base.stats->error_drop_packets, 1);
             return -1;
         }
 
-        if (send_bytes < VXLAN_HEADER_LEN + length)
+        if (send_bytes < VXLAN_HEADER_LEN + payload_len)
         {
             output->error_info.nb_partial_sends++;
-
-            bytes_stats_add(&output->base.stats->error_drop_bytes, VXLAN_HEADER_LEN + length - send_bytes);
-            bytes_stats_add(&output->base.stats->fwd_bytes, send_bytes);
-            packets_stats_add(&output->base.stats->fwd_packets, 1);
             return -1;
         }
 
-        bytes_stats_add(&output->base.stats->fwd_bytes, VXLAN_HEADER_LEN + length);
-        packets_stats_add(&output->base.stats->fwd_packets, 1);
         return 0;
     } while (true);
 }
 
-int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const uint8_t *pkt_data, int direct)
+// Sends the packet, split into fragments when configured. Returns 0 only when every datagram was sent in full.
+static int send_whole_or_split(vxlan_output_t *output, const struct pcap_pkthdr *header, const uint8_t *pkt_data,
+                               size_t frame_len, int direct)
 {
-    vxlan_output_t *output = (vxlan_output_t *)self;
+    // Fast path
+    if (output->split.max_payload_size <= 0 || frame_len <= (size_t)output->split.max_payload_size)
+        return do_send_packet(output, header->ts, pkt_data, frame_len, direct);
+
+    packet_parse_result_t parse_result;
+    if (!parse_packet(pkt_data, frame_len, &parse_result))
+        return do_send_packet(output, header->ts, pkt_data, frame_len, direct);
+
+    int fragment_count = calculate_fragment_count(&parse_result, output->split.max_payload_size);
+    if (fragment_count == 1)
+        return do_send_packet(output, header->ts, pkt_data, frame_len, direct);
 
-    int32_t caplen = header->caplen;
-    if (output->slice > 0 && output->slice < caplen)
+    for (int i = 0; i < fragment_count; i++)
     {
-        caplen = output->slice;
+        int frag_len = build_fragment(&parse_result, pkt_data, i, output->split.max_payload_size,
+                                      output->split.recalculate_checksum, output->split.fragment_buf);
+        if (frag_len <= 0)
+            return -1;
+
+        if (do_send_packet(output, header->ts, output->split.fragment_buf, frag_len, direct) != 0)
+            return -1;
     }
+    return 0;
+}
+
+int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const uint8_t *pkt_data, int direct)
+{
+    vxlan_output_t *output = (vxlan_output_t *)self;
 
-    size_t length = (size_t)(caplen <= 65535 ? caplen : 65535);
+    const size_t frame_len = output_frame_len(header->caplen, output->slice, VXLAN_MAX_FRAME_LEN);
 
     if (direct == PKT_DIR_UNKNOWN)
     {
-        bytes_stats_add(&output->base.stats->direction_drop_bytes, length);
+        bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len);
         packets_stats_add(&output->base.stats->direction_drop_packets, 1);
         return -1;
     }
 
     if (output->rate_limit_mbps > 0)
     {
-        if (!token_bucket_consume(&output->throttle, VXLAN_HEADER_LEN + length, header->ts))
+        if (!token_bucket_consume(&output->throttle, frame_len, header->ts))
         {
-            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, VXLAN_HEADER_LEN + length);
+            bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len);
             packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1);
             return -1;
         }
@@ -219,35 +237,16 @@ int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, con
         output->error_info.first_pktsec = header->ts.tv_sec;
     }
 
-    // Fast path
-    if (output->split.max_payload_size <= 0 || length <= (size_t)output->split.max_payload_size)
+    // A packet counts once, with its frame_len, whether or not it was split
+    if (send_whole_or_split(output, header, pkt_data, frame_len, direct) != 0)
     {
-        return do_send_packet(output, header->ts, pkt_data, length, direct);
-    }
-
-    packet_parse_result_t parse_result;
-    if (!parse_packet(pkt_data, caplen, &parse_result))
-    {
-        return do_send_packet(output, header->ts, pkt_data, length, direct);
-    }
-
-    int fragment_count = calculate_fragment_count(&parse_result, output->split.max_payload_size);
-    if (fragment_count == 1)
-    {
-        return do_send_packet(output, header->ts, pkt_data, length, direct);
+        bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len);
+        packets_stats_add(&output->base.stats->error_drop_packets, 1);
+        return -1;
     }
 
-    for (int i = 0; i < fragment_count; i++)
-    {
-        int frag_len = build_fragment(&parse_result, pkt_data, i, output->split.max_payload_size,
-                                      output->split.recalculate_checksum, output->split.fragment_buf);
-        if (frag_len <= 0)
-            return 0;
-
-        int ret = do_send_packet(output, header->ts, output->split.fragment_buf, frag_len, direct);
-        if (ret != 0)
-            return ret;
-    }
+    bytes_stats_add(&output->base.stats->fwd_bytes, frame_len);
+    packets_stats_add(&output->base.stats->fwd_packets, 1);
     return 0;
 }
 

cpworker/src/output_vxlan.h

@@ -11,7 +11,8 @@
 #include "output.h"
 #include "ratelimit.h"
 
-#define VXLAN_OUTPUT_BUFSIZE 65551 // 8(VXLAN_HEADER_LEN) + 65535 + 8(capture_time)
+#define VXLAN_MAX_FRAME_LEN 65535
+#define VXLAN_OUTPUT_BUFSIZE 65551 // 8(VXLAN_HEADER_LEN) + 65535(VXLAN_MAX_FRAME_LEN) + 8(capture_time)
 
 typedef struct VxlanOptions
 {
@@ -29,7 +30,7 @@ typedef struct VxlanOptions
     struct
     {
         uint16_t max_payload_size;
-        bool recalculate_checksum;  // Whether to recalculate checksums after splitting (default: false)
+        bool recalculate_checksum; // Whether to recalculate checksums after splitting (default: false)
     } split;
 } vxlan_options_t;