FanoutChannel: Add support for custom context

Summary:
One common pattern with FanoutChannel is to send to new subscribers an initial update indicating the current state of the stream. This is currently the pattern for all uses of FanoutChannel.

This pattern is currently accomplished by adding a transform to the input receiver before passing it to fanout channel. The transform updates the current state of the stream in some shared state object. That shared state object is then captured and used in the getInitialValues function passed to subscribe, in order to let new subscribers know the current state of the stream.

However, this approach can lead to a race condition. In this approach, the transform function is executed (on the transform executor) before the getInitialValues function is executed (on the FanoutChannel executor). If someone adds a new subscriber in between, getInitialValues will use the updated shared state, even though the corresponding update has not yet been sent from the output of the transform to the input of the fanout channel.

To solve this race condition, we need to ensure that new subscribers are not added between the shared state change and the fanning out of the update that led to the shared state change. To do this, this diff adds explicit support for a custom context object that can store state. The context object's update function is called on every update, allowing the shared state to be updated on each new value. The context object is also accessible to the getInitialValues function, which allows sending an update with the current state (based on the context) to new subscribers.

This enables the desired pattern without a race condition, and avoids the need for a transform.

Reviewed By: aary

Differential Revision: D30889893

fbshipit-source-id: 9a79fd5a823db1ae477b6b63170978925b791dda

folly/experimental/channels/FanoutChannel-inl.h

@@ -25,16 +25,18 @@
 namespace folly {
 namespace channels {
 
-template <typename ValueType>
-FanoutChannel<ValueType>::FanoutChannel(TProcessor* processor)
+template <typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType>::FanoutChannel(TProcessor* processor)
     : processor_(processor) {}
 
-template <typename ValueType>
-FanoutChannel<ValueType>::FanoutChannel(FanoutChannel&& other) noexcept
+template <typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType>::FanoutChannel(
+    FanoutChannel&& other) noexcept
     : processor_(std::exchange(other.processor_, nullptr)) {}
 
-template <typename ValueType>
-FanoutChannel<ValueType>& FanoutChannel<ValueType>::operator=(
+template <typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType>&
+FanoutChannel<ValueType, ContextType>::operator=(
     FanoutChannel&& other) noexcept {
   if (&other == this) {
     return *this;
@@ -46,31 +48,33 @@ FanoutChannel<ValueType>& FanoutChannel<ValueType>::operator=(
   return *this;
 }
 
-template <typename ValueType>
-FanoutChannel<ValueType>::~FanoutChannel() {
+template <typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType>::~FanoutChannel() {
   if (processor_ != nullptr) {
     std::move(*this).close(folly::exception_wrapper());
   }
 }
 
-template <typename ValueType>
-FanoutChannel<ValueType>::operator bool() const {
+template <typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType>::operator bool() const {
   return processor_ != nullptr;
 }
 
-template <typename ValueType>
-Receiver<ValueType> FanoutChannel<ValueType>::subscribe(
-    folly::Function<std::vector<ValueType>()> getInitialValues) {
+template <typename ValueType, typename ContextType>
+Receiver<ValueType> FanoutChannel<ValueType, ContextType>::subscribe(
+    folly::Function<std::vector<ValueType>(const ContextType&)>
+        getInitialValues) {
   return processor_->subscribe(std::move(getInitialValues));
 }
 
-template <typename ValueType>
-bool FanoutChannel<ValueType>::anySubscribers() {
+template <typename ValueType, typename ContextType>
+bool FanoutChannel<ValueType, ContextType>::anySubscribers() {
   return processor_->anySubscribers();
 }
 
-template <typename ValueType>
-void FanoutChannel<ValueType>::close(folly::exception_wrapper ex) && {
+template <typename ValueType, typename ContextType>
+void FanoutChannel<ValueType, ContextType>::close(
+    folly::exception_wrapper ex) && {
   processor_->destroyHandle(
       ex ? detail::CloseResult(std::move(ex)) : detail::CloseResult());
   processor_ = nullptr;
@@ -78,11 +82,12 @@ void FanoutChannel<ValueType>::close(folly::exception_wrapper ex) && {
 
 namespace detail {
 
-template <typename ValueType>
+template <typename ValueType, typename ContextType>
 class IFanoutChannelProcessor : public IChannelCallback {
  public:
   virtual Receiver<ValueType> subscribe(
-      folly::Function<std::vector<ValueType>()> getInitialValues) = 0;
+      folly::Function<std::vector<ValueType>(const ContextType&)>
+          getInitialValues) = 0;
 
   virtual bool anySubscribers() = 0;
 
@@ -108,16 +113,20 @@ class IFanoutChannelProcessor : public IChannelCallback {
  * then be deleted once the input receiver transitions to the
  * CancellationProcessed state.
  */
-template <typename ValueType>
-class FanoutChannelProcessor : public IFanoutChannelProcessor<ValueType> {
+template <typename ValueType, typename ContextType>
+class FanoutChannelProcessor
+    : public IFanoutChannelProcessor<ValueType, ContextType> {
  private:
   struct State {
+    State(ContextType _context) : context(std::move(_context)) {}
+
     ChannelState getReceiverState() {
       return detail::getReceiverState(receiver.get());
     }
 
     ChannelBridgePtr<ValueType> receiver;
     FanoutSender<ValueType> fanoutSender;
+    ContextType context;
     bool handleDeleted{false};
   };
 
@@ -125,8 +134,9 @@ class FanoutChannelProcessor : public IFanoutChannelProcessor<ValueType> {
 
  public:
   explicit FanoutChannelProcessor(
-      folly::Executor::KeepAlive<folly::SequencedExecutor> executor)
-      : executor_(std::move(executor)) {}
+      folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
+      ContextType context)
+      : executor_(std::move(executor)), state_(std::move(context)) {}
 
   /**
    * Starts fanning out values from the input receiver to all output receivers.
@@ -150,10 +160,11 @@ class FanoutChannelProcessor : public IFanoutChannelProcessor<ValueType> {
    * receiver.
    */
   Receiver<ValueType> subscribe(
-      folly::Function<std::vector<ValueType>()> getInitialValues) override {
+      folly::Function<std::vector<ValueType>(const ContextType&)>
+          getInitialValues) override {
     auto state = state_.wlock();
-    auto initialValues =
-        getInitialValues ? getInitialValues() : std::vector<ValueType>();
+    auto initialValues = getInitialValues ? getInitialValues(state->context)
+                                          : std::vector<ValueType>();
     if (!state->receiver) {
       auto [receiver, sender] = Channel<ValueType>::create();
       for (auto&& value : initialValues) {
@@ -251,6 +262,7 @@ class FanoutChannelProcessor : public IFanoutChannelProcessor<ValueType> {
       if (inputResult.hasValue()) {
         // We have received a normal value from the input receiver. Write it to
         // all output senders.
+        state->context.update(inputResult.value());
         state->fanoutSender.write(std::move(inputResult.value()));
       } else {
         // The input receiver was closed.
@@ -312,14 +324,15 @@ class FanoutChannelProcessor : public IFanoutChannelProcessor<ValueType> {
 };
 } // namespace detail
 
-template <typename TReceiver, typename ValueType>
-FanoutChannel<ValueType> createFanoutChannel(
+template <typename TReceiver, typename ValueType, typename ContextType>
+FanoutChannel<ValueType, ContextType> createFanoutChannel(
     TReceiver inputReceiver,
-    folly::Executor::KeepAlive<folly::SequencedExecutor> executor) {
-  auto* processor =
-      new detail::FanoutChannelProcessor<ValueType>(std::move(executor));
+    folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
+    ContextType context) {
+  auto* processor = new detail::FanoutChannelProcessor<ValueType, ContextType>(
+      std::move(executor), std::move(context));
   processor->start(std::move(inputReceiver));
-  return FanoutChannel<ValueType>(processor);
+  return FanoutChannel<ValueType, ContextType>(processor);
 }
 } // namespace channels
 } // namespace folly

folly/experimental/channels/FanoutChannel.h

@@ -23,10 +23,15 @@ namespace folly {
 namespace channels {
 
 namespace detail {
-template <typename ValueType>
+template <typename ValueType, typename ContextType>
 class IFanoutChannelProcessor;
 }
 
+template <typename TValue>
+struct NoContext {
+  void update(const TValue&) {}
+};
+
 /**
  * A fanout channel allows fanning out updates from a single input receiver
  * to multiple output receivers.
@@ -35,7 +40,17 @@ class IFanoutChannelProcessor;
  * computes a set of initial values. These initial values will only be sent to
  * the new receiver.
  *
- * Example:
+ * FanoutChannel allows specifying an optional context object. If specified, the
+ * context object must have a void update function:
+ *
+ *   void update(const ValueType&);
+ *
+ * This update function will be called on every value from the input receiver.
+ * The context will be passed to the getInitialUpdates argument to subscribe,
+ * allowing for initial updates to depend on the context. This facilitates the
+ * common pattern of letting new subscribers know where they are starting from.
+ *
+ * Example without context:
  *
  *   // Function that returns a receiver:
  *   Receiver<int> getInputReceiver();
@@ -47,10 +62,28 @@ class IFanoutChannelProcessor;
  *   auto receiver1 = fanoutChannel.subscribe();
  *   auto receiver2 = fanoutChannel.subscribe();
  *   auto receiver3 = fanoutChannel.subscribe([]{ return {1, 2, 3}; });
+ *
+ * Example with context:
+ *
+ *   struct Context {
+ *     int lastValue{-1};
+ *
+ *     void update(const int& value) {
+ *       lastValue = value;
+ *     }
+ *   };
+ *
+ *   auto fanoutChannel =
+ *       createFanoutChannel(getReceiver(), getExecutor(), Context());
+ *   auto receiver1 = fanoutChannel.subscribe(
+ *       [](const Context& context) { return {context.latestValue}; });
+ *   auto receiver2 = fanoutChannel.subscribe(
+ *       [](const Context& context) { return {context.latestValue}; });
+ *   std::move(fanoutChannel).close();
  */
-template <typename ValueType>
+template <typename ValueType, typename ContextType = NoContext<ValueType>>
 class FanoutChannel {
-  using TProcessor = detail::IFanoutChannelProcessor<ValueType>;
+  using TProcessor = detail::IFanoutChannelProcessor<ValueType, ContextType>;
 
  public:
   explicit FanoutChannel(TProcessor* processor);
@@ -73,7 +106,8 @@ class FanoutChannel {
    * getInitialValues, or a deadlock will occur.
    */
   Receiver<ValueType> subscribe(
-      folly::Function<std::vector<ValueType>()> getInitialValues = {});
+      folly::Function<std::vector<ValueType>(const ContextType&)>
+          getInitialValues = {});
 
   /**
    * Returns whether this fanout channel has any output receivers.
@@ -94,10 +128,12 @@ class FanoutChannel {
  */
 template <
     typename ReceiverType,
-    typename ValueType = typename ReceiverType::ValueType>
-FanoutChannel<ValueType> createFanoutChannel(
+    typename ValueType = typename ReceiverType::ValueType,
+    typename ContextType = NoContext<typename ReceiverType::ValueType>>
+FanoutChannel<ValueType, ContextType> createFanoutChannel(
     ReceiverType inputReceiver,
-    folly::Executor::KeepAlive<folly::SequencedExecutor> executor);
+    folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
+    ContextType context = ContextType());
 } // namespace channels
 } // namespace folly
 

folly/experimental/channels/test/FanoutChannelTest.cpp

@@ -60,16 +60,22 @@ class FanoutChannelFixture : public Test {
 };
 
 TEST_F(FanoutChannelFixture, ReceiveValue_FanoutBroadcastsValues) {
+  struct LatestVersion {
+    int version{-1};
+
+    void update(const int& newVersion) { version = newVersion; }
+  };
+
   auto [inputReceiver, sender] = Channel<int>::create();
-  auto fanoutChannel =
-      createFanoutChannel(std::move(inputReceiver), &executor_);
+  auto fanoutChannel = createFanoutChannel(
+      std::move(inputReceiver), &executor_, LatestVersion());
 
   EXPECT_FALSE(fanoutChannel.anySubscribers());
 
   auto [handle1, callback1] = processValues(fanoutChannel.subscribe(
-      []() { return toVector(100); } /* getInitialValues */));
+      [](const auto&) { return toVector(100); } /* getInitialValues */));
   auto [handle2, callback2] = processValues(fanoutChannel.subscribe(
-      []() { return toVector(200); } /* getInitialValues */));
+      [](const auto&) { return toVector(200); } /* getInitialValues */));
 
   EXPECT_TRUE(fanoutChannel.anySubscribers());
   EXPECT_CALL(*callback1, onValue(100));
@@ -84,10 +90,12 @@ TEST_F(FanoutChannelFixture, ReceiveValue_FanoutBroadcastsValues) {
   sender.write(2);
   executor_.drain();
 
-  auto [handle3, callback3] = processValues(fanoutChannel.subscribe(
-      []() { return toVector(300); } /* getInitialValues */));
+  auto [handle3, callback3] = processValues(
+      fanoutChannel.subscribe([](const LatestVersion& latestVersion) {
+        return toVector(latestVersion.version);
+      } /* getInitialValues */));
 
-  EXPECT_CALL(*callback3, onValue(300));
+  EXPECT_CALL(*callback3, onValue(2));
   executor_.drain();
 
   sender.write(3);
@@ -203,9 +211,9 @@ TEST_F(FanoutChannelFixture, VectorBool) {
       createFanoutChannel(std::move(inputReceiver), &executor_);
 
   auto [handle1, callback1] = processValues(fanoutChannel.subscribe(
-      [] { return toVector(true); } /* getInitialValues */));
+      [](const auto&) { return toVector(true); } /* getInitialValues */));
   auto [handle2, callback2] = processValues(fanoutChannel.subscribe(
-      [] { return toVector(false); } /* getInitialValues */));
+      [](const auto&) { return toVector(false); } /* getInitialValues */));
 
   EXPECT_CALL(*callback1, onValue(true));
   EXPECT_CALL(*callback2, onValue(false));