Line data Source code
1 : // Copyright 2020 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package pebble
6 :
7 : import (
8 : "bytes"
9 : "context"
10 : "fmt"
11 : "io"
12 : "runtime/debug"
13 : "runtime/pprof"
14 : "sync"
15 : "sync/atomic"
16 : "unsafe"
17 :
18 : "github.com/cockroachdb/errors"
19 : "github.com/cockroachdb/pebble/internal/base"
20 : "github.com/cockroachdb/pebble/internal/invariants"
21 : "github.com/cockroachdb/pebble/internal/keyspan"
22 : "github.com/cockroachdb/pebble/internal/manifest"
23 : "github.com/cockroachdb/pebble/internal/private"
24 : "github.com/cockroachdb/pebble/objstorage"
25 : "github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
26 : "github.com/cockroachdb/pebble/sstable"
27 : )
28 :
29 : var emptyIter = &errorIter{err: nil}
30 : var emptyKeyspanIter = &errorKeyspanIter{err: nil}
31 :
32 : // filteredAll is a singleton internalIterator implementation used when an
33 : // sstable does contain point keys, but all the keys are filtered by the active
34 : // PointKeyFilters set in the iterator's IterOptions.
35 : //
36 : // filteredAll implements filteredIter, ensuring the level iterator recognizes
37 : // when it may need to return file boundaries to keep the rangeDelIter open
38 : // during mergingIter operation.
39 : var filteredAll = &filteredAllKeysIter{errorIter: errorIter{err: nil}}
40 :
41 : var _ filteredIter = filteredAll
42 :
43 : type filteredAllKeysIter struct {
44 : errorIter
45 : }
46 :
47 1 : func (s *filteredAllKeysIter) MaybeFilteredKeys() bool {
48 1 : return true
49 1 : }
50 :
51 : var tableCacheLabels = pprof.Labels("pebble", "table-cache")
52 :
53 : // tableCacheOpts contains the db specific fields
54 : // of a table cache. This is stored in the tableCacheContainer
55 : // along with the table cache.
56 : // NB: It is important to make sure that the fields in this
57 : // struct are read-only. Since the fields here are shared
58 : // by every single tableCacheShard, if non read-only fields
59 : // are updated, we could have unnecessary evictions of those
60 : // fields, and the surrounding fields from the CPU caches.
61 : type tableCacheOpts struct {
62 : // iterCount keeps track of how many iterators are open. It is used to keep
63 : // track of leaked iterators on a per-db level.
64 : iterCount *atomic.Int32
65 :
66 : loggerAndTracer LoggerAndTracer
67 : cacheID uint64
68 : objProvider objstorage.Provider
69 : opts sstable.ReaderOptions
70 : filterMetrics *sstable.FilterMetricsTracker
71 : sstStatsCollector *sstable.CategoryStatsCollector
72 : }
73 :
74 : // tableCacheContainer contains the table cache and
75 : // fields which are unique to the DB.
76 : type tableCacheContainer struct {
77 : tableCache *TableCache
78 :
79 : // dbOpts contains fields relevant to the table cache
80 : // which are unique to each DB.
81 : dbOpts tableCacheOpts
82 : }
83 :
84 : // newTableCacheContainer will panic if the underlying cache in the table cache
85 : // doesn't match Options.Cache.
86 : func newTableCacheContainer(
87 : tc *TableCache,
88 : cacheID uint64,
89 : objProvider objstorage.Provider,
90 : opts *Options,
91 : size int,
92 : sstStatsCollector *sstable.CategoryStatsCollector,
93 1 : ) *tableCacheContainer {
94 1 : // We will release a ref to table cache acquired here when tableCacheContainer.close is called.
95 1 : if tc != nil {
96 0 : if tc.cache != opts.Cache {
97 0 : panic("pebble: underlying cache for the table cache and db are different")
98 : }
99 0 : tc.Ref()
100 1 : } else {
101 1 : // NewTableCache should create a ref to tc which the container should
102 1 : // drop whenever it is closed.
103 1 : tc = NewTableCache(opts.Cache, opts.Experimental.TableCacheShards, size)
104 1 : }
105 :
106 1 : t := &tableCacheContainer{}
107 1 : t.tableCache = tc
108 1 : t.dbOpts.loggerAndTracer = opts.LoggerAndTracer
109 1 : t.dbOpts.cacheID = cacheID
110 1 : t.dbOpts.objProvider = objProvider
111 1 : t.dbOpts.opts = opts.MakeReaderOptions()
112 1 : t.dbOpts.filterMetrics = &sstable.FilterMetricsTracker{}
113 1 : t.dbOpts.iterCount = new(atomic.Int32)
114 1 : t.dbOpts.sstStatsCollector = sstStatsCollector
115 1 : return t
116 : }
117 :
118 : // Before calling close, make sure that there will be no further need
119 : // to access any of the files associated with the store.
120 1 : func (c *tableCacheContainer) close() error {
121 1 : // We want to do some cleanup work here. Check for leaked iterators
122 1 : // by the DB using this container. Note that we'll still perform cleanup
123 1 : // below in the case that there are leaked iterators.
124 1 : var err error
125 1 : if v := c.dbOpts.iterCount.Load(); v > 0 {
126 0 : err = errors.Errorf("leaked iterators: %d", errors.Safe(v))
127 0 : }
128 :
129 : // Release nodes here.
130 1 : for _, shard := range c.tableCache.shards {
131 1 : if shard != nil {
132 1 : shard.removeDB(&c.dbOpts)
133 1 : }
134 : }
135 1 : return firstError(err, c.tableCache.Unref())
136 : }
137 :
138 : func (c *tableCacheContainer) newIters(
139 : ctx context.Context,
140 : file *manifest.FileMetadata,
141 : opts *IterOptions,
142 : internalOpts internalIterOpts,
143 1 : ) (internalIterator, keyspan.FragmentIterator, error) {
144 1 : return c.tableCache.getShard(file.FileBacking.DiskFileNum).newIters(ctx, file, opts, internalOpts, &c.dbOpts)
145 1 : }
146 :
147 : func (c *tableCacheContainer) newRangeKeyIter(
148 : file *manifest.FileMetadata, opts keyspan.SpanIterOptions,
149 1 : ) (keyspan.FragmentIterator, error) {
150 1 : return c.tableCache.getShard(file.FileBacking.DiskFileNum).newRangeKeyIter(file, opts, &c.dbOpts)
151 1 : }
152 :
153 : // getTableProperties returns the properties associated with the backing physical
154 : // table if the input metadata belongs to a virtual sstable.
155 0 : func (c *tableCacheContainer) getTableProperties(file *fileMetadata) (*sstable.Properties, error) {
156 0 : return c.tableCache.getShard(file.FileBacking.DiskFileNum).getTableProperties(file, &c.dbOpts)
157 0 : }
158 :
159 1 : func (c *tableCacheContainer) evict(fileNum base.DiskFileNum) {
160 1 : c.tableCache.getShard(fileNum).evict(fileNum, &c.dbOpts, false)
161 1 : }
162 :
163 1 : func (c *tableCacheContainer) metrics() (CacheMetrics, FilterMetrics) {
164 1 : var m CacheMetrics
165 1 : for i := range c.tableCache.shards {
166 1 : s := c.tableCache.shards[i]
167 1 : s.mu.RLock()
168 1 : m.Count += int64(len(s.mu.nodes))
169 1 : s.mu.RUnlock()
170 1 : m.Hits += s.hits.Load()
171 1 : m.Misses += s.misses.Load()
172 1 : }
173 1 : m.Size = m.Count * int64(unsafe.Sizeof(sstable.Reader{}))
174 1 : f := c.dbOpts.filterMetrics.Load()
175 1 : return m, f
176 : }
177 :
178 : func (c *tableCacheContainer) estimateSize(
179 : meta *fileMetadata, lower, upper []byte,
180 1 : ) (size uint64, err error) {
181 1 : if meta.Virtual {
182 1 : err = c.withVirtualReader(
183 1 : meta.VirtualMeta(),
184 1 : func(r sstable.VirtualReader) (err error) {
185 1 : size, err = r.EstimateDiskUsage(lower, upper)
186 1 : return err
187 1 : },
188 : )
189 1 : } else {
190 1 : err = c.withReader(
191 1 : meta.PhysicalMeta(),
192 1 : func(r *sstable.Reader) (err error) {
193 1 : size, err = r.EstimateDiskUsage(lower, upper)
194 1 : return err
195 1 : },
196 : )
197 : }
198 1 : if err != nil {
199 0 : return 0, err
200 0 : }
201 1 : return size, nil
202 : }
203 :
204 : // createCommonReader creates a Reader for this file. isShared, if true for
205 : // virtual sstables, is passed into the vSSTable reader so its iterators can
206 : // collapse obsolete points accordingly.
207 : func createCommonReader(
208 : v *tableCacheValue, file *fileMetadata, isShared bool,
209 1 : ) sstable.CommonReader {
210 1 : // TODO(bananabrick): We suffer an allocation if file is a virtual sstable.
211 1 : var cr sstable.CommonReader = v.reader
212 1 : if file.Virtual {
213 1 : virtualReader := sstable.MakeVirtualReader(
214 1 : v.reader, file.VirtualMeta(), isShared,
215 1 : )
216 1 : cr = &virtualReader
217 1 : }
218 1 : return cr
219 : }
220 :
221 : func (c *tableCacheContainer) withCommonReader(
222 : meta *fileMetadata, fn func(sstable.CommonReader) error,
223 1 : ) error {
224 1 : s := c.tableCache.getShard(meta.FileBacking.DiskFileNum)
225 1 : v := s.findNode(meta, &c.dbOpts)
226 1 : defer s.unrefValue(v)
227 1 : if v.err != nil {
228 0 : return v.err
229 0 : }
230 1 : provider := c.dbOpts.objProvider
231 1 : objMeta, err := provider.Lookup(fileTypeTable, meta.FileBacking.DiskFileNum)
232 1 : if err != nil {
233 0 : return err
234 0 : }
235 1 : return fn(createCommonReader(v, meta, objMeta.IsShared()))
236 : }
237 :
238 1 : func (c *tableCacheContainer) withReader(meta physicalMeta, fn func(*sstable.Reader) error) error {
239 1 : s := c.tableCache.getShard(meta.FileBacking.DiskFileNum)
240 1 : v := s.findNode(meta.FileMetadata, &c.dbOpts)
241 1 : defer s.unrefValue(v)
242 1 : if v.err != nil {
243 0 : return v.err
244 0 : }
245 1 : return fn(v.reader)
246 : }
247 :
248 : // withVirtualReader fetches a VirtualReader associated with a virtual sstable.
249 : func (c *tableCacheContainer) withVirtualReader(
250 : meta virtualMeta, fn func(sstable.VirtualReader) error,
251 1 : ) error {
252 1 : s := c.tableCache.getShard(meta.FileBacking.DiskFileNum)
253 1 : v := s.findNode(meta.FileMetadata, &c.dbOpts)
254 1 : defer s.unrefValue(v)
255 1 : if v.err != nil {
256 0 : return v.err
257 0 : }
258 1 : provider := c.dbOpts.objProvider
259 1 : objMeta, err := provider.Lookup(fileTypeTable, meta.FileBacking.DiskFileNum)
260 1 : if err != nil {
261 0 : return err
262 0 : }
263 1 : return fn(sstable.MakeVirtualReader(v.reader, meta, objMeta.IsShared()))
264 : }
265 :
266 1 : func (c *tableCacheContainer) iterCount() int64 {
267 1 : return int64(c.dbOpts.iterCount.Load())
268 1 : }
269 :
270 : // TableCache is a shareable cache for open sstables.
271 : type TableCache struct {
272 : refs atomic.Int64
273 :
274 : cache *Cache
275 : shards []*tableCacheShard
276 : }
277 :
278 : // Ref adds a reference to the table cache. Once tableCache.init returns,
279 : // the table cache only remains valid if there is at least one reference
280 : // to it.
281 0 : func (c *TableCache) Ref() {
282 0 : v := c.refs.Add(1)
283 0 : // We don't want the reference count to ever go from 0 -> 1,
284 0 : // cause a reference count of 0 implies that we've closed the cache.
285 0 : if v <= 1 {
286 0 : panic(fmt.Sprintf("pebble: inconsistent reference count: %d", v))
287 : }
288 : }
289 :
290 : // Unref removes a reference to the table cache.
291 1 : func (c *TableCache) Unref() error {
292 1 : v := c.refs.Add(-1)
293 1 : switch {
294 0 : case v < 0:
295 0 : panic(fmt.Sprintf("pebble: inconsistent reference count: %d", v))
296 1 : case v == 0:
297 1 : var err error
298 1 : for i := range c.shards {
299 1 : // The cache shard is not allocated yet, nothing to close
300 1 : if c.shards[i] == nil {
301 0 : continue
302 : }
303 1 : err = firstError(err, c.shards[i].Close())
304 : }
305 :
306 : // Unref the cache which we create a reference to when the tableCache
307 : // is first instantiated.
308 1 : c.cache.Unref()
309 1 : return err
310 : }
311 0 : return nil
312 : }
313 :
314 : // NewTableCache will create a reference to the table cache. It is the callers responsibility
315 : // to call tableCache.Unref if they will no longer hold a reference to the table cache.
316 1 : func NewTableCache(cache *Cache, numShards int, size int) *TableCache {
317 1 : if size == 0 {
318 0 : panic("pebble: cannot create a table cache of size 0")
319 1 : } else if numShards == 0 {
320 0 : panic("pebble: cannot create a table cache with 0 shards")
321 : }
322 :
323 1 : c := &TableCache{}
324 1 : c.cache = cache
325 1 : c.cache.Ref()
326 1 :
327 1 : c.shards = make([]*tableCacheShard, numShards)
328 1 : for i := range c.shards {
329 1 : c.shards[i] = &tableCacheShard{}
330 1 : c.shards[i].init(size / len(c.shards))
331 1 : }
332 :
333 : // Hold a ref to the cache here.
334 1 : c.refs.Store(1)
335 1 :
336 1 : return c
337 : }
338 :
339 1 : func (c *TableCache) getShard(fileNum base.DiskFileNum) *tableCacheShard {
340 1 : return c.shards[uint64(fileNum)%uint64(len(c.shards))]
341 1 : }
342 :
343 : type tableCacheKey struct {
344 : cacheID uint64
345 : fileNum base.DiskFileNum
346 : }
347 :
348 : type tableCacheShard struct {
349 : hits atomic.Int64
350 : misses atomic.Int64
351 : iterCount atomic.Int32
352 :
353 : size int
354 :
355 : mu struct {
356 : sync.RWMutex
357 : nodes map[tableCacheKey]*tableCacheNode
358 : // The iters map is only created and populated in race builds.
359 : iters map[io.Closer][]byte
360 :
361 : handHot *tableCacheNode
362 : handCold *tableCacheNode
363 : handTest *tableCacheNode
364 :
365 : coldTarget int
366 : sizeHot int
367 : sizeCold int
368 : sizeTest int
369 : }
370 : releasing sync.WaitGroup
371 : releasingCh chan *tableCacheValue
372 : releaseLoopExit sync.WaitGroup
373 : }
374 :
375 1 : func (c *tableCacheShard) init(size int) {
376 1 : c.size = size
377 1 :
378 1 : c.mu.nodes = make(map[tableCacheKey]*tableCacheNode)
379 1 : c.mu.coldTarget = size
380 1 : c.releasingCh = make(chan *tableCacheValue, 100)
381 1 : c.releaseLoopExit.Add(1)
382 1 : go c.releaseLoop()
383 1 :
384 1 : if invariants.RaceEnabled {
385 0 : c.mu.iters = make(map[io.Closer][]byte)
386 0 : }
387 : }
388 :
389 1 : func (c *tableCacheShard) releaseLoop() {
390 1 : pprof.Do(context.Background(), tableCacheLabels, func(context.Context) {
391 1 : defer c.releaseLoopExit.Done()
392 1 : for v := range c.releasingCh {
393 0 : v.release(c)
394 0 : }
395 : })
396 : }
397 :
398 : // checkAndIntersectFilters checks the specific table and block property filters
399 : // for intersection with any available table and block-level properties. Returns
400 : // true for ok if this table should be read by this iterator.
401 : func (c *tableCacheShard) checkAndIntersectFilters(
402 : v *tableCacheValue,
403 : tableFilter func(userProps map[string]string) bool,
404 : blockPropertyFilters []BlockPropertyFilter,
405 : boundLimitedFilter sstable.BoundLimitedBlockPropertyFilter,
406 1 : ) (ok bool, filterer *sstable.BlockPropertiesFilterer, err error) {
407 1 : if tableFilter != nil &&
408 1 : !tableFilter(v.reader.Properties.UserProperties) {
409 0 : return false, nil, nil
410 0 : }
411 :
412 1 : if boundLimitedFilter != nil || len(blockPropertyFilters) > 0 {
413 1 : filterer, err = sstable.IntersectsTable(
414 1 : blockPropertyFilters,
415 1 : boundLimitedFilter,
416 1 : v.reader.Properties.UserProperties,
417 1 : )
418 1 : // NB: IntersectsTable will return a nil filterer if the table-level
419 1 : // properties indicate there's no intersection with the provided filters.
420 1 : if filterer == nil || err != nil {
421 1 : return false, nil, err
422 1 : }
423 : }
424 1 : return true, filterer, nil
425 : }
426 :
427 : func (c *tableCacheShard) newIters(
428 : ctx context.Context,
429 : file *manifest.FileMetadata,
430 : opts *IterOptions,
431 : internalOpts internalIterOpts,
432 : dbOpts *tableCacheOpts,
433 1 : ) (internalIterator, keyspan.FragmentIterator, error) {
434 1 : // TODO(sumeer): constructing the Reader should also use a plumbed context,
435 1 : // since parts of the sstable are read during the construction. The Reader
436 1 : // should not remember that context since the Reader can be long-lived.
437 1 :
438 1 : // Calling findNode gives us the responsibility of decrementing v's
439 1 : // refCount. If opening the underlying table resulted in error, then we
440 1 : // decrement this straight away. Otherwise, we pass that responsibility to
441 1 : // the sstable iterator, which decrements when it is closed.
442 1 : v := c.findNode(file, dbOpts)
443 1 : if v.err != nil {
444 0 : defer c.unrefValue(v)
445 0 : return nil, nil, v.err
446 0 : }
447 :
448 1 : hideObsoletePoints := false
449 1 : var pointKeyFilters []BlockPropertyFilter
450 1 : if opts != nil {
451 1 : // This code is appending (at most one filter) in-place to
452 1 : // opts.PointKeyFilters even though the slice is shared for iterators in
453 1 : // the same iterator tree. This is acceptable since all the following
454 1 : // properties are true:
455 1 : // - The iterator tree is single threaded, so the shared backing for the
456 1 : // slice is being mutated in a single threaded manner.
457 1 : // - Each shallow copy of the slice has its own notion of length.
458 1 : // - The appended element is always the obsoleteKeyBlockPropertyFilter
459 1 : // struct, which is stateless, so overwriting that struct when creating
460 1 : // one sstable iterator is harmless to other sstable iterators that are
461 1 : // relying on that struct.
462 1 : //
463 1 : // An alternative would be to have different slices for different sstable
464 1 : // iterators, but that requires more work to avoid allocations.
465 1 : //
466 1 : // TODO(bilal): for compaction reads of foreign sstables, we do hide
467 1 : // obsolete points (see sstable.Reader.newCompactionIter) but we don't
468 1 : // apply the obsolete block property filter. We could optimize this by
469 1 : // applying the filter.
470 1 : hideObsoletePoints, pointKeyFilters =
471 1 : v.reader.TryAddBlockPropertyFilterForHideObsoletePoints(
472 1 : opts.snapshotForHideObsoletePoints, file.LargestSeqNum, opts.PointKeyFilters)
473 1 : }
474 1 : ok := true
475 1 : var filterer *sstable.BlockPropertiesFilterer
476 1 : var err error
477 1 : if opts != nil {
478 1 : ok, filterer, err = c.checkAndIntersectFilters(v, opts.TableFilter,
479 1 : pointKeyFilters, internalOpts.boundLimitedFilter)
480 1 : }
481 1 : if err != nil {
482 0 : c.unrefValue(v)
483 0 : return nil, nil, err
484 0 : }
485 :
486 1 : provider := dbOpts.objProvider
487 1 : // Check if this file is a foreign file.
488 1 : objMeta, err := provider.Lookup(fileTypeTable, file.FileBacking.DiskFileNum)
489 1 : if err != nil {
490 0 : return nil, nil, err
491 0 : }
492 :
493 : // Note: This suffers an allocation for virtual sstables.
494 1 : cr := createCommonReader(v, file, objMeta.IsShared())
495 1 :
496 1 : // NB: range-del iterator does not maintain a reference to the table, nor
497 1 : // does it need to read from it after creation.
498 1 : rangeDelIter, err := cr.NewRawRangeDelIter()
499 1 : if err != nil {
500 0 : c.unrefValue(v)
501 0 : return nil, nil, err
502 0 : }
503 :
504 : // Assert expected bounds in tests.
505 1 : if invariants.Enabled && rangeDelIter != nil {
506 1 : cmp := base.DefaultComparer.Compare
507 1 : if dbOpts.opts.Comparer != nil {
508 1 : cmp = dbOpts.opts.Comparer.Compare
509 1 : }
510 : // TODO(radu): we should be using AssertBounds, but it currently fails in
511 : // some cases (#3167).
512 1 : rangeDelIter = keyspan.AssertUserKeyBounds(
513 1 : rangeDelIter, file.SmallestPointKey.UserKey, file.LargestPointKey.UserKey, cmp,
514 1 : )
515 : }
516 :
517 1 : if !ok {
518 1 : c.unrefValue(v)
519 1 : // Return an empty iterator. This iterator has no mutable state, so
520 1 : // using a singleton is fine.
521 1 : // NB: We still return the potentially non-empty rangeDelIter. This
522 1 : // ensures the iterator observes the file's range deletions even if the
523 1 : // block property filters exclude all the file's point keys. The range
524 1 : // deletions may still delete keys lower in the LSM in files that DO
525 1 : // match the active filters.
526 1 : //
527 1 : // The point iterator returned must implement the filteredIter
528 1 : // interface, so that the level iterator surfaces file boundaries when
529 1 : // range deletions are present.
530 1 : return filteredAll, rangeDelIter, err
531 1 : }
532 :
533 1 : var iter sstable.Iterator
534 1 : useFilter := true
535 1 : if opts != nil {
536 1 : useFilter = manifest.LevelToInt(opts.level) != 6 || opts.UseL6Filters
537 1 : ctx = objiotracing.WithLevel(ctx, manifest.LevelToInt(opts.level))
538 1 : }
539 1 : tableFormat, err := v.reader.TableFormat()
540 1 : if err != nil {
541 0 : return nil, nil, err
542 0 : }
543 1 : var rp sstable.ReaderProvider
544 1 : if tableFormat >= sstable.TableFormatPebblev3 && v.reader.Properties.NumValueBlocks > 0 {
545 1 : rp = &tableCacheShardReaderProvider{c: c, file: file, dbOpts: dbOpts}
546 1 : }
547 :
548 1 : if objMeta.IsShared() && v.reader.Properties.GlobalSeqNum != 0 {
549 1 : if tableFormat < sstable.TableFormatPebblev4 {
550 0 : return nil, nil, errors.New("pebble: shared ingested sstable has a lower table format than expected")
551 0 : }
552 : // The table is shared and ingested.
553 1 : hideObsoletePoints = true
554 : }
555 1 : var categoryAndQoS sstable.CategoryAndQoS
556 1 : if opts != nil {
557 1 : categoryAndQoS = opts.CategoryAndQoS
558 1 : }
559 1 : if internalOpts.bytesIterated != nil {
560 1 : iter, err = cr.NewCompactionIter(
561 1 : internalOpts.bytesIterated, categoryAndQoS, dbOpts.sstStatsCollector, rp,
562 1 : internalOpts.bufferPool)
563 1 : } else {
564 1 : iter, err = cr.NewIterWithBlockPropertyFiltersAndContextEtc(
565 1 : ctx, opts.GetLowerBound(), opts.GetUpperBound(), filterer, hideObsoletePoints, useFilter,
566 1 : internalOpts.stats, categoryAndQoS, dbOpts.sstStatsCollector, rp)
567 1 : }
568 1 : if err != nil {
569 0 : if rangeDelIter != nil {
570 0 : _ = rangeDelIter.Close()
571 0 : }
572 0 : c.unrefValue(v)
573 0 : return nil, nil, err
574 : }
575 : // NB: v.closeHook takes responsibility for calling unrefValue(v) here. Take
576 : // care to avoid introducing an allocation here by adding a closure.
577 1 : iter.SetCloseHook(v.closeHook)
578 1 :
579 1 : c.iterCount.Add(1)
580 1 : dbOpts.iterCount.Add(1)
581 1 : if invariants.RaceEnabled {
582 0 : c.mu.Lock()
583 0 : c.mu.iters[iter] = debug.Stack()
584 0 : c.mu.Unlock()
585 0 : }
586 1 : return iter, rangeDelIter, nil
587 : }
588 :
589 : func (c *tableCacheShard) newRangeKeyIter(
590 : file *manifest.FileMetadata, opts keyspan.SpanIterOptions, dbOpts *tableCacheOpts,
591 1 : ) (keyspan.FragmentIterator, error) {
592 1 : // Calling findNode gives us the responsibility of decrementing v's
593 1 : // refCount. If opening the underlying table resulted in error, then we
594 1 : // decrement this straight away. Otherwise, we pass that responsibility to
595 1 : // the sstable iterator, which decrements when it is closed.
596 1 : v := c.findNode(file, dbOpts)
597 1 : if v.err != nil {
598 0 : defer c.unrefValue(v)
599 0 : return nil, v.err
600 0 : }
601 :
602 1 : ok := true
603 1 : var err error
604 1 : // Don't filter a table's range keys if the file contains RANGEKEYDELs.
605 1 : // The RANGEKEYDELs may delete range keys in other levels. Skipping the
606 1 : // file's range key blocks may surface deleted range keys below. This is
607 1 : // done here, rather than deferring to the block-property collector in order
608 1 : // to maintain parity with point keys and the treatment of RANGEDELs.
609 1 : if v.reader.Properties.NumRangeKeyDels == 0 {
610 1 : ok, _, err = c.checkAndIntersectFilters(v, nil, opts.RangeKeyFilters, nil)
611 1 : }
612 1 : if err != nil {
613 0 : c.unrefValue(v)
614 0 : return nil, err
615 0 : }
616 1 : if !ok {
617 0 : c.unrefValue(v)
618 0 : // Return the empty iterator. This iterator has no mutable state, so
619 0 : // using a singleton is fine.
620 0 : return emptyKeyspanIter, err
621 0 : }
622 :
623 1 : var iter keyspan.FragmentIterator
624 1 : if file.Virtual {
625 1 : provider := dbOpts.objProvider
626 1 : var objMeta objstorage.ObjectMetadata
627 1 : objMeta, err = provider.Lookup(fileTypeTable, file.FileBacking.DiskFileNum)
628 1 : if err == nil {
629 1 : virtualReader := sstable.MakeVirtualReader(
630 1 : v.reader, file.VirtualMeta(), objMeta.IsShared(),
631 1 : )
632 1 : iter, err = virtualReader.NewRawRangeKeyIter()
633 1 : }
634 1 : } else {
635 1 : iter, err = v.reader.NewRawRangeKeyIter()
636 1 : }
637 :
638 : // iter is a block iter that holds the entire value of the block in memory.
639 : // No need to hold onto a ref of the cache value.
640 1 : c.unrefValue(v)
641 1 :
642 1 : if err != nil {
643 0 : return nil, err
644 0 : }
645 :
646 1 : if iter == nil {
647 1 : // NewRawRangeKeyIter can return nil even if there's no error. However,
648 1 : // the keyspan.LevelIter expects a non-nil iterator if err is nil.
649 1 : return emptyKeyspanIter, nil
650 1 : }
651 :
652 1 : return iter, nil
653 : }
654 :
655 : type tableCacheShardReaderProvider struct {
656 : c *tableCacheShard
657 : file *manifest.FileMetadata
658 : dbOpts *tableCacheOpts
659 : v *tableCacheValue
660 : }
661 :
662 : var _ sstable.ReaderProvider = &tableCacheShardReaderProvider{}
663 :
664 : // GetReader implements sstable.ReaderProvider. Note that it is not the
665 : // responsibility of tableCacheShardReaderProvider to ensure that the file
666 : // continues to exist. The ReaderProvider is used in iterators where the
667 : // top-level iterator is pinning the read state and preventing the files from
668 : // being deleted.
669 : //
670 : // The caller must call tableCacheShardReaderProvider.Close.
671 : //
672 : // Note that currently the Reader returned here is only used to read value
673 : // blocks. This reader shouldn't be used for other purposes like reading keys
674 : // outside of virtual sstable bounds.
675 : //
676 : // TODO(bananabrick): We could return a wrapper over the Reader to ensure
677 : // that the reader isn't used for other purposes.
678 1 : func (rp *tableCacheShardReaderProvider) GetReader() (*sstable.Reader, error) {
679 1 : // Calling findNode gives us the responsibility of decrementing v's
680 1 : // refCount.
681 1 : v := rp.c.findNode(rp.file, rp.dbOpts)
682 1 : if v.err != nil {
683 0 : defer rp.c.unrefValue(v)
684 0 : return nil, v.err
685 0 : }
686 1 : rp.v = v
687 1 : return v.reader, nil
688 : }
689 :
690 : // Close implements sstable.ReaderProvider.
691 1 : func (rp *tableCacheShardReaderProvider) Close() {
692 1 : rp.c.unrefValue(rp.v)
693 1 : rp.v = nil
694 1 : }
695 :
696 : // getTableProperties return sst table properties for target file
697 : func (c *tableCacheShard) getTableProperties(
698 : file *fileMetadata, dbOpts *tableCacheOpts,
699 0 : ) (*sstable.Properties, error) {
700 0 : // Calling findNode gives us the responsibility of decrementing v's refCount here
701 0 : v := c.findNode(file, dbOpts)
702 0 : defer c.unrefValue(v)
703 0 :
704 0 : if v.err != nil {
705 0 : return nil, v.err
706 0 : }
707 0 : return &v.reader.Properties, nil
708 : }
709 :
710 : // releaseNode releases a node from the tableCacheShard.
711 : //
712 : // c.mu must be held when calling this.
713 0 : func (c *tableCacheShard) releaseNode(n *tableCacheNode) {
714 0 : c.unlinkNode(n)
715 0 : c.clearNode(n)
716 0 : }
717 :
718 : // unlinkNode removes a node from the tableCacheShard, leaving the shard
719 : // reference in place.
720 : //
721 : // c.mu must be held when calling this.
722 1 : func (c *tableCacheShard) unlinkNode(n *tableCacheNode) {
723 1 : key := tableCacheKey{n.cacheID, n.fileNum}
724 1 : delete(c.mu.nodes, key)
725 1 :
726 1 : switch n.ptype {
727 0 : case tableCacheNodeHot:
728 0 : c.mu.sizeHot--
729 1 : case tableCacheNodeCold:
730 1 : c.mu.sizeCold--
731 0 : case tableCacheNodeTest:
732 0 : c.mu.sizeTest--
733 : }
734 :
735 1 : if n == c.mu.handHot {
736 1 : c.mu.handHot = c.mu.handHot.prev()
737 1 : }
738 1 : if n == c.mu.handCold {
739 1 : c.mu.handCold = c.mu.handCold.prev()
740 1 : }
741 1 : if n == c.mu.handTest {
742 1 : c.mu.handTest = c.mu.handTest.prev()
743 1 : }
744 :
745 1 : if n.unlink() == n {
746 1 : // This was the last entry in the cache.
747 1 : c.mu.handHot = nil
748 1 : c.mu.handCold = nil
749 1 : c.mu.handTest = nil
750 1 : }
751 :
752 1 : n.links.prev = nil
753 1 : n.links.next = nil
754 : }
755 :
756 0 : func (c *tableCacheShard) clearNode(n *tableCacheNode) {
757 0 : if v := n.value; v != nil {
758 0 : n.value = nil
759 0 : c.unrefValue(v)
760 0 : }
761 : }
762 :
763 : // unrefValue decrements the reference count for the specified value, releasing
764 : // it if the reference count fell to 0. Note that the value has a reference if
765 : // it is present in tableCacheShard.mu.nodes, so a reference count of 0 means
766 : // the node has already been removed from that map.
767 1 : func (c *tableCacheShard) unrefValue(v *tableCacheValue) {
768 1 : if v.refCount.Add(-1) == 0 {
769 0 : c.releasing.Add(1)
770 0 : c.releasingCh <- v
771 0 : }
772 : }
773 :
774 : // findNode returns the node for the table with the given file number, creating
775 : // that node if it didn't already exist. The caller is responsible for
776 : // decrementing the returned node's refCount.
777 1 : func (c *tableCacheShard) findNode(meta *fileMetadata, dbOpts *tableCacheOpts) *tableCacheValue {
778 1 : v := c.findNodeInternal(meta, dbOpts)
779 1 :
780 1 : // Loading a file before its global sequence number is known (eg,
781 1 : // during ingest before entering the commit pipeline) can pollute
782 1 : // the cache with incorrect state. In invariant builds, verify
783 1 : // that the global sequence number of the returned reader matches.
784 1 : if invariants.Enabled {
785 1 : if v.reader != nil && meta.LargestSeqNum == meta.SmallestSeqNum &&
786 1 : v.reader.Properties.GlobalSeqNum != meta.SmallestSeqNum {
787 0 : panic(errors.AssertionFailedf("file %s loaded from table cache with the wrong global sequence number %d",
788 0 : meta, v.reader.Properties.GlobalSeqNum))
789 : }
790 : }
791 1 : return v
792 : }
793 :
794 : func (c *tableCacheShard) findNodeInternal(
795 : meta *fileMetadata, dbOpts *tableCacheOpts,
796 1 : ) *tableCacheValue {
797 1 : if refs := meta.Refs(); refs <= 0 {
798 0 : panic(errors.AssertionFailedf("attempting to load file %s with refs=%d from table cache",
799 0 : meta, refs))
800 : }
801 : // Fast-path for a hit in the cache.
802 1 : c.mu.RLock()
803 1 : key := tableCacheKey{dbOpts.cacheID, meta.FileBacking.DiskFileNum}
804 1 : if n := c.mu.nodes[key]; n != nil && n.value != nil {
805 1 : // Fast-path hit.
806 1 : //
807 1 : // The caller is responsible for decrementing the refCount.
808 1 : v := n.value
809 1 : v.refCount.Add(1)
810 1 : c.mu.RUnlock()
811 1 : n.referenced.Store(true)
812 1 : c.hits.Add(1)
813 1 : <-v.loaded
814 1 : return v
815 1 : }
816 1 : c.mu.RUnlock()
817 1 :
818 1 : c.mu.Lock()
819 1 :
820 1 : n := c.mu.nodes[key]
821 1 : switch {
822 1 : case n == nil:
823 1 : // Slow-path miss of a non-existent node.
824 1 : n = &tableCacheNode{
825 1 : fileNum: meta.FileBacking.DiskFileNum,
826 1 : ptype: tableCacheNodeCold,
827 1 : }
828 1 : c.addNode(n, dbOpts)
829 1 : c.mu.sizeCold++
830 :
831 1 : case n.value != nil:
832 1 : // Slow-path hit of a hot or cold node.
833 1 : //
834 1 : // The caller is responsible for decrementing the refCount.
835 1 : v := n.value
836 1 : v.refCount.Add(1)
837 1 : n.referenced.Store(true)
838 1 : c.hits.Add(1)
839 1 : c.mu.Unlock()
840 1 : <-v.loaded
841 1 : return v
842 :
843 0 : default:
844 0 : // Slow-path miss of a test node.
845 0 : c.unlinkNode(n)
846 0 : c.mu.coldTarget++
847 0 : if c.mu.coldTarget > c.size {
848 0 : c.mu.coldTarget = c.size
849 0 : }
850 :
851 0 : n.referenced.Store(false)
852 0 : n.ptype = tableCacheNodeHot
853 0 : c.addNode(n, dbOpts)
854 0 : c.mu.sizeHot++
855 : }
856 :
857 1 : c.misses.Add(1)
858 1 :
859 1 : v := &tableCacheValue{
860 1 : loaded: make(chan struct{}),
861 1 : }
862 1 : v.refCount.Store(2)
863 1 : // Cache the closure invoked when an iterator is closed. This avoids an
864 1 : // allocation on every call to newIters.
865 1 : v.closeHook = func(i sstable.Iterator) error {
866 1 : if invariants.RaceEnabled {
867 0 : c.mu.Lock()
868 0 : delete(c.mu.iters, i)
869 0 : c.mu.Unlock()
870 0 : }
871 1 : c.unrefValue(v)
872 1 : c.iterCount.Add(-1)
873 1 : dbOpts.iterCount.Add(-1)
874 1 : return nil
875 : }
876 1 : n.value = v
877 1 :
878 1 : c.mu.Unlock()
879 1 :
880 1 : // Note adding to the cache lists must complete before we begin loading the
881 1 : // table as a failure during load will result in the node being unlinked.
882 1 : pprof.Do(context.Background(), tableCacheLabels, func(context.Context) {
883 1 : v.load(
884 1 : loadInfo{
885 1 : backingFileNum: meta.FileBacking.DiskFileNum,
886 1 : smallestSeqNum: meta.SmallestSeqNum,
887 1 : largestSeqNum: meta.LargestSeqNum,
888 1 : }, c, dbOpts)
889 1 : })
890 1 : return v
891 : }
892 :
893 1 : func (c *tableCacheShard) addNode(n *tableCacheNode, dbOpts *tableCacheOpts) {
894 1 : c.evictNodes()
895 1 : n.cacheID = dbOpts.cacheID
896 1 : key := tableCacheKey{n.cacheID, n.fileNum}
897 1 : c.mu.nodes[key] = n
898 1 :
899 1 : n.links.next = n
900 1 : n.links.prev = n
901 1 : if c.mu.handHot == nil {
902 1 : // First element.
903 1 : c.mu.handHot = n
904 1 : c.mu.handCold = n
905 1 : c.mu.handTest = n
906 1 : } else {
907 1 : c.mu.handHot.link(n)
908 1 : }
909 :
910 1 : if c.mu.handCold == c.mu.handHot {
911 1 : c.mu.handCold = c.mu.handCold.prev()
912 1 : }
913 : }
914 :
915 1 : func (c *tableCacheShard) evictNodes() {
916 1 : for c.size <= c.mu.sizeHot+c.mu.sizeCold && c.mu.handCold != nil {
917 0 : c.runHandCold()
918 0 : }
919 : }
920 :
921 0 : func (c *tableCacheShard) runHandCold() {
922 0 : n := c.mu.handCold
923 0 : if n.ptype == tableCacheNodeCold {
924 0 : if n.referenced.Load() {
925 0 : n.referenced.Store(false)
926 0 : n.ptype = tableCacheNodeHot
927 0 : c.mu.sizeCold--
928 0 : c.mu.sizeHot++
929 0 : } else {
930 0 : c.clearNode(n)
931 0 : n.ptype = tableCacheNodeTest
932 0 : c.mu.sizeCold--
933 0 : c.mu.sizeTest++
934 0 : for c.size < c.mu.sizeTest && c.mu.handTest != nil {
935 0 : c.runHandTest()
936 0 : }
937 : }
938 : }
939 :
940 0 : c.mu.handCold = c.mu.handCold.next()
941 0 :
942 0 : for c.size-c.mu.coldTarget <= c.mu.sizeHot && c.mu.handHot != nil {
943 0 : c.runHandHot()
944 0 : }
945 : }
946 :
947 0 : func (c *tableCacheShard) runHandHot() {
948 0 : if c.mu.handHot == c.mu.handTest && c.mu.handTest != nil {
949 0 : c.runHandTest()
950 0 : if c.mu.handHot == nil {
951 0 : return
952 0 : }
953 : }
954 :
955 0 : n := c.mu.handHot
956 0 : if n.ptype == tableCacheNodeHot {
957 0 : if n.referenced.Load() {
958 0 : n.referenced.Store(false)
959 0 : } else {
960 0 : n.ptype = tableCacheNodeCold
961 0 : c.mu.sizeHot--
962 0 : c.mu.sizeCold++
963 0 : }
964 : }
965 :
966 0 : c.mu.handHot = c.mu.handHot.next()
967 : }
968 :
969 0 : func (c *tableCacheShard) runHandTest() {
970 0 : if c.mu.sizeCold > 0 && c.mu.handTest == c.mu.handCold && c.mu.handCold != nil {
971 0 : c.runHandCold()
972 0 : if c.mu.handTest == nil {
973 0 : return
974 0 : }
975 : }
976 :
977 0 : n := c.mu.handTest
978 0 : if n.ptype == tableCacheNodeTest {
979 0 : c.mu.coldTarget--
980 0 : if c.mu.coldTarget < 0 {
981 0 : c.mu.coldTarget = 0
982 0 : }
983 0 : c.unlinkNode(n)
984 0 : c.clearNode(n)
985 : }
986 :
987 0 : c.mu.handTest = c.mu.handTest.next()
988 : }
989 :
990 1 : func (c *tableCacheShard) evict(fileNum base.DiskFileNum, dbOpts *tableCacheOpts, allowLeak bool) {
991 1 : c.mu.Lock()
992 1 : key := tableCacheKey{dbOpts.cacheID, fileNum}
993 1 : n := c.mu.nodes[key]
994 1 : var v *tableCacheValue
995 1 : if n != nil {
996 1 : // NB: This is equivalent to tableCacheShard.releaseNode(), but we perform
997 1 : // the tableCacheNode.release() call synchronously below to ensure the
998 1 : // sstable file descriptor is closed before returning. Note that
999 1 : // tableCacheShard.releasing needs to be incremented while holding
1000 1 : // tableCacheShard.mu in order to avoid a race with Close()
1001 1 : c.unlinkNode(n)
1002 1 : v = n.value
1003 1 : if v != nil {
1004 1 : if !allowLeak {
1005 1 : if t := v.refCount.Add(-1); t != 0 {
1006 0 : dbOpts.loggerAndTracer.Fatalf("sstable %s: refcount is not zero: %d\n%s", fileNum, t, debug.Stack())
1007 0 : }
1008 : }
1009 1 : c.releasing.Add(1)
1010 : }
1011 : }
1012 :
1013 1 : c.mu.Unlock()
1014 1 :
1015 1 : if v != nil {
1016 1 : v.release(c)
1017 1 : }
1018 :
1019 1 : dbOpts.opts.Cache.EvictFile(dbOpts.cacheID, fileNum)
1020 : }
1021 :
1022 : // removeDB evicts any nodes which have a reference to the DB
1023 : // associated with dbOpts.cacheID. Make sure that there will
1024 : // be no more accesses to the files associated with the DB.
1025 1 : func (c *tableCacheShard) removeDB(dbOpts *tableCacheOpts) {
1026 1 : var fileNums []base.DiskFileNum
1027 1 :
1028 1 : c.mu.RLock()
1029 1 : // Collect the fileNums which need to be cleaned.
1030 1 : var firstNode *tableCacheNode
1031 1 : node := c.mu.handHot
1032 1 : for node != firstNode {
1033 1 : if firstNode == nil {
1034 1 : firstNode = node
1035 1 : }
1036 :
1037 1 : if node.cacheID == dbOpts.cacheID {
1038 1 : fileNums = append(fileNums, node.fileNum)
1039 1 : }
1040 1 : node = node.next()
1041 : }
1042 1 : c.mu.RUnlock()
1043 1 :
1044 1 : // Evict all the nodes associated with the DB.
1045 1 : // This should synchronously close all the files
1046 1 : // associated with the DB.
1047 1 : for _, fileNum := range fileNums {
1048 1 : c.evict(fileNum, dbOpts, true)
1049 1 : }
1050 : }
1051 :
1052 1 : func (c *tableCacheShard) Close() error {
1053 1 : c.mu.Lock()
1054 1 : defer c.mu.Unlock()
1055 1 :
1056 1 : // Check for leaked iterators. Note that we'll still perform cleanup below in
1057 1 : // the case that there are leaked iterators.
1058 1 : var err error
1059 1 : if v := c.iterCount.Load(); v > 0 {
1060 0 : if !invariants.RaceEnabled {
1061 0 : err = errors.Errorf("leaked iterators: %d", errors.Safe(v))
1062 0 : } else {
1063 0 : var buf bytes.Buffer
1064 0 : for _, stack := range c.mu.iters {
1065 0 : fmt.Fprintf(&buf, "%s\n", stack)
1066 0 : }
1067 0 : err = errors.Errorf("leaked iterators: %d\n%s", errors.Safe(v), buf.String())
1068 : }
1069 : }
1070 :
1071 1 : for c.mu.handHot != nil {
1072 0 : n := c.mu.handHot
1073 0 : if n.value != nil {
1074 0 : if n.value.refCount.Add(-1) == 0 {
1075 0 : c.releasing.Add(1)
1076 0 : c.releasingCh <- n.value
1077 0 : }
1078 : }
1079 0 : c.unlinkNode(n)
1080 : }
1081 1 : c.mu.nodes = nil
1082 1 : c.mu.handHot = nil
1083 1 : c.mu.handCold = nil
1084 1 : c.mu.handTest = nil
1085 1 :
1086 1 : // Only shutdown the releasing goroutine if there were no leaked
1087 1 : // iterators. If there were leaked iterators, we leave the goroutine running
1088 1 : // and the releasingCh open so that a subsequent iterator close can
1089 1 : // complete. This behavior is used by iterator leak tests. Leaking the
1090 1 : // goroutine for these tests is less bad not closing the iterator which
1091 1 : // triggers other warnings about block cache handles not being released.
1092 1 : if err != nil {
1093 0 : c.releasing.Wait()
1094 0 : return err
1095 0 : }
1096 :
1097 1 : close(c.releasingCh)
1098 1 : c.releasing.Wait()
1099 1 : c.releaseLoopExit.Wait()
1100 1 : return err
1101 : }
1102 :
1103 : type tableCacheValue struct {
1104 : closeHook func(i sstable.Iterator) error
1105 : reader *sstable.Reader
1106 : err error
1107 : loaded chan struct{}
1108 : // Reference count for the value. The reader is closed when the reference
1109 : // count drops to zero.
1110 : refCount atomic.Int32
1111 : }
1112 :
1113 : type loadInfo struct {
1114 : backingFileNum base.DiskFileNum
1115 : largestSeqNum uint64
1116 : smallestSeqNum uint64
1117 : }
1118 :
1119 1 : func (v *tableCacheValue) load(loadInfo loadInfo, c *tableCacheShard, dbOpts *tableCacheOpts) {
1120 1 : // Try opening the file first.
1121 1 : var f objstorage.Readable
1122 1 : var err error
1123 1 : f, err = dbOpts.objProvider.OpenForReading(
1124 1 : context.TODO(), fileTypeTable, loadInfo.backingFileNum, objstorage.OpenOptions{MustExist: true},
1125 1 : )
1126 1 : if err == nil {
1127 1 : cacheOpts := private.SSTableCacheOpts(dbOpts.cacheID, loadInfo.backingFileNum).(sstable.ReaderOption)
1128 1 : v.reader, err = sstable.NewReader(f, dbOpts.opts, cacheOpts, dbOpts.filterMetrics)
1129 1 : }
1130 1 : if err != nil {
1131 0 : v.err = errors.Wrapf(
1132 0 : err, "pebble: backing file %s error", loadInfo.backingFileNum)
1133 0 : }
1134 1 : if v.err == nil && loadInfo.smallestSeqNum == loadInfo.largestSeqNum {
1135 1 : v.reader.Properties.GlobalSeqNum = loadInfo.largestSeqNum
1136 1 : }
1137 1 : if v.err != nil {
1138 0 : c.mu.Lock()
1139 0 : defer c.mu.Unlock()
1140 0 : // Lookup the node in the cache again as it might have already been
1141 0 : // removed.
1142 0 : key := tableCacheKey{dbOpts.cacheID, loadInfo.backingFileNum}
1143 0 : n := c.mu.nodes[key]
1144 0 : if n != nil && n.value == v {
1145 0 : c.releaseNode(n)
1146 0 : }
1147 : }
1148 1 : close(v.loaded)
1149 : }
1150 :
1151 1 : func (v *tableCacheValue) release(c *tableCacheShard) {
1152 1 : <-v.loaded
1153 1 : // Nothing to be done about an error at this point. Close the reader if it is
1154 1 : // open.
1155 1 : if v.reader != nil {
1156 1 : _ = v.reader.Close()
1157 1 : }
1158 1 : c.releasing.Done()
1159 : }
1160 :
1161 : type tableCacheNodeType int8
1162 :
1163 : const (
1164 : tableCacheNodeTest tableCacheNodeType = iota
1165 : tableCacheNodeCold
1166 : tableCacheNodeHot
1167 : )
1168 :
1169 0 : func (p tableCacheNodeType) String() string {
1170 0 : switch p {
1171 0 : case tableCacheNodeTest:
1172 0 : return "test"
1173 0 : case tableCacheNodeCold:
1174 0 : return "cold"
1175 0 : case tableCacheNodeHot:
1176 0 : return "hot"
1177 : }
1178 0 : return "unknown"
1179 : }
1180 :
1181 : type tableCacheNode struct {
1182 : fileNum base.DiskFileNum
1183 : value *tableCacheValue
1184 :
1185 : links struct {
1186 : next *tableCacheNode
1187 : prev *tableCacheNode
1188 : }
1189 : ptype tableCacheNodeType
1190 : // referenced is atomically set to indicate that this entry has been accessed
1191 : // since the last time one of the clock hands swept it.
1192 : referenced atomic.Bool
1193 :
1194 : // Storing the cache id associated with the DB instance here
1195 : // avoids the need to thread the dbOpts struct through many functions.
1196 : cacheID uint64
1197 : }
1198 :
1199 1 : func (n *tableCacheNode) next() *tableCacheNode {
1200 1 : if n == nil {
1201 0 : return nil
1202 0 : }
1203 1 : return n.links.next
1204 : }
1205 :
1206 1 : func (n *tableCacheNode) prev() *tableCacheNode {
1207 1 : if n == nil {
1208 0 : return nil
1209 0 : }
1210 1 : return n.links.prev
1211 : }
1212 :
1213 1 : func (n *tableCacheNode) link(s *tableCacheNode) {
1214 1 : s.links.prev = n.links.prev
1215 1 : s.links.prev.links.next = s
1216 1 : s.links.next = n
1217 1 : s.links.next.links.prev = s
1218 1 : }
1219 :
1220 1 : func (n *tableCacheNode) unlink() *tableCacheNode {
1221 1 : next := n.links.next
1222 1 : n.links.prev.links.next = n.links.next
1223 1 : n.links.next.links.prev = n.links.prev
1224 1 : n.links.prev = n
1225 1 : n.links.next = n
1226 1 : return next
1227 1 : }
|
