@@ -86,13 +86,20 @@ type Skiplist struct {
86
86
testing bool
87
87
}
88
88
89
- // Inserter TODO(peter)
89
+ // An Inserter holds state between consecutive insertions into a skiplist.
90
+ //
91
+ // Inserting a key into a skiplist requires finding the "splice" bracketing the
92
+ // key at every level. The Inserter caches the most recently used splice between
93
+ // inserts, avoiding recomputing the splice from scratch when inserts are
94
+ // sequential. Many workloads commit batches of keys in-order, clustered
95
+ // together within the keyspace, and the Inserter reduces work in these cases.
90
96
type Inserter struct {
91
97
spl [maxHeight ]splice
92
98
height uint32
93
99
}
94
100
95
- // Add TODO(peter)
101
+ // Add inserts a key-value pair into the skiplist. Using an Inserter (rather
102
+ // than calling Skiplist.Add) improves performance for sequential inserts.
96
103
func (ins * Inserter ) Add (list * Skiplist , key base.InternalKey , value []byte ) error {
97
104
return list .addInternal (key , value , ins )
98
105
}
@@ -338,15 +345,17 @@ func (s *Skiplist) newNode(
338
345
339
346
func (s * Skiplist ) randomHeight () uint32 {
340
347
rnd := rand .Uint32 ()
341
-
342
348
h := uint32 (1 )
343
349
for h < maxHeight && rnd <= probabilities [h ] {
344
350
h ++
345
351
}
346
-
347
352
return h
348
353
}
349
354
355
+ // findSplice finds the splice bracketing the given key at every level. The
356
+ // splice is stored on the inserter. If the inserter has a cached splice from a
357
+ // previous call to findSplice, it's used as a starting point for the search.
358
+ // This improves performance for sequential inserts.
350
359
func (s * Skiplist ) findSplice (key base.InternalKey , ins * Inserter ) (found bool ) {
351
360
listHeight := s .Height ()
352
361
var level int
@@ -359,30 +368,46 @@ func (s *Skiplist) findSplice(key base.InternalKey, ins *Inserter) (found bool)
359
368
ins .height = listHeight
360
369
level = int (ins .height )
361
370
} else {
362
- // Our cached height is equal to the list height.
371
+ // Our cached height is equal to the list height. We might be able to
372
+ // use our cached splice if the key is bracketed by the splice. We
373
+ // search for the lowest level with a cached splice that has not been
374
+ // invalidated due to a concurrent insert.
363
375
for ; level < int (listHeight ); level ++ {
364
376
spl := & ins .spl [level ]
365
377
if s .getNext (spl .prev , level ) != spl .next {
366
- // One or more nodes have been inserted between the splice at this
367
- // level.
378
+ // One or more nodes have been inserted between the splice at
379
+ // this level. Keep ascending because the cached splice may
380
+ // still be valid.
368
381
continue
369
382
}
370
- if spl .prev != s .head && ! s .keyIsAfterNode (spl .prev , key ) {
383
+
384
+ // The splice at this level was NOT invalidated. We can stop
385
+ // ascending. If the splice brackets the key, we can reuse all the
386
+ // cached splices at higher levels. Otherwise, we reset the level to
387
+ // the height of the list and we'll descend the skiplist from the
388
+ // top.
389
+ switch {
390
+ case spl .prev != s .head && ! s .keyIsAfterNode (spl .prev , key ):
371
391
// Key lies before splice.
372
392
level = int (listHeight )
373
- break
374
- }
375
- if spl .next != s .tail && s .keyIsAfterNode (spl .next , key ) {
393
+ case spl .next != s .tail && s .keyIsAfterNode (spl .next , key ):
376
394
// Key lies after splice.
377
395
level = int (listHeight )
378
- break
396
+ default :
397
+ // The splice brackets the key! Set prev to the splice's
398
+ // previous node in this level and leave level unchanged. We've
399
+ // avoided needing to search the levels between listHeight and
400
+ // level.
401
+ prev = spl .prev
379
402
}
380
- // The splice brackets the key!
381
- prev = spl .prev
382
403
break
383
404
}
384
405
}
385
406
407
+ // The inserter's cached splice is valid from listHeight to level. We now
408
+ // descend from level to the base level to find the splice in the remaining
409
+ // levels.
410
+
386
411
var next * node
387
412
for level = level - 1 ; level >= 0 ; level -- {
388
413
prevLevelNext := next
0 commit comments