Cache managment (#49)

* code cleanup

* add cache invalidation

* extract cache into own module

* extract mutex

* add mutex test

* fix tests

* fix rename

* cache is internally managed

* test mutexes and bugfix

* add cache test

* add headers

* add sanity check

* cleanup test

* fix it could be 0

Author: devbugging

blockchain/cache.go

@@ -0,0 +1,80 @@
+/*
+ * Flow Playground
+ *
+ * Copyright 2019 Dapper Labs, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package blockchain
+
+import (
+	"fmt"
+	"github.com/dapperlabs/flow-playground-api/model"
+	"github.com/getsentry/sentry-go"
+	"github.com/golang/groupcache/lru"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+)
+
+// newCache returns a new instance of cache with provided capacity.
+func newCache(capacity int) *cache {
+	return &cache{
+		cache: lru.New(capacity),
+	}
+}
+
+type cache struct {
+	cache *lru.Cache
+}
+
+// reset the cache for the ID.
+func (c *cache) reset(ID uuid.UUID) {
+	c.cache.Remove(ID)
+}
+
+// get returns a cached emulator if exists, but also checks if it's stale.
+//
+// based on the executions the function receives it compares that to the emulator block height, since
+// one execution is always one block it can compare the heights to the length. If it finds some executions
+// that are not part of emulator it returns that subset, so they can be applied on top.
+func (c *cache) get(
+	ID uuid.UUID,
+	executions []*model.TransactionExecution,
+) (blockchain, []*model.TransactionExecution, error) {
+	val, ok := c.cache.Get(ID)
+	if !ok {
+		return nil, executions, nil
+	}
+
+	emulator := val.(blockchain)
+	latest, err := emulator.getLatestBlock()
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "cache failure")
+	}
+
+	// this should never happen, sanity check
+	if int(latest.Header.Height) > len(executions) {
+		err := fmt.Errorf("cache failure, block height is higher than executions count")
+		sentry.CaptureException(err)
+		return nil, nil, err
+	}
+
+	// this will return only executions that are missing from the emulator
+	return emulator, executions[latest.Header.Height:], nil
+}
+
+// add new entry in the cache.
+func (c *cache) add(ID uuid.UUID, emulator blockchain) {
+	c.cache.Add(ID, emulator)
+}

blockchain/cache_test.go

@@ -0,0 +1,118 @@
+/*
+ * Flow Playground
+ *
+ * Copyright 2019 Dapper Labs, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package blockchain
+
+import (
+	"fmt"
+	"github.com/dapperlabs/flow-playground-api/model"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func createExecutions(count int) []*model.TransactionExecution {
+	executions := make([]*model.TransactionExecution, count)
+	for i := 0; i < count; i++ {
+		executions[i] = &model.TransactionExecution{
+			ProjectChildID: model.NewProjectChildID(uuid.New(), uuid.New()),
+			Index:          i,
+			Script:         fmt.Sprintf(`transaction { execute { log(%d) } }`, i),
+		}
+	}
+	return executions
+}
+
+func Test_Cache(t *testing.T) {
+
+	t.Run("returns cached emulator", func(t *testing.T) {
+		testID := uuid.New()
+		c := newCache(2)
+
+		em, err := newEmulator()
+		require.NoError(t, err)
+
+		c.add(testID, em)
+
+		cacheEm, exe, err := c.get(testID, nil)
+		require.NoError(t, err)
+		assert.Len(t, exe, 0)
+
+		cacheBlock, err := cacheEm.getLatestBlock()
+		require.NoError(t, err)
+
+		block, err := em.getLatestBlock()
+		require.NoError(t, err)
+
+		assert.Equal(t, block.ID(), cacheBlock.ID())
+	})
+
+	t.Run("returns cached emulator with executions", func(t *testing.T) {
+		testID := uuid.New()
+		c := newCache(2)
+
+		em, err := newEmulator()
+		require.NoError(t, err)
+
+		c.add(testID, em)
+
+		executions := createExecutions(5)
+		for _, exe := range executions {
+			_, _, err := em.executeTransaction(exe.Script, exe.Arguments, nil)
+			require.NoError(t, err)
+		}
+
+		cachedEm, cacheExe, err := c.get(testID, executions)
+		require.NoError(t, err)
+		// cached emulator contains all the executions
+		assert.Len(t, cacheExe, 0)
+		// make sure emulators are same
+		cacheBlock, _ := cachedEm.getLatestBlock()
+		block, _ := em.getLatestBlock()
+		assert.Equal(t, cacheBlock.ID(), block.ID())
+	})
+
+	t.Run("returns cached emulator with missing executions", func(t *testing.T) {
+		testID := uuid.New()
+		c := newCache(2)
+
+		em, err := newEmulator()
+		require.NoError(t, err)
+
+		c.add(testID, em)
+
+		executions := createExecutions(5)
+
+		for i, exe := range executions {
+			if i == 3 {
+				break // miss last two executions
+			}
+			_, _, err := em.executeTransaction(exe.Script, exe.Arguments, nil)
+			require.NoError(t, err)
+		}
+
+		_, cacheExe, err := c.get(testID, executions)
+		require.NoError(t, err)
+
+		// cached emulator missed two executions
+		assert.Len(t, cacheExe, 2)
+		assert.Equal(t, 3, cacheExe[0].Index)
+		assert.Equal(t, 4, cacheExe[1].Index)
+	})
+}

blockchain/emulator.go

@@ -81,7 +81,7 @@ func newEmulator() (*emulator, error) {
 		emu.WithTransactionFeesEnabled(false),
 	)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "failed to create a new emulator instance")
 	}
 
 	return &emulator{

blockchain/mutex.go

@@ -0,0 +1,54 @@
+package blockchain
+
+import (
+	"github.com/getsentry/sentry-go"
+	"github.com/google/uuid"
+	"sync"
+)
+
+func newMutex() *mutex {
+	return &mutex{}
+}
+
+// mutex contains locking logic for projects.
+//
+// this custom implementation of mutex creates per project ID mutex lock, this is needed because
+// we need to restrict access to common resource (emulator) based on project ID, and we can not put a mutex lock
+// on the emulator instance since it takes time to load the emulator in the first place, during which racing conditions may occur.
+// Mutex keeps a map of mutex locks per project ID, and it also keeps a track of obtained locks per that ID so it can, after all
+// the locks have been released remove that lock from the mutex map to not pollute memory.
+type mutex struct {
+	mu        sync.Map
+	muCounter sync.Map
+}
+
+// load retrieves the mutex lock by the project ID and increase the usage counter.
+func (m *mutex) load(uuid uuid.UUID) *sync.RWMutex {
+	counter, _ := m.muCounter.LoadOrStore(uuid, 0)
+	m.muCounter.Store(uuid, counter.(int)+1)
+
+	mu, _ := m.mu.LoadOrStore(uuid, &sync.RWMutex{})
+	return mu.(*sync.RWMutex)
+}
+
+// remove returns the mutex lock by the project ID and decreases usage counter, deleting the map entry if at 0.
+func (m *mutex) remove(uuid uuid.UUID) *sync.RWMutex {
+	mu, ok := m.mu.Load(uuid)
+	if !ok {
+		sentry.CaptureMessage("trying to access non-existing mutex")
+	}
+
+	counter, ok := m.muCounter.Load(uuid)
+	if !ok {
+		sentry.CaptureMessage("trying to access non-existing mutex counter")
+	}
+
+	if counter == 1 { // if last one remove it after
+		m.mu.Delete(uuid)
+		m.muCounter.Delete(uuid)
+	} else {
+		m.muCounter.Store(uuid, counter.(int)-1)
+	}
+
+	return mu.(*sync.RWMutex)
+}

blockchain/mutex_test.go

@@ -0,0 +1,65 @@
+/*
+ * Flow Playground
+ *
+ * Copyright 2019 Dapper Labs, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package blockchain
+
+import (
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func Test_Mutex(t *testing.T) {
+	mutex := newMutex()
+
+	testUuid := uuid.New()
+
+	m := mutex.load(testUuid)
+	m.Lock()
+
+	v, _ := mutex.muCounter.Load(testUuid)
+	assert.Equal(t, 1, v.(int))
+
+	_, exists := mutex.mu.Load(testUuid)
+	assert.True(t, exists)
+
+	m1 := mutex.load(testUuid)
+	locked := m1.TryLock()
+	// should fail since we already have one lock
+	assert.False(t, locked)
+
+	v, _ = mutex.muCounter.Load(testUuid)
+	assert.Equal(t, 2, v.(int))
+
+	mutex.remove(testUuid).Unlock()
+
+	v, _ = mutex.muCounter.Load(testUuid)
+	assert.Equal(t, 1, v.(int))
+
+	locked = m1.TryLock()
+	assert.True(t, locked) // should succeed now
+
+	mutex.remove(testUuid).Unlock()
+
+	// after all locks are released there shouldn't be any counter left
+	_, found := mutex.muCounter.Load(testUuid)
+	assert.False(t, found)
+
+	_, found = mutex.mu.Load(testUuid)
+	assert.False(t, found)
+}