1package paged23import (4 "context"56 "src.rest/cat/012/internal/sheet"7)89// A cursor streams a range's rows for the book's Scan, a chunk at a10// time: a goroutine of its own scans the source and hands the rows over11// in batches, so the source's scan runs on while functions read what12// it read, and taking a row costs a slice index.1314// batchRows is how many rows a batch holds.15const batchRows = 40961617// cursor is a range's rows from where the last chunk ended.18type cursor struct {19 at sheet.Addr // where the next chunk starts20 width int21 batches chan []sheet.LiveCell22 free chan []sheet.LiveCell23 cancel context.CancelFunc24 err error // what the scan failed with, read once batches is closed25 cur []sheet.LiveCell26 i int // the next row of cur27 row []sheet.LiveCell // the row being read28 have bool29}3031// cursor is the cursor over the columns of part from from on: the one32// the last chunk left there, or a new one.33func (b *book) cursor(s snapshot, part sheet.Rect, from sheet.Addr) *cursor {34 k := cursorKey{key(s.name), part.From.Col, part.To.Col}35 if c := b.cursors[k]; c != nil {36 if c.at == from {37 return c38 }39 b.stopCursor(c)40 }41 cols := make([]int, 0, part.To.Col-part.From.Col+1)42 for c := part.From.Col; c <= part.To.Col; c++ {43 cols = append(cols, c)44 }45 c := startCursor(b.ctx, s, from, cols)46 if b.cursors == nil {47 b.cursors = map[cursorKey]*cursor{}48 }49 b.cursors[k] = c50 return c51}5253// startCursor scans the source s's columns cols from the sheet row of54// from on.55func startCursor(ctx context.Context, s snapshot, from sheet.Addr, cols []int) *cursor {56 ctx, cancel := context.WithCancel(ctx)57 c := &cursor{at: from, width: len(cols), batches: make(chan []sheet.LiveCell, 2), free: make(chan []sheet.LiveCell, 3), cancel: cancel}58 go func() {59 defer close(c.batches)60 batch := c.take()61 err := s.h.Scan(ctx, int64(from.Row-1), cols, func(_ int64, vals []sheet.LiveCell) bool {62 if batch = append(batch, vals...); len(batch) < batchRows*c.width {63 return true64 }65 if !c.send(ctx, batch) {66 return false67 }68 batch = c.take()69 return true70 })71 if len(batch) > 0 {72 c.send(ctx, batch)73 }74 if ctx.Err() == nil {75 c.err = err76 }77 }()78 return c79}8081// take is a batch to fill: one read already, or a new one.82func (c *cursor) take() []sheet.LiveCell {83 select {84 case b := <-c.free:85 return b[:0]86 default:87 return make([]sheet.LiveCell, 0, batchRows*c.width)88 }89}9091// send hands a batch over, reporting false once the cursor is stopped.92func (c *cursor) send(ctx context.Context, batch []sheet.LiveCell) bool {93 select {94 case c.batches <- batch:95 return true96 case <-ctx.Done():97 return false98 }99}100101// next is the next row, with false after the last.102func (c *cursor) next() ([]sheet.LiveCell, bool) {103 if c.i*c.width >= len(c.cur) {104 if c.cur != nil {105 select {106 case c.free <- c.cur:107 default:108 }109 }110 b, ok := <-c.batches111 if !ok {112 c.cur = nil113 return nil, false114 }115 c.cur, c.i = b, 0116 }117 row := c.cur[c.i*c.width : (c.i+1)*c.width]118 c.i++119 return row, true120}121122// stopCursor stops c's scan, noting what it failed with.123func (b *book) stopCursor(c *cursor) {124 c.cancel()125 for range c.batches {126 }127 if c.err != nil {128 b.fail(c.err)129 }130}131132// read gives emit the cells of part from the cursor on, until emit133// returns false or the part ends.134func (c *cursor) read(part sheet.Rect, emit func(sheet.Addr, sheet.Value) bool) {135 for c.at.Row <= part.To.Row {136 if !c.have {137 row, ok := c.next()138 if !ok {139 c.at.Row = part.To.Row + 1140 return141 }142 c.row, c.have = row, true143 }144 more := true145 for more && c.at.Col <= part.To.Col {146 v := c.row[c.at.Col-part.From.Col].V147 a := c.at148 c.at.Col++149 more = emit(a, v)150 }151 if c.at.Col > part.To.Col {152 // The row is read: the next chunk starts at the next one's153 // first cell, as the Reader asks for it.154 c.at, c.have = sheet.Addr{Col: part.From.Col, Row: c.at.Row + 1}, false155 }156 if !more {157 return158 }159 }160}