Skip to content

Commit

Permalink
[chaotic-good] Annotate key promises for latent-see (grpc#37830)
Browse files Browse the repository at this point in the history 
Closes grpc#37830

COPYBARA_INTEGRATE_REVIEW=grpc#37830 from ctiller:promising-latent-see 4f7dd1b
PiperOrigin-RevId: 681228010
  • Loading branch information
@ctiller @copybara-github
ctiller authored and copybara-github committed Oct 2, 2024
1 parent cd129b4 commit d6fd0bd
Show file tree
Hide file tree
Showing 5 changed files with 187 additions and 148 deletions.
84 changes: 47 additions & 37 deletions src/core/ext/transport/chaotic_good/client_transport.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,9 @@ auto ChaoticGoodClientTransport::PushFrameIntoCall(ServerFragmentFrame frame,
});
// Wrap the actual sequence with something that owns the call handler so that
// its lifetime extends until the push completes.
return [call_handler, push = std::move(push)]() mutable { return push(); };
return GRPC_LATENT_SEE_PROMISE(
"PushFrameIntoCall",
([call_handler, push = std::move(push)]() mutable { return push(); }));
}

auto ChaoticGoodClientTransport::TransportReadLoop(
Expand Down Expand Up @@ -205,11 +207,15 @@ ChaoticGoodClientTransport::ChaoticGoodClientTransport(
party_arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine.get());
party_ = Party::Make(std::move(party_arena));
party_->Spawn("client-chaotic-writer", TransportWriteLoop(transport),
party_->Spawn("client-chaotic-writer",
GRPC_LATENT_SEE_PROMISE("ClientTransportWriteLoop",
TransportWriteLoop(transport)),
OnTransportActivityDone("write_loop"));
party_->Spawn("client-chaotic-reader",
TransportReadLoop(std::move(transport)),
OnTransportActivityDone("read_loop"));
party_->Spawn(
"client-chaotic-reader",
GRPC_LATENT_SEE_PROMISE("ClientTransportReadLoop",
TransportReadLoop(std::move(transport))),
OnTransportActivityDone("read_loop"));
}

ChaoticGoodClientTransport::~ChaoticGoodClientTransport() { party_.reset(); }
Expand Down Expand Up @@ -265,38 +271,42 @@ auto ChaoticGoodClientTransport::CallOutboundLoop(uint32_t stream_id,
return absl::OkStatus();
});
};
return TrySeq(
// Wait for initial metadata then send it out.
call_handler.PullClientInitialMetadata(),
[send_fragment](ClientMetadataHandle md) mutable {
GRPC_TRACE_LOG(chaotic_good, INFO)
<< "CHAOTIC_GOOD: Sending initial metadata: " << md->DebugString();
ClientFragmentFrame frame;
frame.headers = std::move(md);
return send_fragment(std::move(frame));
},
// Continuously send client frame with client to server messages.
ForEach(OutgoingMessages(call_handler),
[send_fragment,
aligned_bytes = aligned_bytes_](MessageHandle message) mutable {
ClientFragmentFrame frame;
// Construct frame header (flags, header_length and
// trailer_length will be added in serialization).
const uint32_t message_length = message->payload()->Length();
const uint32_t padding =
message_length % aligned_bytes == 0
? 0
: aligned_bytes - message_length % aligned_bytes;
CHECK_EQ((message_length + padding) % aligned_bytes, 0u);
frame.message = FragmentMessage(std::move(message), padding,
message_length);
return send_fragment(std::move(frame));
}),
[send_fragment]() mutable {
ClientFragmentFrame frame;
frame.end_of_stream = true;
return send_fragment(std::move(frame));
});
return GRPC_LATENT_SEE_PROMISE(
"CallOutboundLoop",
TrySeq(
// Wait for initial metadata then send it out.
call_handler.PullClientInitialMetadata(),
[send_fragment](ClientMetadataHandle md) mutable {
GRPC_TRACE_LOG(chaotic_good, INFO)
<< "CHAOTIC_GOOD: Sending initial metadata: "
<< md->DebugString();
ClientFragmentFrame frame;
frame.headers = std::move(md);
return send_fragment(std::move(frame));
},
// Continuously send client frame with client to server messages.
ForEach(OutgoingMessages(call_handler),
[send_fragment, aligned_bytes = aligned_bytes_](
MessageHandle message) mutable {
ClientFragmentFrame frame;
// Construct frame header (flags, header_length and
// trailer_length will be added in serialization).
const uint32_t message_length =
message->payload()->Length();
const uint32_t padding =
message_length % aligned_bytes == 0
? 0
: aligned_bytes - message_length % aligned_bytes;
CHECK_EQ((message_length + padding) % aligned_bytes, 0u);
frame.message = FragmentMessage(std::move(message), padding,
message_length);
return send_fragment(std::move(frame));
}),
[send_fragment]() mutable {
ClientFragmentFrame frame;
frame.end_of_stream = true;
return send_fragment(std::move(frame));
}));
}

void ChaoticGoodClientTransport::StartCall(CallHandler call_handler) {
Expand Down
152 changes: 80 additions & 72 deletions src/core/ext/transport/chaotic_good/server_transport.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -199,25 +199,27 @@ auto ChaoticGoodServerTransport::SendCallInitialMetadataAndBody(
auto ChaoticGoodServerTransport::CallOutboundLoop(
uint32_t stream_id, CallInitiator call_initiator) {
auto outgoing_frames = outgoing_frames_.MakeSender();
return Seq(
Map(SendCallInitialMetadataAndBody(stream_id, outgoing_frames,
call_initiator),
[stream_id](absl::Status main_body_result) {
GRPC_TRACE_VLOG(chaotic_good, 2)
<< "CHAOTIC_GOOD: CallOutboundLoop: stream_id=" << stream_id
<< " main_body_result=" << main_body_result;
return Empty{};
}),
call_initiator.PullServerTrailingMetadata(),
// Capture the call_initator to ensure the underlying call_spine
// is alive until the SendFragment promise completes.
[stream_id, outgoing_frames,
call_initiator](ServerMetadataHandle md) mutable {
ServerFragmentFrame frame;
frame.trailers = std::move(md);
frame.stream_id = stream_id;
return SendFragment(std::move(frame), outgoing_frames, call_initiator);
});
return GRPC_LATENT_SEE_PROMISE(
"CallOutboundLoop",
Seq(Map(SendCallInitialMetadataAndBody(stream_id, outgoing_frames,
call_initiator),
[stream_id](absl::Status main_body_result) {
GRPC_TRACE_VLOG(chaotic_good, 2)
<< "CHAOTIC_GOOD: CallOutboundLoop: stream_id=" << stream_id
<< " main_body_result=" << main_body_result;
return Empty{};
}),
call_initiator.PullServerTrailingMetadata(),
// Capture the call_initator to ensure the underlying call_spine
// is alive until the SendFragment promise completes.
[stream_id, outgoing_frames,
call_initiator](ServerMetadataHandle md) mutable {
ServerFragmentFrame frame;
frame.trailers = std::move(md);
frame.stream_id = stream_id;
return SendFragment(std::move(frame), outgoing_frames,
call_initiator);
}));
}

auto ChaoticGoodServerTransport::DeserializeAndPushFragmentToNewCall(
Expand Down Expand Up @@ -269,57 +271,59 @@ auto ChaoticGoodServerTransport::DeserializeAndPushFragmentToExistingCall(
}

auto ChaoticGoodServerTransport::ReadOneFrame(ChaoticGoodTransport& transport) {
return TrySeq(
transport.ReadFrameBytes(),
[this, transport =
&transport](std::tuple<FrameHeader, BufferPair> frame_bytes) {
const auto& frame_header = std::get<0>(frame_bytes);
auto& buffers = std::get<1>(frame_bytes);
return Switch(
frame_header.type,
Case(FrameType::kSettings,
[]() -> absl::Status {
return absl::InternalError("Unexpected settings frame");
}),
Case(FrameType::kFragment,
[this, &frame_header, &buffers, transport]() {
return If(
frame_header.flags.is_set(0),
[this, &frame_header, &buffers, transport]() {
return DeserializeAndPushFragmentToNewCall(
frame_header, std::move(buffers), *transport);
},
[this, &frame_header, &buffers, transport]() {
return DeserializeAndPushFragmentToExistingCall(
frame_header, std::move(buffers), *transport);
});
}),
Case(FrameType::kCancel,
[this, &frame_header]() {
absl::optional<CallInitiator> call_initiator =
ExtractStream(frame_header.stream_id);
GRPC_TRACE_LOG(chaotic_good, INFO)
<< "Cancel stream " << frame_header.stream_id
<< (call_initiator.has_value() ? " (active)"
: " (not found)");
return If(
call_initiator.has_value(),
[&call_initiator]() {
auto c = std::move(*call_initiator);
return c.SpawnWaitable("cancel", [c]() mutable {
c.Cancel();
return absl::OkStatus();
});
},
[]() -> absl::Status { return absl::OkStatus(); });
}),
Default([frame_header]() {
return absl::InternalError(
absl::StrCat("Unexpected frame type: ",
static_cast<uint8_t>(frame_header.type)));
}));
},
[]() -> LoopCtl<absl::Status> { return Continue{}; });
return GRPC_LATENT_SEE_PROMISE(
"ReadOneFrame",
TrySeq(
transport.ReadFrameBytes(),
[this, transport = &transport](
std::tuple<FrameHeader, BufferPair> frame_bytes) {
const auto& frame_header = std::get<0>(frame_bytes);
auto& buffers = std::get<1>(frame_bytes);
return Switch(
frame_header.type,
Case(FrameType::kSettings,
[]() -> absl::Status {
return absl::InternalError("Unexpected settings frame");
}),
Case(FrameType::kFragment,
[this, &frame_header, &buffers, transport]() {
return If(
frame_header.flags.is_set(0),
[this, &frame_header, &buffers, transport]() {
return DeserializeAndPushFragmentToNewCall(
frame_header, std::move(buffers), *transport);
},
[this, &frame_header, &buffers, transport]() {
return DeserializeAndPushFragmentToExistingCall(
frame_header, std::move(buffers), *transport);
});
}),
Case(FrameType::kCancel,
[this, &frame_header]() {
absl::optional<CallInitiator> call_initiator =
ExtractStream(frame_header.stream_id);
GRPC_TRACE_LOG(chaotic_good, INFO)
<< "Cancel stream " << frame_header.stream_id
<< (call_initiator.has_value() ? " (active)"
: " (not found)");
return If(
call_initiator.has_value(),
[&call_initiator]() {
auto c = std::move(*call_initiator);
return c.SpawnWaitable("cancel", [c]() mutable {
c.Cancel();
return absl::OkStatus();
});
},
[]() -> absl::Status { return absl::OkStatus(); });
}),
Default([frame_header]() {
return absl::InternalError(
absl::StrCat("Unexpected frame type: ",
static_cast<uint8_t>(frame_header.type)));
}));
},
[]() -> LoopCtl<absl::Status> { return Continue{}; }));
}

auto ChaoticGoodServerTransport::TransportReadLoop(
Expand Down Expand Up @@ -360,9 +364,13 @@ ChaoticGoodServerTransport::ChaoticGoodServerTransport(
party_arena->SetContext<grpc_event_engine::experimental::EventEngine>(
event_engine.get());
party_ = Party::Make(std::move(party_arena));
party_->Spawn("server-chaotic-writer", TransportWriteLoop(transport),
party_->Spawn("server-chaotic-writer",
GRPC_LATENT_SEE_PROMISE("ServerTransportWriteLoop",
TransportWriteLoop(transport)),
OnTransportActivityDone("writer"));
party_->Spawn("server-chaotic-reader", TransportReadLoop(transport),
party_->Spawn("server-chaotic-reader",
GRPC_LATENT_SEE_PROMISE("ServerTransportReadLoop",
TransportReadLoop(transport)),
OnTransportActivityDone("reader"));
}

Expand Down
2 changes: 1 addition & 1 deletion src/core/lib/event_engine/posix_engine/posix_endpoint.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ namespace {
// of bytes sent.
ssize_t TcpSend(int fd, const struct msghdr* msg, int* saved_errno,
int additional_flags = 0) {
GRPC_LATENT_SEE_INNER_SCOPE("TcpSend");
GRPC_LATENT_SEE_PARENT_SCOPE("TcpSend");
ssize_t sent_length;
do {
sent_length = sendmsg(fd, msg, SENDMSG_FLAGS | additional_flags);
Expand Down
77 changes: 40 additions & 37 deletions src/core/lib/transport/promise_endpoint.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,24 +106,25 @@ class PromiseEndpoint {
return absl::OkStatus();
};
},
[this]() {
return [write_state = write_state_]() -> Poll<absl::Status> {
// If current write isn't finished return `Pending()`, else
// return write result.
WriteState::State expected = WriteState::kWritten;
if (write_state->state.compare_exchange_strong(
expected, WriteState::kIdle, std::memory_order_acquire,
std::memory_order_relaxed)) {
// State was Written, and we changed it to Idle. We can return
// the result.
return std::move(write_state->result);
}
// State was not Written; since we're polling it must be
// Writing. Assert that and return Pending.
CHECK(expected == WriteState::kWriting);
return Pending();
};
});
GRPC_LATENT_SEE_PROMISE(
"DelayedWrite", ([this]() {
return [write_state = write_state_]() -> Poll<absl::Status> {
// If current write isn't finished return `Pending()`, else
// return write result.
WriteState::State expected = WriteState::kWritten;
if (write_state->state.compare_exchange_strong(
expected, WriteState::kIdle, std::memory_order_acquire,
std::memory_order_relaxed)) {
// State was Written, and we changed it to Idle. We can return
// the result.
return std::move(write_state->result);
}
// State was not Written; since we're polling it must be
// Writing. Assert that and return Pending.
CHECK(expected == WriteState::kWriting);
return Pending();
};
})));
}

// Returns a promise that resolves to `SliceBuffer` with
Expand Down Expand Up @@ -174,25 +175,27 @@ class PromiseEndpoint {
return std::move(ret);
};
},
[this, num_bytes]() {
return [read_state = read_state_,
num_bytes]() -> Poll<absl::StatusOr<SliceBuffer>> {
if (!read_state->complete.load(std::memory_order_acquire)) {
return Pending();
}
// If read succeeds, return `SliceBuffer` with `num_bytes` bytes.
if (read_state->result.ok()) {
SliceBuffer ret;
grpc_slice_buffer_move_first_no_inline(
read_state->buffer.c_slice_buffer(), num_bytes,
ret.c_slice_buffer());
read_state->complete.store(false, std::memory_order_relaxed);
return std::move(ret);
}
read_state->complete.store(false, std::memory_order_relaxed);
return std::move(read_state->result);
};
});
GRPC_LATENT_SEE_PROMISE(
"DelayedRead", ([this, num_bytes]() {
return [read_state = read_state_,
num_bytes]() -> Poll<absl::StatusOr<SliceBuffer>> {
if (!read_state->complete.load(std::memory_order_acquire)) {
return Pending();
}
// If read succeeds, return `SliceBuffer` with `num_bytes`
// bytes.
if (read_state->result.ok()) {
SliceBuffer ret;
grpc_slice_buffer_move_first_no_inline(
read_state->buffer.c_slice_buffer(), num_bytes,
ret.c_slice_buffer());
read_state->complete.store(false, std::memory_order_relaxed);
return std::move(ret);
}
read_state->complete.store(false, std::memory_order_relaxed);
return std::move(read_state->result);
};
})));
}

// Returns a promise that resolves to `Slice` with at least
Expand Down
Loading

0 comments on commit d6fd0bd

Please sign in to comment.