cannam@62: // Copyright (c) 2013-2014 Sandstorm Development Group, Inc. and contributors cannam@62: // Licensed under the MIT License: cannam@62: // cannam@62: // Permission is hereby granted, free of charge, to any person obtaining a copy cannam@62: // of this software and associated documentation files (the "Software"), to deal cannam@62: // in the Software without restriction, including without limitation the rights cannam@62: // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell cannam@62: // copies of the Software, and to permit persons to whom the Software is cannam@62: // furnished to do so, subject to the following conditions: cannam@62: // cannam@62: // The above copyright notice and this permission notice shall be included in cannam@62: // all copies or substantial portions of the Software. cannam@62: // cannam@62: // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR cannam@62: // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, cannam@62: // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE cannam@62: // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER cannam@62: // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, cannam@62: // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN cannam@62: // THE SOFTWARE. cannam@62: cannam@62: #ifndef KJ_ASYNC_UNIX_H_ cannam@62: #define KJ_ASYNC_UNIX_H_ cannam@62: cannam@62: #if _WIN32 cannam@62: #error "This file is Unix-specific. On Windows, include async-win32.h instead." cannam@62: #endif cannam@62: cannam@62: #if defined(__GNUC__) && !KJ_HEADER_WARNINGS cannam@62: #pragma GCC system_header cannam@62: #endif cannam@62: cannam@62: #include "async.h" cannam@62: #include "time.h" cannam@62: #include "vector.h" cannam@62: #include "io.h" cannam@62: #include cannam@62: cannam@62: #if __linux__ && !__BIONIC__ && !defined(KJ_USE_EPOLL) cannam@62: // Default to epoll on Linux, except on Bionic (Android) which doesn't have signalfd.h. cannam@62: #define KJ_USE_EPOLL 1 cannam@62: #endif cannam@62: cannam@62: namespace kj { cannam@62: cannam@62: class UnixEventPort: public EventPort { cannam@62: // An EventPort implementation which can wait for events on file descriptors as well as signals. cannam@62: // This API only makes sense on Unix. cannam@62: // cannam@62: // The implementation uses `poll()` or possibly a platform-specific API (e.g. epoll, kqueue). cannam@62: // To also wait on signals without race conditions, the implementation may block signals until cannam@62: // just before `poll()` while using a signal handler which `siglongjmp()`s back to just before cannam@62: // the signal was unblocked, or it may use a nicer platform-specific API like signalfd. cannam@62: // cannam@62: // The implementation reserves a signal for internal use. By default, it uses SIGUSR1. If you cannam@62: // need to use SIGUSR1 for something else, you must offer a different signal by calling cannam@62: // setReservedSignal() at startup. cannam@62: // cannam@62: // WARNING: A UnixEventPort can only be used in the thread and process that created it. In cannam@62: // particular, note that after a fork(), a UnixEventPort created in the parent process will cannam@62: // not work correctly in the child, even if the parent ceases to use its copy. In particular cannam@62: // note that this means that server processes which daemonize themselves at startup must wait cannam@62: // until after daemonization to create a UnixEventPort. cannam@62: cannam@62: public: cannam@62: UnixEventPort(); cannam@62: ~UnixEventPort() noexcept(false); cannam@62: cannam@62: class FdObserver; cannam@62: // Class that watches an fd for readability or writability. See definition below. cannam@62: cannam@62: Promise onSignal(int signum); cannam@62: // When the given signal is delivered to this thread, return the corresponding siginfo_t. cannam@62: // The signal must have been captured using `captureSignal()`. cannam@62: // cannam@62: // If `onSignal()` has not been called, the signal will remain blocked in this thread. cannam@62: // Therefore, a signal which arrives before `onSignal()` was called will not be "missed" -- the cannam@62: // next call to 'onSignal()' will receive it. Also, you can control which thread receives a cannam@62: // process-wide signal by only calling `onSignal()` on that thread's event loop. cannam@62: // cannam@62: // The result of waiting on the same signal twice at once is undefined. cannam@62: cannam@62: static void captureSignal(int signum); cannam@62: // Arranges for the given signal to be captured and handled via UnixEventPort, so that you may cannam@62: // then pass it to `onSignal()`. This method is static because it registers a signal handler cannam@62: // which applies process-wide. If any other threads exist in the process when `captureSignal()` cannam@62: // is called, you *must* set the signal mask in those threads to block this signal, otherwise cannam@62: // terrible things will happen if the signal happens to be delivered to those threads. If at cannam@62: // all possible, call `captureSignal()` *before* creating threads, so that threads you create in cannam@62: // the future will inherit the proper signal mask. cannam@62: // cannam@62: // To un-capture a signal, simply install a different signal handler and then un-block it from cannam@62: // the signal mask. cannam@62: cannam@62: static void setReservedSignal(int signum); cannam@62: // Sets the signal number which `UnixEventPort` reserves for internal use. If your application cannam@62: // needs to use SIGUSR1, call this at startup (before any calls to `captureSignal()` and before cannam@62: // constructing an `UnixEventPort`) to offer a different signal. cannam@62: cannam@62: Timer& getTimer() { return timerImpl; } cannam@62: cannam@62: // implements EventPort ------------------------------------------------------ cannam@62: bool wait() override; cannam@62: bool poll() override; cannam@62: void wake() const override; cannam@62: cannam@62: private: cannam@62: struct TimerSet; // Defined in source file to avoid STL include. cannam@62: class TimerPromiseAdapter; cannam@62: class SignalPromiseAdapter; cannam@62: cannam@62: TimerImpl timerImpl; cannam@62: cannam@62: SignalPromiseAdapter* signalHead = nullptr; cannam@62: SignalPromiseAdapter** signalTail = &signalHead; cannam@62: cannam@62: TimePoint readClock(); cannam@62: void gotSignal(const siginfo_t& siginfo); cannam@62: cannam@62: friend class TimerPromiseAdapter; cannam@62: cannam@62: #if KJ_USE_EPOLL cannam@62: AutoCloseFd epollFd; cannam@62: AutoCloseFd signalFd; cannam@62: AutoCloseFd eventFd; // Used for cross-thread wakeups. cannam@62: cannam@62: sigset_t signalFdSigset; cannam@62: // Signal mask as currently set on the signalFd. Tracked so we can detect whether or not it cannam@62: // needs updating. cannam@62: cannam@62: bool doEpollWait(int timeout); cannam@62: cannam@62: #else cannam@62: class PollContext; cannam@62: cannam@62: FdObserver* observersHead = nullptr; cannam@62: FdObserver** observersTail = &observersHead; cannam@62: cannam@62: unsigned long long threadId; // actually pthread_t cannam@62: #endif cannam@62: }; cannam@62: cannam@62: class UnixEventPort::FdObserver { cannam@62: // Object which watches a file descriptor to determine when it is readable or writable. cannam@62: // cannam@62: // For listen sockets, "readable" means that there is a connection to accept(). For everything cannam@62: // else, it means that read() (or recv()) will return data. cannam@62: // cannam@62: // The presence of out-of-band data should NOT fire this event. However, the event may cannam@62: // occasionally fire spuriously (when there is actually no data to read), and one thing that can cannam@62: // cause such spurious events is the arrival of OOB data on certain platforms whose event cannam@62: // interfaces fail to distinguish between regular and OOB data (e.g. Mac OSX). cannam@62: // cannam@62: // WARNING: The exact behavior of this class differs across systems, since event interfaces cannam@62: // vary wildly. Be sure to read the documentation carefully and avoid depending on unspecified cannam@62: // behavior. If at all possible, use the higher-level AsyncInputStream interface instead. cannam@62: cannam@62: public: cannam@62: enum Flags { cannam@62: OBSERVE_READ = 1, cannam@62: OBSERVE_WRITE = 2, cannam@62: OBSERVE_URGENT = 4, cannam@62: OBSERVE_READ_WRITE = OBSERVE_READ | OBSERVE_WRITE cannam@62: }; cannam@62: cannam@62: FdObserver(UnixEventPort& eventPort, int fd, uint flags); cannam@62: // Begin watching the given file descriptor for readability. Only one ReadObserver may exist cannam@62: // for a given file descriptor at a time. cannam@62: cannam@62: ~FdObserver() noexcept(false); cannam@62: cannam@62: KJ_DISALLOW_COPY(FdObserver); cannam@62: cannam@62: Promise whenBecomesReadable(); cannam@62: // Resolves the next time the file descriptor transitions from having no data to read to having cannam@62: // some data to read. cannam@62: // cannam@62: // KJ uses "edge-triggered" event notification whenever possible. As a result, it is an error cannam@62: // to call this method when there is already data in the read buffer which has been there since cannam@62: // prior to the last turn of the event loop or prior to creation FdWatcher. In this case, it is cannam@62: // unspecified whether the promise will ever resolve -- it depends on the underlying event cannam@62: // mechanism being used. cannam@62: // cannam@62: // In order to avoid this problem, make sure that you only call `whenBecomesReadable()` cannam@62: // only at times when you know the buffer is empty. You know this for sure when one of the cannam@62: // following happens: cannam@62: // * read() or recv() fails with EAGAIN or EWOULDBLOCK. (You MUST have non-blocking mode cannam@62: // enabled on the fd!) cannam@62: // * The file descriptor is a regular byte-oriented object (like a socket or pipe), cannam@62: // read() or recv() returns fewer than the number of bytes requested, and `atEndHint()` cannam@62: // returns false. This can only happen if the buffer is empty but EOF is not reached. (Note, cannam@62: // though, that for record-oriented file descriptors like Linux's inotify interface, this cannam@62: // rule does not hold, because it could simply be that the next record did not fit into the cannam@62: // space available.) cannam@62: // cannam@62: // It is an error to call `whenBecomesReadable()` again when the promise returned previously cannam@62: // has not yet resolved. If you do this, the previous promise may throw an exception. cannam@62: cannam@62: inline Maybe atEndHint() { return atEnd; } cannam@62: // Returns true if the event system has indicated that EOF has been received. There may still cannam@62: // be data in the read buffer, but once that is gone, there's nothing left. cannam@62: // cannam@62: // Returns false if the event system has indicated that EOF had NOT been received as of the cannam@62: // last turn of the event loop. cannam@62: // cannam@62: // Returns nullptr if the event system does not know whether EOF has been reached. In this cannam@62: // case, the only way to know for sure is to call read() or recv() and check if it returns cannam@62: // zero. cannam@62: // cannam@62: // This hint may be useful as an optimization to avoid an unnecessary system call. cannam@62: cannam@62: Promise whenBecomesWritable(); cannam@62: // Resolves the next time the file descriptor transitions from having no space available in the cannam@62: // write buffer to having some space available. cannam@62: // cannam@62: // KJ uses "edge-triggered" event notification whenever possible. As a result, it is an error cannam@62: // to call this method when there is already space in the write buffer which has been there cannam@62: // since prior to the last turn of the event loop or prior to creation FdWatcher. In this case, cannam@62: // it is unspecified whether the promise will ever resolve -- it depends on the underlying cannam@62: // event mechanism being used. cannam@62: // cannam@62: // In order to avoid this problem, make sure that you only call `whenBecomesWritable()` cannam@62: // only at times when you know the buffer is full. You know this for sure when one of the cannam@62: // following happens: cannam@62: // * write() or send() fails with EAGAIN or EWOULDBLOCK. (You MUST have non-blocking mode cannam@62: // enabled on the fd!) cannam@62: // * write() or send() succeeds but accepts fewer than the number of bytes provided. This can cannam@62: // only happen if the buffer is full. cannam@62: // cannam@62: // It is an error to call `whenBecomesWritable()` again when the promise returned previously cannam@62: // has not yet resolved. If you do this, the previous promise may throw an exception. cannam@62: cannam@62: Promise whenUrgentDataAvailable(); cannam@62: // Resolves the next time the file descriptor's read buffer contains "urgent" data. cannam@62: // cannam@62: // The conditions for availability of urgent data are specific to the file descriptor's cannam@62: // underlying implementation. cannam@62: // cannam@62: // It is an error to call `whenUrgentDataAvailable()` again when the promise returned previously cannam@62: // has not yet resolved. If you do this, the previous promise may throw an exception. cannam@62: // cannam@62: // WARNING: This has some known weird behavior on macOS. See cannam@62: // https://github.com/sandstorm-io/capnproto/issues/374. cannam@62: cannam@62: private: cannam@62: UnixEventPort& eventPort; cannam@62: int fd; cannam@62: uint flags; cannam@62: cannam@62: kj::Maybe>> readFulfiller; cannam@62: kj::Maybe>> writeFulfiller; cannam@62: kj::Maybe>> urgentFulfiller; cannam@62: // Replaced each time `whenBecomesReadable()` or `whenBecomesWritable()` is called. Reverted to cannam@62: // null every time an event is fired. cannam@62: cannam@62: Maybe atEnd; cannam@62: cannam@62: void fire(short events); cannam@62: cannam@62: #if !KJ_USE_EPOLL cannam@62: FdObserver* next; cannam@62: FdObserver** prev; cannam@62: // Linked list of observers which currently have a non-null readFulfiller or writeFulfiller. cannam@62: // If `prev` is null then the observer is not currently in the list. cannam@62: cannam@62: short getEventMask(); cannam@62: #endif cannam@62: cannam@62: friend class UnixEventPort; cannam@62: }; cannam@62: cannam@62: } // namespace kj cannam@62: cannam@62: #endif // KJ_ASYNC_UNIX_H_