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         sys_close(handle[0]);
  80         sys_close(handle[1]);
  81         signal_closed.emit();
  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         return sys_write(handle[1], buf, size);
 104 }
 105
 106 unsigned Pipe::do_read(char *buf, unsigned size)
 107 {
 108         if(size==0)
 109                 return 0;
 110
 111 #ifdef WIN32
 112         // Initiate overlapped read if needed
 113         get_event_handle();
 114
 115         if(priv->overlapped)
 116         {
 117                 DWORD ret;
 118                 if(!GetOverlappedResult(*handle[0], priv->overlapped, &ret, !priv->buf_avail))
 119                         throw system_error("GetOverlappedResult");
 120                 else
 121                 {
 122                         priv->buf_avail += ret;
 123                         delete priv->overlapped;
 124                         priv->overlapped = 0;
 125                 }
 126         }
 127
 128         unsigned ret = min(priv->buf_avail, size);
 129         memcpy(buf, priv->buf_next, ret);
 130         priv->buf_next += ret;
 131         priv->buf_avail -= ret;
 132
 133         // Initiate another overlapped read in case someone is polling us
 134         get_event_handle();
 135 #else
 136         unsigned ret = sys_read(handle[0], buf, size);
 137 #endif
 138
 139         if(ret==0)
 140         {
 141                 eof_flag = true;
 142                 signal_end_of_file.emit();
 143         }
 144
 145         return ret;
 146 }
 147
 148 const Handle &Pipe::get_event_handle()
 149 {
 150 #ifdef WIN32
 151         if(!priv->overlapped && !priv->buf_avail)
 152         {
 153                 priv->overlapped = new OVERLAPPED;
 154                 memset(priv->overlapped, 0, sizeof(OVERLAPPED));
 155                 priv->overlapped->hEvent = *priv->event;
 156
 157                 DWORD ret;
 158                 priv->buf_next = priv->buffer;
 159                 if(!ReadFile(*handle[0], priv->buffer, priv->buf_size, &ret, priv->overlapped))
 160                 {
 161                         unsigned err = GetLastError();
 162                         if(err!=ERROR_IO_PENDING)
 163                                 throw system_error("ReadFile");
 164                 }
 165                 else
 166                 {
 167                         priv->buf_avail = ret;
 168                         delete priv->overlapped;
 169                         priv->overlapped = 0;
 170                         SetEvent(*priv->event);
 171                 }
 172         }
 173
 174         return priv->event;
 175 #else
 176         return handle[0];
 177 #endif
 178 }
 179
 180 } // namespace IO
 181 } // namespace Msp