async-io.h 14.6 KB
Newer Older
Kenton Varda's avatar
Kenton Varda committed
1 2
// Copyright (c) 2013-2014 Sandstorm Development Group, Inc. and contributors
// Licensed under the MIT License:
3
//
Kenton Varda's avatar
Kenton Varda committed
4 5 6 7 8 9
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
10
//
Kenton Varda's avatar
Kenton Varda committed
11 12
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
13
//
Kenton Varda's avatar
Kenton Varda committed
14 15 16 17 18 19 20
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
21 22 23 24 25

#ifndef KJ_ASYNC_IO_H_
#define KJ_ASYNC_IO_H_

#include "async.h"
26
#include "function.h"
27
#include "thread.h"
28
#include "time.h"
29 30 31

namespace kj {

32 33
class UnixEventPort;

34
class AsyncInputStream {
35 36
  // Asynchronous equivalent of InputStream (from io.h).

37 38 39 40
public:
  virtual Promise<size_t> read(void* buffer, size_t minBytes, size_t maxBytes) = 0;
  virtual Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) = 0;

41
  Promise<void> read(void* buffer, size_t bytes);
42 43 44
};

class AsyncOutputStream {
45 46
  // Asynchronous equivalent of OutputStream (from io.h).

47 48 49 50 51 52
public:
  virtual Promise<void> write(const void* buffer, size_t size) = 0;
  virtual Promise<void> write(ArrayPtr<const ArrayPtr<const byte>> pieces) = 0;
};

class AsyncIoStream: public AsyncInputStream, public AsyncOutputStream {
53 54
  // A combination input and output stream.

55
public:
56 57
  virtual void shutdownWrite() = 0;
  // Cleanly shut down just the write end of the stream, while keeping the read end open.
58 59 60
};

class ConnectionReceiver {
61 62
  // Represents a server socket listening on a port.

63 64
public:
  virtual Promise<Own<AsyncIoStream>> accept() = 0;
65
  // Accept the next incoming connection.
66 67 68 69 70 71

  virtual uint getPort() = 0;
  // Gets the port number, if applicable (i.e. if listening on IP).  This is useful if you didn't
  // specify a port when constructing the LocalAddress -- one will have been assigned automatically.
};

Kenton Varda's avatar
Kenton Varda committed
72
class NetworkAddress {
73 74 75 76
  // Represents a remote address to which the application can connect.

public:
  virtual Promise<Own<AsyncIoStream>> connect() = 0;
77
  // Make a new connection to this address.
Kenton Varda's avatar
Kenton Varda committed
78 79 80 81 82 83 84
  //
  // The address must not be a wildcard ("*").  If it is an IP address, it must have a port number.

  virtual Own<ConnectionReceiver> listen() = 0;
  // Listen for incoming connections on this address.
  //
  // The address must be local.
85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112

  virtual String toString() = 0;
  // Produce a human-readable string which hopefully can be passed to Network::parseRemoteAddress()
  // to reproduce this address, although whether or not that works of course depends on the Network
  // implementation.  This should be called only to display the address to human users, who will
  // hopefully know what they are able to do with it.
};

class LocalAddress {
  // Represents a local address on which the application can potentially accept connections.

public:
  virtual String toString() = 0;
  // Produce a human-readable string which hopefully can be passed to Network::parseRemoteAddress()
  // to reproduce this address, although whether or not that works of course depends on the Network
  // implementation.  This should be called only to display the address to human users, who will
  // hopefully know what they are able to do with it.
};

class Network {
  // Factory for LocalAddress and RemoteAddress instances, representing the network services
  // offered by the operating system.
  //
  // This interface typically represents broad authority, and well-designed code should limit its
  // use to high-level startup code and user interaction.  Low-level APIs should accept
  // LocalAddress and/or RemoteAddress instances directly and work from there, if at all possible.

public:
Kenton Varda's avatar
Kenton Varda committed
113 114
  virtual Promise<Own<NetworkAddress>> parseAddress(StringPtr addr, uint portHint = 0) = 0;
  // Construct a network address from a user-provided string.  The format of the address
115 116 117 118 119 120
  // strings is not specified at the API level, and application code should make no assumptions
  // about them.  These strings should always be provided by humans, and said humans will know
  // what format to use in their particular context.
  //
  // `portHint`, if provided, specifies the "standard" IP port number for the application-level
  // service in play.  If the address turns out to be an IP address (v4 or v6), and it lacks a
Kenton Varda's avatar
Kenton Varda committed
121 122 123
  // port number, this port will be used.  If `addr` lacks a port number *and* `portHint` is
  // omitted, then the returned address will only support listen() (not connect()), and a port
  // will be chosen when listen() is called.
124

Kenton Varda's avatar
Kenton Varda committed
125 126
  virtual Own<NetworkAddress> getSockaddr(const void* sockaddr, uint len) = 0;
  // Construct a network address from a legacy struct sockaddr.
127 128
};

129
struct OneWayPipe {
Kenton Varda's avatar
Kenton Varda committed
130
  // A data pipe with an input end and an output end.  (Typically backed by pipe() system call.)
131

132 133 134 135 136
  Own<AsyncInputStream> in;
  Own<AsyncOutputStream> out;
};

struct TwoWayPipe {
137
  // A data pipe that supports sending in both directions.  Each end's output sends data to the
Kenton Varda's avatar
Kenton Varda committed
138
  // other end's input.  (Typically backed by socketpair() system call.)
139

140 141
  Own<AsyncIoStream> ends[2];
};
142

143 144 145 146 147 148
class AsyncIoProvider {
  // Class which constructs asynchronous wrappers around the operating system's I/O facilities.
  //
  // Generally, the implementation of this interface must integrate closely with a particular
  // `EventLoop` implementation.  Typically, the EventLoop implementation itself will provide
  // an AsyncIoProvider.
149 150

public:
151 152 153
  virtual OneWayPipe newOneWayPipe() = 0;
  // Creates an input/output stream pair representing the ends of a one-way pipe (e.g. created with
  // the pipe(2) system call).
154

155 156 157
  virtual TwoWayPipe newTwoWayPipe() = 0;
  // Creates two AsyncIoStreams representing the two ends of a two-way pipe (e.g. created with
  // socketpair(2) system call).  Data written to one end can be read from the other.
158

159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175
  virtual Network& getNetwork() = 0;
  // Creates a new `Network` instance representing the networks exposed by the operating system.
  //
  // DO NOT CALL THIS except at the highest levels of your code, ideally in the main() function.  If
  // you call this from low-level code, then you are preventing higher-level code from injecting an
  // alternative implementation.  Instead, if your code needs to use network functionality, it
  // should ask for a `Network` as a constructor or method parameter, so that higher-level code can
  // chose what implementation to use.  The system network is essentially a singleton.  See:
  //     http://www.object-oriented-security.org/lets-argue/singletons
  //
  // Code that uses the system network should not make any assumptions about what kinds of
  // addresses it will parse, as this could differ across platforms.  String addresses should come
  // strictly from the user, who will know how to write them correctly for their system.
  //
  // With that said, KJ currently supports the following string address formats:
  // - IPv4: "1.2.3.4", "1.2.3.4:80"
  // - IPv6: "1234:5678::abcd", "[1234:5678::abcd]:80"
Kenton Varda's avatar
Kenton Varda committed
176 177
  // - Local IP wildcard (covers both v4 and v6):  "*", "*:80"
  // - Symbolic names:  "example.com", "example.com:80", "example.com:http", "1.2.3.4:http"
178 179
  // - Unix domain: "unix:/path/to/socket"

180 181 182 183 184 185 186 187 188 189 190 191
  struct PipeThread {
    // A combination of a thread and a two-way pipe that communicates with that thread.
    //
    // The fields are intentionally ordered so that the pipe will be destroyed (and therefore
    // disconnected) before the thread is destroyed (and therefore joined).  Thus if the thread
    // arranges to exit when it detects disconnect, destruction should be clean.

    Own<Thread> thread;
    Own<AsyncIoStream> pipe;
  };

  virtual PipeThread newPipeThread(
192
      Function<void(AsyncIoProvider&, AsyncIoStream&, WaitScope&)> startFunc) = 0;
193
  // Create a new thread and set up a two-way pipe (socketpair) which can be used to communicate
194
  // with it.  One end of the pipe is passed to the thread's start function and the other end of
195 196 197 198 199
  // the pipe is returned.  The new thread also gets its own `AsyncIoProvider` instance and will
  // already have an active `EventLoop` when `startFunc` is called.
  //
  // TODO(someday):  I'm not entirely comfortable with this interface.  It seems to be doing too
  //   much at once but I'm not sure how to cleanly break it down.
200 201

  virtual Timer& getTimer() = 0;
202 203 204 205 206 207
  // Returns a `Timer` based on real time.  Time does not pass while event handlers are running --
  // it only updates when the event loop polls for system events.  This means that calling `now()`
  // on this timer does not require a system call.
  //
  // This timer is not affected by changes to the system date.  It is unspecified whether the timer
  // continues to count while the system is suspended.
208
};
209

210 211 212 213 214 215 216 217
class LowLevelAsyncIoProvider {
  // Similar to `AsyncIoProvider`, but represents a lower-level interface that may differ on
  // different operating systems.  You should prefer to use `AsyncIoProvider` over this interface
  // whenever possible, as `AsyncIoProvider` is portable and friendlier to dependency-injection.
  //
  // On Unix, this interface can be used to import native file descriptors into the async framework.
  // Different implementations of this interface might work on top of different event handling
  // primitives, such as poll vs. epoll vs. kqueue vs. some higher-level event library.
218
  //
219 220 221 222 223
  // On Windows, this interface can be used to import native HANDLEs into the async framework.
  // Different implementations of this interface might work on top of different event handling
  // primitives, such as I/O completion ports vs. completion routines.
  //
  // TODO(port):  Actually implement Windows support.
224

225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255
public:
  // ---------------------------------------------------------------------------
  // Unix-specific stuff

  enum Flags {
    // Flags controlling how to wrap a file descriptor.

    TAKE_OWNERSHIP = 1 << 0,
    // The returned object should own the file descriptor, automatically closing it when destroyed.
    // The close-on-exec flag will be set on the descriptor if it is not already.
    //
    // If this flag is not used, then the file descriptor is not automatically closed and the
    // close-on-exec flag is not modified.

    ALREADY_CLOEXEC = 1 << 1,
    // Indicates that the close-on-exec flag is known already to be set, so need not be set again.
    // Only relevant when combined with TAKE_OWNERSHIP.
    //
    // On Linux, all system calls which yield new file descriptors have flags or variants which
    // set the close-on-exec flag immediately.  Unfortunately, other OS's do not.

    ALREADY_NONBLOCK = 1 << 2
    // Indicates that the file descriptor is known already to be in non-blocking mode, so the flag
    // need not be set again.  Otherwise, all wrap*Fd() methods will enable non-blocking mode
    // automatically.
    //
    // On Linux, all system calls which yield new file descriptors have flags or variants which
    // enable non-blocking mode immediately.  Unfortunately, other OS's do not.
  };

  virtual Own<AsyncInputStream> wrapInputFd(int fd, uint flags = 0) = 0;
256 257
  // Create an AsyncInputStream wrapping a file descriptor.
  //
258
  // `flags` is a bitwise-OR of the values of the `Flags` enum.
259

260
  virtual Own<AsyncOutputStream> wrapOutputFd(int fd, uint flags = 0) = 0;
261 262
  // Create an AsyncOutputStream wrapping a file descriptor.
  //
263
  // `flags` is a bitwise-OR of the values of the `Flags` enum.
264

265
  virtual Own<AsyncIoStream> wrapSocketFd(int fd, uint flags = 0) = 0;
266 267
  // Create an AsyncIoStream wrapping a socket file descriptor.
  //
268 269 270 271 272 273
  // `flags` is a bitwise-OR of the values of the `Flags` enum.

  virtual Promise<Own<AsyncIoStream>> wrapConnectingSocketFd(int fd, uint flags = 0) = 0;
  // Create an AsyncIoStream wrapping a socket that is in the process of connecting.  The returned
  // promise should not resolve until connection has completed -- traditionally indicated by the
  // descriptor becoming writable.
274
  //
275
  // `flags` is a bitwise-OR of the values of the `Flags` enum.
276

277
  virtual Own<ConnectionReceiver> wrapListenSocketFd(int fd, uint flags = 0) = 0;
278 279 280
  // Create an AsyncIoStream wrapping a listen socket file descriptor.  This socket should already
  // have had `bind()` and `listen()` called on it, so it's ready for `accept()`.
  //
281
  // `flags` is a bitwise-OR of the values of the `Flags` enum.
282 283

  virtual Timer& getTimer() = 0;
284 285 286 287 288 289
  // Returns a `Timer` based on real time.  Time does not pass while event handlers are running --
  // it only updates when the event loop polls for system events.  This means that calling `now()`
  // on this timer does not require a system call.
  //
  // This timer is not affected by changes to the system date.  It is unspecified whether the timer
  // continues to count while the system is suspended.
290
};
291

292 293
Own<AsyncIoProvider> newAsyncIoProvider(LowLevelAsyncIoProvider& lowLevel);
// Make a new AsyncIoProvider wrapping a `LowLevelAsyncIoProvider`.
294

295 296 297
struct AsyncIoContext {
  Own<LowLevelAsyncIoProvider> lowLevelProvider;
  Own<AsyncIoProvider> provider;
298
  WaitScope& waitScope;
299 300 301 302

  UnixEventPort& unixEventPort;
  // TEMPORARY: Direct access to underlying UnixEventPort, mainly for waiting on signals. This
  //   field will go away at some point when we have a chance to improve these interfaces.
303
};
304

305
AsyncIoContext setupAsyncIo();
306
// Convenience method which sets up the current thread with everything it needs to do async I/O.
Kenton Varda's avatar
Kenton Varda committed
307
// The returned objects contain an `EventLoop` which is wrapping an appropriate `EventPort` for
308 309 310 311 312 313 314
// doing I/O on the host system, so everything is ready for the thread to start making async calls
// and waiting on promises.
//
// You would typically call this in your main() loop or in the start function of a thread.
// Example:
//
//     int main() {
Kenton Varda's avatar
Kenton Varda committed
315
//       auto ioContext = kj::setupAsyncIo();
316 317
//
//       // Now we can call an async function.
Kenton Varda's avatar
Kenton Varda committed
318
//       Promise<String> textPromise = getHttp(*ioContext.provider, "http://example.com");
319 320 321 322 323 324 325
//
//       // And we can wait for the promise to complete.  Note that you can only use `wait()`
//       // from the top level, not from inside a promise callback.
//       String text = textPromise.wait();
//       print(text);
//       return 0;
//     }
326

327 328 329
}  // namespace kj

#endif  // KJ_ASYNC_IO_H_