X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=adapter%2Fsocketclient%2Fsocketclient.go;h=c9aa2b4b028b27e6b1c7cb3e9de53711d14a4eaa;hb=67cea0df30f7ef348e265c9326f7a9f15ba26240;hp=f281955825fffc7b64419db192c9f352515e2de7;hpb=6fe52d72255456e7d73df9d2f6b4a8f724ed447d;p=govpp.git diff --git a/adapter/socketclient/socketclient.go b/adapter/socketclient/socketclient.go index f281955..c9aa2b4 100644 --- a/adapter/socketclient/socketclient.go +++ b/adapter/socketclient/socketclient.go @@ -1,10 +1,24 @@ +// 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 socketclient import ( "bufio" - "bytes" + "encoding/binary" + "errors" "fmt" - "github.com/fsnotify/fsnotify" "io" "net" "os" @@ -13,115 +27,173 @@ import ( "sync" "time" - "github.com/lunixbochs/struc" - logger "github.com/sirupsen/logrus" + "github.com/fsnotify/fsnotify" + "github.com/sirupsen/logrus" - "git.fd.io/govpp.git/adapter" - "git.fd.io/govpp.git/codec" - "git.fd.io/govpp.git/examples/bin_api/memclnt" + "go.fd.io/govpp/adapter" + "go.fd.io/govpp/binapi/memclnt" + "go.fd.io/govpp/codec" ) const ( - // DefaultSocketName is default VPP API socket file name - DefaultSocketName = "/run/vpp-api.sock" - - sockCreateMsgId = 15 // hard-coded id for sockclnt_create message - govppClientName = "govppsock" // client name used for socket registration + // DefaultSocketName is default VPP API socket file path. + DefaultSocketName = "/run/vpp/api.sock" + // DefaultClientName is used for identifying client in socket registration + DefaultClientName = "govppsock" ) var ( - ConnectTimeout = time.Second * 3 - DisconnectTimeout = time.Second - Debug = os.Getenv("DEBUG_GOVPP_SOCK") != "" - DebugMsgIds = os.Getenv("DEBUG_GOVPP_SOCKMSG") != "" + // DefaultConnectTimeout is default timeout for connecting + DefaultConnectTimeout = time.Second * 3 + // DefaultDisconnectTimeout is default timeout for discconnecting + DefaultDisconnectTimeout = time.Millisecond * 100 + // MaxWaitReady defines maximum duration of waiting for socket file + MaxWaitReady = time.Second * 10 +) - Log = logger.New() // global logger +var ( + debug = strings.Contains(os.Getenv("DEBUG_GOVPP"), "socketclient") + debugMsgIds = strings.Contains(os.Getenv("DEBUG_GOVPP"), "msgtable") + + log logrus.FieldLogger ) -// init initializes global logger, which logs debug level messages to stdout. +// SetLogger sets global logger. +func SetLogger(logger logrus.FieldLogger) { + log = logger +} + func init() { - Log.Out = os.Stdout - if Debug { - Log.Level = logger.DebugLevel + logger := logrus.New() + if debug { + logger.Level = logrus.DebugLevel + logger.Debug("govpp: debug level enabled for socketclient") } + log = logger.WithField("logger", "govpp/socketclient") } -type vppClient struct { - sockAddr string - conn *net.UnixConn - reader *bufio.Reader - cb adapter.MsgCallback +type Client struct { + socketPath string + clientName string + + conn *net.UnixConn + reader *bufio.Reader + writer *bufio.Writer + + connectTimeout time.Duration + disconnectTimeout time.Duration + + msgCallback adapter.MsgCallback clientIndex uint32 msgTable map[string]uint16 sockDelMsgId uint16 writeMu sync.Mutex - quit chan struct{} - wg sync.WaitGroup + + headerPool *sync.Pool + + quit chan struct{} + wg sync.WaitGroup } -func NewVppClient(sockAddr string) *vppClient { - if sockAddr == "" { - sockAddr = DefaultSocketName - } - return &vppClient{ - sockAddr: sockAddr, - cb: nilCallback, +// NewVppClient returns a new Client using socket. +// If socket is empty string DefaultSocketName is used. +func NewVppClient(socket string) *Client { + if socket == "" { + socket = DefaultSocketName + } + return &Client{ + socketPath: socket, + clientName: DefaultClientName, + connectTimeout: DefaultConnectTimeout, + disconnectTimeout: DefaultDisconnectTimeout, + headerPool: &sync.Pool{New: func() interface{} { + return make([]byte, 16) + }}, + msgCallback: func(msgID uint16, data []byte) { + log.Debugf("no callback set, dropping message: ID=%v len=%d", msgID, len(data)) + }, } } -func nilCallback(msgID uint16, data []byte) { - Log.Warnf("no callback set, dropping message: ID=%v len=%d", msgID, len(data)) +// SetClientName sets a client name used for identification. +func (c *Client) SetClientName(name string) { + c.clientName = name +} + +// SetConnectTimeout sets timeout used during connecting. +func (c *Client) SetConnectTimeout(t time.Duration) { + c.connectTimeout = t +} + +// SetDisconnectTimeout sets timeout used during disconnecting. +func (c *Client) SetDisconnectTimeout(t time.Duration) { + c.disconnectTimeout = t +} + +func (c *Client) SetMsgCallback(cb adapter.MsgCallback) { + log.Debug("SetMsgCallback") + c.msgCallback = cb } // WaitReady checks socket file existence and waits for it if necessary -func (c *vppClient) WaitReady() error { - // verify file existence - if _, err := os.Stat(c.sockAddr); err == nil { - return nil - } else if os.IsExist(err) { - return err +func (c *Client) WaitReady() error { + // check if socket already exists + if _, err := os.Stat(c.socketPath); err == nil { + return nil // socket exists, we are ready + } else if !os.IsNotExist(err) { + return err // some other error occurred } - // if not, watch for it + // socket does not exist, watch for it watcher, err := fsnotify.NewWatcher() if err != nil { return err } defer func() { if err := watcher.Close(); err != nil { - Log.Errorf("failed to close file watcher: %v", err) + log.Debugf("failed to close file watcher: %v", err) } }() - path := filepath.Dir(c.sockAddr) - if err := watcher.Add(path); err != nil { + + // start directory watcher + if err := watcher.Add(filepath.Dir(c.socketPath)); err != nil { return err } + timeout := time.NewTimer(MaxWaitReady) for { - ev := <-watcher.Events - if ev.Name == path { - if (ev.Op & fsnotify.Create) == fsnotify.Create { - // socket ready + select { + case <-timeout.C: + return fmt.Errorf("timeout waiting (%s) for socket file: %s", MaxWaitReady, c.socketPath) + + case e := <-watcher.Errors: + return e + + case ev := <-watcher.Events: + log.Debugf("watcher event: %+v", ev) + if ev.Name == c.socketPath && (ev.Op&fsnotify.Create) == fsnotify.Create { + // socket created, we are ready return nil } } } } -func (c *vppClient) SetMsgCallback(cb adapter.MsgCallback) { - Log.Debug("SetMsgCallback") - c.cb = cb -} - -func (c *vppClient) Connect() error { - Log.Debugf("Connecting to: %v", c.sockAddr) +func (c *Client) Connect() error { + // check if socket exists + if _, err := os.Stat(c.socketPath); os.IsNotExist(err) { + return fmt.Errorf("VPP API socket file %s does not exist", c.socketPath) + } else if err != nil { + return fmt.Errorf("VPP API socket error: %v", err) + } - if err := c.connect(c.sockAddr); err != nil { + if err := c.connect(c.socketPath); err != nil { return err } if err := c.open(); err != nil { + _ = c.disconnect() return err } @@ -132,312 +204,356 @@ func (c *vppClient) Connect() error { return nil } -func (c *vppClient) connect(sockAddr string) error { - addr, err := net.ResolveUnixAddr("unixpacket", sockAddr) - if err != nil { - Log.Debugln("ResolveUnixAddr error:", err) +func (c *Client) Disconnect() error { + if c.conn == nil { + return nil + } + log.Debugf("Disconnecting..") + + close(c.quit) + + if err := c.conn.CloseRead(); err != nil { + log.Debugf("closing readMsg failed: %v", err) + } + + // wait for readerLoop to return + c.wg.Wait() + + // Don't bother sending a vl_api_sockclnt_delete_t message, + // just close the socket. + if err := c.disconnect(); err != nil { return err } - conn, err := net.DialUnix("unixpacket", nil, addr) + return nil +} + +const defaultBufferSize = 4096 + +func (c *Client) connect(sockAddr string) error { + addr := &net.UnixAddr{Name: sockAddr, Net: "unix"} + + log.Debugf("Connecting to: %v", c.socketPath) + + conn, err := net.DialUnix("unix", nil, addr) if err != nil { - Log.Debugln("Dial error:", err) - return err + // we try different type of socket for backwards compatbility with VPP<=19.04 + if strings.Contains(err.Error(), "wrong type for socket") { + addr.Net = "unixpacket" + log.Debugf("%s, retrying connect with type unixpacket", err) + conn, err = net.DialUnix("unixpacket", nil, addr) + } + if err != nil { + log.Debugf("Connecting to socket %s failed: %s", addr, err) + return err + } } c.conn = conn - c.reader = bufio.NewReader(c.conn) + log.Debugf("Connected to socket (local addr: %v)", c.conn.LocalAddr().(*net.UnixAddr)) - Log.Debugf("Connected to socket: %v", addr) + c.reader = bufio.NewReaderSize(c.conn, defaultBufferSize) + c.writer = bufio.NewWriterSize(c.conn, defaultBufferSize) return nil } -func (c *vppClient) open() error { - msgCodec := new(codec.MsgCodec) +func (c *Client) disconnect() error { + log.Debugf("Closing socket") + if err := c.conn.Close(); err != nil { + log.Debugln("Closing socket failed:", err) + return err + } + return nil +} +const ( + sockCreateMsgId = 15 // hard-coded sockclnt_create message ID + createMsgContext = byte(123) + deleteMsgContext = byte(124) +) + +func (c *Client) open() error { + var msgCodec = codec.DefaultCodec + + // Request socket client create req := &memclnt.SockclntCreate{ - Name: []byte(govppClientName), + Name: c.clientName, } msg, err := msgCodec.EncodeMsg(req, sockCreateMsgId) if err != nil { - Log.Debugln("Encode error:", err) + log.Debugln("Encode error:", err) return err } // set non-0 context - msg[5] = 123 + msg[5] = createMsgContext - if err := c.write(msg); err != nil { - Log.Debugln("Write error: ", err) + if err := c.writeMsg(msg); err != nil { + log.Debugln("Write error: ", err) return err } - - readDeadline := time.Now().Add(ConnectTimeout) - if err := c.conn.SetReadDeadline(readDeadline); err != nil { - return err - } - msgReply, err := c.read() + msgReply, err := c.readMsgTimeout(nil, c.connectTimeout) if err != nil { - Log.Println("Read error:", err) - return err - } - // reset read deadline - if err := c.conn.SetReadDeadline(time.Time{}); err != nil { + log.Println("Read error:", err) return err } - //log.Printf("Client got (%d): % 0X", len(msgReply), msgReply) - reply := new(memclnt.SockclntCreateReply) if err := msgCodec.DecodeMsg(msgReply, reply); err != nil { - Log.Println("Decode error:", err) + log.Println("Decoding sockclnt_create_reply failed:", err) return err + } else if reply.Response != 0 { + return fmt.Errorf("sockclnt_create_reply: response error (%d)", reply.Response) } - Log.Debugf("SockclntCreateReply: Response=%v Index=%v Count=%v", + log.Debugf("SockclntCreateReply: Response=%v Index=%v Count=%v", reply.Response, reply.Index, reply.Count) c.clientIndex = reply.Index c.msgTable = make(map[string]uint16, reply.Count) for _, x := range reply.MessageTable { - name := string(bytes.TrimSuffix(bytes.Split(x.Name, []byte{0x00})[0], []byte{0x13})) + msgName := strings.Split(x.Name, "\x00")[0] + name := strings.TrimSuffix(msgName, "\x13") c.msgTable[name] = x.Index if strings.HasPrefix(name, "sockclnt_delete_") { c.sockDelMsgId = x.Index } - if DebugMsgIds { - Log.Debugf(" - %4d: %q", x.Index, name) + if debugMsgIds { + log.Debugf(" - %4d: %q", x.Index, name) } } return nil } -func (c *vppClient) Disconnect() error { - if c.conn == nil { - return nil +func (c *Client) GetMsgID(msgName string, msgCrc string) (uint16, error) { + if msgID, ok := c.msgTable[msgName+"_"+msgCrc]; ok { + return msgID, nil } - Log.Debugf("Disconnecting..") - - close(c.quit) - - // force readerLoop to timeout - if err := c.conn.SetReadDeadline(time.Now()); err != nil { - return err + return 0, &adapter.UnknownMsgError{ + MsgName: msgName, + MsgCrc: msgCrc, } +} - // wait for readerLoop to return - c.wg.Wait() +func (c *Client) SendMsg(context uint32, data []byte) error { + if len(data) < 10 { + return fmt.Errorf("invalid message data, length must be at least 10 bytes") + } + setMsgRequestHeader(data, c.clientIndex, context) - if err := c.close(); err != nil { - return err + if debug { + log.Debugf("sendMsg (%d) context=%v client=%d: % 02X", len(data), context, c.clientIndex, data) } - if err := c.conn.Close(); err != nil { - Log.Debugln("Close socket conn failed:", err) + if err := c.writeMsg(data); err != nil { + log.Debugln("writeMsg error: ", err) return err } return nil } -func (c *vppClient) close() error { - msgCodec := new(codec.MsgCodec) - - req := &memclnt.SockclntDelete{ - Index: c.clientIndex, - } - msg, err := msgCodec.EncodeMsg(req, c.sockDelMsgId) - if err != nil { - Log.Debugln("Encode error:", err) - return err - } - // set non-0 context - msg[5] = 124 +// setMsgRequestHeader sets client index and context in the message request header +// +// Message request has following structure: +// +// type msgRequestHeader struct { +// MsgID uint16 +// ClientIndex uint32 +// Context uint32 +// } +// +func setMsgRequestHeader(data []byte, clientIndex, context uint32) { + // message ID is already set + binary.BigEndian.PutUint32(data[2:6], clientIndex) + binary.BigEndian.PutUint32(data[6:10], context) +} - Log.Debugf("sending socklntDel (%d byes): % 0X\n", len(msg), msg) - if err := c.write(msg); err != nil { - Log.Debugln("Write error: ", err) - return err - } +func (c *Client) writeMsg(msg []byte) error { + // we lock to prevent mixing multiple message writes + c.writeMu.Lock() + defer c.writeMu.Unlock() - readDeadline := time.Now().Add(DisconnectTimeout) - if err := c.conn.SetReadDeadline(readDeadline); err != nil { - return err - } - msgReply, err := c.read() + header := c.headerPool.Get().([]byte) + err := writeMsgHeader(c.writer, header, len(msg)) if err != nil { - Log.Debugln("Read error:", err) - if nerr, ok := err.(net.Error); ok && nerr.Timeout() { - // we accept timeout for reply - return nil - } return err } - // reset read deadline - if err := c.conn.SetReadDeadline(time.Time{}); err != nil { + c.headerPool.Put(header) + + if err := writeMsgData(c.writer, msg, c.writer.Size()); err != nil { return err } - reply := new(memclnt.SockclntDeleteReply) - if err := msgCodec.DecodeMsg(msgReply, reply); err != nil { - Log.Debugln("Decode error:", err) + if err := c.writer.Flush(); err != nil { return err } - Log.Debugf("SockclntDeleteReply: Response=%v", reply.Response) + log.Debugf(" -- writeMsg done") return nil } -func (c *vppClient) GetMsgID(msgName string, msgCrc string) (uint16, error) { - msg := msgName + "_" + msgCrc - msgID, ok := c.msgTable[msg] - if !ok { - return 0, fmt.Errorf("unknown message: %q", msg) - } - return msgID, nil -} +func writeMsgHeader(w io.Writer, header []byte, dataLen int) error { + binary.BigEndian.PutUint32(header[8:12], uint32(dataLen)) -type reqHeader struct { - //MsgID uint16 - ClientIndex uint32 - Context uint32 -} - -func (c *vppClient) SendMsg(context uint32, data []byte) error { - h := &reqHeader{ - ClientIndex: c.clientIndex, - Context: context, - } - buf := new(bytes.Buffer) - if err := struc.Pack(buf, h); err != nil { + n, err := w.Write(header) + if err != nil { return err } - copy(data[2:], buf.Bytes()) - - Log.Debugf("SendMsg (%d) context=%v client=%d: data: % 02X", len(data), context, c.clientIndex, data) - - if err := c.write(data); err != nil { - Log.Debugln("write error: ", err) - return err + if debug { + log.Debugf(" - header sent (%d/%d): % 0X", n, len(header), header) } return nil } -func (c *vppClient) write(msg []byte) error { - h := &msgheader{ - Data_len: uint32(len(msg)), - } - buf := new(bytes.Buffer) - if err := struc.Pack(buf, h); err != nil { - return err - } - header := buf.Bytes() - - // we lock to prevent mixing multiple message sends - c.writeMu.Lock() - defer c.writeMu.Unlock() - - var w io.Writer = c.conn - - if n, err := w.Write(header); err != nil { - return err - } else { - Log.Debugf(" - header sent (%d/%d): % 0X", n, len(header), header) - } - if n, err := w.Write(msg); err != nil { - return err - } else { - Log.Debugf(" - msg sent (%d/%d): % 0X", n, len(msg), msg) +func writeMsgData(w io.Writer, msg []byte, writerSize int) error { + for i := 0; i <= len(msg)/writerSize; i++ { + x := i*writerSize + writerSize + if x > len(msg) { + x = len(msg) + } + if debug { + log.Debugf(" - x=%v i=%v len=%v mod=%v", x, i, len(msg), len(msg)/writerSize) + } + n, err := w.Write(msg[i*writerSize : x]) + if err != nil { + return err + } + if debug { + log.Debugf(" - data sent x=%d (%d/%d): % 0X", x, n, len(msg), msg) + } } - return nil } -type msgHeader struct { - MsgID uint16 - Context uint32 -} - -func (c *vppClient) readerLoop() { +func (c *Client) readerLoop() { defer c.wg.Done() + defer log.Debugf("reader loop done") + + var buf [8192]byte + for { select { case <-c.quit: - Log.Debugf("reader quit") return default: } - msg, err := c.read() + msg, err := c.readMsg(buf[:]) if err != nil { if isClosedError(err) { return } - Log.Debugf("READ FAILED: %v", err) + log.Debugf("readMsg error: %v", err) continue } - h := new(msgHeader) - if err := struc.Unpack(bytes.NewReader(msg), h); err != nil { - Log.Debugf("unpacking header failed: %v", err) - continue + + msgID, context := getMsgReplyHeader(msg) + if debug { + log.Debugf("recvMsg (%d) msgID=%d context=%v", len(msg), msgID, context) } - Log.Debugf("recvMsg (%d) msgID=%d context=%v", len(msg), h.MsgID, h.Context) - c.cb(h.MsgID, msg) + c.msgCallback(msgID, msg) } } -type msgheader struct { - Q int `struc:"uint64"` - Data_len uint32 `struc:"uint32"` - Gc_mark_timestamp uint32 `struc:"uint32"` - //data [0]uint8 +// getMsgReplyHeader gets message ID and context from the message reply header +// +// Message reply has following structure: +// +// type msgReplyHeader struct { +// MsgID uint16 +// Context uint32 +// } +// +func getMsgReplyHeader(msg []byte) (msgID uint16, context uint32) { + msgID = binary.BigEndian.Uint16(msg[0:2]) + context = binary.BigEndian.Uint32(msg[2:6]) + return } -func (c *vppClient) read() ([]byte, error) { - Log.Debug("reading next msg..") - - header := make([]byte, 16) +func (c *Client) readMsgTimeout(buf []byte, timeout time.Duration) ([]byte, error) { + // set read deadline + readDeadline := time.Now().Add(timeout) + if err := c.conn.SetReadDeadline(readDeadline); err != nil { + return nil, err + } - n, err := io.ReadAtLeast(c.reader, header, 16) + // read message + msgReply, err := c.readMsg(buf) if err != nil { return nil, err - } else if n == 0 { - Log.Debugln("zero bytes header") - return nil, nil } - if n != 16 { - Log.Debug("invalid header data (%d): % 0X", n, header[:n]) - return nil, fmt.Errorf("invalid header (expected 16 bytes, got %d)", n) + + // reset read deadline + if err := c.conn.SetReadDeadline(time.Time{}); err != nil { + return nil, err } - Log.Debugf(" - read header %d bytes: % 0X", n, header) - h := &msgheader{} - if err := struc.Unpack(bytes.NewReader(header[:]), h); err != nil { + return msgReply, nil +} + +func (c *Client) readMsg(buf []byte) ([]byte, error) { + log.Debug("reading msg..") + + header := c.headerPool.Get().([]byte) + msgLen, err := readMsgHeader(c.reader, header) + if err != nil { return nil, err } - Log.Debugf(" - decoded header: %+v", h) + c.headerPool.Put(header) + + msg, err := readMsgData(c.reader, buf, msgLen) + + log.Debugf(" -- readMsg done (buffered: %d)", c.reader.Buffered()) + + return msg, nil +} + +func readMsgHeader(r io.Reader, header []byte) (int, error) { + n, err := io.ReadAtLeast(r, header, 16) + if err != nil { + return 0, err + } + if n == 0 { + log.Debugln("zero bytes header") + return 0, nil + } else if n != 16 { + log.Debugf("invalid header (%d bytes): % 0X", n, header[:n]) + return 0, fmt.Errorf("invalid header (expected 16 bytes, got %d)", n) + } + + dataLen := binary.BigEndian.Uint32(header[8:12]) + + return int(dataLen), nil +} - msgLen := int(h.Data_len) - msg := make([]byte, msgLen) +func readMsgData(r io.Reader, buf []byte, dataLen int) ([]byte, error) { + var msg []byte + if buf == nil || len(buf) < dataLen { + msg = make([]byte, dataLen) + } else { + msg = buf[0:dataLen] + } - n, err = c.reader.Read(msg) + n, err := r.Read(msg) if err != nil { return nil, err } - Log.Debugf(" - read msg %d bytes (%d buffered)", n, c.reader.Buffered()) + if debug { + log.Debugf(" - read data (%d bytes): % 0X", n, msg[:n]) + } - if msgLen > n { - remain := msgLen - n - Log.Debugf("continue read for another %d bytes", remain) + if dataLen > n { + remain := dataLen - n + log.Debugf("continue reading remaining %d bytes", remain) view := msg[n:] for remain > 0 { - - nbytes, err := c.reader.Read(view) + nbytes, err := r.Read(view) if err != nil { return nil, err } else if nbytes == 0 { @@ -445,7 +561,7 @@ func (c *vppClient) read() ([]byte, error) { } remain -= nbytes - Log.Debugf("another data received: %d bytes (remain: %d)", nbytes, remain) + log.Debugf("another data received: %d bytes (remain: %d)", nbytes, remain) view = view[nbytes:] } @@ -455,7 +571,7 @@ func (c *vppClient) read() ([]byte, error) { } func isClosedError(err error) bool { - if err == io.EOF { + if errors.Is(err, io.EOF) { return true } return strings.HasSuffix(err.Error(), "use of closed network connection")