diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill.go new file mode 100644 index 00000000..a3b79267 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "fmt" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// A store filling is said before it is full (design 47 G §181, in the assigner's part): the assignment graph revisions' object store +// says `filling` of the seat mesh-wants when it passes 80 % of its cap, and `full` when the next revision does not fit: the assigner then keeps +// its last revision and writes no new want, and says `room` when the store is back below 80 % and goes on. Nothing is dropped to make room. + +// fillOf is the store's fill now: its bytes and cap, when the bucket says them and has a cap. +func (g *graphKeeper) fillOf(ctx context.Context) (used, capacity int64, ok bool) { + f, is := g.objects.(assigngraph.Filler) + if !is { + return 0, 0, false + } + used, capacity, err := f.Fill(ctx) + return used, capacity, err == nil && capacity > 0 +} + +func (p *passer) sayFill(event, about string, used, capacity int64, refused string) { + at := p.clock().UTC() + body := meshwants.Fill{Store: assigngraph.RevisionsObjects, Bytes: used, Cap: capacity, Refused: refused, At: &at} + if err := p.wantsSay(event, body, event+"-"+about); err != nil { + p.logf("%s of %s is not said: %v", event, assigngraph.RevisionsObjects, err) + } +} + +// fillAtStart judges the store once at the process's first pass, when no event is pending: a store past 80 % says filling (full when at +// its cap), so a fill reached while the process was down is said too. +func (p *passer) fillAtStart(ctx context.Context) { + g := p.graph + used, capacity, ok := g.fillOf(ctx) + if !ok { + return + } + switch { + case used >= capacity: + g.full = true + p.sayFill(meshwants.EventFull, "start", used, capacity, "the next revision") + case used*5 >= capacity*4: + p.sayFill(meshwants.EventFilling, "start", used, capacity, "") + } +} + +// fullBefore says the next revision does not fit its store, and keeps nothing: it says `full` once, and `room` once the store is below +// 80 % again. It answers true while the store is full. +func (p *passer) fullBefore(ctx context.Context, c assigngraph.Computed) bool { + g := p.graph + used, capacity, ok := g.fillOf(ctx) + if !ok { + return false + } + if g.full && used*5 < capacity*4 { + g.full = false + p.sayFill(meshwants.EventRoom, fmt.Sprint(used), used, capacity, "") + } + if used+assigngraph.NewBytes(ctx, g.objects, c) <= capacity { + g.full = false + return false + } + if !g.full { + g.full = true + p.sayFill(meshwants.EventFull, fmt.Sprint(used), used, capacity, "the next revision") + } + p.logf("the assignment graph revision is not kept: %s is full (%d of %d bytes), the last revision stays and no want is written until there is room", assigngraph.RevisionsObjects, used, capacity) + return true +} + +// fillingAfter says `filling` when keeping a revision took the store past 80 % of its cap. +func (p *passer) fillingAfter(ctx context.Context, before int64) { + g := p.graph + used, capacity, ok := g.fillOf(ctx) + if ok && before*5 < capacity*4 && used*5 >= capacity*4 { + p.sayFill(meshwants.EventFilling, fmt.Sprint(used), used, capacity, "") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill_test.go new file mode 100644 index 00000000..af05f547 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphfill_test.go @@ -0,0 +1,83 @@ +package main + +import ( + "context" + "testing" + "time" + + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// filled is an object bucket with a fill: the bytes it holds count beside what is stored. +type filled struct { + objs map[string][]byte + base int64 + cap int64 +} + +func (f *filled) Put(_ context.Context, name string, data []byte) error { + f.objs[name] = data + return nil +} +func (f *filled) Get(_ context.Context, name string) ([]byte, error) { + if d, ok := f.objs[name]; ok { + return d, nil + } + return nil, assigngraph.ErrNoObject +} +func (f *filled) Names(context.Context) ([]string, error) { + out := []string{} + for n := range f.objs { + out = append(out, n) + } + return out, nil +} +func (f *filled) Fill(context.Context) (int64, int64, error) { + used := f.base + for _, d := range f.objs { + used += int64(len(d)) + } + return used, f.cap, nil +} + +// Design 47 G §181 (AS13): the assignment graph revisions' store says filling past 80 % of its cap, refuses a revision that does not fit +// with `full` (the last revision stays, no want is written), and says `room` below 80 % again; at the start a store already past 80 % is said. +func TestTheAssignmentGraphStoreSaysFillingFullAndRoom(t *testing.T) { + ctx := context.Background() + p := newPasser(kvSet{}, time.Now()) + var events []string + p.publishWants = func(event string, body any, id string) error { + if _, ok := body.(meshwants.Fill); ok { + events = append(events, event) + } + return nil + } + o := &filled{objs: map[string][]byte{}, base: 500, cap: 1000} + p.graph = &graphKeeper{objects: o, halves: func() map[string]mcc.Half { return nil }} + p.keepGraph(ctx) // the first pass judges at the start: 500 of 1000 is below 80 % + if len(events) != 0 { + t.Fatalf("events %v", events) + } + o.base = 700 + p.graph.started = false + p.keepGraph(ctx) + if len(events) != 1 || events[0] != meshwants.EventFilling { + t.Fatalf("at the start past 80 %%: %v", events) + } + c := assigngraph.Computed{Revision: meshwants.AssignmentRevision{Entries: []meshwants.NodeEntryRef{}}} + o.base = 990 + if !p.fullBefore(ctx, c) || !p.graph.full || events[len(events)-1] != meshwants.EventFull { + t.Fatalf("a revision that does not fit: events %v", events) + } + p.fullBefore(ctx, c) + if n := len(events); n != 2 { + t.Errorf("full is said once: %v", events) + } + o.base = 100 + if p.fullBefore(ctx, c) || p.graph.full || events[len(events)-1] != meshwants.EventRoom { + t.Errorf("room: events %v", events) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go index e3b2e22b..77980e47 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -36,6 +36,8 @@ type graphKeeper struct { said string // cycled is the cycles last said (assigngraph.Cycles). cycled string + // full says the next revision did not fit the store, so no want is written (graphfill.go); started says the first pass judged the fill. + full, started bool // pins removes a revision that ceased to be current once nothing pins it (pins.go); nil keeps every revision. pins *pinKeeper // computed is the revision computed by the last pass, with its entries: the edges the wants wait on (actingwaits.go). @@ -68,9 +70,17 @@ func (p *passer) keepGraph(ctx context.Context) { p.logf("the assignment graph holds a cycle among the waits within one step, and its entries wait for each other: %s", cycles) } } + if !g.started { + g.started = true + p.fillAtStart(ctx) + } if g.has && g.ok && assigngraph.Same(g.last, c.Revision) { return } + used, _, _ := g.fillOf(ctx) + if p.fullBefore(ctx, c) { + return + } rev, fresh, err := assigngraph.Keep(ctx, g.objects, c, meshwants.EventAssignmentRevision, p.clock()) if err != nil { p.logf("the assignment graph revision is not kept, so no want is said as wanted until it is: %v", err) @@ -86,6 +96,7 @@ func (p *passer) keepGraph(ctx context.Context) { if !fresh { return } + p.fillingAfter(ctx, used) if g.pins != nil { g.pins.judge() // the revision before ceased to be current } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go index 73540150..e1833237 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go @@ -99,3 +99,20 @@ func (b *busObjects) Delete(ctx context.Context, name string) error { } return nil } + +// Fill is the bytes the bucket holds and its cap (assigngraph.Filler). +func (b *busObjects) Fill(ctx context.Context) (int64, int64, error) { + o, err := b.open(ctx) + if err != nil { + return 0, 0, err + } + st, err := o.Status(ctx) + if err != nil { + return 0, 0, err + } + info, err := b.js.Stream(ctx, "OBJ_"+st.Bucket()) + if err != nil { + return int64(st.Size()), 0, err + } + return int64(st.Size()), info.CachedInfo().Config.MaxBytes, nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index e0d97613..aad8b9c1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -205,7 +205,9 @@ func (p *passer) clock() time.Time { func (p *passer) pass(ctx context.Context) { p.assignmentsSeat(ctx) p.keepGraph(ctx) - p.wants(ctx) + if p.graph == nil || !p.graph.full { // a full assignment graph store keeps the last revision and writes no new want (design 47 G §181) + p.wants(ctx) + } p.puts(ctx) p.nodeWants(ctx) p.backups(ctx) diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/keep.go b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go index 7ce8600e..c97b5e9f 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/keep.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go @@ -149,3 +149,29 @@ func Event(rev meshwants.AssignmentRevision) meshwants.AssignmentRevisionEvent { return meshwants.AssignmentRevisionEvent{V: rev.V, Revision: rev.Revision, Kind: rev.Kind, Cause: rev.Cause, Group: rev.Group, Heads: rev.Heads, Catalogs: rev.Catalogs, At: rev.At} } + +// Filler is an object bucket that says its fill: the bytes used and its cap in bytes (0: no cap known). +type Filler interface { + Fill(ctx context.Context) (used, capacity int64, err error) +} + +// NewBytes is what keeping c next would add to the bucket: each entry object not held yet and the list (design 47 G §181). +func NewBytes(ctx context.Context, o Objects, c Computed) int64 { + names, err := o.Names(ctx) + if err != nil { + return 0 + } + held := make(map[string]bool, len(names)) + for _, n := range names { + held[n] = true + } + var size int64 + for _, ref := range c.Revision.Entries { + if e, ok := c.Entries[ref.Digest]; ok && !held[EntryObject(ref.Digest)] { + data, _ := json.Marshal(e) + size += int64(len(data)) + } + } + list, _ := json.Marshal(c.Revision) + return size + int64(len(list)) +}