mirror of https://github.com/grpc/grpc.git
This reverts commit 11a8f66cca.
pull/31481/head
AJ Heller
2 years ago
committed by
GitHub
parent
27206c981c
commit
821443e9b5
20 changed files with 421 additions and 859 deletions
@ -0,0 +1,148 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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.
|
||||
|
||||
#ifndef GRPC_CORE_LIB_PROMISE_CALL_PUSH_PULL_H |
||||
#define GRPC_CORE_LIB_PROMISE_CALL_PUSH_PULL_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <assert.h> |
||||
|
||||
#include <type_traits> |
||||
|
||||
#include "absl/types/variant.h" |
||||
|
||||
#include "src/core/lib/gprpp/bitset.h" |
||||
#include "src/core/lib/gprpp/construct_destruct.h" |
||||
#include "src/core/lib/promise/detail/promise_like.h" |
||||
#include "src/core/lib/promise/detail/status.h" |
||||
#include "src/core/lib/promise/poll.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
namespace promise_detail { |
||||
|
||||
template <typename FMain, typename FPush, typename FPull> |
||||
class CallPushPull { |
||||
public: |
||||
CallPushPull(FMain f_main, FPush f_push, FPull f_pull) |
||||
: push_(std::move(f_push)), pull_(std::move(f_pull)) { |
||||
Construct(&main_, std::move(f_main)); |
||||
} |
||||
|
||||
CallPushPull(const CallPushPull&) = delete; |
||||
CallPushPull& operator=(const CallPushPull&) = delete; |
||||
CallPushPull(CallPushPull&& other) noexcept |
||||
: done_(other.done_), |
||||
push_(std::move(other.push_)), |
||||
pull_(std::move(other.pull_)) { |
||||
assert(!done_.is_set(kDoneMain)); |
||||
Construct(&main_, std::move(other.main_)); |
||||
} |
||||
|
||||
CallPushPull& operator=(CallPushPull&& other) noexcept { |
||||
assert(!done_.is_set(kDoneMain)); |
||||
done_ = other.done_; |
||||
assert(!done_.is_set(kDoneMain)); |
||||
push_ = std::move(other.push_); |
||||
main_ = std::move(other.main_); |
||||
pull_ = std::move(other.pull_); |
||||
return *this; |
||||
} |
||||
|
||||
~CallPushPull() { |
||||
if (done_.is_set(kDoneMain)) { |
||||
Destruct(&result_); |
||||
} else { |
||||
Destruct(&main_); |
||||
} |
||||
} |
||||
|
||||
using Result = |
||||
typename PollTraits<decltype(std::declval<PromiseLike<FMain>>()())>::Type; |
||||
|
||||
Poll<Result> operator()() { |
||||
if (!done_.is_set(kDonePush)) { |
||||
auto p = push_(); |
||||
if (auto* status = absl::get_if<kPollReadyIdx>(&p)) { |
||||
if (IsStatusOk(*status)) { |
||||
done_.set(kDonePush); |
||||
} else { |
||||
return StatusCast<Result>(std::move(*status)); |
||||
} |
||||
} |
||||
} |
||||
if (!done_.is_set(kDoneMain)) { |
||||
auto p = main_(); |
||||
if (auto* status = absl::get_if<kPollReadyIdx>(&p)) { |
||||
done_.set(kDoneMain); |
||||
Destruct(&main_); |
||||
Construct(&result_, std::move(*status)); |
||||
} |
||||
} |
||||
if (!done_.is_set(kDonePull)) { |
||||
auto p = pull_(); |
||||
if (auto* status = absl::get_if<kPollReadyIdx>(&p)) { |
||||
if (IsStatusOk(*status)) { |
||||
done_.set(kDonePull); |
||||
} else { |
||||
return StatusCast<Result>(std::move(*status)); |
||||
} |
||||
} |
||||
} |
||||
if (done_.all()) return std::move(result_); |
||||
return Pending{}; |
||||
} |
||||
|
||||
private: |
||||
enum { kDonePull = 0, kDoneMain = 1, kDonePush = 2 }; |
||||
BitSet<3> done_; |
||||
GPR_NO_UNIQUE_ADDRESS PromiseLike<FPush> push_; |
||||
union { |
||||
PromiseLike<FMain> main_; |
||||
Result result_; |
||||
}; |
||||
GPR_NO_UNIQUE_ADDRESS PromiseLike<FPull> pull_; |
||||
}; |
||||
|
||||
} // namespace promise_detail
|
||||
|
||||
// For promises representing calls a common pattern emerges:
|
||||
// There's a process pushing data down the stack, a process handling the main
|
||||
// call part, and a process pulling data back up the stack.
|
||||
//
|
||||
// This can reasonably be represented by the right combinations of TryJoins and
|
||||
// Maps, but since the structure is fundamental to the domain we introduce
|
||||
// this simple helper to make it easier to write the common case.
|
||||
//
|
||||
// It takes three promises: the main call, the push and the pull.
|
||||
// When polling, the push is polled first, then the main call (descending the
|
||||
// stack), then the pull (as we ascend once more).
|
||||
//
|
||||
// If the push or the pull fail early, then the entire call fails.
|
||||
// If the main part of the call fails, we wait until both push and pull are also
|
||||
// done.
|
||||
//
|
||||
// This strategy minimizes repolls.
|
||||
template <typename FMain, typename FPush, typename FPull> |
||||
promise_detail::CallPushPull<FMain, FPush, FPull> CallPushPull(FMain f_main, |
||||
FPush f_push, |
||||
FPull f_pull) { |
||||
return promise_detail::CallPushPull<FMain, FPush, FPull>( |
||||
std::move(f_main), std::move(f_push), std::move(f_pull)); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_CORE_LIB_PROMISE_CALL_PUSH_PULL_H
|
||||
@ -1,341 +0,0 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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.
|
||||
|
||||
#ifndef GRPC_CORE_LIB_PROMISE_TRY_CONCURRENTLY_H |
||||
#define GRPC_CORE_LIB_PROMISE_TRY_CONCURRENTLY_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <stddef.h> |
||||
|
||||
#include <cstdint> |
||||
#include <utility> |
||||
|
||||
#include "absl/types/variant.h" |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/gprpp/construct_destruct.h" |
||||
#include "src/core/lib/promise/detail/promise_like.h" |
||||
#include "src/core/lib/promise/detail/status.h" |
||||
#include "src/core/lib/promise/poll.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
namespace promise_detail { |
||||
|
||||
template <typename Promise> |
||||
struct Necessary { |
||||
PromiseLike<Promise> promise; |
||||
static constexpr bool must_complete() { return true; } |
||||
}; |
||||
|
||||
template <typename Promise> |
||||
struct Helper { |
||||
PromiseLike<Promise> promise; |
||||
static constexpr bool must_complete() { return false; } |
||||
}; |
||||
|
||||
// A set of promises that can be polled concurrently.
|
||||
// Fuses them when completed (that is, destroys the promise and records it
|
||||
// completed).
|
||||
// Relies on an external bit field to handle the recording: this saves a bunch
|
||||
// of space, but means the implementation of this type is weird: it's really
|
||||
// super tied to TryConcurrently and no attempt should be made to use this
|
||||
// independently.
|
||||
template <typename... Ts> |
||||
class FusedSet; |
||||
|
||||
template <typename T, typename... Ts> |
||||
class FusedSet<T, Ts...> : public FusedSet<Ts...> { |
||||
public: |
||||
explicit FusedSet(T&& x, Ts&&... xs) |
||||
: FusedSet<Ts...>(std::forward<T>(xs)...) { |
||||
Construct(&wrapper_, std::forward<T>(x)); |
||||
} |
||||
explicit FusedSet(T&& x, FusedSet<Ts...>&& xs) |
||||
: FusedSet<Ts...>(std::forward<FusedSet<Ts...>>(xs)) { |
||||
Construct(&wrapper_, std::forward<T>(x)); |
||||
} |
||||
// Empty destructor: consumers must call Destroy() to ensure cleanup occurs
|
||||
~FusedSet() {} |
||||
|
||||
FusedSet(const FusedSet&) = delete; |
||||
FusedSet& operator=(const FusedSet&) = delete; |
||||
|
||||
// Assumes all 'done_bits' for other are 0 and will be set to 1
|
||||
FusedSet(FusedSet&& other) noexcept : FusedSet<Ts...>(std::move(other)) { |
||||
Construct(&wrapper_, std::move(other.wrapper_)); |
||||
Destruct(&other.wrapper_); |
||||
} |
||||
|
||||
static constexpr size_t Size() { return 1 + sizeof...(Ts); } |
||||
|
||||
static constexpr uint8_t NecessaryBits() { |
||||
return (T::must_complete() ? 1 : 0) | |
||||
(FusedSet<Ts...>::NecessaryBits() << 1); |
||||
} |
||||
|
||||
template <int kDoneBit> |
||||
void Destroy(uint8_t done_bits) { |
||||
if ((done_bits & (1 << kDoneBit)) == 0) { |
||||
Destruct(&wrapper_); |
||||
} |
||||
FusedSet<Ts...>::template Destroy<kDoneBit + 1>(done_bits); |
||||
} |
||||
|
||||
template <typename Result, int kDoneBit> |
||||
Poll<Result> Run(uint8_t& done_bits) { |
||||
if ((done_bits & (1 << kDoneBit)) == 0) { |
||||
auto p = wrapper_.promise(); |
||||
if (auto* status = absl::get_if<kPollReadyIdx>(&p)) { |
||||
done_bits |= (1 << kDoneBit); |
||||
Destruct(&wrapper_); |
||||
if (!IsStatusOk(*status)) { |
||||
return StatusCast<Result>(std::move(*status)); |
||||
} |
||||
} |
||||
} |
||||
return FusedSet<Ts...>::template Run<Result, kDoneBit + 1>(done_bits); |
||||
} |
||||
|
||||
template <typename P> |
||||
FusedSet<P, T, Ts...> With(P x) { |
||||
return FusedSet<P, T, Ts...>(std::move(x), std::move(*this)); |
||||
} |
||||
|
||||
private: |
||||
union { |
||||
T wrapper_; |
||||
}; |
||||
}; |
||||
|
||||
template <> |
||||
class FusedSet<> { |
||||
public: |
||||
static constexpr size_t Size() { return 0; } |
||||
static constexpr uint8_t NecessaryBits() { return 0; } |
||||
|
||||
template <typename Result, int kDoneBit> |
||||
Poll<Result> Run(uint8_t) { |
||||
return Pending{}; |
||||
} |
||||
template <int kDoneBit> |
||||
void Destroy(uint8_t) {} |
||||
|
||||
template <typename P> |
||||
FusedSet<P> With(P x) { |
||||
return FusedSet<P>(std::move(x)); |
||||
} |
||||
}; |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
class TryConcurrently { |
||||
public: |
||||
TryConcurrently(Main main, PreMain pre_main, PostMain post_main) |
||||
: done_bits_(0), |
||||
pre_main_(std::move(pre_main)), |
||||
post_main_(std::move(post_main)) { |
||||
Construct(&main_, std::move(main)); |
||||
} |
||||
|
||||
TryConcurrently(const TryConcurrently&) = delete; |
||||
TryConcurrently& operator=(const TryConcurrently&) = delete; |
||||
TryConcurrently(TryConcurrently&& other) noexcept |
||||
: done_bits_(0), |
||||
pre_main_(std::move(other.pre_main_)), |
||||
post_main_(std::move(other.post_main_)) { |
||||
GPR_DEBUG_ASSERT(other.done_bits_ == 0); |
||||
other.done_bits_ = HelperBits(); |
||||
Construct(&main_, std::move(other.main_)); |
||||
} |
||||
TryConcurrently& operator=(TryConcurrently&& other) noexcept { |
||||
GPR_DEBUG_ASSERT(other.done_bits_ == 0); |
||||
done_bits_ = 0; |
||||
other.done_bits_ = HelperBits(); |
||||
pre_main_ = std::move(other.pre_main_); |
||||
post_main_ = std::move(other.post_main_); |
||||
Construct(&main_, std::move(other.main_)); |
||||
return *this; |
||||
} |
||||
|
||||
~TryConcurrently() { |
||||
if (done_bits_ & 1) { |
||||
Destruct(&result_); |
||||
} else { |
||||
Destruct(&main_); |
||||
} |
||||
pre_main_.template Destroy<1>(done_bits_); |
||||
post_main_.template Destroy<1 + PreMain::Size()>(done_bits_); |
||||
} |
||||
|
||||
using Result = |
||||
typename PollTraits<decltype(std::declval<PromiseLike<Main>>()())>::Type; |
||||
|
||||
Poll<Result> operator()() { |
||||
auto r = pre_main_.template Run<Result, 1>(done_bits_); |
||||
if (auto* status = absl::get_if<Result>(&r)) { |
||||
GPR_DEBUG_ASSERT(!IsStatusOk(*status)); |
||||
return std::move(*status); |
||||
} |
||||
if ((done_bits_ & 1) == 0) { |
||||
auto p = main_(); |
||||
if (auto* status = absl::get_if<kPollReadyIdx>(&p)) { |
||||
done_bits_ |= 1; |
||||
Destruct(&main_); |
||||
Construct(&result_, std::move(*status)); |
||||
} |
||||
} |
||||
r = post_main_.template Run<Result, 1 + PreMain::Size()>(done_bits_); |
||||
if (auto* status = absl::get_if<Result>(&r)) { |
||||
GPR_DEBUG_ASSERT(!IsStatusOk(*status)); |
||||
return std::move(*status); |
||||
} |
||||
if ((done_bits_ & NecessaryBits()) == NecessaryBits()) { |
||||
return std::move(result_); |
||||
} |
||||
return Pending{}; |
||||
} |
||||
|
||||
template <typename P> |
||||
auto NecessaryPush(P p); |
||||
template <typename P> |
||||
auto NecessaryPull(P p); |
||||
template <typename P> |
||||
auto Push(P p); |
||||
template <typename P> |
||||
auto Pull(P p); |
||||
|
||||
private: |
||||
// Bitmask for done_bits_ specifying which promises must be completed prior to
|
||||
// returning ok.
|
||||
constexpr uint8_t NecessaryBits() { |
||||
return 1 | (PreMain::NecessaryBits() << 1) | |
||||
(PostMain::NecessaryBits() << (1 + PreMain::Size())); |
||||
} |
||||
// Bitmask for done_bits_ specifying what all of the promises being complete
|
||||
// would look like.
|
||||
constexpr uint8_t AllBits() { |
||||
return (1 << (1 + PreMain::Size() + PostMain::Size())) - 1; |
||||
} |
||||
// Bitmask of done_bits_ specifying which bits correspond to helper promises -
|
||||
// that is all promises that are not the main one.
|
||||
constexpr uint8_t HelperBits() { return AllBits() ^ 1; } |
||||
|
||||
// done_bits signifies which operations have completed.
|
||||
// Bit 0 is set if main_ has completed.
|
||||
// The next higher bits correspond one per pre-main promise.
|
||||
// The next higher bits correspond one per post-main promise.
|
||||
// So, going from most significant bit to least significant:
|
||||
// +--------------+-------------+--------+
|
||||
// |post_main bits|pre_main bits|main bit|
|
||||
// +--------------+-------------+--------+
|
||||
uint8_t done_bits_; |
||||
PreMain pre_main_; |
||||
union { |
||||
PromiseLike<Main> main_; |
||||
Result result_; |
||||
}; |
||||
PostMain post_main_; |
||||
}; |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
auto MakeTryConcurrently(Main&& main, PreMain&& pre_main, |
||||
PostMain&& post_main) { |
||||
return TryConcurrently<Main, PreMain, PostMain>( |
||||
std::forward<Main>(main), std::forward<PreMain>(pre_main), |
||||
std::forward<PostMain>(post_main)); |
||||
} |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
template <typename P> |
||||
auto TryConcurrently<Main, PreMain, PostMain>::NecessaryPush(P p) { |
||||
GPR_DEBUG_ASSERT(done_bits_ == 0); |
||||
done_bits_ = HelperBits(); |
||||
return MakeTryConcurrently(std::move(main_), |
||||
pre_main_.With(Necessary<P>{std::move(p)}), |
||||
std::move(post_main_)); |
||||
} |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
template <typename P> |
||||
auto TryConcurrently<Main, PreMain, PostMain>::NecessaryPull(P p) { |
||||
GPR_DEBUG_ASSERT(done_bits_ == 0); |
||||
done_bits_ = HelperBits(); |
||||
return MakeTryConcurrently(std::move(main_), std::move(pre_main_), |
||||
post_main_.With(Necessary<P>{std::move(p)})); |
||||
} |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
template <typename P> |
||||
auto TryConcurrently<Main, PreMain, PostMain>::Push(P p) { |
||||
GPR_DEBUG_ASSERT(done_bits_ == 0); |
||||
done_bits_ = HelperBits(); |
||||
return MakeTryConcurrently(std::move(main_), |
||||
pre_main_.With(Helper<P>{std::move(p)}), |
||||
std::move(post_main_)); |
||||
} |
||||
|
||||
template <typename Main, typename PreMain, typename PostMain> |
||||
template <typename P> |
||||
auto TryConcurrently<Main, PreMain, PostMain>::Pull(P p) { |
||||
GPR_DEBUG_ASSERT(done_bits_ == 0); |
||||
done_bits_ = HelperBits(); |
||||
return MakeTryConcurrently(std::move(main_), std::move(pre_main_), |
||||
post_main_.With(Helper<P>{std::move(p)})); |
||||
} |
||||
|
||||
} // namespace promise_detail
|
||||
|
||||
// TryConcurrently runs a set of promises concurrently.
|
||||
// There is a structure to the promises:
|
||||
// - A 'main' promise dominates the others - it must complete before the
|
||||
// overall promise successfully completes. Its result is chosen in the event
|
||||
// of successful completion.
|
||||
// - A set of (optional) push and pull promises to aid main. Push promises are
|
||||
// polled before main, pull promises are polled after. In this way we can
|
||||
// avoid overall wakeup churn - sending a message will tend to push things
|
||||
// down the promise tree as its polled, so that send should be in a push
|
||||
// promise - then as the main promise is polled and it calls into things
|
||||
// lower in the stack they'll already see things there (this reasoning holds
|
||||
// for receiving things and the pull promises too!).
|
||||
// - Each push and pull promise is either necessary or optional.
|
||||
// Necessary promises must complete successfully before the overall promise
|
||||
// completes. Optional promises will just be cancelled once the main promise
|
||||
// completes and any necessary helpers.
|
||||
// - If any of the promises fail, the overall promise fails immediately.
|
||||
// API:
|
||||
// This function, TryConcurrently, is used to create a TryConcurrently promise.
|
||||
// It takes a single argument, being the main promise. That promise also has
|
||||
// a set of methods for attaching push and pull promises. The act of attachment
|
||||
// returns a new TryConcurrently promise with previous contained promises moved
|
||||
// out.
|
||||
// The methods exposed:
|
||||
// - Push, NecessaryPush: attach a push promise (with the first variant being
|
||||
// optional, the second necessary).
|
||||
// - Pull, NecessaryPull: attach a pull promise, with variants as above.
|
||||
// Example:
|
||||
// TryConcurrently(call_next_filter(std::move(call_args)))
|
||||
// .Push(send_messages_promise)
|
||||
// .Pull(recv_messages_promise)
|
||||
template <typename Main> |
||||
auto TryConcurrently(Main main) { |
||||
return promise_detail::MakeTryConcurrently(std::move(main), |
||||
promise_detail::FusedSet<>(), |
||||
promise_detail::FusedSet<>()); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_CORE_LIB_PROMISE_TRY_CONCURRENTLY_H
|
||||
@ -0,0 +1,77 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 "src/core/lib/promise/call_push_pull.h" |
||||
|
||||
#include <utility> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
TEST(CallPushPullTest, Empty) { |
||||
auto p = CallPushPull([] { return absl::OkStatus(); }, |
||||
[] { return absl::OkStatus(); }, |
||||
[] { return absl::OkStatus(); }); |
||||
EXPECT_EQ(p(), Poll<absl::Status>(absl::OkStatus())); |
||||
} |
||||
|
||||
TEST(CallPushPullTest, Paused) { |
||||
auto p = CallPushPull([]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }); |
||||
EXPECT_EQ(p(), Poll<absl::Status>(Pending{})); |
||||
} |
||||
|
||||
TEST(CallPushPullTest, OneReady) { |
||||
auto a = CallPushPull([]() -> Poll<absl::Status> { return absl::OkStatus(); }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }); |
||||
EXPECT_EQ(a(), Poll<absl::Status>(Pending{})); |
||||
auto b = CallPushPull([]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return absl::OkStatus(); }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }); |
||||
EXPECT_EQ(b(), Poll<absl::Status>(Pending{})); |
||||
auto c = |
||||
CallPushPull([]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return absl::OkStatus(); }); |
||||
EXPECT_EQ(c(), Poll<absl::Status>(Pending{})); |
||||
} |
||||
|
||||
TEST(CallPushPullTest, OneFailed) { |
||||
auto a = CallPushPull( |
||||
[]() -> Poll<absl::Status> { return absl::UnknownError("bah"); }, |
||||
[]() -> Poll<absl::Status> { return absl::OkStatus(); }, |
||||
[]() -> Poll<absl::Status> { return absl::OkStatus(); }); |
||||
EXPECT_EQ(a(), Poll<absl::Status>(absl::UnknownError("bah"))); |
||||
auto b = CallPushPull( |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return absl::UnknownError("humbug"); }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }); |
||||
EXPECT_EQ(b(), Poll<absl::Status>(absl::UnknownError("humbug"))); |
||||
auto c = CallPushPull( |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return Pending{}; }, |
||||
[]() -> Poll<absl::Status> { return absl::UnknownError("wha"); }); |
||||
EXPECT_EQ(c(), Poll<absl::Status>(absl::UnknownError("wha"))); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
@ -1,160 +0,0 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 "src/core/lib/promise/try_concurrently.h" |
||||
|
||||
#include <algorithm> |
||||
#include <iosfwd> |
||||
#include <memory> |
||||
#include <string> |
||||
#include <vector> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
class PromiseFactory { |
||||
public: |
||||
// Create a promise that resolves to Ok but has a memory allocation (to verify
|
||||
// destruction)
|
||||
auto OkPromise(std::string tag) { |
||||
return [this, tag = std::move(tag), |
||||
p = std::make_unique<absl::Status>(absl::OkStatus())]() mutable { |
||||
order_.push_back(tag); |
||||
return std::move(*p); |
||||
}; |
||||
} |
||||
|
||||
// Create a promise that never resolves and carries a memory allocation
|
||||
auto NeverPromise(std::string tag) { |
||||
return [this, tag = std::move(tag), |
||||
p = std::make_unique<Pending>()]() -> Poll<absl::Status> { |
||||
order_.push_back(tag); |
||||
return *p; |
||||
}; |
||||
} |
||||
|
||||
// Create a promise that fails and carries a memory allocation
|
||||
auto FailPromise(std::string tag) { |
||||
return [this, p = std::make_unique<absl::Status>(absl::UnknownError(tag)), |
||||
tag = std::move(tag)]() mutable { |
||||
order_.push_back(tag); |
||||
return std::move(*p); |
||||
}; |
||||
} |
||||
|
||||
// Finish one round and return a vector of strings representing which promises
|
||||
// were polled and in which order.
|
||||
std::vector<std::string> Finish() { return std::exchange(order_, {}); } |
||||
|
||||
private: |
||||
std::vector<std::string> order_; |
||||
}; |
||||
|
||||
std::ostream& operator<<(std::ostream& out, const Poll<absl::Status>& p) { |
||||
return out << PollToString( |
||||
p, [](const absl::Status& s) { return s.ToString(); }); |
||||
} |
||||
|
||||
TEST(TryConcurrentlyTest, Immediate) { |
||||
PromiseFactory pf; |
||||
auto a = TryConcurrently(pf.OkPromise("1")); |
||||
EXPECT_EQ(a(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1"})); |
||||
auto b = TryConcurrently(pf.OkPromise("1")).NecessaryPush(pf.OkPromise("2")); |
||||
EXPECT_EQ(b(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"2", "1"})); |
||||
auto c = TryConcurrently(pf.OkPromise("1")).NecessaryPull(pf.OkPromise("2")); |
||||
EXPECT_EQ(c(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1", "2"})); |
||||
auto d = TryConcurrently(pf.OkPromise("1")) |
||||
.NecessaryPull(pf.OkPromise("2")) |
||||
.NecessaryPush(pf.OkPromise("3")); |
||||
EXPECT_EQ(d(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"3", "1", "2"})); |
||||
auto e = TryConcurrently(pf.OkPromise("1")).Push(pf.NeverPromise("2")); |
||||
EXPECT_EQ(e(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"2", "1"})); |
||||
auto f = TryConcurrently(pf.OkPromise("1")).Pull(pf.NeverPromise("2")); |
||||
EXPECT_EQ(f(), Poll<absl::Status>(absl::OkStatus())); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1", "2"})); |
||||
} |
||||
|
||||
TEST(TryConcurrentlyTest, Paused) { |
||||
PromiseFactory pf; |
||||
auto a = TryConcurrently(pf.NeverPromise("1")); |
||||
EXPECT_EQ(a(), Poll<absl::Status>(Pending{})); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1"})); |
||||
auto b = |
||||
TryConcurrently(pf.OkPromise("1")).NecessaryPush(pf.NeverPromise("2")); |
||||
EXPECT_EQ(b(), Poll<absl::Status>(Pending{})); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"2", "1"})); |
||||
auto c = |
||||
TryConcurrently(pf.OkPromise("1")).NecessaryPull(pf.NeverPromise("2")); |
||||
EXPECT_EQ(c(), Poll<absl::Status>(Pending{})); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1", "2"})); |
||||
} |
||||
|
||||
TEST(TryConcurrentlyTest, OneFailed) { |
||||
PromiseFactory pf; |
||||
auto a = TryConcurrently(pf.FailPromise("bah")); |
||||
EXPECT_EQ(a(), Poll<absl::Status>(absl::UnknownError("bah"))); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"bah"})); |
||||
auto b = TryConcurrently(pf.NeverPromise("1")) |
||||
.NecessaryPush(pf.FailPromise("humbug")); |
||||
EXPECT_EQ(b(), Poll<absl::Status>(absl::UnknownError("humbug"))); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"humbug"})); |
||||
auto c = TryConcurrently(pf.NeverPromise("1")) |
||||
.NecessaryPull(pf.FailPromise("wha")); |
||||
EXPECT_EQ(c(), Poll<absl::Status>(absl::UnknownError("wha"))); |
||||
EXPECT_EQ(pf.Finish(), std::vector<std::string>({"1", "wha"})); |
||||
} |
||||
|
||||
// A pointer to an int designed to cause a double free if it's double destructed
|
||||
// (to flush out bugs)
|
||||
class ProblematicPointer { |
||||
public: |
||||
ProblematicPointer() : p_(new int(0)) {} |
||||
~ProblematicPointer() { delete p_; } |
||||
ProblematicPointer(const ProblematicPointer&) = delete; |
||||
ProblematicPointer& operator=(const ProblematicPointer&) = delete; |
||||
// NOLINTNEXTLINE: we want to allocate during move
|
||||
ProblematicPointer(ProblematicPointer&& other) : p_(new int(*other.p_ + 1)) {} |
||||
ProblematicPointer& operator=(ProblematicPointer&& other) = delete; |
||||
|
||||
private: |
||||
int* p_; |
||||
}; |
||||
|
||||
TEST(TryConcurrentlyTest, MoveItMoveIt) { |
||||
auto a = |
||||
TryConcurrently([x = ProblematicPointer()]() { return absl::OkStatus(); }) |
||||
.NecessaryPull( |
||||
[x = ProblematicPointer()]() { return absl::OkStatus(); }) |
||||
.NecessaryPush( |
||||
[x = ProblematicPointer()]() { return absl::OkStatus(); }) |
||||
.Push([x = ProblematicPointer()]() { return absl::OkStatus(); }) |
||||
.Pull([x = ProblematicPointer()]() { return absl::OkStatus(); }); |
||||
auto b = std::move(a); |
||||
auto c = std::move(b); |
||||
c(); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue