QMP connection pool

Author: sjmiller609

lib/hypervisor/qemu/pool.go

@@ -0,0 +1,57 @@
+package qemu
+
+import (
+	"sync"
+)
+
+// clientPool manages singleton QMP connections per socket path.
+// QEMU's QMP socket only allows one connection at a time, so we must
+// reuse existing connections rather than creating new ones.
+var clientPool = struct {
+	sync.RWMutex
+	clients map[string]*QEMU
+}{
+	clients: make(map[string]*QEMU),
+}
+
+// GetOrCreate returns an existing QEMU client for the socket path,
+// or creates a new one if none exists.
+func GetOrCreate(socketPath string) (*QEMU, error) {
+	// Try read lock first for existing connection
+	clientPool.RLock()
+	if client, ok := clientPool.clients[socketPath]; ok {
+		clientPool.RUnlock()
+		return client, nil
+	}
+	clientPool.RUnlock()
+
+	// Need to create new connection - acquire write lock
+	clientPool.Lock()
+	defer clientPool.Unlock()
+
+	// Double-check after acquiring write lock
+	if client, ok := clientPool.clients[socketPath]; ok {
+		return client, nil
+	}
+
+	// Create new client
+	client, err := newClient(socketPath)
+	if err != nil {
+		return nil, err
+	}
+
+	clientPool.clients[socketPath] = client
+	return client, nil
+}
+
+// Remove closes and removes a client from the pool.
+// Called automatically on errors to allow fresh reconnection.
+func Remove(socketPath string) {
+	clientPool.Lock()
+	defer clientPool.Unlock()
+
+	if client, ok := clientPool.clients[socketPath]; ok {
+		client.client.Close()
+		delete(clientPool.clients, socketPath)
+	}
+}

lib/hypervisor/qemu/qemu.go

@@ -11,16 +11,23 @@ import (
 
 // QEMU implements hypervisor.Hypervisor for QEMU VMM.
 type QEMU struct {
-	client *Client
+	client     *Client
+	socketPath string // for self-removal from pool on error
 }
 
-// New creates a new QEMU client for an existing QMP socket.
+// New returns a QEMU client for the given socket path.
+// Uses a connection pool to ensure only one connection per socket exists.
 func New(socketPath string) (*QEMU, error) {
+	return GetOrCreate(socketPath)
+}
+
+// newClient creates a new QEMU client (internal, used by pool).
+func newClient(socketPath string) (*QEMU, error) {
 	client, err := NewClient(socketPath)
 	if err != nil {
 		return nil, fmt.Errorf("create qemu client: %w", err)
 	}
-	return &QEMU{client: client}, nil
+	return &QEMU{client: client, socketPath: socketPath}, nil
 }
 
 // Verify QEMU implements the interface
@@ -40,18 +47,29 @@ func (q *QEMU) Capabilities() hypervisor.Capabilities {
 // DeleteVM removes the VM configuration from QEMU.
 // This sends a graceful shutdown signal to the guest.
 func (q *QEMU) DeleteVM(ctx context.Context) error {
-	return q.client.SystemPowerdown()
+	if err := q.client.SystemPowerdown(); err != nil {
+		Remove(q.socketPath)
+		return err
+	}
+	return nil
 }
 
 // Shutdown stops the QEMU process.
 func (q *QEMU) Shutdown(ctx context.Context) error {
-	return q.client.Quit()
+	if err := q.client.Quit(); err != nil {
+		Remove(q.socketPath)
+		return err
+	}
+	// Connection is gone after quit, remove from pool
+	Remove(q.socketPath)
+	return nil
 }
 
 // GetVMInfo returns current VM state.
 func (q *QEMU) GetVMInfo(ctx context.Context) (*hypervisor.VMInfo, error) {
 	status, err := q.client.Status()
 	if err != nil {
+		Remove(q.socketPath)
 		return nil, fmt.Errorf("query status: %w", err)
 	}
 
@@ -85,12 +103,20 @@ func (q *QEMU) GetVMInfo(ctx context.Context) (*hypervisor.VMInfo, error) {
 
 // Pause suspends VM execution.
 func (q *QEMU) Pause(ctx context.Context) error {
-	return q.client.Stop()
+	if err := q.client.Stop(); err != nil {
+		Remove(q.socketPath)
+		return err
+	}
+	return nil
 }
 
 // Resume continues VM execution.
 func (q *QEMU) Resume(ctx context.Context) error {
-	return q.client.Continue()
+	if err := q.client.Continue(); err != nil {
+		Remove(q.socketPath)
+		return err
+	}
+	return nil
 }
 
 // Snapshot creates a VM snapshot.

lib/instances/manager.go

@@ -264,7 +264,7 @@ func (m *manager) RotateLogs(ctx context.Context, maxBytes int64, maxFiles int)
 			m.paths.InstanceHypemanLog(inst.Id),
 		}
 		for _, logPath := range logPaths {
-			if err := rotateLogIfNeeded(logPath, maxBytes, maxFiles); err != nil {
+		if err := rotateLogIfNeeded(logPath, maxBytes, maxFiles); err != nil {
 				lastErr = err // Continue with other logs, but track error
 			}
 		}