kkm
7 years ago
commit
51e41ae7d5
648 changed files with 29115 additions and 9095 deletions
@ -1,6 +1,6 @@ |
||||
--- |
||||
Checks: 'modernize-use-nullptr,google-build-namespaces,google-build-explicit-make-pair,readability-function-size,performance-*' |
||||
WarningsAsErrors: 'modernize-use-nullptr,google-build-namespaces,google-build-explicit-make-pair,readability-function-size,performance-*' |
||||
Checks: 'modernize-use-nullptr,google-build-namespaces,google-build-explicit-make-pair,readability-function-size,performance-*,bugprone-*' |
||||
WarningsAsErrors: 'modernize-use-nullptr,google-build-namespaces,google-build-explicit-make-pair,readability-function-size,performance-*,bugprone-*' |
||||
CheckOptions: |
||||
- key: readability-function-size.StatementThreshold |
||||
value: '450' |
||||
|
||||
@ -0,0 +1,2 @@ |
||||
daysUntilLock: 90 |
||||
lockComment: false |
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,32 @@ |
||||
# Polling Engine Usage on gRPC client and Server |
||||
|
||||
_Author: Sree Kuchibhotla (@sreecha) - Sep 2018_ |
||||
|
||||
|
||||
This document talks about how polling engine is used in gRPC core (both on client and server code paths). |
||||
|
||||
## gRPC client |
||||
|
||||
### Relation between Call, Channel (sub-channels), Completion queue, `grpc_pollset` |
||||
- A gRPC Call is tied to a channel (more specifically a sub-channel) and a completion queue for the lifetime of the call. |
||||
- Once a _sub-channel_ is picked for the call, the file-descriptor (socket fd in case of TCP channels) is added to the pollset corresponding to call's completion queue. (Recall that as per [grpc-cq](grpc-cq.md), a completion queue has a pollset by default) |
||||
|
||||
 |
||||
|
||||
|
||||
### Making progress on Async `connect()` on sub-channels (`grpc_pollset_set` usecase) |
||||
- A gRPC channel is created between a client and a 'target'. The 'target' may resolve in to one or more backend servers. |
||||
- A sub-channel is the 'connection' from a client to the backend server |
||||
- While establishing sub-cannels (i.e connections) to the backends, gRPC issues async [`connect()`](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/tcp_client_posix.cc#L296) calls which may not complete right away. When the `connect()` eventually succeeds, the socket fd is make 'writable' |
||||
- This means that the polling engine must be monitoring all these sub-channel `fd`s for writable events and we need to make sure there is a polling thread that monitors all these fds |
||||
- To accomplish this, the `grpc_pollset_set` is used the following way (see picture below) |
||||
|
||||
 |
||||
|
||||
## gRPC server |
||||
|
||||
- The listening fd (i.e., the socket fd corresponding to the server listening port) is added to each of the server completion queues. Note that in gRPC we use SO_REUSEPORT option and create multiple listening fds but all of them map to the same listening port |
||||
- A new incoming channel is assigned to some server completion queue picked randomly (note that we currently [round-robin](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/tcp_server_posix.cc#L231) over the server completion queues) |
||||
|
||||
 |
||||
|
||||
@ -0,0 +1,64 @@ |
||||
# gRPC Completion Queue |
||||
|
||||
_Author: Sree Kuchibhotla (@sreecha) - Sep 2018_ |
||||
|
||||
Code: [completion_queue.cc](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/surface/completion_queue.cc) |
||||
|
||||
This document gives an overview of completion queue architecture and focuses mainly on the interaction between completion queue and the Polling engine layer. |
||||
|
||||
## Completion queue attributes |
||||
Completion queue has two attributes |
||||
|
||||
- Completion_type: |
||||
- GRPC_CQ_NEXT: grpc_completion_queue_next() can be called (but not grpc_completion_queue_pluck()) |
||||
- GRPC_CQ_PLUCK: grpc_completion_queue_pluck() can be called (but not grpc_completion_queue_next()) |
||||
- GRPC_CQ_CALLBACK: The tags in the queue are function pointers to callbacks. Also, neither next() nor pluck() can be called on this |
||||
|
||||
- Polling_type: |
||||
- GRPC_CQ_NON_POLLING: Threads calling completion_queue_next/pluck do not do any polling |
||||
- GRPC_CQ_DEFAULT_POLLING: Threads calling completion_queue_next/pluck do polling |
||||
- GRPC_CQ_NON_LISTENING: Functionally similar to default polling except for a boolean attribute that states that the cq is non-listening. This is used by the grpc-server code to not associate any listening sockets with this completion-queue’s pollset |
||||
|
||||
|
||||
## Details |
||||
|
||||
 |
||||
|
||||
|
||||
### **grpc\_completion\_queue\_next()** & **grpc_completion_queue_pluck()** APIS |
||||
|
||||
|
||||
``` C++ |
||||
grpc_completion_queue_next(cq, deadline)/pluck(cq, deadline, tag) { |
||||
while(true) { |
||||
\\ 1. If an event is queued in the completion queue, dequeue and return |
||||
\\ (in case of pluck() dequeue only if the tag is the one we are interested in) |
||||
|
||||
\\ 2. If completion queue shutdown return |
||||
|
||||
\\ 3. In case of pluck, add (tag, worker) pair to the tag<->worker map on the cq |
||||
|
||||
\\ 4. Call grpc_pollset_work(cq’s-pollset, deadline) to do polling |
||||
\\ Note that if this function found some fds to be readable/writable/error, |
||||
\\ it would have scheduled those closures (which may queue completion events |
||||
\\ on SOME completion queue - not necessarily this one) |
||||
} |
||||
} |
||||
``` |
||||
|
||||
### Queuing a completion event (i.e., "tag") |
||||
|
||||
``` C++ |
||||
grpc_cq_end_op(cq, tag) { |
||||
\\ 1. Queue the tag in the event queue |
||||
|
||||
\\ 2. Find the pollset corresponding to the completion queue |
||||
\\ (i) If the cq is of type GRPC_CQ_NEXT, then KICK ANY worker |
||||
\\ i.e., call grpc_pollset_kick(pollset, nullptr) |
||||
\\ (ii) If the cq is of type GRPC_CQ_PLUCK, then search the tag<->worker |
||||
\\ map on the completion queue to find the worker. Then specifically |
||||
\\ kick that worker i.e call grpc_pollset_kick(pollset, worker) |
||||
} |
||||
|
||||
``` |
||||
|
||||
@ -0,0 +1,154 @@ |
||||
# Polling Engines |
||||
|
||||
_Author: Sree Kuchibhotla (@sreecha) - Sep 2018_ |
||||
|
||||
|
||||
## Why do we need a 'polling engine' ? |
||||
|
||||
Polling engine component was created for the following reasons: |
||||
|
||||
- gRPC code deals with a bunch of file descriptors on which events like descriptor being readable/writable/error have to be monitored |
||||
- gRPC code knows the actions to perform when such events happen |
||||
- For example: |
||||
- `grpc_endpoint` code calls `recvmsg` call when the fd is readable and `sendmsg` call when the fd is writable |
||||
- ` tcp_client` connect code issues async `connect` and finishes creating the client once the fd is writable (i.e when the `connect` actually finished) |
||||
- gRPC needed some component that can "efficiently" do the above operations __using the threads provided by the applications (i.e., not create any new threads)__. Also by "efficiently" we mean optimized for latency and throughput |
||||
|
||||
|
||||
## Polling Engine Implementations in gRPC |
||||
There are multiple polling engine implementations depending on the OS and the OS version. Fortunately all of them expose the same interface |
||||
|
||||
- Linux: |
||||
|
||||
- **`epollex`** (default but requires kernel version >= 4.5), |
||||
- `epoll1` (If `epollex` is not available and glibc version >= 2.9) |
||||
- `poll` (If kernel does not have epoll support) |
||||
- `poll-cv` (If explicitly configured) |
||||
- Mac: **`poll`** (default), `poll-cv` (If explicitly configured) |
||||
- Windows: (no name) |
||||
- One-off polling engines: |
||||
- AppEngine platform: **`poll-cv`** (default) |
||||
- NodeJS : `libuv` polling engine implementation (requires different compile `#define`s) |
||||
|
||||
## Polling Engine Interface |
||||
|
||||
### Opaque Structures exposed by the polling engine |
||||
The following are the **Opaque** structures exposed by Polling Engine interface (NOTE: Different polling engine implementations have different definitions of these structures) |
||||
|
||||
- **grpc_fd:** Structure representing a file descriptor |
||||
- **grpc_pollset:** A set of one or more grpc_fds that are ‘polled’ for readable/writable/error events. One grpc_fd can be in multiple `grpc_pollset`s |
||||
- **grpc_pollset_worker:** Structure representing a ‘polling thread’ - more specifically, the thread that calls `grpc_pollset_work()` API |
||||
- **grpc_pollset_set:** A group of `grpc_fds`, `grpc_pollsets` and `grpc_pollset_sets` (yes, a `grpc_pollset_set` can contain other `grpc_pollset_sets`) |
||||
|
||||
### Polling engine API |
||||
|
||||
#### grpc_fd |
||||
- **grpc\_fd\_notify\_on\_[read|write|error]** |
||||
- Signature: `grpc_fd_notify_on_(grpc_fd* fd, grpc_closure* closure)` |
||||
- Register a [closure](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/closure.h#L67) to be called when the fd becomes readable/writable or has an error (In grpc parlance, we refer to this act as “arming the fd”) |
||||
- The closure is called exactly once per event. I.e once the fd becomes readable (or writable or error), the closure is fired and the fd is ‘unarmed’. To be notified again, the fd has to be armed again. |
||||
|
||||
- **grpc_fd_shutdown** |
||||
- Signature: `grpc_fd_shutdown(grpc_fd* fd)` |
||||
- Any current (or future) closures registered for readable/writable/error events are scheduled immediately with an error |
||||
|
||||
- **grpc_fd_orphan** |
||||
- Signature: `grpc_fd_orphan(grpc_fd* fd, grpc_closure* on_done, int* release_fd, char* reason)` |
||||
- Release the `grpc_fd` structure and call `on_done` closure when the operation is complete |
||||
- If `release_fd` is set to `nullptr`, then `close()` the underlying fd as well. If not, put the underlying fd in `release_fd` (and do not call `close()`) |
||||
- `release_fd` set to non-null in cases where the underlying fd is NOT owned by grpc core (like for example the fds used by C-Ares DNS resolver ) |
||||
|
||||
#### grpc_pollset |
||||
|
||||
- **grpc_pollset_add_fd ** |
||||
- Signature: `grpc_pollset_add_fd(grpc_pollset* ps, grpc_fd *fd)` |
||||
- Add fd to pollset |
||||
> **NOTE**: There is no `grpc_pollset_remove_fd`. This is because calling `grpc_fd_orphan()` will effectively remove the fd from all the pollsets it’s a part of |
||||
|
||||
- ** grpc_pollset_work ** |
||||
- Signature: `grpc_pollset_work(grpc_pollset* ps, grpc_pollset_worker** worker, grpc_millis deadline)` |
||||
> **NOTE**: `grpc_pollset_work()` requires the pollset mutex to be locked before calling it. Shortly after calling `grpc_pollset_work()`, the function populates the `*worker` pointer (among other things) and releases the mutex. Once `grpc_pollset_work()` returns, the `*worker` pointer is **invalid** and should not be used anymore. See the code in `completion_queue.cc` to see how this is used. |
||||
- Poll the fds in the pollset for events AND return when ANY of the following is true: |
||||
- Deadline expired |
||||
- Some fds in the pollset were found to be readable/writable/error and those associated closures were ‘scheduled’ (but not necessarily executed) |
||||
- worker is “kicked” (see `grpc_pollset_kick` for more details) |
||||
|
||||
- **grpc_pollset_kick** |
||||
- Signature: `grpc_pollset_kick(grpc_pollset* ps, grpc_pollset_worker* worker)` |
||||
- “Kick the worker” i.e Force the worker to return from grpc_pollset_work() |
||||
- If `worker == nullptr`, kick ANY worker active on that pollset |
||||
|
||||
#### grpc_pollset_set |
||||
|
||||
- **grpc\_pollset\_set\_[add|del]\_fd** |
||||
- Signature: `grpc_pollset_set_[add|del]_fd(grpc_pollset_set* pss, grpc_fd *fd)` |
||||
Add/Remove fd to the `grpc_pollset_set` |
||||
|
||||
- **grpc\_pollset\_set_[add|del]\_pollset** |
||||
- Signature: `grpc_pollset_set_[add|del]_pollset(grpc_pollset_set* pss, grpc_pollset* ps)` |
||||
- What does adding a pollset to a pollset_set mean ? |
||||
- It means that calling `grpc_pollset_work()` on the pollset will also poll all the fds in the pollset_set i.e semantically, it is similar to adding all the fds inside pollset_set to the pollset. |
||||
- This guarantee is no longer true once the pollset is removed from the pollset_set |
||||
|
||||
- **grpc\_pollset\_set_[add|del]\_pollset\_set** |
||||
- Signature: `grpc_pollset_set_[add|del]_pollset_set(grpc_pollset_set* bag, grpc_pollset_set* item)` |
||||
- Semantically, this is similar to adding all the fds in the ‘bag’ pollset_set to the ‘item’ pollset_set |
||||
|
||||
|
||||
#### Recap: |
||||
|
||||
__Relation between grpc_pollset_worker, grpc_pollset and grpc_fd:__ |
||||
|
||||
 |
||||
|
||||
__grpc_pollset_set__ |
||||
|
||||
 |
||||
|
||||
|
||||
## Polling Engine Implementations |
||||
|
||||
### epoll1 |
||||
|
||||
 |
||||
|
||||
Code at `src/core/lib/iomgr/ev_epoll1_posix.cc` |
||||
|
||||
- The logic to choose a designated poller is quite complicated. Pollsets are internally sharded into what are called `pollset_neighborhood` (a structure internal to `epoll1` polling engine implementation). `grpc_pollset_workers` that call `grpc_pollset_work` on a given pollset are all queued in a linked-list against the `grpc_pollset`. The head of the linked list is called "root worker" |
||||
|
||||
- There are as many neighborhoods as the number of cores. A pollset is put in a neighborhood based on the CPU core of the root worker thread. When picking the next designated poller, we always try to find another worker on the current pollset. If there are no more workers in the current pollset, a `pollset_neighborhood` listed is scanned to pick the next pollset and worker that could be the new designated poller. |
||||
- NOTE: There is room to tune this implementation. All we really need is good way to maintain a list of `grpc_pollset_workers` with a way to group them per-pollset (needed to implement `grpc_pollset_kick` semantics) and a way randomly select a new designated poller |
||||
|
||||
- See [`begin_worker()`](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/ev_epoll1_linux.cc#L729) function to see how a designated poller is chosen. Similarly [`end_worker()`](https://github.com/grpc/grpc/blob/v1.15.1/src/core/lib/iomgr/ev_epoll1_linux.cc#L916) function is called by the worker that was just out of `epoll_wait()` and will have to choose a new designated poller) |
||||
|
||||
|
||||
### epollex |
||||
|
||||
 |
||||
|
||||
Code at `src/core/lib/iomgr/ev_epollex_posix.cc` |
||||
|
||||
- FDs are added to multiple epollsets with EPOLLEXCLUSIVE flag. This prevents multiple worker threads from waking up from polling whenever the fd is readable/writable |
||||
|
||||
- A few observations: |
||||
|
||||
- If multiple pollsets are pointing to the same `Pollable`, then the `pollable` MUST be either empty or of type `PO_FD` (i.e single-fd) |
||||
- A multi-pollable has one-and-only-one incoming link from a pollset |
||||
- The same FD can be in multiple `Pollable`s (even if one of the `Pollable`s is of type PO_FD) |
||||
- There cannot be two `Pollable`s of type PO_FD for the same fd |
||||
|
||||
- Why do we need `Pollable` of type PO_FD and PO_EMTPY ? |
||||
- The main reason is the Sync client API |
||||
- We create one new completion queue per call. If we didn’t have PO_EMPTY and PO_FD type pollables, then every call on a given channel will effectively have to create a `Pollable` and hence an epollset. This is because every completion queue automatically creates a pollset and the channel fd will have to be put in that pollset. This clearly requires an epollset to put that fd. Creating an epollset per call (even if we delete the epollset once the call is completed) would mean a lot of sys calls to create/delete epoll fds. This is clearly not a good idea. |
||||
- With these new types of `Pollable`s, all pollsets (corresponding to the new per-call completion queue) will initially point to PO_EMPTY global epollset. Then once the channel fd is added to the pollset, the pollset will point to the `Pollable` of type PO_FD containing just that fd (i.e it will reuse the existing `Pollable`). This way, the epoll fd creation/deletion churn is avoided. |
||||
|
||||
|
||||
### Other polling engine implementations (poll and windows polling engine) |
||||
- **poll** polling engine: gRPC's `poll` polling engine is quite complicated. It uses the `poll()` function to do the polling (and hence it is for platforms like osx where epoll is not available) |
||||
- The implementation is further complicated by the fact that poll() is level triggered (just keep this in mind in case you wonder why the code at `src/core/lib/iomgr/ev_poll_posix.cc` is written a certain/seemingly complicated way :)) |
||||
|
||||
- **Polling engine on Windows**: Windows polling engine looks nothing like other polling engines |
||||
- Unlike the grpc polling engines for Unix systems (epollex, epoll1 and poll) Windows endpoint implementation and polling engine implementations are very closely tied together |
||||
- Windows endpoint read/write API implementations use the Windows IO API which require specifying an [I/O completion port](https://docs.microsoft.com/en-us/windows/desktop/fileio/i-o-completion-ports) |
||||
- In Windows polling engine’s grpc_pollset_work() implementation, ONE of the threads is chosen to wait on the I/O completion port while other threads wait on a condition variable (much like the turnstile polling in epollex/epoll1) |
||||
|
||||
|
After Width: | Height: | Size: 45 KiB |
|
After Width: | Height: | Size: 55 KiB |
|
After Width: | Height: | Size: 41 KiB |
|
After Width: | Height: | Size: 35 KiB |
|
After Width: | Height: | Size: 51 KiB |
|
After Width: | Height: | Size: 24 KiB |
|
After Width: | Height: | Size: 31 KiB |
|
After Width: | Height: | Size: 41 KiB |
@ -0,0 +1,164 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 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 GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
#define GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpc/impl/codegen/grpc_types.h> |
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc { |
||||
namespace internal { |
||||
|
||||
/// An exception-safe way of invoking a user-specified callback function
|
||||
template <class Func, class Arg> |
||||
void CatchingCallback(Func&& func, Arg&& arg) { |
||||
#if GRPC_ALLOW_EXCEPTIONS |
||||
try { |
||||
func(arg); |
||||
} catch (...) { |
||||
// nothing to return or change here, just don't crash the library
|
||||
} |
||||
#else // GRPC_ALLOW_EXCEPTIONS
|
||||
func(arg); |
||||
#endif // GRPC_ALLOW_EXCEPTIONS
|
||||
} |
||||
|
||||
// The contract on these tags is that they are single-shot. They must be
|
||||
// constructed and then fired at exactly one point. There is no expectation
|
||||
// that they can be reused without reconstruction.
|
||||
|
||||
class CallbackWithStatusTag |
||||
: public grpc_experimental_completion_queue_functor { |
||||
public: |
||||
// always allocated against a call arena, no memory free required
|
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithStatusTag)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithStatusTag(grpc_call* call, std::function<void(Status)> f, |
||||
CompletionQueueTag* ops) |
||||
: call_(call), func_(std::move(f)), ops_(ops) { |
||||
g_core_codegen_interface->grpc_call_ref(call); |
||||
functor_run = &CallbackWithStatusTag::StaticRun; |
||||
} |
||||
~CallbackWithStatusTag() {} |
||||
Status* status_ptr() { return &status_; } |
||||
|
||||
// force_run can not be performed on a tag if operations using this tag
|
||||
// have been sent to PerformOpsOnCall. It is intended for error conditions
|
||||
// that are detected before the operations are internally processed.
|
||||
void force_run(Status s) { |
||||
status_ = std::move(s); |
||||
Run(true); |
||||
} |
||||
|
||||
private: |
||||
grpc_call* call_; |
||||
std::function<void(Status)> func_; |
||||
CompletionQueueTag* ops_; |
||||
Status status_; |
||||
|
||||
static void StaticRun(grpc_experimental_completion_queue_functor* cb, |
||||
int ok) { |
||||
static_cast<CallbackWithStatusTag*>(cb)->Run(static_cast<bool>(ok)); |
||||
} |
||||
void Run(bool ok) { |
||||
void* ignored = ops_; |
||||
|
||||
GPR_CODEGEN_ASSERT(ops_->FinalizeResult(&ignored, &ok)); |
||||
GPR_CODEGEN_ASSERT(ignored == ops_); |
||||
|
||||
// Last use of func_ or status_, so ok to move them out
|
||||
CatchingCallback(std::move(func_), std::move(status_)); |
||||
|
||||
func_ = nullptr; // reset to clear this out for sure
|
||||
status_ = Status(); // reset to clear this out for sure
|
||||
g_core_codegen_interface->grpc_call_unref(call_); |
||||
} |
||||
}; |
||||
|
||||
class CallbackWithSuccessTag |
||||
: public grpc_experimental_completion_queue_functor { |
||||
public: |
||||
// always allocated against a call arena, no memory free required
|
||||
static void operator delete(void* ptr, std::size_t size) { |
||||
assert(size == sizeof(CallbackWithSuccessTag)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { assert(0); } |
||||
|
||||
CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f, |
||||
CompletionQueueTag* ops) |
||||
: call_(call), func_(std::move(f)), ops_(ops) { |
||||
g_core_codegen_interface->grpc_call_ref(call); |
||||
functor_run = &CallbackWithSuccessTag::StaticRun; |
||||
} |
||||
|
||||
CompletionQueueTag* ops() { return ops_; } |
||||
|
||||
// force_run can not be performed on a tag if operations using this tag
|
||||
// have been sent to PerformOpsOnCall. It is intended for error conditions
|
||||
// that are detected before the operations are internally processed.
|
||||
void force_run(bool ok) { Run(ok); } |
||||
|
||||
private: |
||||
grpc_call* call_; |
||||
std::function<void(bool)> func_; |
||||
CompletionQueueTag* ops_; |
||||
|
||||
static void StaticRun(grpc_experimental_completion_queue_functor* cb, |
||||
int ok) { |
||||
static_cast<CallbackWithSuccessTag*>(cb)->Run(static_cast<bool>(ok)); |
||||
} |
||||
void Run(bool ok) { |
||||
void* ignored = ops_; |
||||
bool new_ok = ok; |
||||
GPR_CODEGEN_ASSERT(ops_->FinalizeResult(&ignored, &new_ok)); |
||||
GPR_CODEGEN_ASSERT(ignored == ops_); |
||||
|
||||
// Last use of func_, so ok to move it out for rvalue call above
|
||||
CatchingCallback(std::move(func_), ok); |
||||
|
||||
func_ = nullptr; // reset to clear this out for sure
|
||||
g_core_codegen_interface->grpc_call_unref(call_); |
||||
} |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CALLBACK_COMMON_H
|
||||
@ -0,0 +1,95 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 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 GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
#define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H |
||||
|
||||
#include <functional> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc { |
||||
|
||||
class Channel; |
||||
class ClientContext; |
||||
class CompletionQueue; |
||||
|
||||
namespace internal { |
||||
class RpcMethod; |
||||
|
||||
/// Perform a callback-based unary call
|
||||
/// TODO(vjpai): Combine as much as possible with the blocking unary call code
|
||||
template <class InputMessage, class OutputMessage> |
||||
void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CallbackUnaryCallImpl<InputMessage, OutputMessage> x( |
||||
channel, method, context, request, result, on_completion); |
||||
} |
||||
|
||||
template <class InputMessage, class OutputMessage> |
||||
class CallbackUnaryCallImpl { |
||||
public: |
||||
CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method, |
||||
ClientContext* context, const InputMessage* request, |
||||
OutputMessage* result, |
||||
std::function<void(Status)> on_completion) { |
||||
CompletionQueue* cq = channel->CallbackCQ(); |
||||
GPR_CODEGEN_ASSERT(cq != nullptr); |
||||
Call call(channel->CreateCall(method, context, cq)); |
||||
|
||||
using FullCallOpSet = |
||||
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, |
||||
CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>, |
||||
CallOpClientSendClose, CallOpClientRecvStatus>; |
||||
|
||||
auto* ops = new (g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call.call(), sizeof(FullCallOpSet))) FullCallOpSet; |
||||
|
||||
auto* tag = new (g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call.call(), sizeof(CallbackWithStatusTag))) |
||||
CallbackWithStatusTag(call.call(), on_completion, ops); |
||||
|
||||
// TODO(vjpai): Unify code with sync API as much as possible
|
||||
Status s = ops->SendMessage(*request); |
||||
if (!s.ok()) { |
||||
tag->force_run(s); |
||||
return; |
||||
} |
||||
ops->SendInitialMetadata(context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
ops->RecvInitialMetadata(context); |
||||
ops->RecvMessage(result); |
||||
ops->AllowNoMessage(); |
||||
ops->ClientSendClose(); |
||||
ops->ClientRecvStatus(context, tag->status_ptr()); |
||||
ops->set_cq_tag(tag); |
||||
call.PerformOps(ops); |
||||
} |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
|
||||
@ -0,0 +1,44 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 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 GRPCPP_IMPL_CODEGEN_CLIENT_INTERCEPTOR_H |
||||
#define GRPCPP_IMPL_CODEGEN_CLIENT_INTERCEPTOR_H |
||||
|
||||
#include <grpcpp/impl/codegen/interceptor.h> |
||||
|
||||
namespace grpc { |
||||
namespace experimental { |
||||
class ClientInterceptor { |
||||
public: |
||||
virtual ~ClientInterceptor() {} |
||||
|
||||
virtual void Intercept(InterceptorBatchMethods* methods) = 0; |
||||
}; |
||||
|
||||
class ClientRpcInfo {}; |
||||
|
||||
class ClientInterceptorFactoryInterface { |
||||
public: |
||||
virtual ~ClientInterceptorFactoryInterface() {} |
||||
virtual ClientInterceptor* CreateClientInterceptor(ClientRpcInfo* info) = 0; |
||||
}; |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_CLIENT_INTERCEPTOR_H
|
||||
@ -0,0 +1,67 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 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 GRPCPP_IMPL_CODEGEN_INTERCEPTOR_H |
||||
#define GRPCPP_IMPL_CODEGEN_INTERCEPTOR_H |
||||
|
||||
namespace grpc { |
||||
namespace experimental { |
||||
class InterceptedMessage { |
||||
public: |
||||
template <class M> |
||||
bool Extract(M* msg); // returns false if definitely invalid extraction
|
||||
template <class M> |
||||
M* MutableExtract(); |
||||
uint64_t length(); // length on wire
|
||||
}; |
||||
|
||||
enum class InterceptionHookPoints { |
||||
/* The first two in this list are for clients and servers */ |
||||
PRE_SEND_INITIAL_METADATA, |
||||
PRE_SEND_MESSAGE, |
||||
PRE_SEND_STATUS /* server only */, |
||||
/* The following three are for hijacked clients only and can only be
|
||||
registered by the global interceptor */ |
||||
PRE_RECV_INITIAL_METADATA, |
||||
PRE_RECV_MESSAGE, |
||||
PRE_RECV_STATUS, |
||||
/* The following two are for all clients and servers */ |
||||
POST_RECV_INITIAL_METADATA, |
||||
POST_RECV_MESSAGE, |
||||
POST_RECV_STATUS /* client only */, |
||||
POST_RECV_CLOSE /* server only */, |
||||
NUM_INTERCEPTION_HOOKS |
||||
}; |
||||
|
||||
class InterceptorBatchMethods { |
||||
public: |
||||
virtual ~InterceptorBatchMethods(); |
||||
// Queries to check whether the current batch has an interception hook point
|
||||
// of type \a type
|
||||
virtual bool QueryInterceptionHookPoint(InterceptionHookPoints type) = 0; |
||||
// Calling this will signal that the interceptor is done intercepting the
|
||||
// current batch of the RPC
|
||||
virtual void Proceed() = 0; |
||||
// Calling this indicates that the interceptor has hijacked the RPC (only
|
||||
// valid if the batch contains send_initial_metadata on the client side)
|
||||
virtual void Hijack() = 0; |
||||
}; |
||||
} // namespace experimental
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_INTERCEPTOR_H
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue