aysnc send/recv, seriliaze/deserialize using threadpool.

[helinwang]

This PR added performance optimization: send/recv can use threadpool to serialize/deserialize tensor. (before this PR tries to add retry logic, now this PR only handles the threadpool performance optimization).

Fix: #7801

[gongweibao]

我看了一下文档，忽然有一个疑惑，既然这里是block住的（以前我看例子以为是Finish只是注册一下reply，status, tag等,Next才是block的。）

This function will return when either:

when the server's response message and status have been received.
when the server has returned a non-OK status (no message expected in this case).
when the call failed for some reason and the library generated a non-OK status.

那为何还需要completion_queue.Next(), 这里的Finish是说收到了消息但是消息还没有全部返回完毕？
我觉得grpc这个接口这么设计有点反人类。

我们还是需要找一个正经的Async接口。^_^

[gongweibao]

而且，现在server端是通过计数来生成名字的。/~https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/operators/recv_op.cc#L83

一旦重复发送就出错了。/~https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/operators/recv_op.cc#L133

[typhoonzero]

感觉既然叫Finish也是有道理的。

[gongweibao]

我测试了一下,这里的server response应该不是user level的消息，应该是grpc的框架把caller的数据发送完毕以后接收到response,提示数据发送完毕或者其他情况。

This function will return when either:
    when the server's response message and status have been received.
    when the server has returned a non-OK status (no message expected in this case).
    when the call failed for some reason and the library generated a non-OK status.

测试方式是：即便server端不处理client端发送的消息，只要框架成功接收，client端也是收到了server response. Server response is not user RPC reply.

不过以前的实现确实不是fully async。需要thread_pool去发送。

[gongweibao]

另外，猜测：Status可能是被复用的。RPC的过程中一直在被更新。

[helinwang]

谢谢！看来Finish并不收到处理结果。

[gongweibao]

如果参数多会有太多的thread生成出来，可能会有问题。

[typhoonzero]

嗯，不建议在每次call中创建thread或者可以用threadpool

[helinwang]

Thanks! Done.

[Yancey1989]

Sorry for later review, the threads number was set by std::thread::hardware_concurrency in the global ThreadPool, it's usually equal to CPU core numbers but not always. I think it's too few for IO threads, maybe we need another threadpool to deal with the IO thread. But we can merge this PR firstly, and do more improve for future.

[helinwang]

@gongweibao @typhoonzero thanks for the feedback! I removed the retry code since the error can not be reproduced anymore on the latest develop branch. And the retry logic need more thinking before we implement it.

This PR now only do concurrent send / recv, serialization / deserialization.

[typhoonzero]

LGTM! Just for reminding, parrallel_for also use the same thread pool, it's a singleton. When we need to do multi-thread + multi-node training we must separate these two threadpools.

[gongweibao]

赞
LGTM++

[helinwang]

Thanks! @typhoonzero @Yancey1989 I would not worry too much about the threadpool, because this PR for the most part only run serialization and deserialization on the threadpool. The send and recv are scheduled internally by grpc, since @gongweibao implemented grpc aysnc send/recv. I know there is a grpc::Finish() call in our threadpool, but @gongweibao tested it does not block until the server returns a response.

[gongweibao]

but @gongweibao tested it does not block until the server returns a response.

Some explanations:
The client does blocks until the server returns a response. The client doesn't block until the server returns an RPC reply.
A response is not an RPC reply.