[meshnet] Use gRPC streaming

third_party/meshnet/daemon/grpcwire/grpcwire.go

@@ -328,7 +328,17 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error {
 		return err
 	}
 
-	remote, err := grpc.Dial(url, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	dialOpts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithInitialWindowSize(4 * 1024 * 1024),     // 4MB stream window
+		grpc.WithInitialConnWindowSize(16 * 1024 * 1024), // 16MB connection window
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallRecvMsgSize(64*1024*1024),
+			grpc.MaxCallSendMsgSize(64*1024*1024),
+		),
+	}
+
+	remote, err := grpc.Dial(url, dialOpts...)
 	if err != nil {
 		grpcOvrlyLogger.Infof("RecvFrmLocalPodThread:Failed to connect to remote %s/%d", url, wire.LocalNodeIfaceID)
 		return err
@@ -343,6 +353,19 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error {
 }
 
 func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WireProtocolClient, wire *GRPCWire, locIfNm string) error {
+	var stream mpb.WireProtocol_SendToStreamClient
+	getStream := func() (mpb.WireProtocol_SendToStreamClient, error) {
+		if stream != nil {
+			return stream, nil
+		}
+		st, err := wireClient.SendToStream(ctx)
+		if err != nil {
+			return nil, err
+		}
+		stream = st
+		return stream, nil
+	}
+
 	type readResult struct {
 		buf *[]byte
 		n   int
@@ -365,6 +388,9 @@ func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WirePr
 		case <-wire.StopC:
 			grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: closing connection with remote peer-iface@peer-node-ip: %d@%s/%d from %s@%s",
 				wire.WireIfaceIDOnPeerNode, wire.PeerNodeIP, wire.LocalNodeIfaceID, wire.LocalPodName, wire.LocalPodIfaceName)
+			if stream != nil {
+				_, _ = stream.CloseAndRecv()
+			}
 			return io.EOF
 		case res := <-readChan:
 			bufPtr := res.buf
@@ -411,12 +437,18 @@ func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WirePr
 				grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: unusually large packet received from local pod (may be GRO enabled). size: %d, pkt:%s", n, pktType)
 			}
 
-			ok, err := wireClient.SendToOnce(ctx, payload)
-			packetPool.Put(bufPtr)
-			if err != nil || !ok.Response {
-				grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not deliver pkt %s@%s@%s. Peer not ready, remote iface id %d. err=%v",
-					wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, peerIntfID, err)
+			st, err := getStream()
+			if err != nil {
+				packetPool.Put(bufPtr)
+				grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not get stream for %s@%s: %v", wire.LocalPodName, wire.LocalNodeIfaceName, err)
+				continue
 			}
+
+			if err := st.Send(payload); err != nil {
+				grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not send packet over stream %s@%s: %v", wire.LocalPodName, wire.LocalNodeIfaceName, err)
+				stream = nil // reset stream for reconnect on next packet
+			}
+			packetPool.Put(bufPtr)
 		}
 	}
 }

third_party/meshnet/daemon/grpcwire/grpcwire_test.go

@@ -33,8 +33,29 @@ func (m *mockWireProtocolClient) SendToOnce(ctx context.Context, in *mpb.Packet,
 	return &mpb.BoolResponse{Response: true}, nil
 }
 
+type mockStreamClient struct {
+	grpc.ClientStream
+	mockClient *mockWireProtocolClient
+}
+
+func (s *mockStreamClient) Send(in *mpb.Packet) error {
+	if s.mockClient.sendDelay > 0 {
+		time.Sleep(s.mockClient.sendDelay)
+	}
+	s.mockClient.mu.Lock()
+	defer s.mockClient.mu.Unlock()
+	frameCopy := make([]byte, len(in.Frame))
+	copy(frameCopy, in.Frame)
+	s.mockClient.receivedFrames = append(s.mockClient.receivedFrames, frameCopy)
+	return nil
+}
+
+func (s *mockStreamClient) CloseAndRecv() (*mpb.BoolResponse, error) {
+	return &mpb.BoolResponse{Response: true}, nil
+}
+
 func (m *mockWireProtocolClient) SendToStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[mpb.Packet, mpb.BoolResponse], error) {
-	return nil, errors.New("unimplemented")
+	return &mockStreamClient{mockClient: m}, nil
 }
 
 func TestForwardPackets_NoCorruption(t *testing.T) {

third_party/meshnet/daemon/meshnet/handler.go

@@ -3,6 +3,7 @@ package meshnet
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 
 	"github.com/openconfig/kne/third_party/meshnet/api/types/v1beta1"
@@ -414,6 +415,39 @@ func (m *Meshnet) SendToOnce(ctx context.Context, pkt *mpb.Packet) (*mpb.BoolRes
 	return &mpb.BoolResponse{Response: true}, nil
 }
 
+// ------------------------------------------------------------------------------------------------------
+func (m *Meshnet) SendToStream(stream mpb.WireProtocol_SendToStreamServer) error {
+	for {
+		pkt, err := stream.Recv()
+		if err == io.EOF {
+			return stream.SendAndClose(&mpb.BoolResponse{Response: true})
+		}
+		if err != nil {
+			return err
+		}
+
+		if pkt.RemotIntfId <= 0 {
+			continue
+		}
+
+		wrHandle, err := grpcwire.GetHostIntfHndl(pkt.RemotIntfId)
+		if err != nil {
+			log.WithFields(log.Fields{
+				"daemon":  "meshnetd",
+				"overlay": "gRPC",
+			}).Debugf("SendToStream (wire id - %v): Could not find local handle. err:%v", pkt.RemotIntfId, err)
+			continue
+		}
+
+		if _, err := wrHandle.Write(pkt.Frame); err != nil {
+			log.WithFields(log.Fields{
+				"daemon":  "meshnetd",
+				"overlay": "gRPC",
+			}).Errorf("SendToStream (wire id - %v): Could not write packet(%d bytes) to local interface. err:%v", pkt.RemotIntfId, len(pkt.Frame), err)
+		}
+	}
+}
+
 // ---------------------------------------------------------------------------------------------------------------
 func (m *Meshnet) AddGRPCWireRemote(ctx context.Context, wireDef *mpb.WireDef) (*mpb.WireCreateResponse, error) {
 	stopC := make(chan struct{})

third_party/meshnet/daemon/meshnet/meshnet.go

@@ -103,12 +103,20 @@ func New(cfg Config) (*Meshnet, error) {
 	// If the link type is GRPC then set the GRPC logging level to LevelNone
 	// Otherwise there will be GRPC log for every packet sent as for link type GRPC, GRPC is also the data-plane. This is too
 	// much of log that does not help in debugging and K8S does log rotation very frequently.
+	defaultOpts := []grpc.ServerOption{
+		grpc.InitialWindowSize(4 * 1024 * 1024),     // 4MB stream window
+		grpc.InitialConnWindowSize(16 * 1024 * 1024), // 16MB connection window
+		grpc.MaxRecvMsgSize(64 * 1024 * 1024),
+		grpc.MaxSendMsgSize(64 * 1024 * 1024),
+	}
+	allOpts := append(defaultOpts, cfg.GRPCOpts...)
+
 	var svr *grpc.Server
 	lnkTyp := os.Getenv("INTER_NODE_LINK_TYPE")
 	if lnkTyp == wireutil.INTER_NODE_LINK_GRPC {
-		svr = grpc.NewServer(cfg.GRPCOpts...)
+		svr = grpc.NewServer(allOpts...)
 	} else {
-		svr = newServerWithLogging(cfg.GRPCOpts...)
+		svr = newServerWithLogging(allOpts...)
 	}
 
 	m := &Meshnet{