package main // Where disk-load learns the data items it measures (novox/hq ADR 0577 clause 2): the node-engine's state data // (delivery.EngineDataBucket), one key `.` per module placed on this node that declares data items, as the // engine composed them. disk-load watches its own node's keys and composes nothing: it reads no resolved values, no // manifests and no placement, so a manifest its mesh-sdk cannot judge never silences it, and a round reads one watched // map instead of several hundred keys over the tunnel (issue 1955). import ( "context" "errors" "fmt" "sync" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) // Watched is this node's data items as the engine's state data holds them, by module, kept from a watch of its keys. type Watched struct { Node string mu sync.Mutex byKey map[string]string // key to its module, for a delete that carries no value byModule map[string][]catalogue.PlacedDataItem ready bool } // ErrNotWatchedYet is a round asked before the watch handed over the engine's current keys: no item is known yet, and // none is taken for gone. var ErrNotWatchedYet = errors.New("the node-engine's state " + delivery.EngineDataBucket + " is not watched yet: " + "no data item of this machine is known") // Take keeps one change of a key and answers the module it is of and whether its items changed. func (w *Watched) Take(ch stdio.StateChange) (string, bool, error) { w.mu.Lock() defer w.mu.Unlock() if w.byKey == nil { w.byKey, w.byModule = map[string]string{}, map[string][]catalogue.PlacedDataItem{} } if ch.Op != stdio.OpPut { module, known := w.byKey[ch.Key] if !known { return "", false, nil } delete(w.byKey, ch.Key) delete(w.byModule, module) return module, true, nil } d, err := delivery.DecodeEngineData(ch.Value) if err != nil { return "", false, fmt.Errorf("%s: %w", ch.Key, err) } if d.Node != w.Node { return "", false, fmt.Errorf("%s names the machine %q, and this is %s", ch.Key, d.Node, w.Node) } for _, it := range d.Items { if it.Module != d.Module { return "", false, fmt.Errorf("%s holds an item of %q, and is %s's", ch.Key, it.Module, d.Module) } } changed := !sameItems(w.byModule[d.Module], d.Items) w.byKey[ch.Key] = d.Module w.byModule[d.Module] = d.Items return d.Module, changed, nil } // Ready says the watch handed over every current key: from now on an item missing is an item gone. func (w *Watched) Ready() { w.mu.Lock() w.ready = true w.mu.Unlock() } // All is every module's items, as the Keeper reads them; refused until the watch is ready. func (w *Watched) All(context.Context) (map[string][]catalogue.PlacedDataItem, error) { w.mu.Lock() defer w.mu.Unlock() if !w.ready { return nil, ErrNotWatchedYet } out := make(map[string][]catalogue.PlacedDataItem, len(w.byModule)) for m, its := range w.byModule { out[m] = append([]catalogue.PlacedDataItem(nil), its...) } return out, nil } func sameItems(a, b []catalogue.PlacedDataItem) bool { if len(a) != len(b) { return false } for i := range a { if a[i] != b[i] { return false } } return true } // Follow watches this node's keys of the engine's state data, and asks the keeper to measure a module whose items changed // after the watch began; it returns once the current keys were handed over (stdio.KeptState.Follow). func (w *Watched) Follow(k *Keeper) error { pattern, err := delivery.EngineDataNodeKeys(w.Node) if err != nil { return err } stdio.State(delivery.EngineDataRead).Follow(pattern, func(ch stdio.StateChange) error { module, changed, err := w.Take(ch) if err != nil || !changed || ch.Current { return err } k.Due(module) return nil }) w.Ready() // The handed-over set is complete now: a module whose key went while the watch was down is retired once. k.RetireUnkeyed(context.Background()) // A round refused while the watch caught up runs now, not at the next hour (the delivery's readings within minutes). go k.RoundIfRefused(context.Background()) return nil }