From 027ed2dce63dd1565c0d5c6238b09be793eb7a83 Mon Sep 17 00:00:00 2001 From: Arthur Lu Date: Fri, 10 Jul 2026 20:39:58 +0000 Subject: [PATCH] improved synchronization mechanism using rwmutexes --- app/app.go | 28 +++++---- app/model.go | 163 +++++++++++++++++++++++++-------------------------- app/types.go | 41 ++++--------- 3 files changed, 110 insertions(+), 122 deletions(-) diff --git a/app/app.go b/app/app.go index 24becac..40886a6 100644 --- a/app/app.go +++ b/app/app.go @@ -69,26 +69,26 @@ func Run() { }) router.GET("/", func(c *gin.Context) { - v, err := cluster.Get() + callback := Callback{} + v, err := GetCluster(&callback, &cluster) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) - return } else { c.JSON(http.StatusOK, gin.H{"cluster": v}) - return } + callback.Invoke() }) router.GET("/nodes/:node", func(c *gin.Context) { nodeid := c.Param("node") - node, err := cluster.GetNode(nodeid) + callback := Callback{} + node, err := GetNode(&callback, &cluster, nodeid) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) - return } else { c.JSON(http.StatusOK, gin.H{"node": node}) - return } + callback.Invoke() }) router.GET("/nodes/:node/instances/:vmid", func(c *gin.Context) { @@ -98,14 +98,14 @@ func Run() { c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("%s could not be converted to vmid (uint)", c.Param("instance"))}) return } - instance, err := cluster.GetInstance(nodeid, vmid) + callback := Callback{} + instance, err := GetInstance(&callback, &cluster, nodeid, vmid) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) - return } else { c.JSON(http.StatusOK, gin.H{"instance": instance}) - return } + callback.Invoke() }) router.POST("/sync", func(c *gin.Context) { @@ -125,12 +125,15 @@ func Run() { log.Printf("[INFO] starting %s sync\n", nodeName) - node, err := cluster.GetNode(nodeName) + callback := Callback{} + node, err := GetNode(&callback, &cluster, nodeName) if err != nil { log.Printf("[ERR ] failed to sync %s: %s", nodeName, err.Error()) c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + callback.Invoke() return } + callback.Invoke() err = node.Sync() if err != nil { @@ -150,12 +153,15 @@ func Run() { log.Printf("[INFO] starting %s.%d sync\n", nodeName, vmid) - instance, err := cluster.GetInstance(nodeName, vmid) + callback := Callback{} + instance, err := GetInstance(&callback, &cluster, 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()}) + callback.Invoke() return } + callback.Invoke() err = instance.Sync() if err != nil { diff --git a/app/model.go b/app/model.go index ada2cf4..4308bb0 100644 --- a/app/model.go +++ b/app/model.go @@ -16,20 +16,40 @@ func (cluster *Cluster) Init(pve ProxmoxClient) { cluster.pve = pve } -func (cluster *Cluster) Get() (*Cluster, error) { - // aquire cluster lock - cluster.lock.Lock() - defer cluster.lock.Unlock() - if cluster.OK { - return cluster, nil - } else { +func GetCluster(callback *Callback, cluster *Cluster) (*Cluster, error) { + // obtain read lock on whole cluster tree + cluster.lock.RLock() + callback.Add(cluster.lock.RUnlock) + + if !cluster.OK { return nil, fmt.Errorf("cluster state is invalid") } + + for _, node := range cluster.Nodes { + node.lock.RLock() + defer node.lock.RUnlock() + callback.Add(node.lock.RUnlock) + for _, instance := range node.Instances { + instance.lock.RLock() + callback.Add(instance.lock.RUnlock) + } + } + + return cluster, nil } func (cluster *Cluster) Sync() error { + // obtain write lock on whole tree cluster.lock.Lock() defer cluster.lock.Unlock() + for _, n := range cluster.Nodes { + n.lock.Lock() + defer n.lock.Unlock() + for _, i := range n.Instances { + i.lock.Lock() + defer i.lock.Unlock() + } + } err := cluster.Build() if err != nil { @@ -44,17 +64,11 @@ func (cluster *Cluster) Sync() error { } cluster.OK = true + return nil } -// hard sync cluster -// caller MUST lock cluster BEFORE entering func (cluster *Cluster) Build() error { - // 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) @@ -72,13 +86,11 @@ func (cluster *Cluster) Build() error { wg.Go(func() error { // rebuild node node := Node{} - node.lock.Lock() node.Name = nodeName node.cluster = cluster err := node.Build() - //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) } @@ -87,8 +99,6 @@ func (cluster *Cluster) Build() error { cluster.Nodes[nodeName] = &node cluster.NodesLock.Unlock() - node.lock.Unlock() - return err }) } @@ -104,14 +114,7 @@ func (cluster *Cluster) Build() error { return nil } -// resolve membership of instances to pools -// caller MUST lock cluster BEFORE entering func (cluster *Cluster) ResolvePoolMembership() error { - // 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 @@ -155,11 +158,9 @@ func (cluster *Cluster) ResolvePoolMembership() error { return nil } -// get a node in the cluster -func (cluster *Cluster) GetNode(nodeName string) (*Node, error) { - // aquire cluster lock - cluster.lock.Lock() - defer cluster.lock.Unlock() +func GetNode(callback *Callback, cluster *Cluster, nodeName string) (*Node, error) { + cluster.lock.RLock() + callback.Add(cluster.lock.RUnlock) if !cluster.OK { return nil, fmt.Errorf("cluster state is invalid") @@ -169,23 +170,37 @@ func (cluster *Cluster) GetNode(nodeName string) (*Node, error) { node, ok := cluster.Nodes[nodeName] if !ok { return nil, fmt.Errorf("%s not in cluster", nodeName) - } else { - // aquire node lock to wait in case of a concurrent write - node.lock.Lock() - defer node.lock.Unlock() - - return node, nil } + + node.lock.RLock() + callback.Add(node.lock.RUnlock) + for _, instance := range node.Instances { + instance.lock.RLock() + callback.Add(instance.lock.RUnlock) + } + + return node, nil } func (node *Node) Sync() error { - node.lock.Lock() - defer node.lock.Unlock() - cluster := node.cluster - cluster.lock.Lock() + // obtain write lock on this node subtree, and obtain write lock all instances for ResolvePoolMembership, obtain read lock on everything else + cluster.lock.Lock() // todo relax defer cluster.lock.Unlock() + for _, n := range cluster.Nodes { + if n != node { // if other node, read lock + n.lock.RLock() + defer n.lock.RUnlock() + } else { // if this node, write lock + n.lock.Lock() + defer n.lock.Unlock() + } + for _, i := range n.Instances { // need write lock on all instances + i.lock.Lock() + defer i.lock.Unlock() + } + } err := node.Build() if err != nil { @@ -203,15 +218,7 @@ func (node *Node) Sync() error { return nil } -// 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 @@ -253,9 +260,7 @@ func (node *Node) Build() error { log.Printf("[ERR ] error encountered while syncing vm %s.%d: %s", nodeName, vmid, err) } - node.InstancesLock.Lock() node.Instances[instanceID] = &instance - node.InstancesLock.Unlock() instance.lock.Unlock() @@ -314,10 +319,10 @@ func (node *Node) Build() error { return nil } -func (cluster *Cluster) GetInstance(nodeName string, vmid uint64) (*Instance, error) { - // aquire cluster lock - cluster.lock.Lock() - defer cluster.lock.Unlock() +func GetInstance(callback *Callback, cluster *Cluster, nodeName string, vmid uint64) (*Instance, error) { + // obtain read lock on whole tree, but do not hold cluster and node read lock indefinitely + cluster.lock.RLock() + callback.Add(cluster.lock.RUnlock) if !cluster.OK { return nil, fmt.Errorf("cluster state is invalid") @@ -328,32 +333,34 @@ func (cluster *Cluster) GetInstance(nodeName string, vmid uint64) (*Instance, er if !ok { return nil, fmt.Errorf("%s not in cluster", nodeName) } - - // aquire node lock - node.lock.Lock() - defer node.lock.Unlock() + node.lock.RLock() + callback.Add(node.lock.RUnlock) // get instance instance, ok := node.Instances[InstanceID(vmid)] if !ok { return nil, fmt.Errorf("vmid %d not in node %s", vmid, node.Name) } else { - // aquire instance lock to wait in case of a concurrent write - instance.lock.Lock() - defer instance.lock.Unlock() - + instance.lock.RLock() + callback.Add(instance.lock.RUnlock) return instance, nil } } func (instance *Instance) Sync() error { - instance.lock.Lock() - defer instance.lock.Unlock() - cluster := instance.node.cluster - cluster.lock.Lock() + // obtain write lock on this instance subtree, and obtain write lock all instances for ResolvePoolMembership, obtain read lock on everything else + cluster.lock.Lock() // todo relax defer cluster.lock.Unlock() + for _, n := range cluster.Nodes { + n.lock.RLock() + defer n.lock.RUnlock() + for _, i := range n.Instances { // need write lock on all instances + i.lock.Lock() + defer i.lock.Unlock() + } + } err := instance.Build() if err != nil { @@ -371,15 +378,7 @@ func (instance *Instance) Sync() error { return nil } -// 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 @@ -410,7 +409,7 @@ func (instance *Instance) Build() error { for volid := range instance.configDisks { wg.Go(func() error { - err = instance.RebuildVolume(node, volid) + err = instance.BuildVolume(node, volid) if err != nil { log.Printf("[ERR ] error rebuilding volume %s: %s", volid, err) } @@ -420,7 +419,7 @@ func (instance *Instance) Build() error { for netid := range instance.configNets { wg.Go(func() error { - err = instance.RebuildNet(node, netid) + err = instance.BuildNet(node, netid) if err != nil { log.Printf("[ERR ] error rebuilding net %s: %s", netid, err) return err @@ -431,7 +430,7 @@ func (instance *Instance) Build() error { for deviceid := range instance.configHostPCIs { wg.Go(func() error { - err = instance.RebuildDevice(node, deviceid) + err = instance.BuildDevice(node, deviceid) if err != nil { log.Printf("[ERR ] error rebuilding pci %s: %s", deviceid, err) } @@ -445,7 +444,7 @@ func (instance *Instance) Build() error { } if instance.Type == VM { - err = instance.RebuildBoot(node) + err = instance.BuildBoot(node) if err != nil { log.Printf("[ERR ] error rebuilding boot: %s", err) return err @@ -459,7 +458,7 @@ func (instance *Instance) Build() error { return nil } -func (instance *Instance) RebuildVolume(node *Node, volid string) error { +func (instance *Instance) BuildVolume(node *Node, volid string) error { volumeDataString := instance.configDisks[volid] volume, err := GetVolumeInfo(node, volumeDataString) @@ -478,7 +477,7 @@ func (instance *Instance) RebuildVolume(node *Node, volid string) error { return nil } -func (instance *Instance) RebuildNet(node *Node, netid string) error { +func (instance *Instance) BuildNet(node *Node, netid string) error { net := instance.configNets[netid] netinfo, err := GetNetInfo(net) @@ -494,7 +493,7 @@ func (instance *Instance) RebuildNet(node *Node, netid string) error { return nil } -func (instance *Instance) RebuildDevice(node *Node, deviceid string) error { +func (instance *Instance) BuildDevice(node *Node, deviceid string) error { instanceDevice, ok := instance.configHostPCIs[deviceid] if !ok { // if device does not exist log.Printf("[WARN] %s not found in devices on node %s", deviceid, node.Name) @@ -520,7 +519,7 @@ func (instance *Instance) RebuildDevice(node *Node, deviceid string) error { return nil } -func (instance *Instance) RebuildBoot(node *Node) error { +func (instance *Instance) BuildBoot(node *Node) error { instance.Boot = BootOrder{} eligibleBoot := map[string]bool{} diff --git a/app/types.go b/app/types.go index 9c0d136..88a07e9 100644 --- a/app/types.go +++ b/app/types.go @@ -14,9 +14,9 @@ import ( // override Nodes map to new custom Node type type Cluster struct { paas.Cluster - lock MutexWithCheck + lock sync.RWMutex pve ProxmoxClient - NodesLock sync.Mutex + NodesLock sync.Mutex // lock for Nodes map Nodes map[string]*Node `json:"nodes"` OK bool } @@ -25,7 +25,7 @@ type Cluster struct { // override Instances map to new custom Instance type type Node struct { paas.Node - lock MutexWithCheck + lock sync.RWMutex InstancesLock sync.Mutex // lock for Instances map Instances map[InstanceID]*Instance `json:"instances"` pvenode *proxmox.Node @@ -42,7 +42,7 @@ const CT InstanceType = paas.CT // add mutex and various config objects type Instance struct { paas.Instance - lock MutexWithCheck + lock sync.RWMutex VolumesLock sync.Mutex // lock for Volumes map NetsLock sync.Mutex // lock for Nets map DevicesLock sync.Mutex // lock for Devices map @@ -65,33 +65,16 @@ type FunctionID = paas.FunctionID type Function = paas.Function type BootOrder = paas.BootOrder -// wraps a mutex and mutex state value under a single mutex interface with support for IsLocked check -// highly cursed -type MutexWithCheck struct { - ml sync.Mutex // wrapper mutex - v sync.Mutex // wrapped mutex - l bool // wrapped state +type Callback struct { + f []func() } -func (m *MutexWithCheck) Lock() { - m.ml.Lock() - defer m.ml.Unlock() - - m.v.Lock() - m.l = true +func (c *Callback) Add(nf func()) { + c.f = append(c.f, nf) } -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 +func (c *Callback) Invoke() { + for _, f := range c.f { + f() + } }