Initial commit

go.mod

···
+module github.com/jwkohnen/lrmap
+
+go 1.16

lrmap.go

···
+package lrmap
+
+import (
+	"fmt"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+)
+
+type (
+	LRMap struct {
+		mu           sync.Mutex
+		left         mapT
+		right        mapT
+		readMap      unsafe.Pointer
+		writeMap     *mapT
+		opLog        []op
+		readHandlers map[*ReadHandler]struct{}
+	}
+
+	mapT map[Key]Value
+
+	Key   int
+	Value int
+)
+
+func New() *LRMap {
+	m := &LRMap{
+		left:         make(map[Key]Value),
+		right:        make(map[Key]Value),
+		readHandlers: make(map[*ReadHandler]struct{}),
+	}
+
+	m.swap()
+
+	return m
+}
+
+func (m *LRMap) Set(k Key, v Value) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	(*(m.writeMap))[k] = v
+
+	m.opLog = append(m.opLog, op{t: opSet, k: k, v: &v})
+}
+
+func (m *LRMap) Delete(k Key) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	delete(*(m.writeMap), k)
+
+	m.opLog = append(m.opLog, op{t: opDelete, k: k})
+}
+
+func (m *LRMap) Flush() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.swap()
+
+	m.waitForReaders()
+
+	// redo all operations from the op log into the new write map (old read map) to sync up.
+	for _, o := range m.opLog {
+		switch o.t {
+		case opSet:
+			(*(m.writeMap))[o.k] = *(o.v)
+		case opDelete:
+			delete(*(m.writeMap), o.k)
+		default:
+			panic(fmt.Errorf("op(%d) not implemented", o.t))
+		}
+	}
+
+	// drop the op log completely and let the GC remove all references to stale keys and values
+	m.opLog = nil
+}
+
+func (m *LRMap) NewReadHandler() *ReadHandler {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	rh := &ReadHandler{lrm: m}
+
+	m.readHandlers[rh] = struct{}{}
+
+	// sharp-edged finalizer: https://crawshaw.io/blog/sharp-edged-finalizers
+	_, file, line, _ := runtime.Caller(1)
+	runtime.SetFinalizer(rh, func(*ReadHandler) {
+		panic(fmt.Errorf("reader illegal state: must call Close() (created at %s:%d)", file, line))
+	})
+
+	return rh
+}
+
+func (m *LRMap) swap() {
+	var (
+		r = (*mapT)(m.readMap)
+		w *mapT
+	)
+
+	switch r {
+	case nil:
+		// initial case: start with left map as reader, right map as writer
+		r = &(m.left)
+		w = &(m.right)
+	case &(m.left):
+		r = &(m.right)
+		w = &(m.left)
+	case &(m.right):
+		r = &(m.left)
+		w = &(m.right)
+	default:
+		panic(fmt.Errorf("illegal pointer: %v", r))
+	}
+
+	atomic.StorePointer(&(m.readMap), unsafe.Pointer(r))
+	m.writeMap = w
+}
+
+func (m *LRMap) waitForReaders() {
+	readers := make(map[*ReadHandler]uint64)
+	for rh := range m.readHandlers {
+		epoch := atomic.LoadUint64(&(rh.epoch))
+		if epoch%2 == 1 {
+			readers[rh] = epoch
+		}
+	}
+
+	delay := time.Microsecond
+
+	for {
+		for reader, epoch := range readers {
+			if e := atomic.LoadUint64(&(reader.epoch)); e != epoch {
+				delete(readers, reader)
+			}
+		}
+
+		if len(readers) == 0 {
+			return
+		}
+
+		time.Sleep(delay)
+
+		delay *= 10
+		if delay > 5*time.Second {
+			delay = 5 * time.Second
+		}
+
+	}
+}
+
+func (m *LRMap) getMapAtomic() mapT {
+	return *(*mapT)(atomic.LoadPointer(&(m.readMap)))
+}
+
+type opType int8
+
+const (
+	opSet opType = iota
+	opDelete
+)
+
+type op struct {
+	t opType
+	k Key
+	v *Value
+}
+
+type ReadHandler struct {
+	lrm    *LRMap
+	live   mapT
+	epoch  uint64
+	closed bool
+}
+
+func (r *ReadHandler) Enter() {
+	atomic.AddUint64(&r.epoch, 1)
+	r.live = r.lrm.getMapAtomic()
+}
+
+func (r *ReadHandler) Leave() {
+	atomic.AddUint64(&r.epoch, 1)
+}
+
+func (r *ReadHandler) Get(k Key) Value {
+	v, _ := r.GetOK(k)
+	return v
+}
+
+func (r *ReadHandler) GetOK(k Key) (Value, bool) {
+	if r.epoch%2 == 0 {
+		panic("reader illegal state: must Enter() before operating on data")
+	}
+
+	v, ok := r.live[k]
+	return v, ok
+}
+
+func (r *ReadHandler) Close() {
+	r.lrm.mu.Lock()
+	defer r.lrm.mu.Unlock()
+
+	if r.closed {
+		panic("reader illegal state: must not Close() more than once")
+	}
+
+	delete(r.lrm.readHandlers, r)
+	runtime.SetFinalizer(r, nil)
+	r.closed = true
+}

lrmap_test.go

···
+package lrmap
+
+import (
+	"math"
+	"testing"
+)
+
+func TestHundred(t *testing.T) {
+	lrm := New()
+
+	rh := lrm.NewReadHandler()
+	defer rh.Close()
+
+	for i := 1; i < 100; i += 2 {
+		k, v := Key(i), Value(i)
+
+		lrm.Set(k, v)
+	}
+
+	lrm.Flush()
+
+	for i := 1; i < 100; i += 2 {
+		k, v := Key(i), Value(i)
+
+		rh.Enter()
+		if _v := rh.Get(k); _v != v {
+
+			t.Errorf("Get(%d) want %d, got %d", k, v, _v)
+		}
+		rh.Leave()
+	}
+
+	for i := 0; i < 100; i += 2 {
+		k, v := Key(i), Value(i)
+		lrm.Set(k, v)
+	}
+
+	lrm.Flush()
+
+	for i := 0; i < 100; i++ {
+		rh.Enter()
+		k, v := Key(i), Value(i)
+
+		if _v := rh.Get(k); _v != v {
+			t.Errorf("Get(%d) want %d, got %d", k, v, _v)
+		}
+		rh.Leave()
+	}
+
+	lrm.Flush()
+
+	rh.Enter()
+	for i := 0; i < 100; i++ {
+		k, v := Key(i), Value(i)
+
+		if _v := rh.Get(k); _v != v {
+			t.Errorf("Get(%d) want %d, got %d", k, v, _v)
+		}
+	}
+	rh.Leave()
+
+	for i := 0; i < 100; i += 2 {
+		k := Key(i)
+		lrm.Delete(k)
+	}
+
+	lrm.Flush()
+
+	rh.Enter()
+	for i := 0; i < 100; i++ {
+		k, v := Key(i), Value(i)
+
+		_v, ok := rh.GetOK(k)
+		if i%2 == 0 {
+			if ok || _v != 0 {
+				t.Errorf("GetOK(%d), want (0, false), got (%d, %t)", i, _v, ok)
+			}
+		} else {
+			if !ok || _v != v {
+				t.Errorf("GetOK(%d), want (%d, true), got (%d, %t)", i, v, _v, ok)
+			}
+		}
+	}
+	rh.Leave()
+
+	lrm.Flush()
+
+	rh.Enter()
+	for i := 0; i < 100; i++ {
+		k, v := Key(i), Value(i)
+
+		_v, ok := rh.GetOK(k)
+		if i%2 == 0 {
+			if ok || _v != 0 {
+				t.Errorf("GetOK(%d), want (0, false), got (%d, %t)", i, _v, ok)
+			}
+		} else {
+			if !ok || _v != v {
+				t.Errorf("GetOK(%d), want (%d, true), got (%d, %t)", i, v, _v, ok)
+			}
+		}
+	}
+	rh.Leave()
+}
+
+func TestEpochOverflow(t *testing.T) {
+	var maxEven uint64 = math.MaxUint64 - 1
+	t.Logf("maxEven(%d)", maxEven)
+	if maxEven%2 == 1 {
+		t.Errorf("maxEven(%d) is odd", maxEven)
+	}
+
+	maxOdd := maxEven + 1
+	t.Logf("maxOdd(%d)", maxOdd)
+	if maxOdd%2 == 0 {
+		t.Errorf("maxOdd(%d) is even", maxOdd)
+	}
+
+	overflow := maxOdd + 1
+	t.Logf("overflow(%d)", overflow)
+	if overflow%2 == 1 {
+		t.Errorf("overflow(%d) is odd", overflow)
+	}
+}