[lxc-devel] [lxd/master] lxd/instance/qmp: Merge Go routines

[stgraber on Github]

A non-text attachment was scrubbed...
Name: not available
Type: text/x-mailbox
Size: 354 bytes
Desc: not available
URL: <http://lists.linuxcontainers.org/pipermail/lxc-devel/attachments/20201028/395f7757/attachment.bin>
-------------- next part --------------
From d735de0a151eda78d7f73b157f91789ad38d396e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?St=C3=A9phane=20Graber?= <stgraber at ubuntu.com>
Date: Wed, 28 Oct 2020 17:44:24 -0400
Subject: [PATCH] lxd/instance/qmp: Merge Go routines
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: Stéphane Graber <stgraber at ubuntu.com>
---
 lxd/instance/drivers/qmp/monitor.go | 72 +++++++++++++++--------------
 1 file changed, 37 insertions(+), 35 deletions(-)

diff --git a/lxd/instance/drivers/qmp/monitor.go b/lxd/instance/drivers/qmp/monitor.go
index e02bc96590..3e4d3d1dd8 100644
--- a/lxd/instance/drivers/qmp/monitor.go
+++ b/lxd/instance/drivers/qmp/monitor.go
@@ -75,47 +75,40 @@ func Connect(path string, serialCharDev string, eventHandler func(name string, d
 }
 
 func (m *Monitor) run() error {
-	// Start ringbuffer monitoring go routine.
-	go func() {
-		for {
-			// Read the ringbuffer.
-			resp, err := m.qmp.Run([]byte(fmt.Sprintf(`{"execute": "ringbuf-read", "arguments": {"device": "%s", "size": %d, "format": "utf8"}}`, m.serialCharDev, RingbufSize)))
-			if err != nil {
-				m.Disconnect()
-				return
-			}
-
-			// Decode the response.
-			var respDecoded struct {
-				Return string `json:"return"`
-			}
+	// Ringbuffer monitoring function.
+	checkBuffer := func() {
+		// Read the ringbuffer.
+		resp, err := m.qmp.Run([]byte(fmt.Sprintf(`{"execute": "ringbuf-read", "arguments": {"device": "%s", "size": %d, "format": "utf8"}}`, m.serialCharDev, RingbufSize)))
+		if err != nil {
+			// Failure to send a command, assume disconnected/crashed.
+			m.Disconnect()
+			return
+		}
 
-			err = json.Unmarshal(resp, &respDecoded)
-			if err != nil {
-				continue
-			}
+		// Decode the response.
+		var respDecoded struct {
+			Return string `json:"return"`
+		}
 
-			// Extract the last entry.
-			entries := strings.Split(respDecoded.Return, "\n")
-			if len(entries) > 1 {
-				status := entries[len(entries)-2]
+		err = json.Unmarshal(resp, &respDecoded)
+		if err != nil {
+			// Received bad data, assume disconnected/crashed.
+			m.Disconnect()
+			return
+		}
 
-				if status == "STARTED" {
-					m.agentReady = true
-				} else if status == "STOPPED" {
-					m.agentReady = false
-				}
-			}
+		// Extract the last entry.
+		entries := strings.Split(respDecoded.Return, "\n")
+		if len(entries) > 1 {
+			status := entries[len(entries)-2]
 
-			// Wait until next read or cancel.
-			select {
-			case <-m.chDisconnect:
-				return
-			case <-time.After(10 * time.Second):
-				continue
+			if status == "STARTED" {
+				m.agentReady = true
+			} else if status == "STOPPED" {
+				m.agentReady = false
 			}
 		}
-	}()
+	}
 
 	// Start event monitoring go routine.
 	chEvents, err := m.qmp.Events()
@@ -124,7 +117,11 @@ func (m *Monitor) run() error {
 	}
 
 	go func() {
+		// Initial read from the ringbuffer.
+		go checkBuffer()
+
 		for {
+			// Wait for an event, disconnection or timeout.
 			select {
 			case <-m.chDisconnect:
 				return
@@ -136,6 +133,11 @@ func (m *Monitor) run() error {
 				if m.eventHandler != nil {
 					m.eventHandler(e.Event, e.Data)
 				}
+			case <-time.After(10 * time.Second):
+				// Check if the ringbuffer was updated (non-blocking).
+				go checkBuffer()
+
+				continue
 			}
 		}
 	}()