annotate osx/include/kj/async.h @ 68:85d5306e114e

Remove subrepo - trying to avoid these now
author Chris Cannam
date Tue, 04 Dec 2018 10:27:12 +0000
parents 0994c39f1e94
children
rev   line source
cannam@62 1 // Copyright (c) 2013-2014 Sandstorm Development Group, Inc. and contributors
cannam@62 2 // Licensed under the MIT License:
cannam@62 3 //
cannam@62 4 // Permission is hereby granted, free of charge, to any person obtaining a copy
cannam@62 5 // of this software and associated documentation files (the "Software"), to deal
cannam@62 6 // in the Software without restriction, including without limitation the rights
cannam@62 7 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
cannam@62 8 // copies of the Software, and to permit persons to whom the Software is
cannam@62 9 // furnished to do so, subject to the following conditions:
cannam@62 10 //
cannam@62 11 // The above copyright notice and this permission notice shall be included in
cannam@62 12 // all copies or substantial portions of the Software.
cannam@62 13 //
cannam@62 14 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
cannam@62 15 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
cannam@62 16 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
cannam@62 17 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
cannam@62 18 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
cannam@62 19 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
cannam@62 20 // THE SOFTWARE.
cannam@62 21
cannam@62 22 #ifndef KJ_ASYNC_H_
cannam@62 23 #define KJ_ASYNC_H_
cannam@62 24
cannam@62 25 #if defined(__GNUC__) && !KJ_HEADER_WARNINGS
cannam@62 26 #pragma GCC system_header
cannam@62 27 #endif
cannam@62 28
cannam@62 29 #include "async-prelude.h"
cannam@62 30 #include "exception.h"
cannam@62 31 #include "refcount.h"
cannam@62 32
cannam@62 33 namespace kj {
cannam@62 34
cannam@62 35 class EventLoop;
cannam@62 36 class WaitScope;
cannam@62 37
cannam@62 38 template <typename T>
cannam@62 39 class Promise;
cannam@62 40 template <typename T>
cannam@62 41 class ForkedPromise;
cannam@62 42 template <typename T>
cannam@62 43 class PromiseFulfiller;
cannam@62 44 template <typename T>
cannam@62 45 struct PromiseFulfillerPair;
cannam@62 46
cannam@62 47 template <typename Func, typename T>
cannam@62 48 using PromiseForResult = Promise<_::JoinPromises<_::ReturnType<Func, T>>>;
cannam@62 49 // Evaluates to the type of Promise for the result of calling functor type Func with parameter type
cannam@62 50 // T. If T is void, then the promise is for the result of calling Func with no arguments. If
cannam@62 51 // Func itself returns a promise, the promises are joined, so you never get Promise<Promise<T>>.
cannam@62 52
cannam@62 53 // =======================================================================================
cannam@62 54 // Promises
cannam@62 55
cannam@62 56 template <typename T>
cannam@62 57 class Promise: protected _::PromiseBase {
cannam@62 58 // The basic primitive of asynchronous computation in KJ. Similar to "futures", but designed
cannam@62 59 // specifically for event loop concurrency. Similar to E promises and JavaScript Promises/A.
cannam@62 60 //
cannam@62 61 // A Promise represents a promise to produce a value of type T some time in the future. Once
cannam@62 62 // that value has been produced, the promise is "fulfilled". Alternatively, a promise can be
cannam@62 63 // "broken", with an Exception describing what went wrong. You may implicitly convert a value of
cannam@62 64 // type T to an already-fulfilled Promise<T>. You may implicitly convert the constant
cannam@62 65 // `kj::READY_NOW` to an already-fulfilled Promise<void>. You may also implicitly convert a
cannam@62 66 // `kj::Exception` to an already-broken promise of any type.
cannam@62 67 //
cannam@62 68 // Promises are linear types -- they are moveable but not copyable. If a Promise is destroyed
cannam@62 69 // or goes out of scope (without being moved elsewhere), any ongoing asynchronous operations
cannam@62 70 // meant to fulfill the promise will be canceled if possible. All methods of `Promise` (unless
cannam@62 71 // otherwise noted) actually consume the promise in the sense of move semantics. (Arguably they
cannam@62 72 // should be rvalue-qualified, but at the time this interface was created compilers didn't widely
cannam@62 73 // support that yet and anyway it would be pretty ugly typing kj::mv(promise).whatever().) If
cannam@62 74 // you want to use one Promise in two different places, you must fork it with `fork()`.
cannam@62 75 //
cannam@62 76 // To use the result of a Promise, you must call `then()` and supply a callback function to
cannam@62 77 // call with the result. `then()` returns another promise, for the result of the callback.
cannam@62 78 // Any time that this would result in Promise<Promise<T>>, the promises are collapsed into a
cannam@62 79 // simple Promise<T> that first waits for the outer promise, then the inner. Example:
cannam@62 80 //
cannam@62 81 // // Open a remote file, read the content, and then count the
cannam@62 82 // // number of lines of text.
cannam@62 83 // // Note that none of the calls here block. `file`, `content`
cannam@62 84 // // and `lineCount` are all initialized immediately before any
cannam@62 85 // // asynchronous operations occur. The lambda callbacks are
cannam@62 86 // // called later.
cannam@62 87 // Promise<Own<File>> file = openFtp("ftp://host/foo/bar");
cannam@62 88 // Promise<String> content = file.then(
cannam@62 89 // [](Own<File> file) -> Promise<String> {
cannam@62 90 // return file.readAll();
cannam@62 91 // });
cannam@62 92 // Promise<int> lineCount = content.then(
cannam@62 93 // [](String text) -> int {
cannam@62 94 // uint count = 0;
cannam@62 95 // for (char c: text) count += (c == '\n');
cannam@62 96 // return count;
cannam@62 97 // });
cannam@62 98 //
cannam@62 99 // For `then()` to work, the current thread must have an active `EventLoop`. Each callback
cannam@62 100 // is scheduled to execute in that loop. Since `then()` schedules callbacks only on the current
cannam@62 101 // thread's event loop, you do not need to worry about two callbacks running at the same time.
cannam@62 102 // You will need to set up at least one `EventLoop` at the top level of your program before you
cannam@62 103 // can use promises.
cannam@62 104 //
cannam@62 105 // To adapt a non-Promise-based asynchronous API to promises, use `newAdaptedPromise()`.
cannam@62 106 //
cannam@62 107 // Systems using promises should consider supporting the concept of "pipelining". Pipelining
cannam@62 108 // means allowing a caller to start issuing method calls against a promised object before the
cannam@62 109 // promise has actually been fulfilled. This is particularly useful if the promise is for a
cannam@62 110 // remote object living across a network, as this can avoid round trips when chaining a series
cannam@62 111 // of calls. It is suggested that any class T which supports pipelining implement a subclass of
cannam@62 112 // Promise<T> which adds "eventual send" methods -- methods which, when called, say "please
cannam@62 113 // invoke the corresponding method on the promised value once it is available". These methods
cannam@62 114 // should in turn return promises for the eventual results of said invocations. Cap'n Proto,
cannam@62 115 // for example, implements the type `RemotePromise` which supports pipelining RPC requests -- see
cannam@62 116 // `capnp/capability.h`.
cannam@62 117 //
cannam@62 118 // KJ Promises are based on E promises:
cannam@62 119 // http://wiki.erights.org/wiki/Walnut/Distributed_Computing#Promises
cannam@62 120 //
cannam@62 121 // KJ Promises are also inspired in part by the evolving standards for JavaScript/ECMAScript
cannam@62 122 // promises, which are themselves influenced by E promises:
cannam@62 123 // http://promisesaplus.com/
cannam@62 124 // https://github.com/domenic/promises-unwrapping
cannam@62 125
cannam@62 126 public:
cannam@62 127 Promise(_::FixVoid<T> value);
cannam@62 128 // Construct an already-fulfilled Promise from a value of type T. For non-void promises, the
cannam@62 129 // parameter type is simply T. So, e.g., in a function that returns `Promise<int>`, you can
cannam@62 130 // say `return 123;` to return a promise that is already fulfilled to 123.
cannam@62 131 //
cannam@62 132 // For void promises, use `kj::READY_NOW` as the value, e.g. `return kj::READY_NOW`.
cannam@62 133
cannam@62 134 Promise(kj::Exception&& e);
cannam@62 135 // Construct an already-broken Promise.
cannam@62 136
cannam@62 137 inline Promise(decltype(nullptr)) {}
cannam@62 138
cannam@62 139 template <typename Func, typename ErrorFunc = _::PropagateException>
cannam@62 140 PromiseForResult<Func, T> then(Func&& func, ErrorFunc&& errorHandler = _::PropagateException())
cannam@62 141 KJ_WARN_UNUSED_RESULT;
cannam@62 142 // Register a continuation function to be executed when the promise completes. The continuation
cannam@62 143 // (`func`) takes the promised value (an rvalue of type `T`) as its parameter. The continuation
cannam@62 144 // may return a new value; `then()` itself returns a promise for the continuation's eventual
cannam@62 145 // result. If the continuation itself returns a `Promise<U>`, then `then()` shall also return
cannam@62 146 // a `Promise<U>` which first waits for the original promise, then executes the continuation,
cannam@62 147 // then waits for the inner promise (i.e. it automatically "unwraps" the promise).
cannam@62 148 //
cannam@62 149 // In all cases, `then()` returns immediately. The continuation is executed later. The
cannam@62 150 // continuation is always executed on the same EventLoop (and, therefore, the same thread) which
cannam@62 151 // called `then()`, therefore no synchronization is necessary on state shared by the continuation
cannam@62 152 // and the surrounding scope. If no EventLoop is running on the current thread, `then()` throws
cannam@62 153 // an exception.
cannam@62 154 //
cannam@62 155 // You may also specify an error handler continuation as the second parameter. `errorHandler`
cannam@62 156 // must be a functor taking a parameter of type `kj::Exception&&`. It must return the same
cannam@62 157 // type as `func` returns (except when `func` returns `Promise<U>`, in which case `errorHandler`
cannam@62 158 // may return either `Promise<U>` or just `U`). The default error handler simply propagates the
cannam@62 159 // exception to the returned promise.
cannam@62 160 //
cannam@62 161 // Either `func` or `errorHandler` may, of course, throw an exception, in which case the promise
cannam@62 162 // is broken. When compiled with -fno-exceptions, the framework will still detect when a
cannam@62 163 // recoverable exception was thrown inside of a continuation and will consider the promise
cannam@62 164 // broken even though a (presumably garbage) result was returned.
cannam@62 165 //
cannam@62 166 // If the returned promise is destroyed before the callback runs, the callback will be canceled
cannam@62 167 // (it will never run).
cannam@62 168 //
cannam@62 169 // Note that `then()` -- like all other Promise methods -- consumes the promise on which it is
cannam@62 170 // called, in the sense of move semantics. After returning, the original promise is no longer
cannam@62 171 // valid, but `then()` returns a new promise.
cannam@62 172 //
cannam@62 173 // *Advanced implementation tips:* Most users will never need to worry about the below, but
cannam@62 174 // it is good to be aware of.
cannam@62 175 //
cannam@62 176 // As an optimization, if the callback function `func` does _not_ return another promise, then
cannam@62 177 // execution of `func` itself may be delayed until its result is known to be needed. The
cannam@62 178 // expectation here is that `func` is just doing some transformation on the results, not
cannam@62 179 // scheduling any other actions, therefore the system doesn't need to be proactive about
cannam@62 180 // evaluating it. This way, a chain of trivial then() transformations can be executed all at
cannam@62 181 // once without repeatedly re-scheduling through the event loop. Use the `eagerlyEvaluate()`
cannam@62 182 // method to suppress this behavior.
cannam@62 183 //
cannam@62 184 // On the other hand, if `func` _does_ return another promise, then the system evaluates `func`
cannam@62 185 // as soon as possible, because the promise it returns might be for a newly-scheduled
cannam@62 186 // long-running asynchronous task.
cannam@62 187 //
cannam@62 188 // As another optimization, when a callback function registered with `then()` is actually
cannam@62 189 // scheduled, it is scheduled to occur immediately, preempting other work in the event queue.
cannam@62 190 // This allows a long chain of `then`s to execute all at once, improving cache locality by
cannam@62 191 // clustering operations on the same data. However, this implies that starvation can occur
cannam@62 192 // if a chain of `then()`s takes a very long time to execute without ever stopping to wait for
cannam@62 193 // actual I/O. To solve this, use `kj::evalLater()` to yield control; this way, all other events
cannam@62 194 // in the queue will get a chance to run before your callback is executed.
cannam@62 195
cannam@62 196 Promise<void> ignoreResult() KJ_WARN_UNUSED_RESULT { return then([](T&&) {}); }
cannam@62 197 // Convenience method to convert the promise to a void promise by ignoring the return value.
cannam@62 198 //
cannam@62 199 // You must still wait on the returned promise if you want the task to execute.
cannam@62 200
cannam@62 201 template <typename ErrorFunc>
cannam@62 202 Promise<T> catch_(ErrorFunc&& errorHandler) KJ_WARN_UNUSED_RESULT;
cannam@62 203 // Equivalent to `.then(identityFunc, errorHandler)`, where `identifyFunc` is a function that
cannam@62 204 // just returns its input.
cannam@62 205
cannam@62 206 T wait(WaitScope& waitScope);
cannam@62 207 // Run the event loop until the promise is fulfilled, then return its result. If the promise
cannam@62 208 // is rejected, throw an exception.
cannam@62 209 //
cannam@62 210 // wait() is primarily useful at the top level of a program -- typically, within the function
cannam@62 211 // that allocated the EventLoop. For example, a program that performs one or two RPCs and then
cannam@62 212 // exits would likely use wait() in its main() function to wait on each RPC. On the other hand,
cannam@62 213 // server-side code generally cannot use wait(), because it has to be able to accept multiple
cannam@62 214 // requests at once.
cannam@62 215 //
cannam@62 216 // If the promise is rejected, `wait()` throws an exception. If the program was compiled without
cannam@62 217 // exceptions (-fno-exceptions), this will usually abort. In this case you really should first
cannam@62 218 // use `then()` to set an appropriate handler for the exception case, so that the promise you
cannam@62 219 // actually wait on never throws.
cannam@62 220 //
cannam@62 221 // `waitScope` is an object proving that the caller is in a scope where wait() is allowed. By
cannam@62 222 // convention, any function which might call wait(), or which might call another function which
cannam@62 223 // might call wait(), must take `WaitScope&` as one of its parameters. This is needed for two
cannam@62 224 // reasons:
cannam@62 225 // * `wait()` is not allowed during an event callback, because event callbacks are themselves
cannam@62 226 // called during some other `wait()`, and such recursive `wait()`s would only be able to
cannam@62 227 // complete in LIFO order, which might mean that the outer `wait()` ends up waiting longer
cannam@62 228 // than it is supposed to. To prevent this, a `WaitScope` cannot be constructed or used during
cannam@62 229 // an event callback.
cannam@62 230 // * Since `wait()` runs the event loop, unrelated event callbacks may execute before `wait()`
cannam@62 231 // returns. This means that anyone calling `wait()` must be reentrant -- state may change
cannam@62 232 // around them in arbitrary ways. Therefore, callers really need to know if a function they
cannam@62 233 // are calling might wait(), and the `WaitScope&` parameter makes this clear.
cannam@62 234 //
cannam@62 235 // TODO(someday): Implement fibers, and let them call wait() even when they are handling an
cannam@62 236 // event.
cannam@62 237
cannam@62 238 ForkedPromise<T> fork() KJ_WARN_UNUSED_RESULT;
cannam@62 239 // Forks the promise, so that multiple different clients can independently wait on the result.
cannam@62 240 // `T` must be copy-constructable for this to work. Or, in the special case where `T` is
cannam@62 241 // `Own<U>`, `U` must have a method `Own<U> addRef()` which returns a new reference to the same
cannam@62 242 // (or an equivalent) object (probably implemented via reference counting).
cannam@62 243
cannam@62 244 _::SplitTuplePromise<T> split();
cannam@62 245 // Split a promise for a tuple into a tuple of promises.
cannam@62 246 //
cannam@62 247 // E.g. if you have `Promise<kj::Tuple<T, U>>`, `split()` returns
cannam@62 248 // `kj::Tuple<Promise<T>, Promise<U>>`.
cannam@62 249
cannam@62 250 Promise<T> exclusiveJoin(Promise<T>&& other) KJ_WARN_UNUSED_RESULT;
cannam@62 251 // Return a new promise that resolves when either the original promise resolves or `other`
cannam@62 252 // resolves (whichever comes first). The promise that didn't resolve first is canceled.
cannam@62 253
cannam@62 254 // TODO(someday): inclusiveJoin(), or perhaps just join(), which waits for both completions
cannam@62 255 // and produces a tuple?
cannam@62 256
cannam@62 257 template <typename... Attachments>
cannam@62 258 Promise<T> attach(Attachments&&... attachments) KJ_WARN_UNUSED_RESULT;
cannam@62 259 // "Attaches" one or more movable objects (often, Own<T>s) to the promise, such that they will
cannam@62 260 // be destroyed when the promise resolves. This is useful when a promise's callback contains
cannam@62 261 // pointers into some object and you want to make sure the object still exists when the callback
cannam@62 262 // runs -- after calling then(), use attach() to add necessary objects to the result.
cannam@62 263
cannam@62 264 template <typename ErrorFunc>
cannam@62 265 Promise<T> eagerlyEvaluate(ErrorFunc&& errorHandler) KJ_WARN_UNUSED_RESULT;
cannam@62 266 Promise<T> eagerlyEvaluate(decltype(nullptr)) KJ_WARN_UNUSED_RESULT;
cannam@62 267 // Force eager evaluation of this promise. Use this if you are going to hold on to the promise
cannam@62 268 // for awhile without consuming the result, but you want to make sure that the system actually
cannam@62 269 // processes it.
cannam@62 270 //
cannam@62 271 // `errorHandler` is a function that takes `kj::Exception&&`, like the second parameter to
cannam@62 272 // `then()`, except that it must return void. We make you specify this because otherwise it's
cannam@62 273 // easy to forget to handle errors in a promise that you never use. You may specify nullptr for
cannam@62 274 // the error handler if you are sure that ignoring errors is fine, or if you know that you'll
cannam@62 275 // eventually wait on the promise somewhere.
cannam@62 276
cannam@62 277 template <typename ErrorFunc>
cannam@62 278 void detach(ErrorFunc&& errorHandler);
cannam@62 279 // Allows the promise to continue running in the background until it completes or the
cannam@62 280 // `EventLoop` is destroyed. Be careful when using this: since you can no longer cancel this
cannam@62 281 // promise, you need to make sure that the promise owns all the objects it touches or make sure
cannam@62 282 // those objects outlive the EventLoop.
cannam@62 283 //
cannam@62 284 // `errorHandler` is a function that takes `kj::Exception&&`, like the second parameter to
cannam@62 285 // `then()`, except that it must return void.
cannam@62 286 //
cannam@62 287 // This function exists mainly to implement the Cap'n Proto requirement that RPC calls cannot be
cannam@62 288 // canceled unless the callee explicitly permits it.
cannam@62 289
cannam@62 290 kj::String trace();
cannam@62 291 // Returns a dump of debug info about this promise. Not for production use. Requires RTTI.
cannam@62 292 // This method does NOT consume the promise as other methods do.
cannam@62 293
cannam@62 294 private:
cannam@62 295 Promise(bool, Own<_::PromiseNode>&& node): PromiseBase(kj::mv(node)) {}
cannam@62 296 // Second parameter prevent ambiguity with immediate-value constructor.
cannam@62 297
cannam@62 298 template <typename>
cannam@62 299 friend class Promise;
cannam@62 300 friend class EventLoop;
cannam@62 301 template <typename U, typename Adapter, typename... Params>
cannam@62 302 friend Promise<U> newAdaptedPromise(Params&&... adapterConstructorParams);
cannam@62 303 template <typename U>
cannam@62 304 friend PromiseFulfillerPair<U> newPromiseAndFulfiller();
cannam@62 305 template <typename>
cannam@62 306 friend class _::ForkHub;
cannam@62 307 friend class _::TaskSetImpl;
cannam@62 308 friend Promise<void> _::yield();
cannam@62 309 friend class _::NeverDone;
cannam@62 310 template <typename U>
cannam@62 311 friend Promise<Array<U>> joinPromises(Array<Promise<U>>&& promises);
cannam@62 312 friend Promise<void> joinPromises(Array<Promise<void>>&& promises);
cannam@62 313 };
cannam@62 314
cannam@62 315 template <typename T>
cannam@62 316 class ForkedPromise {
cannam@62 317 // The result of `Promise::fork()` and `EventLoop::fork()`. Allows branches to be created.
cannam@62 318 // Like `Promise<T>`, this is a pass-by-move type.
cannam@62 319
cannam@62 320 public:
cannam@62 321 inline ForkedPromise(decltype(nullptr)) {}
cannam@62 322
cannam@62 323 Promise<T> addBranch();
cannam@62 324 // Add a new branch to the fork. The branch is equivalent to the original promise.
cannam@62 325
cannam@62 326 private:
cannam@62 327 Own<_::ForkHub<_::FixVoid<T>>> hub;
cannam@62 328
cannam@62 329 inline ForkedPromise(bool, Own<_::ForkHub<_::FixVoid<T>>>&& hub): hub(kj::mv(hub)) {}
cannam@62 330
cannam@62 331 friend class Promise<T>;
cannam@62 332 friend class EventLoop;
cannam@62 333 };
cannam@62 334
cannam@62 335 constexpr _::Void READY_NOW = _::Void();
cannam@62 336 // Use this when you need a Promise<void> that is already fulfilled -- this value can be implicitly
cannam@62 337 // cast to `Promise<void>`.
cannam@62 338
cannam@62 339 constexpr _::NeverDone NEVER_DONE = _::NeverDone();
cannam@62 340 // The opposite of `READY_NOW`, return this when the promise should never resolve. This can be
cannam@62 341 // implicitly converted to any promise type. You may also call `NEVER_DONE.wait()` to wait
cannam@62 342 // forever (useful for servers).
cannam@62 343
cannam@62 344 template <typename Func>
cannam@62 345 PromiseForResult<Func, void> evalLater(Func&& func) KJ_WARN_UNUSED_RESULT;
cannam@62 346 // Schedule for the given zero-parameter function to be executed in the event loop at some
cannam@62 347 // point in the near future. Returns a Promise for its result -- or, if `func()` itself returns
cannam@62 348 // a promise, `evalLater()` returns a Promise for the result of resolving that promise.
cannam@62 349 //
cannam@62 350 // Example usage:
cannam@62 351 // Promise<int> x = evalLater([]() { return 123; });
cannam@62 352 //
cannam@62 353 // The above is exactly equivalent to:
cannam@62 354 // Promise<int> x = Promise<void>(READY_NOW).then([]() { return 123; });
cannam@62 355 //
cannam@62 356 // If the returned promise is destroyed before the callback runs, the callback will be canceled
cannam@62 357 // (never called).
cannam@62 358 //
cannam@62 359 // If you schedule several evaluations with `evalLater` during the same callback, they are
cannam@62 360 // guaranteed to be executed in order.
cannam@62 361
cannam@62 362 template <typename Func>
cannam@62 363 PromiseForResult<Func, void> evalNow(Func&& func) KJ_WARN_UNUSED_RESULT;
cannam@62 364 // Run `func()` and return a promise for its result. `func()` executes before `evalNow()` returns.
cannam@62 365 // If `func()` throws an exception, the exception is caught and wrapped in a promise -- this is the
cannam@62 366 // main reason why `evalNow()` is useful.
cannam@62 367
cannam@62 368 template <typename T>
cannam@62 369 Promise<Array<T>> joinPromises(Array<Promise<T>>&& promises);
cannam@62 370 // Join an array of promises into a promise for an array.
cannam@62 371
cannam@62 372 // =======================================================================================
cannam@62 373 // Hack for creating a lambda that holds an owned pointer.
cannam@62 374
cannam@62 375 template <typename Func, typename MovedParam>
cannam@62 376 class CaptureByMove {
cannam@62 377 public:
cannam@62 378 inline CaptureByMove(Func&& func, MovedParam&& param)
cannam@62 379 : func(kj::mv(func)), param(kj::mv(param)) {}
cannam@62 380
cannam@62 381 template <typename... Params>
cannam@62 382 inline auto operator()(Params&&... params)
cannam@62 383 -> decltype(kj::instance<Func>()(kj::instance<MovedParam&&>(), kj::fwd<Params>(params)...)) {
cannam@62 384 return func(kj::mv(param), kj::fwd<Params>(params)...);
cannam@62 385 }
cannam@62 386
cannam@62 387 private:
cannam@62 388 Func func;
cannam@62 389 MovedParam param;
cannam@62 390 };
cannam@62 391
cannam@62 392 template <typename Func, typename MovedParam>
cannam@62 393 inline CaptureByMove<Func, Decay<MovedParam>> mvCapture(MovedParam&& param, Func&& func) {
cannam@62 394 // Hack to create a "lambda" which captures a variable by moving it rather than copying or
cannam@62 395 // referencing. C++14 generalized captures should make this obsolete, but for now in C++11 this
cannam@62 396 // is commonly needed for Promise continuations that own their state. Example usage:
cannam@62 397 //
cannam@62 398 // Own<Foo> ptr = makeFoo();
cannam@62 399 // Promise<int> promise = callRpc();
cannam@62 400 // promise.then(mvCapture(ptr, [](Own<Foo>&& ptr, int result) {
cannam@62 401 // return ptr->finish(result);
cannam@62 402 // }));
cannam@62 403
cannam@62 404 return CaptureByMove<Func, Decay<MovedParam>>(kj::fwd<Func>(func), kj::mv(param));
cannam@62 405 }
cannam@62 406
cannam@62 407 // =======================================================================================
cannam@62 408 // Advanced promise construction
cannam@62 409
cannam@62 410 template <typename T>
cannam@62 411 class PromiseFulfiller {
cannam@62 412 // A callback which can be used to fulfill a promise. Only the first call to fulfill() or
cannam@62 413 // reject() matters; subsequent calls are ignored.
cannam@62 414
cannam@62 415 public:
cannam@62 416 virtual void fulfill(T&& value) = 0;
cannam@62 417 // Fulfill the promise with the given value.
cannam@62 418
cannam@62 419 virtual void reject(Exception&& exception) = 0;
cannam@62 420 // Reject the promise with an error.
cannam@62 421
cannam@62 422 virtual bool isWaiting() = 0;
cannam@62 423 // Returns true if the promise is still unfulfilled and someone is potentially waiting for it.
cannam@62 424 // Returns false if fulfill()/reject() has already been called *or* if the promise to be
cannam@62 425 // fulfilled has been discarded and therefore the result will never be used anyway.
cannam@62 426
cannam@62 427 template <typename Func>
cannam@62 428 bool rejectIfThrows(Func&& func);
cannam@62 429 // Call the function (with no arguments) and return true. If an exception is thrown, call
cannam@62 430 // `fulfiller.reject()` and then return false. When compiled with exceptions disabled,
cannam@62 431 // non-fatal exceptions are still detected and handled correctly.
cannam@62 432 };
cannam@62 433
cannam@62 434 template <>
cannam@62 435 class PromiseFulfiller<void> {
cannam@62 436 // Specialization of PromiseFulfiller for void promises. See PromiseFulfiller<T>.
cannam@62 437
cannam@62 438 public:
cannam@62 439 virtual void fulfill(_::Void&& value = _::Void()) = 0;
cannam@62 440 // Call with zero parameters. The parameter is a dummy that only exists so that subclasses don't
cannam@62 441 // have to specialize for <void>.
cannam@62 442
cannam@62 443 virtual void reject(Exception&& exception) = 0;
cannam@62 444 virtual bool isWaiting() = 0;
cannam@62 445
cannam@62 446 template <typename Func>
cannam@62 447 bool rejectIfThrows(Func&& func);
cannam@62 448 };
cannam@62 449
cannam@62 450 template <typename T, typename Adapter, typename... Params>
cannam@62 451 Promise<T> newAdaptedPromise(Params&&... adapterConstructorParams);
cannam@62 452 // Creates a new promise which owns an instance of `Adapter` which encapsulates the operation
cannam@62 453 // that will eventually fulfill the promise. This is primarily useful for adapting non-KJ
cannam@62 454 // asynchronous APIs to use promises.
cannam@62 455 //
cannam@62 456 // An instance of `Adapter` will be allocated and owned by the returned `Promise`. A
cannam@62 457 // `PromiseFulfiller<T>&` will be passed as the first parameter to the adapter's constructor,
cannam@62 458 // and `adapterConstructorParams` will be forwarded as the subsequent parameters. The adapter
cannam@62 459 // is expected to perform some asynchronous operation and call the `PromiseFulfiller<T>` once
cannam@62 460 // it is finished.
cannam@62 461 //
cannam@62 462 // The adapter is destroyed when its owning Promise is destroyed. This may occur before the
cannam@62 463 // Promise has been fulfilled. In this case, the adapter's destructor should cancel the
cannam@62 464 // asynchronous operation. Once the adapter is destroyed, the fulfillment callback cannot be
cannam@62 465 // called.
cannam@62 466 //
cannam@62 467 // An adapter implementation should be carefully written to ensure that it cannot accidentally
cannam@62 468 // be left unfulfilled permanently because of an exception. Consider making liberal use of
cannam@62 469 // `PromiseFulfiller<T>::rejectIfThrows()`.
cannam@62 470
cannam@62 471 template <typename T>
cannam@62 472 struct PromiseFulfillerPair {
cannam@62 473 Promise<_::JoinPromises<T>> promise;
cannam@62 474 Own<PromiseFulfiller<T>> fulfiller;
cannam@62 475 };
cannam@62 476
cannam@62 477 template <typename T>
cannam@62 478 PromiseFulfillerPair<T> newPromiseAndFulfiller();
cannam@62 479 // Construct a Promise and a separate PromiseFulfiller which can be used to fulfill the promise.
cannam@62 480 // If the PromiseFulfiller is destroyed before either of its methods are called, the Promise is
cannam@62 481 // implicitly rejected.
cannam@62 482 //
cannam@62 483 // Although this function is easier to use than `newAdaptedPromise()`, it has the serious drawback
cannam@62 484 // that there is no way to handle cancellation (i.e. detect when the Promise is discarded).
cannam@62 485 //
cannam@62 486 // You can arrange to fulfill a promise with another promise by using a promise type for T. E.g.
cannam@62 487 // `newPromiseAndFulfiller<Promise<U>>()` will produce a promise of type `Promise<U>` but the
cannam@62 488 // fulfiller will be of type `PromiseFulfiller<Promise<U>>`. Thus you pass a `Promise<U>` to the
cannam@62 489 // `fulfill()` callback, and the promises are chained.
cannam@62 490
cannam@62 491 // =======================================================================================
cannam@62 492 // TaskSet
cannam@62 493
cannam@62 494 class TaskSet {
cannam@62 495 // Holds a collection of Promise<void>s and ensures that each executes to completion. Memory
cannam@62 496 // associated with each promise is automatically freed when the promise completes. Destroying
cannam@62 497 // the TaskSet itself automatically cancels all unfinished promises.
cannam@62 498 //
cannam@62 499 // This is useful for "daemon" objects that perform background tasks which aren't intended to
cannam@62 500 // fulfill any particular external promise, but which may need to be canceled (and thus can't
cannam@62 501 // use `Promise::detach()`). The daemon object holds a TaskSet to collect these tasks it is
cannam@62 502 // working on. This way, if the daemon itself is destroyed, the TaskSet is detroyed as well,
cannam@62 503 // and everything the daemon is doing is canceled.
cannam@62 504
cannam@62 505 public:
cannam@62 506 class ErrorHandler {
cannam@62 507 public:
cannam@62 508 virtual void taskFailed(kj::Exception&& exception) = 0;
cannam@62 509 };
cannam@62 510
cannam@62 511 TaskSet(ErrorHandler& errorHandler);
cannam@62 512 // `loop` will be used to wait on promises. `errorHandler` will be executed any time a task
cannam@62 513 // throws an exception, and will execute within the given EventLoop.
cannam@62 514
cannam@62 515 ~TaskSet() noexcept(false);
cannam@62 516
cannam@62 517 void add(Promise<void>&& promise);
cannam@62 518
cannam@62 519 kj::String trace();
cannam@62 520 // Return debug info about all promises currently in the TaskSet.
cannam@62 521
cannam@62 522 private:
cannam@62 523 Own<_::TaskSetImpl> impl;
cannam@62 524 };
cannam@62 525
cannam@62 526 // =======================================================================================
cannam@62 527 // The EventLoop class
cannam@62 528
cannam@62 529 class EventPort {
cannam@62 530 // Interfaces between an `EventLoop` and events originating from outside of the loop's thread.
cannam@62 531 // All such events come in through the `EventPort` implementation.
cannam@62 532 //
cannam@62 533 // An `EventPort` implementation may interface with low-level operating system APIs and/or other
cannam@62 534 // threads. You can also write an `EventPort` which wraps some other (non-KJ) event loop
cannam@62 535 // framework, allowing the two to coexist in a single thread.
cannam@62 536
cannam@62 537 public:
cannam@62 538 virtual bool wait() = 0;
cannam@62 539 // Wait for an external event to arrive, sleeping if necessary. Once at least one event has
cannam@62 540 // arrived, queue it to the event loop (e.g. by fulfilling a promise) and return.
cannam@62 541 //
cannam@62 542 // This is called during `Promise::wait()` whenever the event queue becomes empty, in order to
cannam@62 543 // wait for new events to populate the queue.
cannam@62 544 //
cannam@62 545 // It is safe to return even if nothing has actually been queued, so long as calling `wait()` in
cannam@62 546 // a loop will eventually sleep. (That is to say, false positives are fine.)
cannam@62 547 //
cannam@62 548 // Returns true if wake() has been called from another thread. (Precisely, returns true if
cannam@62 549 // no previous call to wait `wait()` nor `poll()` has returned true since `wake()` was last
cannam@62 550 // called.)
cannam@62 551
cannam@62 552 virtual bool poll() = 0;
cannam@62 553 // Check if any external events have arrived, but do not sleep. If any events have arrived,
cannam@62 554 // add them to the event queue (e.g. by fulfilling promises) before returning.
cannam@62 555 //
cannam@62 556 // This may be called during `Promise::wait()` when the EventLoop has been executing for a while
cannam@62 557 // without a break but is still non-empty.
cannam@62 558 //
cannam@62 559 // Returns true if wake() has been called from another thread. (Precisely, returns true if
cannam@62 560 // no previous call to wait `wait()` nor `poll()` has returned true since `wake()` was last
cannam@62 561 // called.)
cannam@62 562
cannam@62 563 virtual void setRunnable(bool runnable);
cannam@62 564 // Called to notify the `EventPort` when the `EventLoop` has work to do; specifically when it
cannam@62 565 // transitions from empty -> runnable or runnable -> empty. This is typically useful when
cannam@62 566 // integrating with an external event loop; if the loop is currently runnable then you should
cannam@62 567 // arrange to call run() on it soon. The default implementation does nothing.
cannam@62 568
cannam@62 569 virtual void wake() const;
cannam@62 570 // Wake up the EventPort's thread from another thread.
cannam@62 571 //
cannam@62 572 // Unlike all other methods on this interface, `wake()` may be called from another thread, hence
cannam@62 573 // it is `const`.
cannam@62 574 //
cannam@62 575 // Technically speaking, `wake()` causes the target thread to cease sleeping and not to sleep
cannam@62 576 // again until `wait()` or `poll()` has returned true at least once.
cannam@62 577 //
cannam@62 578 // The default implementation throws an UNIMPLEMENTED exception.
cannam@62 579 };
cannam@62 580
cannam@62 581 class EventLoop {
cannam@62 582 // Represents a queue of events being executed in a loop. Most code won't interact with
cannam@62 583 // EventLoop directly, but instead use `Promise`s to interact with it indirectly. See the
cannam@62 584 // documentation for `Promise`.
cannam@62 585 //
cannam@62 586 // Each thread can have at most one current EventLoop. To make an `EventLoop` current for
cannam@62 587 // the thread, create a `WaitScope`. Async APIs require that the thread has a current EventLoop,
cannam@62 588 // or they will throw exceptions. APIs that use `Promise::wait()` additionally must explicitly
cannam@62 589 // be passed a reference to the `WaitScope` to make the caller aware that they might block.
cannam@62 590 //
cannam@62 591 // Generally, you will want to construct an `EventLoop` at the top level of your program, e.g.
cannam@62 592 // in the main() function, or in the start function of a thread. You can then use it to
cannam@62 593 // construct some promises and wait on the result. Example:
cannam@62 594 //
cannam@62 595 // int main() {
cannam@62 596 // // `loop` becomes the official EventLoop for the thread.
cannam@62 597 // MyEventPort eventPort;
cannam@62 598 // EventLoop loop(eventPort);
cannam@62 599 //
cannam@62 600 // // Now we can call an async function.
cannam@62 601 // Promise<String> textPromise = getHttp("http://example.com");
cannam@62 602 //
cannam@62 603 // // And we can wait for the promise to complete. Note that you can only use `wait()`
cannam@62 604 // // from the top level, not from inside a promise callback.
cannam@62 605 // String text = textPromise.wait();
cannam@62 606 // print(text);
cannam@62 607 // return 0;
cannam@62 608 // }
cannam@62 609 //
cannam@62 610 // Most applications that do I/O will prefer to use `setupAsyncIo()` from `async-io.h` rather
cannam@62 611 // than allocate an `EventLoop` directly.
cannam@62 612
cannam@62 613 public:
cannam@62 614 EventLoop();
cannam@62 615 // Construct an `EventLoop` which does not receive external events at all.
cannam@62 616
cannam@62 617 explicit EventLoop(EventPort& port);
cannam@62 618 // Construct an `EventLoop` which receives external events through the given `EventPort`.
cannam@62 619
cannam@62 620 ~EventLoop() noexcept(false);
cannam@62 621
cannam@62 622 void run(uint maxTurnCount = maxValue);
cannam@62 623 // Run the event loop for `maxTurnCount` turns or until there is nothing left to be done,
cannam@62 624 // whichever comes first. This never calls the `EventPort`'s `sleep()` or `poll()`. It will
cannam@62 625 // call the `EventPort`'s `setRunnable(false)` if the queue becomes empty.
cannam@62 626
cannam@62 627 bool isRunnable();
cannam@62 628 // Returns true if run() would currently do anything, or false if the queue is empty.
cannam@62 629
cannam@62 630 private:
cannam@62 631 EventPort& port;
cannam@62 632
cannam@62 633 bool running = false;
cannam@62 634 // True while looping -- wait() is then not allowed.
cannam@62 635
cannam@62 636 bool lastRunnableState = false;
cannam@62 637 // What did we last pass to port.setRunnable()?
cannam@62 638
cannam@62 639 _::Event* head = nullptr;
cannam@62 640 _::Event** tail = &head;
cannam@62 641 _::Event** depthFirstInsertPoint = &head;
cannam@62 642
cannam@62 643 Own<_::TaskSetImpl> daemons;
cannam@62 644
cannam@62 645 bool turn();
cannam@62 646 void setRunnable(bool runnable);
cannam@62 647 void enterScope();
cannam@62 648 void leaveScope();
cannam@62 649
cannam@62 650 friend void _::detach(kj::Promise<void>&& promise);
cannam@62 651 friend void _::waitImpl(Own<_::PromiseNode>&& node, _::ExceptionOrValue& result,
cannam@62 652 WaitScope& waitScope);
cannam@62 653 friend class _::Event;
cannam@62 654 friend class WaitScope;
cannam@62 655 };
cannam@62 656
cannam@62 657 class WaitScope {
cannam@62 658 // Represents a scope in which asynchronous programming can occur. A `WaitScope` should usually
cannam@62 659 // be allocated on the stack and serves two purposes:
cannam@62 660 // * While the `WaitScope` exists, its `EventLoop` is registered as the current loop for the
cannam@62 661 // thread. Most operations dealing with `Promise` (including all of its methods) do not work
cannam@62 662 // unless the thread has a current `EventLoop`.
cannam@62 663 // * `WaitScope` may be passed to `Promise::wait()` to synchronously wait for a particular
cannam@62 664 // promise to complete. See `Promise::wait()` for an extended discussion.
cannam@62 665
cannam@62 666 public:
cannam@62 667 inline explicit WaitScope(EventLoop& loop): loop(loop) { loop.enterScope(); }
cannam@62 668 inline ~WaitScope() { loop.leaveScope(); }
cannam@62 669 KJ_DISALLOW_COPY(WaitScope);
cannam@62 670
cannam@62 671 private:
cannam@62 672 EventLoop& loop;
cannam@62 673 friend class EventLoop;
cannam@62 674 friend void _::waitImpl(Own<_::PromiseNode>&& node, _::ExceptionOrValue& result,
cannam@62 675 WaitScope& waitScope);
cannam@62 676 };
cannam@62 677
cannam@62 678 } // namespace kj
cannam@62 679
cannam@62 680 #include "async-inl.h"
cannam@62 681
cannam@62 682 #endif // KJ_ASYNC_H_