source/io/pipe.cpp

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