pdns/portsmplexer.cc

   1 #if defined(__sun__) && defined(__svr4__)
   2 #ifdef HAVE_CONFIG_H
   3 #include "config.h"
   4 #endif
   5 #include <port.h>
   6 #include <sys/port_impl.h>
   7 #endif
   8 #include <unistd.h>
   9 #include "mplexer.hh"
  10 #include "sstuff.hh"
  11 #include <iostream>
  12
  13 #include "misc.hh"
  14
  15 #include "namespaces.hh"
  16
  17 class PortsFDMultiplexer : public FDMultiplexer
  18 {
  19 public:
  20   PortsFDMultiplexer(unsigned int maxEventsHint);
  21   ~PortsFDMultiplexer()
  22   {
  23     close(d_portfd);
  24   }
  25
  26   int run(struct timeval* tv, int timeout = 500) override;
  27   void getAvailableFDs(std::vector<int>& fds, int timeout) override;
  28
  29   void addFD(int fd, FDMultiplexer::EventKind kind) override;
  30   void removeFD(int fd, FDMultiplexer::EventKind kind) override;
  31
  32   string getName() const override
  33   {
  34     return "solaris completion ports";
  35   }
  36
  37 private:
  38   int d_portfd;
  39   std::vector<port_event_t> d_pevents;
  40 };
  41
  42 static FDMultiplexer* makePorts(unsigned int maxEventsHint)
  43 {
  44   return new PortsFDMultiplexer(maxEventsHint);
  45 }
  46
  47 static struct PortsRegisterOurselves
  48 {
  49   PortsRegisterOurselves()
  50   {
  51     FDMultiplexer::getMultiplexerMap().emplace(0, &makePorts); // priority 0!
  52   }
  53 } doItPorts;
  54
  55 PortsFDMultiplexer::PortsFDMultiplexer(unsigned int maxEventsHint) :
  56   d_pevents(maxEventsHint)
  57 {
  58   d_portfd = port_create(); // not hard max
  59   if (d_portfd < 0) {
  60     throw FDMultiplexerException("Setting up port: " + stringerror());
  61   }
  62 }
  63
  64 static int convertEventKind(FDMultiplexer::EventKind kind)
  65 {
  66   switch (kind) {
  67   case FDMultiplexer::EventKind::Read:
  68     return POLLIN;
  69   case FDMultiplexer::EventKind::Write:
  70     return POLLOUT;
  71   case FDMultiplexer::EventKind::Both:
  72     return POLLIN | POLLOUT;
  73   }
  74   throw std::runtime_error("Unhandled event kind in the ports multiplexer");
  75 }
  76
  77 void PortsFDMultiplexer::addFD(int fd, FDMultiplexer::EventKind kind)
  78 {
  79   if (port_associate(d_portfd, PORT_SOURCE_FD, fd, convertEventKind(kind), 0) < 0) {
  80     throw FDMultiplexerException("Adding fd to port set: " + stringerror());
  81   }
  82 }
  83
  84 void PortsFDMultiplexer::removeFD(int fd, FDMultiplexer::EventKind)
  85 {
  86   if (port_dissociate(d_portfd, PORT_SOURCE_FD, fd) < 0 && errno != ENOENT) { // it appears under some circumstances, ENOENT will be returned, without this being an error. Apache has this same "fix"
  87     throw FDMultiplexerException("Removing fd from port set: " + stringerror());
  88   }
  89 }
  90
  91 void PortsFDMultiplexer::getAvailableFDs(std::vector<int>& fds, int timeout)
  92 {
  93   struct timespec timeoutspec;
  94   timeoutspec.tv_sec = timeout / 1000;
  95   timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
  96   unsigned int numevents = 1;
  97   int ret = port_getn(d_portfd, d_pevents.data(), min(PORT_MAX_LIST, static_cast<int>(d_pevents.size())), &numevents, timeout != -1 ? &timeoutspec : nullptr);
  98
  99   /* port_getn has an unusual API - (ret == -1, errno == ETIME) can
 100      mean partial success; you must check (*numevents) in this case
 101      and process anything in there, otherwise you'll never see any
 102      events from that object again. We don't care about pure timeouts
 103      (ret == -1, errno == ETIME, *numevents == 0) so we don't bother
 104      with that case. */
 105   if (ret == -1 && errno != ETIME) {
 106     if (errno != EINTR) {
 107       throw FDMultiplexerException("completion port_getn returned error: " + stringerror());
 108     }
 109
 110     // EINTR is not really an error
 111     return;
 112   }
 113
 114   if (numevents == 0) {
 115     // nothing
 116     return;
 117   }
 118
 119   fds.reserve(numevents);
 120
 121   for (unsigned int n = 0; n < numevents; ++n) {
 122     const auto fd = d_pevents[n].portev_object;
 123
 124     /* we need to re-associate the FD */
 125     if ((d_pevents[n].portev_events & POLLIN || d_pevents[n].portev_events & POLLERR || d_pevents[n].portev_events & POLLHUP)) {
 126       if (d_readCallbacks.count(fd)) {
 127         if (port_associate(d_portfd, PORT_SOURCE_FD, fd, d_writeCallbacks.count(fd) > 0 ? POLLIN | POLLOUT : POLLIN, 0) < 0) {
 128           throw FDMultiplexerException("Unable to add fd back to ports (read): " + stringerror());
 129         }
 130       }
 131     }
 132     else if ((d_pevents[n].portev_events & POLLOUT || d_pevents[n].portev_events & POLLERR)) {
 133       if (d_writeCallbacks.count(fd)) {
 134         if (port_associate(d_portfd, PORT_SOURCE_FD, fd, d_readCallbacks.count(fd) > 0 ? POLLIN | POLLOUT : POLLOUT, 0) < 0) {
 135           throw FDMultiplexerException("Unable to add fd back to ports (write): " + stringerror());
 136         }
 137       }
 138     }
 139     else {
 140       /* not registered, this is unexpected */
 141       continue;
 142     }
 143
 144     fds.push_back(fd);
 145   }
 146 }
 147
 148 int PortsFDMultiplexer::run(struct timeval* now, int timeout)
 149 {
 150   InRun guard(d_inrun);
 151
 152   struct timespec timeoutspec;
 153   timeoutspec.tv_sec = timeout / 1000;
 154   timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
 155   unsigned int numevents = 1;
 156   int ret = port_getn(d_portfd, d_pevents.data(), min(PORT_MAX_LIST, static_cast<int>(d_pevents.size())), &numevents, timeout != -1 ? &timeoutspec : nullptr);
 157
 158   /* port_getn has an unusual API - (ret == -1, errno == ETIME) can
 159      mean partial success; you must check (*numevents) in this case
 160      and process anything in there, otherwise you'll never see any
 161      events from that object again. We don't care about pure timeouts
 162      (ret == -1, errno == ETIME, *numevents == 0) so we don't bother
 163      with that case. */
 164   if (ret == -1 && errno != ETIME) {
 165     if (errno != EINTR) {
 166       throw FDMultiplexerException("completion port_getn returned error: " + stringerror());
 167     }
 168     // EINTR is not really an error
 169     gettimeofday(now, nullptr);
 170     return 0;
 171   }
 172   gettimeofday(now, nullptr);
 173   if (!numevents) {
 174     // nothing
 175     return 0;
 176   }
 177
 178   int count = 0;
 179   for (unsigned int n = 0; n < numevents; ++n) {
 180     if (d_pevents[n].portev_events & POLLIN || d_pevents[n].portev_events & POLLERR || d_pevents[n].portev_events & POLLHUP) {
 181       const auto& iter = d_readCallbacks.find(d_pevents[n].portev_object);
 182       if (iter != d_readCallbacks.end()) {
 183         iter->d_callback(iter->d_fd, iter->d_parameter);
 184         count++;
 185         if (d_readCallbacks.count(d_pevents[n].portev_object) && port_associate(d_portfd, PORT_SOURCE_FD, d_pevents[n].portev_object, d_writeCallbacks.count(d_pevents[n].portev_object) ? POLLIN | POLLOUT : POLLIN, 0) < 0) {
 186           throw FDMultiplexerException("Unable to add fd back to ports (read): " + stringerror());
 187         }
 188       }
 189     }
 190     if (d_pevents[n].portev_events & POLLOUT || d_pevents[n].portev_events & POLLERR) {
 191       const auto& iter = d_writeCallbacks.find(d_pevents[n].portev_object);
 192       if (iter != d_writeCallbacks.end()) {
 193         iter->d_callback(iter->d_fd, iter->d_parameter);
 194         count++;
 195         if (d_writeCallbacks.count(d_pevents[n].portev_object) && port_associate(d_portfd, PORT_SOURCE_FD, d_pevents[n].portev_object, d_readCallbacks.count(d_pevents[n].portev_object) ? POLLIN | POLLOUT : POLLOUT, 0) < 0) {
 196           throw FDMultiplexerException("Unable to add fd back to ports (write): " + stringerror());
 197         }
 198       }
 199     }
 200   }
 201
 202   return count;
 203 }
 204
 205 #if 0
 206 void acceptData(int fd, boost::any& parameter)
 207 {
 208   cout<<"Have data on fd "<<fd<<endl;
 209   Socket* sock=boost::any_cast<Socket*>(parameter);
 210   string packet;
 211   IPEndpoint rem;
 212   sock->recvFrom(packet, rem);
 213   cout<<"Received "<<packet.size()<<" bytes!\n";
 214 }
 215
 216
 217 int main()
 218 {
 219   Socket s(AF_INET, SOCK_DGRAM);
 220
 221   IPEndpoint loc("0.0.0.0", 2000);
 222   s.bind(loc);
 223
 224   PortsFDMultiplexer sfm;
 225
 226   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 227
 228   for(int n=0; n < 100 ; ++n) {
 229     sfm.run();
 230   }
 231   sfm.removeReadFD(s.getHandle());
 232   sfm.removeReadFD(s.getHandle());
 233 }
 234 #endif