feat: add in-memory lock

cmd/layotto/main.go

@@ -109,6 +109,7 @@ import (
 	"mosn.io/layotto/components/lock"
 	lock_consul "mosn.io/layotto/components/lock/consul"
 	lock_etcd "mosn.io/layotto/components/lock/etcd"
+	lock_inmemory "mosn.io/layotto/components/lock/in-memory"
 	lock_mongo "mosn.io/layotto/components/lock/mongo"
 	lock_redis "mosn.io/layotto/components/lock/redis"
 	lock_zookeeper "mosn.io/layotto/components/lock/zookeeper"
@@ -346,6 +347,9 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_lock.NewFactory("mongo", func() lock.LockStore {
 				return lock_mongo.NewMongoLock(log.DefaultLogger)
 			}),
+			runtime_lock.NewFactory("in-memory", func() lock.LockStore {
+				return lock_inmemory.NewInMemoryLock()
+			}),
 		),
 
 		// bindings

cmd/layotto_multiple_api/main.go

@@ -103,6 +103,7 @@ import (
 	"mosn.io/layotto/components/lock"
 	lock_consul "mosn.io/layotto/components/lock/consul"
 	lock_etcd "mosn.io/layotto/components/lock/etcd"
+	lock_inmemory "mosn.io/layotto/components/lock/in-memory"
 	lock_redis "mosn.io/layotto/components/lock/redis"
 	lock_zookeeper "mosn.io/layotto/components/lock/zookeeper"
 	runtime_lock "mosn.io/layotto/pkg/runtime/lock"
@@ -340,6 +341,9 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_lock.NewFactory("consul", func() lock.LockStore {
 				return lock_consul.NewConsulLock(log.DefaultLogger)
 			}),
+			runtime_lock.NewFactory("in-memory", func() lock.LockStore {
+				return lock_inmemory.NewInMemoryLock()
+			}),
 		),
 
 		// bindings

components/go.mod

@@ -18,11 +18,11 @@ require (
 	github.com/hashicorp/consul/api v1.3.0
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/minio/minio-go/v7 v7.0.15
-	github.com/mitchellh/mapstructure v1.4.1
+	github.com/mitchellh/mapstructure v1.4.1 // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/spf13/afero v1.2.2 // indirect
 	github.com/stretchr/testify v1.7.0
-	github.com/tencentyun/cos-go-sdk-v5 v0.7.33 // indirect
+	github.com/tencentyun/cos-go-sdk-v5 v0.7.33
 	github.com/valyala/fasthttp v1.26.0
 	github.com/zouyx/agollo/v4 v4.0.7
 	go.beyondstorage.io/services/hdfs v0.3.0
@@ -31,6 +31,7 @@ require (
 	go.etcd.io/etcd/client/v3 v3.5.0
 	go.etcd.io/etcd/server/v3 v3.5.0
 	go.mongodb.org/mongo-driver v1.8.0
+	go.uber.org/atomic v1.7.0
 	golang.org/x/net v0.0.0-20210614182718-04defd469f4e // indirect
 	golang.org/x/oauth2 v0.0.0-20201208152858-08078c50e5b5 // indirect
 	google.golang.org/grpc v1.38.0

components/go.sum

@@ -506,13 +506,8 @@ github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvW
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.10.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
-github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/klauspost/compress v1.13.1 h1:wXr2uRxZTJXHLly6qhJabee5JqIhTRoLBhDOA74hDEQ=
 github.com/klauspost/compress v1.13.1/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
-github.com/klauspost/compress v1.13.5/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
-github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
-github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/cpuid v1.2.3/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid v1.3.1 h1:5JNjFYYQrZeKRJ0734q51WCEEn2huer72Dc7K+R/b6s=
 github.com/klauspost/cpuid v1.3.1/go.mod h1:bYW4mA6ZgKPob1/Dlai2LviZJO7KGI3uoWLd42rAQw4=

components/lock/in-memory/in_memory_lock.go

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * 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 in_memory
+
+import (
+	"go.uber.org/atomic"
+	"mosn.io/layotto/components/lock"
+	"sync"
+	"time"
+)
+
+type InMemoryLock struct {
+	features []lock.Feature
+	data     *sync.Map
+	wLock    sync.Mutex
+}
+
+type lockData struct {
+	key        string
+	owner      string
+	expireTime time.Time
+	lock       *atomic.Int32
+}
+
+func NewInMemoryLock() *InMemoryLock {
+	return &InMemoryLock{
+		features: make([]lock.Feature, 0),
+		data:     &sync.Map{},
+	}
+}
+
+func (s *InMemoryLock) Init(_ lock.Metadata) error {
+	return nil
+}
+
+func (s *InMemoryLock) Features() []lock.Feature {
+	return s.features
+}
+
+func (s *InMemoryLock) TryLock(req *lock.TryLockRequest) (*lock.TryLockResponse, error) {
+	item, ok := s.data.Load(req.ResourceId)
+	if !ok {
+		newItem := &lockData{
+			key:  req.ResourceId,
+			lock: &atomic.Int32{},
+		}
+		s.wLock.Lock()
+		item, _ = s.data.LoadOrStore(req.ResourceId, newItem)
+		s.wLock.Unlock()
+	}
+
+	//0 unlock, 1 lock
+	d := item.(*lockData)
+
+	//check expire
+	if d.owner != "" && time.Now().Before(d.expireTime) {
+		s.wLock.Lock()
+		//double check
+		s.data.Delete(req.ResourceId)
+		item = &lockData{
+			key:  req.ResourceId,
+			lock: &atomic.Int32{},
+		}
+		s.data.Store(req.ResourceId, item)
+		s.wLock.Unlock()
+	}
+
+	if !d.lock.CAS(0, 1) {
+		//lock failed
+		return &lock.TryLockResponse{
+			Success: false,
+		}, nil
+	}
+
+	d.owner = req.LockOwner
+	d.expireTime = time.Now().Add(time.Second * time.Duration(req.Expire))
+	return &lock.TryLockResponse{
+		Success: true,
+	}, nil
+}
+
+func (s *InMemoryLock) Unlock(req *lock.UnlockRequest) (*lock.UnlockResponse, error) {
+	item, ok := s.data.Load(req.ResourceId)
+
+	if !ok {
+		return &lock.UnlockResponse{
+			Status: lock.LOCK_UNEXIST,
+		}, nil
+	}
+
+	d := item.(*lockData)
+	if d.lock.Load() != 1 {
+		return &lock.UnlockResponse{
+			Status: lock.SUCCESS,
+		}, nil
+	}
+
+	if d.owner != req.LockOwner {
+		return &lock.UnlockResponse{
+			Status: lock.LOCK_BELONG_TO_OTHERS,
+		}, nil
+	}
+
+	if !d.lock.CAS(1, 0) {
+		return &lock.UnlockResponse{
+			Status: lock.LOCK_UNEXIST,
+		}, nil
+	}
+
+	d.owner = ""
+	return &lock.UnlockResponse{
+		Status: lock.SUCCESS,
+	}, nil
+}

components/lock/in-memory/in_memory_lock_test.go

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2021 Layotto Authors
+ *
+ * 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 in_memory
+
+import (
+	"github.com/stretchr/testify/assert"
+	"mosn.io/layotto/components/lock"
+	"testing"
+	"time"
+)
+
+func TestNew(t *testing.T) {
+	s := NewInMemoryLock()
+	assert.NotNil(t, s)
+}
+
+func TestInit(t *testing.T) {
+	s := NewInMemoryLock()
+	assert.NotNil(t, s)
+
+	err := s.Init(lock.Metadata{})
+	assert.NoError(t, err)
+}
+
+func TestFeatures(t *testing.T) {
+	s := NewInMemoryLock()
+	assert.NotNil(t, s)
+
+	f := s.Features()
+	assert.NotNil(t, f)
+	assert.Equal(t, 0, len(f))
+}
+
+func TestTryLock(t *testing.T) {
+	s := NewInMemoryLock()
+	assert.NotNil(t, s)
+
+	req := &lock.TryLockRequest{
+		ResourceId: "key111",
+		LockOwner:  "own",
+		Expire:     3,
+	}
+
+	var err error
+	var resp *lock.TryLockResponse
+	resp, err = s.TryLock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.True(t, resp.Success)
+
+	resp, err = s.TryLock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.False(t, resp.Success)
+
+	req = &lock.TryLockRequest{
+		ResourceId: "key112",
+		LockOwner:  "own",
+		Expire:     1,
+	}
+
+	resp, err = s.TryLock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.True(t, resp.Success)
+
+	req = &lock.TryLockRequest{
+		ResourceId: "key112",
+		LockOwner:  "own",
+		Expire:     1,
+	}
+
+	resp, err = s.TryLock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.False(t, resp.Success)
+
+	time.Sleep(2 * time.Second)
+
+	resp, err = s.TryLock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.True(t, resp.Success)
+
+}
+
+func TestUnLock(t *testing.T) {
+	s := NewInMemoryLock()
+	assert.NotNil(t, s)
+
+	req := &lock.UnlockRequest{
+		ResourceId: "key111",
+		LockOwner:  "own",
+	}
+
+	var err error
+	var resp *lock.UnlockResponse
+	resp, err = s.Unlock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.Equal(t, lock.LOCK_UNEXIST, resp.Status)
+
+	lockReq := &lock.TryLockRequest{
+		ResourceId: "key111",
+		LockOwner:  "own",
+		Expire:     10,
+	}
+
+	var lockResp *lock.TryLockResponse
+	lockResp, err = s.TryLock(lockReq)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.True(t, lockResp.Success)
+
+	resp, err = s.Unlock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.Equal(t, lock.SUCCESS, resp.Status)
+
+	lockResp, err = s.TryLock(lockReq)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.True(t, lockResp.Success)
+
+	req.LockOwner = "1"
+
+	resp, err = s.Unlock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.Equal(t, lock.LOCK_BELONG_TO_OTHERS, resp.Status)
+
+	req.ResourceId = "11"
+	resp, err = s.Unlock(req)
+	assert.NoError(t, err)
+	assert.NotNil(t, req)
+	assert.Equal(t, lock.LOCK_UNEXIST, resp.Status)
+
+}

configs/config_in_memory.json

@@ -32,6 +32,12 @@
                           }
                         }
                       },
+                      "lock": {
+                        "in-memory": {
+                          "metadata": {
+                          }
+                        }
+                      },
                       "pub_subs": {
                         "in-memory": {
                           "metadata": {