PPP: Optimize write buffer management

Extend the write buffer handling in gathdlc.c to minimize stalling and
process switching during large PPP transfers. The single write buffer
is replaced by a queue of buffers, allowing for much larger emission
windows without hugely impacting memory consumption. This reduces the
time required to send 50 MB between a couple of local PPP interfaces on
my laptop from ~53s to ~3s.

1 files changed, 71 insertions, 22 deletions

diff --git a/gatchat/gathdlc.c b/gatchat/gathdlc.c
index 6c39e6c0..7c454540 100644
--- a/gatchat/gathdlc.c
+++ b/gatchat/gathdlc.c
@@ -37,7 +37,9 @@
 #include "gatio.h"
 #include "gathdlc.h"
 
-#define BUFFER_SIZE 2048
+#define BUFFER_SIZE	(2 * 2048)
+#define MAX_BUFFERS	64	/* Maximum number of in-flight write buffers */
+#define HDLC_OVERHEAD	256	/* Rough estimate of HDLC protocol overhead */
 
 #define HDLC_FLAG	0x7e	/* Flag sequence */
 #define HDLC_ESCAPE	0x7d	/* Asynchronous control escape */
@@ -51,7 +53,7 @@
 struct _GAtHDLC {
 	gint ref_count;
 	GAtIO *io;
-	struct ring_buffer *write_buffer;
+	GQueue *write_queue;	/* Write buffer queue */
 	unsigned char *decode_buffer;
 	guint decode_offset;
 	guint16 decode_fcs;
@@ -201,6 +203,7 @@ GAtHDLC *g_at_hdlc_new_from_io(GAtIO *io)
 {
 	GAtHDLC *hdlc;
 	unsigned char *buf;
+	struct ring_buffer* write_buffer;
 
 	if (io == NULL)
 		return NULL;
@@ -218,16 +221,22 @@ GAtHDLC *g_at_hdlc_new_from_io(GAtIO *io)
 	hdlc->xmit_accm[3] = 0x60000000; /* 0x7d, 0x7e */
 	hdlc->recv_accm = ~0U;
 
-	hdlc->write_buffer = ring_buffer_new(BUFFER_SIZE * 2);
-	if (!hdlc->write_buffer)
+	write_buffer = ring_buffer_new(BUFFER_SIZE);
+	if (!write_buffer)
+		goto error;
+
+	hdlc->write_queue = g_queue_new();
+	if (!hdlc->write_queue)
 		goto error;
 
+	g_queue_push_tail(hdlc->write_queue, write_buffer);
+
 	/* Write an initial 0x7e as wakeup character */
-	buf = ring_buffer_write_ptr(hdlc->write_buffer, 0);
+	buf = ring_buffer_write_ptr(write_buffer, 0);
 	*buf = HDLC_FLAG;
-	ring_buffer_write_advance(hdlc->write_buffer, 1);
+	ring_buffer_write_advance(write_buffer, 1);
 
-	hdlc->decode_buffer = g_try_malloc(BUFFER_SIZE * 2);
+	hdlc->decode_buffer = g_try_malloc(BUFFER_SIZE);
 	if (!hdlc->decode_buffer)
 		goto error;
 
@@ -239,8 +248,11 @@ GAtHDLC *g_at_hdlc_new_from_io(GAtIO *io)
 	return hdlc;
 
 error:
-	if (hdlc->write_buffer)
-		ring_buffer_free(hdlc->write_buffer);
+	if (hdlc->write_queue)
+		g_queue_free(hdlc->write_queue);
+
+	if (write_buffer)
+		ring_buffer_free(write_buffer);
 
 	if (hdlc->decode_buffer)
 		g_free(hdlc->decode_buffer);
@@ -277,6 +289,8 @@ GAtHDLC *g_at_hdlc_ref(GAtHDLC *hdlc)
 
 void g_at_hdlc_unref(GAtHDLC *hdlc)
 {
+	struct ring_buffer *write_buffer;
+
 	if (hdlc == NULL)
 		return;
 
@@ -294,7 +308,11 @@ void g_at_hdlc_unref(GAtHDLC *hdlc)
 	g_at_io_unref(hdlc->io);
 	hdlc->io = NULL;
 
-	ring_buffer_free(hdlc->write_buffer);
+	while ((write_buffer = g_queue_pop_head(hdlc->write_queue)))
+		ring_buffer_free(write_buffer);
+
+	g_queue_free(hdlc->write_queue);
+
 	g_free(hdlc->decode_buffer);
 
 	if (hdlc->in_read_handler)
@@ -328,15 +346,32 @@ static gboolean can_write_data(gpointer data)
 	unsigned int len;
 	unsigned char *buf;
 	gsize bytes_written;
+	struct ring_buffer* write_buffer;
 
-	len = ring_buffer_len_no_wrap(hdlc->write_buffer);
-	buf = ring_buffer_read_ptr(hdlc->write_buffer, 0);
+	/* Write data out from the head of the queue */
+	write_buffer = g_queue_peek_head(hdlc->write_queue);
+
+	len = ring_buffer_len_no_wrap(write_buffer);
+	buf = ring_buffer_read_ptr(write_buffer, 0);
 
 	bytes_written = g_at_io_write(hdlc->io, (gchar *) buf, len);
 	hdlc_record(hdlc->record_fd, FALSE, buf, bytes_written);
-	ring_buffer_drain(hdlc->write_buffer, bytes_written);
+	ring_buffer_drain(write_buffer, bytes_written);
 
-	if (ring_buffer_len(hdlc->write_buffer) > 0)
+	if (ring_buffer_len(write_buffer) > 0)
+		return TRUE;
+
+	/* All data in current buffer is written, free it
+	 * unless it's the last buffer in the queue.
+	 */
+	if ((ring_buffer_len(write_buffer) == 0) &&
+			(g_queue_get_length(hdlc->write_queue) > 1)) {
+		write_buffer = g_queue_pop_head(hdlc->write_queue);
+		ring_buffer_free(write_buffer);
+		write_buffer = g_queue_peek_head(hdlc->write_queue);
+	}
+
+	if (ring_buffer_len(write_buffer) > 0)
 		return TRUE;
 
 	return FALSE;
@@ -370,19 +405,33 @@ GAtIO *g_at_hdlc_get_io(GAtHDLC *hdlc)
 
 gboolean g_at_hdlc_send(GAtHDLC *hdlc, const unsigned char *data, gsize size)
 {
-	unsigned int avail = ring_buffer_avail(hdlc->write_buffer);
-	unsigned int wrap = ring_buffer_avail_no_wrap(hdlc->write_buffer);
-	unsigned char *buf = ring_buffer_write_ptr(hdlc->write_buffer, 0);
+	struct ring_buffer* write_buffer = g_queue_peek_tail(hdlc->write_queue);
+
+	unsigned int avail = ring_buffer_avail(write_buffer);
+	unsigned int wrap = ring_buffer_avail_no_wrap(write_buffer);
+	unsigned char *buf;
 	unsigned char tail[2];
 	unsigned int i = 0;
 	guint16 fcs = HDLC_INITFCS;
 	gboolean escape = FALSE;
 	gsize pos = 0;
 
-	if (avail < size)
-		return FALSE;
+	if (avail < size + HDLC_OVERHEAD) {
+		if (g_queue_get_length(hdlc->write_queue) > MAX_BUFFERS)
+			return FALSE;	/* Too many pending buffers */
+
+		write_buffer = ring_buffer_new(BUFFER_SIZE);
+		if (write_buffer == NULL)
+			return FALSE;
+
+		g_queue_push_tail(hdlc->write_queue, write_buffer);
+
+		avail = ring_buffer_avail(write_buffer);
+		wrap = ring_buffer_avail_no_wrap(write_buffer);
+	}
 
 	i = 0;
+	buf = ring_buffer_write_ptr(write_buffer, 0);
 
 	while (pos < avail && i < size) {
 		if (escape == TRUE) {
@@ -401,7 +450,7 @@ gboolean g_at_hdlc_send(GAtHDLC *hdlc, const unsigned char *data, gsize size)
 		pos++;
 
 		if (pos == wrap)
-			buf = ring_buffer_write_ptr(hdlc->write_buffer, pos);
+			buf = ring_buffer_write_ptr(write_buffer, pos);
 	}
 
 	if (i < size)
@@ -428,7 +477,7 @@ gboolean g_at_hdlc_send(GAtHDLC *hdlc, const unsigned char *data, gsize size)
 		pos++;
 
 		if (pos == wrap)
-			buf = ring_buffer_write_ptr(hdlc->write_buffer, pos);
+			buf = ring_buffer_write_ptr(write_buffer, pos);
 	}
 
 	if (i < sizeof(tail))
@@ -440,7 +489,7 @@ gboolean g_at_hdlc_send(GAtHDLC *hdlc, const unsigned char *data, gsize size)
 	*buf = HDLC_FLAG;
 	pos++;
 
-	ring_buffer_write_advance(hdlc->write_buffer, pos);
+	ring_buffer_write_advance(write_buffer, pos);
 
 	g_at_io_set_write_handler(hdlc->io, can_write_data, hdlc);