[Cbe-oss-dev] [PATCH 2/3] libspe2: Fix blocking in spe_in_mbox_write w/ SPE_MBOX_ANY_NONBLOCKING

[Kazunori Asayama]

The current implementation of spe_in_mbox_write in libspe2 has
possibility to be blocked even if SPE_MBOX_ANY_NONBLOCKING is passed
as behavior flag, when another thread calls spe_in_mbox_write with
SPE_MBOX_ALL_BLOCKING at the same time (*see the scenario below).

To solve this problem essentially and simply, I propose to wait for
the wbox FD by select() and perform non-blocking write(), instead of
checking the "wbox_stat" node.

Here is a patch to implement this solution.

----
(*) One possible scenario:

  1. Thread A calls _base_spe_in_mbox_write() with SPE_MBOX_ALL_BLOCKING
     behavior flag.
  2. Thread A calls and returns from open_if_closed() in
     _base_spe_in_mbox_write().
  3. Another thread B calls _base_spe_in_mbox_write() with
     SPE_MBOX_ANY_NONBLOCKING behavior flag.
  4. Thread B checks available space by reading
     FD_WBOX_STAT.
  5. Thread B decides to write to FD_WBOX.
  6. Thread A fills all available space of FD_WBOX by calling write().
  7. Thread B is blocked at write() system call !
----

Signed-off-by: Kazunori Asayama <asayama at sm.sony.co.jp>

Index: libspe2-public/spebase/create.c
===================================================================
--- libspe2-public.orig/spebase/create.c
+++ libspe2-public/spebase/create.c
@@ -45,7 +45,7 @@ static const struct fd_attr spe_fd_attr[
 	[FD_MBOX_STAT]	= { .name = "mbox_stat", .mode = O_RDONLY, },
 	[FD_IBOX]	= { .name = "ibox",      .mode = O_RDONLY, },
 	[FD_IBOX_STAT]	= { .name = "ibox_stat", .mode = O_RDONLY, },
-	[FD_WBOX]	= { .name = "wbox",      .mode = O_WRONLY, },
+	[FD_WBOX]	= { .name = "wbox",      .mode = O_WRONLY | O_NONBLOCK, },
 	[FD_WBOX_STAT]	= { .name = "wbox_stat", .mode = O_RDONLY, },
 	[FD_SIG1]	= { .name = "signal1",   .mode = O_RDWR, },
 	[FD_SIG2]	= { .name = "signal2",   .mode = O_RDWR, },
Index: libspe2-public/spebase/mbox.c
===================================================================
--- libspe2-public.orig/spebase/mbox.c
+++ libspe2-public/spebase/mbox.c
@@ -21,6 +21,7 @@
 #include <fcntl.h>
 #include <stdio.h>
 #include <unistd.h>
+#include <sys/select.h>
 
 #include "create.h"
 #include "mbox.h"
@@ -31,6 +32,71 @@
  */
 
 
+static int write_immediate(int fd, const void *buf, size_t count)
+{
+	size_t total = 0;
+
+	while (total < count) {
+		/* Non-blocking mode is asssumed. */
+		int rc = write(fd, (const char *)buf + total, count - total);
+		if (rc == -1) {
+			if (errno == EAGAIN) {
+				break;
+			}
+			else {
+				return -1;
+			}
+		}
+		else if (rc == 0) {
+			break;
+		}
+		total += rc;
+	}
+
+	return total;
+}
+
+static int write_any(int fd, const void *buf, size_t count)
+{
+	while (count > 0) {
+		int rc;
+		fd_set fds;
+
+		FD_ZERO(&fds);
+		FD_SET(fd, &fds);
+
+		rc = select(fd + 1, NULL, &fds, NULL, NULL);
+		if (rc == -1) {
+			return -1;
+		}
+		else if (rc > 0) {
+			rc = write_immediate(fd, buf, count);
+			if (rc) return rc;
+		}
+	}
+
+	return 0;
+}
+
+static int write_all(int fd, const void *buf, size_t count)
+{
+	size_t total = 0;
+
+	while (total < count) {
+		int rc = write_any(fd, (const char *)buf + total, count - total);
+		if (rc == -1) {
+			return -1;
+		}
+		else if (rc == 0) {
+			break;
+		}
+		total += rc;
+	}
+
+	return total;
+}
+
+
 int _base_spe_out_mbox_read(spe_context_ptr_t spectx, 
                         unsigned int mbox_data[], 
                         int count)
@@ -54,84 +120,37 @@ int _base_spe_in_mbox_write(spe_context_
                         int count, 
                         int behavior_flag)
 {
-	int rc, ret;
-	int written, items;
-	unsigned int * ptr;
+	int rc;
 
-	ptr = mbox_data;
+	if (mbox_data == NULL || count < 1){
+		errno = EINVAL;
+		return -1;
+	}
 
 	switch (behavior_flag) {
 	case SPE_MBOX_ALL_BLOCKING: // write all, even if blocking
-		rc = 0;
-		written = 0;
-		while (count > 0) {
-			rc = write(open_if_closed(spectx,FD_WBOX, 0), ptr, 4*count);
-			DEBUG_PRINTF("%s write rc: %d\n", __FUNCTION__, rc);
-			if (rc == -1) {// something else went wrong
-				errno = EIO;
-				break; 
-			}
-			items   = rc/4; 
-			count   -= items;
-			written += items;
-			ptr     += items;
-		}
-		rc = written;
+		rc = write_all(open_if_closed(spectx,FD_WBOX, 0), mbox_data, 4*count);
 		break;
 
 	case  SPE_MBOX_ANY_BLOCKING: // write at least one, even if blocking
-		rc = 0;
-		written = 0;
-		_base_spe_context_lock(spectx, FD_WBOX);
-		while (count > 0) {
-			rc = write(open_if_closed(spectx,FD_WBOX, 1), ptr, 4*count);
-			DEBUG_PRINTF("%s write rc: %d\n", __FUNCTION__, rc);
-			if (rc == -1) {// something else went wrong
-				errno = EIO;
-				break; 
-			}
-			items   = rc/4; 
-			count   -= items;
-			written += items;
-			ptr     += items;
-			rc = read(open_if_closed(spectx,FD_WBOX_STAT, 0), &ret, 4);
-			DEBUG_PRINTF("%s stat ret: %d\n", __FUNCTION__, ret);
-			if (ret == 0) break; // no more space
-
-		}
-		_base_spe_context_unlock(spectx, FD_WBOX);
-		rc = written;
+		rc = write_any(open_if_closed(spectx,FD_WBOX, 0), mbox_data, 4*count);
 		break;
 
 	case  SPE_MBOX_ANY_NONBLOCKING: // only write, if non blocking
-		rc = 0;
-		written = 0;
-		_base_spe_context_lock(spectx, FD_WBOX);
-		while (count > 0) {
-			rc = read(open_if_closed(spectx,FD_WBOX_STAT, 0), &ret, 4*count);
-			DEBUG_PRINTF("%s stat ret: %d\n", __FUNCTION__, ret);
-			if (ret == 0) break; // no more space
-			rc = write(open_if_closed(spectx,FD_WBOX, 1), ptr, 4);
-			DEBUG_PRINTF("%s write rc: %d\n", __FUNCTION__, rc);
-			if (rc == -1) {// something else went wrong
-				errno = EIO;
-				break; 
-			}
-			items   = rc/4; 
-			count   -= items;
-			written += items;
-			ptr     += items;
-		}
-		_base_spe_context_unlock(spectx, FD_WBOX);
-		rc = written;
+		rc = write_immediate(open_if_closed(spectx,FD_WBOX, 0), mbox_data, 4*count);
 		break;
 
 	default:
 		errno = EINVAL;
-		rc = -1;
+		return -1;
 	}
 
-	return rc;
+	if (rc == -1) {
+		errno = EIO;
+		return -1;
+	}
+
+	return rc / 4;
 }
 
 int _base_spe_in_mbox_status(spe_context_ptr_t spectx)