Fixed recvfd leaks

pkg/networkservice/chains/nsmgr/unix_test.go

@@ -14,7 +14,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//+build !windows
+// +build !windows
 
 package nsmgr_test
 
@@ -38,6 +38,10 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/tools/sandbox"
 )
 
+const (
+	linuxOsName = "linux"
+)
+
 func Test_Local_NoURLUsecase(t *testing.T) {
 	t.Cleanup(func() { goleak.VerifyNone(t) })
 
@@ -84,7 +88,7 @@ func Test_Local_NoURLUsecase(t *testing.T) {
 }
 
 func Test_MultiForwarderSendfd(t *testing.T) {
-	if runtime.GOOS != "linux" {
+	if runtime.GOOS != linuxOsName {
 		t.Skip("sendfd works only on linux")
 	}
 	t.Cleanup(func() { goleak.VerifyNone(t) })
@@ -160,3 +164,49 @@ func Test_MultiForwarderSendfd(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 1, counter.Closes())
 }
+
+func Test_TimeoutRecvfd(t *testing.T) {
+	if runtime.GOOS != linuxOsName {
+		t.Skip("recvfd works only on linux")
+	}
+	t.Cleanup(func() { goleak.VerifyNone(t) })
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+
+	domain := sandbox.NewBuilder(ctx, t).
+		UseUnixSockets().
+		SetNodeSetup(func(ctx context.Context, node *sandbox.Node, _ int) {
+			node.NewNSMgr(ctx, "nsmgr", nil, sandbox.GenerateTestToken, nsmgr.NewServer)
+			node.NewForwarder(ctx, &registry.NetworkServiceEndpoint{
+				Name:                "forwarder-1",
+				NetworkServiceNames: []string{"forwarder"},
+				NetworkServiceLabels: map[string]*registry.NetworkServiceLabels{
+					"forwarder": {
+						Labels: map[string]string{
+							"p2p": "true",
+						},
+					},
+				},
+			}, sandbox.GenerateTestToken, recvfd.NewServer())
+		}).
+		Build()
+
+	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
+
+	nsReg, err := nsRegistryClient.Register(ctx, defaultRegistryService())
+	require.NoError(t, err)
+
+	nseReg := defaultRegistryEndpoint(nsReg.Name)
+
+	domain.Nodes[0].NewEndpoint(ctx, nseReg, sandbox.GenerateTestToken)
+
+	nsc := domain.Nodes[0].NewClient(ctx, sandbox.GenerateExpiringToken(0), kernel.NewClient(), sendfd.NewClient())
+
+	request := defaultRequest(nsReg.Name)
+
+	conn, err := nsc.Request(ctx, request.Clone())
+	require.Nil(t, err)
+	require.NotNil(t, conn)
+	require.Equal(t, 4, len(conn.Path.PathSegments))
+}

pkg/networkservice/common/mechanisms/recvfd/server.go

@@ -78,18 +78,21 @@ func (r *recvFDServer) Request(ctx context.Context, request *networkservice.Netw
 }
 
 func (r *recvFDServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	// Clean up the fileMap no matter what happens
+	defer r.fileMaps.Delete(conn.GetId())
+
 	// Get the grpcfd.FDRecver
 	recv, ok := grpcfd.FromContext(ctx)
 	if !ok {
+		r.closeFiles(conn)
 		return next.Server(ctx).Close(ctx, conn)
 	}
+
 	// Get the fileMap
 	fileMap, _ := r.fileMaps.LoadOrStore(conn.GetId(), &perConnectionFileMap{
 		filesByInodeURL:    make(map[string]*os.File),
 		inodeURLbyFilename: make(map[string]*url.URL),
 	})
-	// Clean up the fileMap no matter what happens
-	defer r.fileMaps.Delete(conn.GetId())
 
 	// Recv the FD and Swap the Inode for a file in InodeURL in Parameters
 	err := recvFDAndSwapInodeToFile(ctx, fileMap, conn.GetMechanism().GetParameters(), recv)
@@ -107,3 +110,15 @@ func (r *recvFDServer) Close(ctx context.Context, conn *networkservice.Connectio
 	err = swapFileToInode(fileMap, conn.GetMechanism().GetParameters(), true)
 	return &empty.Empty{}, err
 }
+
+func (r *recvFDServer) closeFiles(conn *networkservice.Connection) {
+	fileMap, _ := r.fileMaps.LoadOrStore(conn.GetId(), &perConnectionFileMap{
+		filesByInodeURL:    make(map[string]*os.File),
+		inodeURLbyFilename: make(map[string]*url.URL),
+	})
+
+	for inodeURLStr, file := range fileMap.filesByInodeURL {
+		delete(fileMap.filesByInodeURL, inodeURLStr)
+		_ = file.Close()
+	}
+}

pkg/registry/common/recvfd/server.go

@@ -98,18 +98,21 @@ func (r *recvfdNseServer) Unregister(ctx context.Context, endpoint *registry.Net
 	if endpoint.GetName() == "" {
 		return nil, errors.New("invalid endpoint specified")
 	}
+
+	// Clean up the fileMap no matter what happens
+	defer r.fileMaps.Delete(endpoint.GetName())
+
 	// Get the grpcfd.FDRecver
 	recv, ok := grpcfd.FromContext(ctx)
 	if !ok {
+		r.closeFiles(endpoint)
 		return next.NetworkServiceEndpointRegistryServer(ctx).Unregister(ctx, endpoint)
 	}
 	// Get the fileMap
 	fileMap, _ := r.fileMaps.LoadOrStore(endpoint.GetName(), &perEndpointFileMap{
 		filesByInodeURL:    make(map[string]*os.File),
 		inodeURLbyFilename: make(map[string]*url.URL),
 	})
-	// Clean up the fileMap no matter what happens
-	defer r.fileMaps.Delete(endpoint.GetName())
 
 	// Recv the FD and Swap the Inode for a file in InodeURL in Parameters
 	endpoint = endpoint.Clone()
@@ -210,3 +213,15 @@ func swapFileToInode(fileMap *perEndpointFileMap, endpoint *registry.NetworkServ
 	})
 	return nil
 }
+
+func (r *recvfdNseServer) closeFiles(endpoint *registry.NetworkServiceEndpoint) {
+	fileMap, _ := r.fileMaps.LoadOrStore(endpoint.GetName(), &perEndpointFileMap{
+		filesByInodeURL:    make(map[string]*os.File),
+		inodeURLbyFilename: make(map[string]*url.URL),
+	})
+
+	for inodeURLStr, file := range fileMap.filesByInodeURL {
+		delete(fileMap.filesByInodeURL, inodeURLStr)
+		_ = file.Close()
+	}
+}