Adds configurable leases to locks to support lock expiry

.vscode/settings.json

@@ -0,0 +1,2 @@
+{
+}

cmd/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"os"
+	"time"
 
 	"github.com/SystemBuilders/LocKey/internal/lockservice"
 	"github.com/SystemBuilders/LocKey/internal/lockservice/node"
@@ -12,7 +13,7 @@ func main() {
 	zerolog.New(os.Stdout).With()
 
 	log := zerolog.New(os.Stdout).With().Logger().Level(zerolog.GlobalLevel())
-	ls := lockservice.NewSimpleLockService(log)
+	ls := lockservice.NewSimpleLockService(log, 5*time.Second)
 
 	scfg := lockservice.NewSimpleConfig("127.0.0.1", "1234")
 	node.Start(ls, *scfg)

internal/lockclient/cache/error.go

@@ -9,6 +9,6 @@ func (e Error) Error() string { return string(e) }
 // Rule of thumb, all errors start with a small letter and end with no full stop.
 const (
 	ErrElementDoesntExist   = Error("element doesn't exist in the cache")
-	ErrElementAlreadyExists = Error("element already exists in the cache")
+	ErrElementAlreadyExists = Error("")
 	ErrCacheDoesntExist     = Error("cache doesn't exist")
 )

internal/lockclient/simple_client.go

@@ -435,7 +435,9 @@ func (sc *SimpleClient) startSession(processID id.ID) {
 		sc.sessionTimers[processID] = timerChan
 		sc.mu.Unlock()
 		// Sessions last for 200ms.
-		time.Sleep(200 * time.Millisecond)
+		// changed to 10s just to test lock expiry
+		// TODO: Make the desired length of session expiry user configurable
+		time.Sleep(10000 * time.Millisecond)
 
 		sc.mu.Lock()
 		sc.sessionTimers[processID] <- struct{}{}

internal/lockclient/simple_client_test.go

@@ -17,7 +17,8 @@ func TestLockService(t *testing.T) {
 
 	log := zerolog.New(os.Stdout).With().Logger().Level(zerolog.GlobalLevel())
 	scfg := lockservice.NewSimpleConfig("http://127.0.0.1", "1234")
-	ls := lockservice.NewSimpleLockService(log)
+	duration := 2 * time.Second // 2 second expiry
+	ls := lockservice.NewSimpleLockService(log, duration)
 
 	quit := make(chan bool, 1)
 	go func() {
@@ -154,14 +155,34 @@ func TestLockService(t *testing.T) {
 			t.Errorf("acquire: got %q want %q", got, want)
 		}
 
-		// Wait for the session to expire
-		time.Sleep(500 * time.Millisecond)
+		// Wait for the lock's lease to expire
+		time.Sleep(3 * time.Second)
+
 		got = sc.Release(d, session)
-		want = ErrSessionNonExistent
+		want = lockservice.ErrCantReleaseFile
 		if got != want {
 			t.Errorf("release: got %q want %q", got, want)
 		}
 	})
+	t.Run("try acquiring after lock expiry; should succeed", func(t *testing.T) {
+		sc := NewSimpleClient(scfg, log, nil)
+		session := sc.Connect()
+		d := lockservice.NewObjectDescriptor("test2")
+
+		got := sc.Acquire(d, session)
+		var want error
+		if got != want {
+			t.Errorf("acquire: got %q want %q", got, want)
+		}
+
+		// Wait for the lock's lease to expire
+		time.Sleep(3 * time.Second)
+
+		got = sc.Acquire(d, session)
+		if got != want {
+			t.Errorf("acquire: got %q want %q", got, want)
+		}
+	})
 
 	quit <- true
 	return
@@ -173,7 +194,7 @@ func BenchmarkLocKeyWithoutCache(b *testing.B) {
 
 	log := zerolog.New(os.Stdout).With().Logger().Level(zerolog.GlobalLevel())
 	scfg := lockservice.NewSimpleConfig("http://127.0.0.1", "1234")
-	ls := lockservice.NewSimpleLockService(log)
+	ls := lockservice.NewSimpleLockService(log, 5)
 
 	quit := make(chan bool, 1)
 	go func() {
@@ -211,7 +232,8 @@ func BenchmarkLocKeyWithCache(b *testing.B) {
 
 	log := zerolog.New(os.Stdout).With().Logger().Level(zerolog.GlobalLevel())
 	scfg := lockservice.NewSimpleConfig("http://127.0.0.1", "1234")
-	ls := lockservice.NewSimpleLockService(log)
+	duration := 2 * time.Second // 2 second expiry
+	ls := lockservice.NewSimpleLockService(log, duration)
 
 	quit := make(chan bool, 1)
 	go func() {

internal/lockservice/simpleLockService.go

@@ -1,15 +1,27 @@
 package lockservice
 
 import (
+	"fmt"
 	"sync"
+	"time"
 
 	"github.com/rs/zerolog"
 )
 
 // SafeLockMap is the lockserver's data structure
 type SafeLockMap struct {
-	LockMap map[string]string
-	Mutex   sync.Mutex
+	LockMap       map[string]*LockMapEntry
+	LeaseDuration time.Duration
+	Mutex         sync.Mutex
+}
+
+// LockMapEntry defines the structure for objects placed
+// in the LockMap. It consists of the owner of the lock
+// that is acquired and the timestamp at which the
+// acquisition took place.
+type LockMapEntry struct {
+	owner     string
+	timestamp time.Time
 }
 
 // SimpleConfig implements Config.
@@ -87,6 +99,14 @@ func (sd *LockDescriptor) Owner() string {
 	return sd.UserID
 }
 
+// NewLockMapEntry returns an instance of a LockMapEntry
+func NewLockMapEntry(owner string, timestamp time.Time) *LockMapEntry {
+	return &LockMapEntry{
+		owner:     owner,
+		timestamp: timestamp,
+	}
+}
+
 // NewSimpleConfig returns an instance of the SimpleConfig.
 func NewSimpleConfig(IPAddr, PortAddr string) *SimpleConfig {
 	return &SimpleConfig{
@@ -111,63 +131,100 @@ func NewObjectDescriptor(ObjectID string) *ObjectDescriptor {
 }
 
 // NewSimpleLockService creates and returns a new lock service ready to use.
-func NewSimpleLockService(log zerolog.Logger) *SimpleLockService {
+func NewSimpleLockService(log zerolog.Logger, leaseDuration time.Duration) *SimpleLockService {
 	safeLockMap := &SafeLockMap{
-		LockMap: make(map[string]string),
+		LockMap: make(map[string]*LockMapEntry),
 	}
+	safeLockMap.LeaseDuration = leaseDuration
 	return &SimpleLockService{
 		log:     log,
 		lockMap: safeLockMap,
 	}
 }
+func fmtDuration(d time.Duration) string {
+	d = d.Round(time.Minute)
+	h := d / time.Hour
+	d -= h * time.Hour
+	ms := d / time.Microsecond
+	return fmt.Sprintf("%02d:%02d", h, ms)
+}
+
+// getCurrentDuration returns the duration between the current time
+// and the time at which a lock was required
+func compareDuration(timestamp time.Time, lease time.Duration) bool {
+	// fmt.Printf("current: %s timestamp: %s duration: %s %s\n", time.Now().String(), timestamp.String(), fmtDuration(time.Now().Sub(timestamp)))
+	intDuration := int64(time.Now().Sub(timestamp))
+	intLease := int64(lease)
+	fmt.Printf("%d %d \n", intDuration, intLease)
+	if time.Now().Sub(timestamp) > lease {
+		return true
+	}
+	return false
+}
 
 // Acquire function lets a client acquire a lock on an object.
 func (ls *SimpleLockService) Acquire(sd Descriptors) error {
 	ls.lockMap.Mutex.Lock()
-	if _, ok := ls.lockMap.LockMap[sd.ID()]; ok {
+
+	// If the lock is not present in the LockMap or
+	// the lock has expired, then allow the acquisition
+	if _, ok := ls.lockMap.LockMap[sd.ID()]; !ok || (ok && (compareDuration(ls.lockMap.LockMap[sd.ID()].timestamp, ls.lockMap.LeaseDuration))) {
+		ls.lockMap.LockMap[sd.ID()] = NewLockMapEntry(sd.Owner(), time.Now())
 		ls.lockMap.Mutex.Unlock()
 		ls.
 			log.
 			Debug().
 			Str("descriptor", sd.ID()).
-			Msg("can't acquire, already been acquired")
-		return ErrFileacquired
+			Str("owner", ls.lockMap.LockMap[sd.ID()].owner).
+			Time("timestamp", ls.lockMap.LockMap[sd.ID()].timestamp).
+			Msg("locked")
+		return nil
 	}
-	ls.lockMap.LockMap[sd.ID()] = sd.Owner()
 	ls.lockMap.Mutex.Unlock()
 	ls.
 		log.
 		Debug().
 		Str("descriptor", sd.ID()).
-		Str("owner", sd.Owner()).
-		Msg("locked")
-	return nil
+		Msg("can't acquire, already been acquired")
+	return ErrFileacquired
+
 }
 
 // Release lets a client to release a lock on an object.
+// TODO: Prevent a lock from being released if it has expired. !!!!!!
 func (ls *SimpleLockService) Release(sd Descriptors) error {
 	ls.lockMap.Mutex.Lock()
 	// Only the entity that posseses the lock for this object
 	// is allowed to release the lock
-	if ls.lockMap.LockMap[sd.ID()] == sd.Owner() {
-		delete(ls.lockMap.LockMap, sd.ID())
+	if _, ok := ls.lockMap.LockMap[sd.ID()]; !ok {
 		ls.
 			log.
 			Debug().
 			Str("descriptor", sd.ID()).
-			Str("owner", sd.Owner()).
-			Msg("released")
+			Msg("can't release, hasn't been acquired")
 		ls.lockMap.Mutex.Unlock()
-		return nil
-	} else if _, ok := ls.lockMap.LockMap[sd.ID()]; !ok {
+		return ErrCantReleaseFile
+
+	} else if compareDuration(ls.lockMap.LockMap[sd.ID()].timestamp, ls.lockMap.LeaseDuration) {
+		// lease expired
 		ls.
 			log.
 			Debug().
 			Str("descriptor", sd.ID()).
-			Msg("can't release, hasn't been acquired")
+			Msg("can't release, lease of lock has expired")
 		ls.lockMap.Mutex.Unlock()
 		return ErrCantReleaseFile
 
+	} else if ls.lockMap.LockMap[sd.ID()].owner == sd.Owner() {
+		delete(ls.lockMap.LockMap, sd.ID())
+		ls.
+			log.
+			Debug().
+			Str("descriptor", sd.ID()).
+			Str("owner", sd.Owner()).
+			Msg("released")
+		ls.lockMap.Mutex.Unlock()
+		return nil
 	} else {
 		ls.
 			log.
@@ -186,14 +243,14 @@ func (ls *SimpleLockService) Release(sd Descriptors) error {
 func (ls *SimpleLockService) CheckAcquired(sd Descriptors) (string, bool) {
 	ls.lockMap.Mutex.Lock()
 	id := sd.ID()
-	if owner, ok := ls.lockMap.LockMap[id]; ok {
+	if entry, ok := ls.lockMap.LockMap[id]; ok {
 		ls.lockMap.Mutex.Unlock()
 		ls.
 			log.
 			Debug().
 			Str("descriptor", id).
 			Msg("checkacquire success")
-		return owner, true
+		return entry.owner, true
 	}
 	ls.
 		log.