source/pipe.cpp

   1 #ifndef WIN32
   2 #include <fcntl.h>
   3 #include <errno.h>
   4 #endif
   5 #include <msp/strings/formatter.h>
   6 #include "pipe.h"
   7
   8 using namespace std;
   9
  10 namespace Msp {
  11 namespace IO {
  12
  13 Pipe::Pipe()
  14 {
  15 #ifdef WIN32
  16         string name = format("\\\\.\\pipe\\%u.%p", GetCurrentProcessId(), this);
  17         handle[0] = CreateNamedPipe(name.c_str(), PIPE_ACCESS_INBOUND|FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE, 1, 1024, 1024, 0, 0);
  18         if(handle[0]==INVALID_HANDLE_VALUE)
  19                 throw SystemError("Unable to create pipe", GetLastError());
  20
  21         handle[1] = CreateFile(name.c_str(), GENERIC_WRITE, 0, 0, OPEN_EXISTING, 0, 0);
  22         if(handle[1]==INVALID_HANDLE_VALUE)
  23         {
  24                 unsigned err = GetLastError();
  25                 CloseHandle(handle[0]);
  26                 throw SystemError("Unable to create pipe", err);
  27         }
  28
  29         overlapped = 0;
  30         event = CreateEvent(0, true, false, 0);
  31         buf_size = 1024;
  32         buffer = new char[buf_size];
  33         buf_avail = 0;
  34         buf_next = buffer;
  35 #else
  36         if(pipe(handle)==-1)
  37                 throw SystemError("Unable to create pipe", errno);
  38 #endif
  39
  40         set_events(P_INPUT);
  41 }
  42
  43 Pipe::~Pipe()
  44 {
  45         close();
  46 }
  47
  48 void Pipe::close()
  49 {
  50         set_events(P_NONE);
  51
  52         signal_flush_required.emit();
  53 #ifdef WIN32
  54         CloseHandle(handle[0]);
  55         CloseHandle(handle[1]);
  56 #else
  57         ::close(handle[0]);
  58         ::close(handle[1]);
  59         signal_closed.emit();
  60 #endif
  61 }
  62
  63 void Pipe::set_block(bool b)
  64 {
  65         mode = (mode&~M_NONBLOCK);
  66         if(b)
  67                 mode = (mode|M_NONBLOCK);
  68
  69 #ifndef WIN32
  70         int flags = fcntl(handle[0], F_GETFD);
  71         fcntl(handle[0], F_SETFL, (flags&O_NONBLOCK)|(b?0:O_NONBLOCK));
  72         flags = fcntl(handle[1], F_GETFD);
  73         fcntl(handle[1], F_SETFL, (flags&O_NONBLOCK)|(b?0:O_NONBLOCK));
  74 #endif
  75 }
  76
  77 unsigned Pipe::do_write(const char *buf, unsigned size)
  78 {
  79         if(size==0)
  80                 return 0;
  81
  82 #ifdef WIN32
  83         DWORD ret;
  84         if(!WriteFile(handle[1], buf, size, &ret, 0))
  85                 throw SystemError("Writing to pipe failed", GetLastError());
  86 #else
  87         int ret = ::write(handle[1], buf, size);
  88         if(ret==-1)
  89         {
  90                 if(errno==EAGAIN)
  91                         return 0;
  92                 else
  93                         throw SystemError("Writing to pipe failed", errno);
  94         }
  95 #endif
  96
  97         return ret;
  98 }
  99
 100 unsigned Pipe::do_read(char *buf, unsigned size)
 101 {
 102         if(size==0)
 103                 return 0;
 104
 105 #ifdef WIN32
 106         // Initiate overlapped read if needed
 107         get_event_handle();
 108
 109         if(overlapped)
 110         {
 111                 DWORD ret;
 112                 if(!GetOverlappedResult(handle[0], overlapped, &ret, !buf_avail))
 113                         throw SystemError("Reading from pipe failed", GetLastError());
 114                 else
 115                 {
 116                         buf_avail += ret;
 117                         delete overlapped;
 118                         overlapped = 0;
 119                 }
 120         }
 121
 122         unsigned ret = min(buf_avail, size);
 123         memcpy(buf, buf_next, ret);
 124         buf_next += ret;
 125         buf_avail -= ret;
 126
 127         // Initiate another overlapped read in case someone is polling us
 128         get_event_handle();
 129 #else
 130         int ret = ::read(handle[0], buf, size);
 131         if(ret==-1)
 132         {
 133                 if(errno==EAGAIN)
 134                         return 0;
 135                 else
 136                         throw SystemError("Reading from pipe failed", errno);
 137         }
 138 #endif
 139
 140         if(ret==0)
 141         {
 142                 eof_flag = true;
 143                 signal_end_of_file.emit();
 144         }
 145
 146         return ret;
 147 }
 148
 149 Handle Pipe::get_event_handle()
 150 {
 151 #ifdef WIN32
 152         if(!overlapped && !buf_avail)
 153         {
 154                 overlapped = new OVERLAPPED;
 155                 memset(overlapped, 0, sizeof(OVERLAPPED));
 156                 overlapped->hEvent = event;
 157
 158                 DWORD ret;
 159                 buf_next = buffer;
 160                 if(!ReadFile(handle[0], buffer, buf_size, &ret, overlapped))
 161                 {
 162                         unsigned err = GetLastError();
 163                         if(err!=ERROR_IO_PENDING)
 164                                 throw SystemError("Failed to start an overlapped read", err);
 165                 }
 166                 else
 167                 {
 168                         buf_avail = ret;
 169                         delete overlapped;
 170                         overlapped = 0;
 171                         SetEvent(event);
 172                 }
 173         }
 174
 175         return event;
 176 #else
 177         return handle[0];
 178 #endif
 179 }
 180
 181 } // namespace IO
 182 } // namespace Msp