annotate win64-msvc/include/kj/async.h @ 62:0994c39f1e94

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