ovsdb: add internal receive loop for notifications and callbacks

Author: mdlayher

ovsdb/client.go

@@ -18,8 +18,12 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
 	"net"
+	"strings"
+	"sync"
+	"sync/atomic"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 )
@@ -28,6 +32,20 @@ import (
 type Client struct {
 	c  *jsonrpc.Conn
 	ll *log.Logger
+
+	// Incremented atomically when sending RPCs.
+	rpcID *int64
+
+	// Callbacks for RPC responses.
+	cbMu      sync.RWMutex
+	callbacks map[int]chan rpcResponse
+
+	wg *sync.WaitGroup
+}
+
+type rpcResponse struct {
+	Result json.RawMessage
+	Error  error
 }
 
 // An OptionFunc is a function which can configure a Client.
@@ -60,14 +78,41 @@ func New(conn net.Conn, options ...OptionFunc) (*Client, error) {
 		}
 	}
 
+	// Set up RPC request IDs.
+	var rpcID int64
+	client.rpcID = &rpcID
+
+	// Set up the JSON-RPC connection.
 	client.c = jsonrpc.NewConn(conn, client.ll)
 
+	// Set up callbacks.
+	client.callbacks = make(map[int]chan rpcResponse)
+
+	// Start up any background routines.
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	// Handle all incoming RPC responses and notifications.
+	go func() {
+		defer wg.Done()
+		client.listen()
+	}()
+
+	client.wg = &wg
+
 	return client, nil
 }
 
+// requestID returns the next available request ID for an RPC.
+func (c *Client) requestID() int {
+	return int(atomic.AddInt64(c.rpcID, 1))
+}
+
 // Close closes a Client's connection.
 func (c *Client) Close() error {
-	return c.c.Close()
+	err := c.c.Close()
+	c.wg.Wait()
+	return err
 }
 
 // ListDatabases returns the name of all databases known to the OVSDB server.
@@ -82,6 +127,11 @@ func (c *Client) ListDatabases() ([]string, error) {
 
 // rpc performs a single RPC request, and checks the response for errors.
 func (c *Client) rpc(method string, out interface{}, args ...interface{}) error {
+	// Unmarshal results into empty struct if no out specified.
+	if out == nil {
+		out = &struct{}{}
+	}
+
 	// Captures any OVSDB errors.
 	r := result{
 		Reply: out,
@@ -90,10 +140,24 @@ func (c *Client) rpc(method string, out interface{}, args ...interface{}) error
 	req := jsonrpc.Request{
 		Method: method,
 		Params: args,
-		// Let the client handle the request ID.
+		ID:     c.requestID(),
+	}
+
+	// Add callback for this RPC ID to return results via channel.
+	ch := make(chan rpcResponse, 0)
+	c.addCallback(req.ID, ch)
+
+	if err := c.c.Send(req); err != nil {
+		return err
+	}
+
+	// Wait for callback to fire.
+	res := <-ch
+	if err := res.Error; err != nil {
+		return err
 	}
 
-	if err := c.c.Execute(req, &r); err != nil {
+	if err := json.Unmarshal(res.Result, &r); err != nil {
 		return err
 	}
 
@@ -105,6 +169,70 @@ func (c *Client) rpc(method string, out interface{}, args ...interface{}) error
 	return nil
 }
 
+// listen starts an RPC receive loop that can return RPC results to
+// clients via a callback.
+func (c *Client) listen() {
+	for {
+		res, err := c.c.Receive()
+		if err != nil {
+			// EOF or closed connection means time to stop serving.
+			if err == io.EOF || strings.Contains(err.Error(), "use of closed network") {
+				return
+			}
+
+			// For any other connection errors, just keep trying.
+			continue
+		}
+
+		// TODO(mdlayher): deal with RPC notifications.
+
+		// Handle any JSON-RPC top-level errors.
+		if err := res.Err(); err != nil {
+			c.doCallback(*res.ID, rpcResponse{
+				Error: err,
+			})
+			continue
+		}
+
+		// Return RPC results via callback.
+		c.doCallback(*res.ID, rpcResponse{
+			Result: res.Result,
+		})
+	}
+}
+
+// addCallback registers a callback for an RPC response for the specified ID,
+// and accepts a channel to return the results on.
+func (c *Client) addCallback(id int, ch chan rpcResponse) {
+	c.cbMu.Lock()
+	defer c.cbMu.Unlock()
+
+	if _, ok := c.callbacks[id]; ok {
+		// This ID was already registered.
+		panicf("OVSDB callback with ID %d already registered", id)
+	}
+
+	c.callbacks[id] = ch
+}
+
+// doCallback performs a callback for an RPC response and clears the
+// callback on completion.
+func (c *Client) doCallback(id int, res rpcResponse) {
+	c.cbMu.Lock()
+	defer c.cbMu.Unlock()
+
+	ch, ok := c.callbacks[id]
+	if !ok {
+		// Nobody is listening to this callback.
+		panicf("OVSDB callback with ID %d has no listeners", id)
+	}
+
+	// Return result, clean up channel, and remove this callback.
+	ch <- res
+	close(ch)
+	delete(c.callbacks, id)
+}
+
 // A result is used to unmarshal JSON-RPC results, and to check for any errors.
 type result struct {
 	Reply interface{}
@@ -143,3 +271,7 @@ type Error struct {
 func (e *Error) Error() string {
 	return fmt.Sprintf("%s: %s: %s", e.Err, e.Details, e.Syntax)
 }
+
+func panicf(format string, a ...interface{}) {
+	panic(fmt.Sprintf(format, a...))
+}

ovsdb/client_integration_test.go

@@ -15,31 +15,62 @@
 package ovsdb_test
 
 import (
-	"os"
+	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb"
 	"github.com/google/go-cmp/cmp"
 )
 
 func TestClientIntegration(t *testing.T) {
-	// Assume the standard Linux location for the socket.
-	const sock = "/var/run/openvswitch/db.sock"
-	if _, err := os.Open(sock); err != nil {
-		t.Skipf("could not access %q: %v", sock, err)
-	}
-
-	c, err := ovsdb.Dial("unix", sock)
-	if err != nil {
-		t.Fatalf("failed to dial: %v", err)
-	}
+	c := dialOVSDB(t)
 	defer c.Close()
 
 	t.Run("databases", func(t *testing.T) {
 		testClientDatabases(t, c)
 	})
 }
 
+func TestClientIntegrationConcurrent(t *testing.T) {
+	c := dialOVSDB(t)
+	defer c.Close()
+
+	const n = 512
+
+	// Wait for all goroutines to start before performing RPCs,
+	// wait for them all to exit before ending the test.
+	var startWG, doneWG sync.WaitGroup
+	startWG.Add(n)
+	doneWG.Add(n)
+
+	// Block all goroutines until they're done spinning up.
+	sigC := make(chan struct{}, 0)
+
+	for i := 0; i < n; i++ {
+		go func(c *ovsdb.Client) {
+			// Block goroutines until all are spun up.
+			startWG.Done()
+			<-sigC
+
+			for j := 0; j < 4; j++ {
+				_, err := c.ListDatabases()
+				if err != nil {
+					panic(fmt.Sprintf("failed to query concurrently: %v", err))
+				}
+			}
+
+			doneWG.Done()
+		}(c)
+	}
+
+	// Unblock all goroutines once they're all spun up, and wait
+	// for them all to finish reading.
+	startWG.Wait()
+	close(sigC)
+	doneWG.Wait()
+}
+
 func testClientDatabases(t *testing.T, c *ovsdb.Client) {
 	dbs, err := c.ListDatabases()
 	if err != nil {
@@ -52,3 +83,16 @@ func testClientDatabases(t *testing.T, c *ovsdb.Client) {
 		t.Fatalf("unexpected databases (-want +got):\n%s", diff)
 	}
 }
+
+func dialOVSDB(t *testing.T) *ovsdb.Client {
+	t.Helper()
+
+	// Assume the standard Linux location for the socket.
+	const sock = "/var/run/openvswitch/db.sock"
+	c, err := ovsdb.Dial("unix", sock)
+	if err != nil {
+		t.Skipf("could not access %q: %v", sock, err)
+	}
+
+	return c
+}

ovsdb/client_test.go

@@ -15,25 +15,45 @@
 package ovsdb_test
 
 import (
+	"encoding/json"
 	"fmt"
+	"log"
+	"os"
 	"testing"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb"
 	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 	"github.com/google/go-cmp/cmp"
 )
 
-func TestClientError(t *testing.T) {
+func TestClientJSONRPCError(t *testing.T) {
 	const str = "some error"
 
-	c, done := testClient(t, func(_ jsonrpc.Request) jsonrpc.Response {
+	c, _, done := testClient(t, func(_ jsonrpc.Request) jsonrpc.Response {
 		return jsonrpc.Response{
-			ID: 1,
-			Result: &ovsdb.Error{
+			ID:    intPtr(1),
+			Error: str,
+		}
+	})
+	defer done()
+
+	_, err := c.ListDatabases()
+	if err == nil {
+		t.Fatal("expected an error, but none occurred")
+	}
+}
+
+func TestClientOVSDBError(t *testing.T) {
+	const str = "some error"
+
+	c, _, done := testClient(t, func(_ jsonrpc.Request) jsonrpc.Response {
+		return jsonrpc.Response{
+			ID: intPtr(1),
+			Result: mustMarshalJSON(t, &ovsdb.Error{
 				Err:     str,
 				Details: "malformed",
 				Syntax:  "{}",
-			},
+			}),
 		}
 	})
 	defer done()
@@ -56,7 +76,7 @@ func TestClientError(t *testing.T) {
 func TestClientListDatabases(t *testing.T) {
 	want := []string{"Open_vSwitch", "test"}
 
-	c, done := testClient(t, func(req jsonrpc.Request) jsonrpc.Response {
+	c, _, done := testClient(t, func(req jsonrpc.Request) jsonrpc.Response {
 		if diff := cmp.Diff("list_dbs", req.Method); diff != "" {
 			panicf("unexpected RPC method (-want +got):\n%s", diff)
 		}
@@ -66,8 +86,8 @@ func TestClientListDatabases(t *testing.T) {
 		}
 
 		return jsonrpc.Response{
-			ID:     1,
-			Result: want,
+			ID:     intPtr(1),
+			Result: mustMarshalJSON(t, want),
 		}
 	})
 	defer done()
@@ -82,17 +102,35 @@ func TestClientListDatabases(t *testing.T) {
 	}
 }
 
-func testClient(t *testing.T, fn jsonrpc.TestFunc) (*ovsdb.Client, func()) {
+func testClient(t *testing.T, fn jsonrpc.TestFunc) (*ovsdb.Client, chan<- *jsonrpc.Response, func()) {
 	t.Helper()
 
-	conn, done := jsonrpc.TestNetConn(t, fn)
+	conn, notifC, done := jsonrpc.TestNetConn(t, fn)
 
-	c, err := ovsdb.New(conn)
+	c, err := ovsdb.New(conn, ovsdb.Debug(log.New(os.Stderr, "", 0)))
 	if err != nil {
 		t.Fatalf("failed to dial: %v", err)
 	}
 
-	return c, done
+	return c, notifC, func() {
+		_ = c.Close()
+		done()
+	}
+}
+
+func mustMarshalJSON(t *testing.T, v interface{}) []byte {
+	t.Helper()
+
+	b, err := json.Marshal(v)
+	if err != nil {
+		t.Fatalf("failed to marshal JSON: %v", err)
+	}
+
+	return b
+}
+
+func intPtr(i int) *int {
+	return &i
 }
 
 func panicf(format string, a ...interface{}) {