6bad3618b0da9dcb00a54dcc7b3d5c807cc31e5c
[folly.git] / folly / experimental / fibers / FiberManager-inl.h
1 /*
2  * Copyright 2016 Facebook, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  *   http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 #pragma once
17
18 #include <cassert>
19
20 #include <folly/CPortability.h>
21 #include <folly/Memory.h>
22 #include <folly/Optional.h>
23 #include <folly/Portability.h>
24 #include <folly/ScopeGuard.h>
25 #ifdef __APPLE__
26 #include <folly/ThreadLocal.h>
27 #endif
28 #include <folly/experimental/fibers/Baton.h>
29 #include <folly/experimental/fibers/Fiber.h>
30 #include <folly/experimental/fibers/LoopController.h>
31 #include <folly/experimental/fibers/Promise.h>
32 #include <folly/futures/Promise.h>
33 #include <folly/futures/Try.h>
34
35 namespace folly {
36 namespace fibers {
37
38 namespace {
39
40 inline FiberManager::Options preprocessOptions(FiberManager::Options opts) {
41 #ifdef FOLLY_SANITIZE_ADDRESS
42   /* ASAN needs a lot of extra stack space.
43      16x is a conservative estimate, 8x also worked with tests
44      where it mattered.  Note that overallocating here does not necessarily
45      increase RSS, since unused memory is pretty much free. */
46   opts.stackSize *= 16;
47 #endif
48   return opts;
49 }
50
51 } // anonymous
52
53 inline void FiberManager::ensureLoopScheduled() {
54   if (isLoopScheduled_) {
55     return;
56   }
57
58   isLoopScheduled_ = true;
59   loopController_->schedule();
60 }
61
62 inline intptr_t FiberManager::activateFiber(Fiber* fiber) {
63   DCHECK_EQ(activeFiber_, (Fiber*)nullptr);
64
65 #ifdef FOLLY_SANITIZE_ADDRESS
66   registerFiberActivationWithAsan(fiber);
67 #endif
68
69   activeFiber_ = fiber;
70   return jumpContext(&mainContext_, &fiber->fcontext_, fiber->data_);
71 }
72
73 inline intptr_t FiberManager::deactivateFiber(Fiber* fiber) {
74   DCHECK_EQ(activeFiber_, fiber);
75
76 #ifdef FOLLY_SANITIZE_ADDRESS
77   registerFiberDeactivationWithAsan(fiber);
78 #endif
79
80   activeFiber_ = nullptr;
81   return jumpContext(&fiber->fcontext_, &mainContext_, 0);
82 }
83
84 inline void FiberManager::runReadyFiber(Fiber* fiber) {
85   SCOPE_EXIT {
86     assert(currentFiber_ == nullptr);
87     assert(activeFiber_ == nullptr);
88   };
89
90   assert(
91       fiber->state_ == Fiber::NOT_STARTED ||
92       fiber->state_ == Fiber::READY_TO_RUN);
93   currentFiber_ = fiber;
94   fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
95   if (observer_) {
96     observer_->starting(reinterpret_cast<uintptr_t>(fiber));
97   }
98
99   while (fiber->state_ == Fiber::NOT_STARTED ||
100          fiber->state_ == Fiber::READY_TO_RUN) {
101     activateFiber(fiber);
102     if (fiber->state_ == Fiber::AWAITING_IMMEDIATE) {
103       try {
104         immediateFunc_();
105       } catch (...) {
106         exceptionCallback_(std::current_exception(), "running immediateFunc_");
107       }
108       immediateFunc_ = nullptr;
109       fiber->state_ = Fiber::READY_TO_RUN;
110     }
111   }
112
113   if (fiber->state_ == Fiber::AWAITING) {
114     awaitFunc_(*fiber);
115     awaitFunc_ = nullptr;
116     if (observer_) {
117       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
118     }
119     currentFiber_ = nullptr;
120     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
121   } else if (fiber->state_ == Fiber::INVALID) {
122     assert(fibersActive_ > 0);
123     --fibersActive_;
124     // Making sure that task functor is deleted once task is complete.
125     // NOTE: we must do it on main context, as the fiber is not
126     // running at this point.
127     fiber->func_ = nullptr;
128     fiber->resultFunc_ = nullptr;
129     if (fiber->finallyFunc_) {
130       try {
131         fiber->finallyFunc_();
132       } catch (...) {
133         exceptionCallback_(std::current_exception(), "running finallyFunc_");
134       }
135       fiber->finallyFunc_ = nullptr;
136     }
137     // Make sure LocalData is not accessible from its destructor
138     if (observer_) {
139       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
140     }
141     currentFiber_ = nullptr;
142     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
143     fiber->localData_.reset();
144     fiber->rcontext_.reset();
145
146     if (fibersPoolSize_ < options_.maxFibersPoolSize ||
147         options_.fibersPoolResizePeriodMs > 0) {
148       fibersPool_.push_front(*fiber);
149       ++fibersPoolSize_;
150     } else {
151       delete fiber;
152       assert(fibersAllocated_ > 0);
153       --fibersAllocated_;
154     }
155   } else if (fiber->state_ == Fiber::YIELDED) {
156     if (observer_) {
157       observer_->stopped(reinterpret_cast<uintptr_t>(fiber));
158     }
159     currentFiber_ = nullptr;
160     fiber->rcontext_ = RequestContext::setContext(std::move(fiber->rcontext_));
161     fiber->state_ = Fiber::READY_TO_RUN;
162     yieldedFibers_.push_back(*fiber);
163   }
164 }
165
166 inline bool FiberManager::loopUntilNoReady() {
167   // Support nested FiberManagers
168   auto originalFiberManager = this;
169   std::swap(currentFiberManager_, originalFiberManager);
170
171   SCOPE_EXIT {
172     isLoopScheduled_ = false;
173     if (!readyFibers_.empty()) {
174       ensureLoopScheduled();
175     }
176     std::swap(currentFiberManager_, originalFiberManager);
177     CHECK_EQ(this, originalFiberManager);
178   };
179
180   bool hadRemoteFiber = true;
181   while (hadRemoteFiber) {
182     hadRemoteFiber = false;
183
184     while (!readyFibers_.empty()) {
185       auto& fiber = readyFibers_.front();
186       readyFibers_.pop_front();
187       runReadyFiber(&fiber);
188     }
189
190     remoteReadyQueue_.sweep([this, &hadRemoteFiber](Fiber* fiber) {
191       runReadyFiber(fiber);
192       hadRemoteFiber = true;
193     });
194
195     remoteTaskQueue_.sweep([this, &hadRemoteFiber](RemoteTask* taskPtr) {
196       std::unique_ptr<RemoteTask> task(taskPtr);
197       auto fiber = getFiber();
198       if (task->localData) {
199         fiber->localData_ = *task->localData;
200       }
201       fiber->rcontext_ = std::move(task->rcontext);
202
203       fiber->setFunction(std::move(task->func));
204       fiber->data_ = reinterpret_cast<intptr_t>(fiber);
205       if (observer_) {
206         observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
207       }
208       runReadyFiber(fiber);
209       hadRemoteFiber = true;
210     });
211   }
212
213   if (observer_) {
214     for (auto& yielded : yieldedFibers_) {
215       observer_->runnable(reinterpret_cast<uintptr_t>(&yielded));
216     }
217   }
218   readyFibers_.splice(readyFibers_.end(), yieldedFibers_);
219
220   return fibersActive_ > 0;
221 }
222
223 // We need this to be in a struct, not inlined in addTask, because clang crashes
224 // otherwise.
225 template <typename F>
226 struct FiberManager::AddTaskHelper {
227   class Func;
228
229   static constexpr bool allocateInBuffer =
230       sizeof(Func) <= Fiber::kUserBufferSize;
231
232   class Func {
233    public:
234     Func(F&& func, FiberManager& fm) : func_(std::forward<F>(func)), fm_(fm) {}
235
236     void operator()() {
237       try {
238         func_();
239       } catch (...) {
240         fm_.exceptionCallback_(
241             std::current_exception(), "running Func functor");
242       }
243       if (allocateInBuffer) {
244         this->~Func();
245       } else {
246         delete this;
247       }
248     }
249
250    private:
251     F func_;
252     FiberManager& fm_;
253   };
254 };
255
256 template <typename F>
257 void FiberManager::addTask(F&& func) {
258   typedef AddTaskHelper<F> Helper;
259
260   auto fiber = getFiber();
261   initLocalData(*fiber);
262
263   if (Helper::allocateInBuffer) {
264     auto funcLoc = static_cast<typename Helper::Func*>(fiber->getUserBuffer());
265     new (funcLoc) typename Helper::Func(std::forward<F>(func), *this);
266
267     fiber->setFunction(std::ref(*funcLoc));
268   } else {
269     auto funcLoc = new typename Helper::Func(std::forward<F>(func), *this);
270
271     fiber->setFunction(std::ref(*funcLoc));
272   }
273
274   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
275   readyFibers_.push_back(*fiber);
276   if (observer_) {
277     observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
278   }
279
280   ensureLoopScheduled();
281 }
282
283 template <typename F>
284 auto FiberManager::addTaskFuture(F&& func)
285     -> folly::Future<typename std::result_of<F()>::type> {
286   using T = typename std::result_of<F()>::type;
287   folly::Promise<T> p;
288   auto f = p.getFuture();
289   addTaskFinally(
290       [func = std::forward<F>(func)]() mutable { return func(); },
291       [p = std::move(p)](folly::Try<T> && t) mutable {
292         p.setTry(std::move(t));
293       });
294   return f;
295 }
296
297 template <typename F>
298 void FiberManager::addTaskRemote(F&& func) {
299   auto task = [&]() {
300     auto currentFm = getFiberManagerUnsafe();
301     if (currentFm && currentFm->currentFiber_ &&
302         currentFm->localType_ == localType_) {
303       return folly::make_unique<RemoteTask>(
304           std::forward<F>(func), currentFm->currentFiber_->localData_);
305     }
306     return folly::make_unique<RemoteTask>(std::forward<F>(func));
307   }();
308   auto insertHead = [&]() {
309     return remoteTaskQueue_.insertHead(task.release());
310   };
311   loopController_->scheduleThreadSafe(std::ref(insertHead));
312 }
313
314 template <typename F>
315 auto FiberManager::addTaskRemoteFuture(F&& func)
316     -> folly::Future<typename std::result_of<F()>::type> {
317   folly::Promise<typename std::result_of<F()>::type> p;
318   auto f = p.getFuture();
319   addTaskRemote(
320       [ p = std::move(p), func = std::forward<F>(func), this ]() mutable {
321         auto t = folly::makeTryWith(std::forward<F>(func));
322         runInMainContext([&]() { p.setTry(std::move(t)); });
323       });
324   return f;
325 }
326
327 template <typename X>
328 struct IsRvalueRefTry {
329   static const bool value = false;
330 };
331 template <typename T>
332 struct IsRvalueRefTry<folly::Try<T>&&> {
333   static const bool value = true;
334 };
335
336 // We need this to be in a struct, not inlined in addTaskFinally, because clang
337 // crashes otherwise.
338 template <typename F, typename G>
339 struct FiberManager::AddTaskFinallyHelper {
340   class Func;
341
342   typedef typename std::result_of<F()>::type Result;
343
344   class Finally {
345    public:
346     Finally(G finally, FiberManager& fm)
347         : finally_(std::move(finally)), fm_(fm) {}
348
349     void operator()() {
350       try {
351         finally_(std::move(*result_));
352       } catch (...) {
353         fm_.exceptionCallback_(
354             std::current_exception(), "running Finally functor");
355       }
356
357       if (allocateInBuffer) {
358         this->~Finally();
359       } else {
360         delete this;
361       }
362     }
363
364    private:
365     friend class Func;
366
367     G finally_;
368     folly::Optional<folly::Try<Result>> result_;
369     FiberManager& fm_;
370   };
371
372   class Func {
373    public:
374     Func(F func, Finally& finally)
375         : func_(std::move(func)), result_(finally.result_) {}
376
377     void operator()() {
378       result_ = folly::makeTryWith(std::move(func_));
379
380       if (allocateInBuffer) {
381         this->~Func();
382       } else {
383         delete this;
384       }
385     }
386
387    private:
388     F func_;
389     folly::Optional<folly::Try<Result>>& result_;
390   };
391
392   static constexpr bool allocateInBuffer =
393       sizeof(Func) + sizeof(Finally) <= Fiber::kUserBufferSize;
394 };
395
396 template <typename F, typename G>
397 void FiberManager::addTaskFinally(F&& func, G&& finally) {
398   typedef typename std::result_of<F()>::type Result;
399
400   static_assert(
401       IsRvalueRefTry<typename FirstArgOf<G>::type>::value,
402       "finally(arg): arg must be Try<T>&&");
403   static_assert(
404       std::is_convertible<
405           Result,
406           typename std::remove_reference<
407               typename FirstArgOf<G>::type>::type::element_type>::value,
408       "finally(Try<T>&&): T must be convertible from func()'s return type");
409
410   auto fiber = getFiber();
411   initLocalData(*fiber);
412
413   typedef AddTaskFinallyHelper<
414       typename std::decay<F>::type,
415       typename std::decay<G>::type>
416       Helper;
417
418   if (Helper::allocateInBuffer) {
419     auto funcLoc = static_cast<typename Helper::Func*>(fiber->getUserBuffer());
420     auto finallyLoc =
421         static_cast<typename Helper::Finally*>(static_cast<void*>(funcLoc + 1));
422
423     new (finallyLoc) typename Helper::Finally(std::forward<G>(finally), *this);
424     new (funcLoc) typename Helper::Func(std::forward<F>(func), *finallyLoc);
425
426     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
427   } else {
428     auto finallyLoc =
429         new typename Helper::Finally(std::forward<G>(finally), *this);
430     auto funcLoc =
431         new typename Helper::Func(std::forward<F>(func), *finallyLoc);
432
433     fiber->setFunctionFinally(std::ref(*funcLoc), std::ref(*finallyLoc));
434   }
435
436   fiber->data_ = reinterpret_cast<intptr_t>(fiber);
437   readyFibers_.push_back(*fiber);
438   if (observer_) {
439     observer_->runnable(reinterpret_cast<uintptr_t>(fiber));
440   }
441
442   ensureLoopScheduled();
443 }
444
445 template <typename F>
446 typename std::result_of<F()>::type FiberManager::runInMainContext(F&& func) {
447   if (UNLIKELY(activeFiber_ == nullptr)) {
448     return func();
449   }
450
451   typedef typename std::result_of<F()>::type Result;
452
453   folly::Try<Result> result;
454   auto f = [&func, &result]() mutable {
455     result = folly::makeTryWith(std::forward<F>(func));
456   };
457
458   immediateFunc_ = std::ref(f);
459   activeFiber_->preempt(Fiber::AWAITING_IMMEDIATE);
460
461   return std::move(result).value();
462 }
463
464 inline FiberManager& FiberManager::getFiberManager() {
465   assert(currentFiberManager_ != nullptr);
466   return *currentFiberManager_;
467 }
468
469 inline FiberManager* FiberManager::getFiberManagerUnsafe() {
470   return currentFiberManager_;
471 }
472
473 inline bool FiberManager::hasActiveFiber() const {
474   return activeFiber_ != nullptr;
475 }
476
477 inline void FiberManager::yield() {
478   assert(currentFiberManager_ == this);
479   assert(activeFiber_ != nullptr);
480   assert(activeFiber_->state_ == Fiber::RUNNING);
481   activeFiber_->preempt(Fiber::YIELDED);
482 }
483
484 template <typename T>
485 T& FiberManager::local() {
486   if (std::type_index(typeid(T)) == localType_ && currentFiber_) {
487     return currentFiber_->localData_.get<T>();
488   }
489   return localThread<T>();
490 }
491
492 template <typename T>
493 T& FiberManager::localThread() {
494 #ifndef __APPLE__
495   static thread_local T t;
496   return t;
497 #else // osx doesn't support thread_local
498   static ThreadLocal<T> t;
499   return *t;
500 #endif
501 }
502
503 inline void FiberManager::initLocalData(Fiber& fiber) {
504   auto fm = getFiberManagerUnsafe();
505   if (fm && fm->currentFiber_ && fm->localType_ == localType_) {
506     fiber.localData_ = fm->currentFiber_->localData_;
507   }
508   fiber.rcontext_ = RequestContext::saveContext();
509 }
510
511 template <typename LocalT>
512 FiberManager::FiberManager(
513     LocalType<LocalT>,
514     std::unique_ptr<LoopController> loopController__,
515     Options options)
516     : loopController_(std::move(loopController__)),
517       stackAllocator_(options.useGuardPages),
518       options_(preprocessOptions(std::move(options))),
519       exceptionCallback_([](std::exception_ptr eptr, std::string context) {
520         try {
521           std::rethrow_exception(eptr);
522         } catch (const std::exception& e) {
523           LOG(DFATAL) << "Exception " << typeid(e).name() << " with message '"
524                       << e.what() << "' was thrown in "
525                       << "FiberManager with context '" << context << "'";
526           throw;
527         } catch (...) {
528           LOG(DFATAL) << "Unknown exception was thrown in FiberManager with "
529                       << "context '" << context << "'";
530           throw;
531         }
532       }),
533       timeoutManager_(std::make_shared<TimeoutController>(*loopController_)),
534       fibersPoolResizer_(*this),
535       localType_(typeid(LocalT)) {
536   loopController_->setFiberManager(this);
537 }
538
539 template <typename F>
540 typename FirstArgOf<F>::type::value_type inline await(F&& func) {
541   typedef typename FirstArgOf<F>::type::value_type Result;
542
543   folly::Try<Result> result;
544
545   Baton baton;
546   baton.wait([&func, &result, &baton]() mutable {
547     func(Promise<Result>(result, baton));
548   });
549
550   return folly::moveFromTry(result);
551 }
552 }
553 }