summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/H5FDstream.c984
-rw-r--r--src/H5FDstream.h53
2 files changed, 1037 insertions, 0 deletions
diff --git a/src/H5FDstream.c b/src/H5FDstream.c
new file mode 100644
index 0000000..cd0e6e6
--- /dev/null
+++ b/src/H5FDstream.c
@@ -0,0 +1,984 @@
+/*
+ * Copyright © 2000 The author.
+ * The author prefers this code not be used for military purposes.
+ *
+ *
+ * Author: Thomas Radke <tradke@aei-potsdam.mpg.de>
+ * Tuesday, September 12, 2000
+ *
+ * Purpose: This code provides the Stream Virtual File Driver.
+ * It is very much based on the core VFD which keeps an
+ * entire HDF5 data file to be processed in main memory.
+ * In addition to that, the memory image of the file is
+ * read from/written to a socket during an open/flush operation.
+ */
+
+#include <netdb.h> /* gethostbyname */
+#include <sys/types.h> /* socket stuff */
+#include <sys/socket.h> /* socket stuff */
+#include <netinet/in.h> /* socket stuff */
+#include <netinet/tcp.h> /* socket stuff */
+
+#include <H5Eprivate.h> /* error handling */
+#include <H5FDpublic.h> /* VFD structures */
+#include <H5MMprivate.h> /* memory allocation */
+#include <H5Ppublic.h> /* files */
+#include <H5FDstream.h> /* Stream VFD header */
+
+
+/* Only build this driver if it was configured with --with-Stream-VFD */
+#ifdef H5_HAVE_STREAM
+
+/* Some useful macros */
+#undef MAX
+#define MAX(x,y) ((x) > (y) ? (x) : (y))
+#undef MIN
+#define MIN(x,y) ((x) < (y) ? (x) : (y))
+
+/* Uncomment this to switch on debugging output */
+/* #define DEBUG 1 */
+
+/* The driver identification number, initialized at runtime */
+static hid_t H5FD_STREAM_g = 0;
+
+/*
+ * The description of a file belonging to this driver. The `eoa' and `eof'
+ * determine the amount of hdf5 address space in use and the high-water mark
+ * of the file (the current size of the underlying memory).
+ */
+typedef struct H5FD_stream_t
+{
+ H5FD_t pub; /* public stuff, must be first */
+ H5FD_stream_fapl_t fapl; /* file access property list */
+ unsigned char *mem; /* the underlying memory */
+ haddr_t eoa; /* end of allocated region */
+ haddr_t eof; /* current allocated size */
+ int socket; /* socket to write / read from */
+ hbool_t dirty; /* flag indicating unflushed data */
+ hbool_t internal_socket; /* flag indicating an internal socket */
+} H5FD_stream_t;
+
+/* Allocate memory in multiples of this size (in bytes) by default */
+#define H5FD_STREAM_INCREMENT 8192
+
+/* default backlog argument for listen call */
+#define H5FD_STREAM_BACKLOG 1
+
+/* default file access property list */
+static const H5FD_stream_fapl_t default_fapl =
+{
+ H5FD_STREAM_INCREMENT, /* address space allocation blocksize */
+ -1, /* no external socket descriptor */
+ TRUE, /* enable I/O on socket */
+ H5FD_STREAM_BACKLOG, /* default backlog for listen(2) */
+ NULL, /* do not broadcast received files */
+ NULL /* argument to READ broadcast routine */
+};
+
+/*
+ * These macros check for overflow of various quantities. These macros
+ * assume that file_offset_t is signed and haddr_t and size_t are unsigned.
+ *
+ * ADDR_OVERFLOW: Checks whether a file address of type `haddr_t'
+ * is too large to be represented by the second argument
+ * of the file seek function.
+ *
+ * SIZE_OVERFLOW: Checks whether a buffer size of type `hsize_t' is too
+ * large to be represented by the `size_t' type.
+ *
+ * REGION_OVERFLOW: Checks whether an address and size pair describe data
+ * which can be addressed entirely in memory.
+ */
+#ifdef HAVE_LSEEK64
+# define file_offset_t off64_t
+#else
+# define file_offset_t off_t
+#endif
+#define MAXADDR (((haddr_t)1<<(8*sizeof(file_offset_t)-1))-1)
+#define ADDR_OVERFLOW(A) (HADDR_UNDEF==(A) || \
+ ((A) & ~(haddr_t)MAXADDR))
+#define SIZE_OVERFLOW(Z) ((Z) & ~(hsize_t)MAXADDR)
+#define REGION_OVERFLOW(A,Z) (ADDR_OVERFLOW(A) || SIZE_OVERFLOW(Z) || \
+ HADDR_UNDEF==(A)+(Z) || \
+ (size_t)((A)+(Z))<(size_t)(A))
+
+/* Function prototypes */
+static void *H5FD_stream_fapl_get (H5FD_t *_stream);
+static H5FD_t *H5FD_stream_open (const char *name, unsigned flags,
+ hid_t fapl_id, haddr_t maxaddr);
+static herr_t H5FD_stream_flush (H5FD_t *_stream);
+static herr_t H5FD_stream_close (H5FD_t *_stream);
+static haddr_t H5FD_stream_get_eoa (H5FD_t *_stream);
+static herr_t H5FD_stream_set_eoa (H5FD_t *_stream, haddr_t addr);
+static haddr_t H5FD_stream_get_eof (H5FD_t *_stream);
+static herr_t H5FD_stream_read (H5FD_t *_stream, hid_t fapl_id, haddr_t addr,
+ hsize_t size, void *buf);
+static herr_t H5FD_stream_write (H5FD_t *_stream, H5FD_mem_t type,
+ hid_t fapl_id, haddr_t addr,
+ hsize_t size, const void *buf);
+
+/* The Stream VFD's class information structure */
+static const H5FD_class_t H5FD_stream_g =
+{
+ "stream", /* name */
+ MAXADDR, /* maxaddr */
+ NULL, /* sb_size */
+ NULL, /* sb_encode */
+ NULL, /* sb_decode */
+ sizeof (H5FD_stream_fapl_t), /* fapl_size */
+ H5FD_stream_fapl_get, /* fapl_get */
+ NULL, /* fapl_copy */
+ NULL, /* fapl_free */
+ 0, /* dxpl_size */
+ NULL, /* dxpl_copy */
+ NULL, /* dxpl_free */
+ H5FD_stream_open, /* open */
+ H5FD_stream_close, /* close */
+ NULL, /* query */
+ NULL, /* cmp */
+ NULL, /* alloc */
+ NULL, /* free */
+ H5FD_stream_get_eoa, /* get_eoa */
+ H5FD_stream_set_eoa, /* set_eoa */
+ H5FD_stream_get_eof, /* get_eof */
+ H5FD_stream_read, /* read */
+ H5FD_stream_write, /* write */
+ H5FD_stream_flush, /* flush */
+ H5FD_FLMAP_SINGLE, /* fl_map */
+};
+
+/* Interface initialization */
+#define INTERFACE_INIT H5FD_stream_init
+static intn interface_initialize_g = 0;
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_init
+ *
+ * Purpose: Initialize this driver by registering it with the library.
+ *
+ * Return: Success: The driver ID for the Stream driver.
+ * Failure: Negative.
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+hid_t H5FD_stream_init (void)
+{
+ FUNC_ENTER (H5FD_stream_init, FAIL);
+
+ if (H5I_VFL != H5Iget_type (H5FD_STREAM_g))
+ H5FD_STREAM_g = H5FDregister (&H5FD_stream_g);
+
+ FUNC_LEAVE (H5FD_STREAM_g);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5Pset_fapl_stream
+ *
+ * Purpose: Modify the file access property list to use the Stream
+ * driver defined in this source file. The INCREMENT specifies
+ * how much to grow the memory each time we need more.
+ * If a valid SOCKET argument is given this will be used
+ * by the driver instead of parsing the 'hostname:port' filename
+ * and opening a socket internally.
+ *
+ * Return: Success: Non-negative
+ * Failure: Negative
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+herr_t H5Pset_fapl_stream (hid_t fapl_id, H5FD_stream_fapl_t *fapl)
+{
+ herr_t result;
+ H5FD_stream_fapl_t user_fapl;
+
+
+ FUNC_ENTER (H5FD_set_fapl_stream, FAIL);
+ H5TRACE2 ("e", "ix", fapl_id, fapl);
+
+ if (H5P_FILE_ACCESS != H5Pget_class (fapl_id))
+ HRETURN_ERROR (H5E_PLIST, H5E_BADTYPE, FAIL, "not a fapl");
+
+ if (fapl)
+ {
+ if (! fapl->do_socket_io && fapl->broadcast_fn == NULL)
+ HRETURN_ERROR (H5E_ARGS, H5E_BADVALUE, FAIL,
+ "read broadcast function pointer is NULL");
+
+ user_fapl = *fapl;
+ if (fapl->increment == 0)
+ {
+ user_fapl.increment = H5FD_STREAM_INCREMENT;
+ }
+ result = H5Pset_driver (fapl_id, H5FD_STREAM, &user_fapl);
+ }
+ else
+ {
+ result = H5Pset_driver (fapl_id, H5FD_STREAM, &default_fapl);
+ }
+
+ FUNC_LEAVE (result);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5Pget_fapl_stream
+ *
+ * Purpose: Queries properties set by the H5Pset_fapl_stream() function.
+ *
+ * Return: Success: Non-negative
+ * Failure: Negative
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+herr_t H5Pget_fapl_stream(hid_t fapl_id, H5FD_stream_fapl_t *fapl /* out */)
+{
+ H5FD_stream_fapl_t *this_fapl;
+
+
+ FUNC_ENTER (H5Pget_fapl_stream, FAIL);
+ H5TRACE2("e","ix",fapl_id,fapl);
+
+ if (H5P_FILE_ACCESS != H5Pget_class (fapl_id))
+ HRETURN_ERROR (H5E_PLIST, H5E_BADTYPE, FAIL, "not a fapl");
+ if (H5FD_STREAM != H5Pget_driver (fapl_id))
+ HRETURN_ERROR (H5E_PLIST, H5E_BADVALUE, FAIL, "incorrect VFL driver");
+ if (NULL == (this_fapl = H5Pget_driver_info (fapl_id)))
+ HRETURN_ERROR (H5E_PLIST, H5E_BADVALUE, FAIL, "bad VFL driver info");
+
+ if (fapl)
+ *fapl = *this_fapl;
+
+ FUNC_LEAVE (SUCCEED);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_fapl_get
+ *
+ * Purpose: Returns a copy of the file access properties
+ *
+ * Return: Success: Ptr to new file access properties
+ * Failure: NULL
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static void *H5FD_stream_fapl_get (H5FD_t *_stream)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+ H5FD_stream_fapl_t *fapl;
+
+
+ FUNC_ENTER (H5FD_stream_fapl_get, NULL);
+
+ if ((fapl = H5MM_calloc (sizeof (H5FD_stream_fapl_t))) == NULL)
+ HRETURN_ERROR (H5E_RESOURCE, H5E_NOSPACE, NULL, "memory allocation failed");
+
+ *fapl = stream->fapl;
+
+ FUNC_LEAVE (fapl);
+}
+
+
+static int H5FDstream_open_socket(const char *filename, int o_flags,
+ unsigned int backlog,
+ const char **errormsg,
+ H5E_major_t *major, H5E_minor_t *minor)
+{
+ struct sockaddr_in sin;
+ struct hostent *he;
+ unsigned short int port;
+ int sock;
+ char *hostname;
+ const char *separator, *tmp;
+ const int on = 1;
+
+
+ sock = -1;
+ *errormsg = NULL;
+
+ /* Parse "hostname:port" from filename argument */
+ for (separator = filename; *separator != ':' && *separator; separator++);
+ if (separator == filename || !*separator)
+ {
+ *errormsg = "invalid host address";
+ }
+ else
+ {
+ tmp = separator;
+ if (! tmp [1])
+ {
+ *errormsg = "no port number";
+ }
+ while (*errormsg == NULL && *++tmp)
+ {
+ if (! isdigit (*tmp))
+ {
+ *errormsg = "invalid port number";
+ }
+ }
+ }
+
+ /* Return if parsing the filename failed */
+ if (*errormsg)
+ {
+ *major = H5E_ARGS; *minor = H5E_BADVALUE;
+ return (sock);
+ }
+
+ hostname = (char *) H5MM_malloc (separator - filename + 1);
+
+ /* Return if out of memory */
+ if (hostname == NULL)
+ {
+ *major = H5E_RESOURCE; *minor = H5E_NOSPACE;
+ *errormsg = "memory allocation failed";
+ return (sock);
+ }
+
+ strncpy (hostname, filename, separator - filename);
+ hostname [separator - filename] = 0;
+ port = atoi (separator + 1);
+
+ memset (&sin, 0, sizeof (sin));
+ sin.sin_family = AF_INET;
+ sin.sin_port = htons (port);
+
+ if (! (he = gethostbyname (hostname)))
+ *errormsg = "unable to get host address";
+ else if ((sock = socket (AF_INET, SOCK_STREAM, 0)) < 0)
+ *errormsg = "unable to open socket";
+
+ if (*errormsg == NULL)
+ {
+ if (O_RDONLY == o_flags)
+ {
+ memcpy (&sin.sin_addr, he->h_addr, he->h_length);
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: connecting to host '%s' port %d\n",
+ hostname, port);
+#endif
+ if (connect (sock, (struct sockaddr *) &sin, sizeof (sin)) < 0)
+ *errormsg = "unable to connect";
+ }
+ else
+ {
+ sin.sin_addr.s_addr = INADDR_ANY;
+ if (fcntl (sock, F_SETFL, O_NONBLOCK) < 0)
+ *errormsg = "unable to set non-blocking mode for socket";
+ else if (setsockopt (sock, IPPROTO_TCP, TCP_NODELAY, &on, sizeof(on)) < 0)
+ *errormsg = "unable to set socket option TCP_NODELAY";
+ else if (setsockopt (sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) < 0)
+ *errormsg = "unable to set socket option SO_REUSEADDR";
+ else if (bind (sock, (struct sockaddr *) &sin, sizeof (sin)) < 0)
+ *errormsg = "unable to bind socket";
+ else if (listen (sock, backlog) < 0)
+ *errormsg = "unable to listen on socket";
+ }
+ }
+
+ H5MM_xfree (hostname);
+
+ /* Return if opening the socket failed */
+ if (*errormsg)
+ {
+ if (sock >= 0)
+ {
+ close (sock);
+ sock = -1;
+ }
+ *major = H5E_FILE; *minor = H5E_CANTOPENFILE;
+ }
+
+ return (sock);
+}
+
+
+static void H5FDstream_read_from_socket (H5FD_stream_t *stream,
+ const char **errormsg,
+ H5E_major_t *major, H5E_minor_t *minor)
+{
+ ssize_t size;
+ size_t max_size = 0;
+ unsigned char *ptr;
+
+
+ *errormsg = NULL;
+ stream->eof = 0;
+ stream->mem = NULL;
+
+ while (1)
+ {
+ if (max_size <= 0)
+ {
+ /*
+ * Allocate initial buffer as increment + 1
+ * to prevent unnecessary reallocation
+ * if increment is exactly a multiple of the filesize
+ */
+ max_size = stream->fapl.increment;
+ if (! stream->mem)
+ max_size++;
+ ptr = H5MM_realloc (stream->mem, stream->eof + max_size);
+ if (! ptr)
+ {
+ *major = H5E_RESOURCE; *minor = H5E_NOSPACE;
+ *errormsg = "unable to allocate file space buffer";
+ return;
+ }
+ stream->mem = ptr;
+ ptr += stream->eof;
+ }
+
+ /* now receive the next chunk of data */
+ size = read (stream->socket, ptr, max_size);
+ if (size < 0 && (EINTR == errno || EAGAIN == errno))
+ continue;
+ if (size < 0)
+ {
+ *major = H5E_IO; *minor = H5E_READERROR;
+ *errormsg = "error reading from file from socket";
+ return;
+ }
+ if (! size)
+ break;
+ max_size -= (size_t) size;
+ stream->eof += (haddr_t) size;
+ ptr += size;
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: read %d bytes (%d total) from socket\n",
+ (int) size, (int) stream->eof);
+#endif
+ }
+
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: read total of %d bytes from socket\n",
+ (int) stream->eof);
+#endif
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_open
+ *
+ * Purpose: Opens an HDF5 file in memory.
+ *
+ * Return: Success: A pointer to a new file data structure. The
+ * public fields will be initialized by the
+ * caller, which is always H5FD_open().
+ * Failure: NULL
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static H5FD_t *H5FD_stream_open (const char *filename,
+ unsigned flags,
+ hid_t fapl_id,
+ haddr_t maxaddr)
+{
+ H5FD_stream_t _stream, *stream;
+ const H5FD_stream_fapl_t *fapl;
+ int o_flags;
+ H5E_major_t major;
+ H5E_minor_t minor;
+ const char *errormsg;
+
+
+ FUNC_ENTER (H5FD_stream_open, NULL);
+
+ /* Check arguments */
+ if (filename == NULL|| *filename == '\0')
+ HRETURN_ERROR (H5E_ARGS, H5E_BADVALUE, NULL,"invalid file name");
+ if (maxaddr == 0 || HADDR_UNDEF == maxaddr)
+ HRETURN_ERROR (H5E_ARGS, H5E_BADRANGE, NULL, "bogus maxaddr");
+ if (ADDR_OVERFLOW (maxaddr))
+ HRETURN_ERROR (H5E_ARGS, H5E_OVERFLOW, NULL, "maxaddr overflow");
+
+ /* Build the open flags */
+ o_flags = (H5F_ACC_RDWR & flags) ? O_RDWR : O_RDONLY;
+ if (H5F_ACC_TRUNC & flags) o_flags |= O_TRUNC;
+ if (H5F_ACC_CREAT & flags) o_flags |= O_CREAT;
+ if (H5F_ACC_EXCL & flags) o_flags |= O_EXCL;
+
+ if ((O_RDWR & o_flags) && ! (O_CREAT & o_flags))
+ HRETURN_ERROR (H5E_ARGS, H5E_UNSUPPORTED, NULL,
+ "open stream for read/write not supported");
+
+ if (H5P_DEFAULT != fapl_id)
+ fapl = H5Pget_driver_info (fapl_id);
+ if (fapl == NULL)
+ fapl = &default_fapl;
+
+ /* zero out file structure and set file access property list */
+ memset (&_stream, 0, sizeof (_stream));
+ _stream.fapl = *fapl;
+
+ errormsg = NULL;
+
+ /* if an external socket is provided with the file access property list
+ we use that, otherwise the filename argument is parsed and a socket
+ is opened internally */
+ if (fapl->do_socket_io)
+ {
+ if (fapl->socket >= 0)
+ {
+ _stream.internal_socket = FALSE;
+ _stream.socket = fapl->socket;
+ }
+ else
+ {
+ _stream.internal_socket = TRUE;
+ _stream.socket = H5FDstream_open_socket (filename, o_flags, fapl->backlog,
+ &errormsg, &major, &minor);
+ }
+ }
+ else
+ {
+ _stream.socket = -1;
+ }
+
+ /* read the data from socket into memory */
+ if (O_RDONLY == o_flags)
+ {
+ if (errormsg == NULL && fapl->do_socket_io)
+ {
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: reading file from socket\n");
+#endif
+ H5FDstream_read_from_socket (&_stream, &errormsg, &major, &minor);
+ }
+
+ /* Now call the user's broadcast routine if given */
+ if (fapl->broadcast_fn)
+ {
+ if ((fapl->broadcast_fn) (&_stream.mem, &_stream.eof,
+ fapl->broadcast_arg) < 0)
+ {
+ /* don't override previous error codes */
+ if (errormsg == NULL)
+ {
+ major = H5E_IO; minor = H5E_READERROR;
+ errormsg = "broadcast error";
+ }
+ }
+
+ /* check for filesize of zero bytes */
+ if (errormsg == NULL && _stream.eof == 0)
+ {
+ major = H5E_IO; minor = H5E_READERROR;
+ errormsg = "zero filesize";
+ }
+ }
+
+ /* For files which are read from a socket:
+ the opened socket is not needed anymore */
+ if (errormsg == NULL)
+ {
+ if (_stream.internal_socket && _stream.socket >= 0)
+ close (_stream.socket);
+ _stream.socket = -1;
+ }
+ }
+
+ /* Create the new file struct */
+ stream = NULL;
+ if (errormsg == NULL)
+ {
+ stream = (H5FD_stream_t *) H5MM_calloc (sizeof (H5FD_stream_t));
+ if (stream == NULL)
+ {
+ major = H5E_RESOURCE; minor = H5E_NOSPACE;
+ errormsg = "unable to allocate file struct";
+ }
+ else
+ {
+ *stream = _stream;
+ }
+ }
+
+ if (errormsg)
+ {
+ if (_stream.mem)
+ H5MM_xfree (_stream.mem);
+ if (_stream.internal_socket && _stream.socket >= 0)
+ close (_stream.socket);
+ HRETURN_ERROR (major, minor, NULL, errormsg);
+ }
+
+ FUNC_LEAVE ((H5FD_t *) stream);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_flush
+ *
+ * Purpose: Flushes the file via sockets to any connected clients
+ * if its dirty flag is set.
+ *
+ * Return: Success: 0
+ * Failure: -1
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t H5FD_stream_flush (H5FD_t *_stream)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+ haddr_t size;
+ unsigned char *ptr;
+ struct sockaddr from;
+#if !(defined(linux) || defined(sun))
+ typedef int socklen_t;
+#endif
+ socklen_t fromlen;
+ int sock;
+
+
+ FUNC_ENTER (H5FD_stream_flush, FAIL);
+
+ /* Write to backing store */
+ if (stream->dirty && stream->socket >= 0)
+ {
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: accepting client connections\n");
+#endif
+ fromlen = sizeof (from);
+ while ((sock = accept (stream->socket, &from, &fromlen)) >= 0)
+ {
+ if (fcntl (sock, F_SETFL, O_NONBLOCK) < 0)
+ {
+ close (sock);
+ continue; /* continue the loop for other clients' requests */
+ }
+
+ size = stream->eof;
+ ptr = stream->mem;
+
+ while (size)
+ {
+ ssize_t n = write (sock, ptr, size);
+ if (n < 0 && (EINTR == errno || EAGAIN == errno))
+ continue;
+ if (n < 0)
+ {
+ close (sock);
+ /* FIXME: continue the loop for other clients here */
+ HRETURN_ERROR (H5E_IO, H5E_WRITEERROR, FAIL,
+ "error writing to socket");
+ }
+ ptr += (size_t) n;
+ size -= (size_t) n;
+#ifdef DEBUG
+ fprintf (stderr, "Stream VFD: wrote %d bytes to socket, %d in total, "
+ "%d left\n", (int) n, ptr - stream->mem, (int) size);
+#endif
+ }
+ close (sock);
+ }
+ stream->dirty = FALSE;
+ }
+
+ FUNC_LEAVE (SUCCEED);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_close
+ *
+ * Purpose: Closes the file.
+ *
+ * Return: Success: 0
+ * Failure: -1
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t H5FD_stream_close (H5FD_t *_stream)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+
+
+ FUNC_ENTER (H5FD_stream_close, FAIL);
+
+ /* Flush */
+ if (H5FD_stream_flush (_stream) < 0)
+ HRETURN_ERROR (H5E_FILE, H5E_CANTFLUSH, FAIL, "unable to flush file");
+
+ /* Release resources */
+ if (stream->socket >= 0 && stream->internal_socket)
+ {
+ close (stream->socket);
+ }
+ if (stream->mem)
+ {
+ H5MM_xfree (stream->mem);
+ }
+ memset (stream, 0, sizeof (H5FD_stream_t));
+
+ FUNC_LEAVE (0);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_get_eoa
+ *
+ * Purpose: Gets the end-of-address marker for the file. The EOA marker
+ * is the first address past the last byte allocated in the
+ * format address space.
+ *
+ * Return: Success: The end-of-address marker.
+ * Failure: HADDR_UNDEF
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static haddr_t H5FD_stream_get_eoa (H5FD_t *_stream)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+
+
+ FUNC_ENTER (H5FD_stream_get_eoa, HADDR_UNDEF);
+
+ FUNC_LEAVE (stream->eoa);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_set_eoa
+ *
+ * Purpose: Set the end-of-address marker for the file. This function is
+ * called shortly after an existing HDF5 file is opened in order
+ * to tell the driver where the end of the HDF5 data is located.
+ *
+ * Return: Success: 0
+ * Failure: -1
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t H5FD_stream_set_eoa (H5FD_t *_stream,
+ haddr_t addr)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+
+
+ FUNC_ENTER (H5FD_stream_set_eoa, FAIL);
+
+ if (ADDR_OVERFLOW (addr))
+ HRETURN_ERROR (H5E_ARGS, H5E_OVERFLOW, FAIL, "address overflow");
+
+ stream->eoa = addr;
+
+ FUNC_LEAVE (0);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_get_eof
+ *
+ * Purpose: Returns the end-of-file marker, which is the greater of
+ * either the size of the underlying memory or the HDF5
+ * end-of-address markers.
+ *
+ * Return: Success: End of file address, the first address past
+ * the end of the "file", either the memory
+ * or the HDF5 file.
+ * Failure: HADDR_UNDEF
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static haddr_t H5FD_stream_get_eof (H5FD_t *_stream)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+
+
+ FUNC_ENTER (H5FD_stream_get_eof, HADDR_UNDEF);
+
+ FUNC_LEAVE (MAX (stream->eof, stream->eoa));
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_read
+ *
+ * Purpose: Reads SIZE bytes of data from FILE beginning at address ADDR
+ * into buffer BUF according to data transfer properties in
+ * DXPL_ID.
+ *
+ * Return: Success: 0
+ * Result is stored in caller-supplied buffer BUF
+ * Failure: -1
+ * Contents of buffer BUF are undefined
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t H5FD_stream_read (H5FD_t *_stream,
+ hid_t UNUSED dxpl_id,
+ haddr_t addr,
+ hsize_t size,
+ void *buf /*out*/)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+ ssize_t nbytes;
+
+
+ FUNC_ENTER (H5FD_stream_read, FAIL);
+
+ assert (stream && stream->pub.cls);
+ assert (buf);
+
+ /* Check for overflow conditions */
+ if (HADDR_UNDEF == addr)
+ HRETURN_ERROR (H5E_IO, H5E_OVERFLOW, FAIL, "file address overflowed");
+ if (REGION_OVERFLOW (addr, size))
+ HRETURN_ERROR (H5E_IO, H5E_OVERFLOW, FAIL, "file address overflowed");
+ if (addr + size > stream->eoa)
+ HRETURN_ERROR (H5E_IO, H5E_OVERFLOW, FAIL, "file address overflowed");
+
+ /* Read the part which is before the EOF marker */
+ if (addr < stream->eof)
+ {
+ nbytes = MIN (size, stream->eof - addr);
+ memcpy (buf, stream->mem + addr, nbytes);
+ size -= nbytes;
+ addr += nbytes;
+ buf = (char *) buf + nbytes;
+ }
+
+ /* Read zeros for the part which is after the EOF markers */
+ if (size > 0)
+ memset (buf, 0, size);
+
+ FUNC_LEAVE (SUCCEED);
+}
+
+
+/*-------------------------------------------------------------------------
+ * Function: H5FD_stream_write
+ *
+ * Purpose: Writes SIZE bytes of data to FILE beginning at address ADDR
+ * from buffer BUF according to data transfer properties in
+ * DXPL_ID.
+ *
+ * Return: Success: Zero
+ * Failure: -1
+ *
+ * Programmer: Thomas Radke
+ * Tuesday, September 12, 2000
+ *
+ * Modifications:
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t H5FD_stream_write (H5FD_t *_stream,
+ H5FD_mem_t UNUSED type,
+ hid_t UNUSED dxpl_id,
+ haddr_t addr,
+ hsize_t size,
+ const void *buf)
+{
+ H5FD_stream_t *stream = (H5FD_stream_t *) _stream;
+
+
+ FUNC_ENTER (H5FD_stream_write, FAIL);
+
+ assert (stream && stream->pub.cls);
+ assert (buf);
+
+ /* Check for overflow conditions */
+ if (REGION_OVERFLOW (addr, size))
+ HRETURN_ERROR (H5E_IO, H5E_OVERFLOW, FAIL, "file address overflowed");
+ if (addr + size > stream->eoa)
+ HRETURN_ERROR (H5E_IO, H5E_OVERFLOW, FAIL, "file address overflowed");
+
+ /*
+ * Allocate more memory if necessary, careful of overflow. Also, if the
+ * allocation fails then the file should remain in a usable state. Be
+ * careful of non-Posix realloc() that doesn't understand what to do when
+ * the first argument is null.
+ */
+ if (addr + size > stream->eof)
+ {
+ unsigned char *x;
+ haddr_t new_eof = stream->fapl.increment *
+ ((addr+size) / stream->fapl.increment);
+
+
+ if ((addr+size) % stream->fapl.increment)
+ {
+ new_eof += stream->fapl.increment;
+ }
+ if (stream->mem == NULL)
+ {
+ x = H5MM_malloc (new_eof);
+ }
+ else
+ {
+ x = H5MM_realloc (stream->mem, new_eof);
+ }
+ if (x == NULL)
+ HRETURN_ERROR (H5E_RESOURCE, H5E_NOSPACE, FAIL,
+ "unable to allocate memory block");
+ stream->mem = x;
+ stream->eof = new_eof;
+ }
+
+ /* Write from BUF to memory */
+ memcpy (stream->mem + addr, buf, size);
+ stream->dirty = TRUE;
+
+ FUNC_LEAVE (SUCCEED);
+}
+
+#endif /* H5_HAVE_STREAM */
diff --git a/src/H5FDstream.h b/src/H5FDstream.h
new file mode 100644
index 0000000..c8acd65
--- /dev/null
+++ b/src/H5FDstream.h
@@ -0,0 +1,53 @@
+/*
+ * Copyright © 2000 The author.
+ * The author prefers this code not be used for military purposes.
+ *
+ *
+ * Author: Thomas Radke <tradke@aei-potsdam.mpg.de>
+ * Tuesday, September 12, 2000
+ *
+ * Purpose: The public header file for the Stream Virtual File Driver.
+ */
+#ifndef H5FDstream_H
+#define H5FDstream_H
+
+#ifdef H5_HAVE_STREAM
+
+#include <H5Ipublic.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define H5FD_STREAM (H5FD_stream_init())
+
+/* prototype for read broadcast callback routine */
+typedef int (*H5FD_stream_broadcast_t) (unsigned char **file,
+ haddr_t *len,
+ void *arg);
+
+/* Driver-specific file access properties */
+typedef struct H5FD_stream_fapl_t
+{
+ size_t increment; /* how much to grow memory in reallocs */
+ int socket; /* external socket descriptor */
+ hbool_t do_socket_io; /* do I/O on socket */
+ unsigned int backlog; /* backlog argument for listen call */
+ H5FD_stream_broadcast_t broadcast_fn; /* READ broadcast callback */
+ void *broadcast_arg; /* READ broadcast callback user argument*/
+} H5FD_stream_fapl_t;
+
+
+__DLL__ hid_t H5FD_stream_init (void);
+__DLL__ herr_t H5Pset_fapl_stream (hid_t fapl_id,
+ H5FD_stream_fapl_t *fapl);
+__DLL__ herr_t H5Pget_fapl_stream (hid_t fapl_id,
+ H5FD_stream_fapl_t *fapl /*out*/ );
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* H5_HAVE_STREAM */
+
+#endif /* H5FDstream_H */