Commit 7f69cb31 authored by Andrew Smith's avatar Andrew Smith Committed by Facebook GitHub Bot

Add FanoutSender abstraction to channels framework

Summary:
The existing FanoutChannel construct takes in an input receiver, allows the creation of output receivers, and broadcasts all received values to the output receivers. It is good for the scenario where you already have an input receiver that you want to forward to multiple output receivers.

However, for the scenario for which there is no existing input receiver, it is a bit heavy weight. To use it in this scenario, you have to create a separate channel, pass its receiver as the input receiver to the FanoutChannel, and write values to the sender.

In addition to having to create a separate input channel that is otherwise unnecessary, it is not optimized for the case where there is a single output receiver. In many cases, there is typically only one output receiver. It would be ideal to have a construct that uses the same amount of memory as a simple channel, in the case where there is only one output receiver.

FanoutSender solves both of these problems. FanoutSender does not require an input channel, and instead allows one to directly write values to all output receivers. In addition, in the case where there is only one output receiver, it uses exactly the same amount of memory as a simple channel.

Reviewed By: aary

Differential Revision: D29896797

fbshipit-source-id: 18178f42364d53e84fb07424ed3e8cc0497e21e1
parent b817015d
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
namespace folly {
namespace channels {
template <typename ValueType>
FanoutSender<ValueType>::FanoutSender()
: senders_(static_cast<detail::ChannelBridge<ValueType>*>(nullptr)) {}
template <typename ValueType>
FanoutSender<ValueType>::FanoutSender(FanoutSender&& other) noexcept
: senders_(std::move(other.senders_)) {}
template <typename ValueType>
FanoutSender<ValueType>& FanoutSender<ValueType>::operator=(
FanoutSender&& other) noexcept {
if (&other == this) {
return *this;
}
std::move(*this).close();
senders_ = std::move(senders_);
return *this;
}
template <typename ValueType>
FanoutSender<ValueType>::~FanoutSender() {
std::move(*this).close();
}
template <typename ValueType>
Receiver<ValueType> FanoutSender<ValueType>::subscribe(
std::vector<ValueType> initialValues) {
clearSendersWithClosedReceivers();
auto [newReceiver, newSender] = Channel<ValueType>::create();
for (auto& initialValue : initialValues) {
newSender.write(std::move(initialValue));
}
if (!anyReceivers()) {
// There are currently no output receivers. Store the new output receiver.
senders_.set(detail::senderGetBridge(newSender).release());
} else if (!hasSenderSet()) {
// There is currently exactly one output receiver. Convert to a sender set.
auto senderSet = std::make_unique<
folly::F14FastSet<detail::ChannelBridgePtr<ValueType>>>();
senderSet->insert(detail::ChannelBridgePtr<ValueType>(getSingleSender()));
senderSet->insert(std::move(detail::senderGetBridge(newSender)));
senders_.set(senderSet.release());
} else {
// There are currently more than one output receivers. Add the new receiver
// to the existing sender set.
auto* senderSet = getSenderSet();
senderSet->insert(std::move(detail::senderGetBridge(newSender)));
}
return std::move(newReceiver);
}
template <typename ValueType>
bool FanoutSender<ValueType>::anyReceivers() {
clearSendersWithClosedReceivers();
return hasSenderSet() || getSingleSender() != nullptr;
}
template <typename ValueType>
template <typename U>
void FanoutSender<ValueType>::write(U&& element) {
clearSendersWithClosedReceivers();
if (!anyReceivers()) {
// There are currently no output receivers to write to.
return;
} else if (!hasSenderSet()) {
// There is exactly one output receiver. Write the value to that receiver.
getSingleSender()->senderPush(std::forward<U>(element));
} else {
// There are more than one output receivers. Write the value to all
// receivers.
for (auto& senderBridge : *getSenderSet()) {
senderBridge->senderPush(element);
}
}
}
template <typename ValueType>
void FanoutSender<ValueType>::close(folly::exception_wrapper ex) && {
clearSendersWithClosedReceivers();
if (!anyReceivers()) {
// There are no output receivers to close.
return;
} else if (!hasSenderSet()) {
// There is exactly one output receiver to close.
if (ex) {
getSingleSender()->senderClose(ex);
} else {
getSingleSender()->senderClose();
}
// Delete the output receiver.
(detail::ChannelBridgePtr<ValueType>(getSingleSender()));
senders_.set(static_cast<detail::ChannelBridge<ValueType>*>(nullptr));
} else {
// There are more than one output receivers to close.
auto senderSet =
std::unique_ptr<F14FastSet<detail::ChannelBridgePtr<ValueType>>>(
getSenderSet());
senders_.set(static_cast<detail::ChannelBridge<ValueType>*>(nullptr));
for (auto& senderBridge : *senderSet) {
if (ex) {
senderBridge->senderClose(ex);
} else {
senderBridge->senderClose();
}
}
}
}
template <typename ValueType>
bool FanoutSender<ValueType>::hasSenderSet() {
return senders_.index() == 1;
}
template <typename ValueType>
detail::ChannelBridge<ValueType>* FanoutSender<ValueType>::getSingleSender() {
return senders_.get(folly::tag_t<detail::ChannelBridge<ValueType>>{});
}
template <typename ValueType>
F14FastSet<detail::ChannelBridgePtr<ValueType>>*
FanoutSender<ValueType>::getSenderSet() {
return senders_.get(
folly::tag_t<folly::F14FastSet<detail::ChannelBridgePtr<ValueType>>>{});
}
template <typename ValueType>
void FanoutSender<ValueType>::clearSendersWithClosedReceivers() {
if (hasSenderSet()) {
// There are currently more than one output receivers. Check all of them to
// see if any have been cancelled.
auto* senderSet = getSenderSet();
for (auto it = senderSet->begin(); it != senderSet->end();) {
auto values = it->get()->senderGetValues();
if (!values.empty()) {
it->get()->senderClose();
it = senderSet->erase(it);
} else {
++it;
}
}
if (senderSet->empty()) {
// After erasing all cancelled output receivers, there are no remaining
// receivers.
(std::unique_ptr<F14FastSet<detail::ChannelBridgePtr<ValueType>>>(
senderSet));
senders_.set(static_cast<detail::ChannelBridge<ValueType>*>(nullptr));
senderSet = nullptr;
} else if (senderSet->size() == 1) {
// After erasing all cancelled output receivers, there is exactly one
// remaining receiver. Move it out of the set, and store it as a single
// receiver.
auto senderSetUniqPtr =
std::unique_ptr<F14FastSet<detail::ChannelBridgePtr<ValueType>>>(
senderSet);
senderSetUniqPtr->eraseInto(senderSet->begin(), [&](auto&& senderBridge) {
senders_.set(senderBridge.release());
});
senderSet = nullptr;
}
} else {
auto* bridge = getSingleSender();
if (bridge != nullptr) {
// There is currently exactly one output receiver. Check to see if it has
// been cancelled.
auto values = bridge->senderGetValues();
if (!values.empty()) {
bridge->senderClose();
senders_.set(static_cast<detail::ChannelBridge<ValueType>*>(nullptr));
// Delete the output receiver.
(detail::ChannelBridgePtr<ValueType>(bridge));
}
}
}
}
} // namespace channels
} // namespace folly
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <folly/container/F14Set.h>
#include <folly/experimental/channels/Channel.h>
#include <folly/experimental/channels/detail/PointerVariant.h>
namespace folly {
namespace channels {
/**
* A FanoutSender allows fanning out updates to multiple output receivers.
* Values can be written as with a normal Sender. When there is only one output
* receiver, the memory used by a FanoutSender (and the corresponding output
* receiver) is the same as the memory used by a normal channel.
*
* When a new output receiver is added, an optional vector of initial values
* can be provided. These initial values will only be sent to the new receiver.
*
* Memory used by closed receivers is reclaimed lazily (when iterating over
* receivers).
*
* Example:
*
* FanoutSender<int> fanoutSender;
* auto receiver1 = fanoutSender.getNewReceiver();
* auto receiver2 = fanoutSender.getNewReceiver();
* auto receiver3 = fanoutSender.getNewReceiver({1, 2, 3});
* std::move(fanoutSender).close();
*/
template <typename ValueType>
class FanoutSender {
public:
FanoutSender();
FanoutSender(FanoutSender&& other) noexcept;
FanoutSender& operator=(FanoutSender&& other) noexcept;
~FanoutSender();
/**
* Returns a new output receiver that will receive all values written to the
* FanoutSender. If the initialValues parameter is provided, the given values
* will (only) go to the new output receiver.
*/
Receiver<ValueType> subscribe(std::vector<ValueType> initialValues = {});
/**
* Returns whether this fanout sender has any active output receivers.
*/
bool anyReceivers();
/**
* Sends the given value to all corresponding receivers.
*/
template <typename U = ValueType>
void write(U&& element);
/**
* Closes the fanout sender.
*/
void close(folly::exception_wrapper ex = folly::exception_wrapper()) &&;
private:
bool hasSenderSet();
detail::ChannelBridge<ValueType>* getSingleSender();
folly::F14FastSet<detail::ChannelBridgePtr<ValueType>>* getSenderSet();
void clearSendersWithClosedReceivers();
detail::PointerVariant<
detail::ChannelBridge<ValueType>,
folly::F14FastSet<detail::ChannelBridgePtr<ValueType>>>
senders_;
};
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/FanoutSender-inl.h>
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <fmt/format.h>
#include <folly/Demangle.h>
namespace folly {
namespace channels {
namespace detail {
/**
* A PointerVariant stores a pointer of one of two possible types.
*/
template <typename FirstType, typename SecondType>
class PointerVariant {
public:
template <typename T>
explicit PointerVariant(T* pointer) {
set(pointer);
}
PointerVariant(PointerVariant&& other) noexcept
: storage_(std::exchange(other.storage_, 0)) {}
PointerVariant& operator=(PointerVariant&& other) noexcept {
storage_ = std::exchange(other.storage_, 0);
return *this;
}
/**
* Returns the zero-based index of the type that is currently held.
*/
size_t index() { return static_cast<size_t>(storage_ & kTypeMask); }
/**
* Returns the pointer stored in the PointerVariant, if the type matches the
* first type. If the stored type does not match the first type, an exception
* will be thrown.
*/
inline FirstType* get(folly::tag_t<FirstType>) {
ensureCorrectType(false /* secondType */);
return reinterpret_cast<FirstType*>(storage_ & kPointerMask);
}
/**
* Returns the pointer stored in the PointerVariant, if the type matches the
* second type. If the stored type does not match the second type, an
* exception will be thrown.
*/
inline SecondType* get(folly::tag_t<SecondType>) {
ensureCorrectType(true /* secondType */);
return reinterpret_cast<SecondType*>(storage_ & kPointerMask);
}
/**
* Store a new pointer of type FirstType in the PointerVariant.
*/
void set(FirstType* pointer) {
storage_ = reinterpret_cast<intptr_t>(pointer);
}
/**
* Store a new pointer of type SecondType in the PointerVariant.
*/
void set(SecondType* pointer) {
storage_ = reinterpret_cast<intptr_t>(pointer) | kTypeMask;
}
private:
void ensureCorrectType(bool secondType) {
if (secondType != !!(storage_ & kTypeMask)) {
throw std::runtime_error(fmt::format(
"Incorrect type specified. Given: {}, Stored: {}",
secondType ? folly::demangle(typeid(SecondType).name())
: folly::demangle(typeid(FirstType).name()),
storage_ & kTypeMask ? folly::demangle(typeid(SecondType).name())
: folly::demangle(typeid(FirstType).name())));
}
}
static constexpr intptr_t kTypeMask = 1;
static constexpr intptr_t kPointerMask = ~kTypeMask;
intptr_t storage_;
};
} // namespace detail
} // namespace channels
} // namespace folly
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/channels/detail/PointerVariant.h>
#include <folly/portability/GTest.h>
namespace folly {
namespace channels {
namespace detail {
using namespace std::string_literals;
TEST(PointerVariantTest, Basic) {
int64_t intVal1 = 100;
int64_t intVal2 = 200;
std::string strVal1 = "100"s;
std::string strVal2 = "200"s;
PointerVariant<int64_t, std::string> var(&intVal1);
EXPECT_EQ(*var.get(folly::tag_t<int64_t>{}), 100);
var.set(&intVal2);
EXPECT_EQ(*var.get(folly::tag_t<int64_t>{}), 200);
var.set(&strVal1);
EXPECT_EQ(*var.get(folly::tag_t<std::string>{}), "100"s);
var.set(&strVal2);
EXPECT_EQ(*var.get(folly::tag_t<std::string>{}), "200"s);
}
TEST(PointerVariantTest, Get_IncorrectType) {
int64_t intVal = 100;
std::string strVal = "100"s;
PointerVariant<int64_t, std::string> var(&intVal);
EXPECT_THROW(var.get(folly::tag_t<std::string>{}), std::runtime_error);
var.set(&strVal);
EXPECT_THROW(var.get(folly::tag_t<int64_t>{}), std::runtime_error);
}
} // namespace detail
} // namespace channels
} // namespace folly
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/executors/ManualExecutor.h>
#include <folly/executors/SerialExecutor.h>
#include <folly/experimental/channels/ConsumeChannel.h>
#include <folly/experimental/channels/FanoutSender.h>
#include <folly/experimental/channels/test/ChannelTestUtil.h>
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
namespace folly {
namespace channels {
using namespace testing;
class FanoutSenderFixture : public Test {
protected:
FanoutSenderFixture() {}
using TCallback = StrictMock<MockNextCallback<int>>;
std::pair<ChannelCallbackHandle, TCallback*> processValues(
Receiver<int> receiver) {
auto callback = std::make_unique<TCallback>();
auto callbackPtr = callback.get();
auto handle = consumeChannelWithCallback(
std::move(receiver),
&executor_,
[cbk = std::move(callback)](
folly::Try<int> resultTry) mutable -> folly::coro::Task<bool> {
(*cbk)(std::move(resultTry));
co_return true;
});
return std::make_pair(std::move(handle), callbackPtr);
}
StrictMock<MockNextCallback<std::string>> createCallback() {
return StrictMock<MockNextCallback<std::string>>();
}
folly::ManualExecutor executor_;
};
TEST_F(FanoutSenderFixture, WriteValue_FanoutBroadcastsValues) {
auto fanoutSender = FanoutSender<int>();
auto [handle1, callback1] =
processValues(fanoutSender.subscribe(toVector(100)));
auto [handle2, callback2] =
processValues(fanoutSender.subscribe(toVector(200)));
EXPECT_CALL(*callback1, onValue(100));
EXPECT_CALL(*callback2, onValue(200));
EXPECT_CALL(*callback1, onValue(1));
EXPECT_CALL(*callback2, onValue(1));
EXPECT_CALL(*callback1, onValue(2));
EXPECT_CALL(*callback2, onValue(2));
EXPECT_CALL(*callback1, onClosed());
EXPECT_CALL(*callback2, onClosed());
executor_.drain();
EXPECT_TRUE(fanoutSender.anyReceivers());
fanoutSender.write(1);
fanoutSender.write(2);
executor_.drain();
std::move(fanoutSender).close();
executor_.drain();
}
TEST_F(FanoutSenderFixture, InputThrows_AllOutputReceiversGetException) {
auto fanoutSender = FanoutSender<int>();
auto [handle1, callback1] = processValues(fanoutSender.subscribe());
auto [handle2, callback2] = processValues(fanoutSender.subscribe());
EXPECT_CALL(*callback1, onValue(1));
EXPECT_CALL(*callback2, onValue(1));
EXPECT_CALL(*callback1, onRuntimeError("std::runtime_error: Error"));
EXPECT_CALL(*callback2, onRuntimeError("std::runtime_error: Error"));
executor_.drain();
EXPECT_TRUE(fanoutSender.anyReceivers());
fanoutSender.write(1);
executor_.drain();
std::move(fanoutSender).close(std::runtime_error("Error"));
executor_.drain();
}
TEST_F(FanoutSenderFixture, ReceiversCancelled) {
auto fanoutSender = FanoutSender<int>();
auto [handle1, callback1] = processValues(fanoutSender.subscribe());
auto [handle2, callback2] = processValues(fanoutSender.subscribe());
EXPECT_CALL(*callback1, onValue(1));
EXPECT_CALL(*callback2, onValue(1));
EXPECT_CALL(*callback1, onCancelled());
EXPECT_CALL(*callback2, onValue(2));
EXPECT_CALL(*callback2, onCancelled());
executor_.drain();
EXPECT_TRUE(fanoutSender.anyReceivers());
fanoutSender.write(1);
executor_.drain();
EXPECT_TRUE(fanoutSender.anyReceivers());
handle1.reset();
fanoutSender.write(2);
executor_.drain();
EXPECT_TRUE(fanoutSender.anyReceivers());
handle2.reset();
fanoutSender.write(3);
executor_.drain();
EXPECT_FALSE(fanoutSender.anyReceivers());
std::move(fanoutSender).close();
executor_.drain();
}
} // namespace channels
} // namespace folly
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment