Allow begin to process Unregisters without Registers (#1542)

* Allow begin process Unregisters without Registers Signed-off-by: Nikita Skrynnik <[email protected]> * delete one test + fix go leaks Signed-off-by: Nikita Skrynnik <[email protected]> * fix test using mutexes Signed-off-by: Nikita Skrynnik <[email protected]> * fix more unit tests Signed-off-by: Nikita Skrynnik <[email protected]> * fix some tests + fix linter issues Signed-off-by: Nikita Skrynnik <[email protected]> * rerun CI Signed-off-by: Nikita Skrynnik <[email protected]> * test StressTest for begin server Signed-off-by: Nikita Skrynnik <[email protected]> * delete unneseccary test Signed-off-by: Nikita Skrynnik <[email protected]> * fix dial chain element to fix test Signed-off-by: Nikita Skrynnik <[email protected]> * fix linter issues Signed-off-by: Nikita Skrynnik <[email protected]> * cleanup Signed-off-by: Nikita Skrynnik <[email protected]> * minor fix Signed-off-by: Nikita Skrynnik <[email protected]> * rerun CI Signed-off-by: Nikita Skrynnik <[email protected]> * test again Signed-off-by: Nikita Skrynnik <[email protected]> * test again Signed-off-by: Nikita Skrynnik <[email protected]> * disable one test Signed-off-by: Nikita Skrynnik <[email protected]> * add some logs Signed-off-by: Nikita Skrynnik <[email protected]> * add more logs Signed-off-by: Nikita Skrynnik <[email protected]> * reduce count of events to 10 Signed-off-by: Nikita Skrynnik <[email protected]> * skip fifo test Signed-off-by: Nikita Skrynnik <[email protected]> * fix golang linter issues Signed-off-by: Nikita Skrynnik <[email protected]> * rerun CI Signed-off-by: Nikita Skrynnik <[email protected]> * cleanup Signed-off-by: Nikita Skrynnik <[email protected]> * apply some review comments Signed-off-by: Nikita Skrynnik <[email protected]> * cleanup Signed-off-by: Nikita Skrynnik <[email protected]> * fix review comments related to dial Signed-off-by: Nikita Skrynnik <[email protected]> * fix review comments Signed-off-by: Nikita Skrynnik <[email protected]> * cleanup Signed-off-by: Nikita Skrynnik <[email protected]> * rerun CI Signed-off-by: Nikita Skrynnik <[email protected]> * fix more review comments Signed-off-by: Nikita Skrynnik <[email protected]> --------- Signed-off-by: Nikita Skrynnik <[email protected]>
networkservicemesh · Nov 8, 2023 · 20150d6 · 20150d6
1 parent 85ca154
commit 20150d6
Show file tree

Hide file tree

Showing 7 changed files with 414 additions and 129 deletions.
diff --git a/pkg/registry/common/begin/benchmark_test.go b/pkg/registry/common/begin/benchmark_test.go
@@ -0,0 +1,186 @@
+// Copyright (c) 2023 Cisco 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 begin_test
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/networkservicemesh/api/pkg/api/registry"
+
+	"github.com/networkservicemesh/sdk/pkg/registry/common/begin"
+	"github.com/networkservicemesh/sdk/pkg/registry/core/chain"
+	"github.com/networkservicemesh/sdk/pkg/registry/core/next"
+)
+
+var (
+	count = 1000
+)
+
+type dataRaceServer struct {
+	count int
+}
+
+func (s *dataRaceServer) Register(ctx context.Context, in *registry.NetworkServiceEndpoint) (*registry.NetworkServiceEndpoint, error) {
+	s.count++
+	return next.NetworkServiceEndpointRegistryServer(ctx).Register(ctx, in)
+}
+
+func (s *dataRaceServer) Find(query *registry.NetworkServiceEndpointQuery, server registry.NetworkServiceEndpointRegistry_FindServer) error {
+	return next.NetworkServiceEndpointRegistryServer(server.Context()).Find(query, server)
+}
+
+func (s *dataRaceServer) Unregister(ctx context.Context, in *registry.NetworkServiceEndpoint) (*empty.Empty, error) {
+	s.count--
+	return next.NetworkServiceEndpointRegistryServer(ctx).Unregister(ctx, in)
+}
+
+func BenchmarkBegin_RegisterSameIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+		&dataRaceServer{count: 0},
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(count)
+	b.ResetTimer()
+	for i := 0; i < count; i++ {
+		go func() {
+			_, _ = server.Register(context.Background(), &registry.NetworkServiceEndpoint{Name: "1"})
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}
+
+func BenchmarkBegin_UnregisterSameIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+		&dataRaceServer{count: 0},
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(count)
+	b.ResetTimer()
+	for i := 0; i < count; i++ {
+		go func() {
+			_, _ = server.Unregister(context.Background(), &registry.NetworkServiceEndpoint{Name: "1"})
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}
+
+func BenchmarkBegin_RegisterUnregisterSameIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+		&dataRaceServer{count: 0},
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(2 * count)
+	b.ResetTimer()
+	go func() {
+		for i := 0; i < count; i++ {
+			go func() {
+				_, _ = server.Register(context.Background(), &registry.NetworkServiceEndpoint{Name: "1"})
+				wg.Done()
+			}()
+		}
+	}()
+
+	go func() {
+		for i := 0; i < count; i++ {
+			go func() {
+				_, _ = server.Unregister(context.Background(), &registry.NetworkServiceEndpoint{Name: "1"})
+				wg.Done()
+			}()
+		}
+	}()
+
+	wg.Wait()
+}
+
+func BenchmarkBegin_RegisterDifferentIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(count)
+	b.ResetTimer()
+	for i := 0; i < count; i++ {
+		local := i
+		go func() {
+			_, _ = server.Register(context.Background(), &registry.NetworkServiceEndpoint{Name: fmt.Sprint(local)})
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}
+
+func BenchmarkBegin_UnregisterDifferentIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(count)
+	b.ResetTimer()
+	for i := 0; i < count; i++ {
+		local := i
+		go func() {
+			_, _ = server.Unregister(context.Background(), &registry.NetworkServiceEndpoint{Name: fmt.Sprint(local)})
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}
+
+func BenchmarkBegin_RegisterUnregisterDifferentIDs(b *testing.B) {
+	server := chain.NewNetworkServiceEndpointRegistryServer(
+		begin.NewNetworkServiceEndpointRegistryServer(),
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(2 * count)
+	b.ResetTimer()
+	go func() {
+		for i := 0; i < count; i++ {
+			local := i
+			go func() {
+				_, _ = server.Register(context.Background(), &registry.NetworkServiceEndpoint{Name: fmt.Sprint(local)})
+				wg.Done()
+			}()
+		}
+	}()
+
+	go func() {
+		for i := 0; i < count; i++ {
+			local := i
+			go func() {
+				_, _ = server.Unregister(context.Background(), &registry.NetworkServiceEndpoint{Name: fmt.Sprint(local)})
+				wg.Done()
+			}()
+		}
+	}()
+
+	wg.Wait()
+}
diff --git a/pkg/registry/common/begin/close_client_test.go b/pkg/registry/common/begin/close_client_test.go
@@ -1,4 +1,4 @@
-// Copyright (c) 2022 Cisco and/or its affiliates.
+// Copyright (c) 2022-2023 Cisco and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -18,7 +18,6 @@ package begin_test
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	"github.com/golang/protobuf/ptypes/empty"
@@ -85,46 +84,3 @@ func (m *markClient) Unregister(ctx context.Context, in *registry.NetworkService
 	assert.Equal(m.t, mark, in.GetNetworkServiceLabels()[mark].Labels[mark])
 	return next.NetworkServiceEndpointRegistryClient(ctx).Unregister(ctx, in, opts...)
 }
-
-func TestDoubleCloseClient(t *testing.T) {
-	t.Cleanup(func() { goleak.VerifyNone(t) })
-	client := chain.NewNetworkServiceEndpointRegistryClient(
-		begin.NewNetworkServiceEndpointRegistryClient(),
-		&doubleCloseClient{t: t},
-	)
-	id := "1"
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	conn, err := client.Register(ctx, &registry.NetworkServiceEndpoint{
-		Name: id,
-	})
-	assert.NotNil(t, t, conn)
-	assert.NoError(t, err)
-	conn = conn.Clone()
-	_, err = client.Unregister(ctx, conn)
-	assert.NoError(t, err)
-	_, err = client.Unregister(ctx, conn)
-	assert.NoError(t, err)
-}
-
-type doubleCloseClient struct {
-	t *testing.T
-	sync.Once
-}
-
-func (s *doubleCloseClient) Register(ctx context.Context, in *registry.NetworkServiceEndpoint, opts ...grpc.CallOption) (*registry.NetworkServiceEndpoint, error) {
-	return next.NetworkServiceEndpointRegistryClient(ctx).Register(ctx, in, opts...)
-}
-
-func (s *doubleCloseClient) Find(ctx context.Context, in *registry.NetworkServiceEndpointQuery, opts ...grpc.CallOption) (registry.NetworkServiceEndpointRegistry_FindClient, error) {
-	return next.NetworkServiceEndpointRegistryClient(ctx).Find(ctx, in, opts...)
-}
-
-func (s *doubleCloseClient) Unregister(ctx context.Context, in *registry.NetworkServiceEndpoint, opts ...grpc.CallOption) (*empty.Empty, error) {
-	count := 1
-	s.Do(func() {
-		count++
-	})
-	assert.Equal(s.t, 2, count, "Close has been called more than once")
-	return next.NetworkServiceEndpointRegistryClient(ctx).Unregister(ctx, in, opts...)
-}
diff --git a/pkg/registry/common/begin/close_server_test.go b/pkg/registry/common/begin/close_server_test.go
@@ -1,4 +1,4 @@
-// Copyright (c) 2022 Cisco and/or its affiliates.
+// Copyright (c) 2022-2023 Cisco and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -18,20 +18,16 @@ package begin_test
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	"github.com/networkservicemesh/api/pkg/api/registry"
 
 	"github.com/networkservicemesh/sdk/pkg/registry/common/begin"
-	"github.com/networkservicemesh/sdk/pkg/registry/common/null"
 	"github.com/networkservicemesh/sdk/pkg/registry/core/adapters"
 	"github.com/networkservicemesh/sdk/pkg/registry/core/chain"
-	"github.com/networkservicemesh/sdk/pkg/registry/core/next"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/goleak"
-	"google.golang.org/protobuf/types/known/emptypb"
 )
 
 func TestCloseServer(t *testing.T) {
@@ -55,39 +51,3 @@ func TestCloseServer(t *testing.T) {
 	_, err = server.Unregister(ctx, conn)
 	assert.NoError(t, err)
 }
-
-func TestDoubleCloseServer(t *testing.T) {
-	t.Cleanup(func() { goleak.VerifyNone(t) })
-	server := chain.NewNetworkServiceEndpointRegistryServer(
-		begin.NewNetworkServiceEndpointRegistryServer(),
-		&doubleCloseServer{t: t, NetworkServiceEndpointRegistryServer: null.NewNetworkServiceEndpointRegistryServer()},
-	)
-	id := "1"
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	conn, err := server.Register(ctx, &registry.NetworkServiceEndpoint{
-		Name: id,
-	})
-	assert.NotNil(t, t, conn)
-	assert.NoError(t, err)
-	conn = conn.Clone()
-	_, err = server.Unregister(ctx, conn)
-	assert.NoError(t, err)
-	_, err = server.Unregister(ctx, conn)
-	assert.NoError(t, err)
-}
-
-type doubleCloseServer struct {
-	t *testing.T
-	sync.Once
-	registry.NetworkServiceEndpointRegistryServer
-}
-
-func (s *doubleCloseServer) Unregister(ctx context.Context, in *registry.NetworkServiceEndpoint) (*emptypb.Empty, error) {
-	count := 1
-	s.Do(func() {
-		count++
-	})
-	assert.Equal(s.t, 2, count, "Close has been called more than once")
-	return next.NetworkServiceEndpointRegistryServer(ctx).Unregister(ctx, in)
-}