Bug 1059813: Wrap socket I/O operations in |UnixSocketRawData|, r=qdot

This patch moves the I/O operations for sending and receiving data
in |SocketIOBase| into |UnixSocketRawData|. This change allows to
add a clean interface to |UnixSocketRawData| and later replace the
class by other implementations.

ipc/unixsocket/SocketBase.cpp

@@ -7,8 +7,9 @@
  */
 
 #include "SocketBase.h"
+#include <errno.h>
 #include <string.h>
-#include "nsThreadUtils.h"
+#include <unistd.h>
 
 namespace mozilla {
 namespace ipc {
@@ -18,22 +19,80 @@ namespace ipc {
 //
 
 UnixSocketRawData::UnixSocketRawData(size_t aSize)
-: mSize(aSize)
+: mSize(0)
 , mCurrentWriteOffset(0)
+, mAvailableSpace(aSize)
 {
-  mData = new uint8_t[mSize];
+  mData = new uint8_t[mAvailableSpace];
 }
 
 UnixSocketRawData::UnixSocketRawData(const void* aData, size_t aSize)
 : mSize(aSize)
 , mCurrentWriteOffset(0)
+, mAvailableSpace(aSize)
 {
   MOZ_ASSERT(aData || !mSize);
 
-  mData = new uint8_t[mSize];
+  mData = new uint8_t[mAvailableSpace];
   memcpy(mData, aData, mSize);
 }
 
+ssize_t
+UnixSocketRawData::Receive(int aFd)
+{
+  if (!GetTrailingSpace()) {
+    if (!GetLeadingSpace()) {
+      return -1; /* buffer is full */
+    }
+    /* free up space at the end of data buffer */
+    if (GetSize() <= GetLeadingSpace()) {
+      memcpy(mData, GetData(), GetSize());
+    } else {
+      memmove(mData, GetData(), GetSize());
+    }
+    mCurrentWriteOffset = 0;
+  }
+
+  ssize_t res =
+    TEMP_FAILURE_RETRY(read(aFd, GetTrailingBytes(), GetTrailingSpace()));
+
+  if (res < 0) {
+    /* I/O error */
+    return -1;
+  } else if (!res) {
+    /* EOF or peer shutdown sending */
+    return 0;
+  }
+
+  mSize += res;
+
+  return res;
+}
+
+ssize_t
+UnixSocketRawData::Send(int aFd)
+{
+  if (!GetSize()) {
+    return 0;
+  }
+
+  ssize_t res = TEMP_FAILURE_RETRY(write(aFd, GetData(), GetSize()));
+
+  if (res < 0) {
+    if (errno == EAGAIN || errno == EWOULDBLOCK) {
+      return 0; /* socket is blocked; try again later */
+    }
+    return -1;
+  } else if (!res) {
+    /* nothing written */
+    return 0;
+  }
+
+  Consume(res);
+
+  return res;
+}
+
 //
 // SocketConsumerBase
 //

ipc/unixsocket/SocketBase.h

@@ -9,8 +9,6 @@
 #ifndef mozilla_ipc_SocketBase_h
 #define mozilla_ipc_SocketBase_h
 
-#include <errno.h>
-#include <unistd.h>
 #include "base/message_loop.h"
 #include "nsAutoPtr.h"
 #include "nsTArray.h"
@@ -47,6 +45,61 @@ public:
    * beforehand (for example, when being assigned strings)
    */
   UnixSocketRawData(const void* aData, size_t aSize);
+
+  /**
+   * Receives data from aFd at the end of the buffer. The returned value
+   * is the number of newly received bytes, or 0 if the peer shut down
+   * its connection, or a negative value on errors.
+   */
+  ssize_t Receive(int aFd);
+
+  /**
+   * Sends data to aFd from the beginning of the buffer. The returned value
+   * is the number of bytes written, or a negative value on error.
+   */
+  ssize_t Send(int aFd);
+
+  const uint8_t* GetData() const
+  {
+    return mData + mCurrentWriteOffset;
+  }
+
+  size_t GetSize() const
+  {
+    return mSize;
+  }
+
+  void Consume(size_t aSize)
+  {
+    MOZ_ASSERT(aSize <= mSize);
+
+    mSize -= aSize;
+    mCurrentWriteOffset += aSize;
+  }
+
+protected:
+  size_t GetLeadingSpace() const
+  {
+    return mCurrentWriteOffset;
+  }
+
+  size_t GetTrailingSpace() const
+  {
+    return mAvailableSpace - (mCurrentWriteOffset + mSize);
+  }
+
+  size_t GetAvailableSpace() const
+  {
+    return mAvailableSpace;
+  }
+
+  void* GetTrailingBytes()
+  {
+    return mData + mCurrentWriteOffset + mSize;
+  }
+
+private:
+  size_t mAvailableSpace;
 };
 
 enum SocketConnectionStatus {
@@ -331,9 +384,7 @@ public:
     nsAutoPtr<UnixSocketRawData> incoming(
       new UnixSocketRawData(mMaxReadSize));
 
-    ssize_t res =
-      TEMP_FAILURE_RETRY(read(aFd, incoming->mData, incoming->mSize));
-
+    ssize_t res = incoming->Receive(aFd);
     if (res < 0) {
       /* an I/O error occured */
       nsRefPtr<nsRunnable> r = new SocketIORequestClosingRunnable<T>(aIO);
@@ -346,8 +397,6 @@ public:
       return 0;
     }
 
-    incoming->mSize = res;
-
 #ifdef MOZ_TASK_TRACER
     // Make unix socket creation events to be the source events of TaskTracer,
     // and originate the rest correlation tasks from here.
@@ -367,38 +416,24 @@ public:
     MOZ_ASSERT(aFd >= 0);
     MOZ_ASSERT(aIO);
 
-    do {
-      if (!HasPendingData()) {
-        return NS_OK;
-      }
-
+    while (HasPendingData()) {
       UnixSocketRawData* outgoing = mOutgoingQ.ElementAt(0);
-      MOZ_ASSERT(outgoing->mSize);
-
-      const uint8_t* data = outgoing->mData + outgoing->mCurrentWriteOffset;
-      size_t size = outgoing->mSize - outgoing->mCurrentWriteOffset;
-
-      ssize_t res = TEMP_FAILURE_RETRY(write(aFd, data, size));
 
+      ssize_t res = outgoing->Send(aFd);
       if (res < 0) {
-        if (errno == EAGAIN || errno == EWOULDBLOCK) {
-          return NS_OK; /* no more data available */
-        }
-        /* an error occored */
+        /* an I/O error occured */
         nsRefPtr<nsRunnable> r = new SocketIORequestClosingRunnable<T>(aIO);
         NS_DispatchToMainThread(r);
         return NS_ERROR_FAILURE;
-      } else if (!res) {
-        return NS_OK; /* nothing written */
+      } else if (!res && outgoing->GetSize()) {
+        /* I/O is currently blocked; try again later */
+        return NS_OK;
       }
-
-      outgoing->mCurrentWriteOffset += res;
-
-      if (outgoing->mCurrentWriteOffset == outgoing->mSize) {
+      if (!outgoing->GetSize()) {
         mOutgoingQ.RemoveElementAt(0);
         delete outgoing;
       }
-    } while (true);
+    }
 
     return NS_OK;
   }