1// Package paged reads linked sources in place: Parquet files and SQLite2// tables or queries too big for any grid, linked as tables on tabs of3// their own (the engine's source.go). Its Host answers what a4// workbook's formulas and pivot tables ask of them (sheet.SourceHost)5// by streaming the sources in the background, keeping the answers until6// a source's file changes; its Pages are the rows a source's tab shows,7// read a page at a time as it scrolls.8//9// The Host and Pages belong to the goroutine that owns the workbook (the10// UI's). What reads files runs as Jobs on other goroutines, over what11// the Job took with it, and hands its result back to be stored, so the12// owner never waits on a file.13package paged1415import (16 "context"17 "os"18 "strings"19 "time"2021 "src.rest/cat/012/internal/fileio"22 "src.rest/cat/012/internal/sheet"23)2425// Linked is a source as the workbook links it, its file resolved.26type Linked struct {27 Name string28 Spec fileio.SourceSpec29}3031// Opener opens what a spec names: fileio.OpenSource, or a stand-in.32type Opener func(ctx context.Context, spec fileio.SourceSpec) (fileio.Source, error)3334// Host answers a workbook's questions about its linked sources.35type Host struct {36 // Budget is how many cells a function that holds what it reads may37 // read of a source, and a range read whole may have: max-cells.38 Budget int39 // Open opens a source; fileio.OpenSource unless replaced.40 Open Opener41 // Debounce is how long a changed file must hold still before it's42 // read again.43 Debounce time.Duration44 // Now is the clock; tests replace it.45 Now func() time.Time4647 srcs map[string]*source // by name key48 answers map[string]answer // by question key49 asked map[string]bool // questions queued or running50 fresh []*Job // jobs not yet handed out51}5253// source is a linked source as the host has it.54type source struct {55 name string56 spec fileio.SourceSpec57 h fileio.Source // nil until opened58 shape sheet.SourceShape59 err string60 gen int // counts openings; answers of earlier ones are stale61 opening bool // a job is opening it62 read stamp // the file when opened63 seen stamp // the file when last polled, different from read64 seenAt time.Time65}6667// stamp is what tells a file changed.68type stamp struct {69 size int6470 mod time.Time71}7273// answer is a stored answer, with the generation of each source it read.74type answer struct {75 a sheet.SourceAnswer76 gens map[string]int77}7879// NewHost is a host with budget max-cells.80func NewHost(budget int) *Host {81 return &Host{Budget: budget, Open: fileio.OpenSource, Debounce: 300 * time.Millisecond, Now: time.Now,82 srcs: map[string]*source{}, answers: map[string]answer{}, asked: map[string]bool{}}83}8485func key(name string) string { return strings.ToLower(name) }8687// Link makes the host's sources the workbook's: new ones and those88// reading something else are opened, those gone are closed. It returns89// whether anything changed.90func (h *Host) Link(list []Linked) bool {91 changed := false92 keep := map[string]bool{}93 for _, l := range list {94 k := key(l.Name)95 keep[k] = true96 s := h.srcs[k]97 if s != nil && s.spec == l.Spec {98 continue99 }100 if s != nil && s.h != nil {101 s.h.Close()102 }103 gen := 0104 if s != nil {105 gen = s.gen + 1106 }107 s = &source{name: l.Name, spec: l.Spec, gen: gen}108 h.srcs[k] = s109 h.reopen(s)110 changed = true111 }112 for k, s := range h.srcs {113 if !keep[k] {114 if s.h != nil {115 s.h.Close()116 }117 delete(h.srcs, k)118 changed = true119 }120 }121 return changed122}123124// reopen queues a job opening s.125func (h *Host) reopen(s *source) {126 s.opening = true127 h.fresh = append(h.fresh, &Job{kind: jobOpen, name: s.name, spec: s.spec, gen: s.gen, open: h.Open})128}129130// Reload reads the source named name again, as when its file changed:131// what was worked out of it is worked out again.132func (h *Host) Reload(name string) {133 s := h.srcs[key(name)]134 if s == nil || s.opening {135 return136 }137 if s.h != nil {138 s.h.Close()139 s.h = nil140 }141 s.gen++142 h.reopen(s)143}144145// Close lets every source go.146func (h *Host) Close() {147 for _, s := range h.srcs {148 if s.h != nil {149 s.h.Close()150 }151 }152 clear(h.srcs)153}154155// Shape is what the host found of the source named name: its shape, or156// why it can't be read, and whether it has been opened.157func (h *Host) Shape(name string) (sheet.SourceShape, string, bool) {158 s := h.srcs[key(name)]159 if s == nil || s.h == nil && s.err == "" {160 return sheet.SourceShape{}, "", false161 }162 return s.shape, s.err, true163}164165// Source is the opened source named name, for its tab's pages.166func (h *Host) Source(name string) (fileio.Source, int, bool) {167 s := h.srcs[key(name)]168 if s == nil || s.h == nil {169 return nil, 0, false170 }171 return s.h, s.gen, true172}173174// Answer returns the answer to q, or false while it isn't known, queuing175// q to be worked out (sheet.SourceHost).176func (h *Host) Answer(q sheet.SourceQuestion) (sheet.SourceAnswer, bool) {177 k := q.Key()178 if a, ok := h.answers[k]; ok && h.current(a.gens) {179 return a.a, true180 }181 if h.asked[k] {182 return sheet.SourceAnswer{}, false183 }184 j := &Job{kind: jobAnswer, q: q, key: k, budget: h.Budget, srcs: map[string]snapshot{}}185 for _, name := range q.Sources() {186 s := h.srcs[key(name)]187 switch {188 case s == nil:189 return sheet.SourceAnswer{V: sheet.ErrRef, Why: "There's no linked source named " + name}, true190 case s.err != "":191 return sheet.SourceAnswer{V: sheet.ErrRef, Why: s.name + ": " + s.err}, true192 case s.h == nil:193 return sheet.SourceAnswer{}, false // opening: asked again once it's open194 }195 j.srcs[key(name)] = snapshot{name: s.name, h: s.h, shape: s.shape, gen: s.gen, dir: s.spec.TempDir}196 }197 h.asked[k] = true198 h.fresh = append(h.fresh, j)199 return sheet.SourceAnswer{}, false200}201202// current reports whether answers read with the sources' generations203// gens are still the sources'.204func (h *Host) current(gens map[string]int) bool {205 for k, g := range gens {206 if s := h.srcs[k]; s == nil || s.gen != g {207 return false208 }209 }210 return true211}212213// Jobs hands out the jobs queued since the last call, to run in the214// background, each once.215func (h *Host) Jobs() []*Job {216 out := h.fresh217 h.fresh = nil218 return out219}220221// Pending reports whether jobs are queued or running.222func (h *Host) Pending() bool { return len(h.fresh) > 0 || len(h.asked) > 0 || h.opening() }223224func (h *Host) opening() bool {225 for _, s := range h.srcs {226 if s.opening {227 return true228 }229 }230 return false231}232233// Done is what storing a job's result tells the workbook.234type Done struct {235 // Keys are the questions answered.236 Keys []string237 // Opened names a source opened, or failing to, with its shape.238 Opened string239 Shape sheet.SourceShape240 Err string241}242243// Store keeps what the job j found, run and back on the owner's244// goroutine, and says what the workbook is to be told.245func (h *Host) Store(j *Job) Done {246 if j.kind == jobOpen {247 return h.storeOpen(j)248 }249 delete(h.asked, j.key)250 gens := map[string]int{}251 for k, snap := range j.srcs {252 gens[k] = snap.gen253 }254 if !h.current(gens) {255 return Done{} // a source changed while it ran: asked again256 }257 h.answers[j.key] = answer{a: j.ans, gens: gens}258 return Done{Keys: []string{j.key}}259}260261func (h *Host) storeOpen(j *Job) Done {262 s := h.srcs[key(j.name)]263 if s == nil || s.gen != j.gen {264 if j.h != nil {265 j.h.Close()266 }267 return Done{}268 }269 s.opening = false270 if s.h != nil && s.h != j.h {271 s.h.Close()272 }273 s.h, s.shape, s.err, s.read = j.h, j.shape, "", j.stamp274 if j.err != nil {275 s.h, s.shape, s.err = nil, sheet.SourceShape{}, describe(j.err)276 }277 h.forget(key(s.name))278 return Done{Opened: s.name, Shape: s.shape, Err: s.err}279}280281// forget drops the answers that read the source with key k.282func (h *Host) forget(k string) {283 for qk, a := range h.answers {284 if _, ok := a.gens[k]; ok {285 delete(h.answers, qk)286 }287 }288}289290// describe is an error as the context line says it.291func describe(err error) string {292 if os.IsNotExist(err) {293 return "the file isn't there"294 }295 return err.Error()296}297298// Poll looks at each source's file, once they're open, and queues the299// ones that changed and have held still for Debounce to be read again,300// reporting whether it queued any. A file that goes away is noticed the301// same way.302func (h *Host) Poll() bool {303 now, queued := h.Now(), false304 for _, s := range h.srcs {305 if s.opening {306 continue307 }308 st, err := os.Stat(s.spec.Path)309 cur := stamp{}310 if err == nil {311 cur = stamp{st.Size(), st.ModTime()}312 }313 switch {314 case cur == s.read:315 s.seen = stamp{}316 continue317 case cur != s.seen:318 s.seen, s.seenAt = cur, now319 continue320 case now.Sub(s.seenAt) < h.Debounce:321 continue322 }323 h.Reload(s.name) // jobs reading it fail, and their answers are dropped324 queued = true325 }326 return queued327}