From e0c06582abc40236a383d0260a1f914576d813b6 Mon Sep 17 00:00:00 2001 From: Arthur Lu Date: Tue, 7 Jul 2026 23:13:18 +0000 Subject: [PATCH] new mutex model --- app/app.go | 38 ++++++++------- app/model.go | 125 +++++++++++++++++++++++++++++-------------------- app/proxmox.go | 30 ++++++------ app/types.go | 35 ++++++++++++-- 4 files changed, 141 insertions(+), 87 deletions(-) diff --git a/app/app.go b/app/app.go index b323e99..24becac 100644 --- a/app/app.go +++ b/app/app.go @@ -34,13 +34,10 @@ func Run() { cluster := Cluster{} cluster.Init(client) - start := time.Now() log.Printf("[INFO] starting cluster sync\n") err := cluster.Sync() if err != nil { log.Printf("[ERR ] error encountered while syncing cluster: %s", err) - } else { - log.Printf("[INFO] synced cluster in %d ms\n", time.Since(start).Milliseconds()) } // set repeating update for full rebuilds @@ -53,15 +50,11 @@ func Run() { case <-channel: return case <-ticker.C: - start := time.Now() log.Printf("[INFO] starting cluster sync\n") err := cluster.Sync() if err != nil { log.Printf("[ERR ] error encountered while syncing cluster: %s", err) - } else { - log.Printf("[INFO] synced cluster in %d ms", time.Since(start).Milliseconds()) } - } } }() @@ -116,7 +109,6 @@ func Run() { }) router.POST("/sync", func(c *gin.Context) { - start := time.Now() log.Printf("[INFO] starting cluster sync\n") err := cluster.Sync() if err != nil { @@ -124,21 +116,26 @@ func Run() { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } else { - log.Printf("[INFO] synced cluster in %fs\n", time.Since(start).Seconds()) return } }) router.POST("/nodes/:node/sync", func(c *gin.Context) { nodeName := c.Param("node") - start := time.Now() + log.Printf("[INFO] starting %s sync\n", nodeName) - err := SyncNode(&cluster, nodeName) + + node, err := cluster.GetNode(nodeName) + if err != nil { + log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error()) + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + err = node.Sync() if err != nil { log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error()) c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) - } else { - log.Printf("[INFO] synced %s in %fs\n", nodeName, time.Since(start).Seconds()) } }) @@ -146,17 +143,24 @@ func Run() { nodeName := c.Param("node") vmid, err := strconv.ParseUint(c.Param("vmid"), 10, 64) if err != nil { + log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error()) c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))}) return } - start := time.Now() + log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid) - err = SyncInstance(&cluster, nodeName, vmid) + + instance, err := cluster.GetInstance(nodeName, vmid) + if err != nil { + log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error()) + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + err = instance.Sync() if err != nil { log.Printf("[ERR ] failed to sync %s.%d: %s", nodeName, vmid, err.Error()) c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) - } else { - log.Printf("[INFO] synced %s in %fs\n", nodeName, time.Since(start).Seconds()) } }) diff --git a/app/model.go b/app/model.go index df6770f..ada2cf4 100644 --- a/app/model.go +++ b/app/model.go @@ -28,6 +28,9 @@ func (cluster *Cluster) Get() (*Cluster, error) { } func (cluster *Cluster) Sync() error { + cluster.lock.Lock() + defer cluster.lock.Unlock() + err := cluster.Build() if err != nil { cluster.OK = false @@ -45,10 +48,14 @@ func (cluster *Cluster) Sync() error { } // hard sync cluster +// caller MUST lock cluster BEFORE entering func (cluster *Cluster) Build() error { - // aquire lock on cluster, release on return - cluster.lock.Lock() - defer cluster.lock.Unlock() + // check to confirm cluster is locked before making any changes + if !cluster.lock.IsLocked() { + return fmt.Errorf("cluster was not locked before calling cluster.Sync()") + } + + start := time.Now() cluster.Nodes = make(map[string]*Node) @@ -57,17 +64,16 @@ func (cluster *Cluster) Build() error { // get all nodes nodes, err := cluster.pve.Nodes() if err != nil { - cluster.lock.Unlock() return err } // for each node: for _, nodeName := range nodes { wg.Go(func() error { - start := time.Now() - // rebuild node node := Node{} + node.lock.Lock() + node.Name = nodeName node.cluster = cluster @@ -75,14 +81,14 @@ func (cluster *Cluster) Build() error { //err := cluster.BuildNode(nodeName) if err != nil { // if an error was encountered, continue and log the error log.Printf("[ERR ] error encountered while syncing node %s: %s", nodeName, err) - } else { - log.Printf("[INFO] synced node %s in %d ms", nodeName, time.Since(start).Milliseconds()) } cluster.NodesLock.Lock() cluster.Nodes[nodeName] = &node cluster.NodesLock.Unlock() + node.lock.Unlock() + return err }) } @@ -93,13 +99,20 @@ func (cluster *Cluster) Build() error { return err } + log.Printf("[INFO] built cluster in %d ms", time.Since(start).Milliseconds()) + return nil } +// resolve membership of instances to pools +// caller MUST lock cluster BEFORE entering func (cluster *Cluster) ResolvePoolMembership() error { - // aquire lock on cluster, release on return - cluster.lock.Lock() - defer cluster.lock.Unlock() + // check to confirm cluster is locked before making any changes + if !cluster.lock.IsLocked() { + return fmt.Errorf("cluster was not locked before calling cluster.ResolvePoolMembership()") + } + + start := time.Now() //clear existing pool memberships for _, node := range cluster.Nodes { @@ -137,6 +150,8 @@ func (cluster *Cluster) ResolvePoolMembership() error { } } + log.Printf("[INFO] resovled cluster instance pool memberships in %d ms", time.Since(start).Milliseconds()) + return nil } @@ -163,14 +178,16 @@ func (cluster *Cluster) GetNode(nodeName string) (*Node, error) { } } -func SyncNode(cluster *Cluster, nodeName string) error { - node, err := cluster.GetNode(nodeName) - if err != nil { - cluster.OK = false - return err - } +func (node *Node) Sync() error { + node.lock.Lock() + defer node.lock.Unlock() - err = node.Build() + cluster := node.cluster + + cluster.lock.Lock() + defer cluster.lock.Unlock() + + err := node.Build() if err != nil { cluster.OK = false return err @@ -188,20 +205,23 @@ func SyncNode(cluster *Cluster, nodeName string) error { // hard sync node // returns error if the node could not be reached +// caller MUST lock node BEFORE entering func (node *Node) Build() error { + // check to confirm cluster is locked before making any changes + if !node.lock.IsLocked() { + return fmt.Errorf("node was not locked before calling node.Build()") + } + + start := time.Now() + nodeName := node.Name cluster := node.cluster - n, err := cluster.pve.Node(nodeName) + err := cluster.pve.Node(node, nodeName) if err != nil && cluster.Nodes[nodeName] == nil { // node is unreachable and did not exist previously // return an error because we requested to sync a node that was not already in the cluster return fmt.Errorf("error retrieving %s: %s", nodeName, err.Error()) } - *node = *n // copies lock value, which should be unlocked - - // aquire lock on node, release on return - node.lock.Lock() - defer node.lock.Unlock() wg, _ := errgroup.WithContext(context.Background()) @@ -219,10 +239,11 @@ func (node *Node) Build() error { } for _, vmid := range vms { wg.Go(func() error { - start := time.Now() instanceID := InstanceID(paas.SafeUint64(vmid)) instance := Instance{} + instance.lock.Lock() + instance.VMID = instanceID instance.Type = VM instance.node = node @@ -230,14 +251,14 @@ func (node *Node) Build() error { err := instance.Build() if err != nil { // if an error was encountered, continue and log the error log.Printf("[ERR ] error encountered while syncing vm %s.%d: %s", nodeName, vmid, err) - } else { - log.Printf("[INFO] synced vm %s.%d in %d ms", nodeName, vmid, time.Since(start).Milliseconds()) } node.InstancesLock.Lock() node.Instances[instanceID] = &instance node.InstancesLock.Unlock() + instance.lock.Unlock() + return nil }) } @@ -249,10 +270,10 @@ func (node *Node) Build() error { } for _, vmid := range cts { wg.Go(func() error { - start := time.Now() - instanceID := InstanceID(paas.SafeUint64(vmid)) instance := Instance{} + instance.lock.Lock() + instance.VMID = instanceID instance.Type = CT instance.node = node @@ -260,15 +281,14 @@ func (node *Node) Build() error { err := instance.Build() if err != nil { // if an error was encountered, continue and log the error log.Printf("[ERR ] error encountered while syncing ct %s.%d: %s", nodeName, vmid, err) - } else { - log.Printf("[INFO] synced ct %s.%d in %d ms", nodeName, vmid, time.Since(start).Milliseconds()) - } node.InstancesLock.Lock() node.Instances[instanceID] = &instance node.InstancesLock.Unlock() + instance.lock.Unlock() + return nil }) } @@ -289,9 +309,7 @@ func (node *Node) Build() error { node.cluster = cluster - //cluster.NodesLock.Lock() - //cluster.Nodes[nodeName] = node - //cluster.NodesLock.Unlock() + log.Printf("[INFO] built node %s in %d ms", node.Name, time.Since(start).Milliseconds()) return nil } @@ -328,14 +346,16 @@ func (cluster *Cluster) GetInstance(nodeName string, vmid uint64) (*Instance, er } } -func SyncInstance(cluster *Cluster, nodeName string, vmid uint64) error { - instance, err := cluster.GetInstance(nodeName, vmid) - if err != nil { - cluster.OK = false - return err - } +func (instance *Instance) Sync() error { + instance.lock.Lock() + defer instance.lock.Unlock() - err = instance.Build() + cluster := instance.node.cluster + + cluster.lock.Lock() + defer cluster.lock.Unlock() + + err := instance.Build() if err != nil { cluster.OK = false return err @@ -353,17 +373,24 @@ func SyncInstance(cluster *Cluster, nodeName string, vmid uint64) error { // hard sync instance // returns error if the instance could not be reached +// caller MUST lock cluster BEFORE entering func (instance *Instance) Build() error { + // check to confirm cluster is locked before making any changes + if !instance.lock.IsLocked() { + return fmt.Errorf("instance was not locked before calling instance.Build()") + } + + start := time.Now() + vmid := instance.VMID instancetype := instance.Type node := instance.node - var i *Instance var err error switch instancetype { case VM: - i, err = node.VirtualMachine(vmid) + err = node.VirtualMachine(instance, vmid) case CT: - i, err = node.Container(vmid) + err = node.Container(instance, vmid) } @@ -372,12 +399,6 @@ func (instance *Instance) Build() error { return fmt.Errorf("error retrieving %s.%d: %s", node.Name, vmid, err.Error()) } - *instance = *i // copiues lock vlaue which should be unlocked - - // aquire lock on instance, release on return - instance.lock.Lock() - defer instance.lock.Unlock() - wg, _ := errgroup.WithContext(context.Background()) if err != nil && node.Instances[vmid] != nil { // node is unreachable and did exist previously @@ -433,6 +454,8 @@ func (instance *Instance) Build() error { instance.node = node + log.Printf("[INFO] built instance %s.%d in %d ms", instance.node.Name, instance.VMID, time.Since(start).Milliseconds()) + return nil } diff --git a/app/proxmox.go b/app/proxmox.go index e93a667..66d920d 100644 --- a/app/proxmox.go +++ b/app/proxmox.go @@ -73,8 +73,8 @@ func (pve ProxmoxClient) Nodes() ([]string, error) { } // Gets a Node's resources but does not recursively expand instances -func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { - host := Node{} +func (pve ProxmoxClient) Node(host *Node, nodeName string) error { + //host := Node{} host.Devices = make(map[DeviceBus]*Device) host.Instances = make(map[InstanceID]*Instance) host.storage = make(map[string][]*proxmox.StorageContent) @@ -82,14 +82,14 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { // get pve node node, err := pve.client.Node(context.Background(), nodeName) if err != nil { - return &host, err + return err } // get pve node devices devices := []PVEDevice{} err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/hardware/pci", nodeName), &devices) if err != nil { - return &host, err + return err } // populate host devices from pve node devices for _, device := range devices { @@ -119,7 +119,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { proctypes := []PVEProctype{} err = pve.client.Get(context.Background(), fmt.Sprintf("/nodes/%s/capabilities/qemu/cpu", nodeName), &proctypes) if err != nil { - return &host, err + return err } // populate host proctypes from pve node proctypes for _, proctype := range proctypes { @@ -130,7 +130,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { wg, _ := errgroup.WithContext(context.Background()) storages, err := node.Storages(context.Background()) if err != nil { - return &host, err + return err } // populate host storage content from pve node storages for _, storage := range storages { @@ -147,7 +147,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { } err = wg.Wait() if err != nil { - return &host, err + return err } // set host basic values @@ -157,7 +157,7 @@ func (pve ProxmoxClient) Node(nodeName string) (*Node, error) { host.Swap = node.Swap.Total host.pvenode = node - return &host, err + return err } // Get all VM IDs on specified host @@ -174,11 +174,10 @@ func (host *Node) VirtualMachines() ([]uint, error) { } // Get a VM's CPU, Memory but does not recursively link Devices, Disks, Drives, Nets -func (host *Node) VirtualMachine(VMID InstanceID) (*Instance, error) { - instance := Instance{} +func (host *Node) VirtualMachine(instance *Instance, VMID InstanceID) error { vm, err := host.pvenode.VirtualMachine(context.Background(), int(VMID)) if err != nil { - return &instance, err + return err } config := vm.VirtualMachineConfig @@ -198,7 +197,7 @@ func (host *Node) VirtualMachine(VMID InstanceID) (*Instance, error) { instance.Nets = make(map[NetID]*Net) instance.Devices = make(map[DeviceID]*Device) - return &instance, nil + return nil } func MergeVMDisksAndUnused(vmc *proxmox.VirtualMachineConfig) map[string]string { @@ -221,11 +220,10 @@ func (host *Node) Containers() ([]uint, error) { } // Get a CT's CPU, Memory, Swap but does not recursively link Devices, Disks, Drives, Nets -func (host *Node) Container(VMID InstanceID) (*Instance, error) { - instance := Instance{} +func (host *Node) Container(instance *Instance, VMID InstanceID) error { ct, err := host.pvenode.Container(context.Background(), int(VMID)) if err != nil { - return &instance, err + return err } config := ct.ContainerConfig @@ -243,7 +241,7 @@ func (host *Node) Container(VMID InstanceID) (*Instance, error) { instance.Volumes = make(map[VolumeID]*Volume) instance.Nets = make(map[NetID]*Net) - return &instance, nil + return nil } func MergeCTDisksAndUnused(cc *proxmox.ContainerConfig) map[string]string { diff --git a/app/types.go b/app/types.go index aa58f43..183a20c 100644 --- a/app/types.go +++ b/app/types.go @@ -14,7 +14,7 @@ import ( // override Nodes map to new custom Node type type Cluster struct { paas.Cluster - lock sync.Mutex + lock MutexWithCheck pve ProxmoxClient NodesLock sync.Mutex Nodes map[string]*Node `json:"nodes"` @@ -24,8 +24,8 @@ type Cluster struct { // add mutex and pve api Node object // override Instances map to new custom Instance type type Node struct { - lock sync.Mutex paas.Node + lock MutexWithCheck InstancesLock sync.Mutex // lock for Instances map Instances map[InstanceID]*Instance `json:"instances"` pvenode *proxmox.Node @@ -41,8 +41,8 @@ const CT InstanceType = paas.CT // add mutex and various config objects type Instance struct { - lock sync.Mutex paas.Instance + lock MutexWithCheck VolumesLock sync.Mutex // lock for Volumes map NetsLock sync.Mutex // lock for Nets map DevicesLock sync.Mutex // lock for Devices map @@ -64,3 +64,32 @@ type Device = paas.Device type FunctionID = paas.FunctionID type Function = paas.Function type BootOrder = paas.BootOrder + +type MutexWithCheck struct { + ml sync.Mutex + v sync.Mutex + l bool +} + +func (m *MutexWithCheck) Lock() { + m.ml.Lock() + defer m.ml.Unlock() + + m.v.Lock() + m.l = true +} + +func (m *MutexWithCheck) Unlock() { + m.ml.Lock() + defer m.ml.Unlock() + + m.v.Unlock() + m.l = false +} + +func (m *MutexWithCheck) IsLocked() bool { + m.ml.Lock() + defer m.ml.Unlock() + + return m.l +}