Skip to content

Commit a15fea6

Browse files
Ankit Kumarmeta-codesync[bot]
Ankit Kumar
authored and
meta-codesync[bot]
committed
Rename server handler to ThriftServerBackpressureHandler
Summary: Rename `WriteBufferBackpressureHandler` to reflect its role pausing inbound admission while accepted responses continue toward the transport. Update the handler header, pipeline registrations and helper names, Buck targets, test fixtures, benchmark, and Thrift layer reference. Reviewed By: evanjzou Differential Revision: D125545452 fbshipit-source-id: 3c74ff92c23bf9fbc9fd361f7eba568e58424211
1 parent bcac0d6 commit a15fea6

5 files changed

Lines changed: 61 additions & 59 deletions

File tree

‎thrift/lib/cpp2/fast_thrift/thrift/server/ThriftServerConnectionFactory.cpp‎

Lines changed: 27 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -55,14 +55,14 @@
5555
#include <thrift/lib/cpp2/fast_thrift/thrift/server/adapter/ThriftServerCompositeAppAdapter.h>
5656
#include <thrift/lib/cpp2/fast_thrift/thrift/server/common/Event.h>
5757
#include <thrift/lib/cpp2/fast_thrift/thrift/server/common/ReadBackpressureArbiter.h>
58+
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerBackpressureHandler.h>
5859
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerChecksumHandler.h>
5960
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerCompressionHandler.h>
6061
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerConnectionCloseHandler.h>
6162
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerRequestContextHandler.h>
6263
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerRequestLifecycleHandler.h>
6364
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerSetupHandler.h>
6465
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerStreamMuxHandler.h>
65-
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/WriteBufferBackpressureHandler.h>
6666

6767
namespace apache::thrift::fast_thrift::thrift::server {
6868

@@ -114,7 +114,7 @@ HANDLER_TAG(thrift_server_request_lifecycle_handler);
114114
HANDLER_TAG(thrift_server_compression_handler);
115115
HANDLER_TAG(thrift_server_checksum_handler);
116116
HANDLER_TAG(thrift_server_connection_close_handler);
117-
HANDLER_TAG(write_buffer_backpressure_handler);
117+
HANDLER_TAG(thrift_server_backpressure_handler);
118118
HANDLER_TAG(thrift_server_setup_handler);
119119
HANDLER_TAG(thrift_server_stream_mux_handler);
120120

@@ -296,17 +296,17 @@ PipelineOwner finishStaticThriftPipeline(
296296
}
297297
}
298298

299-
template <bool WithWriteBuffer, bool WithExtensions, typename Builder>
300-
PipelineOwner addStaticWriteBuffer(
299+
template <bool WithBackpressure, bool WithExtensions, typename Builder>
300+
PipelineOwner addStaticBackpressure(
301301
Builder&& builder,
302302
const ThriftServerConnectionFactoryConfig& config,
303303
ExtensionStateStore& extensionStates,
304304
ReadBackpressureArbiter& readBackpressureArbiter) {
305-
if constexpr (WithWriteBuffer) {
305+
if constexpr (WithBackpressure) {
306306
return finishStaticThriftPipeline<WithExtensions>(
307307
std::forward<Builder>(builder)
308-
.template addNextDuplexTemplate<WriteBufferBackpressureHandler>(
309-
write_buffer_backpressure_handler_tag,
308+
.template addNextDuplexTemplate<ThriftServerBackpressureHandler>(
309+
thrift_server_backpressure_handler_tag,
310310
&readBackpressureArbiter),
311311
config,
312312
extensionStates);
@@ -318,7 +318,7 @@ PipelineOwner addStaticWriteBuffer(
318318

319319
template <
320320
bool WithChecksum,
321-
bool WithWriteBuffer,
321+
bool WithBackpressure,
322322
bool WithExtensions,
323323
typename Builder>
324324
PipelineOwner addStaticChecksum(
@@ -327,7 +327,7 @@ PipelineOwner addStaticChecksum(
327327
ExtensionStateStore& extensionStates,
328328
ReadBackpressureArbiter& readBackpressureArbiter) {
329329
if constexpr (WithChecksum) {
330-
return addStaticWriteBuffer<WithWriteBuffer, WithExtensions>(
330+
return addStaticBackpressure<WithBackpressure, WithExtensions>(
331331
std::forward<Builder>(builder)
332332
.template addNextDuplexTemplate<ThriftServerChecksumHandler>(
333333
thrift_server_checksum_handler_tag)
@@ -339,7 +339,7 @@ PipelineOwner addStaticChecksum(
339339
extensionStates,
340340
readBackpressureArbiter);
341341
} else {
342-
return addStaticWriteBuffer<WithWriteBuffer, WithExtensions>(
342+
return addStaticBackpressure<WithBackpressure, WithExtensions>(
343343
std::forward<Builder>(builder)
344344
.template addNextDuplexTemplate<ThriftServerConnectionCloseHandler>(
345345
thrift_server_connection_close_handler_tag,
@@ -353,15 +353,15 @@ PipelineOwner addStaticChecksum(
353353

354354
template <
355355
bool WithChecksum,
356-
bool WithWriteBuffer,
356+
bool WithBackpressure,
357357
bool WithExtensions,
358358
typename Builder>
359359
PipelineOwner addStaticCompression(
360360
Builder&& builder,
361361
const ThriftServerConnectionFactoryConfig& config,
362362
ExtensionStateStore& extensionStates,
363363
ReadBackpressureArbiter& readBackpressureArbiter) {
364-
return addStaticChecksum<WithChecksum, WithWriteBuffer, WithExtensions>(
364+
return addStaticChecksum<WithChecksum, WithBackpressure, WithExtensions>(
365365
std::forward<Builder>(builder)
366366
.template addNextDuplexTemplate<ThriftServerCompressionHandler>(
367367
thrift_server_compression_handler_tag),
@@ -374,7 +374,7 @@ template <
374374
bool WithStats,
375375
bool WithHeaders,
376376
bool WithChecksum,
377-
bool WithWriteBuffer,
377+
bool WithBackpressure,
378378
bool WithExtensions,
379379
typename TailAdapter>
380380
PipelineOwner buildStaticThriftPipeline(
@@ -396,7 +396,7 @@ PipelineOwner buildStaticThriftPipeline(
396396
.setAllocator(allocator);
397397
auto& readBackpressureArbiter = connContext->readBackpressureArbiter();
398398
if constexpr (WithStats) {
399-
return addStaticCompression<WithChecksum, WithWriteBuffer, WithExtensions>(
399+
return addStaticCompression<WithChecksum, WithBackpressure, WithExtensions>(
400400
std::move(builder)
401401
.template addNextDuplex<
402402
ThriftMetricsHandler<Direction::Server, ServerStatsShard>>(
@@ -411,7 +411,7 @@ PipelineOwner buildStaticThriftPipeline(
411411
extensionStates,
412412
readBackpressureArbiter);
413413
} else {
414-
return addStaticCompression<WithChecksum, WithWriteBuffer, WithExtensions>(
414+
return addStaticCompression<WithChecksum, WithBackpressure, WithExtensions>(
415415
std::move(builder)
416416
.template addNextDuplexTemplate<ThriftServerRequestContextHandler>(
417417
thrift_server_request_context_handler_tag,
@@ -429,7 +429,7 @@ template <
429429
bool WithStats,
430430
bool WithHeaders,
431431
bool WithChecksum,
432-
bool WithWriteBuffer,
432+
bool WithBackpressure,
433433
typename TailAdapter>
434434
PipelineOwner selectStaticExtensions(
435435
bool withExtensions,
@@ -446,7 +446,7 @@ PipelineOwner selectStaticExtensions(
446446
WithStats,
447447
WithHeaders,
448448
WithChecksum,
449-
WithWriteBuffer,
449+
WithBackpressure,
450450
true>(
451451
evb,
452452
transportAdapter,
@@ -461,7 +461,7 @@ PipelineOwner selectStaticExtensions(
461461
WithStats,
462462
WithHeaders,
463463
WithChecksum,
464-
WithWriteBuffer,
464+
WithBackpressure,
465465
false>(
466466
evb,
467467
transportAdapter,
@@ -478,7 +478,7 @@ template <
478478
bool WithHeaders,
479479
bool WithChecksum,
480480
typename TailAdapter>
481-
PipelineOwner selectStaticWriteBuffer(
481+
PipelineOwner selectStaticBackpressure(
482482
const ThriftServerConnectionFactoryConfig& config,
483483
folly::EventBase* evb,
484484
ThriftServerTransportAdapter* transportAdapter,
@@ -523,7 +523,7 @@ PipelineOwner selectStaticChecksum(
523523
ExtensionStateStore& extensionStates,
524524
ServerStatsShard* statsShard) {
525525
if (config.enableChecksum) {
526-
return selectStaticWriteBuffer<WithStats, WithHeaders, true>(
526+
return selectStaticBackpressure<WithStats, WithHeaders, true>(
527527
config,
528528
evb,
529529
transportAdapter,
@@ -533,7 +533,7 @@ PipelineOwner selectStaticChecksum(
533533
extensionStates,
534534
statsShard);
535535
}
536-
return selectStaticWriteBuffer<WithStats, WithHeaders, false>(
536+
return selectStaticBackpressure<WithStats, WithHeaders, false>(
537537
config,
538538
evb,
539539
transportAdapter,
@@ -900,8 +900,8 @@ detail::ThriftServerConnectionFactoryT<Mode>::buildConnectionImpl(
900900
ThriftServerConnectionCloseHandler<channel_pipeline::detail::ContextImpl>;
901901
using RequestLifecycleHandler = ThriftServerRequestLifecycleHandler<
902902
channel_pipeline::detail::ContextImpl>;
903-
using WriteBufferHandler =
904-
WriteBufferBackpressureHandler<channel_pipeline::detail::ContextImpl>;
903+
using BackpressureHandler =
904+
ThriftServerBackpressureHandler<channel_pipeline::detail::ContextImpl>;
905905
using SetupHandler =
906906
ThriftServerSetupHandler<channel_pipeline::detail::ContextImpl>;
907907
using MuxHandler =
@@ -969,14 +969,14 @@ detail::ThriftServerConnectionFactoryT<Mode>::buildConnectionImpl(
969969
thrift_server_connection_close_handler_tag,
970970
config_.drainTimeout,
971971
config_.reapTimeout);
972-
// Write-buffer handler sits between the context handler and the drain
972+
// Backpressure handler sits between the context handler and the drain
973973
// handler. Placed above drain (closer to head) so its inbound
974974
// Backpressure signal propagates upstream toward the transport, and
975-
// outbound responses from the tail traverse drain → write-buffer →
975+
// outbound responses from the tail traverse drain → backpressure →
976976
// head.
977977
if (config_.enableWriteBufferBackpressure) {
978-
thriftPipelineBuilder.template addNextDuplex<WriteBufferHandler>(
979-
write_buffer_backpressure_handler_tag, &readBackpressureArbiter);
978+
thriftPipelineBuilder.template addNextDuplex<BackpressureHandler>(
979+
thrift_server_backpressure_handler_tag, &readBackpressureArbiter);
980980
}
981981
// Embedder-registered handlers go after all built-ins, in registration
982982
// order — the first sits closest to the head, the last immediately above

thrift/lib/cpp2/fast_thrift/thrift/server/handler/WriteBufferBackpressureHandler.h renamed to thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerBackpressureHandler.h

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,13 +30,12 @@ namespace apache::thrift::fast_thrift::thrift {
3030
/**
3131
* Pauses inbound admission while outbound writes report backpressure.
3232
* Responses for accepted requests always continue toward the transport.
33-
* The historical name is retained for pipeline registrations.
3433
*/
3534
template <typename Context>
36-
class WriteBufferBackpressureHandler {
35+
class ThriftServerBackpressureHandler {
3736
public:
38-
WriteBufferBackpressureHandler() = default;
39-
explicit WriteBufferBackpressureHandler(
37+
ThriftServerBackpressureHandler() = default;
38+
explicit ThriftServerBackpressureHandler(
4039
ReadBackpressureArbiter* arbiter) noexcept
4140
: arbiter_(arbiter) {}
4241

thrift/lib/cpp2/fast_thrift/thrift/server/handler/bench/WriteBufferBackpressureHandlerBench.cpp renamed to thrift/lib/cpp2/fast_thrift/thrift/server/handler/bench/ThriftServerBackpressureHandlerBench.cpp

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
*/
1616

1717
/**
18-
* WriteBufferBackpressureHandler microbenchmarks.
18+
* ThriftServerBackpressureHandler microbenchmarks.
1919
*
2020
* The hot path is `onWrite` in the unsaturated case — a single flag
2121
* check + forward. This bench guards that hot path against regressions
@@ -33,7 +33,7 @@
3333
#include <thrift/lib/cpp2/fast_thrift/channel_pipeline/Common.h>
3434
#include <thrift/lib/cpp2/fast_thrift/channel_pipeline/TypeErasedBox.h>
3535
#include <thrift/lib/cpp2/fast_thrift/thrift/server/common/Messages.h>
36-
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/WriteBufferBackpressureHandler.h>
36+
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerBackpressureHandler.h>
3737

3838
using namespace folly;
3939
using namespace apache::thrift::fast_thrift::channel_pipeline;
@@ -85,7 +85,7 @@ BENCHMARK(OnWrite_PassThrough, iters) {
8585
BenchmarkSuspender suspender;
8686

8787
ReadBackpressureArbiter arbiter;
88-
WriteBufferBackpressureHandler<BenchCtx> handler(&arbiter);
88+
ThriftServerBackpressureHandler<BenchCtx> handler(&arbiter);
8989
BenchCtx ctx;
9090

9191
std::vector<TypeErasedBox> requests;
@@ -107,7 +107,7 @@ BENCHMARK(OnWrite_ForwardWhileBackpressured, iters) {
107107
BenchmarkSuspender suspender;
108108

109109
ReadBackpressureArbiter arbiter;
110-
WriteBufferBackpressureHandler<BenchCtx> handler(&arbiter);
110+
ThriftServerBackpressureHandler<BenchCtx> handler(&arbiter);
111111
BenchCtx ctx;
112112

113113
// Arm the backpressure flag with one priming write.
@@ -136,7 +136,7 @@ BENCHMARK(OnWriteReady_ReleaseReadPause, iters) {
136136
BenchmarkSuspender suspender;
137137

138138
ReadBackpressureArbiter arbiter;
139-
WriteBufferBackpressureHandler<BenchCtx> handler(&arbiter);
139+
ThriftServerBackpressureHandler<BenchCtx> handler(&arbiter);
140140
BenchCtx ctx;
141141

142142
const size_t cycles = (iters + kDrainBatch - 1) / kDrainBatch;

thrift/lib/cpp2/fast_thrift/thrift/server/handler/test/WriteBufferBackpressureHandlerTest.cpp renamed to thrift/lib/cpp2/fast_thrift/thrift/server/handler/test/ThriftServerBackpressureHandlerTest.cpp

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
* limitations under the License.
1515
*/
1616

17-
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/WriteBufferBackpressureHandler.h>
17+
#include <thrift/lib/cpp2/fast_thrift/thrift/server/handler/ThriftServerBackpressureHandler.h>
1818

1919
#include <vector>
2020

@@ -77,8 +77,9 @@ ThriftServerResponseMessage makeResponse(uint32_t streamId) {
7777
.payload = ThriftInitialResponsePayload{.streamId = streamId}};
7878
}
7979

80-
TEST(WriteBufferBackpressureHandlerTest, ResponsesContinueWhileReadsArePaused) {
81-
WriteBufferBackpressureHandler<FakeContext> handler;
80+
TEST(
81+
ThriftServerBackpressureHandlerTest, ResponsesContinueWhileReadsArePaused) {
82+
ThriftServerBackpressureHandler<FakeContext> handler;
8283
FakeContext ctx;
8384
ctx.nextWriteResult = Result::Backpressure;
8485

@@ -109,8 +110,8 @@ TEST(WriteBufferBackpressureHandlerTest, ResponsesContinueWhileReadsArePaused) {
109110
Result::Success);
110111
}
111112

112-
TEST(WriteBufferBackpressureHandlerTest, ReadinessWaitsForOtherPauseOwners) {
113-
WriteBufferBackpressureHandler<FakeContext> handler;
113+
TEST(ThriftServerBackpressureHandlerTest, ReadinessWaitsForOtherPauseOwners) {
114+
ThriftServerBackpressureHandler<FakeContext> handler;
114115
FakeContext ctx;
115116
ctx.arbiter.acquire();
116117
ctx.nextWriteResult = Result::Backpressure;
@@ -122,8 +123,9 @@ TEST(WriteBufferBackpressureHandlerTest, ReadinessWaitsForOtherPauseOwners) {
122123
EXPECT_EQ(ctx.fakePipeline.onReadReadyCalls, 0);
123124
}
124125

125-
TEST(WriteBufferBackpressureHandlerTest, InactiveReleasesPauseWithoutResuming) {
126-
WriteBufferBackpressureHandler<FakeContext> handler;
126+
TEST(
127+
ThriftServerBackpressureHandlerTest, InactiveReleasesPauseWithoutResuming) {
128+
ThriftServerBackpressureHandler<FakeContext> handler;
127129
FakeContext ctx;
128130
ctx.nextWriteResult = Result::Backpressure;
129131
(void)handler.onWrite(ctx, erase_and_box(makeResponse(1)));
@@ -137,8 +139,8 @@ TEST(WriteBufferBackpressureHandlerTest, InactiveReleasesPauseWithoutResuming) {
137139
EXPECT_EQ(ctx.cancelAwaitWriteReadyCalls, 1);
138140
}
139141

140-
TEST(WriteBufferBackpressureHandlerTest, ErrorsRemainTerminalUnderPressure) {
141-
WriteBufferBackpressureHandler<FakeContext> handler;
142+
TEST(ThriftServerBackpressureHandlerTest, ErrorsRemainTerminalUnderPressure) {
143+
ThriftServerBackpressureHandler<FakeContext> handler;
142144
FakeContext ctx;
143145
ctx.nextWriteResult = Result::Backpressure;
144146
(void)handler.onWrite(ctx, erase_and_box(makeResponse(1)));
@@ -153,17 +155,18 @@ TEST(WriteBufferBackpressureHandlerTest, ErrorsRemainTerminalUnderPressure) {
153155
EXPECT_EQ(ctx.arbiter.holderCount(), 1);
154156
}
155157

156-
#if defined(FAST_THRIFT_WRITE_BUFFER_PIPELINE_TESTS)
158+
#if defined(FAST_THRIFT_SERVER_BACKPRESSURE_PIPELINE_TESTS)
157159
HANDLER_TAG(read_backpressure);
158160

159161
TEST(
160-
WriteBufferBackpressureHandlerPipelineTest, WritesFlowWhileReadsArePaused) {
162+
ThriftServerBackpressureHandlerPipelineTest,
163+
WritesFlowWhileReadsArePaused) {
161164
folly::EventBase evb;
162165
cp::test::MockHeadHandler head;
163166
cp::test::MockTailHandler tail;
164167
cp::test::TestAllocator allocator;
165168
ReadBackpressureArbiter arbiter;
166-
using Handler = WriteBufferBackpressureHandler<cp::detail::ContextImpl>;
169+
using Handler = ThriftServerBackpressureHandler<cp::detail::ContextImpl>;
167170
auto pipeline = cp::PipelineBuilder<
168171
cp::test::MockHeadHandler,
169172
cp::test::MockTailHandler,

0 commit comments

Comments
 (0)