Enable skiplist values that are up to 32 bits in length (#472)

Currently, inline skiplist values are limited to a length of 16 bits.
This commit increases that limit to 32 bits. It does this by making
node.value a uint64, and then packing the value offset and size into
it, as two uint32 values.

Because node.value is accessed with atomic.LoadUint64, it must be
aligned on a 64-bit boundary. This is guaranteed by Arena.putNode,
which is changed to always align on a 64-bit boundary rather than
on a pointer boundary (which would incorrectly align on 32-bit
boundary on a 32-bit machine).

Author: andy-kimball

skl/arena.go

@@ -25,7 +25,12 @@ import (
 
 const (
 	offsetSize = int(unsafe.Sizeof(uint32(0)))
-	ptrAlign   = int(unsafe.Sizeof(uintptr(0))) - 1
+
+	// Always align nodes on 64-bit boundaries, even on 32-bit architectures,
+	// so that the node.value field is 64-bit aligned. This is necessary because
+	// node.getValueOffset uses atomic.LoadUint64, which expects its input
+	// pointer to be 64-bit aligned.
+	nodeAlign  = int(unsafe.Sizeof(uint64(0))) - 1
 )
 
 // Arena should be lock-free.
@@ -61,14 +66,14 @@ func (s *Arena) putNode(height int) uint32 {
 	unusedSize := (maxHeight - height) * offsetSize
 
 	// Pad the allocation with enough bytes to ensure pointer alignment.
-	l := uint32(MaxNodeSize - unusedSize + ptrAlign)
+	l := uint32(MaxNodeSize - unusedSize + nodeAlign)
 	n := atomic.AddUint32(&s.n, l)
 	y.AssertTruef(int(n) <= len(s.buf),
 		"Arena too small, toWrite:%d newTotal:%d limit:%d",
 		l, n, len(s.buf))
 
 	// Return the aligned offset.
-	m := (n - l + uint32(ptrAlign)) & ^uint32(ptrAlign)
+	m := (n - l + uint32(nodeAlign)) & ^uint32(nodeAlign)
 	return m
 }
 
@@ -115,8 +120,8 @@ func (s *Arena) getKey(offset uint32, size uint16) []byte {
 
 // getVal returns byte slice at offset. The given size should be just the value
 // size and should NOT include the meta bytes.
-func (s *Arena) getVal(offset uint32, size uint16) (ret y.ValueStruct) {
-	ret.Decode(s.buf[offset : offset+uint32(size)])
+func (s *Arena) getVal(offset uint32, size uint32) (ret y.ValueStruct) {
+	ret.Decode(s.buf[offset : offset+size])
 	return
 }
 

skl/skl.go

@@ -53,9 +53,8 @@ type node struct {
 	// Multiple parts of the value are encoded as a single uint64 so that it
 	// can be atomically loaded and stored:
 	//   value offset: uint32 (bits 0-31)
-	//   value size  : uint16 (bits 32-47)
-	// 12 bytes are allocated to ensure 8 byte alignment also on 32bit systems.
-	value [12]byte
+	//   value size  : uint32 (bits 32-63)
+	value uint64
 
 	// A byte slice is 24 bytes. We are trying to save space here.
 	keyOffset uint32 // Immutable. No need to lock to access key.
@@ -109,17 +108,17 @@ func newNode(arena *Arena, key []byte, v y.ValueStruct, height int) *node {
 	node.keyOffset = arena.putKey(key)
 	node.keySize = uint16(len(key))
 	node.height = uint16(height)
-	*node.value64BitAlignedPtr() = encodeValue(arena.putVal(v), v.EncodedSize())
+	node.value = encodeValue(arena.putVal(v), v.EncodedSize())
 	return node
 }
 
-func encodeValue(valOffset uint32, valSize uint16) uint64 {
+func encodeValue(valOffset uint32, valSize uint32) uint64 {
 	return uint64(valSize)<<32 | uint64(valOffset)
 }
 
-func decodeValue(value uint64) (valOffset uint32, valSize uint16) {
+func decodeValue(value uint64) (valOffset uint32, valSize uint32) {
 	valOffset = uint32(value)
-	valSize = uint16(value >> 32)
+	valSize = uint32(value >> 32)
 	return
 }
 
@@ -135,15 +134,8 @@ func NewSkiplist(arenaSize int64) *Skiplist {
 	}
 }
 
-func (s *node) value64BitAlignedPtr() *uint64 {
-	if uintptr(unsafe.Pointer(&s.value))%8 == 0 {
-		return (*uint64)(unsafe.Pointer(&s.value))
-	}
-	return (*uint64)(unsafe.Pointer(&s.value[4]))
-}
-
-func (s *node) getValueOffset() (uint32, uint16) {
-	value := atomic.LoadUint64(s.value64BitAlignedPtr())
+func (s *node) getValueOffset() (uint32, uint32) {
+	value := atomic.LoadUint64(&s.value)
 	return decodeValue(value)
 }
 
@@ -154,7 +146,7 @@ func (s *node) key(arena *Arena) []byte {
 func (s *node) setValue(arena *Arena, v y.ValueStruct) {
 	valOffset := arena.putVal(v)
 	value := encodeValue(valOffset, v.EncodedSize())
-	atomic.StoreUint64(s.value64BitAlignedPtr(), value)
+	atomic.StoreUint64(&s.value, value)
 }
 
 func (s *node) getNextOffset(h int) uint32 {

y/iterator.go

@@ -47,14 +47,14 @@ func sizeVarint(x uint64) (n int) {
 }
 
 // EncodedSize is the size of the ValueStruct when encoded
-func (v *ValueStruct) EncodedSize() uint16 {
+func (v *ValueStruct) EncodedSize() uint32 {
 	sz := len(v.Value) + 2 // meta, usermeta.
 	if v.ExpiresAt == 0 {
-		return uint16(sz + 1)
+		return uint32(sz + 1)
 	}
 
 	enc := sizeVarint(v.ExpiresAt)
-	return uint16(sz + enc)
+	return uint32(sz + enc)
 }
 
 // Decode uses the length of the slice to infer the length of the Value field.