[Tarantool-patches] [RAFT 02/10] raft: relay status updates to followers

[Vladislav Shpilevoy]

Hi! Thanks for the patch!

> diff --git a/src/box/relay.cc b/src/box/relay.cc
> index a7843a8c2..be252cad1 100644
> --- a/src/box/relay.cc
> +++ b/src/box/relay.cc
> @@ -773,13 +774,40 @@ relay_send_initial_join_row(struct xstream *stream, struct xrow_header *row)
>  		relay_send(relay, row);
>  }
>  
> +static void
> +relay_send_raft(struct relay *relay, struct raft_request *req)
> +{
> +	struct xrow_header packet;
> +	xrow_encode_raft(&packet, &fiber()->gc, req);
> +	relay_send(relay, &packet);
> +}
> +
> +static void
> +relay_send_raft_msg(struct cmsg *msg)
> +{
> +	struct raft_broadcast_msg *raft_msg = (struct raft_broadcast_msg *)msg;
> +	struct relay *relay = container_of(msg->route[0].pipe, struct relay,
> +					   tx_pipe);
> +	relay_send_raft(relay, &raft_msg->req);
> +}
> +
> +void
> +relay_push_raft_msg(struct relay *relay, struct cmsg *msg,
> +		    struct cmsg_hop *route)
> +{
> +	route[0].f = relay_send_raft_msg;
> +	route[0].pipe = &relay->tx_pipe;
> +	route[1].f = raft_free_msg;
> +	route[1].pipe = NULL;
> +	cmsg_init(msg, route);
> +	cpipe_push(&relay->relay_pipe, msg);
> +}
> +
>  /** Send a single row to the client. */
>  static void
>  relay_send_row(struct xstream *stream, struct xrow_header *packet)
>  {
>  	struct relay *relay = container_of(stream, struct relay, stream);
> -	assert(iproto_type_is_dml(packet->type) ||
> -	       iproto_type_is_synchro_request(packet->type));
>  	if (packet->group_id == GROUP_LOCAL) {
>  		/*
>  		 * We do not relay replica-local rows to other
> @@ -796,6 +824,8 @@ relay_send_row(struct xstream *stream, struct xrow_header *packet)
>  		packet->group_id = GROUP_DEFAULT;
>  		packet->bodycnt = 0;
>  	}
> +	assert(iproto_type_is_dml(packet->type) ||
> +	       iproto_type_is_synchro_request(packet->type));

Why did you move this check, if Raft uses relay_send() anyway? Not
relay_send_row().

>  	/* Check if the rows from the instance are filtered. */
>  	if ((1 << packet->replica_id & relay->id_filter) != 0)
>  		return;