fix data races in proxy server

Signed-off-by: Matus Mrekaj <[email protected]>
Change-Id: I932d560548ee816e28683243a7318a2a7fbbb24a

Author: matus-mrekaj

cmd/vpp-proxy/main.go

@@ -120,7 +120,10 @@ func runClient() {
 }
 
 func runServer() {
-	p := proxy.NewServer()
+	p, err := proxy.NewServer()
+	if err != nil {
+		log.Fatalln(err)
+	}
 
 	statsAdapter := statsclient.NewStatsClient(*statsSocket)
 	binapiAdapter := socketclient.NewVppClient(*binapiSocket)
@@ -135,5 +138,7 @@ func runServer() {
 	}
 	defer p.DisconnectBinapi()
 
-	p.ListenAndServe(*proxyAddr)
+	if err := p.ListenAndServe(*proxyAddr); err != nil {
+		log.Fatalln(err)
+	}
 }

core/connection.go

@@ -111,6 +111,9 @@ type Connection struct {
 
 	lastReplyLock sync.Mutex // lock for the last reply
 	lastReply     time.Time  // time of the last received reply from VPP
+
+	msgControlPing      api.Message
+	msgControlPingReply api.Message
 }
 
 func newConnection(binapi adapter.VppAPI, attempts int, interval time.Duration) *Connection {
@@ -122,14 +125,16 @@ func newConnection(binapi adapter.VppAPI, attempts int, interval time.Duration)
 	}
 
 	c := &Connection{
-		vppClient:     binapi,
-		maxAttempts:   attempts,
-		recInterval:   interval,
-		codec:         &codec.MsgCodec{},
-		msgIDs:        make(map[string]uint16),
-		msgMap:        make(map[uint16]api.Message),
-		channels:      make(map[uint16]*Channel),
-		subscriptions: make(map[uint16][]*subscriptionCtx),
+		vppClient:           binapi,
+		maxAttempts:         attempts,
+		recInterval:         interval,
+		codec:               &codec.MsgCodec{},
+		msgIDs:              make(map[string]uint16),
+		msgMap:              make(map[uint16]api.Message),
+		channels:            make(map[uint16]*Channel),
+		subscriptions:       make(map[uint16][]*subscriptionCtx),
+		msgControlPing:      msgControlPing,
+		msgControlPingReply: msgControlPingReply,
 	}
 	binapi.SetMsgCallback(c.msgCallback)
 	return c
@@ -314,7 +319,7 @@ func (c *Connection) healthCheckLoop(connChan chan ConnectionEvent) {
 		}
 
 		// send the control ping request
-		ch.reqChan <- &vppRequest{msg: msgControlPing}
+		ch.reqChan <- &vppRequest{msg: c.msgControlPing}
 
 		for {
 			// expect response within timeout period
@@ -427,12 +432,12 @@ func (c *Connection) retrieveMessageIDs() (err error) {
 		}
 		n++
 
-		if c.pingReqID == 0 && msg.GetMessageName() == msgControlPing.GetMessageName() {
+		if c.pingReqID == 0 && msg.GetMessageName() == c.msgControlPing.GetMessageName() {
 			c.pingReqID = msgID
-			msgControlPing = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
-		} else if c.pingReplyID == 0 && msg.GetMessageName() == msgControlPingReply.GetMessageName() {
+			c.msgControlPing = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
+		} else if c.pingReplyID == 0 && msg.GetMessageName() == c.msgControlPingReply.GetMessageName() {
 			c.pingReplyID = msgID
-			msgControlPingReply = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
+			c.msgControlPingReply = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
 		}
 
 		if debugMsgIDs {

core/control_ping.go

@@ -1,6 +1,8 @@
 package core
 
-import "git.fd.io/govpp.git/api"
+import (
+	"git.fd.io/govpp.git/api"
+)
 
 var (
 	msgControlPing      api.Message = new(ControlPing)

core/request_handler.go

@@ -110,7 +110,7 @@ func (c *Connection) processRequest(ch *Channel, req *vppRequest) error {
 
 	if req.multi {
 		// send a control ping to determine end of the multipart response
-		pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)
+		pingData, _ := c.codec.EncodeMsg(c.msgControlPing, c.pingReqID)
 
 		log.WithFields(logger.Fields{
 			"channel":  ch.id,

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/golang/protobuf v1.3.2 // indirect
 	github.com/hpcloud/tail v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0 // indirect
-	github.com/lunixbochs/struc v0.0.0-20180408203800-02e4c2afbb2a
+	github.com/lunixbochs/struc v0.0.0-20190916212049-a5c72983bc42
 	github.com/onsi/ginkgo v1.8.0 // indirect
 	github.com/onsi/gomega v1.1.0
 	github.com/pkg/profile v1.2.1

go.sum

@@ -17,6 +17,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lunixbochs/struc v0.0.0-20180408203800-02e4c2afbb2a h1:axFx97V2Lyke5LbeygrJlzc07mwVhHt2ZHeI/Nv8Aq4=
 github.com/lunixbochs/struc v0.0.0-20180408203800-02e4c2afbb2a/go.mod h1:iOJu9pApjjmEmNq7PqlA5R9mDu/HMF5EM3llWKX/TyA=
+github.com/lunixbochs/struc v0.0.0-20190916212049-a5c72983bc42 h1:PzBD7QuxXSgSu61TKXxRwVGzWO5d9QZ0HxFFpndZMCg=
+github.com/lunixbochs/struc v0.0.0-20190916212049-a5c72983bc42/go.mod h1:vy1vK6wD6j7xX6O6hXe621WabdtNkou2h7uRtTfRMyg=
 github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
 github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.1.0 h1:e3YP4dN/HYPpGh29X1ZkcxcEICsOls9huyVCRBaxjq8=

proxy/client.go

@@ -1,8 +1,22 @@
+//  Copyright (c) 2019 Cisco and/or its affiliates.
+//
+//  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 proxy
 
 import (
 	"fmt"
-	"log"
+	"git.fd.io/govpp.git/core"
 	"net/rpc"
 	"reflect"
 	"time"
@@ -40,7 +54,8 @@ func (c *Client) NewStatsClient() (*StatsClient, error) {
 // NewBinapiClient returns new BinapiClient which implements api.Channel.
 func (c *Client) NewBinapiClient() (*BinapiClient, error) {
 	binapi := &BinapiClient{
-		rpc: c.rpc,
+		rpc:     c.rpc,
+		timeout: core.DefaultReplyTimeout,
 	}
 	return binapi, nil
 }
@@ -103,27 +118,31 @@ func (s *StatsClient) GetBufferStats(bufStats *api.BufferStats) error {
 }
 
 type BinapiClient struct {
-	rpc *rpc.Client
+	rpc     *rpc.Client
+	timeout time.Duration
 }
 
 func (b *BinapiClient) SendRequest(msg api.Message) api.RequestCtx {
 	req := &requestCtx{
-		rpc: b.rpc,
-		req: msg,
+		rpc:     b.rpc,
+		timeout: b.timeout,
+		req:     msg,
 	}
-	log.Printf("SendRequest: %T %+v", msg, msg)
+	log.Debugf("SendRequest: %T %+v", msg, msg)
 	return req
 }
 
 type requestCtx struct {
-	rpc *rpc.Client
-	req api.Message
+	rpc     *rpc.Client
+	req     api.Message
+	timeout time.Duration
 }
 
 func (r *requestCtx) ReceiveReply(msg api.Message) error {
 	req := BinapiRequest{
 		Msg:      r.req,
 		ReplyMsg: msg,
+		Timeout:  r.timeout,
 	}
 	resp := BinapiResponse{}
 
@@ -140,16 +159,18 @@ func (r *requestCtx) ReceiveReply(msg api.Message) error {
 
 func (b *BinapiClient) SendMultiRequest(msg api.Message) api.MultiRequestCtx {
 	req := &multiRequestCtx{
-		rpc: b.rpc,
-		req: msg,
+		rpc:     b.rpc,
+		timeout: b.timeout,
+		req:     msg,
 	}
-	log.Printf("SendMultiRequest: %T %+v", msg, msg)
+	log.Debugf("SendMultiRequest: %T %+v", msg, msg)
 	return req
 }
 
 type multiRequestCtx struct {
-	rpc *rpc.Client
-	req api.Message
+	rpc     *rpc.Client
+	req     api.Message
+	timeout time.Duration
 
 	index   int
 	replies []api.Message
@@ -162,6 +183,7 @@ func (r *multiRequestCtx) ReceiveReply(msg api.Message) (stop bool, err error) {
 			Msg:      r.req,
 			ReplyMsg: msg,
 			IsMulti:  true,
+			Timeout:  r.timeout,
 		}
 		resp := BinapiResponse{}
 
@@ -189,24 +211,23 @@ func (b *BinapiClient) SubscribeNotification(notifChan chan api.Message, event a
 }
 
 func (b *BinapiClient) SetReplyTimeout(timeout time.Duration) {
-	req := BinapiTimeoutRequest{Timeout: timeout}
-	resp := BinapiTimeoutResponse{}
-	if err := b.rpc.Call("BinapiRPC.SetTimeout", req, &resp); err != nil {
-		log.Println(err)
-	}
+	b.timeout = timeout
 }
 
 func (b *BinapiClient) CheckCompatiblity(msgs ...api.Message) error {
+	msgNamesCrscs := make([]string, 0, len(msgs))
+
 	for _, msg := range msgs {
-		req := BinapiCompatibilityRequest{
-			MsgName: msg.GetMessageName(),
-			Crc:     msg.GetCrcString(),
-		}
-		resp := BinapiCompatibilityResponse{}
-		if err := b.rpc.Call("BinapiRPC.Compatibility", req, &resp); err != nil {
-			return err
-		}
+		msgNamesCrscs = append(msgNamesCrscs, msg.GetMessageName()+"_"+msg.GetCrcString())
 	}
+
+	req := BinapiCompatibilityRequest{MsgNameCrcs: msgNamesCrscs}
+	resp := BinapiCompatibilityResponse{}
+
+	if err := b.rpc.Call("BinapiRPC.Compatibility", req, &resp); err != nil {
+		return err
+	}
+
 	return nil
 }
 

proxy/log.go

@@ -0,0 +1,30 @@
+package proxy
+
+import (
+	"github.com/sirupsen/logrus"
+	"os"
+)
+
+var (
+	debug = os.Getenv("DEBUG_GOVPP_PROXY") != ""
+
+	log = logrus.New()
+)
+
+func init() {
+	log.Out = os.Stdout
+	if debug {
+		log.Level = logrus.DebugLevel
+		log.Debugf("govpp/proxy: debug mode enabled")
+	}
+}
+
+// SetLogger sets global logger to l.
+func SetLogger(l *logrus.Logger) {
+	log = l
+}
+
+// SetLogLevel sets global logger level to lvl.
+func SetLogLevel(lvl logrus.Level) {
+	log.Level = lvl
+}