mirror of
https://github.com/ipfs/kubo.git
synced 2025-07-02 03:28:25 +08:00
feat(fsrepo): protect with a repo lockfile
NB: daemon is one spot the repo lock is typically acquired
This commit is contained in:
@ -12,7 +12,7 @@ import (
|
||||
cmdsHttp "github.com/jbenet/go-ipfs/commands/http"
|
||||
core "github.com/jbenet/go-ipfs/core"
|
||||
commands "github.com/jbenet/go-ipfs/core/commands"
|
||||
daemon "github.com/jbenet/go-ipfs/core/daemon"
|
||||
fsrepo "github.com/jbenet/go-ipfs/repo/fsrepo"
|
||||
util "github.com/jbenet/go-ipfs/util"
|
||||
"github.com/jbenet/go-ipfs/util/debugerror"
|
||||
)
|
||||
@ -89,13 +89,13 @@ func daemonFunc(req cmds.Request) (interface{}, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// acquire the daemon lock _before_ constructing a node. we need to make
|
||||
// acquire the repo lock _before_ constructing a node. we need to make
|
||||
// sure we are permitted to access the resources (datastore, etc.)
|
||||
lock, err := daemon.Lock(req.Context().ConfigRoot)
|
||||
if err != nil {
|
||||
repo := fsrepo.At(req.Context().ConfigRoot)
|
||||
if err := repo.Open(); err != nil {
|
||||
return nil, debugerror.Errorf("Couldn't obtain lock. Is another daemon already running?")
|
||||
}
|
||||
defer lock.Close()
|
||||
defer repo.Close()
|
||||
|
||||
// OK!!! Now we're ready to construct the node.
|
||||
// make sure we construct an online node.
|
||||
|
@ -20,7 +20,6 @@ import (
|
||||
cmdsCli "github.com/jbenet/go-ipfs/commands/cli"
|
||||
cmdsHttp "github.com/jbenet/go-ipfs/commands/http"
|
||||
core "github.com/jbenet/go-ipfs/core"
|
||||
daemon "github.com/jbenet/go-ipfs/core/daemon"
|
||||
repo "github.com/jbenet/go-ipfs/repo"
|
||||
config "github.com/jbenet/go-ipfs/repo/config"
|
||||
fsrepo "github.com/jbenet/go-ipfs/repo/fsrepo"
|
||||
@ -392,7 +391,7 @@ func commandShouldRunOnDaemon(details cmdDetails, req cmds.Request, root *cmds.C
|
||||
|
||||
// at this point need to know whether daemon is running. we defer
|
||||
// to this point so that some commands dont open files unnecessarily.
|
||||
daemonLocked := daemon.Locked(req.Context().ConfigRoot)
|
||||
daemonLocked := fsrepo.LockedByOtherProcess(req.Context().ConfigRoot)
|
||||
|
||||
if daemonLocked {
|
||||
|
||||
|
@ -10,6 +10,7 @@ import (
|
||||
repo "github.com/jbenet/go-ipfs/repo"
|
||||
common "github.com/jbenet/go-ipfs/repo/common"
|
||||
config "github.com/jbenet/go-ipfs/repo/config"
|
||||
lockfile "github.com/jbenet/go-ipfs/repo/fsrepo/lock"
|
||||
opener "github.com/jbenet/go-ipfs/repo/fsrepo/opener"
|
||||
util "github.com/jbenet/go-ipfs/util"
|
||||
debugerror "github.com/jbenet/go-ipfs/util/debugerror"
|
||||
@ -24,10 +25,13 @@ var (
|
||||
// If an operation is used when repo is Open and the operation does not
|
||||
// change the repo's state, the package lock does not need to be acquired.
|
||||
openerCounter *opener.Counter
|
||||
|
||||
lockfiles map[string]io.Closer
|
||||
)
|
||||
|
||||
func init() {
|
||||
openerCounter = opener.NewCounter()
|
||||
lockfiles = make(map[string]io.Closer)
|
||||
}
|
||||
|
||||
// FSRepo represents an IPFS FileSystem Repo. It is not thread-safe.
|
||||
@ -74,6 +78,15 @@ func Remove(path string) error {
|
||||
return os.RemoveAll(path)
|
||||
}
|
||||
|
||||
// LockedByOtherProcess returns true if the FSRepo is locked by another
|
||||
// process. If true, then the repo cannot be opened by this process.
|
||||
func LockedByOtherProcess(repoPath string) bool {
|
||||
openerCounter.Lock()
|
||||
defer openerCounter.Unlock()
|
||||
// NB: the lock is only held when repos are Open
|
||||
return lockfile.Locked(repoPath) && openerCounter.NumOpeners(repoPath) == 0
|
||||
}
|
||||
|
||||
// Open returns an error if the repo is not initialized.
|
||||
func (r *FSRepo) Open() error {
|
||||
openerCounter.Lock()
|
||||
@ -118,9 +131,7 @@ func (r *FSRepo) Open() error {
|
||||
return debugerror.Errorf("logs: %s", err)
|
||||
}
|
||||
|
||||
r.state = opened
|
||||
openerCounter.AddOpener(r.path)
|
||||
return nil
|
||||
return transitionToOpened(r)
|
||||
}
|
||||
|
||||
// Config returns the FSRepo's config. This method must not be called if the
|
||||
@ -217,8 +228,7 @@ func (r *FSRepo) Close() error {
|
||||
if r.state != opened {
|
||||
return debugerror.Errorf("repo is %s", r.state)
|
||||
}
|
||||
openerCounter.RemoveOpener(r.path)
|
||||
return nil // TODO release repo lock
|
||||
return transitionToClosed(r)
|
||||
}
|
||||
|
||||
var _ io.Closer = &FSRepo{}
|
||||
@ -258,3 +268,36 @@ func initCheckDir(path string) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// transitionToOpened manages the state transition to |opened|. Caller must hold
|
||||
// openerCounter lock.
|
||||
func transitionToOpened(r *FSRepo) error {
|
||||
r.state = opened
|
||||
if countBefore := openerCounter.NumOpeners(r.path); countBefore == 0 { // #first
|
||||
closer, err := lockfile.Lock(r.path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lockfiles[r.path] = closer
|
||||
}
|
||||
return openerCounter.AddOpener(r.path)
|
||||
}
|
||||
|
||||
// transitionToClosed manages the state transition to |closed|. Caller must
|
||||
// hold openerCounter lock.
|
||||
func transitionToClosed(r *FSRepo) error {
|
||||
r.state = closed
|
||||
if err := openerCounter.RemoveOpener(r.path); err != nil {
|
||||
return err
|
||||
}
|
||||
if countAfter := openerCounter.NumOpeners(r.path); countAfter == 0 {
|
||||
closer, ok := lockfiles[r.path]
|
||||
if !ok {
|
||||
return errors.New("package error: lockfile is not held")
|
||||
}
|
||||
if err := closer.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
@ -1,4 +1,4 @@
|
||||
package daemon
|
||||
package lock
|
||||
|
||||
import (
|
||||
"io"
|
||||
@ -10,6 +10,7 @@ import (
|
||||
)
|
||||
|
||||
// LockFile is the filename of the daemon lock, relative to config dir
|
||||
// TODO rename repo lock and hide name
|
||||
const LockFile = "daemon.lock"
|
||||
|
||||
func Lock(confdir string) (io.Closer, error) {
|
||||
@ -23,7 +24,6 @@ func Locked(confdir string) bool {
|
||||
}
|
||||
if lk, err := Lock(confdir); err != nil {
|
||||
return true
|
||||
|
||||
} else {
|
||||
lk.Close()
|
||||
return false
|
@ -38,15 +38,17 @@ func (l *Counter) NumOpeners(repoPath string) int {
|
||||
|
||||
// AddOpener messages that an FSRepo holds a handle to the repo at this path.
|
||||
// This method is not thread-safe. The caller must have this object locked.
|
||||
func (l *Counter) AddOpener(repoPath string) {
|
||||
func (l *Counter) AddOpener(repoPath string) error {
|
||||
l.repos[key(repoPath)]++
|
||||
return nil
|
||||
}
|
||||
|
||||
// RemoveOpener messgaes that an FSRepo no longer holds a handle to the repo at
|
||||
// this path. This method is not thread-safe. The caller must have this object
|
||||
// locked.
|
||||
func (l *Counter) RemoveOpener(repoPath string) {
|
||||
func (l *Counter) RemoveOpener(repoPath string) error {
|
||||
l.repos[key(repoPath)]--
|
||||
return nil
|
||||
}
|
||||
|
||||
func key(repoPath string) string {
|
||||
|
Reference in New Issue
Block a user