Commit 120926cd authored by Andrew Smith's avatar Andrew Smith Committed by Facebook GitHub Bot

Channels framework

Summary: A channel is a sender and receiver pair that allows one component to send values to another. A sender and receiver pair is similar to an AsyncPipe and AsyncGenerator pair. However, unlike AsyncPipe/AsyncGenerator, senders and receivers can be used by higher level transformation abstractions that are much more memory-efficient than using AsyncGenerator directly. These higher level abstractions do not require long-lived coroutine frames to wait on incoming values.

Reviewed By: aary

Differential Revision: D29158424

fbshipit-source-id: 88c51d0f9d73677a04906197f4c44fe84ac01cdb
parent af0a489d
/*
* 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/experimental/channels/detail/ChannelBridge.h>
namespace folly {
namespace channels {
template <typename TValue>
class Receiver;
template <typename TValue>
class Sender;
namespace detail {
template <typename TValue>
ChannelBridgePtr<TValue>& senderGetBridge(Sender<TValue>& sender);
template <typename TValue>
bool receiverWait(
Receiver<TValue>& receiver, detail::IChannelCallback* receiverCallback);
template <typename TValue>
std::optional<folly::Try<TValue>> receiverGetValue(Receiver<TValue>& receiver);
template <typename TValue>
std::pair<detail::ChannelBridgePtr<TValue>, detail::ReceiverQueue<TValue>>
receiverUnbuffer(Receiver<TValue>&& receiver);
} // 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.
*/
#pragma once
#include <folly/CancellationToken.h>
#include <folly/Synchronized.h>
#include <folly/experimental/channels/detail/ChannelBridge.h>
#include <folly/experimental/coro/Coroutine.h>
namespace folly {
namespace channels {
namespace detail {
template <typename TValue>
ChannelBridgePtr<TValue>& senderGetBridge(Sender<TValue>& sender) {
return sender.bridge_;
}
template <typename TValue>
bool receiverWait(
Receiver<TValue>& receiver, detail::IChannelCallback* callback) {
if (!receiver.buffer_.empty()) {
return false;
}
return receiver.bridge_->receiverWait(callback);
}
template <typename TValue>
std::optional<folly::Try<TValue>> receiverGetValue(Receiver<TValue>& receiver) {
if (receiver.buffer_.empty()) {
receiver.buffer_ = receiver.bridge_->receiverGetValues();
if (receiver.buffer_.empty()) {
return std::nullopt;
}
}
auto result = std::move(receiver.buffer_.front());
receiver.buffer_.pop();
return result;
}
template <typename TValue>
std::pair<detail::ChannelBridgePtr<TValue>, detail::ReceiverQueue<TValue>>
receiverUnbuffer(Receiver<TValue>&& receiver) {
return std::make_pair(
std::move(receiver.bridge_), std::move(receiver.buffer_));
}
} // namespace detail
template <typename TValue>
class Receiver<TValue>::Waiter : public detail::IChannelCallback {
public:
Waiter(Receiver<TValue>* receiver, folly::CancellationToken cancelToken)
: state_(State{.receiver = receiver}),
cancelCallback_(makeCancellationCallback(std::move(cancelToken))) {}
bool await_ready() const noexcept {
// We are ready immediately if the receiver is either cancelled or closed.
return state_.withRLock(
[&](const State& state) { return state.cancelled || !state.receiver; });
}
bool await_suspend(folly::coro::coroutine_handle<> awaitingCoroutine) {
return state_.withWLock([&](State& state) {
if (state.cancelled || !state.receiver ||
!receiverWait(*state.receiver, this)) {
// We will not suspend at all if the receiver is either cancelled or
// closed.
return false;
}
state.awaitingCoroutine = awaitingCoroutine;
return true;
});
}
std::optional<TValue> await_resume() {
auto result = getResult();
if (!result.hasValue() && !result.hasException()) {
return std::nullopt;
}
return std::move(result.value());
}
folly::Try<TValue> await_resume_try() { return getResult(); }
protected:
struct State {
Receiver<TValue>* receiver;
folly::coro::coroutine_handle<> awaitingCoroutine;
bool cancelled{false};
};
std::unique_ptr<folly::CancellationCallback> makeCancellationCallback(
folly::CancellationToken cancelToken) {
if (!cancelToken.canBeCancelled()) {
return nullptr;
}
return std::make_unique<folly::CancellationCallback>(
std::move(cancelToken), [this] {
auto receiver = state_.withWLock([&](State& state) {
state.cancelled = true;
return std::exchange(state.receiver, nullptr);
});
if (receiver) {
std::move(*receiver).cancel();
}
});
}
void consume(detail::ChannelBridgeBase*) override { resume(); }
void canceled(detail::ChannelBridgeBase*) override { resume(); }
void resume() {
auto awaitingCoroutine = state_.withWLock([&](State& state) {
return std::exchange(state.awaitingCoroutine, nullptr);
});
awaitingCoroutine.resume();
}
folly::Try<TValue> getResult() {
cancelCallback_.reset();
return state_.withWLock([&](State& state) {
if (state.cancelled) {
return folly::Try<TValue>(
folly::make_exception_wrapper<folly::OperationCancelled>());
}
if (!state.receiver) {
return folly::Try<TValue>();
}
auto result =
std::move(detail::receiverGetValue(*state.receiver).value());
if (!result.hasValue()) {
std::move(*state.receiver).cancel();
state.receiver = nullptr;
}
return result;
});
}
folly::Synchronized<State> state_;
std::unique_ptr<folly::CancellationCallback> cancelCallback_;
};
template <typename TValue>
struct Receiver<TValue>::NextSemiAwaitable {
public:
explicit NextSemiAwaitable(
Receiver<TValue>* receiver,
std::optional<folly::CancellationToken> cancelToken = std::nullopt)
: receiver_(receiver), cancelToken_(std::move(cancelToken)) {}
[[nodiscard]] Waiter operator co_await() {
return Waiter(receiver_, cancelToken_.value_or(folly::CancellationToken()));
}
friend NextSemiAwaitable co_withCancellation(
folly::CancellationToken cancelToken, NextSemiAwaitable&& awaitable) {
if (awaitable.cancelToken_.has_value()) {
return std::move(awaitable);
}
return NextSemiAwaitable(awaitable.receiver_, std::move(cancelToken));
}
private:
Receiver<TValue>* receiver_;
std::optional<folly::CancellationToken> cancelToken_;
};
} // 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/experimental/channels/Channel-fwd.h>
#include <folly/experimental/channels/detail/ChannelBridge.h>
namespace folly {
namespace channels {
/*
* A channel is a sender and receiver pair that allows one component to send
* values to another. A sender and receiver pair is similar to an AsyncPipe and
* AsyncGenerator pair. However, unlike AsyncPipe/AsyncGenerator, senders and
* receivers can be used by memory-efficient higher level transformation
* abstractions.
*
* Typical usage:
* auto [receiver, sender] = Channel<T>::create();
* sender.write(val1);
* auto val2 = co_await receiver.next();
*/
template <typename TValue>
class Channel {
public:
/**
* Creates a new channel with a sender/receiver pair. The channel will be
* closed if the sender is destroyed, and will be cancelled if the receiver is
* destroyed.
*/
static std::pair<Receiver<TValue>, Sender<TValue>> create() {
auto senderBridge = detail::ChannelBridge<TValue>::create();
auto receiverBridge = senderBridge->copy();
return std::make_pair(
Receiver<TValue>(std::move(receiverBridge)),
Sender<TValue>(std::move(senderBridge)));
}
};
/**
* A sender sends values to be consumed by a receiver.
*/
template <typename TValue>
class Sender {
public:
friend Channel<TValue>;
using ValueType = TValue;
Sender(Sender&& other) noexcept : bridge_(std::move(other.bridge_)) {}
Sender& operator=(Sender&& other) noexcept {
if (this == &other) {
return *this;
}
if (bridge_) {
std::move(*this).close();
}
bridge_ = std::move(other.bridge_);
return *this;
}
~Sender() {
if (bridge_) {
std::move(*this).close();
}
}
/**
* Returns whether or not this sender instance is valid. This will return
* false if the sender was closed or moved away.
*/
explicit operator bool() const { return bridge_ != nullptr; }
/**
* Writes a value into the pipe.
*/
template <typename U = TValue>
void write(U&& element) {
if (!bridge_->isSenderClosed()) {
bridge_->senderPush(std::forward<U>(element));
}
}
/**
* Closes the pipe without an exception.
*/
void close() && {
if (!bridge_->isSenderClosed()) {
bridge_->senderClose();
}
bridge_ = nullptr;
}
/**
* Closes the pipe with an exception.
*/
void close(folly::exception_wrapper exception) && {
if (!bridge_->isSenderClosed()) {
bridge_->senderClose(std::move(exception));
}
bridge_ = nullptr;
}
/**
* Returns whether or not the corresponding receiver has been cancelled or
* destroyed.
*/
bool isReceiverCancelled() {
if (bridge_->isSenderClosed()) {
return true;
}
auto values = bridge_->senderGetValues();
if (!values.empty()) {
bridge_->senderClose();
return true;
}
return false;
}
private:
friend detail::ChannelBridgePtr<TValue>& detail::senderGetBridge<>(
Sender<TValue>&);
explicit Sender(detail::ChannelBridgePtr<TValue> bridge)
: bridge_(std::move(bridge)) {}
detail::ChannelBridgePtr<TValue> bridge_;
};
/**
* A receiver that receives values sent by a sender. There are two ways that a
* receiver can be consumed directly:
*
* 1. Call co_await receiver.next() to get the next value. See the docstring of
* next() for more details.
*
* 2. Call consumeChannelWithCallback to get a callback when each value comes
* in. See ConsumeChannel.h for more details. This uses less memory than
* #1, as it only needs to allocate coroutine frames when processing values
* (rather than always having such frames allocated when waiting for values).
*
* A receiver may also be passed to framework primitives that consume the
* receiver (such as transform or merge). Like #2, these primitives do not
* require coroutine frames to be allocated when waiting for values to come in.
*/
template <typename TValue>
class Receiver {
class Waiter;
struct NextSemiAwaitable;
public:
friend Channel<TValue>;
using ValueType = TValue;
Receiver() {}
Receiver(Receiver&& other) noexcept
: bridge_(std::move(other.bridge_)), buffer_(std::move(other.buffer_)) {}
Receiver& operator=(Receiver&& other) noexcept {
if (this == &other) {
return *this;
}
if (bridge_ != nullptr) {
std::move(*this).cancel();
}
bridge_ = std::move(other.bridge_);
buffer_ = std::move(other.buffer_);
return *this;
}
~Receiver() {
if (bridge_ != nullptr) {
std::move(*this).cancel();
}
}
/**
* Returns whether or not this receiver instance is valid. This will return
* false if the receiver was cancelled or moved away.
*/
explicit operator bool() const { return bridge_ != nullptr; }
/**
* Returns the next value sent by a sender. The behavior is the same as the
* behavior of next() on folly::coro::AsyncGenerator<TValue>.
*
* If consumed directly with co_await, next() will return an std::optional:
*
* std::optional<TValue> value = co_await receiver.next();
*
* - If a value is sent, the std::optional will contain the value.
* - If the channel is closed by the sender with no exception, the optional
* will be empty.
* - If the channel is closed by the sender with an exception, next() will
* throw the exception.
* - If the next() call was cancelled, next() will throw an exception of
* type folly::OperationCancelled.
*
* If consumed with folly::coro::co_awaitTry, this will return a folly::Try:
*
* folly::Try<TValue> value = co_await folly::coro::co_awaitTry(
* receiver.next());
*
* - If a value is sent, the folly::Try will contain the value.
* - If the channel is closed by the sender with no exception, the try will
* be empty (with no value or exception).
* - If the channel is closed by the sender with an exception, the try will
* contain the exception.
* - If the next() call was cancelled, the try will contain an exception of
* type folly::OperationCancelled.
*/
NextSemiAwaitable next() { return NextSemiAwaitable(*this ? this : nullptr); }
/**
* Cancels this receiver. If the receiver is currently being consumed, the
* consumer will receive a folly::OperationCancelled exception.
*/
void cancel() && {
bridge_->receiverCancel();
bridge_ = nullptr;
buffer_.clear();
}
private:
explicit Receiver(detail::ChannelBridgePtr<TValue> bridge)
: bridge_(std::move(bridge)) {}
friend bool detail::receiverWait<>(
Receiver<TValue>&, detail::IChannelCallback*);
friend std::optional<folly::Try<TValue>> detail::receiverGetValue<>(
Receiver<TValue>&);
friend std::
pair<detail::ChannelBridgePtr<TValue>, detail::ReceiverQueue<TValue>>
detail::receiverUnbuffer<>(Receiver<TValue>&& receiver);
detail::ChannelBridgePtr<TValue> bridge_;
detail::ReceiverQueue<TValue> buffer_;
};
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/Channel-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 <folly/IntrusiveList.h>
#include <folly/ScopeGuard.h>
#include <folly/experimental/channels/Channel.h>
namespace folly {
namespace channels {
namespace detail {
class ChannelCallbackProcessor : public IChannelCallback {
public:
virtual void onHandleDestroyed() = 0;
};
} // namespace detail
/**
* A callback handle for a consumption operation on a channel. The consumption
* operation will be cancelled when this handle is destroyed.
*/
class ChannelCallbackHandle {
public:
ChannelCallbackHandle() : processor_(nullptr) {}
explicit ChannelCallbackHandle(detail::ChannelCallbackProcessor* processor)
: processor_(processor) {}
~ChannelCallbackHandle() {
if (processor_) {
processor_->onHandleDestroyed();
}
}
ChannelCallbackHandle(ChannelCallbackHandle&& other) noexcept
: processor_(std::exchange(other.processor_, nullptr)) {}
ChannelCallbackHandle& operator=(ChannelCallbackHandle&& other) {
if (&other == this) {
return *this;
}
reset();
processor_ = std::exchange(other.processor_, nullptr);
return *this;
}
void reset() {
if (processor_) {
processor_->onHandleDestroyed();
processor_ = nullptr;
}
}
private:
detail::ChannelCallbackProcessor* processor_;
};
namespace detail {
/**
* A wrapper around a ChannelCallbackHandle that belongs to an intrusive linked
* list. When the holder is destroyed, the object will automatically be unlinked
* from the linked list that it is in (if any).
*/
struct ChannelCallbackHandleHolder {
explicit ChannelCallbackHandleHolder(ChannelCallbackHandle _handle)
: handle(std::move(_handle)) {}
ChannelCallbackHandleHolder(ChannelCallbackHandleHolder&& other) noexcept
: handle(std::move(other.handle)) {
hook.swap_nodes(other.hook);
}
ChannelCallbackHandleHolder& operator=(
ChannelCallbackHandleHolder&& other) noexcept {
if (&other == this) {
return *this;
}
handle = std::move(other.handle);
hook.unlink();
hook.swap_nodes(other.hook);
return *this;
}
void requestCancellation() { handle.reset(); }
ChannelCallbackHandle handle;
folly::IntrusiveListHook hook;
};
template <typename TValue, typename OnNextFunc>
class ChannelCallbackProcessorImplWithList;
} // namespace detail
/**
* A list of channel callback handles. When consumeChannelWithCallback is
* invoked with a list, a cancellation handle is automatically added to the list
* for the consumption operation. Similarly, when a consumption operation is
* completed, the handle is automatically removed from the lists.
*
* If the list still has any cancellation handles remaining when the list is
* destroyed, cancellation is triggered for each handle in the list.
*
* This list is not thread safe.
*/
class ChannelCallbackHandleList {
public:
ChannelCallbackHandleList() {}
ChannelCallbackHandleList(ChannelCallbackHandleList&& other) noexcept {
holders_.swap(other.holders_);
}
ChannelCallbackHandleList& operator=(
ChannelCallbackHandleList&& other) noexcept {
if (&other == this) {
return *this;
}
holders_.swap(other.holders_);
return *this;
}
~ChannelCallbackHandleList() { clear(); }
void clear() {
for (auto& holder : holders_) {
holder.requestCancellation();
}
holders_.clear();
}
private:
template <typename TValue, typename OnNextFunc>
friend class detail::ChannelCallbackProcessorImplWithList;
void add(detail::ChannelCallbackHandleHolder& holder) {
holders_.push_back(holder);
}
using ChannelCallbackHandleListImpl = folly::IntrusiveList<
detail::ChannelCallbackHandleHolder,
&detail::ChannelCallbackHandleHolder::hook>;
ChannelCallbackHandleListImpl holders_;
};
} // namespace channels
} // namespace folly
This diff is collapsed.
/*
* 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/Executor.h>
#include <folly/IntrusiveList.h>
#include <folly/experimental/channels/Channel.h>
#include <folly/experimental/channels/ChannelCallbackHandle.h>
#include <folly/experimental/coro/Task.h>
namespace folly {
namespace channels {
/**
* This function takes a Receiver, and consumes updates from that receiver with
* a callback.
*
* This function returns a ChannelCallbackHandle. On destruction of this handle,
* the callback will receive a try containing an exception of type
* folly::OperationCancelled. If an active callback is running at the time the
* cancellation request is received, cancellation will be requested on the
* ambient cancellation token of the callback.
*
* The callback is run for each received value on the given executor. A try
* is passed to the callback with the result:
*
* - If a value is sent, the folly::Try will contain the value.
* - If the channel is closed by the sender with no exception, the try will
* be empty (with no value or exception).
* - If the channel is closed by the sender with an exception, the try will
* contain the exception.
* - If the channel was cancelled (by the destruction of the returned
* handle), the try will contain an exception of type
* folly::OperationCancelled.
*
* If the callback returns false or throws a folly::OperationCancelled
* exception, the channel will be cancelled and no further values will be
* received.
*/
template <
typename TReceiver,
typename OnNextFunc,
typename TValue = typename TReceiver::ValueType,
std::enable_if_t<
std::is_constructible_v<
folly::Function<folly::coro::Task<bool>(folly::Try<TValue>)>,
OnNextFunc>,
int> = 0>
ChannelCallbackHandle consumeChannelWithCallback(
TReceiver receiver,
folly::Executor::KeepAlive<> executor,
OnNextFunc onNext);
/**
* This overload is similar to the previous overload. However, unlike the
* previous overload (which returns a handle that allows cancellation of that
* specific consumption operation), this overload accepts a list of handles and
* returns void. This overload will immediately add a handle to the list, and
* will remove itself from the list if the channel is closed or cancelled.
*
* When the passed-in handle list is destroyed by the caller, all handles
* remaining in the list will trigger cancellation on their corresponding
* consumption operations.
*
* Note that ChannelCallbackHandleList is not thread safe. This means that all
* operations using a particular list (including start and cancellation) need to
* run on the same serial executor passed to this function.
*/
template <
typename TReceiver,
typename OnNextFunc,
typename TValue = typename TReceiver::ValueType,
std::enable_if_t<
std::is_constructible_v<
folly::Function<folly::coro::Task<bool>(folly::Try<TValue>)>,
OnNextFunc>,
int> = 0>
void consumeChannelWithCallback(
TReceiver receiver,
folly::Executor::KeepAlive<> executor,
OnNextFunc onNext,
ChannelCallbackHandleList& callbackHandles);
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/ConsumeChannel-inl.h>
This diff is collapsed.
/*
* 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/executors/SequencedExecutor.h>
#include <folly/experimental/channels/Channel.h>
namespace folly {
namespace channels {
namespace detail {
template <typename TValue>
class IFanoutChannelProcessor;
}
/**
* A fanout channel allows one to fan out updates from a single input receiver
* to multiple output receivers.
*
* When a new output receiver is added, an optional function will be run that
* computes a set of initial values. These initial values will only be sent to
* the new receiver.
*
* Example:
*
* // Function that returns a receiver:
* Receiver<int> getInputReceiver();
*
* // Function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* auto fanoutChannel = createFanoutChannel(getReceiver(), getExecutor());
* auto receiver1 = fanoutChannel.newReceiver();
* auto receiver2 = fanoutChannel.newReceiver();
* auto receiver3 = fanoutChannel.newReceiver([]{ return {1, 2, 3}; });
* std::move(fanoutChannel).close();
*/
template <typename TValue>
class FanoutChannel {
using TProcessor = detail::IFanoutChannelProcessor<TValue>;
public:
explicit FanoutChannel(TProcessor* processor);
FanoutChannel(FanoutChannel&& other) noexcept;
FanoutChannel& operator=(FanoutChannel&& other) noexcept;
~FanoutChannel();
/**
* Returns whether this FanoutChannel is a valid object.
*/
explicit operator bool() const;
/**
* Returns a new output receiver that will receive all values from the input
* receiver. If a getInitialValues parameter is provided, it will be executed
* to determine the set of initial values that will (only) go to the new input
* receiver.
*/
Receiver<TValue> getNewReceiver(
folly::Function<std::vector<TValue>()> getInitialValues = {});
/**
* Returns whether this fanout channel has any output receivers.
*/
bool anyReceivers();
/**
* Closes the fanout channel.
*/
void close(std::optional<folly::exception_wrapper> ex = std::nullopt) &&;
private:
TProcessor* processor_;
};
/**
* Creates a new fanout channel that fans out updates from an input receiver.
*/
template <typename TReceiver, typename TValue = typename TReceiver::ValueType>
FanoutChannel<TValue> createFanoutChannel(
TReceiver inputReceiver,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor);
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/FanoutChannel-inl.h>
This diff is collapsed.
/*
* 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/executors/SequencedExecutor.h>
#include <folly/experimental/channels/Channel.h>
namespace folly {
namespace channels {
/**
* Merge takes a list of receivers, and returns a new receiver that receives
* all updates from all input receivers. If any input receiver closes with
* an exception, the exception is forwarded and the channel is closed. If any
* input receiver closes without an exception, the channel continues to merge
* values from the other input receivers until all input receivers are closed.
*
* @param inputReceivers: The collection of input receivers to merge.
*
* @param executor: A SequencedExecutor used to merge input values.
*
* Example:
*
* // Example function that returns a list of receivers
* std::vector<Receiver<int>> getReceivers();
*
* // Example function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* Receiver<int> mergedReceiver = merge(getReceivers(), getExecutor());
*/
template <typename TReceiver, typename TValue = typename TReceiver::ValueType>
Receiver<TValue> merge(
std::vector<TReceiver> inputReceivers,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor);
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/Merge-inl.h>
This diff is collapsed.
/*
* 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/executors/SequencedExecutor.h>
#include <folly/experimental/channels/Channel.h>
namespace folly {
namespace channels {
namespace detail {
template <typename TValue, typename TSubscriptionId>
class IMergeChannelProcessor;
}
/**
* A merge channel allows one to merge multiple receivers into a single output
* receiver. The set of receivers being merged can be changed at runtime. Each
* receiver is added with a subscription ID, that can be used to remove the
* receiver at a later point.
*
* Example:
*
* // Example function that returns a receiver for a given entity:
* Receiver<int> subscribe(std::string entity);
*
* // Example function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* auto [outputReceiver, mergeChannel]
* = createMergeChannel<int, std::string>(getExecutor());
* mergeChannel.addNewReceiver("abc", subscribe("abc"));
* mergeChannel.addNewReceiver("def", subscribe("def"));
* mergeChannel.removeReceiver("abc");
* std::move(mergeChannel).close();
*/
template <typename TValue, typename TSubscriptionId>
class MergeChannel {
using TProcessor = detail::IMergeChannelProcessor<TValue, TSubscriptionId>;
public:
explicit MergeChannel(
detail::IMergeChannelProcessor<TValue, TSubscriptionId>* processor);
MergeChannel(MergeChannel&& other) noexcept;
MergeChannel& operator=(MergeChannel&& other) noexcept;
~MergeChannel();
/**
* Returns whether this MergeChannel is a valid object.
*/
explicit operator bool() const;
/**
* Adds a new receiver to be merged, along with a given subscription ID. If
* the subscription ID matches the ID of an existing receiver, that existing
* receiver is replaced with the new one (and changes to the old receiver will
* no longer be merged). An added receiver can later be removed by passing the
* subscription ID to removeReceiver.
*/
template <typename TReceiver>
void addNewReceiver(TSubscriptionId subscriptionId, TReceiver receiver);
/**
* Removes the receiver added with the given subscription ID. The receiver
* will be asynchronously removed, so the consumer may still receive some
* values from this receiver after this call.
*/
void removeReceiver(TSubscriptionId subscriptionId);
/**
* Closes the merge channel.
*/
void close(std::optional<folly::exception_wrapper> ex = std::nullopt) &&;
private:
TProcessor* processor_;
};
/**
* Creates a new merge channel.
*
* @param executor: The SequencedExecutor to use for merging values.
*/
template <typename TValue, typename TSubscriptionId>
std::pair<Receiver<TValue>, MergeChannel<TValue, TSubscriptionId>>
createMergeChannel(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor);
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/MergeChannel-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/CancellationToken.h>
#include <folly/experimental/channels/Channel.h>
#include <folly/experimental/channels/ConsumeChannel.h>
#include <folly/experimental/channels/Producer.h>
#include <folly/experimental/coro/Task.h>
namespace folly {
namespace channels {
template <typename TValue>
Producer<TValue>::KeepAlive::KeepAlive(Producer<TValue>* ptr) : ptr_(ptr) {}
template <typename TValue>
Producer<TValue>::KeepAlive::~KeepAlive() {
if (ptr_ && --ptr_->refCount_ == 0) {
auto deleteTask =
folly::coro::co_invoke([ptr = ptr_]() -> folly::coro::Task<void> {
delete ptr;
co_return;
});
std::move(deleteTask).scheduleOn(ptr_->getExecutor()).start();
}
}
template <typename TValue>
Producer<TValue>::KeepAlive::KeepAlive(
Producer<TValue>::KeepAlive&& other) noexcept
: ptr_(std::exchange(other.ptr_, nullptr)) {}
template <typename TValue>
typename Producer<TValue>::KeepAlive& Producer<TValue>::KeepAlive::operator=(
Producer<TValue>::KeepAlive&& other) noexcept {
if (&other == this) {
return *this;
}
ptr_ = std::exchange(other.ptr_, nullptr);
return *this;
}
template <typename TValue>
Producer<TValue>::Producer(
Sender<TValue> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor)
: sender_(std::move(detail::senderGetBridge(sender))),
executor_(std::move(executor)) {
CHECK(sender_->senderWait(this));
}
template <typename TValue>
void Producer<TValue>::write(TValue value) {
executor_->add([this, value = std::move(value)]() mutable {
sender_->senderPush(std::move(value));
});
}
template <typename TValue>
void Producer<TValue>::close(std::optional<folly::exception_wrapper> ex) {
executor_->add([this, ex = std::move(ex)]() mutable {
if (ex.has_value()) {
sender_->senderClose(std::move(ex.value()));
} else {
sender_->senderClose();
}
});
}
template <typename TValue>
bool Producer<TValue>::isClosed() {
return sender_->isSenderClosed();
}
template <typename TValue>
folly::Executor::KeepAlive<folly::SequencedExecutor>
Producer<TValue>::getExecutor() {
return executor_;
}
template <typename TValue>
typename Producer<TValue>::KeepAlive Producer<TValue>::getKeepAlive() {
refCount_.fetch_add(1, std::memory_order_relaxed);
return KeepAlive(this);
}
template <typename TValue>
void Producer<TValue>::consume(detail::ChannelBridgeBase*) {
onClosed().scheduleOn(getExecutor()).start([=](auto) {
// Decrement ref count
KeepAlive(this);
});
}
template <typename TValue>
void Producer<TValue>::canceled(detail::ChannelBridgeBase* bridge) {
consume(bridge);
}
namespace detail {
template <typename TProducer>
class ProducerImpl : public TProducer {
template <typename ProducerType, typename... Args>
friend Receiver<typename ProducerType::ValueType> makeProducer(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
Args&&... args);
public:
using TProducer::TProducer;
private:
void ensureMakeProducerUsedForCreation() override {}
};
} // namespace detail
template <typename TProducer, typename... Args>
Receiver<typename TProducer::ValueType> makeProducer(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
Args&&... args) {
using TValue = typename TProducer::ValueType;
auto [receiver, sender] = Channel<TValue>::create();
new detail::ProducerImpl<TProducer>(
std::move(sender), std::move(executor), std::forward<Args>(args)...);
return std::move(receiver);
}
} // 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/executors/SequencedExecutor.h>
#include <folly/experimental/channels/ChannelCallbackHandle.h>
#include <folly/experimental/coro/Task.h>
namespace folly {
namespace channels {
/**
* A Producer is a base class for an object that produces a channel. The
* subclass can call write to write a new value to the channel, and close to
* close the channel. It is a useful way to generate output values for a
* receiver, without having to keep alive an extraneous object that produces
* those values.
*
* When the consumer of the channel stops consuming, the onClosed function will
* be called. The subclass should cancel any ongoing work in this function.
* After onCancelled is called, the object will be deleted once the last
* outstanding KeepAlive is destroyed.
*
* Example:
* // Function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* // Function that returns output values
* std::vector<int> getLatestOutputValues();
*
* // Example producer implementation
* class PollingProducer : public Producer<int> {
* public:
* PollingProducer(
* Sender<int> sender,
* folly::Executor::KeepAlive<folly::SequencedExecutor> executor)
* : Producer<int>(std::move(sender), std::move(executor)) {
* // Start polling for values.
* folly::coro::co_withCancellation(
* cancelSource_.getToken(),
* [=, keepAlive = getKeepAlive()]() {
* return pollForOutputValues();
* })
* .scheduleOn(getExecutor())
* .start();
* }
*
* folly::coro::Task<void> onClosed() override {
* // The consumer has stopped consuming our values. Stop polling.
* cancelSource_.requestCancellation();
* }
*
* private:
* folly::coro::Task<void> pollForOutputValues() {
* auto cancelToken = co_await folly::coro::co_current_cancellation_token;
* while (!cancelToken.isCancellationRequested()) {
* auto outputValues = getLatestOutputValues();
* for (auto& outputValue : outputValues) {
* write(std::move(outputValue));
* }
* }
* co_await folly::coro::sleep(std::chrono::seconds(1));
* }
*
* folly::CancellationSource cancelSource_;
* };
*
* // Producer usage
* Receiver<int> receiver = makeProducer<PollingProducer>(getExecutor());
*/
template <typename TValue>
class Producer : public detail::IChannelCallback {
public:
using ValueType = TValue;
protected:
/**
* This object will ensure that the corresponding Producer that created it
* will not be destroyed.
*/
class KeepAlive {
public:
~KeepAlive();
KeepAlive(KeepAlive&&) noexcept;
KeepAlive& operator=(KeepAlive&&) noexcept;
private:
friend class Producer<TValue>;
explicit KeepAlive(Producer<TValue>* ptr);
Producer<TValue>* ptr_;
};
Producer(
Sender<TValue> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor);
virtual ~Producer() override = default;
/**
* Writes a value into the channel.
*/
void write(TValue value);
/**
* Closes the channel.
*/
void close(std::optional<folly::exception_wrapper> ex = std::nullopt);
/**
* Returns whether or not this producer is closed or cancelled.
*/
bool isClosed();
/**
* Returns the executor used for this producer.
*/
folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
/**
* Returns a KeepAlive object. This object will not be destroyed before all
* KeepAlive objects are destroyed.
*/
KeepAlive getKeepAlive();
/**
* Called when the corresponding receiver is cancelled, or the sender is
* closed.
*/
virtual folly::coro::Task<void> onClosed() { co_return; }
/**
* If you get an error that this function is not implemented, do not
* implement it. Instead, create your object with makeProducer
* below.
*/
virtual void ensureMakeProducerUsedForCreation() = 0;
private:
template <typename TProducer, typename... Args>
friend Receiver<typename TProducer::ValueType> makeProducer(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
Args&&... args);
void consume(detail::ChannelBridgeBase* bridge) override;
void canceled(detail::ChannelBridgeBase* bridge) override;
detail::ChannelBridgePtr<TValue> sender_;
folly::Executor::KeepAlive<folly::SequencedExecutor> executor_;
std::atomic<int> refCount_{1};
};
/**
* Creates a new object that extends the Producer class, and returns a receiver.
* The receiver will receive any values produced by the producer. See the
* description of the Producer class for information on how to implement a
* producer.
*/
template <typename TProducer, typename... Args>
Receiver<typename TProducer::ValueType> makeProducer(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
Args&&... args);
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/Producer-inl.h>
This diff is collapsed.
/*
* 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/executors/SequencedExecutor.h>
#include <folly/experimental/channels/Channel.h>
namespace folly {
namespace channels {
/**
* Returns an output receiver that applies a given transformation function to
* each value from an input receiver.
*
* The TransformValue function takes a Try<TInputValue>, and returns a
* folly::coro::AsyncGenerator<TOutputValue>.
*
* - If the TransformValue function yields one or more output values, those
* output values are sent to the output receiver.
*
* - If the TransformValue function throws an OnClosedException, the output
* receiver is closed (without an exception).
*
* - If the TransformValue function throws any other type of exception, the
* output receiver is closed with that exception.
*
* If the input receiver was closed, the TransformValue function is called with
* a Try containing an exception (either OnClosedException if the input receiver
* was closed without an exception, or the closure exception if the input
* receiver was closed with an exception). In this case, regardless of what the
* TransformValue function returns, the output receiver will be closed
* (potentially after receiving the last output values the TransformValue
* function returned, if any).
*
* @param inputReceiver: The input receiver.
*
* @param executor: A folly::SequencedExecutor used to transform the values.
*
* @param transformValue: A function as described above.
*
* Example:
*
* // Function that returns a receiver
* Receiver<int> getInputReceiver();
*
* // Function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* Receiver<std::string> outputReceiver = transform(
* getInputReceiver(),
* getExecutor(),
* [](folly::Try<int> try) -> folly::coro::AsyncGenerator<std::string&&> {
* co_yield folly::to<std::string>(try.value());
* });
*/
template <
typename TReceiver,
typename TransformValueFunc,
typename TInputValue = typename TReceiver::ValueType,
typename TOutputValue = typename folly::invoke_result_t<
TransformValueFunc,
folly::Try<TInputValue>>::value_type>
Receiver<TOutputValue> transform(
TReceiver inputReceiver,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
TransformValueFunc transformValue);
/**
* This function is similar to the above transform function. However, instead of
* taking a single input receiver, it takes an initialization function that
* returns a std::pair<std::vector<TOutputValue>, Receiver<TInputValue>>.
*
* - If the InitializeTransform function returns successfully, the vector's
* output values will be immediately sent to the output receiver. The input
* receiver is then processed as described in the transform function's
* documentation, until it is closed (without an exception). At that point,
* the InitializationTransform is re-run, and the transform begins anew.
*
* - If the InitializeTransform function throws an OnClosedException, the
* output receiver is closed (with no exception).
*
* - If the InitializeTransform function throws any other type of exception,
* the output receiver is closed with that exception.
*
* - If the TransformValue function throws any exception other than
* OnClosedException, the output receiver is closed with that exception.
*
* @param executor: A folly::SequencedExecutor used to transform the values.
*
* @param initializeTransform: The InitializeTransform function as described
* above.
*
* @param transformValue: A function as described above.
*
* Example:
*
* // Function that returns a receiver
* Receiver<int> getInputReceiver();
*
* // Function that returns an executor
* folly::Executor::KeepAlive<folly::SequencedExecutor> getExecutor();
*
* Receiver<std::string> outputReceiver = transform(
* getExecutor(),
* []() -> folly::coro::Task<
* std::pair<std::vector<std::string>, Receiver<int>> {
* co_return std::make_pair(
* std::vector<std::string>({"Initialized"}),
* getInputReceiver());
* },
* [](folly::Try<int> try) -> folly::coro::AsyncGenerator<std::string&&> {
* co_yield folly::to<std::string>(try.value());
* });
*
*/
template <
typename InitializeTransformFunc,
typename TransformValueFunc,
typename TReceiver = typename folly::invoke_result_t<
InitializeTransformFunc>::StorageType::second_type,
typename TInputValue = typename TReceiver::ValueType,
typename TOutputValue = typename folly::invoke_result_t<
TransformValueFunc,
folly::Try<TInputValue>>::value_type>
Receiver<TOutputValue> resumableTransform(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
InitializeTransformFunc initializeTransform,
TransformValueFunc transformValue);
/**
* An OnClosedException passed to a transform callback indicates that the input
* channel was closed. An OnClosedException can also be thrown by a transform
* callback, which will close the output channel.
*/
struct OnClosedException : public std::exception {
const char* what() const noexcept override {
return "A transform has closed the channel.";
}
};
} // namespace channels
} // namespace folly
#include <folly/experimental/channels/Transform-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 <folly/Try.h>
#include <folly/experimental/channels/detail/AtomicQueue.h>
namespace folly {
namespace channels {
namespace detail {
class ChannelBridgeBase {};
class IChannelCallback {
public:
virtual ~IChannelCallback() = default;
virtual void consume(ChannelBridgeBase* bridge) = 0;
virtual void canceled(ChannelBridgeBase* bridge) = 0;
};
using SenderQueue = typename folly::channels::detail::Queue<folly::Unit>;
template <typename TValue>
using ReceiverQueue =
typename folly::channels::detail::Queue<folly::Try<TValue>>;
template <typename TValue>
class ChannelBridge : public ChannelBridgeBase {
public:
struct Deleter {
void operator()(ChannelBridge<TValue>* ptr) { ptr->decref(); }
};
using Ptr = std::unique_ptr<ChannelBridge<TValue>, Deleter>;
static Ptr create() { return Ptr(new ChannelBridge<TValue>()); }
Ptr copy() {
auto refCount = refCount_.fetch_add(1, std::memory_order_relaxed);
DCHECK(refCount > 0);
return Ptr(this);
}
// These should only be called from the sender thread
template <typename U = TValue>
void senderPush(U&& value) {
receiverQueue_.push(
folly::Try<TValue>(std::forward<U>(value)),
static_cast<ChannelBridgeBase*>(this));
}
bool senderWait(IChannelCallback* callback) {
return senderQueue_.wait(callback, static_cast<ChannelBridgeBase*>(this));
}
IChannelCallback* cancelSenderWait() { return senderQueue_.cancelCallback(); }
void senderClose() {
if (!isSenderClosed()) {
receiverQueue_.push(
folly::Try<TValue>(), static_cast<ChannelBridgeBase*>(this));
senderQueue_.close(static_cast<ChannelBridgeBase*>(this));
}
}
void senderClose(folly::exception_wrapper ex) {
if (!isSenderClosed()) {
receiverQueue_.push(
folly::Try<TValue>(std::move(ex)),
static_cast<ChannelBridgeBase*>(this));
senderQueue_.close(static_cast<ChannelBridgeBase*>(this));
}
}
bool isSenderClosed() { return senderQueue_.isClosed(); }
SenderQueue senderGetValues() {
return senderQueue_.getMessages(static_cast<ChannelBridgeBase*>(this));
}
// These should only be called from the receiver thread
void receiverCancel() {
if (!isReceiverCancelled()) {
senderQueue_.push(folly::Unit(), static_cast<ChannelBridgeBase*>(this));
receiverQueue_.close(static_cast<ChannelBridgeBase*>(this));
}
}
bool isReceiverCancelled() { return receiverQueue_.isClosed(); }
bool receiverWait(IChannelCallback* callback) {
return receiverQueue_.wait(callback, static_cast<ChannelBridgeBase*>(this));
}
IChannelCallback* cancelReceiverWait() {
return receiverQueue_.cancelCallback();
}
ReceiverQueue<TValue> receiverGetValues() {
return receiverQueue_.getMessages(static_cast<ChannelBridgeBase*>(this));
}
private:
using ReceiverAtomicQueue = typename folly::channels::detail::
AtomicQueue<IChannelCallback, folly::Try<TValue>>;
using SenderAtomicQueue = typename folly::channels::detail::
AtomicQueue<IChannelCallback, folly::Unit>;
void decref() {
if (refCount_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
delete this;
}
}
ReceiverAtomicQueue receiverQueue_;
SenderAtomicQueue senderQueue_;
std::atomic<int8_t> refCount_{1};
};
template <typename TValue>
using ChannelBridgePtr = typename ChannelBridge<TValue>::Ptr;
template <typename TValue>
struct ChannelBridgeHash {
using is_transparent = std::true_type;
bool operator()(ChannelBridgePtr<TValue> const& arg) const {
return std::hash<ChannelBridgePtr<TValue>>()(arg);
}
bool operator()(ChannelBridge<TValue>* const& arg) const {
return std::hash<ChannelBridge<TValue>*>()(arg);
}
};
template <typename TValue>
struct ChannelBridgeEqual {
using is_transparent = std::true_type;
bool operator()(
ChannelBridgePtr<TValue> const& lhs,
ChannelBridgePtr<TValue> const& rhs) const {
return lhs.get() == rhs.get();
}
bool operator()(
ChannelBridge<TValue>* const& lhs,
ChannelBridgePtr<TValue> const& rhs) const {
return lhs == rhs.get();
}
bool operator()(
ChannelBridgePtr<TValue> const& lhs,
ChannelBridge<TValue>* const& rhs) const {
return lhs.get() == rhs;
}
};
} // 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.
*/
#pragma once
#include <optional>
#include <folly/ExceptionWrapper.h>
#include <folly/Function.h>
#include <folly/ScopeGuard.h>
#include <folly/executors/SequencedExecutor.h>
#include <folly/experimental/channels/Channel.h>
#include <folly/experimental/coro/Task.h>
namespace folly {
namespace channels {
namespace detail {
struct CloseResult {
CloseResult() {}
explicit CloseResult(folly::exception_wrapper _exception)
: exception(std::move(_exception)) {}
std::optional<folly::exception_wrapper> exception;
};
enum class ChannelState {
Active,
CancellationTriggered,
CancellationProcessed
};
template <typename TSender>
ChannelState getSenderState(TSender* sender) {
if (sender == nullptr) {
return ChannelState::CancellationProcessed;
} else if (sender->isSenderClosed()) {
return ChannelState::CancellationTriggered;
} else {
return ChannelState::Active;
}
}
template <typename TReceiver>
ChannelState getReceiverState(TReceiver* receiver) {
if (receiver == nullptr) {
return ChannelState::CancellationProcessed;
} else if (receiver->isReceiverCancelled()) {
return ChannelState::CancellationTriggered;
} else {
return ChannelState::Active;
}
}
inline std::ostream& operator<<(std::ostream& os, ChannelState state) {
switch (state) {
case ChannelState::Active:
return os << "Active";
case ChannelState::CancellationTriggered:
return os << "CancellationTriggered";
case ChannelState::CancellationProcessed:
return os << "CancellationProcessed";
default:
return os << "Should never be hit";
}
}
/**
* A cancellation callback that wraps an existing channel callback. When the
* callback is fired, this object will trigger cancellation on its cancellation
* source (in addition to firing the wrapped callback).
*/
template <typename TSender>
class SenderCancellationCallback : public IChannelCallback {
public:
explicit SenderCancellationCallback(
TSender& sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
IChannelCallback* channelCallback)
: sender_(sender),
executor_(std::move(executor)),
channelCallback_(channelCallback) {
if (channelCallback_ == nullptr) {
// The sender was already canceled runOperationWithSenderCancellation was
// even called. This means the cancelled callback already was fired, so
// we will not set the callback to fire here.
cancelSource_.requestCancellation();
return;
}
CHECK(sender_);
if (!sender_->senderWait(this)) {
// The sender was cancelled after runOperationWithSenderCancellation was
// called, but before we had a chance to start the operation. This means
// that the cancelled callback was never called. We will therefore set it
// to fire here, when the operation is complete.
cancelSource_.requestCancellation();
callbackToFire_.setValue(CallbackToFire::Consume);
}
}
folly::coro::Task<void> onTaskCompleted() {
if (!channelCallback_) {
co_return;
}
auto callbackToFire = std::optional<CallbackToFire>();
if (callbackToFire_.isFulfilled()) {
// The callback was fired.
callbackToFire = co_await callbackToFire_.getSemiFuture();
} else {
// The callback has not yet been fired.
if (!sender_->cancelSenderWait()) {
// The sender has been cancelled, but the callback has not been called
// yet. Wait for the callback to be called.
callbackToFire = co_await callbackToFire_.getSemiFuture();
} else if (!sender_->senderWait(channelCallback_)) {
// The sender was cancelled between the call to cancelSenderWait and
// the call to senderWait. This means that the cancelled callback was
// never called. We will therefore set it to fire here.
callbackToFire = CallbackToFire::Consume;
}
}
if (callbackToFire.has_value()) {
switch (callbackToFire.value()) {
case CallbackToFire::Consume:
channelCallback_->consume(sender_.get());
co_return;
case CallbackToFire::Canceled:
channelCallback_->canceled(sender_.get());
co_return;
}
}
// The sender has not yet been cancelled, and we are now back in the state
// where the sender is waiting on the user-provided callback. We are done.
}
/**
* Returns a cancellation token that will trigger when the sender
*/
folly::CancellationToken getCancellationToken() {
return cancelSource_.getToken();
}
/**
* Requests cancellation, and triggers the consume function on the callback
* if the callback was not previously triggered.
*/
void consume(ChannelBridgeBase*) override {
executor_->add([=]() {
cancelSource_.requestCancellation();
CHECK(!callbackToFire_.isFulfilled());
callbackToFire_.setValue(CallbackToFire::Consume);
});
}
/**
* Requests cancellation, and triggers the canceled function on the callback
* if the callback was not previously triggered.
*/
void canceled(ChannelBridgeBase*) override {
executor_->add([=]() {
cancelSource_.requestCancellation();
CHECK(!callbackToFire_.isFulfilled());
callbackToFire_.setValue(CallbackToFire::Canceled);
});
}
private:
enum class CallbackToFire { Consume, Canceled };
TSender& sender_;
folly::Executor::KeepAlive<folly::SequencedExecutor> executor_;
IChannelCallback* channelCallback_;
folly::CancellationSource cancelSource_;
folly::Promise<CallbackToFire> callbackToFire_;
};
/**
* Any object that produces an output receiver (transform, merge,
* MergeChannel, etc) will listen for a cancellation signal from that output
* receiver. Once the consumer of the output receiver stops consuming, a
* callback will be called that triggers these objects to start cleaning
* themselves up (and eventually destroy themselves).
*
* However, when one of these objects decides to run a user coroutine, they
* would like that user coroutine to be able to get notified when that
* cancellation signal is received. That allows the coroutine to stop any
* long-running operations quickly, rather than running a long time when the
* consumer of the output receiver no longer cares about the result.
*
* This function enables that behavior. It will run the provided operation
* coroutine. While that coroutine is running, it will listen to cancellation
* events from the output receiver (through its sender). If it receives a
* cancellation signal from the sender, it will trigger cancellation of the
* operation coroutine.
*
* Once the coroutine finishes, it will then call the given channel callback
* to notify it of the cancellation event (the same way that callback would
* have been notified if no coroutine had been started). It will also resume
* waiting on the channel callback.
*
* @param executor: The executor to run the coroutine on.
*
* @param sender: The sender to use to listen for cancellation. If this is
* null, we will assume that cancellation already occurred.
*
* @param alreadyStartedWaiting: Whether or not the caller already started
* listening for a cancellation signal from the output receiver. If so, this
* function will temporarily stop waiting with that callback (so it can listen
* for the cancellation signal to stop the coroutine).
*
* @param channelCallbackToRestore: The channel callback to restore once the
* coroutine operation is complete.
*
* @param operation: The operation to run.
*/
template <typename TSender>
void runOperationWithSenderCancellation(
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
TSender& sender,
bool alreadyStartedWaiting,
IChannelCallback* channelCallbackToRestore,
folly::coro::Task<void> operation) noexcept {
if (alreadyStartedWaiting && (!sender || !sender->cancelSenderWait())) {
// The output receiver was cancelled before starting this operation
// (indicating that the channel callback already ran).
channelCallbackToRestore = nullptr;
}
folly::coro::co_invoke(
[&sender,
executor,
channelCallbackToRestore,
operation = std::move(operation)]() mutable -> folly::coro::Task<void> {
auto senderCancellationCallback = SenderCancellationCallback(
sender, executor, channelCallbackToRestore);
auto result =
co_await folly::coro::co_awaitTry(folly::coro::co_withCancellation(
senderCancellationCallback.getCancellationToken(),
std::move(operation)));
if (result.hasException()) {
LOG(FATAL) << fmt::format(
"Unexpected exception when running coroutine operation with "
"sender cancellation: {}",
result.exception().what());
}
co_await senderCancellationCallback.onTaskCompleted();
})
.scheduleOn(executor)
.start();
}
} // 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/experimental/channels/Channel.h>
#include <folly/experimental/channels/test/ChannelTestUtil.h>
#include <folly/portability/GTest.h>
namespace folly {
namespace channels {
using namespace testing;
class ChannelFixture : public Test,
public WithParamInterface<ConsumptionMode>,
public ChannelConsumerBase<int> {
protected:
ChannelFixture() : ChannelConsumerBase(GetParam()) {}
~ChannelFixture() override {
cancellationSource_.requestCancellation();
if (!continueConsuming_.isFulfilled()) {
continueConsuming_.setValue(false);
}
executor_.drain();
}
folly::Executor::KeepAlive<> getExecutor() override { return &executor_; }
void onNext(folly::Try<int> result) override { onNext_(std::move(result)); }
folly::ManualExecutor executor_;
StrictMock<MockNextCallback<int>> onNext_;
};
TEST_P(ChannelFixture, SingleWriteBeforeNext_ThenCancelled) {
auto [receiver, sender] = Channel<int>::create();
sender.write(1);
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onCancelled());
startConsuming(std::move(receiver));
executor_.drain();
EXPECT_FALSE(sender.isReceiverCancelled());
cancellationSource_.requestCancellation();
executor_.drain();
EXPECT_TRUE(sender.isReceiverCancelled());
}
TEST_P(ChannelFixture, SingleWriteAfterNext_ThenCancelled) {
auto [receiver, sender] = Channel<int>::create();
startConsuming(std::move(receiver));
executor_.drain();
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onCancelled());
sender.write(1);
executor_.drain();
EXPECT_FALSE(sender.isReceiverCancelled());
cancellationSource_.requestCancellation();
executor_.drain();
EXPECT_TRUE(sender.isReceiverCancelled());
}
TEST_P(ChannelFixture, MultipleWrites_ThenStopConsumingByReturningFalse) {
auto [receiver, sender] = Channel<int>::create();
startConsuming(std::move(receiver));
executor_.drain();
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onValue(2));
EXPECT_CALL(onNext_, onValue(3));
sender.write(1);
sender.write(2);
executor_.drain();
sender.write(3);
sender.write(4);
continueConsuming_ = folly::SharedPromise<bool>();
continueConsuming_.setValue(false);
executor_.drain();
}
TEST_P(
ChannelFixture,
MultipleWrites_ThenStopConsumingByThrowingOperationCancelled) {
auto [receiver, sender] = Channel<int>::create();
startConsuming(std::move(receiver));
executor_.drain();
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onValue(2));
EXPECT_CALL(onNext_, onValue(3));
sender.write(1);
sender.write(2);
executor_.drain();
sender.write(3);
sender.write(4);
continueConsuming_ = folly::SharedPromise<bool>();
continueConsuming_.setException(folly::OperationCancelled());
executor_.drain();
}
TEST_P(ChannelFixture, Close_NoException_BeforeSubscribe) {
auto [receiver, sender] = Channel<int>::create();
std::move(sender).close();
EXPECT_CALL(onNext_, onClosed());
startConsuming(std::move(receiver));
executor_.drain();
}
TEST_P(ChannelFixture, Close_NoException_AfterSubscribeAndWrite) {
auto [receiver, sender] = Channel<int>::create();
sender.write(1);
std::move(sender).close();
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onClosed());
startConsuming(std::move(receiver));
executor_.drain();
}
TEST_P(ChannelFixture, Close_DueToDestruction_BeforeSubscribe) {
auto [receiver, sender] = Channel<int>::create();
{ auto toDestroy = std::move(sender); }
EXPECT_CALL(onNext_, onClosed());
startConsuming(std::move(receiver));
executor_.drain();
}
TEST_P(ChannelFixture, Close_DueToDestruction_AfterSubscribeAndWrite) {
auto [receiver, sender] = Channel<int>::create();
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onClosed());
startConsuming(std::move(receiver));
sender.write(1);
{ auto toDestroy = std::move(sender); }
executor_.drain();
}
TEST_P(ChannelFixture, Close_Exception_BeforeSubscribe) {
auto [receiver, sender] = Channel<int>::create();
std::move(sender).close(std::runtime_error("Error"));
EXPECT_CALL(onNext_, onRuntimeError("std::runtime_error: Error"));
startConsuming(std::move(receiver));
executor_.drain();
}
TEST_P(ChannelFixture, Close_Exception_AfterSubscribeAndWrite) {
auto [receiver, sender] = Channel<int>::create();
sender.write(1);
std::move(sender).close(std::runtime_error("Error"));
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onRuntimeError("std::runtime_error: Error"));
startConsuming(std::move(receiver));
executor_.drain();
}
TEST_P(ChannelFixture, CancellationRespected) {
auto [receiver, sender] = Channel<int>::create();
EXPECT_CALL(onNext_, onValue(1));
continueConsuming_ = folly::SharedPromise<bool>();
sender.write(1);
startConsuming(std::move(receiver));
executor_.drain();
EXPECT_FALSE(sender.isReceiverCancelled());
cancellationSource_.requestCancellation();
executor_.drain();
EXPECT_TRUE(sender.isReceiverCancelled());
}
INSTANTIATE_TEST_CASE_P(
Channel_Coro_WithTry,
ChannelFixture,
testing::Values(ConsumptionMode::CoroWithTry));
INSTANTIATE_TEST_CASE_P(
Channel_Coro_WithoutTry,
ChannelFixture,
testing::Values(ConsumptionMode::CoroWithoutTry));
INSTANTIATE_TEST_CASE_P(
Channel_Callback_WithHandle,
ChannelFixture,
testing::Values(ConsumptionMode::CallbackWithHandle));
INSTANTIATE_TEST_CASE_P(
Channel_Callback_WithHandleList,
ChannelFixture,
testing::Values(ConsumptionMode::CallbackWithHandleList));
class ChannelFixtureStress : public Test,
public WithParamInterface<ConsumptionMode> {
protected:
ChannelFixtureStress()
: producer_(std::make_unique<StressTestProducer<int>>(
[value = 0]() mutable { return value++; })),
consumer_(std::make_unique<StressTestConsumer<int>>(
GetParam(), [lastReceived = -1](int value) mutable {
EXPECT_EQ(value, ++lastReceived);
})) {}
static constexpr std::chrono::milliseconds kTestTimeout =
std::chrono::milliseconds{5000};
std::unique_ptr<StressTestProducer<int>> producer_;
std::unique_ptr<StressTestConsumer<int>> consumer_;
};
TEST_P(ChannelFixtureStress, Close_NoException) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
consumer_->startConsuming(std::move(receiver));
/* sleep override */
std::this_thread::sleep_for(kTestTimeout);
producer_->stopProducing();
EXPECT_EQ(consumer_->waitForClose().get(), CloseType::NoException);
}
TEST_P(ChannelFixtureStress, Close_Exception) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::runtime_error("Error"));
consumer_->startConsuming(std::move(receiver));
/* sleep override */
std::this_thread::sleep_for(kTestTimeout);
producer_->stopProducing();
EXPECT_EQ(consumer_->waitForClose().get(), CloseType::Exception);
}
TEST_P(ChannelFixtureStress, Cancelled) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
consumer_->startConsuming(std::move(receiver));
/* sleep override */
std::this_thread::sleep_for(kTestTimeout);
consumer_->cancel();
EXPECT_EQ(consumer_->waitForClose().get(), CloseType::Cancelled);
producer_->stopProducing();
}
TEST_P(ChannelFixtureStress, Close_NoException_ThenCancelledImmediately) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
consumer_->startConsuming(std::move(receiver));
/* sleep override */
std::this_thread::sleep_for(kTestTimeout);
producer_->stopProducing();
consumer_->cancel();
EXPECT_THAT(
consumer_->waitForClose().get(),
AnyOf(Eq(CloseType::NoException), Eq(CloseType::Cancelled)));
}
TEST_P(ChannelFixtureStress, Cancelled_ThenClosedImmediately_NoException) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
consumer_->startConsuming(std::move(receiver));
/* sleep override */
std::this_thread::sleep_for(kTestTimeout);
consumer_->cancel();
producer_->stopProducing();
EXPECT_THAT(
consumer_->waitForClose().get(),
AnyOf(Eq(CloseType::NoException), Eq(CloseType::Cancelled)));
}
INSTANTIATE_TEST_CASE_P(
Channel_Coro_WithTry,
ChannelFixtureStress,
testing::Values(ConsumptionMode::CoroWithTry));
INSTANTIATE_TEST_CASE_P(
Channel_Coro_WithoutTry,
ChannelFixtureStress,
testing::Values(ConsumptionMode::CoroWithoutTry));
INSTANTIATE_TEST_CASE_P(
Channel_Callback_WithHandle,
ChannelFixtureStress,
testing::Values(ConsumptionMode::CallbackWithHandle));
INSTANTIATE_TEST_CASE_P(
Channel_Callback_WithHandleList,
ChannelFixtureStress,
testing::Values(ConsumptionMode::CallbackWithHandleList));
} // 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/CPUThreadPoolExecutor.h>
#include <folly/experimental/channels/ConsumeChannel.h>
#include <folly/experimental/coro/DetachOnCancel.h>
#include <folly/experimental/coro/Sleep.h>
#include <folly/futures/SharedPromise.h>
#include <folly/portability/GMock.h>
namespace folly {
namespace channels {
template <typename T, typename... Others>
std::vector<T> toVector(T firstItem, Others... items) {
std::vector<T> itemsVector;
itemsVector.push_back(std::move(firstItem));
[[maybe_unused]] int dummy[] = {
(itemsVector.push_back(std::move(items)), 0)...};
return itemsVector;
}
template <typename TValue>
class MockNextCallback {
public:
void operator()(folly::Try<TValue> result) {
if (result.hasValue()) {
onValue(result.value());
} else if (result.template hasException<folly::OperationCancelled>()) {
onCancelled();
} else if (result.template hasException<std::runtime_error>()) {
onRuntimeError(result.exception().what().toStdString());
} else if (result.hasException()) {
LOG(FATAL) << "Unexpected exception: " << result.exception().what();
} else {
onClosed();
}
}
MOCK_METHOD1_T(onValue, void(TValue));
MOCK_METHOD0(onClosed, void());
MOCK_METHOD0(onCancelled, void());
MOCK_METHOD1(onRuntimeError, void(std::string));
};
enum class ConsumptionMode {
CoroWithTry,
CoroWithoutTry,
CallbackWithHandle,
CallbackWithHandleList
};
template <typename TValue>
class ChannelConsumerBase {
public:
explicit ChannelConsumerBase(ConsumptionMode mode) : mode_(mode) {
continueConsuming_.setValue(true);
}
ChannelConsumerBase(ChannelConsumerBase&&) = default;
ChannelConsumerBase& operator=(ChannelConsumerBase&&) = default;
virtual ~ChannelConsumerBase() = default;
virtual folly::Executor::KeepAlive<> getExecutor() = 0;
virtual void onNext(folly::Try<TValue> result) = 0;
void startConsuming(Receiver<TValue> receiver) {
folly::coro::co_withCancellation(
cancellationSource_.getToken(), processValuesCoro(std::move(receiver)))
.scheduleOn(getExecutor())
.start();
}
folly::coro::Task<void> processValuesCoro(Receiver<TValue> receiver) {
auto executor = co_await folly::coro::co_current_executor;
if (mode_ == ConsumptionMode::CoroWithTry ||
mode_ == ConsumptionMode::CoroWithoutTry) {
do {
folly::Try<TValue> resultTry;
if (mode_ == ConsumptionMode::CoroWithTry) {
resultTry = co_await folly::coro::co_awaitTry(receiver.next());
} else if (mode_ == ConsumptionMode::CoroWithoutTry) {
try {
auto result = co_await receiver.next();
if (result.has_value()) {
resultTry = folly::Try<TValue>(result.value());
} else {
resultTry = folly::Try<TValue>();
}
} catch (const std::exception& ex) {
resultTry = folly::Try<TValue>(
folly::exception_wrapper(std::current_exception(), ex));
} catch (...) {
resultTry = folly::Try<TValue>(
folly::exception_wrapper(std::current_exception()));
}
} else {
LOG(FATAL) << "Unknown consumption mode";
}
bool hasValue = resultTry.hasValue();
onNext(std::move(resultTry));
if (!hasValue) {
co_return;
}
} while (co_await folly::coro::detachOnCancel(
continueConsuming_.getSemiFuture()));
} else if (mode_ == ConsumptionMode::CallbackWithHandle) {
auto callbackHandle = consumeChannelWithCallback(
std::move(receiver),
getExecutor(),
[=](folly::Try<TValue> resultTry) -> folly::coro::Task<bool> {
onNext(std::move(resultTry));
co_return co_await folly::coro::detachOnCancel(
continueConsuming_.getSemiFuture());
});
cancelCallback_ = std::make_unique<folly::CancellationCallback>(
co_await folly::coro::co_current_cancellation_token,
[=, handle = std::move(callbackHandle)]() mutable {
handle.reset();
});
} else if (mode_ == ConsumptionMode::CallbackWithHandleList) {
auto callbackHandleList = ChannelCallbackHandleList();
consumeChannelWithCallback(
std::move(receiver),
getExecutor(),
[=](folly::Try<TValue> resultTry) -> folly::coro::Task<bool> {
onNext(std::move(resultTry));
co_return co_await folly::coro::detachOnCancel(
continueConsuming_.getSemiFuture());
},
callbackHandleList);
cancelCallback_ = std::make_unique<folly::CancellationCallback>(
co_await folly::coro::co_current_cancellation_token,
[=, handleList = std::move(callbackHandleList)]() mutable {
executor->add([handleList = std::move(handleList)]() mutable {
handleList.clear();
});
});
} else {
LOG(FATAL) << "Unknown consumption mode";
}
}
protected:
ConsumptionMode mode_;
folly::CancellationSource cancellationSource_;
folly::SharedPromise<bool> continueConsuming_;
std::unique_ptr<folly::CancellationCallback> cancelCallback_;
};
enum class CloseType { NoException, Exception, Cancelled };
template <typename TValue>
class StressTestConsumer : public ChannelConsumerBase<TValue> {
public:
StressTestConsumer(
ConsumptionMode mode, folly::Function<void(TValue)> onValue)
: ChannelConsumerBase<TValue>(mode),
executor_(std::make_unique<folly::CPUThreadPoolExecutor>(1)),
onValue_(std::move(onValue)) {}
StressTestConsumer(StressTestConsumer&&) = delete;
StressTestConsumer&& operator=(StressTestConsumer&&) = delete;
~StressTestConsumer() override {
this->cancellationSource_.requestCancellation();
if (!this->continueConsuming_.isFulfilled()) {
this->continueConsuming_.setValue(false);
}
executor_.reset();
}
folly::Executor::KeepAlive<> getExecutor() override {
return executor_.get();
}
void onNext(folly::Try<TValue> result) override {
if (result.hasValue()) {
onValue_(std::move(result.value()));
} else if (result.template hasException<folly::OperationCancelled>()) {
closedType_.setValue(CloseType::Cancelled);
} else if (result.hasException()) {
EXPECT_TRUE(result.template hasException<std::runtime_error>());
closedType_.setValue(CloseType::Exception);
} else {
closedType_.setValue(CloseType::NoException);
}
}
void cancel() { this->cancellationSource_.requestCancellation(); }
folly::SemiFuture<CloseType> waitForClose() {
return closedType_.getSemiFuture();
}
private:
std::unique_ptr<folly::CPUThreadPoolExecutor> executor_;
folly::Function<void(TValue)> onValue_;
folly::Promise<CloseType> closedType_;
};
template <typename TValue>
class StressTestProducer {
public:
explicit StressTestProducer(folly::Function<TValue()> getNextValue)
: executor_(std::make_unique<folly::CPUThreadPoolExecutor>(1)),
getNextValue_(std::move(getNextValue)) {}
StressTestProducer(StressTestProducer&&) = delete;
StressTestProducer&& operator=(StressTestProducer&&) = delete;
~StressTestProducer() {
if (executor_) {
stopProducing();
executor_.reset();
}
}
void startProducing(
Sender<TValue> sender,
std::optional<folly::exception_wrapper> closeException) {
auto produceTask = folly::coro::co_invoke(
[=,
sender = std::move(sender),
ex = std::move(closeException)]() mutable -> folly::coro::Task<void> {
for (int i = 1; !stopped_.load(std::memory_order_relaxed); i++) {
if (i % 1000 == 0) {
co_await folly::coro::sleep(std::chrono::milliseconds(100));
}
sender.write(getNextValue_());
}
if (ex.has_value()) {
std::move(sender).close(std::move(ex.value()));
} else {
std::move(sender).close();
}
co_return;
});
std::move(produceTask).scheduleOn(executor_.get()).start();
}
void stopProducing() { stopped_.store(true); }
private:
std::unique_ptr<folly::CPUThreadPoolExecutor> executor_;
folly::Function<TValue()> getNextValue_;
std::atomic<bool> stopped_{false};
};
} // 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/FanoutChannel.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 FanoutChannelFixture : public Test {
protected:
FanoutChannelFixture() {}
~FanoutChannelFixture() { executor_.drain(); }
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(FanoutChannelFixture, ReceiveValue_FanoutBroadcastsValues) {
auto [inputReceiver, sender] = Channel<int>::create();
auto fanoutChannel =
createFanoutChannel(std::move(inputReceiver), &executor_);
auto [handle1, callback1] = processValues(fanoutChannel.getNewReceiver(
[] { return toVector(100); } /* getInitialValues */));
auto [handle2, callback2] = processValues(fanoutChannel.getNewReceiver(
[] { return toVector(200); } /* getInitialValues */));
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(fanoutChannel.anyReceivers());
sender.write(1);
sender.write(2);
executor_.drain();
std::move(sender).close();
executor_.drain();
EXPECT_FALSE(fanoutChannel.anyReceivers());
}
TEST_F(FanoutChannelFixture, InputClosed_AllOutputReceiversClose) {
auto [inputReceiver, sender] = Channel<int>::create();
auto fanoutChannel =
createFanoutChannel(std::move(inputReceiver), &executor_);
auto [handle1, callback1] = processValues(fanoutChannel.getNewReceiver());
auto [handle2, callback2] = processValues(fanoutChannel.getNewReceiver());
EXPECT_CALL(*callback1, onValue(1));
EXPECT_CALL(*callback2, onValue(1));
EXPECT_CALL(*callback1, onClosed());
EXPECT_CALL(*callback2, onClosed());
executor_.drain();
EXPECT_TRUE(fanoutChannel.anyReceivers());
sender.write(1);
executor_.drain();
std::move(sender).close();
executor_.drain();
EXPECT_FALSE(fanoutChannel.anyReceivers());
}
TEST_F(FanoutChannelFixture, InputThrows_AllOutputReceiversGetException) {
auto [inputReceiver, sender] = Channel<int>::create();
auto fanoutChannel =
createFanoutChannel(std::move(inputReceiver), &executor_);
auto [handle1, callback1] = processValues(fanoutChannel.getNewReceiver());
auto [handle2, callback2] = processValues(fanoutChannel.getNewReceiver());
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(fanoutChannel.anyReceivers());
sender.write(1);
executor_.drain();
std::move(sender).close(std::runtime_error("Error"));
executor_.drain();
EXPECT_FALSE(fanoutChannel.anyReceivers());
}
TEST_F(FanoutChannelFixture, ReceiversCancelled) {
auto [inputReceiver, sender] = Channel<int>::create();
auto fanoutChannel =
createFanoutChannel(std::move(inputReceiver), &executor_);
auto [handle1, callback1] = processValues(fanoutChannel.getNewReceiver());
auto [handle2, callback2] = processValues(fanoutChannel.getNewReceiver());
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(fanoutChannel.anyReceivers());
sender.write(1);
executor_.drain();
EXPECT_TRUE(fanoutChannel.anyReceivers());
handle1.reset();
sender.write(2);
executor_.drain();
EXPECT_TRUE(fanoutChannel.anyReceivers());
handle2.reset();
sender.write(3);
executor_.drain();
EXPECT_FALSE(fanoutChannel.anyReceivers());
std::move(sender).close();
executor_.drain();
EXPECT_FALSE(fanoutChannel.anyReceivers());
}
class FanoutChannelFixtureStress : public Test {
protected:
FanoutChannelFixtureStress()
: producer_(makeProducer()),
consumers_(toVector(makeConsumer(), makeConsumer(), makeConsumer())) {}
static std::unique_ptr<StressTestProducer<int>> makeProducer() {
return std::make_unique<StressTestProducer<int>>(
[value = 0]() mutable { return value++; });
}
static std::unique_ptr<StressTestConsumer<int>> makeConsumer() {
return std::make_unique<StressTestConsumer<int>>(
ConsumptionMode::CallbackWithHandle,
[lastReceived = -1](int value) mutable {
if (lastReceived == -1) {
lastReceived = value;
} else {
EXPECT_EQ(value, ++lastReceived);
}
});
}
static void sleepFor(std::chrono::milliseconds duration) {
/* sleep override */
std::this_thread::sleep_for(duration);
}
static constexpr std::chrono::milliseconds kTestTimeout =
std::chrono::milliseconds{10};
std::unique_ptr<StressTestProducer<int>> producer_;
std::vector<std::unique_ptr<StressTestConsumer<int>>> consumers_;
};
TEST_F(FanoutChannelFixtureStress, HandleClosed) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
folly::CPUThreadPoolExecutor fanoutChannelExecutor(1);
auto fanoutChannel = createFanoutChannel(
std::move(receiver),
folly::SerialExecutor::create(&fanoutChannelExecutor));
consumers_.at(0)->startConsuming(fanoutChannel.getNewReceiver());
consumers_.at(1)->startConsuming(fanoutChannel.getNewReceiver());
sleepFor(kTestTimeout / 3);
consumers_.at(2)->startConsuming(fanoutChannel.getNewReceiver());
sleepFor(kTestTimeout / 3);
consumers_.at(0)->cancel();
EXPECT_EQ(consumers_.at(0)->waitForClose().get(), CloseType::Cancelled);
sleepFor(kTestTimeout / 3);
std::move(fanoutChannel).close();
EXPECT_EQ(consumers_.at(1)->waitForClose().get(), CloseType::NoException);
EXPECT_EQ(consumers_.at(2)->waitForClose().get(), CloseType::NoException);
}
TEST_F(FanoutChannelFixtureStress, InputChannelClosed) {
auto [receiver, sender] = Channel<int>::create();
producer_->startProducing(std::move(sender), std::nullopt /* closeEx */);
folly::CPUThreadPoolExecutor fanoutChannelExecutor(1);
auto fanoutChannel = createFanoutChannel(
std::move(receiver),
folly::SerialExecutor::create(&fanoutChannelExecutor));
consumers_.at(0)->startConsuming(fanoutChannel.getNewReceiver());
consumers_.at(1)->startConsuming(fanoutChannel.getNewReceiver());
sleepFor(kTestTimeout / 3);
consumers_.at(2)->startConsuming(fanoutChannel.getNewReceiver());
sleepFor(kTestTimeout / 3);
consumers_.at(0)->cancel();
EXPECT_EQ(consumers_.at(0)->waitForClose().get(), CloseType::Cancelled);
sleepFor(kTestTimeout / 3);
producer_->stopProducing();
EXPECT_EQ(consumers_.at(1)->waitForClose().get(), CloseType::NoException);
EXPECT_EQ(consumers_.at(2)->waitForClose().get(), CloseType::NoException);
}
} // namespace channels
} // namespace folly
This diff is collapsed.
This diff is collapsed.
/*
* 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/experimental/channels/Producer.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 ProducerFixture : public Test {
protected:
ProducerFixture() {}
~ProducerFixture() { executor_.drain(); }
ChannelCallbackHandle processValues(Receiver<int> receiver) {
return consumeChannelWithCallback(
std::move(receiver),
&executor_,
[=](folly::Try<int> resultTry) -> folly::coro::Task<bool> {
onNext_(std::move(resultTry));
co_return true;
});
}
folly::ManualExecutor executor_;
StrictMock<MockNextCallback<int>> onNext_;
};
TEST_F(ProducerFixture, Write_ThenCloseWithoutException) {
class TestProducer : public Producer<int> {
public:
TestProducer(
Sender<int> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor)
: Producer<int>(std::move(sender), std::move(executor)) {
write(1);
close();
}
};
auto receiver = makeProducer<TestProducer>(&executor_);
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onClosed());
auto callbackHandle = processValues(std::move(receiver));
executor_.drain();
}
TEST_F(ProducerFixture, Write_ThenCloseWithException) {
class TestProducer : public Producer<int> {
public:
TestProducer(
Sender<int> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor)
: Producer<int>(std::move(sender), std::move(executor)) {
write(1);
close(std::runtime_error("Error"));
}
};
auto receiver = makeProducer<TestProducer>(&executor_);
EXPECT_CALL(onNext_, onValue(1));
EXPECT_CALL(onNext_, onRuntimeError("std::runtime_error: Error"));
auto callbackHandle = processValues(std::move(receiver));
executor_.drain();
}
TEST_F(ProducerFixture, KeepAliveExists_DelaysDestruction) {
class TestProducer : public Producer<int> {
public:
TestProducer(
Sender<int> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
folly::SemiFuture<folly::Unit> future,
bool& destructed)
: Producer<int>(std::move(sender), std::move(executor)),
destructed_(destructed) {
folly::coro::co_invoke(
[keepAlive = getKeepAlive(),
future = std::move(future)]() mutable -> folly::coro::Task<void> {
co_await std::move(future);
})
.scheduleOn(getExecutor())
.start();
}
~TestProducer() { destructed_ = true; }
bool& destructed_;
};
auto promise = folly::Promise<folly::Unit>();
bool destructed = false;
auto receiver = makeProducer<TestProducer>(
&executor_, promise.getSemiFuture(), destructed);
EXPECT_CALL(onNext_, onCancelled());
auto callbackHandle = processValues(std::move(receiver));
executor_.drain();
callbackHandle.reset();
executor_.drain();
EXPECT_FALSE(destructed);
promise.setValue();
executor_.drain();
EXPECT_TRUE(destructed);
}
TEST_F(
ProducerFixture,
ConsumerStopsConsumingReceiver_OnCancelledCalled_ThenDestructed) {
class TestProducer : public Producer<int> {
public:
TestProducer(
Sender<int> sender,
folly::Executor::KeepAlive<folly::SequencedExecutor> executor,
folly::Promise<folly::Unit> onCancelledStarted,
folly::SemiFuture<folly::Unit> onCancelledCompleted,
bool& destructed)
: Producer<int>(std::move(sender), std::move(executor)),
onCancelledStarted_(std::move(onCancelledStarted)),
onCancelledCompleted_(std::move(onCancelledCompleted)),
destructed_(destructed) {}
folly::coro::Task<void> onClosed() override {
onCancelledStarted_.setValue();
co_await std::move(onCancelledCompleted_);
}
~TestProducer() override { destructed_ = true; }
folly::Promise<folly::Unit> onCancelledStarted_;
folly::SemiFuture<folly::Unit> onCancelledCompleted_;
bool& destructed_;
};
auto onCancelledStartedPromise = folly::Promise<folly::Unit>();
auto onCancelledStartedFuture = onCancelledStartedPromise.getSemiFuture();
auto onCancelledCompletedPromise = folly::Promise<folly::Unit>();
auto onCancelledCompletedFuture = onCancelledCompletedPromise.getSemiFuture();
bool destructed = false;
auto receiver = makeProducer<TestProducer>(
&executor_,
std::move(onCancelledStartedPromise),
std::move(onCancelledCompletedFuture),
destructed);
EXPECT_CALL(onNext_, onCancelled());
auto callbackHandle = processValues(std::move(receiver));
executor_.drain();
EXPECT_FALSE(onCancelledStartedFuture.isReady());
EXPECT_FALSE(destructed);
callbackHandle.reset();
executor_.drain();
EXPECT_TRUE(onCancelledStartedFuture.isReady());
EXPECT_FALSE(destructed);
onCancelledCompletedPromise.setValue();
executor_.drain();
EXPECT_TRUE(destructed);
}
} // namespace channels
} // namespace folly
This diff is collapsed.
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