You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
GH-55396 : [C++][FlightRPC] Add async gRPC server - #55419
Here we add an async gRPC Flight server whose handlers return arrow::Future instead of blocking and address the issue from #55396
The synchronous transport runs every RPC on a gRPC worker thread that the handler occupies until it returns. Throughput is bounded by the thread pool which grows at most about one thread per second rather than by the work in flight. With the async transport a handler returns a future immediately, and no gRPC thread is held while the answer is pending.
Deriving from AsyncGenericFlightServerBase is what selects the async transport. FlightServerBase is unchanged and still uses the synchronous path.
Every handler returns a future. Complete it from your own thread or pool when the answer is ready. Handlers that do run inline must still return promptly: blocking a gRPC callback thread starves new connections multiplexed onto it.
An unoverridden handler answers Status::NotImplemented.
Handler
Returns
ListFlightsAsync
Future<shared_ptr<AsyncFlightListing>>
GetFlightInfoAsync
Future<shared_ptr<FlightInfo>>
PollFlightInfoAsync
Future<shared_ptr<PollInfo>>
GetSchemaAsync
Future<shared_ptr<SchemaResult>>
DoGetAsync
Future<shared_ptr<AsyncFlightDataStream>>
DoActionAsync
Future<shared_ptr<AsyncResultStream>>
ListActionsAsync
Future<vector<ActionType>>
DoExchangeAsync
Future<>
CreateDoPutListener
shared_ptr<AsyncFlightDataListener> (plain, not a future)
Streams
Pull-based iterators. The transport requests one item at a time, so a producer never runs ahead of the consumer.
DoPut is not answered by a handler method. The transport hands each decoded
upload to a per RPC listener from CreateDoPutListener, returning nullptr
refuses the upload with UNIMPLEMENTED.
OnNext runs once per decoded message. chunk.data is the RecordBatch, or nullptr for a metadata-only message.
OnDescriptor returning non-OK rejects the upload; the client sees that status when it closes the writer.
OnFinish runs exactly once, however the upload ends. A cancel and a clean end do not both report.
Cancel is safe from any thread at any time. It returns OK when the cancel was handed to the transport, or Invalid when no upload is in flight.
writer()->WriteMetadata() may be called from any thread, including after the RPC has finished, the transport guards that race.
Authentication
Auth goes through hooks on the server class:
virtual Status Handshake(const ServerCallContext& context,
const std::string& request,
std::string* response);
virtual Status ValidateToken(const ServerCallContext& context,
const std::string& token,
std::string* peer_identity);
A blocking ServerAuthHandler in FlightServerOptions is not supported: Init() rejects it with NotImplemented rather than silently servicing unauthenticated calls.
Other new API
FlightStreamChunk(std::shared_ptr<RecordBatch>, std::shared_ptr<Buffer>): build a chunk directly (types.h).
AsyncFlightMessageDecoder: decode a FlightData message from a buffer or from an already-deserialized message (flight_data_decoder.h).
Requirements
The async server needs gRPC's generic callback API:
cpp/examples/arrow/flight_async_generic_doget_example.cc: DoGet. Each batch is produced on the stream's own worker thread --batch_delay_ms after the transport asks, so the delay never occupies a gRPC thread.
cpp/examples/arrow/flight_async_generic_doput_example.cc: DoPut. OnDescriptor / OnSchemaDecoded / OnNext / OnFinish, and the PutResult metadata channel.
Are there any user-facing changes?
Was AI used for this PR?
In accordance to the AI generation guidelines, please disclose below whether and how AI was used in this PR.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
Unresolved lifetime and synchronization bugs can cause use-after-free, deadlocks, and cancellation failures, with additional defects breaking Windows builds and mutual-TLS peer identity propagation.
Adds a future-based C++ Flight server for GH-55396, using gRPC’s generic callback API to avoid holding worker threads while application results are pending. The existing synchronous server remains available.
Changes:
Introduces asynchronous handlers, streaming interfaces, upload listeners, and authentication hooks.
Adds callback reactors and shares server lifecycle and serialization utilities.
Integrates version-gated builds, RPC tests, decoder tests, and upload/download examples.
File
Description
cpp/src/arrow/util/windows_fixup.h
Stops undefining Windows IN and OUT macros.
cpp/src/arrow/flight/types.h
Declares a chunk constructor accepting data and metadata.
Dictionary messages remain queued until a record batch arrives. A stream sending many dictionary deltas or replacements before its next batch retains every message body, and HasRecordBatch() rescans the growing queue on each Consume(), producing unbounded retained memory and quadratic work. Decode dictionary messages incrementally instead of waiting for a record batch to drain them, while retaining only the dictionary state needed for subsequent batches.
Queue PutResults or expose completion backpressure
A listener that acknowledges each batch can receive the next OnNext() before the preceding PutResult's OnWriteDone(), so ordinary consecutive uploads can hit this branch and lose acknowledgements or fail the upload. WriteMetadata() returns only a Status, so callers have no completion future to await as this error requests. Queue and serialize PutResults, or expose an asynchronous completion/backpressure API so the next acknowledgement can be issued reliably.
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
Unresolved concurrency and lifecycle defects can deadlock uploads, report success despite commit failures, deliver callbacks after termination, and retain unbounded buffers or resources.
If the client cancels while OnDescriptor() is pending, DoPut reports OnFinish(), but completing the descriptor future with OK still delivers OnSchemaDecoded() or OnNext() afterward. The listener can therefore receive data after releasing resources or rolling back in its terminal callback. Coordinate decoder callbacks with upload termination so this continuation cannot deliver data after OnFinish().
Dictionary messages remain in queue_ until a record batch arrives, so a client sending dictionary replacements without record batches causes unbounded buffer retention and repeated scans of the growing queue. Messages without application metadata also bypass OnNext, so listener backpressure cannot bound this growth. Process dictionaries incrementally into the IPC dictionary state instead of retaining every message until the next record batch.
Propagate OnFinish failures before completing the RPC
On a clean upload, FinishUpload calls this and immediately finishes the RPC with OK, regardless of the OnFinish future's result. A listener that commits its upload asynchronously in OnFinish can therefore fail—even return an already-failed future—while the client's Close still succeeds. Chain successful RPC completion to this future and propagate its failure, without blocking the gRPC callback thread.
This calls the listener's OnFinish while holding write_mutex(). If another thread is in writer()->WriteMetadata(), it can hold writer_mutex_ while waiting for write_mutex(); an OnFinish that writes metadata then waits for writer_mutex_, deadlocking both threads and preventing RPC shutdown. Release the transport lock before invoking FinishUpload here, as OnWriteDone already does.
Break pending-future ownership cycles during teardown
If a listing or result stream retains its pending NextAsync() future, the reactor, iterator, continuation, and HoldToken form an ownership cycle that cancellation does not break, retaining the reactor and stream after shutdown. Break the cycle during teardown while keeping late callbacks safe; simply removing the hold would introduce a use-after-free. The same issue affects DoPut listeners that retain a pending OnNext() future in async_grpc_do_put.cc:131–132.
Sequential OnNext callbacks that acknowledge each batch can reach this guard before the previous PutResult's OnWriteDone, rejecting an acknowledgement or failing an otherwise valid upload. WriteMetadata returns only a Status, so the listener has no completion signal to await before the next write. Queue and serialize PutResults, or expose an awaitable write completion so listeners can apply backpressure.
Honor the configured memory manager in async decoders
The async service never receives the configured memory manager, so DoPut and DoExchange decode host-backed buffers even when FlightServerOptions::memory_manager requests another allocation destination. This ignores the incoming-allocation contract; the synchronous reader applies Buffer::ViewOrCopy() in transport_server_internal.cc:46. Pass the manager through to both async decoders and apply it to message bodies, or explicitly reject unsupported managers during Init().
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
github-actions
Bot
added
Component: C++
Component: FlightRPC
Component: Documentation
labels
Rationale for this change
Address the issue: #55396
What changes are included in this PR?
Here we add an async gRPC Flight server whose handlers return
arrow::Futureinstead of blocking and address the issue from #55396The synchronous transport runs every RPC on a gRPC worker thread that the handler occupies until it returns. Throughput is bounded by the thread pool which grows at most about one thread per second rather than by the work in flight. With the async transport a handler returns a future immediately, and no gRPC thread is held while the answer is pending.
Deriving from
AsyncGenericFlightServerBaseis what selects the async transport.FlightServerBaseis unchanged and still uses the synchronous path.Every handler returns a future. Complete it from your own thread or pool when the answer is ready. Handlers that do run inline must still return promptly: blocking a gRPC callback thread starves new connections multiplexed onto it.
An unoverridden handler answers
Status::NotImplemented.ListFlightsAsyncFuture<shared_ptr<AsyncFlightListing>>GetFlightInfoAsyncFuture<shared_ptr<FlightInfo>>PollFlightInfoAsyncFuture<shared_ptr<PollInfo>>GetSchemaAsyncFuture<shared_ptr<SchemaResult>>DoGetAsyncFuture<shared_ptr<AsyncFlightDataStream>>DoActionAsyncFuture<shared_ptr<AsyncResultStream>>ListActionsAsyncFuture<vector<ActionType>>DoExchangeAsyncFuture<>CreateDoPutListenershared_ptr<AsyncFlightDataListener>(plain, not a future)Streams
Pull-based iterators. The transport requests one item at a time, so a producer never runs ahead of the consumer.
NextAsync()reports the end of a stream asstd::nullopt.DoExchangegets both halves:DoPut uses listener
DoPut is not answered by a handler method. The transport hands each decoded
upload to a per RPC listener from
CreateDoPutListener, returningnullptrrefuses the upload with
UNIMPLEMENTED.OnNextruns once per decoded message.chunk.datais theRecordBatch, ornullptrfor a metadata-only message.OnDescriptorreturning non-OK rejects the upload; the client sees that status when it closes the writer.OnFinishruns exactly once, however the upload ends. A cancel and a clean end do not both report.Cancelis safe from any thread at any time. It returnsOKwhen the cancel was handed to the transport, orInvalidwhen no upload is in flight.writer()->WriteMetadata()may be called from any thread, including after the RPC has finished, the transport guards that race.Authentication
Auth goes through hooks on the server class:
A blocking
ServerAuthHandlerinFlightServerOptionsis not supported:Init()rejects it withNotImplementedrather than silently servicing unauthenticated calls.Other new API
FlightStreamChunk(std::shared_ptr<RecordBatch>, std::shared_ptr<Buffer>): build a chunk directly (types.h).AsyncFlightMessageDecoder: decode aFlightDatamessage from a buffer or from an already-deserialized message (flight_data_decoder.h).Requirements
The async server needs gRPC's generic callback API:
Init()fails withNotImplementedAre these changes tested?
New files:
Examples:
--batch_delay_msafter the transport asks, so the delay never occupies a gRPC thread.OnDescriptor/OnSchemaDecoded/OnNext/OnFinish, and thePutResultmetadata channel.Are there any user-facing changes?
Was AI used for this PR?
In accordance to the AI generation guidelines, please disclose below whether and how AI was used in this PR.
PR code and description written by:
Reviewed before submission by: