Use cases of AsyncNotifyWhenDone

[Debashish Deka]

In the greeter example provided in the GitHub repo, we see this line: (https://github.com/grpc/grpc/blob/v1.24.0/examples/cpp/helloworld/greeter_async_server.cc)

responder_.Finish(reply_, Status::OK, this);

Because of the "this" pointer argument, cq_.Next() triggers the event and we clear the CallData object inside in the "FINISH" state as per the example provided.

I tried to use "AsyncNotifyWhenDone" just before RequestsayHello():

ctx_.AsyncNotifyWhenDone((void*)(extFunction)) where extFunction is some random function declared in the file. I found that, cq_.Next() triggered a new event with tag value equal to the extFunction address.

So, we get two events now, one because of the Finish call and the other due to the AsyncNotifyWhenDone() call. I want to ask, what could be a use of using AsyncNotifyWhenDone
? I could not find any special requirements apart from deleting the CallData instance.

If I am wrong in part of the question. Please correct me.

Thank you! 
 

 

[Acorn Pooley]

My understanding is that AsyncNotifyWhenDone is useful if you want to be notified that the request was cancelled (either explicitly cancelled by the caller or implicitly cancelled because the connection was dropped).  When you see the AsyncNotifyWhenDone you can check IsCancelled to see if the rpc got cancelled.  But you also get the AsyncNotifyWhenDone after the rpc is finished.  I'm not sure about the order of the Finish and Done events - I'm not sure it is  guaranteed they always come in the same order.  If you delete the "this" pointer in either event then you have to be careful not to also delete it (or otherwise dereference it) in the other event.  I deal with this by using a "tag" that is not actually a pointer - I use an index into an array of currently executing rpcs plus some "salt" bits that make each tag unique.

If you don't care if the rpc gets cancelled then I don't think there is much use for the AsyncNotifyWhenDone AFAICT.

Cheers,

Acorn

[mailvi...@gmail.com]

Thank You

[Vijay Pai]

Acorn, thanks for the detailed and correct response. I'll go one step further than your fourth sentence, though, and say that there is explicitly no guarantee about the ordering of the Finish and Done events (or, for that matter, any concurrent operations on the CQ). Your solution for dealing with that sounds perfect.

- Vijay

[Ctmahapa95]

Alternatively you can keep track of number of async operations in progress and a flag set by asyncnotifywhendone as done by contributed code from Arpit of Electronic Arts for managing gRPC async server state. You can delete (like Arpit’s code) or reset (like Vijay’s code in the benchmark code) when both counter is 0 and the flag is true.

--
You received this message because you are subscribed to the Google Groups "grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email to grpc-io+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/04d9623e-b5f4-4551-8630-9b35a335f00e%40googlegroups.com.

[Acorn Pooley]

Someone (offline) asked me to elaborate on how I use the "tag" as briefly described in my previous post.

The tag is a void*, meaning that on a 64 bit computer it is 64 bits of information.

I use some of those bits as an "rpc identifier" and some of those bits as an indication of what this tag represents.

For example, lets say I decide to use bits 0-20 to identify my rpc, and bits 21-23 to identify the action.

Action can be one of the values

enum Action {

   RPC_STARTED = 1,

   RPC_REQUEST_RECEIVED = 2,

   RPC_RESPONSE_SENT = 3,

   RPC_FINISHED = 4,

   RPC_ASYNC_DONE = 5,

} ;

and int rpc_id can be any value from 0 to 0xfffff  (20 bits).

When I create a new RPC object I assign it a new rpc_id from an incrementing counter.

When I call any grpc function that requires a tag, I use a tag that looks like

     void* tag = ((void*)(rpc_id | (action << 20)));

Then when I get a tag out of the completion queue, I can tell what the tag indicates:

     int rpc_id = (uint32_t)(tag) & 0xfffff;

     Action action = (Action)(((uint32_t)(tag) >> 21) & 0x7);

Now I can look up the rpc object with something like

     // I implement RpcIdToRpcPointer() as a lookup into an array of currently running rpc

     // objects.  It could also be implemented as a map lookup.

     MyRpcClass* rpc = RpcIdToRpcPointer(rpc_id);  

     if (rpc == nullptr) {

          LOG("Ignoring tag from bad rpc_id which no longer exists\n");

     }

And what I do with it depends on the action.

This allows me to distinguish a AsyncNotifyWhenDone tag (which I would handle by deleting the rpc object) vs a RPC_FINISHED (which I might also handle by deleting the rpc object) vs a RESPONSE_SENT tag (which I would handle by sending the next streaming response) vs various other actions.

Since the RPC_FINISHED and RPC_ASYNC_DONE both cause me to delete the RPC, it is important that I do not use a pointer to the rpc object as a tag.  If one of those events causes the rpc to be deleted, and then later I get the tag for the other event, I could end up deleting the object twice (big problem).  By instead using an rpc_id to look up the rpc I can tell if I get a tag from an rpc that has already been deleted (the RpcIdToRpcPointer() returns nullptr).  When RpcIdToRpcPointer() returns nullptr I just assume the rpc was previously deleted and ignore the tag.

The salt:

It would be nice if grpc would tell you whether there are any tags "pending" (waiting for an event to occur, or waiting in a completion queue if the event already occurred).  But there does not seem to be a reliable way to do this.  It is unclear to me what happens to tags associated with streaming response sent, or streaming request received, if the rpc is cancelled.  So I have to assume when I delete the rpc object that some tags for that rpc may pop out of the completion queue at some later time.  Since I don't know how long I have to wait before that might happen, I have to be careful about reusing an rpc_id for a new rpc if there may be tags for the old rpc that can come out of the completion queue later.

To address this, I use some of the bits of the tag as a "salt" to minimize the chance that a tag for an old deleted rpc is mistaken for a newer rpc with the same rpc_id.

For example, I can use bits 24-31 for my salt.  When I first use a particular rpc_id I set the salt to 0.  When that rpc gets deleted and I use the same rpc_id for a new rpc object, I increment the salt value to 1.  Now I generate my tag as

     void* tag = ((void*)(rpc_id | (action << 20) | (salt << 24)));

and when I get a tag from the completion queue I do

     int rpc_id = (uint32_t)(tag) & 0xfffff;

     Action action = (Action)(((uint32_t)(tag) >> 21) & 0x7);

     int salt = (((uint32_t)(tag) >> 24) & 0xff);

     MyRpcClass* rpc = RpcIdToRpcPointer(rpc_id, salt);

     // Note: on a 64 bit computer you can use many more bits for the salt and the rpc_id.

Now my  RpcIdToRpcPointer() function looks up the rpc object using the rpc_id (as a key for a map lookup or an index into an array of rpc objects), and then compares the salt to the salt stored in the rpc object.  If they do not match then I assume it is a tag for an old rpc and I ignore the tag.

This seems pretty complicated and I wish there were a simpler way to ensure that no tags from an "old" rpc are going to come out of a completion queue.  But when using AsyncNotifyWhenDone with async bidirectional streaming RPCs, it seems that something like this is necessary to ensure tags are not confused. I'd be interested to hear how other folks handle this.

Acorn