src/dasynq/dasynq-epoll.h

   1 #include <system_error>
   2 #include <mutex>
   3 #include <type_traits>
   4 #include <unordered_map>
   5 #include <vector>
   6
   7 #include <sys/epoll.h>
   8 #include <sys/signalfd.h>
   9 #include <sys/types.h>
  10 #include <sys/wait.h>
  11
  12 #include <unistd.h>
  13 #include <signal.h>
  14
  15 namespace dasynq {
  16
  17 template <class Base> class epoll_loop;
  18
  19 class epoll_traits
  20 {
  21     template <class Base> friend class epoll_loop;
  22
  23     public:
  24
  25     class sigdata_t
  26     {
  27         template <class Base> friend class epoll_loop;
  28
  29         struct signalfd_siginfo info;
  30
  31         public:
  32         // mandatory:
  33         int get_signo() { return info.ssi_signo; }
  34         int get_sicode() { return info.ssi_code; }
  35         pid_t get_sipid() { return info.ssi_pid; }
  36         uid_t get_siuid() { return info.ssi_uid; }
  37         void * get_siaddr() { return reinterpret_cast<void *>(info.ssi_addr); }
  38         int get_sistatus() { return info.ssi_status; }
  39         int get_sival_int() { return info.ssi_int; }
  40         void * get_sival_ptr() { return reinterpret_cast<void *>(info.ssi_ptr); }
  41
  42         // XSI
  43         int get_sierrno() { return info.ssi_errno; }
  44
  45         // XSR (streams) OB (obselete)
  46         int get_siband() { return info.ssi_band; }
  47
  48         // Linux:
  49         int32_t get_sifd() { return info.ssi_fd; }
  50         uint32_t get_sittimerid() { return info.ssi_tid; }
  51         uint32_t get_sioverrun() { return info.ssi_overrun; }
  52         uint32_t get_sitrapno() { return info.ssi_trapno; }
  53         uint32_t get_siutime() { return info.ssi_utime; }
  54         uint32_t get_sistime() { return info.ssi_stime; }
  55         // Field exposed by Linux kernel but not Glibc:
  56         // uint16_t get_siaddr_lsb() { return info.ssi_addr_lsb; }
  57
  58         void set_signo(int signo) { info.ssi_signo = signo; }
  59     };
  60
  61     class fd_r;
  62
  63     // File descriptor optional storage. If the mechanism can return the file descriptor, this
  64     // class will be empty, otherwise it can hold a file descriptor.
  65     class fd_s {
  66         friend class fd_r;
  67
  68         // Epoll doesn't return the file descriptor (it can, but it can't return both file
  69         // descriptor and user data).
  70         int fd;
  71
  72         public:
  73         fd_s(int fd_p) noexcept : fd(fd_p) { }
  74     };
  75
  76     // File descriptor reference (passed to event callback). If the mechanism can return the
  77     // file descriptor, this class holds the file descriptor. Otherwise, the file descriptor
  78     // must be stored in an fd_s instance.
  79     class fd_r {
  80         public:
  81         int get_fd(fd_s ss)
  82         {
  83             return ss.fd;
  84         }
  85     };
  86
  87     constexpr static bool has_bidi_fd_watch = true;
  88     constexpr static bool has_separate_rw_fd_watches = false;
  89     constexpr static bool interrupt_after_fd_add = false;
  90     constexpr static bool interrupt_after_signal_add = false;
  91     constexpr static bool supports_non_oneshot_fd = true;
  92 };
  93
  94
  95 template <class Base> class epoll_loop : public Base
  96 {
  97     int epfd; // epoll fd
  98     int sigfd; // signalfd fd; -1 if not initialised
  99     sigset_t sigmask;
 100
 101     std::unordered_map<int, void *> sigdataMap;
 102
 103     // Base contains:
 104     //   lock - a lock that can be used to protect internal structure.
 105     //          receive*() methods will be called with lock held.
 106     //   receive_signal(sigdata_t &, user *) noexcept
 107     //   receive_fd_event(fd_r, user *, int flags) noexcept
 108
 109     using sigdata_t = epoll_traits::sigdata_t;
 110     using fd_r = typename epoll_traits::fd_r;
 111
 112     void process_events(epoll_event *events, int r)
 113     {
 114         std::lock_guard<decltype(Base::lock)> guard(Base::lock);
 115
 116         for (int i = 0; i < r; i++) {
 117             void * ptr = events[i].data.ptr;
 118
 119             if (ptr == &sigfd) {
 120                 // Signal
 121                 sigdata_t siginfo;
 122                 while (true) {
 123                     int r = read(sigfd, &siginfo.info, sizeof(siginfo.info));
 124                     if (r == -1) break;
 125                     auto iter = sigdataMap.find(siginfo.get_signo());
 126                     if (iter != sigdataMap.end()) {
 127                         void *userdata = (*iter).second;
 128                         if (Base::receive_signal(*this, siginfo, userdata)) {
 129                             sigdelset(&sigmask, siginfo.get_signo());
 130                         }
 131                     }
 132                 }
 133                 signalfd(sigfd, &sigmask, SFD_NONBLOCK | SFD_CLOEXEC);
 134             }
 135             else {
 136                 int flags = 0;
 137                 (events[i].events & EPOLLIN) && (flags |= IN_EVENTS);
 138                 (events[i].events & EPOLLHUP) && (flags |= IN_EVENTS);
 139                 (events[i].events & EPOLLOUT) && (flags |= OUT_EVENTS);
 140                 (events[i].events & EPOLLERR) && (flags |= IN_EVENTS | OUT_EVENTS | ERR_EVENTS);
 141                 auto r = Base::receive_fd_event(*this, fd_r(), ptr, flags);
 142                 if (std::get<0>(r) != 0) {
 143                     enable_fd_watch_nolock(fd_r().get_fd(std::get<1>(r)), ptr, std::get<0>(r));
 144                 }
 145             }
 146         }
 147     }
 148
 149     public:
 150
 151     /**
 152      * epoll_loop constructor.
 153      *
 154      * Throws std::system_error or std::bad_alloc if the event loop cannot be initialised.
 155      */
 156     epoll_loop() : sigfd(-1)
 157     {
 158         epfd = epoll_create1(EPOLL_CLOEXEC);
 159         if (epfd == -1) {
 160             throw std::system_error(errno, std::system_category());
 161         }
 162         sigemptyset(&sigmask);
 163         Base::init(this);
 164     }
 165
 166     ~epoll_loop()
 167     {
 168         close(epfd);
 169         if (sigfd != -1) {
 170             close(sigfd);
 171         }
 172     }
 173
 174     //        fd:  file descriptor to watch
 175     //  userdata:  data to associate with descriptor
 176     //     flags:  IN_EVENTS | OUT_EVENTS | ONE_SHOT
 177     // soft_fail:  true if unsupported file descriptors should fail by returning false instead
 178     //             of throwing an exception
 179     // returns: true on success; false if file descriptor type isn't supported and soft_fail == true
 180     // throws:  std::system_error or std::bad_alloc on failure
 181     bool add_fd_watch(int fd, void *userdata, int flags, bool enabled = true, bool soft_fail = false)
 182     {
 183         struct epoll_event epevent;
 184         // epevent.data.fd = fd;
 185         epevent.data.ptr = userdata;
 186         epevent.events = 0;
 187
 188         if (flags & ONE_SHOT) {
 189             epevent.events = EPOLLONESHOT;
 190         }
 191         if ((flags & IN_EVENTS) && enabled) {
 192             epevent.events |= EPOLLIN;
 193         }
 194         if ((flags & OUT_EVENTS) && enabled) {
 195             epevent.events |= EPOLLOUT;
 196         }
 197
 198         if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &epevent) == -1) {
 199             if (soft_fail && errno == EPERM) {
 200                 return false;
 201             }
 202             throw new std::system_error(errno, std::system_category());
 203         }
 204         return true;
 205     }
 206
 207     bool add_bidi_fd_watch(int fd, void *userdata, int flags, bool emulate)
 208     {
 209         // No implementation.
 210         throw std::system_error(std::make_error_code(std::errc::not_supported));
 211     }
 212
 213     // flags specifies which watch to remove; ignored if the loop doesn't support
 214     // separate read/write watches.
 215     void remove_fd_watch(int fd, int flags) noexcept
 216     {
 217         epoll_ctl(epfd, EPOLL_CTL_DEL, fd, nullptr);
 218     }
 219
 220     void remove_fd_watch_nolock(int fd, int flags) noexcept
 221     {
 222         remove_fd_watch(fd, flags);
 223     }
 224
 225     void remove_bidi_fd_watch(int fd) noexcept
 226     {
 227         // Shouldn't be called for epoll.
 228         remove_fd_watch(fd, IN_EVENTS | OUT_EVENTS);
 229     }
 230
 231     // Note this will *replace* the old flags with the new, that is,
 232     // it can enable *or disable* read/write events.
 233     void enable_fd_watch(int fd, void *userdata, int flags) noexcept
 234     {
 235         struct epoll_event epevent;
 236         // epevent.data.fd = fd;
 237         epevent.data.ptr = userdata;
 238         epevent.events = 0;
 239
 240         if (flags & ONE_SHOT) {
 241             epevent.events = EPOLLONESHOT;
 242         }
 243         if (flags & IN_EVENTS) {
 244             epevent.events |= EPOLLIN;
 245         }
 246         if (flags & OUT_EVENTS) {
 247             epevent.events |= EPOLLOUT;
 248         }
 249
 250         if (epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &epevent) == -1) {
 251             // Shouldn't be able to fail
 252             // throw new std::system_error(errno, std::system_category());
 253         }
 254     }
 255
 256     void enable_fd_watch_nolock(int fd, void *userdata, int flags)
 257     {
 258         enable_fd_watch(fd, userdata, flags);
 259     }
 260
 261     void disable_fd_watch(int fd, int flags) noexcept
 262     {
 263         struct epoll_event epevent;
 264         // epevent.data.fd = fd;
 265         epevent.data.ptr = nullptr;
 266         epevent.events = 0;
 267
 268         // Epoll documentation says that hangup will still be reported, need to check
 269         // whether this is really the case. Suspect it is really only the case if
 270         // EPOLLIN is set.
 271         if (epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &epevent) == -1) {
 272             // Let's assume that this can't fail.
 273             // throw new std::system_error(errno, std::system_category());
 274         }
 275     }
 276
 277     void disable_fd_watch_nolock(int fd, int flags) noexcept
 278     {
 279         disable_fd_watch(fd, flags);
 280     }
 281
 282     // Note signal should be masked before call.
 283     void add_signal_watch(int signo, void *userdata)
 284     {
 285         std::lock_guard<decltype(Base::lock)> guard(Base::lock);
 286         add_signal_watch_nolock(signo, userdata);
 287     }
 288
 289     // Note signal should be masked before call.
 290     void add_signal_watch_nolock(int signo, void *userdata)
 291     {
 292         sigdataMap[signo] = userdata;
 293
 294         // Modify the signal fd to watch the new signal
 295         bool was_no_sigfd = (sigfd == -1);
 296         sigaddset(&sigmask, signo);
 297         sigfd = signalfd(sigfd, &sigmask, SFD_NONBLOCK | SFD_CLOEXEC);
 298         if (sigfd == -1) {
 299             throw new std::system_error(errno, std::system_category());
 300         }
 301
 302         if (was_no_sigfd) {
 303             // Add the signalfd to the epoll set.
 304             struct epoll_event epevent;
 305             epevent.data.ptr = &sigfd;
 306             epevent.events = EPOLLIN;
 307             // No need for EPOLLONESHOT - we can pull the signals out
 308             // as we see them.
 309             if (epoll_ctl(epfd, EPOLL_CTL_ADD, sigfd, &epevent) == -1) {
 310                 close(sigfd);
 311                 sigfd = -1;
 312                 throw new std::system_error(errno, std::system_category());
 313             }
 314         }
 315     }
 316
 317     // Note, called with lock held:
 318     void rearm_signal_watch_nolock(int signo, void *userdata) noexcept
 319     {
 320         sigaddset(&sigmask, signo);
 321         signalfd(sigfd, &sigmask, SFD_NONBLOCK | SFD_CLOEXEC);
 322     }
 323
 324     void remove_signal_watch_nolock(int signo) noexcept
 325     {
 326         sigdelset(&sigmask, signo);
 327         signalfd(sigfd, &sigmask, 0);
 328     }
 329
 330     void remove_signal_watch(int signo) noexcept
 331     {
 332         std::lock_guard<decltype(Base::lock)> guard(Base::lock);
 333         remove_signal_watch_nolock(signo);
 334     }
 335
 336     // If events are pending, process an unspecified number of them.
 337     // If no events are pending, wait until one event is received and
 338     // process this event (and possibly any other events received
 339     // simultaneously).
 340     // If processing an event removes a watch, there is a possibility
 341     // that the watched event will still be reported (if it has
 342     // occurred) before pull_events() returns.
 343     //
 344     //  do_wait - if false, returns immediately if no events are
 345     //            pending.
 346     void pull_events(bool do_wait)
 347     {
 348         epoll_event events[16];
 349         int r = epoll_wait(epfd, events, 16, do_wait ? -1 : 0);
 350         if (r == -1 || r == 0) {
 351             // signal or no events
 352             return;
 353         }
 354
 355         do {
 356             process_events(events, r);
 357             r = epoll_wait(epfd, events, 16, 0);
 358         } while (r > 0);
 359     }
 360 };
 361
 362 } // end namespace