Stream API & RPC services

[ondrej-fabry]

This issue tracks progress of design & implementation of new Stream API & generated RPC services.
This post will be updated over time to provide all the relevant info.

---

Stream API

The GoVPP originally came with Channels for sending and receiving messages, which supported normal requests (request/reply) types, multi-request (dump/details) and subscriptions (want/event). Internally it came with buffered channels (100 capacity by default).

Old Channel API

Code

// ChannelProvider provides the communication channel with govpp core.
type ChannelProvider interface {
	// NewAPIChannel returns a new channel for communication with VPP via govpp core.
	// It uses default buffer sizes for the request and reply Go channels.
	NewAPIChannel() (Channel, error)

	// NewAPIChannelBuffered returns a new channel for communication with VPP via govpp core.
	// It allows to specify custom buffer sizes for the request and reply Go channels.
	NewAPIChannelBuffered(reqChanBufSize, replyChanBufSize int) (Channel, error)
}

// Channel provides methods for direct communication with VPP channel.
type Channel interface {
	// SendRequest asynchronously sends a request to VPP. Returns a request context, that can be used to call ReceiveReply.
	// In case of any errors by sending, the error will be delivered to ReplyChan (and returned by ReceiveReply).
	SendRequest(msg Message) RequestCtx

	// SendMultiRequest asynchronously sends a multipart request (request to which multiple responses are expected) to VPP.
	// Returns a multipart request context, that can be used to call ReceiveReply.
	// In case of any errors by sending, the error will be delivered to ReplyChan (and returned by ReceiveReply).
	SendMultiRequest(msg Message) MultiRequestCtx

	// SubscribeNotification subscribes for receiving of the specified notification messages via provided Go channel.
	// Note that the caller is responsible for creating the Go channel with preferred buffer size. If the channel's
	// buffer is full, the notifications will not be delivered into it.
	SubscribeNotification(notifChan chan Message, event Message) (SubscriptionCtx, error)

	// SetReplyTimeout sets the timeout for replies from VPP. It represents the maximum time the API waits for a reply
	// from VPP before returning an error.
	SetReplyTimeout(timeout time.Duration)

	// CheckCompatibility checks the compatiblity for the given messages.
	// It will return an error if any of the given messages are not compatible.
	CheckCompatiblity(msgs ...Message) error

	// Close closes the API channel and releases all API channel-related resources
	// in the ChannelProvider.
	Close()
}

// RequestCtx is helper interface which allows to receive reply on request.
type RequestCtx interface {
	// ReceiveReply receives a reply from VPP (blocks until a reply is delivered
	// from VPP, or until an error occurs). The reply will be decoded into the msg
	// argument. Error will be returned if the response cannot be received or decoded.
	ReceiveReply(msg Message) error
}

// MultiRequestCtx is helper interface which allows to receive reply on multi-request.
type MultiRequestCtx interface {
	// ReceiveReply receives a reply from VPP (blocks until a reply is delivered
	// from VPP, or until an error occurs).The reply will be decoded into the msg
	// argument. If the last reply has been already consumed, lastReplyReceived is
	// set to true. Do not use the message itself if lastReplyReceived is
	// true - it won't be filled with actual data.Error will be returned if the
	// response cannot be received or decoded.
	ReceiveReply(msg Message) (lastReplyReceived bool, err error)
}

// SubscriptionCtx is helper interface which allows to control subscription for
// notification events.
type SubscriptionCtx interface {
	// Unsubscribe unsubscribes from receiving the notifications tied to the
	// subscription context.
	Unsubscribe() error
}

However this Channel API could not be used for some new types of multi-requests (dump/details+reply) that VPP later introduced. This new type of requests could not work with the old Channel API since it was not possible to receive message without knowing exactly what type of message comes (receiving message was a parameter not a return value). The implementation of Channels also does several stuff behind the scenes which were either not very controllable by user (sending ControlPing for multi-requests) or just plain inefficient (using reflect to get Retval field).

For this reasons we introduced new API for sending and receiving messages: Stream.

New Stream API

Code

// Connection represents the client connection to VPP API.
//
// NOTE: This API is EXPERIMENTAL.
type Connection interface {
	// NewStream creates a new stream for sending and receiving messages.
	// Context can be used to close the stream using cancel or timeout.
	NewStream(ctx context.Context, options ...StreamOption) (Stream, error)

	// Invoke can be used for a simple request-reply RPC.
	// It creates stream and calls SendMsg with req and RecvMsg which returns
	// reply.
	Invoke(ctx context.Context, req Message, reply Message) error
}

// Stream provides low-level access for sending and receiving messages.
// Users should handle correct type and ordering of messages.
//
// It is not safe to call these methods on the same stream in different
// goroutines.
//
// NOTE: This API is EXPERIMENTAL.
type Stream interface {
	// SendMsg sends a message to the client.
	// It blocks until message is sent to the transport.
	SendMsg(Message) error

	// RecvMsg blocks until a message is received or error occurs.
	RecvMsg() (Message, error)

	// Close closes the stream. Calling SendMsg and RecvMsg will return error
	// after closing stream.
	Close() error
}

// StreamOption allows customizing a Stream. Available options are:
// - WithRequestSize
// - WithReplySize
// - WithReplyTimeout
type StreamOption func(Stream)

This new API is supposed to be less restrictive and much simpler. This means it does not try to do any assumptions about the semantics of the VPP API requests. It only has two methods for messages: SendMsg(Message) error and RecvMsg() (Message, error).

However, the new Stream API does not currently come with completely new implementation and for time/cycles reasons it still uses the channel implementation under the hood. This means it creates a new channel for each stream which brings some extra overhead which is not even necessary or even used.

The plan is to switch the Stream to a new implementation which would get rid of the channels and provide much more efficient way.

TBD

RPC services

With the new Stream API, being so low-level, users would have to do much of the things done by Channels before themselves; e.g. send ControlPings after each dump manually. For this reasons we also introduced generated RPC methods, where each generate binapi package came with a service interface and also client implementation of this interface where all of the boilerplate was automatically generated.

Here is example code using raw sending (Channel): https://github.com/FDio/govpp/blob/master/examples/simple-client/simple_client.go
Here is example code using RPCs (Stream): https://github.com/FDio/govpp/blob/master/examples/rpc-service/rpc_service.go

The generated RPCs have made it possible to get rid of the boilerplate and allow users to use simple methods. For those who are familiar with Protobuf and gRPC, this has been very similar in that way.

The generated code for RPC takes care of:

• matching correct Request/Response message pairs of the particular service
• handling the Retval value and returning error in case it's non-zero
• sending extra ControlPing request after dumps
• handling correct message type for special dump requests (dump/details/reply)
• allowing to pass the context.Context value from the user to the actual requests processing

TBD