folly/wangle/detail/State.h

   1 /*
   2  * Copyright 2014 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
  17 #pragma once
  18
  19 #include <atomic>
  20 #include <mutex>
  21 #include <stdexcept>
  22 #include <vector>
  23
  24 #include <folly/Optional.h>
  25
  26 #include <folly/wangle/Try.h>
  27 #include <folly/wangle/Promise.h>
  28 #include <folly/wangle/Future.h>
  29 #include <folly/wangle/Executor.h>
  30
  31 namespace folly { namespace wangle { namespace detail {
  32
  33 // As of GCC 4.8.1, the std::function in libstdc++ optimizes only for pointers
  34 // to functions, using a helper avoids a call to malloc.
  35 template<typename T>
  36 void empty_callback(Try<T>&&) { }
  37
  38 /** The shared state object for Future and Promise. */
  39 template<typename T>
  40 class State {
  41  public:
  42   // This must be heap-constructed. There's probably a way to enforce that in
  43   // code but since this is just internal detail code and I don't know how
  44   // off-hand, I'm punting.
  45   State() = default;
  46   ~State() {
  47     assert(calledBack_);
  48     assert(detached_ == 2);
  49   }
  50
  51   // not copyable
  52   State(State const&) = delete;
  53   State& operator=(State const&) = delete;
  54
  55   // not movable (see comment in the implementation of Future::then)
  56   State(State&&) noexcept = delete;
  57   State& operator=(State&&) = delete;
  58
  59   Try<T>& getTry() {
  60     return *value_;
  61   }
  62
  63   template <typename F>
  64   void setCallback(F func) {
  65     {
  66       std::lock_guard<decltype(mutex_)> lock(mutex_);
  67
  68       if (callback_) {
  69         throw std::logic_error("setCallback called twice");
  70       }
  71
  72       callback_ = std::move(func);
  73     }
  74
  75     maybeCallback();
  76   }
  77
  78   void fulfil(Try<T>&& t) {
  79     {
  80       std::lock_guard<decltype(mutex_)> lock(mutex_);
  81
  82       if (ready()) {
  83         throw std::logic_error("fulfil called twice");
  84       }
  85
  86       value_ = std::move(t);
  87       assert(ready());
  88     }
  89
  90     maybeCallback();
  91   }
  92
  93   void setException(std::exception_ptr const& e) {
  94     fulfil(Try<T>(e));
  95   }
  96
  97   template <class E> void setException(E const& e) {
  98     fulfil(Try<T>(std::make_exception_ptr<E>(e)));
  99   }
 100
 101   bool ready() const {
 102     return value_.hasValue();
 103   }
 104
 105   typename std::add_lvalue_reference<T>::type value() {
 106     if (ready()) {
 107       return value_->value();
 108     } else {
 109       throw FutureNotReady();
 110     }
 111   }
 112
 113   // Called by a destructing Future
 114   void detachFuture() {
 115     if (!callback_) {
 116       setCallback(empty_callback<T>);
 117     }
 118     activate();
 119     detachOne();
 120   }
 121
 122   // Called by a destructing Promise
 123   void detachPromise() {
 124     if (!ready()) {
 125       setException(BrokenPromise());
 126     }
 127     detachOne();
 128   }
 129
 130   void deactivate() {
 131     std::lock_guard<decltype(mutex_)> lock(mutex_);
 132     active_ = false;
 133   }
 134
 135   void activate() {
 136     {
 137       std::lock_guard<decltype(mutex_)> lock(mutex_);
 138       active_ = true;
 139     }
 140     maybeCallback();
 141   }
 142
 143   bool isActive() { return active_; }
 144
 145   void setExecutor(Executor* x) {
 146     std::lock_guard<decltype(mutex_)> lock(mutex_);
 147     executor_ = x;
 148   }
 149
 150  private:
 151   void maybeCallback() {
 152     std::lock_guard<decltype(mutex_)> lock(mutex_);
 153     if (!calledBack_ &&
 154         value_ && callback_ && isActive()) {
 155       // TODO(5306911) we should probably try/catch here
 156       if (executor_) {
 157         MoveWrapper<folly::Optional<Try<T>>> val(std::move(value_));
 158         MoveWrapper<std::function<void(Try<T>&&)>> cb(std::move(callback_));
 159         executor_->add([cb, val]() mutable { (*cb)(std::move(**val)); });
 160         calledBack_ = true;
 161       } else {
 162         callback_(std::move(*value_));
 163         calledBack_ = true;
 164       }
 165     }
 166   }
 167
 168   void detachOne() {
 169     bool shouldDelete;
 170     {
 171       std::lock_guard<decltype(mutex_)> lock(mutex_);
 172       detached_++;
 173       assert(detached_ == 1 || detached_ == 2);
 174       shouldDelete = (detached_ == 2);
 175     }
 176
 177     if (shouldDelete) {
 178       // we should have already executed the callback with the value
 179       assert(calledBack_);
 180       delete this;
 181     }
 182   }
 183
 184   folly::Optional<Try<T>> value_;
 185   std::function<void(Try<T>&&)> callback_;
 186   bool calledBack_ = false;
 187   unsigned char detached_ = 0;
 188   bool active_ = true;
 189   Executor* executor_ = nullptr;
 190
 191   // this lock isn't meant to protect all accesses to members, only the ones
 192   // that need to be threadsafe: the act of setting value_ and callback_, and
 193   // seeing if they are set and whether we should then continue.
 194   std::recursive_mutex mutex_;
 195 };
 196
 197 template <typename... Ts>
 198 struct VariadicContext {
 199   VariadicContext() : total(0), count(0) {}
 200   Promise<std::tuple<Try<Ts>... > > p;
 201   std::tuple<Try<Ts>... > results;
 202   size_t total;
 203   std::atomic<size_t> count;
 204   typedef Future<std::tuple<Try<Ts>...>> type;
 205 };
 206
 207 template <typename... Ts, typename THead, typename... Fs>
 208 typename std::enable_if<sizeof...(Fs) == 0, void>::type
 209 whenAllVariadicHelper(VariadicContext<Ts...> *ctx, THead&& head, Fs&&... tail) {
 210   head.setCallback_([ctx](Try<typename THead::value_type>&& t) {
 211     std::get<sizeof...(Ts) - sizeof...(Fs) - 1>(ctx->results) = std::move(t);
 212     if (++ctx->count == ctx->total) {
 213       ctx->p.setValue(std::move(ctx->results));
 214       delete ctx;
 215     }
 216   });
 217 }
 218
 219 template <typename... Ts, typename THead, typename... Fs>
 220 typename std::enable_if<sizeof...(Fs) != 0, void>::type
 221 whenAllVariadicHelper(VariadicContext<Ts...> *ctx, THead&& head, Fs&&... tail) {
 222   head.setCallback_([ctx](Try<typename THead::value_type>&& t) {
 223     std::get<sizeof...(Ts) - sizeof...(Fs) - 1>(ctx->results) = std::move(t);
 224     if (++ctx->count == ctx->total) {
 225       ctx->p.setValue(std::move(ctx->results));
 226       delete ctx;
 227     }
 228   });
 229   // template tail-recursion
 230   whenAllVariadicHelper(ctx, std::forward<Fs>(tail)...);
 231 }
 232
 233 template <typename T>
 234 struct WhenAllContext {
 235   explicit WhenAllContext() : count(0), total(0) {}
 236   Promise<std::vector<Try<T> > > p;
 237   std::vector<Try<T> > results;
 238   std::atomic<size_t> count;
 239   size_t total;
 240 };
 241
 242 template <typename T>
 243 struct WhenAnyContext {
 244   explicit WhenAnyContext(size_t n) : done(false), ref_count(n) {};
 245   Promise<std::pair<size_t, Try<T>>> p;
 246   std::atomic<bool> done;
 247   std::atomic<size_t> ref_count;
 248   void decref() {
 249     if (--ref_count == 0) {
 250       delete this;
 251     }
 252   }
 253 };
 254
 255 template <typename T>
 256 struct WhenAllLaterContext {
 257   explicit WhenAllLaterContext() : count(0), total(0) {}
 258   std::function<void(std::vector<Try<T>>&&)> fn;
 259   std::vector<Try<T> > results;
 260   std::atomic<size_t> count;
 261   size_t total;
 262 };
 263
 264 }}} // namespace