X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/200f7004f921a68ec40b407dfe31f1db95e98fb9..143a5f355100b837daa428231df0370b525a1f9f:/services/keepstore/volume_unix.go diff --git a/services/keepstore/volume_unix.go b/services/keepstore/volume_unix.go index 8d23d11618..74bee52387 100644 --- a/services/keepstore/volume_unix.go +++ b/services/keepstore/volume_unix.go @@ -1,110 +1,31 @@ -// A UnixVolume is a Volume backed by a locally mounted disk. -// package main import ( + "bytes" "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 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. -// +// A UnixVolume stores and retrieves blocks in a local directory. type UnixVolume struct { - root string // path to this volume - queue chan *IORequest + // 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 } -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) - } - req.reply <- &result - } -} - -func MakeUnixVolume(root string, serialize bool, readonly bool) *UnixVolume { - v := &UnixVolume{ - root: root, - queue: nil, - readonly: readonly, - } - if serialize { - v.queue = make(chan *IORequest) - go v.IOHandler() - } - return v -} - -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 (v *UnixVolume) Put(loc string, block []byte) error { - if v.readonly { - return MethodDisabledError - } - if v.queue == nil { - return v.Write(loc, block) - } - reply := make(chan *IOResponse) - v.queue <- &IORequest{KeepPut, loc, block, reply} - response := <-reply - return response.err -} - func (v *UnixVolume) Touch(loc string) error { if v.readonly { return MethodDisabledError @@ -115,6 +36,10 @@ func (v *UnixVolume) Touch(loc string) error { return err } defer f.Close() + if v.locker != nil { + v.locker.Lock() + defer v.locker.Unlock() + } if e := lockfile(f); e != nil { return e } @@ -133,28 +58,102 @@ func (v *UnixVolume) Mtime(loc string) (time.Time, error) { } } -// Read retrieves a block identified by the locator string "loc", and +// 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) +} + +// 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 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 +// Get returns a nil buffer IFF it returns a non-nil error. +func (v *UnixVolume) Get(loc string) ([]byte, error) { + path := v.blockPath(loc) + stat, err := v.stat(path) + if err != nil { + return nil, err + } + buf := bufs.Get(int(stat.Size())) + err = v.getFunc(path, func(rdr io.Reader) error { + _, err = io.ReadFull(rdr, buf) + return err + }) + if err != nil { + bufs.Put(buf) + return nil, err + } + return buf, nil } -// Write stores a block of data identified by the locator string +// 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) + stat, err := v.stat(path) + if err != nil { + return err + } + bufLen := 1 << 20 + if int64(bufLen) > stat.Size() { + bufLen = int(stat.Size()) + } + cmp := expect + buf := make([]byte, bufLen) + return v.getFunc(path, func(rdr io.Reader) error { + // Loop invariants: all data read so far matched what + // we expected, and the first N bytes of cmp are + // expected to equal the next N bytes read from + // reader. + for { + n, err := rdr.Read(buf) + if n > len(cmp) || bytes.Compare(cmp[:n], buf[:n]) != 0 { + return collisionOrCorrupt(loc[:32], expect[:len(expect)-len(cmp)], buf[:n], rdr) + } + cmp = cmp[n:] + if err == io.EOF { + if len(cmp) != 0 { + return collisionOrCorrupt(loc[:32], expect[:len(expect)-len(cmp)], nil, nil) + } + return nil + } else if err != nil { + return err + } + } + }) +} + +// 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 } @@ -172,8 +171,14 @@ func (v *UnixVolume) Write(loc string, block []byte) error { } 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 { @@ -190,7 +195,7 @@ func (v *UnixVolume) Write(loc string, block []byte) error { } // 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 @@ -216,6 +221,8 @@ func (v *UnixVolume) Status() *VolumeStatus { return &VolumeStatus{v.root, devnum, free, used} } +var blockDirRe = regexp.MustCompile(`^[0-9a-f]+$`) + // 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, IndexTo writes a complete list of blocks. @@ -231,31 +238,54 @@ func (v *UnixVolume) Status() *VolumeStatus { // 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 - } - basename := filepath.Base(path) - if info.IsDir() && - !strings.HasPrefix(basename, prefix) && - !strings.HasPrefix(prefix, basename) { - // Skip directories that do not match - // prefix. We know there is nothing - // interesting inside. - return filepath.SkipDir + var lastErr error = nil + 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 } - if info.IsDir() || - !IsValidLocator(basename) || - !strings.HasPrefix(basename, prefix) { - return nil + name := fileInfo[0].Name() + if !strings.HasPrefix(name, prefix) { + continue } - _, err = fmt.Fprintf(w, "%s+%d %d\n", - basename, info.Size(), info.ModTime().Unix()) - return err - }) + _, err = fmt.Fprint(w, + name, + "+", fileInfo[0].Size(), + " ", fileInfo[0].ModTime().Unix(), + "\n") + } + blockdir.Close() + } } func (v *UnixVolume) Delete(loc string) error { @@ -270,6 +300,10 @@ func (v *UnixVolume) Delete(loc string) error { 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 {