[PATCH 1/4] Do not trigger user disconnect at g_at_shutdown

[PATCH 1/4] Do not trigger user disconnect at g_at_shutdown

[Zhenhua Zhang]

[-- Attachment #1: Type: text/plain, Size: 3724 bytes --]

Similar to gatchat, introduce read_watcher_destroy_notify and do
not trigger user disconnect at g_at_shutdown. Delay destroy of
gatserver until read_watcher is destroyed.
---
 gatchat/gatserver.c |   42 ++++++++++++++++++++++++++++--------------
 1 files changed, 28 insertions(+), 14 deletions(-)

diff --git a/gatchat/gatserver.c b/gatchat/gatserver.c
index d131b0c..95a9e8f 100644
--- a/gatchat/gatserver.c
+++ b/gatchat/gatserver.c
@@ -89,7 +89,7 @@ struct _GAtServer {
 	gint ref_count;				/* Ref count */
 	struct v250_settings v250;		/* V.250 command setting */
 	GIOChannel *channel;			/* Server IO */
-	int server_watch;			/* Watch for server IO */
+	guint read_watch;			/* GSource read id, 0 if none */
 	guint read_so_far;			/* Number of bytes processed */
 	GAtDisconnectFunc user_disconnect;	/* User disconnect func */
 	gpointer user_disconnect_data;		/* User disconnect data */
@@ -98,6 +98,7 @@ struct _GAtServer {
 	struct ring_buffer *read_buf;		/* Current read buffer */
 	guint max_read_attempts;		/* Max reads per select */
 	enum ParserState parser_state;
+	gboolean destroyed;			/* Re-entrancy guard */
 };
 
 static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
@@ -380,7 +381,7 @@ static void new_bytes(GAtServer *p)
 
 	/* We're overflowing the buffer, shutdown the socket */
 	if (p->read_buf && ring_buffer_avail(p->read_buf) == 0)
-		g_source_remove(p->server_watch);
+		g_source_remove(p->read_watch);
 }
 
 static gboolean received_data(GIOChannel *channel, GIOCondition cond,
@@ -431,17 +432,25 @@ static gboolean received_data(GIOChannel *channel, GIOCondition cond,
 	return TRUE;
 }
 
-static void server_watcher_destroy_notify(GAtServer *server)
+static void g_at_server_cleanup(GAtServer *server)
 {
-	server->server_watch = 0;
-
+	/* Cleanup all received data */
 	ring_buffer_free(server->read_buf);
 	server->read_buf = NULL;
 
 	server->channel = NULL;
+}
+
+static void read_watcher_destroy_notify(GAtServer *server)
+{
+	g_at_server_cleanup(server);
+	server->read_watch = 0;
 
 	if (server->user_disconnect)
 		server->user_disconnect(server->user_disconnect_data);
+
+	if (server->destroyed)
+		g_free(server);
 }
 
 static void v250_settings_create(struct v250_settings *v250)
@@ -480,10 +489,10 @@ GAtServer *g_at_server_new(GIOChannel *io)
 	if (!g_at_util_setup_io(server->channel, G_IO_FLAG_NONBLOCK))
 		goto error;
 
-	server->server_watch = g_io_add_watch_full(io, G_PRIORITY_DEFAULT,
+	server->read_watch = g_io_add_watch_full(io, G_PRIORITY_DEFAULT,
 				G_IO_IN | G_IO_HUP | G_IO_ERR | G_IO_NVAL,
 				received_data, server,
-				(GDestroyNotify)server_watcher_destroy_notify);
+				(GDestroyNotify)read_watcher_destroy_notify);
 
 	return server;
 
@@ -520,6 +529,16 @@ void g_at_server_unref(GAtServer *server)
 		return;
 
 	g_at_server_shutdown(server);
+
+	/* glib delays the destruction of the watcher until it exits, this
+	 * means we can't free the data just yet, even though we've been
+	 * destroyed already.  We have to wait until the read_watcher
+	 * destroy function gets called
+	 */
+	if (server->read_watch != 0)
+		server->destroyed = TRUE;
+	else
+		g_free(server);
 }
 
 gboolean g_at_server_shutdown(GAtServer *server)
@@ -531,13 +550,8 @@ gboolean g_at_server_shutdown(GAtServer *server)
 	server->user_disconnect = NULL;
 	server->user_disconnect_data = NULL;
 
-	if (server->server_watch) {
-		g_source_remove(server->server_watch);
-		server->server_watch = 0;
-	}
-
-	g_free(server);
-	server = NULL;
+	if (server->read_watch)
+		g_source_remove(server->read_watch);
 
 	return TRUE;
 }
-- 
1.6.6.1

[PATCH 2/4] Add define for read and write buffer size

[Zhenhua Zhang]

[-- Attachment #1: Type: text/plain, Size: 723 bytes --]

---
 gatchat/gatserver.c |    4 +++-
 1 files changed, 3 insertions(+), 1 deletions(-)

diff --git a/gatchat/gatserver.c b/gatchat/gatserver.c
index 95a9e8f..15bbbff 100644
--- a/gatchat/gatserver.c
+++ b/gatchat/gatserver.c
@@ -32,6 +32,8 @@
 #include "ringbuffer.h"
 #include "gatserver.h"
 
+#define BUF_SIZE 4096
+
 enum ParserState {
 	PARSER_STATE_IDLE,
 	PARSER_STATE_A,
@@ -480,7 +482,7 @@ GAtServer *g_at_server_new(GIOChannel *io)
 	server->ref_count = 1;
 	v250_settings_create(&server->v250);
 	server->channel = io;
-	server->read_buf = ring_buffer_new(4096);
+	server->read_buf = ring_buffer_new(BUF_SIZE);
 	server->max_read_attempts = 3;
 
 	if (!server->read_buf)
-- 
1.6.6.1

[PATCH 3/4] Add write buffer queue for non-blocking write

[Zhenhua Zhang]

[-- Attachment #1: Type: text/plain, Size: 5682 bytes --]

The head of the queue is the data to be written, the tail is the
free buffer to cache data into.

If the tail of queue is full, allocate a new free buffer and
append it at the tail.
---
 gatchat/gatserver.c |  105 ++++++++++++++++++++++++++++++++++++++++++++++-----
 1 files changed, 95 insertions(+), 10 deletions(-)

diff --git a/gatchat/gatserver.c b/gatchat/gatserver.c
index 15bbbff..a136a6f 100644
--- a/gatchat/gatserver.c
+++ b/gatchat/gatserver.c
@@ -92,17 +92,56 @@ struct _GAtServer {
 	struct v250_settings v250;		/* V.250 command setting */
 	GIOChannel *channel;			/* Server IO */
 	guint read_watch;			/* GSource read id, 0 if none */
+	guint write_watch;			/* GSource write id, 0 if none */
 	guint read_so_far;			/* Number of bytes processed */
 	GAtDisconnectFunc user_disconnect;	/* User disconnect func */
 	gpointer user_disconnect_data;		/* User disconnect data */
 	GAtDebugFunc debugf;			/* Debugging output function */
 	gpointer debug_data;			/* Data to pass to debug func */
 	struct ring_buffer *read_buf;		/* Current read buffer */
+	GQueue *write_queue;			/* Write buffer queue */
 	guint max_read_attempts;		/* Max reads per select */
 	enum ParserState parser_state;
 	gboolean destroyed;			/* Re-entrancy guard */
 };
 
+static void g_at_server_wakeup_writer(GAtServer *server);
+
+static struct ring_buffer *allocate_next(GAtServer *server)
+{
+	struct ring_buffer *buf = ring_buffer_new(BUF_SIZE);
+
+	if (!buf)
+		return NULL;
+
+	g_queue_push_tail(server->write_queue, buf);
+
+	return buf;
+}
+
+static void send_common(GAtServer *server, const char *buf, unsigned int len)
+{
+	gsize towrite = len;
+	gsize bytes_written = 0;
+	struct ring_buffer *write_buf;
+
+	write_buf = g_queue_peek_tail(server->write_queue);
+
+	while (bytes_written < towrite) {
+		gsize wbytes = MIN((gsize)ring_buffer_avail(write_buf),
+						towrite - bytes_written);
+
+		bytes_written += ring_buffer_write(write_buf,
+							buf + bytes_written,
+							wbytes);
+
+		if (ring_buffer_avail(write_buf) == 0)
+			write_buf = allocate_next(server);
+	}
+
+	g_at_server_wakeup_writer(server);
+}
+
 static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
 {
 	struct v250_settings v250 = server->v250;
@@ -110,7 +149,7 @@ static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
 	char buf[1024];
 	char t = v250.s3;
 	char r = v250.s4;
-	gsize wbuf;
+	unsigned int len;
 
 	if (v250.quiet)
 		return;
@@ -119,16 +158,13 @@ static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
 		return;
 
 	if (v250.is_v1)
-		snprintf(buf, sizeof(buf), "%c%c%s%c%c", t, r, result_str,
+		len = snprintf(buf, sizeof(buf), "%c%c%s%c%c", t, r, result_str,
 				t, r);
 	else
-		snprintf(buf, sizeof(buf), "%u%c", (unsigned int) result, t);
-
-	g_at_util_debug_chat(FALSE, buf, strlen(buf),
-				server->debugf, server->debug_data);
+		len = snprintf(buf, sizeof(buf), "%u%c", (unsigned int) result,
+				t);
 
-	g_io_channel_write(server->channel, (char *) buf, strlen(buf),
-							&wbuf);
+	send_common(server, buf, MIN(len, sizeof(buf)-1));
 }
 
 static inline gboolean is_at_command_prefix(const char c)
@@ -434,12 +470,31 @@ static gboolean received_data(GIOChannel *channel, GIOCondition cond,
 	return TRUE;
 }
 
+static gboolean can_write_data(GIOChannel *channel, GIOCondition cond,
+				gpointer data)
+{
+	return FALSE;
+}
+
+static void write_queue_free(GQueue *write_queue)
+{
+	struct ring_buffer *write_buf;
+
+	while ((write_buf = g_queue_pop_head(write_queue)))
+		ring_buffer_free(write_buf);
+
+	g_queue_free(write_queue);
+}
+
 static void g_at_server_cleanup(GAtServer *server)
 {
 	/* Cleanup all received data */
 	ring_buffer_free(server->read_buf);
 	server->read_buf = NULL;
 
+	/* Cleanup pending data to write */
+	write_queue_free(server->write_queue);
+
 	server->channel = NULL;
 }
 
@@ -455,6 +510,23 @@ static void read_watcher_destroy_notify(GAtServer *server)
 		g_free(server);
 }
 
+static void write_watcher_destroy_notify(GAtServer *server)
+{
+	server->write_watch = 0;
+}
+
+static void g_at_server_wakeup_writer(GAtServer *server)
+{
+	if (server->write_watch != 0)
+		return;
+
+	server->write_watch = g_io_add_watch_full(server->channel,
+			G_PRIORITY_DEFAULT,
+			G_IO_OUT | G_IO_HUP | G_IO_ERR | G_IO_NVAL,
+			can_write_data, server,
+			(GDestroyNotify)write_watcher_destroy_notify);
+}
+
 static void v250_settings_create(struct v250_settings *v250)
 {
 	v250->s3 = '\r';
@@ -483,11 +555,18 @@ GAtServer *g_at_server_new(GIOChannel *io)
 	v250_settings_create(&server->v250);
 	server->channel = io;
 	server->read_buf = ring_buffer_new(BUF_SIZE);
-	server->max_read_attempts = 3;
-
 	if (!server->read_buf)
 		goto error;
 
+	server->write_queue = g_queue_new();
+	if (!server->write_queue)
+		goto error;
+
+	if (!allocate_next(server))
+		goto error;
+
+	server->max_read_attempts = 3;
+
 	if (!g_at_util_setup_io(server->channel, G_IO_FLAG_NONBLOCK))
 		goto error;
 
@@ -502,6 +581,9 @@ error:
 	if (server->read_buf)
 		ring_buffer_free(server->read_buf);
 
+	if (server->write_queue)
+		write_queue_free(server->write_queue);
+
 	if (server)
 		g_free(server);
 
@@ -552,6 +634,9 @@ gboolean g_at_server_shutdown(GAtServer *server)
 	server->user_disconnect = NULL;
 	server->user_disconnect_data = NULL;
 
+	if (server->write_watch)
+		g_source_remove(server->write_watch);
+
 	if (server->read_watch)
 		g_source_remove(server->read_watch);
 
-- 
1.6.6.1

[PATCH 4/4] Add write server response into non-blocking IO

[Zhenhua Zhang]

[-- Attachment #1: Type: text/plain, Size: 2313 bytes --]

Write server response into non-blocking GIOChannel when we have
G_IO_OUT signal.
---
 gatchat/gatserver.c |   66 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 66 insertions(+), 0 deletions(-)

diff --git a/gatchat/gatserver.c b/gatchat/gatserver.c
index a136a6f..5b5e2f7 100644
--- a/gatchat/gatserver.c
+++ b/gatchat/gatserver.c
@@ -33,6 +33,7 @@
 #include "gatserver.h"
 
 #define BUF_SIZE 4096
+/* #define WRITE_SCHEDULER_DEBUG 1 */
 
 enum ParserState {
 	PARSER_STATE_IDLE,
@@ -473,6 +474,71 @@ static gboolean received_data(GIOChannel *channel, GIOCondition cond,
 static gboolean can_write_data(GIOChannel *channel, GIOCondition cond,
 				gpointer data)
 {
+	GAtServer *server = data;
+	GIOError err;
+	gsize bytes_written;
+	gsize towrite;
+	struct ring_buffer *write_buf;
+	unsigned char *buf;
+	gboolean write_again = FALSE;
+#ifdef WRITE_SCHEDULER_DEBUG
+	int limiter;
+#endif
+
+	if (cond & (G_IO_NVAL | G_IO_HUP | G_IO_ERR))
+		return FALSE;
+
+	if (!server->write_queue)
+		return FALSE;
+
+	/* Write data out from the head of the queue */
+	write_buf = g_queue_peek_head(server->write_queue);
+
+	buf = ring_buffer_read_ptr(write_buf, 0);
+
+	towrite = ring_buffer_len_no_wrap(write_buf);
+	if (towrite < (gsize)ring_buffer_len(write_buf))
+		write_again = TRUE;
+
+#ifdef WRITE_SCHEDULER_DEBUG
+	limiter = towrite;
+
+	if (limiter > 5)
+		limiter = 5;
+#endif
+
+	err = g_io_channel_write(server->channel,
+			(char *)buf,
+#ifdef WRITE_SCHEDULER_DEBUG
+			limiter,
+#else
+			towrite,
+#endif
+			&bytes_written);
+
+	if (err != G_IO_ERROR_NONE) {
+		g_source_remove(server->read_watch);
+		return FALSE;
+	}
+
+	g_at_util_debug_chat(FALSE, (char *)buf, bytes_written, server->debugf,
+				server->debug_data);
+
+	ring_buffer_drain(write_buf, bytes_written);
+
+	/* All data in current buffer is written, free it
+	 * unless it's the last buffer in the queue.
+	 */
+	if ((ring_buffer_len(write_buf) == 0) &&
+			(g_queue_get_length(server->write_queue) != 1)) {
+		write_buf = g_queue_pop_head(server->write_queue);
+		ring_buffer_free(write_buf);
+		return TRUE;
+	}
+
+	if (bytes_written < towrite || write_again == TRUE)
+		return TRUE;
+
 	return FALSE;
 }
 
-- 
1.6.6.1

Re: [PATCH 3/4] Add write buffer queue for non-blocking write

[Zhenhua Zhang]

[-- Attachment #1: Type: text/plain, Size: 5148 bytes --]

Hi Denis,

On 02/26/2010 05:56 PM, Zhang, Zhenhua wrote:
> The head of the queue is the data to be written, the tail is the
> free buffer to cache data into.
>
> If the tail of queue is full, allocate a new free buffer and
> append it at the tail.
> ---
>   gatchat/gatserver.c |  105 ++++++++++++++++++++++++++++++++++++++++++++++-----
>   1 files changed, 95 insertions(+), 10 deletions(-)
>
> +
> +static void send_common(GAtServer *server, const char *buf, unsigned int len)
> +{
> +	gsize towrite = len;
> +	gsize bytes_written = 0;
> +	struct ring_buffer *write_buf;
> +
> +	write_buf = g_queue_peek_tail(server->write_queue);

This is the updated 3/4 patch. For this write_buf, it should alway exist 
since we allocate one buffer in g_at_server_new. So I don't check 
write_buf here.

> +	while (bytes_written<  towrite) {
> +		gsize wbytes = MIN((gsize)ring_buffer_avail(write_buf),
> +						towrite - bytes_written);
> +
> +		bytes_written += ring_buffer_write(write_buf,
> +							buf + bytes_written,
> +							wbytes);
> +
> +		if (ring_buffer_avail(write_buf) == 0)
> +			write_buf = allocate_next(server);
> +	}
> +
> +	g_at_server_wakeup_writer(server);
> +}
> +
>   static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
>   {
>   	struct v250_settings v250 = server->v250;
> @@ -110,7 +149,7 @@ static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
>   	char buf[1024];
>   	char t = v250.s3;
>   	char r = v250.s4;
> -	gsize wbuf;
> +	unsigned int len;
>
>   	if (v250.quiet)
>   		return;
> @@ -119,16 +158,13 @@ static void g_at_server_send_result(GAtServer *server, GAtServerResult result)
>   		return;
>
>   	if (v250.is_v1)
> -		snprintf(buf, sizeof(buf), "%c%c%s%c%c", t, r, result_str,
> +		len = snprintf(buf, sizeof(buf), "%c%c%s%c%c", t, r, result_str,
>   				t, r);
>   	else
> -		snprintf(buf, sizeof(buf), "%u%c", (unsigned int) result, t);
> -
> -	g_at_util_debug_chat(FALSE, buf, strlen(buf),
> -				server->debugf, server->debug_data);
> +		len = snprintf(buf, sizeof(buf), "%u%c", (unsigned int) result,
> +				t);
>
> -	g_io_channel_write(server->channel, (char *) buf, strlen(buf),
> -							&wbuf);
> +	send_common(server, buf, MIN(len, sizeof(buf)-1));

Update to avoid buffer overflow case. However, it's rarely possible in 
g_at_server_result.

Let me know if any comments for the patches. Thanks.

Zhenhua.

>   }
>
>   static inline gboolean is_at_command_prefix(const char c)
> @@ -434,12 +470,31 @@ static gboolean received_data(GIOChannel *channel, GIOCondition cond,
>   	return TRUE;
>   }
>
> +static gboolean can_write_data(GIOChannel *channel, GIOCondition cond,
> +				gpointer data)
> +{
> +	return FALSE;
> +}
> +
> +static void write_queue_free(GQueue *write_queue)
> +{
> +	struct ring_buffer *write_buf;
> +
> +	while ((write_buf = g_queue_pop_head(write_queue)))
> +		ring_buffer_free(write_buf);
> +
> +	g_queue_free(write_queue);
> +}
> +
>   static void g_at_server_cleanup(GAtServer *server)
>   {
>   	/* Cleanup all received data */
>   	ring_buffer_free(server->read_buf);
>   	server->read_buf = NULL;
>
> +	/* Cleanup pending data to write */
> +	write_queue_free(server->write_queue);
> +
>   	server->channel = NULL;
>   }
>
> @@ -455,6 +510,23 @@ static void read_watcher_destroy_notify(GAtServer *server)
>   		g_free(server);
>   }
>
> +static void write_watcher_destroy_notify(GAtServer *server)
> +{
> +	server->write_watch = 0;
> +}
> +
> +static void g_at_server_wakeup_writer(GAtServer *server)
> +{
> +	if (server->write_watch != 0)
> +		return;
> +
> +	server->write_watch = g_io_add_watch_full(server->channel,
> +			G_PRIORITY_DEFAULT,
> +			G_IO_OUT | G_IO_HUP | G_IO_ERR | G_IO_NVAL,
> +			can_write_data, server,
> +			(GDestroyNotify)write_watcher_destroy_notify);
> +}
> +
>   static void v250_settings_create(struct v250_settings *v250)
>   {
>   	v250->s3 = '\r';
> @@ -483,11 +555,18 @@ GAtServer *g_at_server_new(GIOChannel *io)
>   	v250_settings_create(&server->v250);
>   	server->channel = io;
>   	server->read_buf = ring_buffer_new(BUF_SIZE);
> -	server->max_read_attempts = 3;
> -
>   	if (!server->read_buf)
>   		goto error;
>
> +	server->write_queue = g_queue_new();
> +	if (!server->write_queue)
> +		goto error;
> +
> +	if (!allocate_next(server))
> +		goto error;
> +
> +	server->max_read_attempts = 3;
> +
>   	if (!g_at_util_setup_io(server->channel, G_IO_FLAG_NONBLOCK))
>   		goto error;
>
> @@ -502,6 +581,9 @@ error:
>   	if (server->read_buf)
>   		ring_buffer_free(server->read_buf);
>
> +	if (server->write_queue)
> +		write_queue_free(server->write_queue);
> +
>   	if (server)
>   		g_free(server);
>
> @@ -552,6 +634,9 @@ gboolean g_at_server_shutdown(GAtServer *server)
>   	server->user_disconnect = NULL;
>   	server->user_disconnect_data = NULL;
>
> +	if (server->write_watch)
> +		g_source_remove(server->write_watch);
> +
>   	if (server->read_watch)
>   		g_source_remove(server->read_watch);
>

Re: [PATCH 1/4] Do not trigger user disconnect at g_at_shutdown

[Denis Kenzior]

[-- Attachment #1: Type: text/plain, Size: 250 bytes --]

Hi Zhenhua,

> Similar to gatchat, introduce read_watcher_destroy_notify and do
> not trigger user disconnect at g_at_shutdown. Delay destroy of
> gatserver until read_watcher is destroyed.

Patch has been applied, thanks.

Regards,
-Denis

Re: [PATCH 2/4] Add define for read and write buffer size

[Denis Kenzior]

[-- Attachment #1: Type: text/plain, Size: 170 bytes --]

Hi Zhenhua,

> ---
>  gatchat/gatserver.c |    4 +++-
>  1 files changed, 3 insertions(+), 1 deletions(-)
> 

Patch has been applied, thanks.

Regards,
-Denis

Re: [PATCH 3/4] Add write buffer queue for non-blocking write

[Denis Kenzior]

[-- Attachment #1: Type: text/plain, Size: 290 bytes --]

Hi Zhenhua,

> The head of the queue is the data to be written, the tail is the
> free buffer to cache data into.
> 
> If the tail of queue is full, allocate a new free buffer and
> append it at the tail.

Patch has been applied with one minor tweak afterwards.

Thanks,
-Denis

Re: [PATCH 4/4] Add write server response into non-blocking IO

[Denis Kenzior]

[-- Attachment #1: Type: text/plain, Size: 185 bytes --]

Hi Zhenhua,

> Write server response into non-blocking GIOChannel when we have
> G_IO_OUT signal.

Patch has been applied with minor simplification afterwards.

Thanks,
-Denis