forked from TrueCloudLab/restic
cmd/restic: implement lock refresh
This commit is contained in:
parent
781ec903e8
commit
159b9e80c4
3 changed files with 70 additions and 18 deletions
|
@ -4,14 +4,21 @@ import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
|
"sync"
|
||||||
"syscall"
|
"syscall"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/restic/restic"
|
"github.com/restic/restic"
|
||||||
"github.com/restic/restic/debug"
|
"github.com/restic/restic/debug"
|
||||||
"github.com/restic/restic/repository"
|
"github.com/restic/restic/repository"
|
||||||
)
|
)
|
||||||
|
|
||||||
var globalLocks []*restic.Lock
|
var globalLocks struct {
|
||||||
|
locks []*restic.Lock
|
||||||
|
cancelRefresh chan struct{}
|
||||||
|
refreshWG sync.WaitGroup
|
||||||
|
sync.Mutex
|
||||||
|
}
|
||||||
|
|
||||||
func lockRepo(repo *repository.Repository) (*restic.Lock, error) {
|
func lockRepo(repo *repository.Repository) (*restic.Lock, error) {
|
||||||
return lockRepository(repo, false)
|
return lockRepository(repo, false)
|
||||||
|
@ -29,35 +36,69 @@ func lockRepository(repo *repository.Repository, exclusive bool) (*restic.Lock,
|
||||||
|
|
||||||
lock, err := lockFn(repo)
|
lock, err := lockFn(repo)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if restic.IsAlreadyLocked(err) {
|
|
||||||
tpe := ""
|
|
||||||
if exclusive {
|
|
||||||
tpe = " exclusive"
|
|
||||||
}
|
|
||||||
fmt.Fprintf(os.Stderr, "unable to acquire%s lock for operation:\n", tpe)
|
|
||||||
fmt.Fprintln(os.Stderr, err)
|
|
||||||
fmt.Fprintf(os.Stderr, "\nthe `unlock` command can be used to remove stale locks\n")
|
|
||||||
os.Exit(1)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
globalLocks = append(globalLocks, lock)
|
globalLocks.Lock()
|
||||||
|
if globalLocks.cancelRefresh == nil {
|
||||||
|
debug.Log("main.lockRepository", "start goroutine for lock refresh")
|
||||||
|
globalLocks.cancelRefresh = make(chan struct{})
|
||||||
|
globalLocks.refreshWG = sync.WaitGroup{}
|
||||||
|
globalLocks.refreshWG.Add(1)
|
||||||
|
go refreshLocks(&globalLocks.refreshWG, globalLocks.cancelRefresh)
|
||||||
|
}
|
||||||
|
|
||||||
|
globalLocks.locks = append(globalLocks.locks, lock)
|
||||||
|
globalLocks.Unlock()
|
||||||
|
|
||||||
return lock, err
|
return lock, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var refreshInterval = 5 * time.Minute
|
||||||
|
|
||||||
|
func refreshLocks(wg *sync.WaitGroup, done <-chan struct{}) {
|
||||||
|
debug.Log("main.refreshLocks", "start")
|
||||||
|
defer func() {
|
||||||
|
wg.Done()
|
||||||
|
globalLocks.Lock()
|
||||||
|
globalLocks.cancelRefresh = nil
|
||||||
|
globalLocks.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
ticker := time.NewTicker(refreshInterval)
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
debug.Log("main.refreshLocks", "terminate")
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
debug.Log("main.refreshLocks", "refreshing locks")
|
||||||
|
globalLocks.Lock()
|
||||||
|
for _, lock := range globalLocks.locks {
|
||||||
|
err := lock.Refresh()
|
||||||
|
if err != nil {
|
||||||
|
fmt.Fprintf(os.Stderr, "unable to refresh lock: %v\n", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
globalLocks.Unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func unlockRepo(lock *restic.Lock) error {
|
func unlockRepo(lock *restic.Lock) error {
|
||||||
|
globalLocks.Lock()
|
||||||
|
defer globalLocks.Unlock()
|
||||||
|
|
||||||
debug.Log("unlockRepo", "unlocking repository")
|
debug.Log("unlockRepo", "unlocking repository")
|
||||||
if err := lock.Unlock(); err != nil {
|
if err := lock.Unlock(); err != nil {
|
||||||
debug.Log("unlockRepo", "error while unlocking: %v", err)
|
debug.Log("unlockRepo", "error while unlocking: %v", err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := 0; i < len(globalLocks); i++ {
|
for i := 0; i < len(globalLocks.locks); i++ {
|
||||||
if lock == globalLocks[i] {
|
if lock == globalLocks.locks[i] {
|
||||||
globalLocks = append(globalLocks[:i], globalLocks[i+1:]...)
|
globalLocks.locks = append(globalLocks.locks[:i], globalLocks.locks[i+1:]...)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -66,8 +107,11 @@ func unlockRepo(lock *restic.Lock) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
func unlockAll() error {
|
func unlockAll() error {
|
||||||
debug.Log("unlockAll", "unlocking %d locks", len(globalLocks))
|
globalLocks.Lock()
|
||||||
for _, lock := range globalLocks {
|
defer globalLocks.Unlock()
|
||||||
|
|
||||||
|
debug.Log("unlockAll", "unlocking %d locks", len(globalLocks.locks))
|
||||||
|
for _, lock := range globalLocks.locks {
|
||||||
if err := lock.Unlock(); err != nil {
|
if err := lock.Unlock(); err != nil {
|
||||||
debug.Log("unlockAll", "error while unlocking: %v", err)
|
debug.Log("unlockAll", "error while unlocking: %v", err)
|
||||||
return err
|
return err
|
||||||
|
|
|
@ -1,10 +1,12 @@
|
||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
|
||||||
"github.com/jessevdk/go-flags"
|
"github.com/jessevdk/go-flags"
|
||||||
|
"github.com/restic/restic"
|
||||||
"github.com/restic/restic/debug"
|
"github.com/restic/restic/debug"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -26,6 +28,10 @@ func main() {
|
||||||
os.Exit(0)
|
os.Exit(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if restic.IsAlreadyLocked(err) {
|
||||||
|
fmt.Fprintf(os.Stderr, "\nthe `unlock` command can be used to remove stale locks\n")
|
||||||
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|
2
lock.go
2
lock.go
|
@ -225,6 +225,7 @@ func (l *Lock) Stale() bool {
|
||||||
// Refresh refreshes the lock by creating a new file in the backend with a new
|
// Refresh refreshes the lock by creating a new file in the backend with a new
|
||||||
// timestamp. Afterwards the old lock is removed.
|
// timestamp. Afterwards the old lock is removed.
|
||||||
func (l *Lock) Refresh() error {
|
func (l *Lock) Refresh() error {
|
||||||
|
debug.Log("Lock.Refresh", "refreshing lock %v", l.lockID.Str())
|
||||||
id, err := l.createLock()
|
id, err := l.createLock()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
@ -235,6 +236,7 @@ func (l *Lock) Refresh() error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
debug.Log("Lock.Refresh", "new lock ID %v", id.Str())
|
||||||
l.lockID = id
|
l.lockID = id
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|
Loading…
Reference in a new issue