[promise] Introduce map_pipe, cleanup factories (#31430)
* [promise] Introduce map_pipe, cleanup factories * Automated change: Fix sanity tests Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/31462/head
Craig Tiller
2 years ago
committed by
GitHub
parent
ab3d62ae8f
commit
58c628a7ad
29 changed files with 664 additions and 56 deletions
@ -0,0 +1,87 @@ |
||||
// 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_MAP_PIPE_H |
||||
#define GRPC_CORE_LIB_PROMISE_MAP_PIPE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "absl/status/status.h" |
||||
|
||||
#include "src/core/lib/promise/detail/promise_factory.h" |
||||
#include "src/core/lib/promise/for_each.h" |
||||
#include "src/core/lib/promise/map.h" |
||||
#include "src/core/lib/promise/pipe.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
// Apply a (possibly async) mapping function to src, and output into dst.
|
||||
//
|
||||
// In psuedo-code:
|
||||
// for each element in wait_for src.Next:
|
||||
// x = wait_for filter_factory(element)
|
||||
// wait_for dst.Push(x)
|
||||
template <typename T, typename Filter> |
||||
auto MapPipe(PipeReceiver<T> src, PipeSender<T> dst, Filter filter_factory) { |
||||
return ForEach( |
||||
std::move(src), |
||||
[filter_factory = promise_detail::RepeatedPromiseFactory<T, Filter>( |
||||
std::move(filter_factory)), |
||||
dst = std::move(dst)](T t) mutable { |
||||
return TrySeq(filter_factory.Make(std::move(t)), [&dst](T t) { |
||||
return Map(dst.Push(std::move(t)), [](bool successful_push) { |
||||
if (successful_push) { |
||||
return absl::OkStatus(); |
||||
} |
||||
return absl::CancelledError(); |
||||
}); |
||||
}); |
||||
}); |
||||
} |
||||
|
||||
// Helper to intecept a pipe and apply a mapping function.
|
||||
// Each of the `Intercept` constructors will take a PipeSender or PipeReceiver,
|
||||
// construct a new pipe, and then replace the passed in pipe with its new end.
|
||||
// In this way it can interject logic per-element.
|
||||
// Next, the TakeAndRun function will return a promise that can be run to apply
|
||||
// a mapping promise to each element of the pipe.
|
||||
template <typename T> |
||||
class PipeMapper { |
||||
public: |
||||
static PipeMapper Intercept(PipeSender<T>& intercept_sender) { |
||||
PipeMapper<T> r; |
||||
r.interceptor_.sender.Swap(&intercept_sender); |
||||
return r; |
||||
} |
||||
|
||||
static PipeMapper Intercept(PipeReceiver<T>& intercept_receiver) { |
||||
PipeMapper<T> r; |
||||
r.interceptor_.receiver.Swap(&intercept_receiver); |
||||
return r; |
||||
} |
||||
|
||||
template <typename Filter> |
||||
auto TakeAndRun(Filter filter) { |
||||
return MapPipe(std::move(interceptor_.receiver), |
||||
std::move(interceptor_.sender), std::move(filter)); |
||||
} |
||||
|
||||
private: |
||||
PipeMapper() = default; |
||||
Pipe<T> interceptor_; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_CORE_LIB_PROMISE_MAP_PIPE_H
|
||||
@ -0,0 +1,19 @@ |
||||
// 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 <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/promise/pipe.h" |
||||
|
||||
grpc_core::DebugOnlyTraceFlag grpc_trace_promise_pipe(false, "promise_pipe"); |
||||
@ -0,0 +1,153 @@ |
||||
// 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/map_pipe.h" |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include <memory> |
||||
#include <utility> |
||||
|
||||
#include "gmock/gmock.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include <grpc/event_engine/memory_allocator.h> |
||||
|
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/promise/activity.h" |
||||
#include "src/core/lib/promise/detail/basic_seq.h" |
||||
#include "src/core/lib/promise/for_each.h" |
||||
#include "src/core/lib/promise/join.h" |
||||
#include "src/core/lib/promise/map.h" |
||||
#include "src/core/lib/promise/pipe.h" |
||||
#include "src/core/lib/promise/poll.h" |
||||
#include "src/core/lib/promise/seq.h" |
||||
#include "src/core/lib/promise/try_seq.h" |
||||
#include "src/core/lib/resource_quota/arena.h" |
||||
#include "src/core/lib/resource_quota/memory_quota.h" |
||||
#include "src/core/lib/resource_quota/resource_quota.h" |
||||
#include "test/core/promise/test_wakeup_schedulers.h" |
||||
|
||||
using testing::Mock; |
||||
using testing::MockFunction; |
||||
using testing::StrictMock; |
||||
|
||||
namespace grpc_core { |
||||
|
||||
static auto* g_memory_allocator = new MemoryAllocator( |
||||
ResourceQuota::Default()->memory_quota()->CreateMemoryAllocator("test")); |
||||
|
||||
template <typename T> |
||||
class Delayed { |
||||
public: |
||||
explicit Delayed(T x) : x_(x) {} |
||||
|
||||
Poll<T> operator()() { |
||||
Activity::current()->ForceImmediateRepoll(); |
||||
++polls_; |
||||
if (polls_ == 10) return std::move(x_); |
||||
return Pending(); |
||||
} |
||||
|
||||
private: |
||||
int polls_ = 0; |
||||
T x_; |
||||
}; |
||||
|
||||
TEST(MapPipeTest, SendThriceWithPipeInterceptingReceive) { |
||||
int num_received = 0; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
MakeActivity( |
||||
[&num_received] { |
||||
Pipe<int> pipe; |
||||
auto filter = |
||||
PipeMapper<int>::Intercept(pipe.receiver).TakeAndRun([](int x) { |
||||
return Delayed<int>(x + 1); |
||||
}); |
||||
auto sender = std::make_shared<std::unique_ptr<PipeSender<int>>>( |
||||
std::make_unique<PipeSender<int>>(std::move(pipe.sender))); |
||||
return Map( |
||||
Join( |
||||
std::move(filter), |
||||
// Push 3 things into a pipe -- 0, 1, then 2 -- then close.
|
||||
Seq((*sender)->Push(0), [sender] { return (*sender)->Push(1); }, |
||||
[sender] { return (*sender)->Push(2); }, |
||||
[sender] { |
||||
sender->reset(); |
||||
return absl::OkStatus(); |
||||
}), |
||||
// Use a ForEach loop to read them out and verify all values are
|
||||
// seen (but with 1 added).
|
||||
ForEach(std::move(pipe.receiver), |
||||
[&num_received](int i) { |
||||
num_received++; |
||||
EXPECT_EQ(num_received, i); |
||||
return absl::OkStatus(); |
||||
})), |
||||
JustElem<2>()); |
||||
}, |
||||
NoWakeupScheduler(), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }, |
||||
MakeScopedArena(1024, g_memory_allocator)); |
||||
Mock::VerifyAndClearExpectations(&on_done); |
||||
EXPECT_EQ(num_received, 3); |
||||
} |
||||
|
||||
TEST(MapPipeTest, SendThriceWithPipeInterceptingSend) { |
||||
int num_received = 0; |
||||
StrictMock<MockFunction<void(absl::Status)>> on_done; |
||||
EXPECT_CALL(on_done, Call(absl::OkStatus())); |
||||
MakeActivity( |
||||
[&num_received] { |
||||
Pipe<int> pipe; |
||||
auto filter = |
||||
PipeMapper<int>::Intercept(pipe.sender).TakeAndRun([](int x) { |
||||
return Delayed<int>(x + 1); |
||||
}); |
||||
auto sender = std::make_shared<std::unique_ptr<PipeSender<int>>>( |
||||
std::make_unique<PipeSender<int>>(std::move(pipe.sender))); |
||||
return Map( |
||||
Join( |
||||
std::move(filter), |
||||
// Push 3 things into a pipe -- 0, 1, then 2 -- then close.
|
||||
Seq((*sender)->Push(0), [sender] { return (*sender)->Push(1); }, |
||||
[sender] { return (*sender)->Push(2); }, |
||||
[sender] { |
||||
sender->reset(); |
||||
return absl::OkStatus(); |
||||
}), |
||||
// Use a ForEach loop to read them out and verify all values are
|
||||
// seen (but with 1 added).
|
||||
ForEach(std::move(pipe.receiver), |
||||
[&num_received](int i) { |
||||
num_received++; |
||||
EXPECT_EQ(num_received, i); |
||||
return absl::OkStatus(); |
||||
})), |
||||
JustElem<2>()); |
||||
}, |
||||
NoWakeupScheduler(), |
||||
[&on_done](absl::Status status) { on_done.Call(std::move(status)); }, |
||||
MakeScopedArena(1024, g_memory_allocator)); |
||||
Mock::VerifyAndClearExpectations(&on_done); |
||||
EXPECT_EQ(num_received, 3); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue