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