FragmentProtoDisassembler: use a job instead of a timer to wait for more input before submitting an

output packet. Provides a small but noticable speedup.

badvpn-client.8

@@ -60,8 +60,6 @@ badvpn-client \- VPN node daemon for the BadVPN peer-to-peer VPN system
 .br
 .RB "[" --otp " <blowfish/aes> <num> <num-warn>]"
 .br
-.RB "[" --fragmentation-latency " <milliseconds>]"
-.br
 .RE
 )
 .br
@@ -207,12 +205,6 @@ the sending peer sending it to the receiving peer via the server and the receivi
 it via the server. Note that one-time passwords are only useful if clients use TLS to connect to the
 server. The OTP option must match on all peers, except for num-warn.
 .TP
-.BR --fragmentation-latency " <milliseconds>"
-When using UDP transport, sets the maximum latency to sacrifice in order to pack frames into data
-packets more efficiently. If it is >=0, a timer of that many milliseconds is used to wait for further
-frames to put into an incomplete packet since the first chunk of the packet was written. If it is
-<0, packets are sent out immediately. Defaults to 0, which is the recommended setting.
-.TP
 .BR --peer-ssl
 When using TCP transport, enables TLS for data connections. Requires using TLS for server connection.
 For this to work, the peers must trust each others' cerificates, and the cerificates must grant the

client/DatagramPeerIO.c

@@ -130,7 +130,6 @@ int DatagramPeerIO_Init (
     int payload_mtu,
     int socket_mtu,
     struct spproto_security_params sp_params,
-    btime_t latency,
     int num_frames,
     PacketPassInterface *recv_userif,
     int otp_warning_count,
@@ -216,7 +215,7 @@ int DatagramPeerIO_Init (
     // init sending base
     
     // init disassembler
-    FragmentProtoDisassembler_Init(&o->send_disassembler, o->reactor, o->payload_mtu, o->spproto_payload_mtu, -1, latency);
+    FragmentProtoDisassembler_Init(&o->send_disassembler, BReactor_PendingGroup(o->reactor), o->payload_mtu, o->spproto_payload_mtu, -1);
     
     // init encoder
     if (!SPProtoEncoder_Init(&o->send_encoder, FragmentProtoDisassembler_GetOutput(&o->send_disassembler), o->sp_params, otp_warning_count, BReactor_PendingGroup(o->reactor), twd)) {

client/DatagramPeerIO.h

@@ -135,7 +135,6 @@ typedef struct {
  *                   send a FragmentProto chunk with one byte of data over SPProto, i.e. the following has to hold:
  *                   spproto_payload_mtu_for_carrier_mtu(sp_params, socket_mtu) > sizeof(struct fragmentproto_chunk_header)
  * @param sp_params SPProto security parameters
- * @param latency latency parameter to {@link FragmentProtoDisassembler_Init}.
  * @param num_frames num_frames parameter to {@link FragmentProtoAssembler_Init}. Must be >0.
  * @param recv_userif interface to pass received packets to the user. Its MTU must be >=payload_mtu.
  * @param otp_warning_count If using OTPs, after how many encoded packets to call the handler.
@@ -154,7 +153,6 @@ int DatagramPeerIO_Init (
     int payload_mtu,
     int socket_mtu,
     struct spproto_security_params sp_params,
-    btime_t latency,
     int num_frames,
     PacketPassInterface *recv_userif,
     int otp_warning_count,

client/FragmentProtoDisassembler.c

@@ -61,6 +61,23 @@ static void write_chunks (FragmentProtoDisassembler *o)
         o->out_used += sizeof(struct fragmentproto_chunk_header) + chunk_len;
     } while (IN_AVAIL > 0 && OUT_AVAIL > 0);
     
+    // should we finish the output packet?
+    if (OUT_AVAIL <= 0) {
+        // set no output packet
+        o->out = NULL;
+        
+        // stop timer (if it's running)
+        BPending_Unset(&o->finish_job);
+        
+        // finish output
+        PacketRecvInterface_Done(&o->output, o->out_used);
+    } else {
+        // start finish job if we have more space in the output packet. This allows us to
+        // write more data into this output packet if the input has more data, before we
+        // submit it out in the job handler.
+        BPending_Set(&o->finish_job);
+    }
+    
     // have we finished the input packet?
     if (IN_AVAIL == 0) {
         // set no input packet
@@ -72,25 +89,6 @@ static void write_chunks (FragmentProtoDisassembler *o)
         // finish input
         PacketPassInterface_Done(&o->input);
     }
-    
-    // should we finish the output packet?
-    if (OUT_AVAIL <= 0 || o->latency < 0) {
-        // set no output packet
-        o->out = NULL;
-        
-        // stop timer (if it's running)
-        if (o->latency >= 0) {
-            BReactor_RemoveTimer(o->reactor, &o->timer);
-        }
-        
-        // finish output
-        PacketRecvInterface_Done(&o->output, o->out_used);
-    } else {
-        // start timer if we have output and it's not running (output was empty before)
-        if (!BTimer_IsRunning(&o->timer)) {
-            BReactor_SetTimer(o->reactor, &o->timer);
-        }
-    }
 }
 
 static void input_handler_send (FragmentProtoDisassembler *o, uint8_t *data, int data_len)
@@ -140,9 +138,8 @@ static void output_handler_recv (FragmentProtoDisassembler *o, uint8_t *data)
     write_chunks(o);
 }
 
-static void timer_handler (FragmentProtoDisassembler *o)
+static void finish_job_handler (FragmentProtoDisassembler *o)
 {
-    ASSERT(o->latency >= 0)
     ASSERT(o->out)
     ASSERT(o->in_len == -1)
     
@@ -153,7 +150,7 @@ static void timer_handler (FragmentProtoDisassembler *o)
     PacketRecvInterface_Done(&o->output, o->out_used);
 }
 
-void FragmentProtoDisassembler_Init (FragmentProtoDisassembler *o, BReactor *reactor, int input_mtu, int output_mtu, int chunk_mtu, btime_t latency)
+void FragmentProtoDisassembler_Init (FragmentProtoDisassembler *o, BPendingGroup *pg, int input_mtu, int output_mtu, int chunk_mtu)
 {
     ASSERT(input_mtu >= 0)
     ASSERT(input_mtu <= UINT16_MAX)
@@ -161,22 +158,18 @@ void FragmentProtoDisassembler_Init (FragmentProtoDisassembler *o, BReactor *rea
     ASSERT(chunk_mtu > 0 || chunk_mtu < 0)
     
     // init arguments
-    o->reactor = reactor;
     o->output_mtu = output_mtu;
     o->chunk_mtu = chunk_mtu;
-    o->latency = latency;
     
     // init input
-    PacketPassInterface_Init(&o->input, input_mtu, (PacketPassInterface_handler_send)input_handler_send, o, BReactor_PendingGroup(reactor));
+    PacketPassInterface_Init(&o->input, input_mtu, (PacketPassInterface_handler_send)input_handler_send, o, pg);
     PacketPassInterface_EnableCancel(&o->input, (PacketPassInterface_handler_requestcancel)input_handler_requestcancel);
     
     // init output
-    PacketRecvInterface_Init(&o->output, o->output_mtu, (PacketRecvInterface_handler_recv)output_handler_recv, o, BReactor_PendingGroup(reactor));
+    PacketRecvInterface_Init(&o->output, o->output_mtu, (PacketRecvInterface_handler_recv)output_handler_recv, o, pg);
     
-    // init timer
-    if (o->latency >= 0) {
-        BTimer_Init(&o->timer, o->latency, (BTimer_handler)timer_handler, o);
-    }
+    // init finish job
+    BPending_Init(&o->finish_job, pg, (BPending_handler)finish_job_handler, o);
     
     // have no input packet
     o->in_len = -1;
@@ -194,10 +187,8 @@ void FragmentProtoDisassembler_Free (FragmentProtoDisassembler *o)
 {
     DebugObject_Free(&o->d_obj);
 
-    // free timer
-    if (o->latency >= 0) {
-        BReactor_RemoveTimer(o->reactor, &o->timer);
-    }
+    // free finish job
+    BPending_Free(&o->finish_job);
     
     // free output
     PacketRecvInterface_Free(&o->output);

client/FragmentProtoDisassembler.h

@@ -32,8 +32,7 @@
 
 #include <protocol/fragmentproto.h>
 #include <base/DebugObject.h>
-#include <system/BReactor.h>
-#include <system/BTime.h>
+#include <base/BPending.h>
 #include <flow/PacketPassInterface.h>
 #include <flow/PacketRecvInterface.h>
 
@@ -45,13 +44,11 @@
  * Output is with {@link PacketRecvInterface}.
  */
 typedef struct {
-    BReactor *reactor;
     int output_mtu;
     int chunk_mtu;
-    btime_t latency;
     PacketPassInterface input;
     PacketRecvInterface output;
-    BTimer timer;
+    BPending finish_job;
     int in_len;
     uint8_t *in;
     int in_used;
@@ -65,16 +62,12 @@ typedef struct {
  * Initializes the object.
  *
  * @param o the object
- * @param reactor reactor we live in
+ * @param pg pending group we live in
  * @param input_mtu maximum input packet size. Must be >=0 and <=UINT16_MAX.
  * @param output_mtu maximum output packet size. Must be >sizeof(struct fragmentproto_chunk_header).
  * @param chunk_mtu maximum chunk size. Must be >0, or <0 for no explicit limit.
- * @param latency maximum time a pending output packet with some data can wait for more data
- *                before being sent out. If nonnegative, a timer will be used. If negative,
- *                packets will always be sent out immediately. If low latency is desired,
- *                prefer setting this to zero rather than negative.
  */
-void FragmentProtoDisassembler_Init (FragmentProtoDisassembler *o, BReactor *reactor, int input_mtu, int output_mtu, int chunk_mtu, btime_t latency);
+void FragmentProtoDisassembler_Init (FragmentProtoDisassembler *o, BPendingGroup *pg, int input_mtu, int output_mtu, int chunk_mtu);
 
 /**
  * Frees the object.

client/client.c

@@ -102,7 +102,6 @@ struct {
     int otp_mode;
     int otp_num;
     int otp_num_warn;
-    int fragmentation_latency;
     int peer_ssl;
     int peer_tcp_socket_sndbuf;
     int send_buffer_size;
@@ -661,7 +660,6 @@ void print_help (const char *name)
         "            --encryption-mode <blowfish/aes/none>\n"
         "            --hash-mode <md5/sha1/none>\n"
         "            [--otp <blowfish/aes> <num> <num-warn>]\n"
-        "            [--fragmentation-latency <milliseconds>]\n"
         "        )\n"
         "        (transport-mode=tcp?\n"
         "            (ssl? [--peer-ssl])\n"
@@ -714,7 +712,6 @@ int parse_arguments (int argc, char *argv[])
     options.encryption_mode = -1;
     options.hash_mode = -1;
     options.otp_mode = SPPROTO_OTP_MODE_NONE;
-    options.fragmentation_latency = PEER_DEFAULT_UDP_FRAGMENTATION_LATENCY;
     options.peer_ssl = 0;
     options.peer_tcp_socket_sndbuf = -1;
     options.send_buffer_size = PEER_DEFAULT_SEND_BUFFER_SIZE;
@@ -725,8 +722,6 @@ int parse_arguments (int argc, char *argv[])
     options.igmp_last_member_query_time = DEFAULT_IGMP_LAST_MEMBER_QUERY_TIME;
     options.allow_peer_talk_without_ssl = 0;
     
-    int have_fragmentation_latency = 0;
-    
     int i;
     for (i = 1; i < argc; i++) {
         char *arg = argv[i];
@@ -1008,15 +1003,6 @@ int parse_arguments (int argc, char *argv[])
             }
             i += 3;
         }
-        else if (!strcmp(arg, "--fragmentation-latency")) {
-            if (1 >= argc - i) {
-                fprintf(stderr, "%s: requires an argument\n", arg);
-                return 0;
-            }
-            options.fragmentation_latency = atoi(argv[i + 1]);
-            have_fragmentation_latency = 1;
-            i++;
-        }
         else if (!strcmp(arg, "--peer-ssl")) {
             options.peer_ssl = 1;
         }
@@ -1145,11 +1131,6 @@ int parse_arguments (int argc, char *argv[])
         return 0;
     }
     
-    if (!(!have_fragmentation_latency || (options.transport_mode == TRANSPORT_MODE_UDP))) {
-        fprintf(stderr, "False: --fragmentation-latency => UDP\n");
-        return 0;
-    }
-    
     if (!(!options.peer_ssl || (options.ssl && options.transport_mode == TRANSPORT_MODE_TCP))) {
         fprintf(stderr, "False: --peer-ssl => (--ssl && TCP)\n");
         return 0;
@@ -1556,7 +1537,7 @@ int peer_init_link (struct peer_data *peer)
         // init DatagramPeerIO
         if (!DatagramPeerIO_Init(
             &peer->pio.udp.pio, &ss, data_mtu, CLIENT_UDP_MTU, sp_params,
-            options.fragmentation_latency, PEER_UDP_ASSEMBLER_NUM_FRAMES, recv_if,
+            PEER_UDP_ASSEMBLER_NUM_FRAMES, recv_if,
             options.otp_num_warn, &twd, peer,
             (BLog_logfunc)peer_logfunc,
             (DatagramPeerIO_handler_error)peer_udp_pio_handler_error,

client/client.h

@@ -56,8 +56,6 @@
 #define PEER_DEFAULT_MAX_MACS 16
 // maximum number of multicast addresses per peer
 #define PEER_DEFAULT_MAX_GROUPS 16
-// how long we wait for a packet to reach full size before sending it (see FragmentProtoDisassembler latency argument)
-#define PEER_DEFAULT_UDP_FRAGMENTATION_LATENCY 0
 // value related to how much out-of-order input we tolerate (see FragmentProtoAssembler num_frames argument)
 #define PEER_UDP_ASSEMBLER_NUM_FRAMES 4
 // socket send buffer (SO_SNDBUF) for peer TCP connections, <=0 to not set