Native anonymous pipe function

ccutrer · Jun 21, 2011 · ab68bf2 · ab68bf2
1 parent 7aaf0d5
commit ab68bf2
Show file tree

Hide file tree

Showing 4 changed files with 71 additions and 8 deletions.
diff --git a/mordor/streams/handle.cpp b/mordor/streams/handle.cpp
@@ -33,13 +33,12 @@ HandleStream::init(HANDLE hFile, IOManager *ioManager, Scheduler *scheduler,
     m_cancelRead = m_cancelWrite = false;
     m_ioManager = ioManager;
     m_scheduler = scheduler;
-    DWORD type = GetFileType(hFile);
-    if (type == FILE_TYPE_CHAR) {
+    m_type = GetFileType(hFile);
+    if (m_type == FILE_TYPE_CHAR) {
         m_ioManager = NULL;
         CONSOLE_SCREEN_BUFFER_INFO info;
-        if (!GetConsoleScreenBufferInfo(hFile, &info))
-            MORDOR_THROW_EXCEPTION_FROM_LAST_ERROR_API("GetConsoleScreenBufferInfo");
-        m_maxOpSize = info.dwSize.X * info.dwSize.Y / 2;
+        if (GetConsoleScreenBufferInfo(hFile, &info))
+            m_maxOpSize = info.dwSize.X * info.dwSize.Y / 2;
     }
     if (m_ioManager) {
         try {
@@ -123,6 +122,8 @@ HandleStream::read(void *buffer, size_t length)
     Log::Level level = Log::DEBUG;
     if (!ret) {
         if (lastError() == ERROR_HANDLE_EOF) {
+        } else if (m_type == FILE_TYPE_PIPE && lastError() == ERROR_BROKEN_PIPE) {
+            ret = TRUE;
         } else if (m_ioManager) {
             if (lastError() == ERROR_IO_PENDING)
                 level = Log::TRACE;

diff --git a/mordor/streams/handle.h b/mordor/streams/handle.h
@@ -54,6 +54,7 @@ class HandleStream : public Stream
     HANDLE m_hFile;
     bool m_own, m_cancelRead, m_cancelWrite;
     size_t m_maxOpSize;
+    DWORD m_type;
 };
 
 typedef HandleStream NativeStream;

diff --git a/mordor/streams/pipe.cpp b/mordor/streams/pipe.cpp
@@ -5,11 +5,16 @@
 #include <boost/thread/mutex.hpp>
 
 #include "buffer.h"
+#include "file.h"
 #include "mordor/assert.h"
 #include "mordor/fiber.h"
 #include "mordor/scheduler.h"
 #include "stream.h"
 
+#ifdef OSX
+#include <crt_externs.h>
+#endif
+
 namespace Mordor {
 
 static Logger::ptr g_log = Log::lookup("mordor:streams:pipe");
@@ -329,4 +334,47 @@ PipeStream::onRemoteClose(const boost::signals2::slot<void ()> &slot)
     return m_onRemoteClose.connect(slot);
 }
 
+
+std::pair<NativeStream::ptr, NativeStream::ptr>
+anonymousPipe(IOManager *ioManager)
+{
+    std::pair<NativeStream::ptr, NativeStream::ptr> result;
+#ifdef WINDOWS
+    if (ioManager) {
+        // TODO: Implement overlapped I/O for this pipe with either a
+        // not-quite-anonymous pipe, or a socket pair
+        MORDOR_NOTREACHED();
+    } else {
+        HANDLE read = NULL, write = NULL;
+        if (!CreatePipe(&read, &write, NULL, 0))
+            MORDOR_THROW_EXCEPTION_FROM_LAST_ERROR_API("CreatePipe");
+        try {
+            result.first.reset(new HandleStream(read));
+            result.second.reset(new HandleStream(write));
+        } catch (...) {
+            if (!result.first)
+                CloseHandle(read);
+            if (!result.second)
+                CloseHandle(write);
+            throw;
+        }
+    }
+#else
+    int fds[2];
+    if (pipe(fds))
+        MORDOR_THROW_EXCEPTION_FROM_LAST_ERROR_API("pipe");
+    try {
+            result.first.reset(new FDStream(fds[0], ioManager));
+            result.second.reset(new FDStream(fds[1], ioManager));
+        } catch (...) {
+            if (!result.first)
+                close(fds[0]);
+            if (!result.second)
+                close(fds[1]);
+            throw;
+        }
+#endif
+    return result;
+}
+
 }
diff --git a/mordor/streams/pipe.h b/mordor/streams/pipe.h
@@ -6,12 +6,25 @@
 
 #include <boost/shared_ptr.hpp>
 
+#ifdef WINDOWS
+#include "handle.h"
+#else
+#include "fd.h"
+#endif
+
 namespace Mordor {
 
-class Stream;
+class IOManager;
+
+/// Create a user-space only, full-duplex anonymous pipe
+std::pair<Stream::ptr, Stream::ptr> pipeStream(size_t bufferSize = ~0);
 
-std::pair<boost::shared_ptr<Stream>, boost::shared_ptr<Stream> >
-    pipeStream(size_t bufferSize = ~0);
+/// Create a kernel-level, half-duplex anonymous pipe
+///
+/// The Streams created by this function will have a file handle, and are
+/// suitable for usage with native OS APIs
+std::pair<NativeStream::ptr, NativeStream::ptr>
+    anonymousPipe(IOManager *ioManager = NULL);
 
 }