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