pss: Refactor. Step 2. Refactor forward cache (ethersphere#1742)

pss: Refactored forwardCache into a TTLSet + add mockable timer

Author: jpeletier

go.mod

@@ -71,6 +71,7 @@ require (
 	github.com/steakknife/bloomfilter v0.0.0-20180922174646-6819c0d2a570 // indirect
 	github.com/steakknife/hamming v0.0.0-20180906055917-c99c65617cd3 // indirect
 	github.com/syndtr/goleveldb v0.0.0-20190318030020-c3a204f8e965
+	github.com/tilinna/clock v1.0.2
 	github.com/tyler-smith/go-bip39 v0.0.0-20181017060643-dbb3b84ba2ef // indirect
 	github.com/uber-go/atomic v1.4.0 // indirect
 	github.com/uber/jaeger-client-go v0.0.0-20180607151842-f7e0d4744fa6

go.sum

@@ -249,6 +249,8 @@ github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/syndtr/goleveldb v0.0.0-20190318030020-c3a204f8e965 h1:V/AztY/q2oW5ghho7YMgUJQkKvSACHRxpeDyT5DxpIo=
 github.com/syndtr/goleveldb v0.0.0-20190318030020-c3a204f8e965/go.mod h1:9OrXJhf154huy1nPWmuSrkgjPUtUNhA+Zmy+6AESzuA=
+github.com/tilinna/clock v1.0.2 h1:6BO2tyAC9JbPExKH/z9zl44FLu1lImh3nDNKA0kgrkI=
+github.com/tilinna/clock v1.0.2/go.mod h1:ZsP7BcY7sEEz7ktc0IVy8Us6boDrK8VradlKRUGfOao=
 github.com/tyler-smith/go-bip39 v0.0.0-20181017060643-dbb3b84ba2ef h1:luEzjJzktS9eU0CmI0uApXHLP/lKzOoRPrJhd71J8ik=
 github.com/tyler-smith/go-bip39 v0.0.0-20181017060643-dbb3b84ba2ef/go.mod h1:sJ5fKU0s6JVwZjjcUEX2zFOnvq0ASQ2K9Zr6cf67kNs=
 github.com/uber-go/atomic v1.4.0 h1:yOuPqEq4ovnhEjpHmfFwsqBXDYbQeT6Nb0bwD6XnD5o=

pss/internal/ticker/ticker.go

@@ -0,0 +1,54 @@
+package ticker
+
+import (
+	"errors"
+	"time"
+
+	"github.com/tilinna/clock"
+)
+
+// Config defines the necessary information and dependencies to instantiate a Ticker
+type Config struct {
+	Clock    clock.Clock
+	Interval time.Duration
+	Callback func()
+}
+
+// Ticker represents a periodic timer that invokes a callback
+type Ticker struct {
+	quitC chan struct{}
+}
+
+// ErrAlreadyStopped is returned if this service was already stopped and Stop() is called again
+var ErrAlreadyStopped = errors.New("Already stopped")
+
+// New builds a ticker that will call the given callback function periodically
+func New(config *Config) *Ticker {
+
+	tk := &Ticker{
+		quitC: make(chan struct{}),
+	}
+	ticker := config.Clock.NewTicker(config.Interval)
+	go func() {
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				config.Callback()
+			case <-tk.quitC:
+				return
+			}
+		}
+	}()
+	return tk
+}
+
+// Stop stops the timer and releases the goroutine running it.
+func (tk *Ticker) Stop() error {
+	if tk.quitC == nil {
+		return ErrAlreadyStopped
+	}
+	close(tk.quitC)
+	tk.quitC = nil
+	return nil
+}

pss/internal/ticker/ticker_test.go

@@ -0,0 +1,47 @@
+package ticker_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethersphere/swarm/pss/internal/ticker"
+	"github.com/tilinna/clock"
+)
+
+// TestNewTicker tests whether the ticker calls a callback function periodically
+func TestNewTicker(t *testing.T) {
+	var err error
+
+	testClock := clock.NewMock(time.Unix(0, 0))
+	interval := 10 * time.Second
+
+	wg := sync.WaitGroup{}
+	wg.Add(10)
+	tickWait := make(chan bool)
+
+	testTicker := ticker.New(&ticker.Config{
+		Interval: interval,
+		Clock:    testClock,
+		Callback: func() {
+			wg.Done()
+			tickWait <- true
+		},
+	})
+
+	for i := 0; i < 10; i++ {
+		testClock.Add(interval)
+		<-tickWait
+	}
+
+	wg.Wait()
+	err = testTicker.Stop()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = testTicker.Stop()
+	if err != ticker.ErrAlreadyStopped {
+		t.Fatal("Expected Stop() to return ticker.ErrAlreadyStopped when trying to stop an already stopped ticker")
+	}
+}

pss/internal/ttlset/ttlset.go

@@ -0,0 +1,84 @@
+package ttlset
+
+import (
+	"sync"
+	"time"
+
+	"github.com/tilinna/clock"
+)
+
+// Config defines the TTLSet configuration
+type Config struct {
+	EntryTTL time.Duration // time after which items are removed
+	Clock    clock.Clock   // time reference
+}
+
+// TTLSet implements a Set that automatically removes expired keys
+// after a predefined expiration time
+type TTLSet struct {
+	Config
+	set  map[interface{}]setEntry
+	lock sync.RWMutex
+}
+
+type setEntry struct {
+	expiresAt time.Time
+}
+
+// New instances a TTLSet
+func New(config *Config) *TTLSet {
+	ts := &TTLSet{
+		set:    make(map[interface{}]setEntry),
+		Config: *config,
+	}
+	return ts
+}
+
+// Add adds a new key to the set
+func (ts *TTLSet) Add(key interface{}) error {
+	var entry setEntry
+	var ok bool
+
+	ts.lock.Lock()
+	defer ts.lock.Unlock()
+
+	if entry, ok = ts.set[key]; !ok {
+		entry = setEntry{}
+	}
+	entry.expiresAt = ts.Clock.Now().Add(ts.EntryTTL)
+	ts.set[key] = entry
+	return nil
+}
+
+// Has returns whether or not a key is already/still in the set
+func (ts *TTLSet) Has(key interface{}) bool {
+	ts.lock.Lock()
+	defer ts.lock.Unlock()
+
+	entry, ok := ts.set[key]
+	if ok {
+		if entry.expiresAt.After(ts.Clock.Now()) {
+			return true
+		}
+		delete(ts.set, key) // since we're holding the lock, take the chance to delete a expired record
+	}
+	return false
+}
+
+// GC will remove expired entries from the set
+func (ts *TTLSet) GC() {
+	ts.lock.Lock()
+	defer ts.lock.Unlock()
+	for k, v := range ts.set {
+		if v.expiresAt.Before(ts.Clock.Now()) {
+			delete(ts.set, k)
+		}
+	}
+}
+
+// Count returns the number of entries in the set
+func (ts *TTLSet) Count() int {
+	ts.lock.Lock()
+	defer ts.lock.Unlock()
+	return len(ts.set)
+}

pss/internal/ttlset/ttlset_test.go

@@ -0,0 +1,150 @@
+package ttlset_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ethersphere/swarm/pss/internal/ttlset"
+	"github.com/tilinna/clock"
+)
+
+func TestTTLSet(t *testing.T) {
+	var err error
+
+	testClock := clock.NewMock(time.Unix(0, 0))
+
+	testEntryTTL := 10 * time.Second
+	testSet := ttlset.New(&ttlset.Config{
+		EntryTTL: testEntryTTL,
+		Clock:    testClock,
+	})
+
+	key1 := "some key"
+	key2 := "some other key"
+
+	// check adding a key to the set
+	err = testSet.Add(key1)
+	if err != nil {
+		t.Fatal((err))
+	}
+
+	// check if the key is now there:
+	hasKey := testSet.Has(key1)
+	if !(hasKey == true) {
+		t.Fatal("key1 should've been in the set, but Has() returned false")
+	}
+
+	// check if Has() returns false when asked about a key that was never added:
+	hasKey = testSet.Has("some made up key")
+	if !(hasKey == false) {
+		t.Fatal("Has() should have returned false when presented with a key that was never added")
+	}
+
+	// Let some time pass, but not enough to have the key expire:
+	testClock.Add(testEntryTTL / 2)
+
+	// check if the key is still there:
+	hasKey = testSet.Has(key1)
+	if !(hasKey == true) {
+		t.Fatal("key1 should've been in the set, but Has() returned false")
+	}
+
+	// Let some time pass well beyond the expiry time, so key1 expires:
+	testClock.Add(testEntryTTL * 2)
+
+	// Add another key to the set:
+	err = testSet.Add(key2)
+	if err != nil {
+		t.Fatal((err))
+	}
+
+	hasKey = testSet.Has(key1)
+	if !(hasKey == false) {
+		t.Fatal("key1 should've been removed from the set, but Has() returned true")
+	}
+
+	hasKey = testSet.Has(key2)
+	if !(hasKey == true) {
+		t.Fatal("key should remain in the set, but Has() returned false")
+	}
+
+	// Let some time pass well beyond key2's expiry time, so key2 expires:
+	testClock.Add(testEntryTTL * 2)
+
+	hasKey = testSet.Has(key2)
+	if !(hasKey == false) {
+		t.Fatal("key2 should have been wiped, but Has() returned true")
+	}
+}
+
+func TestGC(t *testing.T) {
+	var err error
+
+	testClock := clock.NewMock(time.Unix(0, 0))
+
+	testEntryTTL := 10 * time.Second
+	testSet := ttlset.New(&ttlset.Config{
+		EntryTTL: testEntryTTL,
+		Clock:    testClock,
+	})
+
+	key1 := "some key"
+	key2 := "some later key"
+
+	// check adding a message to the cache
+	err = testSet.Add(key1)
+	if err != nil {
+		t.Fatal((err))
+	}
+
+	// move the clock 2 seconds
+	testClock.Add(2 * time.Second)
+
+	// add a second key which will have a later expiration time
+	err = testSet.Add(key2)
+	if err != nil {
+		t.Fatal((err))
+	}
+
+	count := testSet.Count()
+	if !(count == 2) {
+		t.Fatal("Expected the set to contain 2 keys")
+	}
+
+	testSet.GC() // attempt a cleanup. This cleanup should not affect any of the two keys, since they are not expired.
+
+	count = testSet.Count()
+	if !(count == 2) {
+		t.Fatal("Expected the set to still contain 2 keys")
+	}
+
+	//Now, move the clock forward 9 seconds. This will expire key1 but still keep key2
+	testClock.Add(9 * time.Second)
+	testSet.GC() // invoke the internal cleaning function, which should wipe only key1
+	count = testSet.Count()
+	if !(count == 1) {
+		t.Fatal("Expected the set to now have only 1 key")
+	}
+
+	//Verify if key1 was wiped but key2 persists:
+	hasKey := testSet.Has(key1)
+	if !(hasKey == false) {
+		t.Fatal("Expected the set to have removed key1")
+	}
+
+	hasKey = testSet.Has(key2)
+	if !(hasKey == true) {
+		t.Fatal("Expected the set to still contain key2")
+	}
+
+	//Now, move the clock some more time. This will wipe key2
+	testClock.Add(7 * time.Second)
+	testSet.GC() // invoke the internal cleaning function, which should wipe only key1
+
+	count = testSet.Count()
+	// verify the map is now empty
+	if !(count == 0) {
+		t.Fatal("Expected the set to be empty")
+	}
+
+}