server/comms: fix synchronization issues and drop sleeps in tests #435

buck54321 · 2020-06-01T01:49:05Z

Resolves #394

Eliminate sloppy time.Sleep sync patterns in tests. Improves WSLink.Send synchronization so that a graceful shutdown will wait for messages to finish sending.

Channel patterns are used throughout. There are still 2 time.Sleeps, one for response handler timeout and one for checking that ping-pongs are working.

buck54321 · 2020-06-01T01:53:21Z

client/rpcserver/rpcserver.go

-// Run starts the web server. Satisfies the dex.Runner interface.
-func (s *RPCServer) Run(ctx context.Context) {
+// Connect starts the RPC server. Satisfies the dex.Connector interface.
+func (s *RPCServer) Connect(ctx context.Context) (error, *sync.WaitGroup) {


BTW, I'm going to reverse the order of these return values. I'm just trying to avoid any more conflicts than necessary with #408 for now.

TBH, I'm thinking this PR goes in before #408, even though that creates some resolution work for @kevinstl . I think having the canonical pattern laid out first is good. How would you prefer to proceed, @buck54321 ?

buck54321 · 2020-06-01T02:17:01Z

dex/ws/wslink.go

+	c.sendWG.Add(1)
+	c.outChan <- b
+	go func() {
+		c.writeMtx.Lock()
+		b := <-c.outChan
+		c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+		err := c.conn.WriteMessage(websocket.TextMessage, b)
+		c.writeMtx.Unlock()
+		c.sendWG.Done()
+		if err != nil {
+			c.Disconnect()
+		}
+	}()


This pattern is quite different than before, when there was a persistent goroutine to handle pings and outgoing messages. Doing it like this with the sendWg sync.WaitGroup allows us to have a "graceful" shutdown, and not be racy with any messages sent right before Disconnect is called. The outChan is still used to sequence the sends.

buck54321 · 2020-06-01T02:21:41Z

server/comms/comms_test.go

@@ -34,14 +37,51 @@ func newServer() *Server {
 	}
 }

+func giveItASecond(f func() bool) bool {


Even though we are eliminating time.Sleep for synchronization, there are still some times where a channel isn't convenient or would add significant complexity. giveItASecond can be used to sync to some arbitrary condition, e.g. the client's presence in the server's client map, with a 1 ms delay between checks.

buck54321 · 2020-06-01T02:24:39Z

server/comms/comms_test.go

 	// Send the message if their is a receiver for the current test.
 	if conn.recv != nil {
 		conn.recv <- msg
 	}
-	conn.write++


write wasn't being used.

buck54321 · 2020-06-01T02:27:18Z

server/comms/comms_test.go

-		ListenAddrs: []string{portAddr},
+		ListenAddrs: []string{":0"},
 		RPCKey:      keyPath,
 		RPCCert:     certPath,
 	})
 	if err != nil {
 		t.Fatalf("server constructor error: %v", err)
 	}
+	address := "wss://" + server.listeners[0].Addr().String() + "/ws"


This allows testing in parallel.

JoeGruffins

Working well for me.

dex/ws/wslink.go

JoeGruffins · 2020-06-01T04:40:51Z

dex/ws/wslink.go

+	go func() {
+		c.writeMtx.Lock()
+		b := <-c.outChan
+		c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+		err := c.conn.WriteMessage(websocket.TextMessage, b)
+		c.writeMtx.Unlock()
+		c.sendWG.Done()
+		if err != nil {
+			c.Disconnect()
+		}
+	}()


This looks a bit more memory intensive. Is that ok? I was recently scolded by @davecgh in dcrd for trying to create a goroutine for all rpc requests sent to the server.

Although goroutines are not terribly heavy, we will be churning goroutines quite a bit this way.

To summarize what we're trying to accomplish here: With a sending goroutine that receives on outChan as before we run the risk of filling that channel's buffer in Send. Relying on a channel buffer was convenient, but Send might block if that buffer fills. ~~That maybe is not too bad because the quit channel was the arbiter of connections status and messages were not dropped when the channel was full, just blocking.~~ EDIT: That is a problem as one bad client can hold up the caller for several seconds.

We realized that a queue for the outgoing messages might needed. The present solution solves the queuing issue via the Go scheduler, with a goroutine created for each message sent, at the cost of goroutine churn. This may become a problem; I don't know.

Other solutions:

A more complex implementation is to append under lock or in a for-select loop the messages to a slice of queued outgoing messages, while a the same goroutine doing the append pops entries off and sends them. It's complex because a slice is not a synchronization mechanism. ~~The epochPump is an example, and that's pretty hairy.~~ What might work is a for-select with cases for (1) quit, (2), append to queue, and (3) write message from head/first.

Have each Send block until it can actually WriteMessage itselft. This would involve acquiring a semaphore (using a channel with buffer length 1, not a mutex, to select with quit in another case) or checking quit after the mutex is locked, and it would make Send blocking...

I actually think having Send block might be desirable so that the caller gets the error from WriteMessage, whereas now nil does not mean the send was successful just that marshalling was successful and the message is now queued for send.

Anyway, we've chatted on matrix and I think we can run with this (a goroutine for each message sent) for now. A requirement with Send is that a misbehaving client cannot gum up the works for up to writeWait (seconds). Would we like each connection to have a single goroutine for the life of the connection? Yes, but I think that using a goroutine for each message to handle this concurrency requirement (non-blocking sends) is appropriate given the complexity of alternative solutions for message queuing.

chappjc

ConnectionMaster refactoring works for me. Also, I think we can run with the Send modification that uses a new goroutine as a means to schedule sends without blocking the Send caller. We can consider alternatives that do not require this, but it should work.

chappjc · 2020-06-01T17:49:58Z

server/comms/server.go

@@ -341,12 +349,14 @@ func (s *Server) disconnectClients() {
 }

 // addClient assigns the client an ID and adds it to the map.


Let's also add that it now attempts connection.

chappjc · 2020-06-01T19:22:50Z

dex/ws/wslink.go

+	defer c.Disconnect()
+	defer c.wg.Done()


last in first out, so defer c.Disconnect should be after defer c.wg.Done() to ensure Disconnect gets called.

buck54321 added 2 commits May 29, 2020 18:03

eliminate sleeps from comms tests

866047d

Channel patterns are used throughout. There are still 2 time.Sleeps, one for response handler timeout and one for checking that ping-pongs are working.

fix other sync issues

42fd176

buck54321 commented Jun 1, 2020

View reviewed changes

JoeGruffins mentioned this pull request Jun 1, 2020

dexc: webserver and rpcserve bind failures should shutdown dexc #408

Closed

JoeGruffins approved these changes Jun 1, 2020

View reviewed changes

switch connector return value order

4c64c3c

chappjc reviewed Jun 1, 2020

View reviewed changes

better docstring for addClient

8369fa1

chappjc reviewed Jun 1, 2020

View reviewed changes

fix defer order

701cd7d

chappjc approved these changes Jun 2, 2020

View reviewed changes

chappjc merged commit b90705e into decred:master Jun 2, 2020

kevinstl mentioned this pull request Jun 3, 2020

dexc: webserver and rpcserve bind failures should shutdown dexc #443

Merged

buck54321 mentioned this pull request Jun 16, 2020

server/comms: TestClientRequests fails randomly with "clientCount != 0" #255

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

server/comms: fix synchronization issues and drop sleeps in tests #435

server/comms: fix synchronization issues and drop sleeps in tests #435

buck54321 commented Jun 1, 2020

buck54321 Jun 1, 2020

chappjc Jun 1, 2020

buck54321 Jun 1, 2020

buck54321 Jun 1, 2020

buck54321 Jun 1, 2020

buck54321 Jun 1, 2020

JoeGruffins left a comment

JoeGruffins Jun 1, 2020

chappjc Jun 1, 2020 •

edited

Loading

chappjc Jun 1, 2020 •

edited

Loading

chappjc left a comment

chappjc Jun 1, 2020

chappjc Jun 1, 2020

		@@ -341,12 +349,14 @@ func (s *Server) disconnectClients() {
		}

		// addClient assigns the client an ID and adds it to the map.

server/comms: fix synchronization issues and drop sleeps in tests #435

server/comms: fix synchronization issues and drop sleeps in tests #435

Conversation

buck54321 commented Jun 1, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JoeGruffins left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chappjc Jun 1, 2020 • edited Loading

Choose a reason for hiding this comment

chappjc Jun 1, 2020 • edited Loading

Choose a reason for hiding this comment

chappjc left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chappjc Jun 1, 2020 •

edited

Loading

chappjc Jun 1, 2020 •

edited

Loading