Commit 7c7597c2 authored by Shai Szulanski's avatar Shai Szulanski Committed by Facebook GitHub Bot

Delete unused multiplex and de/materialize algorithms

Summary: The CallbackRecord interface is unfamiliar to users and inconsistent with the general direction of folly::coro, and as a result these algorithms have not seen any usage. Deletes them and moves CallbackRecord into the only file that uses it. (We may eventually want to replace it if we come up with a story for Try<Reference> but I'm not tackling that here).

Differential Revision: D33570204

fbshipit-source-id: 9cfb0def927e5882f582d6c1eb280d97c9ff3e22
parent 7880abd1
/*
* Copyright (c) Meta Platforms, Inc. and 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/coro/Materialize.h>
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
template <typename Reference, typename Value>
AsyncGenerator<Reference, Value> dematerialize(
AsyncGenerator<CallbackRecord<Reference>, CallbackRecord<Value>> source) {
while (auto item = co_await source.next()) {
if (item->hasValue()) {
// Value
co_yield std::move(*item).value();
} else if (item->hasNone()) {
// None
co_return;
} else if (item->hasError()) {
// Exception
std::move(*item).error().throw_exception();
} else {
DCHECK(false);
std::terminate();
}
}
}
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
/*
* Copyright (c) Meta Platforms, Inc. and 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/ExceptionWrapper.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/Coroutine.h>
#include <folly/experimental/coro/Materialize.h>
#include <variant>
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
// Dematerialize the CallbackRecords from an input stream into a stream that
// replays all of the events materialized in the input stream.
//
// The input is a stream of CallbackRecord.
//
// The output is a stream of Value.
//
// Example:
// AsyncGenerator<CallbackRecord<int>> stream();
//
// Task<void> consumer() {
// auto events = dematerialize(stream());
// try {
// while (auto item = co_await events.next()) {
// // Value
// auto&& value = *item;
// std::cout << "value " << value << "\n";
// }
// // None
// std::cout << "end\n";
// } catch (const std::exception& error) {
// // Exception
// std::cout << "error " << error.what() << "\n";
// }
// }
template <typename Reference, typename Value>
AsyncGenerator<Reference, Value> dematerialize(
AsyncGenerator<CallbackRecord<Reference>, CallbackRecord<Value>> source);
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
#include <folly/experimental/coro/Dematerialize-inl.h>
/*
* Copyright (c) Meta Platforms, Inc. and 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.
*/
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
template <typename Reference, typename Value>
AsyncGenerator<CallbackRecord<Reference>, CallbackRecord<Value>> materialize(
AsyncGenerator<Reference, Value> source) {
using EventType = CallbackRecord<Reference>;
folly::exception_wrapper ex;
try {
while (auto item = co_await source.next()) {
co_yield EventType{callback_record_value, *std::move(item)};
}
} catch (...) {
ex = folly::exception_wrapper{std::current_exception()};
}
if (ex) {
co_yield EventType{callback_record_error, std::move(ex)};
} else {
co_yield EventType{callback_record_none};
}
}
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
/*
* Copyright (c) Meta Platforms, Inc. and 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/ExceptionWrapper.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/Coroutine.h>
#include <variant>
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
enum class CallbackRecordSelector { Invalid, Value, None, Error };
constexpr inline std::in_place_index_t<0> const callback_record_value{};
constexpr inline std::in_place_index_t<1> const callback_record_none{};
constexpr inline std::in_place_index_t<2> const callback_record_error{};
//
// CallbackRecord records the result of a single invocation of a callback.
//
// This is very related to Try and expected, but this also records None in
// addition to Value and Error results.
//
// When the callback supports multiple overloads of Value then T would be
// something like a variant<tuple<..>, ..>
//
// When the callback supports multiple overloads of Error then all the errors
// are coerced to folly::exception_wrapper
//
template <class T>
class CallbackRecord {
static void clear(CallbackRecord* that) {
auto selector =
std::exchange(that->selector_, CallbackRecordSelector::Invalid);
if (selector == CallbackRecordSelector::Value) {
detail::deactivate(that->value_);
} else if (selector == CallbackRecordSelector::Error) {
detail::deactivate(that->error_);
}
}
template <class OtherReference>
static void convert_variant(
CallbackRecord* that, const CallbackRecord<OtherReference>& other) {
if (other.hasValue()) {
detail::activate(that->value_, other.value_.get());
} else if (other.hasError()) {
detail::activate(that->error_, other.error_.get());
}
that->selector_ = other.selector_;
}
template <class OtherReference>
static void convert_variant(
CallbackRecord* that, CallbackRecord<OtherReference>&& other) {
if (other.hasValue()) {
detail::activate(that->value_, std::move(other.value_).get());
} else if (other.hasError()) {
detail::activate(that->error_, std::move(other.error_).get());
}
that->selector_ = other.selector_;
}
public:
~CallbackRecord() { clear(this); }
CallbackRecord() noexcept : selector_(CallbackRecordSelector::Invalid) {}
template <class V>
CallbackRecord(const std::in_place_index_t<0>&, V&& v) noexcept(
std::is_nothrow_constructible_v<T, V>)
: CallbackRecord() {
detail::activate(value_, std::forward<V>(v));
selector_ = CallbackRecordSelector::Value;
}
explicit CallbackRecord(const std::in_place_index_t<1>&) noexcept
: selector_(CallbackRecordSelector::None) {}
CallbackRecord(
const std::in_place_index_t<2>&, folly::exception_wrapper e) noexcept
: CallbackRecord() {
detail::activate(error_, std::move(e));
selector_ = CallbackRecordSelector::Error;
}
CallbackRecord(CallbackRecord&& other) noexcept(
std::is_nothrow_move_constructible_v<T>)
: CallbackRecord() {
convert_variant(this, std::move(other));
}
CallbackRecord& operator=(CallbackRecord&& other) noexcept(
std::is_nothrow_move_constructible_v<T>) {
if (&other != this) {
clear(this);
convert_variant(this, std::move(other));
}
return *this;
}
template <class U>
CallbackRecord(CallbackRecord<U>&& other) noexcept(
std::is_nothrow_constructible_v<T, U>)
: CallbackRecord() {
convert_variant(this, std::move(other));
}
bool hasNone() const noexcept {
return selector_ == CallbackRecordSelector::None;
}
bool hasError() const noexcept {
return selector_ == CallbackRecordSelector::Error;
}
decltype(auto) error() & {
DCHECK(hasError());
return error_.get();
}
decltype(auto) error() && {
DCHECK(hasError());
return std::move(error_).get();
}
decltype(auto) error() const& {
DCHECK(hasError());
return error_.get();
}
decltype(auto) error() const&& {
DCHECK(hasError());
return std::move(error_).get();
}
bool hasValue() const noexcept {
return selector_ == CallbackRecordSelector::Value;
}
decltype(auto) value() & {
DCHECK(hasValue());
return value_.get();
}
decltype(auto) value() && {
DCHECK(hasValue());
return std::move(value_).get();
}
decltype(auto) value() const& {
DCHECK(hasValue());
return value_.get();
}
decltype(auto) value() const&& {
DCHECK(hasValue());
return std::move(value_).get();
}
explicit operator bool() const noexcept {
return selector_ != CallbackRecordSelector::Invalid;
}
private:
union {
detail::ManualLifetime<T> value_;
detail::ManualLifetime<folly::exception_wrapper> error_;
};
CallbackRecordSelector selector_;
};
// Materialize the results an input stream into a stream that
// contains all of the events of the input stream.
//
// The output is a stream of CallbackRecord.
//
// Example:
// AsyncGenerator<int> stream();
//
// Task<void> consumer() {
// auto events = materialize(stream());
// while (auto item = co_await events.next()) {
// auto&& event = *item;
// if (event.hasValue()) {
// // Value
// int value = std::move(event).value();
// std::cout << "value " << value << "\n";
// } else if (event.hasNone()) {
// // None
// std::cout << "end\n";
// } else {
// // Exception
// folly::exception_wrapper error = std::move(event).error();
// std::cout << "error " << error.what() << "\n";
// }
// }
// }
template <typename Reference, typename Value>
AsyncGenerator<CallbackRecord<Reference>, CallbackRecord<Value>> materialize(
AsyncGenerator<Reference, Value> source);
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
#include <folly/experimental/coro/Materialize-inl.h>
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#include <folly/Executor.h> #include <folly/Executor.h>
#include <folly/ScopeGuard.h> #include <folly/ScopeGuard.h>
#include <folly/experimental/coro/Baton.h> #include <folly/experimental/coro/Baton.h>
#include <folly/experimental/coro/Materialize.h>
#include <folly/experimental/coro/Mutex.h> #include <folly/experimental/coro/Mutex.h>
#include <folly/experimental/coro/Task.h> #include <folly/experimental/coro/Task.h>
#include <folly/experimental/coro/ViaIfAsync.h> #include <folly/experimental/coro/ViaIfAsync.h>
...@@ -35,6 +34,165 @@ ...@@ -35,6 +34,165 @@
namespace folly { namespace folly {
namespace coro { namespace coro {
namespace detail {
enum class CallbackRecordSelector { Invalid, Value, None, Error };
constexpr inline std::in_place_index_t<0> const callback_record_value{};
constexpr inline std::in_place_index_t<1> const callback_record_none{};
constexpr inline std::in_place_index_t<2> const callback_record_error{};
//
// CallbackRecord records the result of a single invocation of a callback.
//
// This is very related to Try and expected, but this also records None in
// addition to Value and Error results.
//
// When the callback supports multiple overloads of Value then T would be
// something like a variant<tuple<..>, ..>
//
// When the callback supports multiple overloads of Error then all the errors
// are coerced to folly::exception_wrapper
//
template <class T>
class CallbackRecord {
static void clear(CallbackRecord* that) {
auto selector =
std::exchange(that->selector_, CallbackRecordSelector::Invalid);
if (selector == CallbackRecordSelector::Value) {
detail::deactivate(that->value_);
} else if (selector == CallbackRecordSelector::Error) {
detail::deactivate(that->error_);
}
}
template <class OtherReference>
static void convert_variant(
CallbackRecord* that, const CallbackRecord<OtherReference>& other) {
if (other.hasValue()) {
detail::activate(that->value_, other.value_.get());
} else if (other.hasError()) {
detail::activate(that->error_, other.error_.get());
}
that->selector_ = other.selector_;
}
template <class OtherReference>
static void convert_variant(
CallbackRecord* that, CallbackRecord<OtherReference>&& other) {
if (other.hasValue()) {
detail::activate(that->value_, std::move(other.value_).get());
} else if (other.hasError()) {
detail::activate(that->error_, std::move(other.error_).get());
}
that->selector_ = other.selector_;
}
public:
~CallbackRecord() { clear(this); }
CallbackRecord() noexcept : selector_(CallbackRecordSelector::Invalid) {}
template <class V>
CallbackRecord(const std::in_place_index_t<0>&, V&& v) noexcept(
std::is_nothrow_constructible_v<T, V>)
: CallbackRecord() {
detail::activate(value_, std::forward<V>(v));
selector_ = CallbackRecordSelector::Value;
}
explicit CallbackRecord(const std::in_place_index_t<1>&) noexcept
: selector_(CallbackRecordSelector::None) {}
CallbackRecord(
const std::in_place_index_t<2>&, folly::exception_wrapper e) noexcept
: CallbackRecord() {
detail::activate(error_, std::move(e));
selector_ = CallbackRecordSelector::Error;
}
CallbackRecord(CallbackRecord&& other) noexcept(
std::is_nothrow_move_constructible_v<T>)
: CallbackRecord() {
convert_variant(this, std::move(other));
}
CallbackRecord& operator=(CallbackRecord&& other) noexcept(
std::is_nothrow_move_constructible_v<T>) {
if (&other != this) {
clear(this);
convert_variant(this, std::move(other));
}
return *this;
}
template <class U>
CallbackRecord(CallbackRecord<U>&& other) noexcept(
std::is_nothrow_constructible_v<T, U>)
: CallbackRecord() {
convert_variant(this, std::move(other));
}
bool hasNone() const noexcept {
return selector_ == CallbackRecordSelector::None;
}
bool hasError() const noexcept {
return selector_ == CallbackRecordSelector::Error;
}
decltype(auto) error() & {
DCHECK(hasError());
return error_.get();
}
decltype(auto) error() && {
DCHECK(hasError());
return std::move(error_).get();
}
decltype(auto) error() const& {
DCHECK(hasError());
return error_.get();
}
decltype(auto) error() const&& {
DCHECK(hasError());
return std::move(error_).get();
}
bool hasValue() const noexcept {
return selector_ == CallbackRecordSelector::Value;
}
decltype(auto) value() & {
DCHECK(hasValue());
return value_.get();
}
decltype(auto) value() && {
DCHECK(hasValue());
return std::move(value_).get();
}
decltype(auto) value() const& {
DCHECK(hasValue());
return value_.get();
}
decltype(auto) value() const&& {
DCHECK(hasValue());
return std::move(value_).get();
}
explicit operator bool() const noexcept {
return selector_ != CallbackRecordSelector::Invalid;
}
private:
union {
detail::ManualLifetime<T> value_;
detail::ManualLifetime<folly::exception_wrapper> error_;
};
CallbackRecordSelector selector_;
};
} // namespace detail
template <typename Reference, typename Value> template <typename Reference, typename Value>
AsyncGenerator<Reference, Value> merge( AsyncGenerator<Reference, Value> merge(
...@@ -50,7 +208,7 @@ AsyncGenerator<Reference, Value> merge( ...@@ -50,7 +208,7 @@ AsyncGenerator<Reference, Value> merge(
coro::Baton recordPublished; coro::Baton recordPublished;
coro::Baton recordConsumed; coro::Baton recordConsumed;
coro::Baton allTasksCompleted; coro::Baton allTasksCompleted;
CallbackRecord<Reference> record; detail::CallbackRecord<Reference> record;
}; };
auto makeConsumerTask = auto makeConsumerTask =
...@@ -78,8 +236,8 @@ AsyncGenerator<Reference, Value> merge( ...@@ -78,8 +236,8 @@ AsyncGenerator<Reference, Value> merge(
} }
// Publish the value. // Publish the value.
state_->record = CallbackRecord<Reference>{ state_->record = detail::CallbackRecord<Reference>{
callback_record_value, *std::move(item)}; detail::callback_record_value, *std::move(item)};
state_->recordPublished.post(); state_->recordPublished.post();
// Wait until the consumer is finished with it. // Wait until the consumer is finished with it.
...@@ -105,8 +263,8 @@ AsyncGenerator<Reference, Value> merge( ...@@ -105,8 +263,8 @@ AsyncGenerator<Reference, Value> merge(
auto lock = co_await co_viaIfAsync( auto lock = co_await co_viaIfAsync(
state_->executor.get_alias(), state_->mutex.co_scoped_lock()); state_->executor.get_alias(), state_->mutex.co_scoped_lock());
if (!state_->record.hasError()) { if (!state_->record.hasError()) {
state_->record = state_->record = detail::CallbackRecord<Reference>{
CallbackRecord<Reference>{callback_record_error, std::move(ex)}; detail::callback_record_error, std::move(ex)};
state_->recordPublished.post(); state_->recordPublished.post();
} }
}; };
...@@ -134,8 +292,8 @@ AsyncGenerator<Reference, Value> merge( ...@@ -134,8 +292,8 @@ AsyncGenerator<Reference, Value> merge(
auto lock = co_await co_viaIfAsync( auto lock = co_await co_viaIfAsync(
state->executor.get_alias(), state->mutex.co_scoped_lock()); state->executor.get_alias(), state->mutex.co_scoped_lock());
if (!state->record.hasError()) { if (!state->record.hasError()) {
state->record = state->record = detail::CallbackRecord<Reference>{
CallbackRecord<Reference>{callback_record_error, std::move(ex)}; detail::callback_record_error, std::move(ex)};
state->recordPublished.post(); state->recordPublished.post();
} }
} }
...@@ -150,7 +308,8 @@ AsyncGenerator<Reference, Value> merge( ...@@ -150,7 +308,8 @@ AsyncGenerator<Reference, Value> merge(
// Stream not yet been terminated with an error. // Stream not yet been terminated with an error.
// Terminate the stream with the 'end()' signal. // Terminate the stream with the 'end()' signal.
assert(!state->record.hasValue()); assert(!state->record.hasValue());
state->record = CallbackRecord<Reference>{callback_record_none}; state->record =
detail::CallbackRecord<Reference>{detail::callback_record_none};
state->recordPublished.post(); state->recordPublished.post();
} }
}; };
......
/*
* Copyright (c) Meta Platforms, Inc. and 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/coro/Materialize.h>
#include <folly/experimental/coro/Merge.h>
#include <folly/experimental/coro/Transform.h>
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
template <
typename SelectIdFn,
typename Reference,
typename Value,
typename KeyType>
AsyncGenerator<
Enumerated<KeyType, CallbackRecord<Reference>>,
Enumerated<KeyType, CallbackRecord<Value>>>
multiplex(
folly::Executor::KeepAlive<> exec,
AsyncGenerator<AsyncGenerator<Reference, Value>>&& sources,
SelectIdFn&& selectId) {
using EventType = CallbackRecord<Reference>;
using ReferenceType = Enumerated<KeyType, EventType>;
return merge(
std::move(exec),
transform(
std::move(sources),
[selectId = std::forward<SelectIdFn>(selectId)](
AsyncGenerator<Reference, Value>&& item) mutable {
KeyType id = invoke(selectId, item);
return transform(
materialize(std::move(item)),
[id = std::move(id)](EventType&& event) {
return ReferenceType{id, std::move(event)};
});
}));
}
struct MultiplexIdcountFn {
size_t n = 0;
template <typename Inner>
size_t operator()(Inner&&) noexcept {
return n++;
}
};
template <typename Reference, typename Value>
AsyncGenerator<
Enumerated<size_t, CallbackRecord<Reference>>,
Enumerated<size_t, CallbackRecord<Value>>>
multiplex(
folly::Executor::KeepAlive<> exec,
AsyncGenerator<AsyncGenerator<Reference, Value>>&& sources) {
return multiplex(std::move(exec), std::move(sources), MultiplexIdcountFn{});
}
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
/*
* Copyright (c) Meta Platforms, Inc. and 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/ExceptionWrapper.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/Coroutine.h>
#include <folly/experimental/coro/Materialize.h>
#include <tuple>
#include <variant>
#if FOLLY_HAS_COROUTINES
namespace folly {
namespace coro {
template <typename Id, typename Value>
using Enumerated = std::tuple<Id, Value>;
// Multiplex the results of multiple streams into a single stream that
// contains all of the events of the input stream.
//
// The output is a stream of std::tuple<Id, Event> where the first tuple
// element is the result of a call to selectId(innerStream). The default
// selectId returns a size_t set to the index of the stream that the event
// came from and where Event is the CallbackRecord result of
// materialize(innerStream).
//
// Example:
// AsyncGenerator<AsyncGenerator<int>&&> streams();
//
// Task<void> consumer() {
// auto events = multiplex(streams());
// while (auto item != co_await events.next()) {
// auto&& [index, event] = *item;
// if (event.index() == 0) {
// // Value
// int value = std::get<0>(event);
// std::cout << index << " value " << value << "\n";
// } else if (event.index() == 1) {
// // End Of Stream
// std::cout << index << " end\n";
// } else {
// // Exception
// folly::exception_wrapper error = std::get<2>(event);
// std::cout << index << " error " << error.what() << "\n";
// }
// }
// }
template <
typename SelectIdFn,
typename Reference,
typename Value,
typename KeyType = std::decay_t<
invoke_result_t<SelectIdFn&, const AsyncGenerator<Reference, Value>&>>>
AsyncGenerator<
Enumerated<KeyType, CallbackRecord<Reference>>,
Enumerated<KeyType, CallbackRecord<Value>>>
multiplex(
folly::Executor::KeepAlive<> exec,
AsyncGenerator<AsyncGenerator<Reference, Value>>&& sources,
SelectIdFn&& selectId);
template <typename Reference, typename Value>
AsyncGenerator<
Enumerated<size_t, CallbackRecord<Reference>>,
Enumerated<size_t, CallbackRecord<Value>>>
multiplex(
folly::Executor::KeepAlive<> exec,
AsyncGenerator<AsyncGenerator<Reference, Value>>&& sources);
} // namespace coro
} // namespace folly
#endif // FOLLY_HAS_COROUTINES
#include <folly/experimental/coro/Multiplex-inl.h>
/*
* Copyright (c) Meta Platforms, Inc. and 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/Portability.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/BlockingWait.h>
#include <folly/experimental/coro/CurrentExecutor.h>
#include <folly/experimental/coro/Dematerialize.h>
#include <folly/experimental/coro/Task.h>
#include <folly/portability/GTest.h>
#if FOLLY_HAS_COROUTINES
using namespace folly::coro;
class DematerializeTest : public testing::Test {};
TEST_F(DematerializeTest, SimpleStream) {
struct MyError : std::exception {};
const int seenEndOfStream = 100;
const int seenError = 200;
std::vector<int> lastSeen(3, -1);
std::vector<int> expectedSeen = {
{seenEndOfStream, seenEndOfStream, seenError}};
auto selectStream = [](int index) -> AsyncGenerator<int> {
auto ints = [](int n) -> AsyncGenerator<int> {
for (int i = 0; i <= n; ++i) {
co_await co_reschedule_on_current_executor;
co_yield i;
}
};
auto failing = []() -> AsyncGenerator<int> {
co_yield 0;
co_yield 1;
throw MyError{};
};
if (index == 0) {
return ints(4);
} else if (index == 1) {
return ints(3);
}
return failing();
};
auto test = [&](int index) -> Task<void> {
auto generator = dematerialize(materialize(selectStream(index)));
try {
while (auto item = co_await generator.next()) {
auto value = *item;
CHECK(index >= 0 && index <= 2);
CHECK_EQ(lastSeen[index] + 1, value);
lastSeen[index] = value;
}
// None
if (index == 0) {
CHECK_EQ(4, lastSeen[index]);
} else if (index == 1) {
CHECK_EQ(3, lastSeen[index]);
} else {
// Stream 2 should have completed with an error not EndOfStream.
CHECK(false);
}
lastSeen[index] = seenEndOfStream;
} catch (const MyError&) {
// Error
CHECK_EQ(2, index);
CHECK_EQ(1, lastSeen[index]);
lastSeen[index] = seenError;
}
CHECK_EQ(expectedSeen[index], lastSeen[index]);
};
blockingWait(test(0));
blockingWait(test(1));
blockingWait(test(2));
}
TEST_F(DematerializeTest, GeneratorOfRValueReference) {
auto makeGenerator =
[]() -> folly::coro::AsyncGenerator<std::unique_ptr<int>&&> {
co_yield std::make_unique<int>(10);
auto ptr = std::make_unique<int>(20);
co_yield std::move(ptr);
CHECK(ptr == nullptr);
};
folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
auto gen = dematerialize(materialize(makeGenerator()));
auto result = co_await gen.next();
CHECK_EQ(10, **result);
// Don't move it to a local var.
result = co_await gen.next();
CHECK_EQ(20, **result);
auto ptr = *result; // Move it to a local var.
result = co_await gen.next();
CHECK(!result);
}());
}
struct MoveOnly {
explicit MoveOnly(int value) : value_(value) {}
MoveOnly(MoveOnly&& other) noexcept
: value_(std::exchange(other.value_, -1)) {}
~MoveOnly() {}
MoveOnly& operator=(MoveOnly&&) = delete;
int value() const { return value_; }
private:
int value_;
};
TEST_F(DematerializeTest, GeneratorOfMoveOnlyType) {
auto makeGenerator = []() -> folly::coro::AsyncGenerator<MoveOnly> {
MoveOnly rvalue(1);
co_yield std::move(rvalue);
CHECK_EQ(-1, rvalue.value());
co_yield MoveOnly(2);
};
folly::coro::blockingWait([&]() -> folly::coro::Task<void> {
auto gen = dematerialize(materialize(makeGenerator()));
auto result = co_await gen.next();
// NOTE: It's an error to dereference using '*it' as this returns a copy
// of the iterator's reference type, which in this case is 'MoveOnly'.
CHECK_EQ(1, result->value());
result = co_await gen.next();
CHECK_EQ(2, result->value());
result = co_await gen.next();
CHECK(!result);
}());
}
#endif
/*
* Copyright (c) Meta Platforms, Inc. and 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/Portability.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/BlockingWait.h>
#include <folly/experimental/coro/CurrentExecutor.h>
#include <folly/experimental/coro/Materialize.h>
#include <folly/experimental/coro/Task.h>
#include <folly/portability/GTest.h>
#if FOLLY_HAS_COROUTINES
using namespace folly::coro;
class MaterializeTest : public testing::Test {};
TEST_F(MaterializeTest, SimpleStream) {
struct MyError : std::exception {};
const int seenEndOfStream = 100;
const int seenError = 200;
std::vector<int> lastSeen(3, -1);
std::vector<int> expectedSeen = {
{seenEndOfStream, seenEndOfStream, seenError}};
auto selectStream = [](int index) -> AsyncGenerator<int> {
auto ints = [](int n) -> AsyncGenerator<int> {
for (int i = 0; i <= n; ++i) {
co_await co_reschedule_on_current_executor;
co_yield i;
}
};
auto failing = []() -> AsyncGenerator<int> {
co_yield 0;
co_yield 1;
throw MyError{};
};
if (index == 0) {
return ints(4);
} else if (index == 1) {
return ints(3);
}
return failing();
};
auto test = [&](int index) -> Task<void> {
auto generator = materialize(selectStream(index));
while (auto item = co_await generator.next()) {
auto event = std::move(*item);
CHECK(index >= 0 && index <= 2);
if (event.hasValue()) {
auto value = std::move(event).value();
CHECK_EQ(lastSeen[index] + 1, value);
lastSeen[index] = value;
} else if (event.hasNone()) {
// None
if (index == 0) {
CHECK_EQ(4, lastSeen[index]);
} else if (index == 1) {
CHECK_EQ(3, lastSeen[index]);
} else if (event.hasError()) {
// Stream 2 should have completed with an error not EndOfStream.
CHECK(false);
}
lastSeen[index] = seenEndOfStream;
} else {
// Error
CHECK_EQ(2, index);
CHECK_EQ(1, lastSeen[index]);
CHECK(std::move(event).error().is_compatible_with<MyError>());
lastSeen[index] = seenError;
}
}
CHECK_EQ(expectedSeen[index], lastSeen[index]);
};
blockingWait(test(0));
blockingWait(test(1));
blockingWait(test(2));
}
#endif
/*
* Copyright (c) Meta Platforms, Inc. and 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/Portability.h>
#include <folly/experimental/coro/AsyncGenerator.h>
#include <folly/experimental/coro/BlockingWait.h>
#include <folly/experimental/coro/CurrentExecutor.h>
#include <folly/experimental/coro/Multiplex.h>
#include <folly/experimental/coro/Task.h>
#include <folly/portability/GTest.h>
#if FOLLY_HAS_COROUTINES
using namespace folly::coro;
class MultiplexTest : public testing::Test {};
TEST_F(MultiplexTest, SimpleStream) {
struct MyError : std::exception {};
blockingWait([]() -> Task<void> {
auto makeStreamOfStreams = []() -> AsyncGenerator<AsyncGenerator<int>> {
auto ints = [](int n) -> AsyncGenerator<int> {
for (int i = 0; i <= n; ++i) {
co_await co_reschedule_on_current_executor;
co_yield i;
}
};
auto failing = []() -> AsyncGenerator<int> {
co_yield 0;
co_yield 1;
throw MyError{};
};
co_yield ints(4);
co_yield ints(3);
co_yield failing();
};
std::vector<int> lastSeen(3, -1);
const int seenEndOfStream = 100;
const int seenError = 200;
auto generator =
multiplex(co_await co_current_executor, makeStreamOfStreams());
while (auto item = co_await generator.next()) {
auto [index, event] = std::move(*item);
CHECK(index >= 0 && index <= 2);
if (event.hasValue()) {
auto value = std::move(event).value();
CHECK_EQ(lastSeen[index] + 1, value);
lastSeen[index] = value;
} else if (event.hasNone()) {
// None
if (index == 0) {
CHECK_EQ(4, lastSeen[index]);
} else if (index == 1) {
CHECK_EQ(3, lastSeen[index]);
} else {
// Stream 2 should have completed with an error not EndOfStream.
CHECK(false);
}
lastSeen[index] = seenEndOfStream;
} else if (event.hasError()) {
// Error
CHECK_EQ(2, index);
CHECK_EQ(1, lastSeen[index]);
CHECK(std::move(event).error().is_compatible_with<MyError>());
lastSeen[index] = seenError;
}
}
CHECK_EQ(seenEndOfStream, lastSeen[0]);
CHECK_EQ(seenEndOfStream, lastSeen[1]);
CHECK_EQ(seenError, lastSeen[2]);
}());
}
#endif
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