Set memif_rx_mode on a separate chain element

pkg/networkservice/chains/forwarder/server.go

@@ -149,7 +149,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, vppConn
 						mtu.NewClient(vppConn),
 						tag.NewClient(ctx, vppConn),
 						// mechanisms
-						memif.NewClient(vppConn,
+						memif.NewClient(ctx, vppConn,
 							memif.WithChangeNetNS(),
 						),
 						kernel.NewClient(vppConn),

pkg/networkservice/mechanisms/memif/client.go

@@ -38,6 +38,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/tools/postpone"
 
 	"github.com/networkservicemesh/sdk-vpp/pkg/networkservice/mechanisms/memif/memifproxy"
+	"github.com/networkservicemesh/sdk-vpp/pkg/networkservice/mechanisms/memif/memifrxmode"
 )
 
 type memifClient struct {
@@ -47,13 +48,14 @@ type memifClient struct {
 }
 
 // NewClient provides a NetworkServiceClient chain elements that support the memif Mechanism
-func NewClient(vppConn api.Connection, options ...Option) networkservice.NetworkServiceClient {
+func NewClient(chainCtx context.Context, vppConn Connection, options ...Option) networkservice.NetworkServiceClient {
 	opts := &memifOptions{}
 	for _, o := range options {
 		o(opts)
 	}
 
 	return chain.NewNetworkServiceClient(
+		memifrxmode.NewClient(chainCtx, vppConn),
 		&memifClient{
 			vppConn:     vppConn,
 			changeNetNS: opts.changeNetNS,

pkg/networkservice/mechanisms/memif/client_test.go

@@ -34,7 +34,7 @@ import (
 )
 
 func Test_MemifClient_ShouldAppendMechanismIfMemifMechanismMissed(t *testing.T) {
-	c := chain.NewNetworkServiceClient(metadata.NewClient(), memif.NewClient(nil))
+	c := chain.NewNetworkServiceClient(metadata.NewClient(), memif.NewClient(context.Background(), nil))
 
 	req := &networkservice.NetworkServiceRequest{
 		MechanismPreferences: []*networkservice.Mechanism{},
@@ -56,7 +56,7 @@ func Test_MemifClient_ShouldAppendMechanismIfMemifMechanismMissed(t *testing.T)
 }
 
 func Test_MemifClient_ShouldNotDuplicateMechanisms(t *testing.T) {
-	c := chain.NewNetworkServiceClient(metadata.NewClient(), memif.NewClient(nil))
+	c := chain.NewNetworkServiceClient(metadata.NewClient(), memif.NewClient(context.Background(), nil))
 
 	req := &networkservice.NetworkServiceRequest{
 		MechanismPreferences: []*networkservice.Mechanism{

pkg/networkservice/mechanisms/memif/common.go

@@ -29,8 +29,6 @@ import (
 	"time"
 
 	"git.fd.io/govpp.git/api"
-	interfaces "github.com/edwarnicke/govpp/binapi/interface"
-	"github.com/edwarnicke/govpp/binapi/interface_types"
 	"github.com/edwarnicke/govpp/binapi/memif"
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	memifMech "github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/memif"
@@ -44,6 +42,12 @@ import (
 	"github.com/networkservicemesh/sdk-vpp/pkg/tools/ifindex"
 )
 
+// Connection aggregates the api.Connection and api.ChannelProvider interfaces
+type Connection interface {
+	api.Connection
+	api.ChannelProvider
+}
+
 // NetNSInfo contains shared info for server and client
 type NetNSInfo struct {
 	netNS     netns.NsHandle
@@ -151,19 +155,6 @@ func createMemif(ctx context.Context, vppConn api.Connection, socketID uint32, m
 		WithField("vppapi", "MemifCreate").Debug("completed")
 	ifindex.Store(ctx, isClient, rsp.SwIfIndex)
 
-	now = time.Now()
-	if _, err := interfaces.NewServiceClient(vppConn).SwInterfaceSetRxMode(ctx, &interfaces.SwInterfaceSetRxMode{
-		SwIfIndex: rsp.SwIfIndex,
-		Mode:      interface_types.RX_MODE_API_ADAPTIVE,
-	}); err != nil {
-		return errors.WithStack(err)
-	}
-	log.FromContext(ctx).
-		WithField("swIfIndex", rsp.SwIfIndex).
-		WithField("mode", interface_types.RX_MODE_API_ADAPTIVE).
-		WithField("duration", time.Since(now)).
-		WithField("vppapi", "SwInterfaceSetRxMode").Debug("completed")
-
 	if isClient {
 		up.Store(ctx, isClient, true)
 	}

pkg/networkservice/mechanisms/memif/memifrxmode/client.go

@@ -0,0 +1,88 @@
+// Copyright (c) 2022 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.
+
+//go:build linux
+
+package memifrxmode
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+
+	"github.com/golang/protobuf/ptypes/empty"
+
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+
+	memifMech "github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/memif"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
+	"github.com/networkservicemesh/sdk/pkg/tools/postpone"
+
+	"github.com/networkservicemesh/sdk-vpp/pkg/tools/ifindex"
+)
+
+type memifrxmodeClient struct {
+	chainCtx context.Context
+	vppConn  Connection
+}
+
+// NewClient provides a NetworkServiceClient chain elements that support the memif Mechanism
+func NewClient(chainCtx context.Context, vppConn Connection) networkservice.NetworkServiceClient {
+	return &memifrxmodeClient{
+		chainCtx: chainCtx,
+		vppConn:  vppConn,
+	}
+}
+
+func (m *memifrxmodeClient) Request(ctx context.Context, request *networkservice.NetworkServiceRequest, opts ...grpc.CallOption) (*networkservice.Connection, error) {
+	postponeCtxFunc := postpone.ContextWithValues(ctx)
+
+	conn, err := next.Client(ctx).Request(ctx, request, opts...)
+	if err != nil {
+		return nil, err
+	}
+	if mechanism := memifMech.ToMechanism(conn.GetMechanism()); mechanism == nil {
+		return conn, err
+	}
+
+	if ok := load(ctx, metadata.IsClient(m)); !ok {
+		swIfIndex, _ := ifindex.Load(ctx, metadata.IsClient(m))
+
+		cancelCtx, cancel := context.WithCancel(m.chainCtx)
+		store(ctx, metadata.IsClient(m), cancel)
+
+		if err := setRxMode(cancelCtx, m.vppConn, swIfIndex); err != nil {
+			closeCtx, cancelClose := postponeCtxFunc()
+			defer cancelClose()
+
+			if _, closeErr := m.Close(closeCtx, conn, opts...); closeErr != nil {
+				err = errors.Wrapf(err, "connection closed with error: %s", closeErr.Error())
+			}
+
+			return nil, err
+		}
+	}
+	return conn, nil
+}
+
+func (m *memifrxmodeClient) Close(ctx context.Context, conn *networkservice.Connection, opts ...grpc.CallOption) (*empty.Empty, error) {
+	if oldCancel, loaded := loadAndDelete(ctx, metadata.IsClient(m)); loaded {
+		oldCancel()
+	}
+	return next.Client(ctx).Close(ctx, conn, opts...)
+}

pkg/networkservice/mechanisms/memif/memifrxmode/common.go

@@ -0,0 +1,86 @@
+// Copyright (c) 2022 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.
+
+//go:build linux
+
+package memifrxmode
+
+import (
+	"context"
+	"time"
+
+	"git.fd.io/govpp.git/api"
+
+	interfaces "github.com/edwarnicke/govpp/binapi/interface"
+	"github.com/edwarnicke/govpp/binapi/interface_types"
+
+	"github.com/networkservicemesh/sdk/pkg/tools/log"
+)
+
+// Connection aggregates the api.Connection and api.ChannelProvider interfaces
+type Connection interface {
+	api.Connection
+	api.ChannelProvider
+}
+
+func setRxMode(ctx context.Context, vppConn Connection, swIfIndex interface_types.InterfaceIndex) error {
+	apiChannel, err := vppConn.NewAPIChannelBuffered(256, 256)
+	if err != nil {
+		return err
+	}
+
+	notifCh := make(chan api.Message, 256)
+	subscription, err := apiChannel.SubscribeNotification(notifCh, &interfaces.SwInterfaceEvent{})
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		defer apiChannel.Close()
+		defer func() { _ = subscription.Unsubscribe() }()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case rawMsg := <-notifCh:
+				if msg, ok := rawMsg.(*interfaces.SwInterfaceEvent); ok &&
+					msg.SwIfIndex == swIfIndex &&
+					msg.Flags&interface_types.IF_STATUS_API_FLAG_LINK_UP != 0 {
+					now := time.Now()
+					_, err = interfaces.NewServiceClient(vppConn).SwInterfaceSetRxMode(ctx, &interfaces.SwInterfaceSetRxMode{
+						SwIfIndex: swIfIndex,
+						Mode:      interface_types.RX_MODE_API_ADAPTIVE,
+					})
+					if err != nil {
+						log.FromContext(ctx).
+							WithField("swIfIndex", swIfIndex).
+							WithField("mode", interface_types.RX_MODE_API_ADAPTIVE).
+							WithField("duration", time.Since(now)).
+							WithField("vppapi", "SwInterfaceSetRxMode").Debugf("error: %v", err.Error())
+						return
+					}
+					log.FromContext(ctx).
+						WithField("swIfIndex", swIfIndex).
+						WithField("mode", interface_types.RX_MODE_API_ADAPTIVE).
+						WithField("duration", time.Since(now)).
+						WithField("vppapi", "SwInterfaceSetRxMode").Debug("completed")
+					return
+				}
+			}
+		}
+	}()
+	return nil
+}

pkg/networkservice/mechanisms/memif/memifrxmode/doc.go

@@ -0,0 +1,18 @@
+// Copyright (c) 2022 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 memifrxmode provides a NetworkService chain elements to set ADAPTIVE rx mode for memif interfaces
+package memifrxmode

pkg/networkservice/mechanisms/memif/memifrxmode/metadata.go

@@ -0,0 +1,54 @@
+// Copyright (c) 2020-2022 Cisco and/or its affiliates.
+//
+// Copyright (c) 2021-2022 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.
+
+//go:build linux
+
+package memifrxmode
+
+import (
+	"context"
+
+	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
+)
+
+type key struct{}
+
+// store sets the context.CancelFunc stored in per Connection.Id metadata.
+func store(ctx context.Context, isClient bool, cancel context.CancelFunc) {
+	metadata.Map(ctx, isClient).Store(key{}, cancel)
+}
+
+// loadAndDelete deletes the context.CancelFunc stored in per Connection.Id metadata.
+func loadAndDelete(ctx context.Context, isClient bool) (value context.CancelFunc, ok bool) {
+	rawValue, ok := metadata.Map(ctx, isClient).LoadAndDelete(key{})
+	if !ok {
+		return
+	}
+	value, ok = rawValue.(context.CancelFunc)
+	return value, ok
+}
+
+// load returns the context.CancelFunc stored in per Connection.Id metadata.
+func load(ctx context.Context, isClient bool) (ok bool) {
+	rawValue, ok := metadata.Map(ctx, isClient).Load(key{})
+	if !ok {
+		return
+	}
+	_, ok = rawValue.(context.CancelFunc)
+	return ok
+}