Changeset 45

Timestamp:

06/30/08 11:57:16 (5 months ago)

Author:

jtang

Message:

Add sipc_ioctl() call to enable non-blocking read.
(This does not address interrupted read/writes, nor race condition during read, nor Java bindings.)

Files:

• trunk/libsipc/include/sipc/sipc.h (modified) (3 diffs)
• trunk/libsipc/man/man3/sipc_ioctl.3 (added)
• trunk/libsipc/man/man3/sipc_open.3 (modified) (2 diffs)
• trunk/libsipc/man/man3/sipc_recv_data.3 (modified) (3 diffs)
• trunk/libsipc/src/libsipc.map (modified) (1 diff)
• trunk/libsipc/src/sipc.c (modified) (2 diffs)
• trunk/libsipc/src/sipc_internal.h (modified) (1 diff)
• trunk/libsipc/src/sipc_mqueue.c (modified) (7 diffs)
• trunk/libsipc/src/sipc_shm.c (modified) (2 diffs)
• trunk/libsipc/tests/mqueue.c (modified) (4 diffs)
• trunk/libsipc/tests/shm.c (modified) (3 diffs)

trunk/libsipc/include/sipc/sipc.h

@@ -40 +40 @@
-#define SIPC_NUM_TYPES 2
-
+/* SIPC behaviors, for sipc_ioctl() */
+#define SIPC_BLOCK 0
+#define SIPC_NOBLOCK 1
+
-/* sipc_open(char *key, int role, int ipc_type, size_t size)
- *      key             Unique key to identify IPC communication channel,
@@ -62 +66 @@
-void sipc_unlink(const char *key, int ipc_type);
-
+/* Modify the behavior of an open SIPC channel.  Returns 0 on success,
+ * <0 on failure.
+ */
+int sipc_ioctl(sipc_t *sipc, int request);
+
-/* Blocking call to send msglen bytes of data.
- * This can only be called if sender was specified when sipc_init was called.
@@ -67 +76 @@
-int sipc_send_data(sipc_t *sipc, int msg_len);
-
-Blocking call to recieve data.  Data will be allocated and filled and
-len will be set to the length. Returns 0 on success, <0 on failure
+Call to receive data.  Data will be allocated and filled and len
+will be set to the length.  Returns 0 on success, <0 on failure.
-int sipc_recv_data(sipc_t *sipc, char **data, int *len);
-

trunk/libsipc/man/man3/sipc_open.3

@@ -1 @@
-, 2007
+ - 2008
-.\" This file is distributed according to the GNU Lesser General Public License
-.TH "SIPC_OPEN" 3 "2006-08-10" "Linux 2.6" "Linux Programmer's Manual"
@@ -43 +43 @@
-This function returns a pointer to the new IPC handle on success, NULL on error.
-.SH "SEE ALSO"
-
+, \fBsipc_ioctl(3)\fR
-.SH AUTHOR
-David Windsor <dwindsor@tresys.com>

trunk/libsipc/man/man3/sipc_recv_data.3

@@ -24 +24 @@
-.TP
-.I len
-This will hold
+Pointer to
-.PP
-If the IPC channel is a message queue, \fBsipc_recv_data\fR()
@@ -41 +41 @@
-\fBsipc_recv_data\fR(), else undefined behavior results.
-.P
-
+
+
-.SH "RETURN VALUE"
-
-
+
-.I *data
- and
+,
-.I msg_len
-
+
+
+
+
+
+
+
+
+
+
+
+
-.SH "EXAMPLES"
-.nf
@@ -61 +72 @@
-.fi
-.SH "SEE ALSO"
-
+, \fBsipc_ioctl(3)\fR
-.SH AUTHOR
-David Windsor <dwindsor@tresys.com>

trunk/libsipc/src/libsipc.map

@@ -11 +11 @@
-  local: *;
-};
+
+VERS_1.1{
+  global:
+        sipc_ioctl;
+};

trunk/libsipc/src/sipc.c

@@ -53 +53 @@
-        new_sipc->ipc_type = ipc_type;
-        new_sipc->role = role;
+        new_sipc->non_blocking = 0;
-        new_sipc->len = len;
-        new_sipc->msg_len = SIPC_MSGLEN_NOT_SET;
@@ -116 +117 @@
-}
-
+int sipc_ioctl(sipc_t *sipc, int request)
+{
+        if (!sipc) {
+                errno = EINVAL;
+                return -1;
+        }
+
+        switch (request) {
+        case SIPC_BLOCK:
+                sipc->non_blocking = 0;
+                break;
+        case SIPC_NOBLOCK:
+                sipc->non_blocking = IPC_NOWAIT;
+                break;
+        default:
+                errno = EINVAL;
+                return -1;
+        }
+        return 0;
+}
+
-char *sipc_get_data_ptr(sipc_t *sipc)
-{

trunk/libsipc/src/sipc_internal.h

@@ -76 +76 @@
-        int ipc_type;
-        int role;
+        int non_blocking;              /* set to IPC_NOWAIT if reads should be
+                                          nonblocking, 0 to block (default) */
-        union
-        {

trunk/libsipc/src/sipc_mqueue.c

@@ -193 +193 @@
-        int retv = -1, alloc_sz = 0, idx = 0, recv_sz;
-        int msgtxt_sz = SIPC_MQUEUE_MSG_SZ - sizeof(struct msgbuf);
-
-
+
+
+
+
+
+
+
+
-                goto err;
-
-
+
+
-        idx = 0;                       /* Current index into the buffer */
-        *len = 0;                      /* Bytes received so far */
-
-        if (sipc->role != SIPC_RECEIVER) {
+                error = EBADF;
-                sipc_error(sipc, "sipc_send_data called without receiver flag enabled in IPC structure\n");
-                goto err;
@@ -212 +218 @@
-        /* Receive and validate the length of message marker */
-        if (sipc->msg_len == SIPC_MSGLEN_NOT_SET) {
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-                        return 0;
+                }
-
-                if (is_msg_len(mbuf)) {
-                        sipc->msg_len = atoi(mbuf->mtext);
-                        if (sipc->msg_len < 0) {
+                                error = EIO;
-                                sipc_error(sipc, "libsipc: bad message length\n");
-                                goto err;
-                        }
-                        if (sipc->msg_len > sipc->len) {
+                                error = EIO;
-                                sipc_error(sipc, "libsipc: cannot receive buffer of size %zd, can only receive %zd\n",
-                                           sipc->msg_len, sipc->len);
@@ -231 +245 @@
-                        }
-                } else {
+                        error = EIO;
-                        sipc_error(sipc, "libsipc: did not receive length of message, received %ld\n", mbuf->mtype);
-                        goto err;
@@ -239 +254 @@
-         * Stop when we have received the total number of bytes
-         * or upon receiving the end of message marker. */
-
-
+
+
+
+
+
+
-                goto err;
-        }
@@ -253 +272 @@
-                char *tdata = realloc(*data, alloc_sz);
-                if (!tdata) {
+                        error = errno;
-                        sipc_error(sipc, "%s\n", "Out of memory");
-                        free(*data);
-                        *data = NULL;
-                        *len = 0;
-                        goto err;
-                }
@@ -266 +283 @@
-
-                /* Receive the next message */
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-                        goto err;
-                }
@@ -278 +301 @@
-      err:
-        sipc->msg_len = SIPC_MSGLEN_NOT_SET;    /* Unset message length */
+        errno = error;
+        if (retv) {
+                free(*data);
+                *data = NULL;
+                *len = 0;
+        }
-        return retv;
-}

trunk/libsipc/src/sipc_shm.c

@@ -188 +188 @@
-/* Receive a DATA_READY marker from sender and pass the shm pointer to
- * the caller.  This function will block if there is nothing on the
-
+, if the handle is set to blocking mode (default)
- */
-int sipc_shm_recv_data(sipc_t *sipc, char **data, int *len)
-{
-        int mtype = 0;
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
-
-        /* Get a message from the side channel. */
-MQ_BLOCK
+block
-        if (mtype == SIPC_DATA_READY) {
-                /* It is now OK to read shared memory */
@@ -208 +213 @@
-                return 0;
-        } else {
-
-
-
-
+
+
+
+
+
+
+
+
+
+

trunk/libsipc/tests/mqueue.c

@@ -48 +48 @@
-static void do_parent_chunked(void);
-static void do_child_chunked(void);
+static void do_blocked_parent(void);
+static void do_blocked_child(void);
-static int verify_data(char *filename, char *recv_data);
-static inline int is_end_xmit(char *data);
@@ -229 +231 @@
-}
-
+static void test_blocked_mqueue(void)
+{
+        pid_t pid;
+        int status = 0;
+
+        switch ((pid = fork())) {
+        case -1:
+                fprintf(stderr, "fork: %s\n", strerror(errno));
+                return;
+        case 0:
+                sipc_close(reader2_ipc);
+                do_blocked_child();
+                exit(0);
+                break;
+        default:
+                do_blocked_parent();
+                break;
+        }
+
+        wait(&status);
+}
+
-CU_TestInfo mqueue_tests[] = {
-        {"unforked mqueue", test_unforked_mqueue}
@@ -235 +259 @@
-        ,
-        {"forked chunked mqueue", test_chunked_mqueue}
+        ,
+        {"forked blocked mqueue", test_blocked_mqueue}
-        ,
-        CU_TEST_INFO_NULL
@@ -341 +367 @@
-}
-
+/*
+ * Test blocking and non-blocking mode.
+ */
+static void do_blocked_child(void)
+{
+        char *shm_data = sipc_get_data_ptr(writer_ipc);
+        if (!shm_data) {
+                sipc_error(writer_ipc, "Unable to get pointer to data.\n");
+                return;
+        }
+        sleep(2);
+        shm_data[0] = 0x42;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        sleep(2);
+        shm_data[0] = 0x43;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        shm_data[0] = 0x44;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        sipc_close(writer_ipc);
+}
+
+static void do_blocked_parent(void)
+{
+        char *data = NULL;
+        int len = 0, retv, error;
+
+        printf("This should pause for 2 seconds (default read)...\n");
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == 1 && data[0] == 0x42);
+        free(data);
+
+        printf("No blocking here.\n");
+        CU_ASSERT(sipc_ioctl(reader_ipc, SIPC_NOBLOCK) == 0);
+        retv = sipc_recv_data(reader_ipc, &data, &len);
+        error = errno;
+        CU_ASSERT(retv == -1);
+        CU_ASSERT(data == NULL);
+        CU_ASSERT(len == 0);
+        CU_ASSERT(error == EAGAIN);
+
+        printf("This should pause for 3 seconds... (non-blocking read)\n");
+        sleep(3);
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == 1 && data[0] == 0x43);
+        free(data);
+
+        CU_ASSERT(sipc_ioctl(reader_ipc, -1) < 0);
+        CU_ASSERT(sipc_ioctl(reader_ipc, SIPC_BLOCK) == 0);
+
+        printf("This should pause for 1 second... (blocked read)\n");
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == 1 && data[0] == 0x44);
+        free(data);
+}
+
-static int send_end_xmit(sipc_t *ipc)
-{

trunk/libsipc/tests/shm.c

@@ -46 +46 @@
-static void do_binary_child(void);
-static void do_binary_parent(void);
+static void do_blocked_child(void);
+static void do_blocked_parent(void);
-static void send_end_xmit(sipc_t *sipc);
-static int verify_data(char *filename, char *recv_data);
@@ -175 +177 @@
-}
-
+static void test_blocked_shm(void)
+{
+        int status;
+        pid_t pid;
+
+        switch (pid = fork()) {
+        case -1:
+                fprintf(stderr, "fork: %s\n", strerror(errno));
+                return;
+        case 0:
+                sipc_close(reader_ipc);
+                do_blocked_child();
+                /* Exit here to prevent displaying results 2x */
+                exit(0);
+                break;
+        default:
+                do_blocked_parent();
+                break;
+        }
+
+        wait(&status);
+}
+
-CU_TestInfo shm_tests[] = {
-        {"forked shm", test_shm}
-        ,
-        {"binary shm", test_binary_shm}
+        ,
+        {"blocked shm", test_blocked_shm}
-        ,
-        CU_TEST_INFO_NULL
@@ -313 +340 @@
-}
-
+/*
+ * Test blocking and non-blocking mode.
+ */
+static void do_blocked_child(void)
+{
+        char *shm_data = sipc_get_data_ptr(writer_ipc);
+        if (!shm_data) {
+                sipc_error(writer_ipc, "Unable to get pointer to data.\n");
+                return;
+        }
+        sleep(2);
+        shm_data[0] = 0x42;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        sleep(2);
+        shm_data[0] = 0x43;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        shm_data[0] = 0x44;
+        if (sipc_send_data(writer_ipc, 1) < 0) {
+                sipc_error(writer_ipc, "Unable to send data\n");
+                return;
+        }
+        sipc_close(writer_ipc);
+}
+
+static void do_blocked_parent(void)
+{
+        char *data = NULL;
+        int len = 0, retv, error;
+
+        printf("This should pause for 2 seconds (default read)...\n");
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == DATA_LEN && data[0] == 0x42);
+        sipc_shm_recv_done(reader_ipc);
+
+        printf("No blocking here.\n");
+        CU_ASSERT(sipc_ioctl(reader_ipc, SIPC_NOBLOCK) == 0);
+        retv = sipc_recv_data(reader_ipc, &data, &len);
+        error = errno;
+        CU_ASSERT(retv == -1);
+        CU_ASSERT(data == NULL);
+        CU_ASSERT(len == 0);
+        CU_ASSERT(error == EAGAIN);
+
+        printf("This should pause for 3 seconds... (non-blocking read)\n");
+        sleep(3);
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == DATA_LEN && data[0] == 0x43);
+        sipc_shm_recv_done(reader_ipc);
+
+        CU_ASSERT(sipc_ioctl(reader_ipc, -1) < 0);
+        CU_ASSERT(sipc_ioctl(reader_ipc, SIPC_BLOCK) == 0);
+
+        printf("This should pause for 1 second... (blocked read)\n");
+        if (sipc_recv_data(reader_ipc, &data, &len) < 0) {
+                CU_FAIL("Error during first read");
+        }
+        CU_ASSERT(len == DATA_LEN && data[0] == 0x44);
+        sipc_shm_recv_done(reader_ipc);
+}
+
-static void send_end_xmit(sipc_t *sipc)
-{