English version

概述

在一些应用场景中, client或server需要向对面发送大量数据,这些数据非常大或者持续地在产生以至于无法放在一个RPC的附件中。比如一个分布式系统的不同节点间传递replica或snapshot。client/server之间虽然可以通过多次RPC把数据切分后传输过去,但存在如下问题:

  • 如果这些RPC是并行的,无法保证接收端有序地收到数据,拼接数据的逻辑相当复杂。
  • 如果这些RPC是串行的,每次传递都得等待一次网络RTT+处理数据的延时,特别是后者的延时可能是难以预估的。

为了让大块数据以流水线的方式在client/server之间传递, 我们提供了Streaming RPC这种交互模型。Streaming RPC让用户能够在client/service之间建立用户态连接,称为Stream, 同一个TCP连接之上能同时存在多个Stream。 Stream的传输数据以消息为基本单位, 输入端可以源源不断的往Stream中写入消息, 接收端会按输入端写入顺序收到消息。

Streaming RPC保证:

  • 有消息边界。
  • 接收消息的顺序和发送消息的顺序严格一致。
  • 全双工。
  • 支持流控。
  • 提供超时提醒

目前的实现还没有自动切割过大的消息,同一个tcp连接上的多个Stream之间可能有Head-of-line blocking问题,请尽量避免过大的单个消息,实现自动切割后我们会告知并更新文档。

例子见example/streaming_echo_c++

建立Stream

目前Stream都由Client端建立。Client先在本地创建一个Stream,再通过一次RPC(必须使用baidu_std协议)与指定的Service建立一个Stream,如果Service在收到请求之后选择接受这个Stream, 那在response返回Client后Stream就会建立成功。过程中的任何错误都把RPC标记为失败,同时也意味着Stream创建失败。用linux下建立连接的过程打比方,Client先创建socket(创建Stream),再调用connect尝试与远端建立连接(通过RPC建立Stream),远端accept后连接就建立了(service接受后创建成功)。

如果Client尝试向不支持Streaming RPC的老Server建立Stream,将总是失败。

程序中我们用StreamId代表一个Stream,对Stream的读写,关闭操作都将作用在这个Id上。

  1. struct StreamOptions
  2. // The max size of unconsumed data allowed at remote side.
  3. // If |max_buf_size| <= 0, there's no limit of buf size
  4. // default: 2097152 (2M)
  5. int max_buf_size;
  6. // Notify user when there's no data for at least |idle_timeout_ms|
  7. // milliseconds since the last time that on_received_messages or on_idle_timeout
  8. // finished.
  9. // default: -1
  10. long idle_timeout_ms;
  11. // How many messages at most passed to handler->on_received_messages
  12. // default: 1
  13. size_t max_messages_size;
  14. // Handle input message, if handler is NULL, the remote side is not allowd to
  15. // write any message, who will get EBADF on writting
  16. // default: NULL
  17. StreamInputHandler* handler;
  18. };
  19. // [Called at the client side]
  20. // Create a Stream at client-side along with the |cntl|, which will be connected
  21. // when receiving the response with a Stream from server-side. If |options| is
  22. // NULL, the Stream will be created with default options
  23. // Return 0 on success, -1 otherwise
  24. int StreamCreate(StreamId* request_stream, Controller &cntl, const StreamOptions* options);

接受Stream

如果client在RPC上附带了一个Stream, service在收到RPC后可以通过调用StreamAccept接受。接受后Server端对应产生的Stream存放在response_stream中,Server可通过这个Stream向Client发送数据。

  1. // [Called at the server side]
  2. // Accept the Stream. If client didn't create a Stream with the request
  3. // (cntl.has_remote_stream() returns false), this method would fail.
  4. // Return 0 on success, -1 otherwise.
  5. int StreamAccept(StreamId* response_stream, Controller &cntl, const StreamOptions* options);

读取Stream

在建立或者接受一个Stream的时候, 用户可以继承StreamInputHandler并把这个handler填入StreamOptions中. 通过这个handler,用户可以处理对端的写入数据,连接关闭以及idle timeout

  1. class StreamInputHandler {
  2. public:
  3. // 当接收到消息后被调用
  4. virtual int on_received_messages(StreamId id, butil::IOBuf *const messages[], size_t size) = 0;
  5. // 当Stream上长时间没有数据交互后被调用
  6. virtual void on_idle_timeout(StreamId id) = 0;
  7. // 当Stream被关闭时被调用
  8. virtual void on_closed(StreamId id) = 0;
  9. };

第一次收到请求的时间

在client端,如果建立过程是一次同步RPC, 那在等待的线程被唤醒之后,on_received_message就可能会被调用到。 如果是异步RPC请求, 那等到这次请求的done->Run() 执行完毕之后, on_received_message就可能会被调用。

在server端, 当框架传入的done->Run()被调用完之后, on_received_message就可能会被调用。

写入Stream

  1. // Write |message| into |stream_id|. The remote-side handler will received the
  2. // message by the written order
  3. // Returns 0 on success, errno otherwise
  4. // Errno:
  5. // - EAGAIN: |stream_id| is created with positive |max_buf_size| and buf size
  6. // which the remote side hasn't consumed yet excceeds the number.
  7. // - EINVAL: |stream_id| is invalied or has been closed
  8. int StreamWrite(StreamId stream_id, const butil::IOBuf &message);

流控

当存在较多已发送但未接收的数据时,发送端的Write操作会立即失败(返回EAGAIN), 这时候可以通过同步或异步的方式等待对端消费掉数据。

  1. // Wait util the pending buffer size is less than |max_buf_size| or error occurs
  2. // Returns 0 on success, errno otherwise
  3. // Errno:
  4. // - ETIMEDOUT: when |due_time| is not NULL and time expired this
  5. // - EINVAL: the Stream was close during waiting
  6. int StreamWait(StreamId stream_id, const timespec* due_time);
  7. // Async wait
  8. void StreamWait(StreamId stream_id, const timespec *due_time,
  9. void (*on_writable)(StreamId stream_id, void* arg, int error_code),
  10. void *arg);

关闭Stream

  1. // Close |stream_id|, after this function is called:
  2. // - All the following |StreamWrite| would fail
  3. // - |StreamWait| wakes up immediately.
  4. // - Both sides |on_closed| would be notifed after all the pending buffers have
  5. // been received
  6. // This function could be called multiple times without side-effects
  7. int StreamClose(StreamId stream_id);