[promises] Inter-activity pipe (#34188)
Pipe-like type (has a send end, a receive end, and a closing mechanism) for cross-activity transfers. --------- Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/34211/head
Craig Tiller
2 years ago
committed by
GitHub
parent
c405f75a5a
commit
60c6b6bb3b
8 changed files with 429 additions and 0 deletions
@ -0,0 +1,148 @@ |
||||
// Copyright 2023 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_SRC_CORE_LIB_PROMISE_INTER_ACTIVITY_PIPE_H |
||||
#define GRPC_SRC_CORE_LIB_PROMISE_INTER_ACTIVITY_PIPE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include <array> |
||||
#include <utility> |
||||
|
||||
#include "absl/base/thread_annotations.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include "src/core/lib/gprpp/ref_counted.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/gprpp/sync.h" |
||||
#include "src/core/lib/promise/activity.h" |
||||
#include "src/core/lib/promise/poll.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
template <typename T, uint8_t kQueueSize> |
||||
class InterActivityPipe { |
||||
private: |
||||
class Center : public RefCounted<Center, NonPolymorphicRefCount> { |
||||
public: |
||||
Poll<bool> Push(T& value) { |
||||
ReleasableMutexLock lock(&mu_); |
||||
if (closed_) return false; |
||||
if (count_ == kQueueSize) { |
||||
on_available_ = Activity::current()->MakeNonOwningWaker(); |
||||
return Pending{}; |
||||
} |
||||
queue_[(first_ + count_) % kQueueSize] = std::move(value); |
||||
++count_; |
||||
if (count_ == 1) { |
||||
auto on_occupied = std::move(on_occupied_); |
||||
lock.Release(); |
||||
on_occupied.Wakeup(); |
||||
} |
||||
return true; |
||||
} |
||||
|
||||
Poll<absl::optional<T>> Next() { |
||||
ReleasableMutexLock lock(&mu_); |
||||
if (count_ == 0) { |
||||
if (closed_) return absl::nullopt; |
||||
on_occupied_ = Activity::current()->MakeNonOwningWaker(); |
||||
return Pending{}; |
||||
} |
||||
auto value = std::move(queue_[first_]); |
||||
first_ = (first_ + 1) % kQueueSize; |
||||
--count_; |
||||
if (count_ == kQueueSize - 1) { |
||||
auto on_available = std::move(on_available_); |
||||
lock.Release(); |
||||
on_available.Wakeup(); |
||||
} |
||||
return std::move(value); |
||||
} |
||||
|
||||
void MarkClosed() { |
||||
ReleasableMutexLock lock(&mu_); |
||||
if (std::exchange(closed_, true)) return; |
||||
auto on_occupied = std::move(on_occupied_); |
||||
auto on_available = std::move(on_available_); |
||||
lock.Release(); |
||||
on_occupied.Wakeup(); |
||||
on_available.Wakeup(); |
||||
} |
||||
|
||||
private: |
||||
Mutex mu_; |
||||
std::array<T, kQueueSize> queue_ ABSL_GUARDED_BY(mu_); |
||||
bool closed_ ABSL_GUARDED_BY(mu_) = false; |
||||
uint8_t first_ ABSL_GUARDED_BY(mu_) = 0; |
||||
uint8_t count_ ABSL_GUARDED_BY(mu_) = 0; |
||||
Waker on_occupied_ ABSL_GUARDED_BY(mu_); |
||||
Waker on_available_ ABSL_GUARDED_BY(mu_); |
||||
}; |
||||
RefCountedPtr<Center> center_{MakeRefCounted<Center>()}; |
||||
|
||||
public: |
||||
class Sender { |
||||
public: |
||||
explicit Sender(RefCountedPtr<Center> center) |
||||
: center_(std::move(center)) {} |
||||
Sender(const Sender&) = delete; |
||||
Sender& operator=(const Sender&) = delete; |
||||
Sender(Sender&&) noexcept = default; |
||||
Sender& operator=(Sender&&) noexcept = default; |
||||
|
||||
~Sender() { |
||||
if (center_ != nullptr) center_->MarkClosed(); |
||||
} |
||||
|
||||
auto Push(T value) { |
||||
return [center = center_, value = std::move(value)]() mutable { |
||||
return center->Push(value); |
||||
}; |
||||
} |
||||
|
||||
private: |
||||
RefCountedPtr<Center> center_; |
||||
}; |
||||
|
||||
class Receiver { |
||||
public: |
||||
explicit Receiver(RefCountedPtr<Center> center) |
||||
: center_(std::move(center)) {} |
||||
Receiver(const Receiver&) = delete; |
||||
Receiver& operator=(const Receiver&) = delete; |
||||
Receiver(Receiver&&) noexcept = default; |
||||
Receiver& operator=(Receiver&&) noexcept = default; |
||||
|
||||
~Receiver() { |
||||
if (center_ != nullptr) center_->MarkClosed(); |
||||
} |
||||
|
||||
auto Next() { |
||||
return [center = center_]() { return center->Next(); }; |
||||
} |
||||
|
||||
private: |
||||
RefCountedPtr<Center> center_; |
||||
}; |
||||
|
||||
Sender sender{center_}; |
||||
Receiver receiver{center_}; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_LIB_PROMISE_INTER_ACTIVITY_PIPE_H
|
||||
@ -0,0 +1,113 @@ |
||||
// Copyright 2023 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/inter_activity_pipe.h" |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include "src/core/lib/promise/seq.h" |
||||
#include "test/core/promise/test_wakeup_schedulers.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
template <typename F> |
||||
ActivityPtr TestActivity(F f) { |
||||
return MakeActivity(std::move(f), InlineWakeupScheduler{}, |
||||
[](absl::Status status) { EXPECT_TRUE(status.ok()); }); |
||||
} |
||||
|
||||
TEST(InterActivityPipe, CanSendAndReceive) { |
||||
InterActivityPipe<int, 1> pipe; |
||||
bool done = false; |
||||
auto a = TestActivity(Seq(pipe.sender.Push(3), [](bool b) { |
||||
EXPECT_TRUE(b); |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_FALSE(done); |
||||
auto b = |
||||
TestActivity(Seq(pipe.receiver.Next(), [&done](absl::optional<int> n) { |
||||
EXPECT_EQ(n, 3); |
||||
done = true; |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_TRUE(done); |
||||
} |
||||
|
||||
TEST(InterActivityPipe, CanSendTwiceAndReceive) { |
||||
InterActivityPipe<int, 1> pipe; |
||||
bool done = false; |
||||
auto a = TestActivity(Seq( |
||||
pipe.sender.Push(3), |
||||
[&](bool b) { |
||||
EXPECT_TRUE(b); |
||||
return pipe.sender.Push(4); |
||||
}, |
||||
[](bool b) { |
||||
EXPECT_TRUE(b); |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_FALSE(done); |
||||
auto b = TestActivity(Seq( |
||||
pipe.receiver.Next(), |
||||
[&pipe](absl::optional<int> n) { |
||||
EXPECT_EQ(n, 3); |
||||
return pipe.receiver.Next(); |
||||
}, |
||||
[&done](absl::optional<int> n) { |
||||
EXPECT_EQ(n, 4); |
||||
done = true; |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_TRUE(done); |
||||
} |
||||
|
||||
TEST(InterActivityPipe, CanReceiveAndSend) { |
||||
InterActivityPipe<int, 1> pipe; |
||||
bool done = false; |
||||
auto b = |
||||
TestActivity(Seq(pipe.receiver.Next(), [&done](absl::optional<int> n) { |
||||
EXPECT_EQ(n, 3); |
||||
done = true; |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_FALSE(done); |
||||
auto a = TestActivity(Seq(pipe.sender.Push(3), [](bool b) { |
||||
EXPECT_TRUE(b); |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_TRUE(done); |
||||
} |
||||
|
||||
TEST(InterActivityPipe, CanClose) { |
||||
InterActivityPipe<int, 1> pipe; |
||||
bool done = false; |
||||
auto b = |
||||
TestActivity(Seq(pipe.receiver.Next(), [&done](absl::optional<int> n) { |
||||
EXPECT_EQ(n, absl::nullopt); |
||||
done = true; |
||||
return absl::OkStatus(); |
||||
})); |
||||
EXPECT_FALSE(done); |
||||
// Drop the sender
|
||||
{ auto x = std::move(pipe.sender); } |
||||
EXPECT_TRUE(done); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue