cannam@148: // Copyright (c) 2013-2014 Sandstorm Development Group, Inc. and contributors cannam@148: // Licensed under the MIT License: cannam@148: // cannam@148: // Permission is hereby granted, free of charge, to any person obtaining a copy cannam@148: // of this software and associated documentation files (the "Software"), to deal cannam@148: // in the Software without restriction, including without limitation the rights cannam@148: // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell cannam@148: // copies of the Software, and to permit persons to whom the Software is cannam@148: // furnished to do so, subject to the following conditions: cannam@148: // cannam@148: // The above copyright notice and this permission notice shall be included in cannam@148: // all copies or substantial portions of the Software. cannam@148: // cannam@148: // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR cannam@148: // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, cannam@148: // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE cannam@148: // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER cannam@148: // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, cannam@148: // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN cannam@148: // THE SOFTWARE. cannam@148: cannam@148: #ifndef KJ_ASYNC_IO_H_ cannam@148: #define KJ_ASYNC_IO_H_ cannam@148: cannam@148: #if defined(__GNUC__) && !KJ_HEADER_WARNINGS cannam@148: #pragma GCC system_header cannam@148: #endif cannam@148: cannam@148: #include "async.h" cannam@148: #include "function.h" cannam@148: #include "thread.h" cannam@148: #include "time.h" cannam@148: cannam@148: struct sockaddr; cannam@148: cannam@148: namespace kj { cannam@148: cannam@148: #if _WIN32 cannam@148: class Win32EventPort; cannam@148: #else cannam@148: class UnixEventPort; cannam@148: #endif cannam@148: cannam@148: class NetworkAddress; cannam@148: class AsyncOutputStream; cannam@148: cannam@148: // ======================================================================================= cannam@148: // Streaming I/O cannam@148: cannam@148: class AsyncInputStream { cannam@148: // Asynchronous equivalent of InputStream (from io.h). cannam@148: cannam@148: public: cannam@148: virtual Promise read(void* buffer, size_t minBytes, size_t maxBytes); cannam@148: virtual Promise tryRead(void* buffer, size_t minBytes, size_t maxBytes) = 0; cannam@148: cannam@148: Promise read(void* buffer, size_t bytes); cannam@148: cannam@148: virtual Maybe tryGetLength(); cannam@148: // Get the remaining number of bytes that will be produced by this stream, if known. cannam@148: // cannam@148: // This is used e.g. to fill in the Content-Length header of an HTTP message. If unknown, the cannam@148: // HTTP implementation may need to fall back to Transfer-Encoding: chunked. cannam@148: // cannam@148: // The default implementation always returns null. cannam@148: cannam@148: virtual Promise pumpTo( cannam@148: AsyncOutputStream& output, uint64_t amount = kj::maxValue); cannam@148: // Read `amount` bytes from this stream (or to EOF) and write them to `output`, returning the cannam@148: // total bytes actually pumped (which is only less than `amount` if EOF was reached). cannam@148: // cannam@148: // Override this if your stream type knows how to pump itself to certain kinds of output cannam@148: // streams more efficiently than via the naive approach. You can use cannam@148: // kj::dynamicDowncastIfAvailable() to test for stream types you recognize, and if none match, cannam@148: // delegate to the default implementation. cannam@148: // cannam@148: // The default implementation first tries calling output.tryPumpFrom(), but if that fails, it cannam@148: // performs a naive pump by allocating a buffer and reading to it / writing from it in a loop. cannam@148: cannam@148: Promise> readAllBytes(); cannam@148: Promise readAllText(); cannam@148: // Read until EOF and return as one big byte array or string. cannam@148: }; cannam@148: cannam@148: class AsyncOutputStream { cannam@148: // Asynchronous equivalent of OutputStream (from io.h). cannam@148: cannam@148: public: cannam@148: virtual Promise write(const void* buffer, size_t size) = 0; cannam@148: virtual Promise write(ArrayPtr> pieces) = 0; cannam@148: cannam@148: virtual Maybe> tryPumpFrom( cannam@148: AsyncInputStream& input, uint64_t amount = kj::maxValue); cannam@148: // Implements double-dispatch for AsyncInputStream::pumpTo(). cannam@148: // cannam@148: // This method should only be called from within an implementation of pumpTo(). cannam@148: // cannam@148: // This method examines the type of `input` to find optimized ways to pump data from it to this cannam@148: // output stream. If it finds one, it performs the pump. Otherwise, it returns null. cannam@148: // cannam@148: // The default implementation always returns null. cannam@148: }; cannam@148: cannam@148: class AsyncIoStream: public AsyncInputStream, public AsyncOutputStream { cannam@148: // A combination input and output stream. cannam@148: cannam@148: public: cannam@148: virtual void shutdownWrite() = 0; cannam@148: // Cleanly shut down just the write end of the stream, while keeping the read end open. cannam@148: cannam@148: virtual void abortRead() {} cannam@148: // Similar to shutdownWrite, but this will shut down the read end of the stream, and should only cannam@148: // be called when an error has occurred. cannam@148: cannam@148: virtual void getsockopt(int level, int option, void* value, uint* length); cannam@148: virtual void setsockopt(int level, int option, const void* value, uint length); cannam@148: // Corresponds to getsockopt() and setsockopt() syscalls. Will throw an "unimplemented" exception cannam@148: // if the stream is not a socket or the option is not appropriate for the socket type. The cannam@148: // default implementations always throw "unimplemented". cannam@148: cannam@148: virtual void getsockname(struct sockaddr* addr, uint* length); cannam@148: virtual void getpeername(struct sockaddr* addr, uint* length); cannam@148: // Corresponds to getsockname() and getpeername() syscalls. Will throw an "unimplemented" cannam@148: // exception if the stream is not a socket. The default implementations always throw cannam@148: // "unimplemented". cannam@148: // cannam@148: // Note that we don't provide methods that return NetworkAddress because it usually wouldn't cannam@148: // be useful. You can't connect() to or listen() on these addresses, obviously, because they are cannam@148: // ephemeral addresses for a single connection. cannam@148: }; cannam@148: cannam@148: struct OneWayPipe { cannam@148: // A data pipe with an input end and an output end. (Typically backed by pipe() system call.) cannam@148: cannam@148: Own in; cannam@148: Own out; cannam@148: }; cannam@148: cannam@148: struct TwoWayPipe { cannam@148: // A data pipe that supports sending in both directions. Each end's output sends data to the cannam@148: // other end's input. (Typically backed by socketpair() system call.) cannam@148: cannam@148: Own ends[2]; cannam@148: }; cannam@148: cannam@148: class ConnectionReceiver { cannam@148: // Represents a server socket listening on a port. cannam@148: cannam@148: public: cannam@148: virtual Promise> accept() = 0; cannam@148: // Accept the next incoming connection. cannam@148: cannam@148: virtual uint getPort() = 0; cannam@148: // Gets the port number, if applicable (i.e. if listening on IP). This is useful if you didn't cannam@148: // specify a port when constructing the NetworkAddress -- one will have been assigned cannam@148: // automatically. cannam@148: cannam@148: virtual void getsockopt(int level, int option, void* value, uint* length); cannam@148: virtual void setsockopt(int level, int option, const void* value, uint length); cannam@148: // Same as the methods of AsyncIoStream. cannam@148: }; cannam@148: cannam@148: // ======================================================================================= cannam@148: // Datagram I/O cannam@148: cannam@148: class AncillaryMessage { cannam@148: // Represents an ancillary message (aka control message) received using the recvmsg() system cannam@148: // call (or equivalent). Most apps will not use this. cannam@148: cannam@148: public: cannam@148: inline AncillaryMessage(int level, int type, ArrayPtr data); cannam@148: AncillaryMessage() = default; cannam@148: cannam@148: inline int getLevel() const; cannam@148: // Originating protocol / socket level. cannam@148: cannam@148: inline int getType() const; cannam@148: // Protocol-specific message type. cannam@148: cannam@148: template cannam@148: inline Maybe as(); cannam@148: // Interpret the ancillary message as the given struct type. Most ancillary messages are some cannam@148: // sort of struct, so this is a convenient way to access it. Returns nullptr if the message cannam@148: // is smaller than the struct -- this can happen if the message was truncated due to cannam@148: // insufficient ancillary buffer space. cannam@148: cannam@148: template cannam@148: inline ArrayPtr asArray(); cannam@148: // Interpret the ancillary message as an array of items. If the message size does not evenly cannam@148: // divide into elements of type T, the remainder is discarded -- this can happen if the message cannam@148: // was truncated due to insufficient ancillary buffer space. cannam@148: cannam@148: private: cannam@148: int level; cannam@148: int type; cannam@148: ArrayPtr data; cannam@148: // Message data. In most cases you should use `as()` or `asArray()`. cannam@148: }; cannam@148: cannam@148: class DatagramReceiver { cannam@148: // Class encapsulating the recvmsg() system call. You must specify the DatagramReceiver's cannam@148: // capacity in advance; if a received packet is larger than the capacity, it will be truncated. cannam@148: cannam@148: public: cannam@148: virtual Promise receive() = 0; cannam@148: // Receive a new message, overwriting this object's content. cannam@148: // cannam@148: // receive() may reuse the same buffers for content and ancillary data with each call. cannam@148: cannam@148: template cannam@148: struct MaybeTruncated { cannam@148: T value; cannam@148: cannam@148: bool isTruncated; cannam@148: // True if the Receiver's capacity was insufficient to receive the value and therefore the cannam@148: // value is truncated. cannam@148: }; cannam@148: cannam@148: virtual MaybeTruncated> getContent() = 0; cannam@148: // Get the content of the datagram. cannam@148: cannam@148: virtual MaybeTruncated> getAncillary() = 0; cannam@148: // Ancilarry messages received with the datagram. See the recvmsg() system call and the cmsghdr cannam@148: // struct. Most apps don't need this. cannam@148: // cannam@148: // If the returned value is truncated, then the last message in the array may itself be cannam@148: // truncated, meaning its as() method will return nullptr or its asArray() method will cannam@148: // return fewer elements than expected. Truncation can also mean that additional messages were cannam@148: // available but discarded. cannam@148: cannam@148: virtual NetworkAddress& getSource() = 0; cannam@148: // Get the datagram sender's address. cannam@148: cannam@148: struct Capacity { cannam@148: size_t content = 8192; cannam@148: // How much space to allocate for the datagram content. If a datagram is received that is cannam@148: // larger than this, it will be truncated, with no way to recover the tail. cannam@148: cannam@148: size_t ancillary = 0; cannam@148: // How much space to allocate for ancillary messages. As with content, if the ancillary data cannam@148: // is larger than this, it will be truncated. cannam@148: }; cannam@148: }; cannam@148: cannam@148: class DatagramPort { cannam@148: public: cannam@148: virtual Promise send(const void* buffer, size_t size, NetworkAddress& destination) = 0; cannam@148: virtual Promise send(ArrayPtr> pieces, cannam@148: NetworkAddress& destination) = 0; cannam@148: cannam@148: virtual Own makeReceiver( cannam@148: DatagramReceiver::Capacity capacity = DatagramReceiver::Capacity()) = 0; cannam@148: // Create a new `Receiver` that can be used to receive datagrams. `capacity` specifies how much cannam@148: // space to allocate for the received message. The `DatagramPort` must outlive the `Receiver`. cannam@148: cannam@148: virtual uint getPort() = 0; cannam@148: // Gets the port number, if applicable (i.e. if listening on IP). This is useful if you didn't cannam@148: // specify a port when constructing the NetworkAddress -- one will have been assigned cannam@148: // automatically. cannam@148: cannam@148: virtual void getsockopt(int level, int option, void* value, uint* length); cannam@148: virtual void setsockopt(int level, int option, const void* value, uint length); cannam@148: // Same as the methods of AsyncIoStream. cannam@148: }; cannam@148: cannam@148: // ======================================================================================= cannam@148: // Networks cannam@148: cannam@148: class NetworkAddress { cannam@148: // Represents a remote address to which the application can connect. cannam@148: cannam@148: public: cannam@148: virtual Promise> connect() = 0; cannam@148: // Make a new connection to this address. cannam@148: // cannam@148: // The address must not be a wildcard ("*"). If it is an IP address, it must have a port number. cannam@148: cannam@148: virtual Own listen() = 0; cannam@148: // Listen for incoming connections on this address. cannam@148: // cannam@148: // The address must be local. cannam@148: cannam@148: virtual Own bindDatagramPort(); cannam@148: // Open this address as a datagram (e.g. UDP) port. cannam@148: // cannam@148: // The address must be local. cannam@148: cannam@148: virtual Own clone() = 0; cannam@148: // Returns an equivalent copy of this NetworkAddress. cannam@148: cannam@148: virtual String toString() = 0; cannam@148: // Produce a human-readable string which hopefully can be passed to Network::parseAddress() cannam@148: // to reproduce this address, although whether or not that works of course depends on the Network cannam@148: // implementation. This should be called only to display the address to human users, who will cannam@148: // hopefully know what they are able to do with it. cannam@148: }; cannam@148: cannam@148: class Network { cannam@148: // Factory for NetworkAddress instances, representing the network services offered by the cannam@148: // operating system. cannam@148: // cannam@148: // This interface typically represents broad authority, and well-designed code should limit its cannam@148: // use to high-level startup code and user interaction. Low-level APIs should accept cannam@148: // NetworkAddress instances directly and work from there, if at all possible. cannam@148: cannam@148: public: cannam@148: virtual Promise> parseAddress(StringPtr addr, uint portHint = 0) = 0; cannam@148: // Construct a network address from a user-provided string. The format of the address cannam@148: // strings is not specified at the API level, and application code should make no assumptions cannam@148: // about them. These strings should always be provided by humans, and said humans will know cannam@148: // what format to use in their particular context. cannam@148: // cannam@148: // `portHint`, if provided, specifies the "standard" IP port number for the application-level cannam@148: // service in play. If the address turns out to be an IP address (v4 or v6), and it lacks a cannam@148: // port number, this port will be used. If `addr` lacks a port number *and* `portHint` is cannam@148: // omitted, then the returned address will only support listen() and bindDatagramPort() cannam@148: // (not connect()), and an unused port will be chosen each time one of those methods is called. cannam@148: cannam@148: virtual Own getSockaddr(const void* sockaddr, uint len) = 0; cannam@148: // Construct a network address from a legacy struct sockaddr. cannam@148: }; cannam@148: cannam@148: // ======================================================================================= cannam@148: // I/O Provider cannam@148: cannam@148: class AsyncIoProvider { cannam@148: // Class which constructs asynchronous wrappers around the operating system's I/O facilities. cannam@148: // cannam@148: // Generally, the implementation of this interface must integrate closely with a particular cannam@148: // `EventLoop` implementation. Typically, the EventLoop implementation itself will provide cannam@148: // an AsyncIoProvider. cannam@148: cannam@148: public: cannam@148: virtual OneWayPipe newOneWayPipe() = 0; cannam@148: // Creates an input/output stream pair representing the ends of a one-way pipe (e.g. created with cannam@148: // the pipe(2) system call). cannam@148: cannam@148: virtual TwoWayPipe newTwoWayPipe() = 0; cannam@148: // Creates two AsyncIoStreams representing the two ends of a two-way pipe (e.g. created with cannam@148: // socketpair(2) system call). Data written to one end can be read from the other. cannam@148: cannam@148: virtual Network& getNetwork() = 0; cannam@148: // Creates a new `Network` instance representing the networks exposed by the operating system. cannam@148: // cannam@148: // DO NOT CALL THIS except at the highest levels of your code, ideally in the main() function. If cannam@148: // you call this from low-level code, then you are preventing higher-level code from injecting an cannam@148: // alternative implementation. Instead, if your code needs to use network functionality, it cannam@148: // should ask for a `Network` as a constructor or method parameter, so that higher-level code can cannam@148: // chose what implementation to use. The system network is essentially a singleton. See: cannam@148: // http://www.object-oriented-security.org/lets-argue/singletons cannam@148: // cannam@148: // Code that uses the system network should not make any assumptions about what kinds of cannam@148: // addresses it will parse, as this could differ across platforms. String addresses should come cannam@148: // strictly from the user, who will know how to write them correctly for their system. cannam@148: // cannam@148: // With that said, KJ currently supports the following string address formats: cannam@148: // - IPv4: "1.2.3.4", "1.2.3.4:80" cannam@148: // - IPv6: "1234:5678::abcd", "[1234:5678::abcd]:80" cannam@148: // - Local IP wildcard (covers both v4 and v6): "*", "*:80" cannam@148: // - Symbolic names: "example.com", "example.com:80", "example.com:http", "1.2.3.4:http" cannam@148: // - Unix domain: "unix:/path/to/socket" cannam@148: cannam@148: struct PipeThread { cannam@148: // A combination of a thread and a two-way pipe that communicates with that thread. cannam@148: // cannam@148: // The fields are intentionally ordered so that the pipe will be destroyed (and therefore cannam@148: // disconnected) before the thread is destroyed (and therefore joined). Thus if the thread cannam@148: // arranges to exit when it detects disconnect, destruction should be clean. cannam@148: cannam@148: Own thread; cannam@148: Own pipe; cannam@148: }; cannam@148: cannam@148: virtual PipeThread newPipeThread( cannam@148: Function startFunc) = 0; cannam@148: // Create a new thread and set up a two-way pipe (socketpair) which can be used to communicate cannam@148: // with it. One end of the pipe is passed to the thread's start function and the other end of cannam@148: // the pipe is returned. The new thread also gets its own `AsyncIoProvider` instance and will cannam@148: // already have an active `EventLoop` when `startFunc` is called. cannam@148: // cannam@148: // TODO(someday): I'm not entirely comfortable with this interface. It seems to be doing too cannam@148: // much at once but I'm not sure how to cleanly break it down. cannam@148: cannam@148: virtual Timer& getTimer() = 0; cannam@148: // Returns a `Timer` based on real time. Time does not pass while event handlers are running -- cannam@148: // it only updates when the event loop polls for system events. This means that calling `now()` cannam@148: // on this timer does not require a system call. cannam@148: // cannam@148: // This timer is not affected by changes to the system date. It is unspecified whether the timer cannam@148: // continues to count while the system is suspended. cannam@148: }; cannam@148: cannam@148: class LowLevelAsyncIoProvider { cannam@148: // Similar to `AsyncIoProvider`, but represents a lower-level interface that may differ on cannam@148: // different operating systems. You should prefer to use `AsyncIoProvider` over this interface cannam@148: // whenever possible, as `AsyncIoProvider` is portable and friendlier to dependency-injection. cannam@148: // cannam@148: // On Unix, this interface can be used to import native file descriptors into the async framework. cannam@148: // Different implementations of this interface might work on top of different event handling cannam@148: // primitives, such as poll vs. epoll vs. kqueue vs. some higher-level event library. cannam@148: // cannam@148: // On Windows, this interface can be used to import native HANDLEs into the async framework. cannam@148: // Different implementations of this interface might work on top of different event handling cannam@148: // primitives, such as I/O completion ports vs. completion routines. cannam@148: // cannam@148: // TODO(port): Actually implement Windows support. cannam@148: cannam@148: public: cannam@148: // --------------------------------------------------------------------------- cannam@148: // Unix-specific stuff cannam@148: cannam@148: enum Flags { cannam@148: // Flags controlling how to wrap a file descriptor. cannam@148: cannam@148: TAKE_OWNERSHIP = 1 << 0, cannam@148: // The returned object should own the file descriptor, automatically closing it when destroyed. cannam@148: // The close-on-exec flag will be set on the descriptor if it is not already. cannam@148: // cannam@148: // If this flag is not used, then the file descriptor is not automatically closed and the cannam@148: // close-on-exec flag is not modified. cannam@148: cannam@148: #if !_WIN32 cannam@148: ALREADY_CLOEXEC = 1 << 1, cannam@148: // Indicates that the close-on-exec flag is known already to be set, so need not be set again. cannam@148: // Only relevant when combined with TAKE_OWNERSHIP. cannam@148: // cannam@148: // On Linux, all system calls which yield new file descriptors have flags or variants which cannam@148: // set the close-on-exec flag immediately. Unfortunately, other OS's do not. cannam@148: cannam@148: ALREADY_NONBLOCK = 1 << 2 cannam@148: // Indicates that the file descriptor is known already to be in non-blocking mode, so the flag cannam@148: // need not be set again. Otherwise, all wrap*Fd() methods will enable non-blocking mode cannam@148: // automatically. cannam@148: // cannam@148: // On Linux, all system calls which yield new file descriptors have flags or variants which cannam@148: // enable non-blocking mode immediately. Unfortunately, other OS's do not. cannam@148: #endif cannam@148: }; cannam@148: cannam@148: #if _WIN32 cannam@148: typedef uintptr_t Fd; cannam@148: // On Windows, the `fd` parameter to each of these methods must be a SOCKET, and must have the cannam@148: // flag WSA_FLAG_OVERLAPPED (which socket() uses by default, but WSASocket() wants you to specify cannam@148: // explicitly). cannam@148: #else cannam@148: typedef int Fd; cannam@148: // On Unix, any arbitrary file descriptor is supported. cannam@148: #endif cannam@148: cannam@148: virtual Own wrapInputFd(Fd fd, uint flags = 0) = 0; cannam@148: // Create an AsyncInputStream wrapping a file descriptor. cannam@148: // cannam@148: // `flags` is a bitwise-OR of the values of the `Flags` enum. cannam@148: cannam@148: virtual Own wrapOutputFd(Fd fd, uint flags = 0) = 0; cannam@148: // Create an AsyncOutputStream wrapping a file descriptor. cannam@148: // cannam@148: // `flags` is a bitwise-OR of the values of the `Flags` enum. cannam@148: cannam@148: virtual Own wrapSocketFd(Fd fd, uint flags = 0) = 0; cannam@148: // Create an AsyncIoStream wrapping a socket file descriptor. cannam@148: // cannam@148: // `flags` is a bitwise-OR of the values of the `Flags` enum. cannam@148: cannam@148: virtual Promise> wrapConnectingSocketFd( cannam@148: Fd fd, const struct sockaddr* addr, uint addrlen, uint flags = 0) = 0; cannam@148: // Create an AsyncIoStream wrapping a socket and initiate a connection to the given address. cannam@148: // The returned promise does not resolve until connection has completed. cannam@148: // cannam@148: // `flags` is a bitwise-OR of the values of the `Flags` enum. cannam@148: cannam@148: virtual Own wrapListenSocketFd(Fd fd, uint flags = 0) = 0; cannam@148: // Create an AsyncIoStream wrapping a listen socket file descriptor. This socket should already cannam@148: // have had `bind()` and `listen()` called on it, so it's ready for `accept()`. cannam@148: // cannam@148: // `flags` is a bitwise-OR of the values of the `Flags` enum. cannam@148: cannam@148: virtual Own wrapDatagramSocketFd(Fd fd, uint flags = 0); cannam@148: cannam@148: virtual Timer& getTimer() = 0; cannam@148: // Returns a `Timer` based on real time. Time does not pass while event handlers are running -- cannam@148: // it only updates when the event loop polls for system events. This means that calling `now()` cannam@148: // on this timer does not require a system call. cannam@148: // cannam@148: // This timer is not affected by changes to the system date. It is unspecified whether the timer cannam@148: // continues to count while the system is suspended. cannam@148: }; cannam@148: cannam@148: Own newAsyncIoProvider(LowLevelAsyncIoProvider& lowLevel); cannam@148: // Make a new AsyncIoProvider wrapping a `LowLevelAsyncIoProvider`. cannam@148: cannam@148: struct AsyncIoContext { cannam@148: Own lowLevelProvider; cannam@148: Own provider; cannam@148: WaitScope& waitScope; cannam@148: cannam@148: #if _WIN32 cannam@148: Win32EventPort& win32EventPort; cannam@148: #else cannam@148: UnixEventPort& unixEventPort; cannam@148: // TEMPORARY: Direct access to underlying UnixEventPort, mainly for waiting on signals. This cannam@148: // field will go away at some point when we have a chance to improve these interfaces. cannam@148: #endif cannam@148: }; cannam@148: cannam@148: AsyncIoContext setupAsyncIo(); cannam@148: // Convenience method which sets up the current thread with everything it needs to do async I/O. cannam@148: // The returned objects contain an `EventLoop` which is wrapping an appropriate `EventPort` for cannam@148: // doing I/O on the host system, so everything is ready for the thread to start making async calls cannam@148: // and waiting on promises. cannam@148: // cannam@148: // You would typically call this in your main() loop or in the start function of a thread. cannam@148: // Example: cannam@148: // cannam@148: // int main() { cannam@148: // auto ioContext = kj::setupAsyncIo(); cannam@148: // cannam@148: // // Now we can call an async function. cannam@148: // Promise textPromise = getHttp(*ioContext.provider, "http://example.com"); cannam@148: // cannam@148: // // And we can wait for the promise to complete. Note that you can only use `wait()` cannam@148: // // from the top level, not from inside a promise callback. cannam@148: // String text = textPromise.wait(ioContext.waitScope); cannam@148: // print(text); cannam@148: // return 0; cannam@148: // } cannam@148: // cannam@148: // WARNING: An AsyncIoContext can only be used in the thread and process that created it. In cannam@148: // particular, note that after a fork(), an AsyncIoContext created in the parent process will cannam@148: // not work correctly in the child, even if the parent ceases to use its copy. In particular cannam@148: // note that this means that server processes which daemonize themselves at startup must wait cannam@148: // until after daemonization to create an AsyncIoContext. cannam@148: cannam@148: // ======================================================================================= cannam@148: // inline implementation details cannam@148: cannam@148: inline AncillaryMessage::AncillaryMessage( cannam@148: int level, int type, ArrayPtr data) cannam@148: : level(level), type(type), data(data) {} cannam@148: cannam@148: inline int AncillaryMessage::getLevel() const { return level; } cannam@148: inline int AncillaryMessage::getType() const { return type; } cannam@148: cannam@148: template cannam@148: inline Maybe AncillaryMessage::as() { cannam@148: if (data.size() >= sizeof(T)) { cannam@148: return *reinterpret_cast(data.begin()); cannam@148: } else { cannam@148: return nullptr; cannam@148: } cannam@148: } cannam@148: cannam@148: template cannam@148: inline ArrayPtr AncillaryMessage::asArray() { cannam@148: return arrayPtr(reinterpret_cast(data.begin()), data.size() / sizeof(T)); cannam@148: } cannam@148: cannam@148: } // namespace kj cannam@148: cannam@148: #endif // KJ_ASYNC_IO_H_