Fixed recvfd leaks

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

@@ -21,12 +21,10 @@ package recvfd_test
 import (
 	"context"
 	"fmt"
-	"net"
 	"net/url"
 	"os"
 	"path"
 	"runtime"
-	"syscall"
 	"testing"
 	"time"
 
@@ -36,7 +34,7 @@ import (
 	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/cls"
 	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/common"
 	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/kernel"
-	"github.com/stretchr/testify/assert"
+	"github.com/networkservicemesh/sdk/pkg/tools/grpcfdutils"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 	"google.golang.org/grpc"
@@ -58,13 +56,6 @@ type checkRecvfdServer struct {
 	t *testing.T
 }
 
-type notifiableFDTransceiver struct {
-	grpcfd.FDTransceiver
-	net.Addr
-
-	onRecvFile map[string]func()
-}
-
 func (n *checkRecvfdServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
 	return next.Server(ctx).Close(ctx, conn)
 }
@@ -76,52 +67,28 @@ func (n *checkRecvfdServer) Request(ctx context.Context, request *networkservice
 	transceiver, ok := p.Addr.(grpcfd.FDTransceiver)
 	require.True(n.t, ok)
 
-	p.Addr = &notifiableFDTransceiver{
+	p.Addr = &grpcfdutils.NotifiableFDTransceiver{
 		FDTransceiver: transceiver,
 		Addr:          p.Addr,
-		onRecvFile:    n.onRecvFile,
+		OnRecvFile:    n.onRecvFile,
 	}
 
 	return next.Server(ctx).Request(ctx, request)
 }
 
-func (w *notifiableFDTransceiver) RecvFileByURL(urlStr string) (<-chan *os.File, error) {
-	recv, err := w.FDTransceiver.RecvFileByURL(urlStr)
-	if err != nil {
-		return nil, err
-	}
-
-	var fileCh = make(chan *os.File)
-	go func() {
-		for f := range recv {
-			runtime.SetFinalizer(f, func(file *os.File) {
-				onFileClosedFunc, ok := w.onRecvFile[urlStr]
-				if ok {
-					onFileClosedFunc()
-				}
-			})
-			fileCh <- f
-		}
-	}()
-
-	return fileCh, nil
-}
-
 func createFile(fileName string, t *testing.T) (inodeURLStr string, fileClosedContext context.Context, cancelFunc func()) {
 	f, err := os.Create(fileName)
 	require.NoError(t, err, "Failed to create and open a file: %v", err)
 
-	info, err := f.Stat()
-	assert.NoError(t, err)
-
-	stat, ok := info.Sys().(*syscall.Stat_t)
-	assert.True(t, ok)
-
 	err = f.Close()
 	require.NoError(t, err, "Failed to close file: %v", err)
 
 	fileClosedContext, cancelFunc = context.WithCancel(context.Background())
-	inodeURLStr = fmt.Sprintf("inode://%d/%d", stat.Dev, stat.Ino)
+
+	inodeURL, err := grpcfd.FilenameToURL(fileName)
+	require.NoError(t, err)
+
+	inodeURLStr = inodeURL.String()
 
 	return
 }
@@ -150,7 +117,7 @@ func createClient(ctx context.Context, u *url.URL) networkservice.NetworkService
 func TestRecvfdClosesSingleFile(t *testing.T) {
 	t.Cleanup(func() { goleak.VerifyNone(t) })
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
 	var dir = t.TempDir()
@@ -203,7 +170,7 @@ func TestRecvfdClosesSingleFile(t *testing.T) {
 func TestRecvfdClosesMultipleFiles(t *testing.T) {
 	t.Cleanup(func() { goleak.VerifyNone(t) })
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
 	var dir = t.TempDir()

pkg/registry/common/recvfd/server_test.go

@@ -20,9 +20,7 @@ package recvfd_test
 
 import (
 	"context"
-	"net"
 	"net/url"
-	"os"
 	"path"
 	"runtime"
 	"testing"
@@ -45,24 +43,18 @@ import (
 	registryserialize "github.com/networkservicemesh/sdk/pkg/registry/common/serialize"
 	registrychain "github.com/networkservicemesh/sdk/pkg/registry/core/chain"
 	"github.com/networkservicemesh/sdk/pkg/registry/core/next"
+	"github.com/networkservicemesh/sdk/pkg/tools/grpcfdutils"
 	"github.com/networkservicemesh/sdk/pkg/tools/grpcutils"
 	"github.com/networkservicemesh/sdk/pkg/tools/sandbox"
 	"github.com/networkservicemesh/sdk/pkg/tools/token"
 )
 
 type checkNseRecvfdServer struct {
-	onRecvFile func()
+	onRecvFile map[string]func()
 
 	t *testing.T
 }
 
-type notifiableFDTransceiver struct {
-	grpcfd.FDTransceiver
-	net.Addr
-
-	onRecvFile func()
-}
-
 func (n *checkNseRecvfdServer) Unregister(ctx context.Context, endpoint *registry.NetworkServiceEndpoint) (*empty.Empty, error) {
 	return next.NetworkServiceEndpointRegistryServer(ctx).Unregister(ctx, endpoint)
 }
@@ -74,10 +66,10 @@ func (n *checkNseRecvfdServer) Register(ctx context.Context, endpoint *registry.
 	transceiver, ok := p.Addr.(grpcfd.FDTransceiver)
 	require.True(n.t, ok)
 
-	p.Addr = &notifiableFDTransceiver{
+	p.Addr = &grpcfdutils.NotifiableFDTransceiver{
 		FDTransceiver: transceiver,
 		Addr:          p.Addr,
-		onRecvFile:    n.onRecvFile,
+		OnRecvFile:    n.onRecvFile,
 	}
 
 	return next.NetworkServiceEndpointRegistryServer(ctx).Register(ctx, endpoint)
@@ -87,23 +79,15 @@ func (n *checkNseRecvfdServer) Find(query *registry.NetworkServiceEndpointQuery,
 	return next.NetworkServiceEndpointRegistryServer(server.Context()).Find(query, server)
 }
 
-func (w *notifiableFDTransceiver) RecvFileByURL(urlStr string) (<-chan *os.File, error) {
-	recv, err := w.FDTransceiver.RecvFileByURL(urlStr)
-	if err != nil {
-		return nil, err
-	}
+func getFileInfo(fileName string, t *testing.T) (inodeURLStr string, fileClosedContext context.Context, cancelFunc func()) {
+	fileClosedContext, cancelFunc = context.WithCancel(context.Background())
+
+	inodeURL, err := grpcfd.FilenameToURL(fileName)
+	require.NoError(t, err)
+
+	inodeURLStr = inodeURL.String()
 
-	var fileCh = make(chan *os.File)
-	go func() {
-		for f := range recv {
-			runtime.SetFinalizer(f, func(file *os.File) {
-				w.onRecvFile()
-			})
-			fileCh <- f
-		}
-	}()
-
-	return fileCh, nil
+	return
 }
 
 func startServer(ctx context.Context, t *testing.T, testRegistry registryserver.Registry, serveURL *url.URL) {
@@ -116,35 +100,29 @@ func startServer(ctx context.Context, t *testing.T, testRegistry registryserver.
 }
 
 func TestNseRecvfdServerClosesFile(t *testing.T) {
-	var ctx, cancel = context.WithTimeout(context.Background(), time.Minute)
+	var ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	var dir = t.TempDir()
-
-	s, err := os.Create(path.Join(dir, "test.sock"))
-	require.NoError(t, err)
-
-	err = s.Close()
-	require.NoError(t, err)
-
-	fileClosedContext, cancelFunc := context.WithCancel(context.Background())
-
 	var nsRegistry = registrychain.NewNetworkServiceRegistryServer(
 		registryserialize.NewNetworkServiceRegistryServer(),
 		memory.NewNetworkServiceRegistryServer(),
 	)
 
+	var checkRecvfdServer = &checkNseRecvfdServer{
+		t:          t,
+		onRecvFile: make(map[string]func()),
+	}
+
 	var nseRegistry = registrychain.NewNetworkServiceEndpointRegistryServer(
 		registryserialize.NewNetworkServiceEndpointRegistryServer(),
-		&checkNseRecvfdServer{
-			t:          t,
-			onRecvFile: cancelFunc,
-		},
+		checkRecvfdServer,
 		registryrecvfd.NewNetworkServiceEndpointRegistryServer(),
 		memory.NewNetworkServiceEndpointRegistryServer(),
 	)
 
-	var regURL = &url.URL{Scheme: "unix", Path: s.Name()}
+	var dir = t.TempDir()
+	var testFileName = path.Join(dir, t.Name()+".sock")
+	var regURL = &url.URL{Scheme: "unix", Path: testFileName}
 
 	var dialOptions = []grpc.DialOption{
 		grpc.WithTransportCredentials(
@@ -169,15 +147,19 @@ func TestNseRecvfdServerClosesFile(t *testing.T) {
 		))
 
 	startServer(ctx, t, registryserver.NewServer(nsRegistry, nseRegistry), regURL)
-	require.NoError(t, err)
+
+	// setting onRecvFile after starting the server as we're re-creating a socket in grpcutils.ListenAndServe
+	// in registry case we're passing a socket
+	inodeURLStr, fileClosedContext, cancelFunc := getFileInfo(testFileName, t)
+	checkRecvfdServer.onRecvFile[inodeURLStr] = cancelFunc
 
 	var testEndpoint = &registry.NetworkServiceEndpoint{
 		Name:                "test-endpoint",
 		NetworkServiceNames: []string{"test"},
 		Url:                 regURL.String(),
 	}
 
-	_, err = nseClient.Register(ctx, testEndpoint.Clone())
+	_, err := nseClient.Register(ctx, testEndpoint.Clone())
 	require.NoError(t, err)
 
 	_, err = nseClient.Unregister(ctx, testEndpoint.Clone())

pkg/tools/grpcfdutils/transceiver.go

@@ -0,0 +1,54 @@
+// Copyright (c) 2021 Doc.ai and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package grpcfdutils
+
+import (
+	"net"
+	"os"
+	"runtime"
+
+	"github.com/edwarnicke/grpcfd"
+)
+
+type NotifiableFDTransceiver struct {
+	grpcfd.FDTransceiver
+	net.Addr
+
+	OnRecvFile map[string]func()
+}
+
+func (w *NotifiableFDTransceiver) RecvFileByURL(urlStr string) (<-chan *os.File, error) {
+	recv, err := w.FDTransceiver.RecvFileByURL(urlStr)
+	if err != nil {
+		return nil, err
+	}
+
+	var fileCh = make(chan *os.File)
+	go func() {
+		for f := range recv {
+			runtime.SetFinalizer(f, func(file *os.File) {
+				onFileClosedFunc, ok := w.OnRecvFile[urlStr]
+				if ok {
+					onFileClosedFunc()
+				}
+			})
+			fileCh <- f
+		}
+	}()
+
+	return fileCh, nil
+}