folly/experimental/fibers/AddTasks-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 #include <memory>
  17 #include <vector>
  18
  19 #include <folly/experimental/fibers/FiberManager.h>
  20
  21 namespace folly { namespace fibers {
  22
  23 template <typename T>
  24 TaskIterator<T>::TaskIterator(TaskIterator&& other) noexcept
  25     : context_(std::move(other.context_)),
  26       id_(other.id_) {
  27 }
  28
  29 template <typename T>
  30 TaskIterator<T>::TaskIterator(std::shared_ptr<Context> context)
  31     : context_(std::move(context)),
  32       id_(-1) {
  33   assert(context_);
  34 }
  35
  36 template <typename T>
  37 inline bool TaskIterator<T>::hasCompleted() const {
  38   return context_->tasksConsumed < context_->results.size();
  39 }
  40
  41 template <typename T>
  42 inline bool TaskIterator<T>::hasPending() const {
  43   return !context_.unique();
  44 }
  45
  46 template <typename T>
  47 inline bool TaskIterator<T>::hasNext() const {
  48   return hasPending() || hasCompleted();
  49 }
  50
  51 template <typename T>
  52 folly::Try<T> TaskIterator<T>::awaitNextResult() {
  53   assert(hasCompleted() || hasPending());
  54   reserve(1);
  55
  56   size_t i = context_->tasksConsumed++;
  57   id_ = context_->results[i].first;
  58   return std::move(context_->results[i].second);
  59 }
  60
  61 template <typename T>
  62 inline T TaskIterator<T>::awaitNext() {
  63   return std::move(awaitNextResult().value());
  64 }
  65
  66 template <>
  67 inline void TaskIterator<void>::awaitNext() {
  68   awaitNextResult().value();
  69 }
  70
  71 template <typename T>
  72 inline void TaskIterator<T>::reserve(size_t n) {
  73   size_t tasksReady = context_->results.size() - context_->tasksConsumed;
  74
  75   // we don't need to do anything if there are already n or more tasks complete
  76   // or if we have no tasks left to execute.
  77   if (!hasPending() || tasksReady >= n) {
  78     return;
  79   }
  80
  81   n -= tasksReady;
  82   size_t tasksLeft = context_->totalTasks - context_->results.size();
  83   n = std::min(n, tasksLeft);
  84
  85   await(
  86     [this, n](Promise<void> promise) {
  87       context_->tasksToFulfillPromise = n;
  88       context_->promise.assign(std::move(promise));
  89     });
  90 }
  91
  92 template <typename T>
  93 inline size_t TaskIterator<T>::getTaskID() const {
  94   assert(id_ != static_cast<size_t>(-1));
  95   return id_;
  96 }
  97
  98 template <class InputIterator>
  99 TaskIterator<typename std::result_of<
 100   typename std::iterator_traits<InputIterator>::value_type()>::type>
 101 addTasks(InputIterator first, InputIterator last) {
 102   typedef typename std::result_of<
 103     typename std::iterator_traits<InputIterator>::value_type()>::type
 104       ResultType;
 105   typedef TaskIterator<ResultType> IteratorType;
 106
 107   auto context = std::make_shared<typename IteratorType::Context>();
 108   context->totalTasks = std::distance(first, last);
 109   context->results.reserve(context->totalTasks);
 110
 111   for (size_t i = 0; first != last; ++i, ++first) {
 112 #ifdef __clang__
 113 #pragma clang diagnostic push // ignore generalized lambda capture warning
 114 #pragma clang diagnostic ignored "-Wc++1y-extensions"
 115 #endif
 116     addTask(
 117       [i, context, f = std::move(*first)]() {
 118         context->results.emplace_back(i, folly::makeTryWith(std::move(f)));
 119
 120         // Check for awaiting iterator.
 121         if (context->promise.hasValue()) {
 122           if (--context->tasksToFulfillPromise == 0) {
 123             context->promise->setValue();
 124             context->promise.clear();
 125           }
 126         }
 127       }
 128     );
 129 #ifdef __clang__
 130 #pragma clang diagnostic pop
 131 #endif
 132   }
 133
 134   return IteratorType(std::move(context));
 135 }
 136
 137 }}