sendfd doesn't send fd to all connections

[Bolodya1997]

Related issues

1. Continues sendfd doesn't send fd on Forwarder/Endpoint iteration #1111.
2. Needed for Heal rework #1106.

Expected Behavior

In such case:

1. sendfd.Client sends fd on Close or Request.
2. connect.Client makes Close or Request to URL-1.
3. ...
4. connect.Client makes Close or Request to URL-n.

We want fd to be sent to all URL-1, ..., URL-n receivers.

Current Behavior

sendfd.Client sends fd only to URL-1.

Cause

1. sendfd.Client uses grpcfd.(*wrapPerRPCCredentials) for sending fd before the gRPC interaction.
2. grpcfd.(*wrapPerRPCCredentials) captures FDTransceiver from the first gRPC interaction (it would be URL-1 in our case) and sends all fds only to this transceiver:
   1. https://github.com/edwarnicke/grpcfd/blob/39c35952c7c38397177cf36ed7b8026ab88c61ec/per_rpc_transport_credentials.go#L39-L48
   2. https://github.com/edwarnicke/grpcfd/blob/39c35952c7c38397177cf36ed7b8026ab88c61ec/per_rpc_transport_credentials.go#L66-L72

Solution

Analysis

We have 4 different points where we need to use FDTransceiver in terms of linear execution:

1. Client requests Server and sends fd (sendfd.Client).
2. Server processes request and receives fd (recvfd.Server).
3. Server responses to the Client and sends fd (sendfd.Server).
4. Client processes response and receives fd (recvfd.Client).

We have the following:

1. At [1] we can request grpc.Peer, but we will get it only on [4].
2. At [2, 3] we can access grpc.Peer directly from the request context.

So in [2, 3, 4] cases we can simply use grpcfd.FromPeer to get FDTransceiver.

In [1] case we don't actually need FDTransceiver, we only need FDSender.

Edit grpcfd.(*wrapPerRPCCredentials)

1. Implement only FDSender interface.
2. Stop capturing FDTransceiver.

type wrapPerRPCCredentials struct {
	credentials.PerRPCCredentials
	transceiverFuncs []func(FDSender)
	executor         serialize.Executor
}

func (w *wrapPerRPCCredentials) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	<-w.executor.AsyncExec(func() {
		if sender, ok := FromContext(ctx); ok {
			for _, f := range w.senderFuncs {
				f(sender)
			}
		}
	})
	if w.PerRPCCredentials != nil {
		return w.PerRPCCredentials.GetRequestMetadata(ctx, uri...)
	}
	return map[string]string{}, nil
}

...

func (w *wrapPerRPCCredentials) SendFD(fd uintptr) <-chan error {
	out := make(chan error, 1)
	w.executor.AsyncExec(func() {
		w.transceiverFuncs = append(w.transceiverFuncs, func(transceiver FDTransceiver) {
			go joinErrChs(transceiver.SendFD(fd), out)
		})
	})
	return out
}

...

Use grpcfd.FromPeer in recvfd.Client

func (r *recvFDClient) Request(ctx context.Context, request *networkservice.NetworkServiceRequest, opts ...grpc.CallOption) (*networkservice.Connection, error) {
	p := new(peer.Peer)
	opts = append(opts, grpc.Peer(p))

	...

	conn, err := next.Client(ctx).Request(ctx, request, opts...)
	if err != nil {
		return nil, err
	}

	recv, ok := grpcfd.FromPeer(p)
	if !ok {
		return conn, nil
	}

	// <-- use recv

	// Return connection
	return conn, nil
}

func (r *recvFDClient) Close(ctx context.Context, conn *networkservice.Connection, opts ...grpc.CallOption) (*empty.Empty, error) {
	p := new(peer.Peer)
	opts = append(opts, grpc.Peer(p))

	...

	_, err := next.Client(ctx).Close(ctx, conn, opts...)
	if err != nil {
		return nil, err
	}

	recv, ok := grpcfd.FromPeer(p)
	if !ok {
		return &empty.Empty{}, nil
	}

	// <-- use recv

	return &empty.Empty{}, nil
}

[Bolodya1997]

@denis-tingaikin @glazychev-art @edwarnicke
cc