Add disk-load's watched items, their test and its whole test, which the binary's ignore line hid
The ignore line `disk-load` matched the directory cmd/disk-load, so new files there were never added; it now names the built binary alone. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -0,0 +1,119 @@
|
||||
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 `<node>.<module>` 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()
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user