-// A UnixVolume is a Volume backed by a locally mounted disk.
-//
package main
import (
+ "bufio"
+ "errors"
+ "flag"
"fmt"
"io"
"io/ioutil"
"log"
"os"
"path/filepath"
+ "regexp"
"strconv"
"strings"
+ "sync"
"syscall"
"time"
)
-// IORequests are encapsulated Get or Put requests. They are used to
-// implement serialized I/O (i.e. only one read/write operation per
-// volume). When running in serialized mode, the Keep front end sends
-// IORequests on a channel to an IORunner, which handles them one at a
-// time and returns an IOResponse.
-//
-type IOMethod int
-
-const (
- KeepGet IOMethod = iota
- KeepPut
-)
-
-type IORequest struct {
- method IOMethod
- loc string
- data []byte
- reply chan *IOResponse
+type unixVolumeAdder struct {
+ *volumeSet
}
-type IOResponse struct {
- data []byte
- err error
-}
-
-// A UnixVolume has the following properties:
-//
-// root
-// the path to the volume's root directory
-// queue
-// A channel of IORequests. If non-nil, all I/O requests for
-// this volume should be queued on this channel; the result
-// will be delivered on the IOResponse channel supplied in the
-// request.
-//
-type UnixVolume struct {
- root string // path to this volume
- queue chan *IORequest
- readonly bool
-}
-
-func (v *UnixVolume) IOHandler() {
- for req := range v.queue {
- var result IOResponse
- switch req.method {
- case KeepGet:
- result.data, result.err = v.Read(req.loc)
- case KeepPut:
- result.err = v.Write(req.loc, req.data)
+func (vs *unixVolumeAdder) Set(value string) error {
+ if dirs := strings.Split(value, ","); len(dirs) > 1 {
+ log.Print("DEPRECATED: using comma-separated volume list.")
+ for _, dir := range dirs {
+ if err := vs.Set(dir); err != nil {
+ return err
+ }
}
- req.reply <- &result
+ return nil
}
-}
-
-func MakeUnixVolume(root string, serialize bool, readonly bool) *UnixVolume {
- v := &UnixVolume{
- root: root,
- queue: nil,
- readonly: readonly,
+ if len(value) == 0 || value[0] != '/' {
+ return errors.New("Invalid volume: must begin with '/'.")
}
- if serialize {
- v.queue = make(chan *IORequest)
- go v.IOHandler()
+ if _, err := os.Stat(value); err != nil {
+ return err
+ }
+ var locker sync.Locker
+ if flagSerializeIO {
+ locker = &sync.Mutex{}
}
- return v
+ *vs.volumeSet = append(*vs.volumeSet, &UnixVolume{
+ root: value,
+ locker: locker,
+ readonly: flagReadonly,
+ })
+ return nil
}
-func (v *UnixVolume) Get(loc string) ([]byte, error) {
- if v.queue == nil {
- return v.Read(loc)
- }
- reply := make(chan *IOResponse)
- v.queue <- &IORequest{KeepGet, loc, nil, reply}
- response := <-reply
- return response.data, response.err
+func init() {
+ flag.Var(
+ &unixVolumeAdder{&volumes},
+ "volumes",
+ "Deprecated synonym for -volume.")
+ flag.Var(
+ &unixVolumeAdder{&volumes},
+ "volume",
+ "Local storage directory. Can be given more than once to add multiple directories. If none are supplied, the default is to use all directories named \"keep\" that exist in the top level directory of a mount point at startup time. Can be a comma-separated list, but this is deprecated: use multiple -volume arguments instead.")
}
-func (v *UnixVolume) Put(loc string, block []byte) error {
- if v.readonly {
- return MethodDisabledError
+// Discover adds a UnixVolume for every directory named "keep" that is
+// located at the top level of a device- or tmpfs-backed mount point
+// other than "/". It returns the number of volumes added.
+func (vs *unixVolumeAdder) Discover() int {
+ added := 0
+ f, err := os.Open(ProcMounts)
+ if err != nil {
+ log.Fatalf("opening %s: %s", ProcMounts, err)
}
- if v.queue == nil {
- return v.Write(loc, block)
+ scanner := bufio.NewScanner(f)
+ for scanner.Scan() {
+ args := strings.Fields(scanner.Text())
+ if err := scanner.Err(); err != nil {
+ log.Fatalf("reading %s: %s", ProcMounts, err)
+ }
+ dev, mount := args[0], args[1]
+ if mount == "/" {
+ continue
+ }
+ if dev != "tmpfs" && !strings.HasPrefix(dev, "/dev/") {
+ continue
+ }
+ keepdir := mount + "/keep"
+ if st, err := os.Stat(keepdir); err != nil || !st.IsDir() {
+ continue
+ }
+ // Set the -readonly flag (but only for this volume)
+ // if the filesystem is mounted readonly.
+ flagReadonlyWas := flagReadonly
+ for _, fsopt := range strings.Split(args[3], ",") {
+ if fsopt == "ro" {
+ flagReadonly = true
+ break
+ }
+ if fsopt == "rw" {
+ break
+ }
+ }
+ if err := vs.Set(keepdir); err != nil {
+ log.Printf("adding %q: %s", keepdir, err)
+ } else {
+ added++
+ }
+ flagReadonly = flagReadonlyWas
}
- reply := make(chan *IOResponse)
- v.queue <- &IORequest{KeepPut, loc, block, reply}
- response := <-reply
- return response.err
+ return added
}
+// A UnixVolume stores and retrieves blocks in a local directory.
+type UnixVolume struct {
+ // path to the volume's root directory
+ root string
+ // something to lock during IO, typically a sync.Mutex (or nil
+ // to skip locking)
+ locker sync.Locker
+ readonly bool
+}
+
+// Touch sets the timestamp for the given locator to the current time
func (v *UnixVolume) Touch(loc string) error {
if v.readonly {
return MethodDisabledError
return err
}
defer f.Close()
+ if v.locker != nil {
+ v.locker.Lock()
+ defer v.locker.Unlock()
+ }
if e := lockfile(f); e != nil {
return e
}
defer unlockfile(f)
- now := time.Now().Unix()
- utime := syscall.Utimbuf{now, now}
- return syscall.Utime(p, &utime)
+ ts := syscall.NsecToTimespec(time.Now().UnixNano())
+ return syscall.UtimesNano(p, []syscall.Timespec{ts, ts})
}
+// Mtime returns the stored timestamp for the given locator.
func (v *UnixVolume) Mtime(loc string) (time.Time, error) {
p := v.blockPath(loc)
- if fi, err := os.Stat(p); err != nil {
+ fi, err := os.Stat(p)
+ if err != nil {
return time.Time{}, err
- } else {
- return fi.ModTime(), nil
}
+ return fi.ModTime(), nil
}
-// Read retrieves a block identified by the locator string "loc", and
-// returns its contents as a byte slice.
-//
-// If the block could not be opened or read, Read returns a nil slice
-// and the os.Error that was generated.
-//
-// If the block is present but its content hash does not match loc,
-// Read returns the block and a CorruptError. It is the caller's
-// responsibility to decide what (if anything) to do with the
-// corrupted data block.
-//
-func (v *UnixVolume) Read(loc string) ([]byte, error) {
- buf, err := ioutil.ReadFile(v.blockPath(loc))
- return buf, err
+// Lock the locker (if one is in use), open the file for reading, and
+// call the given function if and when the file is ready to read.
+func (v *UnixVolume) getFunc(path string, fn func(io.Reader) error) error {
+ if v.locker != nil {
+ v.locker.Lock()
+ defer v.locker.Unlock()
+ }
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return fn(f)
}
-// Write stores a block of data identified by the locator string
+// stat is os.Stat() with some extra sanity checks.
+func (v *UnixVolume) stat(path string) (os.FileInfo, error) {
+ stat, err := os.Stat(path)
+ if err == nil {
+ if stat.Size() < 0 {
+ err = os.ErrInvalid
+ } else if stat.Size() > BlockSize {
+ err = TooLongError
+ }
+ }
+ return stat, err
+}
+
+// Get retrieves a block, copies it to the given slice, and returns
+// the number of bytes copied.
+func (v *UnixVolume) Get(loc string, buf []byte) (int, error) {
+ path := v.blockPath(loc)
+ stat, err := v.stat(path)
+ if err != nil {
+ return 0, v.translateError(err)
+ }
+ if stat.Size() > int64(len(buf)) {
+ return 0, TooLongError
+ }
+ var read int
+ size := int(stat.Size())
+ err = v.getFunc(path, func(rdr io.Reader) error {
+ read, err = io.ReadFull(rdr, buf[:size])
+ return err
+ })
+ return read, err
+}
+
+// Compare returns nil if Get(loc) would return the same content as
+// expect. It is functionally equivalent to Get() followed by
+// bytes.Compare(), but uses less memory.
+func (v *UnixVolume) Compare(loc string, expect []byte) error {
+ path := v.blockPath(loc)
+ if _, err := v.stat(path); err != nil {
+ return v.translateError(err)
+ }
+ return v.getFunc(path, func(rdr io.Reader) error {
+ return compareReaderWithBuf(rdr, expect, loc[:32])
+ })
+}
+
+// Put stores a block of data identified by the locator string
// "loc". It returns nil on success. If the volume is full, it
// returns a FullError. If the write fails due to some other error,
// that error is returned.
-//
-func (v *UnixVolume) Write(loc string, block []byte) error {
+func (v *UnixVolume) Put(loc string, block []byte) error {
+ if v.readonly {
+ return MethodDisabledError
+ }
if v.IsFull() {
return FullError
}
}
bpath := v.blockPath(loc)
+ if v.locker != nil {
+ v.locker.Lock()
+ defer v.locker.Unlock()
+ }
if _, err := tmpfile.Write(block); err != nil {
log.Printf("%s: writing to %s: %s\n", v, bpath, err)
+ tmpfile.Close()
+ os.Remove(tmpfile.Name())
return err
}
if err := tmpfile.Close(); err != nil {
}
// Status returns a VolumeStatus struct describing the volume's
-// current state.
+// current state, or nil if an error occurs.
//
func (v *UnixVolume) Status() *VolumeStatus {
var fs syscall.Statfs_t
return &VolumeStatus{v.root, devnum, free, used}
}
+var blockDirRe = regexp.MustCompile(`^[0-9a-f]+$`)
+var blockFileRe = regexp.MustCompile(`^[0-9a-f]{32}$`)
+
// IndexTo writes (to the given Writer) a list of blocks found on this
// volume which begin with the specified prefix. If the prefix is an
-// empty string, Index writes a complete list of blocks.
+// empty string, IndexTo writes a complete list of blocks.
//
// Each block is given in the format
//
// e4de7a2810f5554cd39b36d8ddb132ff+67108864 1388701136
//
func (v *UnixVolume) IndexTo(prefix string, w io.Writer) error {
- return filepath.Walk(v.root,
- func(path string, info os.FileInfo, err error) error {
- if err != nil {
- log.Printf("%s: IndexTo Walk error at %s: %s",
- v, path, err)
- return nil
- }
- locator := filepath.Base(path)
- // Skip directories that do not match prefix.
- // We know there is nothing interesting inside.
- if info.IsDir() &&
- !strings.HasPrefix(locator, prefix) &&
- !strings.HasPrefix(prefix, locator) {
- return filepath.SkipDir
+ var lastErr error
+ rootdir, err := os.Open(v.root)
+ if err != nil {
+ return err
+ }
+ defer rootdir.Close()
+ for {
+ names, err := rootdir.Readdirnames(1)
+ if err == io.EOF {
+ return lastErr
+ } else if err != nil {
+ return err
+ }
+ if !strings.HasPrefix(names[0], prefix) && !strings.HasPrefix(prefix, names[0]) {
+ // prefix excludes all blocks stored in this dir
+ continue
+ }
+ if !blockDirRe.MatchString(names[0]) {
+ continue
+ }
+ blockdirpath := filepath.Join(v.root, names[0])
+ blockdir, err := os.Open(blockdirpath)
+ if err != nil {
+ log.Print("Error reading ", blockdirpath, ": ", err)
+ lastErr = err
+ continue
+ }
+ for {
+ fileInfo, err := blockdir.Readdir(1)
+ if err == io.EOF {
+ break
+ } else if err != nil {
+ log.Print("Error reading ", blockdirpath, ": ", err)
+ lastErr = err
+ break
}
- // Skip any file that is not apparently a locator, e.g. .meta files
- if info.IsDir() || !IsValidLocator(locator) {
- return nil
+ name := fileInfo[0].Name()
+ if !strings.HasPrefix(name, prefix) {
+ continue
}
- // Print filenames beginning with prefix
- if !strings.HasPrefix(locator, prefix) {
- return nil
+ if !blockFileRe.MatchString(name) {
+ continue
}
- _, err = fmt.Fprintf(w, "%s+%d %d\n",
- locator, info.Size(), info.ModTime().Unix())
- return err
- })
+ _, err = fmt.Fprint(w,
+ name,
+ "+", fileInfo[0].Size(),
+ " ", fileInfo[0].ModTime().UnixNano(),
+ "\n")
+ }
+ blockdir.Close()
+ }
}
-func (v *UnixVolume) Delete(loc string) error {
+// Trash trashes the block data from the unix storage
+// If trashLifetime == 0, the block is deleted
+// Else, the block is renamed as path/{loc}.trash.{deadline},
+// where deadline = now + trashLifetime
+func (v *UnixVolume) Trash(loc string) error {
// Touch() must be called before calling Write() on a block. Touch()
// also uses lockfile(). This avoids a race condition between Write()
- // and Delete() because either (a) the file will be deleted and Touch()
+ // and Trash() because either (a) the file will be trashed and Touch()
// will signal to the caller that the file is not present (and needs to
// be re-written), or (b) Touch() will update the file's timestamp and
- // Delete() will read the correct up-to-date timestamp and choose not to
- // delete the file.
+ // Trash() will read the correct up-to-date timestamp and choose not to
+ // trash the file.
if v.readonly {
return MethodDisabledError
}
+ if v.locker != nil {
+ v.locker.Lock()
+ defer v.locker.Unlock()
+ }
p := v.blockPath(loc)
f, err := os.OpenFile(p, os.O_RDWR|os.O_APPEND, 0644)
if err != nil {
}
defer unlockfile(f)
- // If the block has been PUT in the last blob_signature_ttl
+ // If the block has been PUT in the last blobSignatureTTL
// seconds, return success without removing the block. This
// protects data from garbage collection until it is no longer
// possible for clients to retrieve the unreferenced blocks
// anyway (because the permission signatures have expired).
if fi, err := os.Stat(p); err != nil {
return err
- } else {
- if time.Since(fi.ModTime()) < blob_signature_ttl {
- return nil
+ } else if time.Since(fi.ModTime()) < blobSignatureTTL {
+ return nil
+ }
+
+ if trashLifetime == 0 {
+ return os.Remove(p)
+ }
+ return os.Rename(p, fmt.Sprintf("%v.trash.%d", p, time.Now().Add(trashLifetime).Unix()))
+}
+
+// Untrash moves block from trash back into store
+// Look for path/{loc}.trash.{deadline} in storage,
+// and rename the first such file as path/{loc}
+func (v *UnixVolume) Untrash(loc string) (err error) {
+ if v.readonly {
+ return MethodDisabledError
+ }
+
+ files, err := ioutil.ReadDir(v.blockDir(loc))
+ if err != nil {
+ return err
+ }
+
+ if len(files) == 0 {
+ return os.ErrNotExist
+ }
+
+ foundTrash := false
+ prefix := fmt.Sprintf("%v.trash.", loc)
+ for _, f := range files {
+ if strings.HasPrefix(f.Name(), prefix) {
+ foundTrash = true
+ err = os.Rename(v.blockPath(f.Name()), v.blockPath(loc))
+ if err == nil {
+ break
+ }
}
}
- return os.Remove(p)
+
+ if foundTrash == false {
+ return os.ErrNotExist
+ }
+
+ return
}
// blockDir returns the fully qualified directory name for the directory
}
// IsFull returns true if the free space on the volume is less than
-// MIN_FREE_KILOBYTES.
+// MinFreeKilobytes.
//
func (v *UnixVolume) IsFull() (isFull bool) {
fullSymlink := v.root + "/full"
}
if avail, err := v.FreeDiskSpace(); err == nil {
- isFull = avail < MIN_FREE_KILOBYTES
+ isFull = avail < MinFreeKilobytes
} else {
log.Printf("%s: FreeDiskSpace: %s\n", v, err)
isFull = false
return fmt.Sprintf("[UnixVolume %s]", v.root)
}
+// Writable returns false if all future Put, Mtime, and Delete calls
+// are expected to fail.
func (v *UnixVolume) Writable() bool {
return !v.readonly
}
+// Replication returns the number of replicas promised by the
+// underlying device (currently assumed to be 1).
+func (v *UnixVolume) Replication() int {
+ return 1
+}
+
// lockfile and unlockfile use flock(2) to manage kernel file locks.
func lockfile(f *os.File) error {
return syscall.Flock(int(f.Fd()), syscall.LOCK_EX)
func unlockfile(f *os.File) error {
return syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
}
+
+// Where appropriate, translate a more specific filesystem error to an
+// error recognized by handlers, like os.ErrNotExist.
+func (v *UnixVolume) translateError(err error) error {
+ switch err.(type) {
+ case *os.PathError:
+ // stat() returns a PathError if the parent directory
+ // (not just the file itself) is missing
+ return os.ErrNotExist
+ default:
+ return err
+ }
+}
+
+var unixTrashLocRegexp = regexp.MustCompile(`/([0-9a-f]{32})\.trash\.(\d+)$`)
+
+// EmptyTrash walks hierarchy looking for {hash}.trash.*
+// and deletes those with deadline < now.
+func (v *UnixVolume) EmptyTrash() {
+ var bytesDeleted, bytesInTrash int64
+ var blocksDeleted, blocksInTrash int
+
+ err := filepath.Walk(v.root, func(path string, info os.FileInfo, err error) error {
+ if err != nil {
+ log.Printf("EmptyTrash: filepath.Walk: %v: %v", path, err)
+ return nil
+ }
+ if info.Mode().IsDir() {
+ return nil
+ }
+ matches := unixTrashLocRegexp.FindStringSubmatch(path)
+ if len(matches) != 3 {
+ return nil
+ }
+ deadline, err := strconv.ParseInt(matches[2], 10, 64)
+ if err != nil {
+ log.Printf("EmptyTrash: %v: ParseInt(%v): %v", path, matches[2], err)
+ return nil
+ }
+ bytesInTrash += info.Size()
+ blocksInTrash++
+ if deadline > time.Now().Unix() {
+ return nil
+ }
+ err = os.Remove(path)
+ if err != nil {
+ log.Printf("EmptyTrash: Remove %v: %v", path, err)
+ return nil
+ }
+ bytesDeleted += info.Size()
+ blocksDeleted++
+ return nil
+ })
+
+ if err != nil {
+ log.Printf("EmptyTrash error for %v: %v", v.String(), err)
+ }
+
+ log.Printf("EmptyTrash stats for %v: Deleted %v bytes in %v blocks. Remaining in trash: %v bytes in %v blocks.", v.String(), bytesDeleted, blocksDeleted, bytesInTrash-bytesDeleted, blocksInTrash-blocksDeleted)
+}
projects / arvados.git / blobdiff