| 278 | } |
| 279 | |
| 280 | Future<R> start() |
| 281 | { |
| 282 | auto self = shared(); |
| 283 | auto weak_self = weak(); |
| 284 | |
| 285 | // Propagating discards: |
| 286 | // |
| 287 | // When the caller does a discard we need to propagate it to |
| 288 | // either the future returned from `iterate` or the future |
| 289 | // returned from `body`. One easy way to do this would be to add |
| 290 | // an `onAny` callback for every future returned from `iterate` |
| 291 | // and `body`, but that would be a slow memory leak that would |
| 292 | // grow over time, especially if the loop was actually |
| 293 | // infinite. Instead, we capture the current future that needs to |
| 294 | // be discarded within a `discard` function that we'll invoke when |
| 295 | // we get a discard. Because there is a race setting the `discard` |
| 296 | // function and reading it out to invoke we have to synchronize |
| 297 | // access using a mutex. An alternative strategy would be to use |
| 298 | // something like `atomic_load` and `atomic_store` with |
| 299 | // `shared_ptr` so that we can swap the current future(s) |
| 300 | // atomically. |
| 301 | promise.future().onDiscard([weak_self]() { |
| 302 | auto self = weak_self.lock(); |
| 303 | if (self) { |
| 304 | // We need to make a copy of the current `discard` function so |
| 305 | // that we can invoke it outside of the `synchronized` block |
| 306 | // in the event that discarding invokes causes the `onAny` |
| 307 | // callbacks that we have added in `run` to execute which may |
| 308 | // deadlock attempting to re-acquire `mutex`! |
| 309 | std::function<void()> f = []() {}; |
| 310 | synchronized (self->mutex) { |
| 311 | f = self->discard; |
| 312 | } |
| 313 | f(); |
| 314 | } |
| 315 | }); |
| 316 | |
| 317 | if (pid.isSome()) { |
| 318 | // Start the loop using `pid` as the execution context. |
| 319 | dispatch(pid.get(), [self]() { |
| 320 | self->run(self->iterate()); |
| 321 | }); |
| 322 | |
| 323 | // TODO(benh): Link with `pid` so that we can discard or abandon |
| 324 | // the promise in the event `pid` terminates and didn't discard |
| 325 | // us so that we can avoid any leaks (memory or otherwise). |
| 326 | } else { |
| 327 | run(iterate()); |
| 328 | } |
| 329 | |
| 330 | return promise.future(); |
| 331 | } |
| 332 | |
| 333 | void run(Future<T> next) |
| 334 | { |