feat(lock): add timeout to repository lock feature

smothiki · smothiki · commit 6fdcf08064d9 · 2016-06-02T16:05:12.000-06:00
diff --git a/boot.go b/boot.go
@@ -53,7 +53,7 @@ func main() {
 				}
 				fs := sys.RealFS()
 				env := sys.RealEnv()
-				pushLock := sshd.NewInMemoryRepositoryLock()
+				pushLock := sshd.NewInMemoryRepositoryLock(cnf.GitLockTimeout())
 				circ := sshd.NewCircuit()
 
 				storageParams, err := conf.GetStorageParams(env)
diff --git a/pkg/sshd/config.go b/pkg/sshd/config.go
@@ -14,9 +14,15 @@ type Config struct {
 	StorageType                  string `envconfig:"BUILDER_STORAGE" default:"minio"`
 	SlugBuilderImagePullPolicy   string `envconfig:"SLUG_BUILDER_IMAGE_PULL_POLICY" default:"Always"`
 	DockerBuilderImagePullPolicy string `envconfig:"DOCKER_BUILDER_IMAGE_PULL_POLICY" default:"Always"`
+	LockTimeout                  int    `envconfig:"GIT_LOCK_TIMEOUT" default:"10"`
 }
 
 // CleanerPollSleepDuration returns c.CleanerPollSleepDurationSec as a time.Duration.
 func (c Config) CleanerPollSleepDuration() time.Duration {
 	return time.Duration(c.CleanerPollSleepDurationSec) * time.Second
 }
+
+//GitLockTimeout return LockTimeout in minutes
+func (c Config) GitLockTimeout() time.Duration {
+	return time.Duration(c.LockTimeout) * time.Minute
+}
diff --git a/pkg/sshd/lock.go b/pkg/sshd/lock.go
@@ -14,41 +14,57 @@ var (
 // RepositoryLock interface that allows the creation of a lock associated
 // with a repository name to avoid simultaneous git operations.
 type RepositoryLock interface {
-	// Lock acquires a lock for a repository. In the case the repository is already locked
-	// it waits until a timeout to get the lock. If it was not possible to get the
-	// lock after the timeout an error is returned.
-	Lock(repoName string, timeout time.Duration) error
+	// Lock acquires a lock for a repository.
+	Lock(repoName string) error
 	// Unlock releases the lock for a repository or returns an error if the specified
-	// name doesn't exist. In the case the repository is already locked it waits until
-	// a timeout to get the lock. If it was not possible to get the lock after the timeout
-	// an error is returned.
-	Unlock(repoName string, timeout time.Duration) error
+	// name doesn't exist.
+	Unlock(repoName string) error
+	// Timeout returns the time duration for which it has to hold the lock
+	Timeout() time.Duration
 }
 
-func wrapInLock(lck RepositoryLock, repoName string, timeout time.Duration, fn func() error) error {
-	if err := lck.Lock(repoName, timeout); err != nil {
+func wrapInLock(lck RepositoryLock, repoName string, fn func() error) error {
+	if err := lck.Lock(repoName); err != nil {
 		return errAlreadyLocked
 	}
-	defer lck.Unlock(repoName, timeout)
-	return fn()
+	timer := time.NewTimer(lck.Timeout())
+	defer timer.Stop()
+	doneCh := make(chan struct{})
+	fnCh := make(chan error)
+	go func() {
+		err := fn()
+		select {
+		case fnCh <- err:
+		case <-doneCh:
+		}
+	}()
+	select {
+	case <-timer.C:
+		lck.Unlock(repoName)
+		defer close(doneCh)
+		return fmt.Errorf("%s lock exceeded timout", repoName)
+	case err := <-fnCh:
+		return err
+	}
 }
 
 // NewInMemoryRepositoryLock returns a new instance of a RepositoryLock.
-func NewInMemoryRepositoryLock() RepositoryLock {
+func NewInMemoryRepositoryLock(timeout time.Duration) RepositoryLock {
 	return &inMemoryRepoLock{
 		mutex:   &sync.RWMutex{},
 		dataMap: make(map[string]bool),
+		timeout: timeout,
 	}
 }
 
 type inMemoryRepoLock struct {
 	mutex   *sync.RWMutex
 	dataMap map[string]bool
+	timeout time.Duration
 }
 
 // Lock aquires a lock associated with the specified name.
-// This implementation ignores the timeout.
-func (rl *inMemoryRepoLock) Lock(repoName string, timeout time.Duration) error {
+func (rl *inMemoryRepoLock) Lock(repoName string) error {
 	rl.mutex.Lock()
 	defer rl.mutex.Unlock()
 
@@ -62,8 +78,8 @@ func (rl *inMemoryRepoLock) Lock(repoName string, timeout time.Duration) error {
 }
 
 // Unlock releases the lock for a repository or returns an error if the specified name doesn't
-// exist. This implementation ignores the timeout.
-func (rl *inMemoryRepoLock) Unlock(repoName string, timeout time.Duration) error {
+// exist.
+func (rl *inMemoryRepoLock) Unlock(repoName string) error {
 	rl.mutex.Lock()
 	defer rl.mutex.Unlock()
 
@@ -78,3 +94,8 @@ func (rl *inMemoryRepoLock) Unlock(repoName string, timeout time.Duration) error
 
 	return nil
 }
+
+// Timeout returns the time duration for which a gitpush should hold the lock
+func (rl *inMemoryRepoLock) Timeout() time.Duration {
+	return rl.timeout
+}
diff --git a/pkg/sshd/lock_test.go b/pkg/sshd/lock_test.go
@@ -15,59 +15,59 @@ const (
 func TestMultipleSameRepoLocks(t *testing.T) {
 	var wg sync.WaitGroup
 	const repo = "repo1"
-	const numTries = 100
-	lck := NewInMemoryRepositoryLock()
-	assert.NoErr(t, lck.Lock(repo, 0*time.Second))
+	const numTries = 0
+	lck := NewInMemoryRepositoryLock(0)
+	assert.NoErr(t, lck.Lock(repo))
 	for i := 0; i < numTries; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			assert.True(t, lck.Lock(repo, 0*time.Second) != nil, "lock of already locked repo should return error")
+			assert.True(t, lck.Lock(repo) != nil, "lock of already locked repo should return error")
 		}()
 	}
 	assert.NoErr(t, waitWithTimeout(&wg, 1*time.Second))
-	assert.NoErr(t, lck.Unlock(repo, 0*time.Second))
+	assert.NoErr(t, lck.Unlock(repo))
 	for i := 0; i < numTries; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			assert.True(t, lck.Unlock(repo, 0*time.Second) != nil, "unlock of already unlocked repo should return error")
+			assert.True(t, lck.Unlock(repo) != nil, "unlock of already unlocked repo should return error")
 		}()
 	}
 	assert.NoErr(t, waitWithTimeout(&wg, 1*time.Second))
 }
 
 func TestSingleLock(t *testing.T) {
-	rl := NewInMemoryRepositoryLock()
+	rl := NewInMemoryRepositoryLock(0)
 	key := "fakeid"
 	callbackCh := make(chan interface{})
 	go lockAndCallback(rl, key, callbackCh)
 	verifyCallbackHappens(t, callbackCh)
 }
 
 func TestSingleLockUnlock(t *testing.T) {
-	rl := NewInMemoryRepositoryLock()
+	rl := NewInMemoryRepositoryLock(0)
 	key := "fakeid"
 	callbackCh := make(chan interface{})
 	go lockAndCallback(rl, key, callbackCh)
 	verifyCallbackHappens(t, callbackCh)
-	err := rl.Unlock(key, time.Duration(0))
+	err := rl.Unlock(key)
 	if err != nil {
 		t.Fatalf("unexpected error %v", err)
 	}
 }
 
 func TestInvalidUnlock(t *testing.T) {
-	rl := NewInMemoryRepositoryLock()
+	rl := NewInMemoryRepositoryLock(0)
 	key := "fakeid"
-	err := rl.Unlock(key, time.Duration(0))
+	err := rl.Unlock(key)
 	if err == nil {
 		t.Fatal("expected error but returned nil")
 	}
 }
 
 func TestDoubleLockUnlock(t *testing.T) {
-	rl := NewInMemoryRepositoryLock()
+	rl := NewInMemoryRepositoryLock(0)
 	key := "fakeid"
 	callbackCh1stLock := make(chan interface{})
 	callbackCh2ndLock := make(chan interface{})
@@ -76,29 +76,29 @@ func TestDoubleLockUnlock(t *testing.T) {
 	verifyCallbackHappens(t, callbackCh1stLock)
 	go lockAndCallback(rl, key, callbackCh2ndLock)
 	verifyCallbackDoesntHappens(t, callbackCh2ndLock)
-	err := rl.Unlock(key, time.Duration(0))
+	err := rl.Unlock(key)
 	if err != nil {
 		t.Fatalf("unexpected error %v", err)
 	}
-	err = rl.Unlock(key, time.Duration(0))
+	err = rl.Unlock(key)
 	if err == nil {
 		t.Fatalf("expected error but returned nil")
 	}
 }
 
 func TestWrapInLock(t *testing.T) {
-	lck := NewInMemoryRepositoryLock()
-	assert.NoErr(t, wrapInLock(lck, "repo", 0*time.Second, func() error {
+	lck := NewInMemoryRepositoryLock(0)
+	assert.NoErr(t, wrapInLock(lck, "repo", func() error {
 		return nil
 	}))
-	lck.Lock("repo", 0*time.Second)
-	assert.Err(t, errAlreadyLocked, wrapInLock(lck, "repo", 0*time.Second, func() error {
+	lck.Lock("repo")
+	assert.Err(t, errAlreadyLocked, wrapInLock(lck, "repo", func() error {
 		return nil
 	}))
 }
 
 func lockAndCallback(rl RepositoryLock, id string, callbackCh chan<- interface{}) {
-	if err := rl.Lock(id, time.Duration(0)); err == nil {
+	if err := rl.Lock(id); err == nil {
 		callbackCh <- true
 	}
 }
diff --git a/pkg/sshd/server.go b/pkg/sshd/server.go
@@ -10,7 +10,6 @@ import (
 	"io/ioutil"
 	"net"
 	"strings"
-	"time"
 
 	"github.com/deis/builder/pkg/controller"
 	"github.com/deis/builder/pkg/git"
@@ -248,7 +247,7 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, conda
 					channel.Stderr().Write([]byte("No repo given"))
 					return err
 				}
-				wrapErr := wrapInLock(s.pushLock, repoName, time.Duration(0), s.runReceive(req, sshconn, channel, repoName, parts, condata))
+				wrapErr := wrapInLock(s.pushLock, repoName, s.runReceive(req, sshconn, channel, repoName, parts, condata))
 				if wrapErr == errAlreadyLocked {
 					log.Info(multiplePush)
 					// The error must be in git format
@@ -262,7 +261,7 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, conda
 
 				var xs uint32
 				if wrapErr != nil {
-					log.Err("Failed git receive: %v", err)
+					log.Err("Failed git receive: %v", wrapErr)
 					xs = 1
 				}
 				sendExitStatus(xs, channel)
diff --git a/pkg/sshd/server_test.go b/pkg/sshd/server_test.go
@@ -21,6 +21,7 @@ func TestGitPktLine(t *testing.T) {
 	str := "hello world"
 	err := gitPktLine(b, str)
 	assert.NoErr(t, err)
+
 	outStr := string(b.Bytes())
 	assert.True(t, len(outStr) > 4, "output string <= 4 chars")
 	assert.Equal(t, outStr[:4], fmt.Sprintf("%04x", len(str)+4), "hex prefix")
@@ -62,7 +63,7 @@ func TestReceive(t *testing.T) {
 	cfg.AddHostKey(key)
 
 	c := NewCircuit()
-	pushLock := NewInMemoryRepositoryLock()
+	pushLock := NewInMemoryRepositoryLock(0)
 	runServer(cfg, c, pushLock, testingServerAddr, time.Duration(0), t)
 
 	// Give server time to initialize.
@@ -116,7 +117,7 @@ func TestPushInvalidArgsLength(t *testing.T) {
 	cfg.AddHostKey(key)
 
 	c := NewCircuit()
-	pushLock := NewInMemoryRepositoryLock()
+	pushLock := NewInMemoryRepositoryLock(0)
 	runServer(cfg, c, pushLock, testingServerAddr, 0*time.Second, t)
 
 	// Give server time to initialize.
@@ -152,7 +153,7 @@ func TestConcurrentPushSameRepo(t *testing.T) {
 	cfg.AddHostKey(key)
 
 	c := NewCircuit()
-	pushLock := NewInMemoryRepositoryLock()
+	pushLock := NewInMemoryRepositoryLock(0)
 	runServer(cfg, c, pushLock, testingServerAddr, 2*time.Second, t)
 
 	// Give server time to initialize.
@@ -216,7 +217,7 @@ func TestConcurrentPushDifferentRepo(t *testing.T) {
 	assert.NoErr(t, err)
 	cfg.AddHostKey(key)
 	c := NewCircuit()
-	pushLock := NewInMemoryRepositoryLock()
+	pushLock := NewInMemoryRepositoryLock(time.Duration(1 * time.Minute))
 	runServer(cfg, c, pushLock, testingServerAddr, time.Duration(0), t)
 	time.Sleep(200 * time.Millisecond)
 	assert.Equal(t, c.State(), ClosedState, "circuit state")

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ func main() {`
`53`	`53`	`}`
`54`	`54`	`fs := sys.RealFS()`
`55`	`55`	`env := sys.RealEnv()`
`56`		`- pushLock := sshd.NewInMemoryRepositoryLock()`
	`56`	`+ pushLock := sshd.NewInMemoryRepositoryLock(cnf.GitLockTimeout())`
`57`	`57`	`circ := sshd.NewCircuit()`
`58`	`58`
`59`	`59`	`storageParams, err := conf.GetStorageParams(env)`