[Libpdtp-svn] r57 - /

[<tar...@pd...>]

Author: tarcieri
Date: 2004-12-20 17:19:43 -0700 (Mon, 20 Dec 2004)
New Revision: 57

Added:
   async.c
   async.h
Log:
Inclusion of previously missing asynchronous I/O code

Added: async.c
===================================================================
--- async.c	2004-12-21 00:12:02 UTC (rev 56)
+++ async.c	2004-12-21 00:19:43 UTC (rev 57)
@@ -0,0 +1,155 @@
+/*
+   async.c - Asynchronous read/write functions
+   Copyright (C)2003-04 Anthony Arcieri
+   All rights reserved.
+
+   Redistribution and use in source and binary forms, with or without
+   modification, are permitted provided that the following conditions
+   are met:
+
+ * Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+
+ * Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+
+ * The name of Anthony Arcieri may not be used to endorse or promote 
+ products derived from this software without specific prior written 
+ permission.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT 
+ OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 
+ SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 
+ TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "machdep.h"
+#include "pdtp.h"
+
+#include <string.h>
+
+#include "alloc.h"
+#include "async.h"
+#include "debug.h"
+#include "multiplexer.h"
+#include "socketops.h"
+
+typedef struct async_io_parameters {
+	uint8_t *buffer;
+	unsigned bytes_transferred, bytes_total;
+	pdtp_async_handler_t callback;
+	void *arg;
+} *async_io_parameters_t;
+
+static int async_read_handler(pdtp_multiplexer_t, pdtp_sock_t, void *arg);
+static int async_write_handler(pdtp_multiplexer_t, pdtp_sock_t, void *arg);
+
+void pdtp__async_read(pdtp_multiplexer_t mplx, pdtp_sock_t sock, unsigned length, pdtp_async_handler_t completion_callback, void *arg)
+{
+	async_io_parameters_t params = NEW(async_io_parameters);
+
+	params->buffer = pdtp__alloc(0, length);
+	params->bytes_transferred = 0;
+	params->bytes_total = length;
+	params->callback = completion_callback;
+	params->arg = arg;
+
+	pdtp__multiplexer_add_socket(mplx, sock, PDTP_MPLX_READ, async_read_handler, params);
+}
+
+void pdtp__async_write(pdtp_multiplexer_t mplx, pdtp_sock_t sock, void *buffer, unsigned length, pdtp_async_handler_t completion_callback, void *arg)
+{
+	async_io_parameters_t params = NEW(async_io_parameters);
+
+	params->buffer = pdtp__alloc(0, length);
+	params->bytes_transferred = 0;
+	params->bytes_total = length;
+	params->callback = completion_callback;
+	params->arg = arg;
+
+	memcpy(params->buffer, buffer, length);
+
+	pdtp__multiplexer_add_socket(mplx, sock, PDTP_MPLX_WRITE, async_write_handler, params);
+}
+
+static int async_read_handler(pdtp_multiplexer_t mplx, pdtp_sock_t sock, void *arg)
+{
+        int n, ret;
+        async_io_parameters_t params = (async_io_parameters_t)arg;
+
+#ifndef NDEBUG
+	debug("async_read_handler invoked, trying to get %d bytes", params->bytes_total - params->bytes_transferred);
+#endif
+
+        if((n = pdtp__atomic_readsock(sock, params->buffer + params->bytes_transferred, params->bytes_total - params->bytes_transferred)) < 1) {
+#ifndef NDEBUG
+		debug("async_read_handler: Read error (%d)", n);
+		perror("pdtp__atomic_readsock");
+#endif
+                pdtp__multiplexer_remove_socket(mplx, sock);
+                ret = params->callback(mplx, sock, 0, -1, params->arg);
+
+                pdtp__free(params->buffer);
+                pdtp__free(params);
+
+                return ret;
+        }
+
+        params->bytes_transferred += n;
+
+#ifndef NDEBUG
+	debug("async_read_handler: %d bytes read, %d remain", n, params->bytes_total - params->bytes_transferred);
+#endif
+
+        if(params->bytes_transferred < params->bytes_total)
+                return 1;
+
+        pdtp__multiplexer_remove_socket(mplx, sock);
+        ret = params->callback(mplx, sock, params->buffer, params->bytes_total, params->arg);
+
+        pdtp__free(params->buffer);
+        pdtp__free(params);
+
+        return ret;
+}
+
+static int async_write_handler(pdtp_multiplexer_t mplx, pdtp_sock_t sock, void *arg)
+{
+        int n, ret;
+        async_io_parameters_t params = (async_io_parameters_t)arg;
+
+#ifndef NDEBUG
+	debug("async_write_handler invoked");
+#endif
+
+        if((n = pdtp__atomic_writesock(sock, params->buffer + params->bytes_transferred, params->bytes_total - params->bytes_transferred)) < 1) {
+                pdtp__multiplexer_remove_socket(mplx, sock);
+                ret = params->callback(mplx, sock, 0, -1, params->arg);
+
+                pdtp__free(params->buffer);
+                pdtp__free(params);
+
+                return ret;
+        }
+
+        params->bytes_transferred += n;
+
+        if(params->bytes_transferred < params->bytes_total)
+                return 1;
+
+        pdtp__multiplexer_remove_socket(mplx, sock);
+        ret = params->callback(mplx, sock, params->buffer, params->bytes_total, params->arg);
+
+        pdtp__free(params->buffer);
+        pdtp__free(params);
+
+        return ret;
+}

Added: async.h
===================================================================
--- async.h	2004-12-21 00:12:02 UTC (rev 56)
+++ async.h	2004-12-21 00:19:43 UTC (rev 57)
@@ -0,0 +1,24 @@
+#ifndef ASYNC_H
+#define ASYNC_H
+
+#include "machdep.h"
+#include "pdtp.h"
+
+#include "multiplexer.h"
+#include "socketops.h"
+
+typedef int (*pdtp_async_handler_t)(pdtp_multiplexer_t, pdtp_sock_t, const void *buffer, int bytes, void *arg);
+
+/* Asynchronous read function.  The specified amount of data will be read
+   from the given socket,  Once this has been completed, the specified
+   completion_callback will be invoked with the data and the given argument */
+void pdtp__async_read(pdtp_multiplexer_t mplx, pdtp_sock_t sock, unsigned length, pdtp_async_handler_t completion_callback, void *arg);
+
+/* Asynchronous write function.  The specified amount of data will be
+   written from the specified buffer to the given socket.  Once this has
+   been completed, the specified completion_callback will be invoked with
+   the given argument */
+void pdtp__async_write(pdtp_multiplexer_t mplx, pdtp_sock_t sock, void *buffer,
+unsigned length, pdtp_async_handler_t completion_callback, void *arg);
+
+#endif /* ASYNC_H */