Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Thread annotations for C++ callback API headers (#25571)

Browse Source 
* Annotate headers

* clang-format
pull/25618/head
Vijay Pai 4 years ago committed by GitHub
parent 65ea33e427
commit dae5624e47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 39 additions and 30 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      include/grpcpp/impl/codegen/client_callback.h
  2. 48
      include/grpcpp/impl/codegen/server_callback.h

21
include/grpcpp/impl/codegen/client_callback.h
Unescape View File

@ -27,6 +27,7 @@
#include <grpcpp/impl/codegen/config.h>
#include <grpcpp/impl/codegen/core_codegen_interface.h>
#include <grpcpp/impl/codegen/status.h>
#include <grpcpp/impl/codegen/sync.h>
namespace grpc {
class Channel;
@ -472,7 +473,7 @@ class ClientCallbackReaderWriterImpl
// there are no tests catching the compiler warning.
static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); }
void StartCall() override {
void StartCall() ABSL_LOCKS_EXCLUDED(start_mu_) override {
// This call initiates two batches, plus any backlog, each with a callback
// 1. Send initial metadata (unless corked) + recv initial metadata
// 2. Any read backlog
@ -521,7 +522,8 @@ class ClientCallbackReaderWriterImpl
call_.PerformOps(&read_ops_);
}
void Write(const Request* msg, ::grpc::WriteOptions options) override {
void Write(const Request* msg, ::grpc::WriteOptions options)
ABSL_LOCKS_EXCLUDED(start_mu_) override {
if (options.is_last_message()) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
@ -544,7 +546,7 @@ class ClientCallbackReaderWriterImpl
}
call_.PerformOps(&write_ops_);
}
void WritesDone() override {
void WritesDone() ABSL_LOCKS_EXCLUDED(start_mu_) override {
writes_done_ops_.ClientSendClose();
writes_done_tag_.Set(
call_.call(),
@ -685,7 +687,7 @@ class ClientCallbackReaderWriterImpl
bool writes_done_ops = false;
bool read_ops = false;
};
StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
StartCallBacklog backlog_ ABSL_GUARDED_BY(start_mu_);
// Minimum of 3 callbacks to pre-register for start ops, StartCall, and finish
std::atomic<intptr_t> callbacks_outstanding_{3};
@ -843,7 +845,7 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
struct StartCallBacklog {
bool read_ops = false;
};
StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
StartCallBacklog backlog_ ABSL_GUARDED_BY(start_mu_);
// Minimum of 2 callbacks to pre-register for start and finish
std::atomic<intptr_t> callbacks_outstanding_{2};
@ -884,7 +886,7 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
// there are no tests catching the compiler warning.
static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); }
void StartCall() override {
void StartCall() ABSL_LOCKS_EXCLUDED(start_mu_) override {
// This call initiates two batches, plus any backlog, each with a callback
// 1. Send initial metadata (unless corked) + recv initial metadata
// 2. Any backlog
@ -916,7 +918,8 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
this->MaybeFinish(/*from_reaction=*/false);
}
void Write(const Request* msg, ::grpc::WriteOptions options) override {
void Write(const Request* msg, ::grpc::WriteOptions options)
ABSL_LOCKS_EXCLUDED(start_mu_) override {
if (GPR_UNLIKELY(options.is_last_message())) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
@ -941,7 +944,7 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
call_.PerformOps(&write_ops_);
}
void WritesDone() override {
void WritesDone() ABSL_LOCKS_EXCLUDED(start_mu_) override {
writes_done_ops_.ClientSendClose();
writes_done_tag_.Set(
call_.call(),
@ -1070,7 +1073,7 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
bool write_ops = false;
bool writes_done_ops = false;
};
StartCallBacklog backlog_ /* GUARDED_BY(start_mu_) */;
StartCallBacklog backlog_ ABSL_GUARDED_BY(start_mu_);
// Minimum of 3 callbacks to pre-register for start ops, StartCall, and finish
std::atomic<intptr_t> callbacks_outstanding_{3};

48
include/grpcpp/impl/codegen/server_callback.h
Unescape View File

@ -29,6 +29,7 @@
#include <grpcpp/impl/codegen/core_codegen_interface.h>
#include <grpcpp/impl/codegen/message_allocator.h>
#include <grpcpp/impl/codegen/status.h>
#include <grpcpp/impl/codegen/sync.h>
namespace grpc {
@ -278,7 +279,7 @@ class ServerBidiReactor : public internal::ServerReactor {
/// Send any initial metadata stored in the RPC context. If not invoked,
/// any initial metadata will be passed along with the first Write or the
/// Finish (if there are no writes).
void StartSendInitialMetadata() {
void StartSendInitialMetadata() ABSL_LOCKS_EXCLUDED(stream_mu_) {
ServerCallbackReaderWriter<Request, Response>* stream =
stream_.load(std::memory_order_acquire);
if (stream == nullptr) {
@ -296,7 +297,7 @@ class ServerBidiReactor : public internal::ServerReactor {
///
/// \param[out] req Where to eventually store the read message. Valid when
/// the library calls OnReadDone
void StartRead(Request* req) {
void StartRead(Request* req) ABSL_LOCKS_EXCLUDED(stream_mu_) {
ServerCallbackReaderWriter<Request, Response>* stream =
stream_.load(std::memory_order_acquire);
if (stream == nullptr) {
@ -325,7 +326,8 @@ class ServerBidiReactor : public internal::ServerReactor {
/// ownership but the caller must ensure that the message is
/// not deleted or modified until OnWriteDone is called.
/// \param[in] options The WriteOptions to use for writing this message
void StartWrite(const Response* resp, ::grpc::WriteOptions options) {
void StartWrite(const Response* resp, ::grpc::WriteOptions options)
ABSL_LOCKS_EXCLUDED(stream_mu_) {
ServerCallbackReaderWriter<Request, Response>* stream =
stream_.load(std::memory_order_acquire);
if (stream == nullptr) {
@ -354,7 +356,7 @@ class ServerBidiReactor : public internal::ServerReactor {
/// \param[in] options The WriteOptions to use for writing this message
/// \param[in] s The status outcome of this RPC
void StartWriteAndFinish(const Response* resp, ::grpc::WriteOptions options,
::grpc::Status s) {
::grpc::Status s) ABSL_LOCKS_EXCLUDED(stream_mu_) {
ServerCallbackReaderWriter<Request, Response>* stream =
stream_.load(std::memory_order_acquire);
if (stream == nullptr) {
@ -389,7 +391,7 @@ class ServerBidiReactor : public internal::ServerReactor {
/// cancelled.
///
/// \param[in] s The status outcome of this RPC
void Finish(::grpc::Status s) {
void Finish(::grpc::Status s) ABSL_LOCKS_EXCLUDED(stream_mu_) {
ServerCallbackReaderWriter<Request, Response>* stream =
stream_.load(std::memory_order_acquire);
if (stream == nullptr) {
@ -481,7 +483,7 @@ class ServerBidiReactor : public internal::ServerReactor {
::grpc::WriteOptions write_options_wanted;
::grpc::Status status_wanted;
};
PreBindBacklog backlog_ /* GUARDED_BY(stream_mu_) */;
PreBindBacklog backlog_ ABSL_GUARDED_BY(stream_mu_);
};
/// \a ServerReadReactor is the interface for a client-streaming RPC.
@ -492,7 +494,7 @@ class ServerReadReactor : public internal::ServerReactor {
~ServerReadReactor() override = default;
/// The following operation initiations are exactly like ServerBidiReactor.
void StartSendInitialMetadata() {
void StartSendInitialMetadata() ABSL_LOCKS_EXCLUDED(reader_mu_) {
ServerCallbackReader<Request>* reader =
reader_.load(std::memory_order_acquire);
if (reader == nullptr) {
@ -505,7 +507,7 @@ class ServerReadReactor : public internal::ServerReactor {
}
reader->SendInitialMetadata();
}
void StartRead(Request* req) {
void StartRead(Request* req) ABSL_LOCKS_EXCLUDED(reader_mu_) {
ServerCallbackReader<Request>* reader =
reader_.load(std::memory_order_acquire);
if (reader == nullptr) {
@ -518,7 +520,7 @@ class ServerReadReactor : public internal::ServerReactor {
}
reader->Read(req);
}
void Finish(::grpc::Status s) {
void Finish(::grpc::Status s) ABSL_LOCKS_EXCLUDED(reader_mu_) {
ServerCallbackReader<Request>* reader =
reader_.load(std::memory_order_acquire);
if (reader == nullptr) {
@ -544,7 +546,8 @@ class ServerReadReactor : public internal::ServerReactor {
// May be overridden by internal implementation details. This is not a public
// customization point.
virtual void InternalBindReader(ServerCallbackReader<Request>* reader) {
virtual void InternalBindReader(ServerCallbackReader<Request>* reader)
ABSL_LOCKS_EXCLUDED(reader_mu_) {
grpc::internal::MutexLock l(&reader_mu_);
if (GPR_UNLIKELY(backlog_.send_initial_metadata_wanted)) {
@ -568,7 +571,7 @@ class ServerReadReactor : public internal::ServerReactor {
Request* read_wanted = nullptr;
::grpc::Status status_wanted;
};
PreBindBacklog backlog_ /* GUARDED_BY(reader_mu_) */;
PreBindBacklog backlog_ ABSL_GUARDED_BY(reader_mu_);
};
/// \a ServerWriteReactor is the interface for a server-streaming RPC.
@ -579,7 +582,7 @@ class ServerWriteReactor : public internal::ServerReactor {
~ServerWriteReactor() override = default;
/// The following operation initiations are exactly like ServerBidiReactor.
void StartSendInitialMetadata() {
void StartSendInitialMetadata() ABSL_LOCKS_EXCLUDED(writer_mu_) {
ServerCallbackWriter<Response>* writer =
writer_.load(std::memory_order_acquire);
if (writer == nullptr) {
@ -595,7 +598,8 @@ class ServerWriteReactor : public internal::ServerReactor {
void StartWrite(const Response* resp) {
StartWrite(resp, ::grpc::WriteOptions());
}
void StartWrite(const Response* resp, ::grpc::WriteOptions options) {
void StartWrite(const Response* resp, ::grpc::WriteOptions options)
ABSL_LOCKS_EXCLUDED(writer_mu_) {
ServerCallbackWriter<Response>* writer =
writer_.load(std::memory_order_acquire);
if (writer == nullptr) {
@ -610,7 +614,7 @@ class ServerWriteReactor : public internal::ServerReactor {
writer->Write(resp, options);
}
void StartWriteAndFinish(const Response* resp, ::grpc::WriteOptions options,
::grpc::Status s) {
::grpc::Status s) ABSL_LOCKS_EXCLUDED(writer_mu_) {
ServerCallbackWriter<Response>* writer =
writer_.load(std::memory_order_acquire);
if (writer == nullptr) {
@ -629,7 +633,7 @@ class ServerWriteReactor : public internal::ServerReactor {
void StartWriteLast(const Response* resp, ::grpc::WriteOptions options) {
StartWrite(resp, options.set_last_message());
}
void Finish(::grpc::Status s) {
void Finish(::grpc::Status s) ABSL_LOCKS_EXCLUDED(writer_mu_) {
ServerCallbackWriter<Response>* writer =
writer_.load(std::memory_order_acquire);
if (writer == nullptr) {
@ -654,7 +658,8 @@ class ServerWriteReactor : public internal::ServerReactor {
friend class ServerCallbackWriter<Response>;
// May be overridden by internal implementation details. This is not a public
// customization point.
virtual void InternalBindWriter(ServerCallbackWriter<Response>* writer) {
virtual void InternalBindWriter(ServerCallbackWriter<Response>* writer)
ABSL_LOCKS_EXCLUDED(writer_mu_) {
grpc::internal::MutexLock l(&writer_mu_);
if (GPR_UNLIKELY(backlog_.send_initial_metadata_wanted)) {
@ -687,7 +692,7 @@ class ServerWriteReactor : public internal::ServerReactor {
::grpc::WriteOptions write_options_wanted;
::grpc::Status status_wanted;
};
PreBindBacklog backlog_ /* GUARDED_BY(writer_mu_) */;
PreBindBacklog backlog_ ABSL_GUARDED_BY(writer_mu_);
};
class ServerUnaryReactor : public internal::ServerReactor {
@ -696,7 +701,7 @@ class ServerUnaryReactor : public internal::ServerReactor {
~ServerUnaryReactor() override = default;
/// StartSendInitialMetadata is exactly like ServerBidiReactor.
void StartSendInitialMetadata() {
void StartSendInitialMetadata() ABSL_LOCKS_EXCLUDED(call_mu_) {
ServerCallbackUnary* call = call_.load(std::memory_order_acquire);
if (call == nullptr) {
grpc::internal::MutexLock l(&call_mu_);
@ -711,7 +716,7 @@ class ServerUnaryReactor : public internal::ServerReactor {
/// Finish is similar to ServerBidiReactor except for one detail.
/// If the status is non-OK, any message will not be sent. Instead,
/// the client will only receive the status and any trailing metadata.
void Finish(::grpc::Status s) {
void Finish(::grpc::Status s) ABSL_LOCKS_EXCLUDED(call_mu_) {
ServerCallbackUnary* call = call_.load(std::memory_order_acquire);
if (call == nullptr) {
grpc::internal::MutexLock l(&call_mu_);
@ -734,7 +739,8 @@ class ServerUnaryReactor : public internal::ServerReactor {
friend class ServerCallbackUnary;
// May be overridden by internal implementation details. This is not a public
// customization point.
virtual void InternalBindCall(ServerCallbackUnary* call) {
virtual void InternalBindCall(ServerCallbackUnary* call)
ABSL_LOCKS_EXCLUDED(call_mu_) {
grpc::internal::MutexLock l(&call_mu_);
if (GPR_UNLIKELY(backlog_.send_initial_metadata_wanted)) {
@ -754,7 +760,7 @@ class ServerUnaryReactor : public internal::ServerReactor {
bool finish_wanted = false;
::grpc::Status status_wanted;
};
PreBindBacklog backlog_ /* GUARDED_BY(call_mu_) */;
PreBindBacklog backlog_ ABSL_GUARDED_BY(call_mu_);
};
namespace internal {

Write Preview
Loading…
Cancel
Save