Description

This PR is the first step in enabling support for streaming data that transits through Dapr runtimes, with the initial focus on sidecar-to-sidecar communication, as well as service invocation(currently limited to HTTP only).

To explain the issue, here's a demo of a service "client" calling into "server" through Dapr, making an HTTP request. The payload is "big" (about 70KB) and for demonstration purposes, both "client" and "server" are slow at producing or consuming the data.

As you can see, the flow is:

1. "client" makes a request to its sidecar
2. The sidecar waits for the entire request to be complete and loads the entire data in memory, then serializes it (to protobuf)
3. "client" app's sidecar then sends the data to "server" app's sidecar (I did not add any "slowdown" in this step, but this does take some time especially with large payloads)
4. "server" app's sidecar waits for the entire payload to be received (from "client" app's sidecar) and loaded in memory, then un-serializes it
5. "server" app's sidecar sends the data to the "server" app

The flow is then repeated exactly the same way (but with roles inverted) for the response.

Because in every step the sidecars read the entire message in memory, this can make it challenging to use Dapr service-to-service invocation with large payloads (for example, images) because it adds significant latency. Additionally, because all the data is kept in memory in both serialized and unserialized forms, there's an increase in memory usage.

This PR changes the behavior, and now service-to-service invocation (same example) behaves this way:

By leveraging streaming, this is much faster, especially when transferring large files or when the producer is slow. Especially the TTFB (Time To First Byte) is much shorter.

What this PR includes, at a higher level:

• Changed the way sidecars invoke one another to use streams. The new method ServiceInvocation.CallLocalStream is a variant of ServiceInvocation.CallLocal that uses streams rather than making unary calls. Internally this uses a buffer of up to 4KB, so messages that are smaller than that (and that are received in a single chunk by the runtime) are transferred in a single block. This should allow having no performance impact for small payloads (compared to unary RPCs).
• In the HTTP API server, service invocation calls are now read as streams, so the runtime can begin invoking the other app as soon as the first bytes come in.
• Likewise, when the sidecar invokes the apps using HTTP, the data is now sent to the app as a stream. Apps can begin receiving data as soon as the runtime gets it.

It also includes a lot of internal changes to support the two things above.

For clients that invoke Dapr via regular HTTP calls and receive responses through a regular HTTP server, no change will be needed. They will be able to enjoy the benefits of streaming for free and without doing anything!

For clients that use the Dapr SDKs, we may need to update the SDKs to leverage streaming when possible and appropriate. That is a separate work item.

Lastly, this PR is not "complete" as it doesn't add support for streams everywhere where we could want them. For example, we will want to support clients that use gRPC too. We will also want to leverage streaming for reading/writing into state stores and bindings.

Talking to @artursouza, he recommended focusing on HTTP service invocation first, and take this as an opportunity to lay the groundwork that will ultimately be useful in supporting all other scenarios. In this regard, this PR does provide some solid foundations for the remaining work.

Implementation details

This PR creates a new gRPC method that is used for sidecar-to-sidecar communication.

Currently, sidecar-to-sidecar communication uses the CallLocal RPC:

// Invokes a method of the specific service.
rpc CallLocal (InternalInvokeRequest) returns (InternalInvokeResponse) {}

This RPC is not removed and is retained for backwards compatibility (e.g. when Dapr 1.8 invokes Dapr vNext), however it is not used otherwise.

A new RPC is added:

// Invokes a method of the specific service using a stream of data.
rpc CallLocalStream (stream InternalInvokeRequestStream) returns (stream InternalInvokeResponseStream) {}

The request and response objects are defined as:

// InternalInvokeRequestStream is a variant of InternalInvokeRequest used in streaming RPCs.
message InternalInvokeRequestStream {
  // Request details.
  // This does not contain any data in message.data.
  InternalInvokeRequest request = 1;

  // Chunk of data.
  StreamPayload payload = 2;
}

// InternalInvokeResponseStream is a variant of InternalInvokeResponse used in streaming RPCs.
message InternalInvokeResponseStream {
  // Response details.
  // This does not contain any data in message.data.
  InternalInvokeResponse response = 1;

  // Chunk of data.
  StreamPayload payload = 2;
}

// Chunk of data sent in a streaming request or response.
message StreamPayload {
  // Data sent in the chunk.
  google.protobuf.Any data = 1;

  // Set to true if this is the last chunk.
  bool complete = 2;
}

In the protos above, InternalInvokeRequest and InternalInvokeResponse are the existing protos, that are used by the unary CallLocal RPC. In the streamed version, they are wrapped in a proto together with a StreamPayload object.

Note that, although the RPC uses bi-directional streaming, it still behaves like a unary call, in the sense that once the connection has been established, first the sender sends the entire request, and then it reads the entire response from the receiver. We use streams to be able to send data in chunks, but the sidecars do not maintain a persistent stream that is used to exchange multiple messages or where communication is constantly bi-directional.

How it works:

1. The caller sidecar invokes CallLocalStream on the target sidecar and establishes the gRPC connection and the streaming RPC.
2. The caller sidecar sends the data in chunks (see below for details on how chunking works).
3. The target sidecar can begin processing the message (i.e. send it to the target app) as soon as the first chunk is received.
4. Once the full payload has been sent, the target sidecar waits for the response from the app (which is read as a stream).
5. The target sidecar sends the response in chunks to the caller sidecar, with the first chunk sent as soon as it's available.

Each chunk is a message of type InternalInvokeRequestStream or InternalInvokeResponseStream sent over the RPC, where:

• The first message in the stream MUST contain an InternalInvokeRequest or InternalInvokeResponse with the required keys present.
• The first message in the stream MAY contain a payload, but that is not required.
• Subsequent messages (any message except the first in the stream) MUST contain a payload and MUST NOT contain any other property.
• The last message in the stream MUST contain a payload with complete set to true. That message is assumed to be the last one from the sender and no more messages are to be sent in the stream after that.

The amount of data contained in payload.data is variable and it's up to the discretion of the sender. In this PR, chunks are at most 4KB in size, although senders may send smaller chunks if they wish. Receivers must not assume that messages will contain any specific number of bytes in the payload.

Note that it's possible for senders to send a single message in the stream. If the data is small and could fit in a single chunk, senders MAY choose to include a payload with complete=true in the first message. In this case, the entire message contains both an InternalInvokeRequest/InternalInvokeResponse with the required keys present AND a payload with data (if any) and complete=true. Receivers should assume that single message to be the entire communication from the sender.

To do

• Figure out a way to ensure backwards compatibility when a runtime that supports streaming (e.g. daprd "vNext") tries to invoke a runtime that does not (e.g. daprd 1.8)
• Update docs
• Review what updates are needed in the SDKs

Issue reference

Fixes #3103
Also impacts other issues such as #4866

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Unit tests passing
• End-to-end tests passing
• Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: @ItalyPaleAle TODO