13937: Adds counters to azure driver.
[arvados.git] / services / keepstore / unix_volume.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package main
6
7 import (
8         "bufio"
9         "context"
10         "flag"
11         "fmt"
12         "io"
13         "io/ioutil"
14         "os"
15         "os/exec"
16         "path/filepath"
17         "regexp"
18         "strconv"
19         "strings"
20         "sync"
21         "sync/atomic"
22         "syscall"
23         "time"
24
25         "github.com/prometheus/client_golang/prometheus"
26 )
27
28 type unixVolumeAdder struct {
29         *Config
30 }
31
32 // String implements flag.Value
33 func (vs *unixVolumeAdder) String() string {
34         return "-"
35 }
36
37 func (vs *unixVolumeAdder) Set(path string) error {
38         if dirs := strings.Split(path, ","); len(dirs) > 1 {
39                 log.Print("DEPRECATED: using comma-separated volume list.")
40                 for _, dir := range dirs {
41                         if err := vs.Set(dir); err != nil {
42                                 return err
43                         }
44                 }
45                 return nil
46         }
47         vs.Config.Volumes = append(vs.Config.Volumes, &UnixVolume{
48                 Root:      path,
49                 ReadOnly:  deprecated.flagReadonly,
50                 Serialize: deprecated.flagSerializeIO,
51         })
52         return nil
53 }
54
55 func init() {
56         VolumeTypes = append(VolumeTypes, func() VolumeWithExamples { return &UnixVolume{} })
57
58         flag.Var(&unixVolumeAdder{theConfig}, "volumes", "see Volumes configuration")
59         flag.Var(&unixVolumeAdder{theConfig}, "volume", "see Volumes configuration")
60 }
61
62 // Discover adds a UnixVolume for every directory named "keep" that is
63 // located at the top level of a device- or tmpfs-backed mount point
64 // other than "/". It returns the number of volumes added.
65 func (vs *unixVolumeAdder) Discover() int {
66         added := 0
67         f, err := os.Open(ProcMounts)
68         if err != nil {
69                 log.Fatalf("opening %s: %s", ProcMounts, err)
70         }
71         scanner := bufio.NewScanner(f)
72         for scanner.Scan() {
73                 args := strings.Fields(scanner.Text())
74                 if err := scanner.Err(); err != nil {
75                         log.Fatalf("reading %s: %s", ProcMounts, err)
76                 }
77                 dev, mount := args[0], args[1]
78                 if mount == "/" {
79                         continue
80                 }
81                 if dev != "tmpfs" && !strings.HasPrefix(dev, "/dev/") {
82                         continue
83                 }
84                 keepdir := mount + "/keep"
85                 if st, err := os.Stat(keepdir); err != nil || !st.IsDir() {
86                         continue
87                 }
88                 // Set the -readonly flag (but only for this volume)
89                 // if the filesystem is mounted readonly.
90                 flagReadonlyWas := deprecated.flagReadonly
91                 for _, fsopt := range strings.Split(args[3], ",") {
92                         if fsopt == "ro" {
93                                 deprecated.flagReadonly = true
94                                 break
95                         }
96                         if fsopt == "rw" {
97                                 break
98                         }
99                 }
100                 if err := vs.Set(keepdir); err != nil {
101                         log.Printf("adding %q: %s", keepdir, err)
102                 } else {
103                         added++
104                 }
105                 deprecated.flagReadonly = flagReadonlyWas
106         }
107         return added
108 }
109
110 // A UnixVolume stores and retrieves blocks in a local directory.
111 type UnixVolume struct {
112         Root                 string // path to the volume's root directory
113         ReadOnly             bool
114         Serialize            bool
115         DirectoryReplication int
116         StorageClasses       []string
117
118         // something to lock during IO, typically a sync.Mutex (or nil
119         // to skip locking)
120         locker sync.Locker
121
122         os osWithStats
123 }
124
125 // DeviceID returns a globally unique ID for the volume's root
126 // directory, consisting of the filesystem's UUID and the path from
127 // filesystem root to storage directory, joined by "/". For example,
128 // the DeviceID for a local directory "/mnt/xvda1/keep" might be
129 // "fa0b6166-3b55-4994-bd3f-92f4e00a1bb0/keep".
130 func (v *UnixVolume) DeviceID() string {
131         giveup := func(f string, args ...interface{}) string {
132                 log.Printf(f+"; using blank DeviceID for volume %s", append(args, v)...)
133                 return ""
134         }
135         buf, err := exec.Command("findmnt", "--noheadings", "--target", v.Root).CombinedOutput()
136         if err != nil {
137                 return giveup("findmnt: %s (%q)", err, buf)
138         }
139         findmnt := strings.Fields(string(buf))
140         if len(findmnt) < 2 {
141                 return giveup("could not parse findmnt output: %q", buf)
142         }
143         fsRoot, dev := findmnt[0], findmnt[1]
144
145         absRoot, err := filepath.Abs(v.Root)
146         if err != nil {
147                 return giveup("resolving relative path %q: %s", v.Root, err)
148         }
149         realRoot, err := filepath.EvalSymlinks(absRoot)
150         if err != nil {
151                 return giveup("resolving symlinks in %q: %s", absRoot, err)
152         }
153
154         // Find path from filesystem root to realRoot
155         var fsPath string
156         if strings.HasPrefix(realRoot, fsRoot+"/") {
157                 fsPath = realRoot[len(fsRoot):]
158         } else if fsRoot == "/" {
159                 fsPath = realRoot
160         } else if fsRoot == realRoot {
161                 fsPath = ""
162         } else {
163                 return giveup("findmnt reports mount point %q which is not a prefix of volume root %q", fsRoot, realRoot)
164         }
165
166         if !strings.HasPrefix(dev, "/") {
167                 return giveup("mount %q device %q is not a path", fsRoot, dev)
168         }
169
170         fi, err := os.Stat(dev)
171         if err != nil {
172                 return giveup("stat %q: %s\n", dev, err)
173         }
174         ino := fi.Sys().(*syscall.Stat_t).Ino
175
176         // Find a symlink in /dev/disk/by-uuid/ whose target is (i.e.,
177         // has the same inode as) the mounted device
178         udir := "/dev/disk/by-uuid"
179         d, err := os.Open(udir)
180         if err != nil {
181                 return giveup("opening %q: %s", udir, err)
182         }
183         uuids, err := d.Readdirnames(0)
184         if err != nil {
185                 return giveup("reading %q: %s", udir, err)
186         }
187         for _, uuid := range uuids {
188                 link := filepath.Join(udir, uuid)
189                 fi, err = os.Stat(link)
190                 if err != nil {
191                         log.Printf("error: stat %q: %s", link, err)
192                         continue
193                 }
194                 if fi.Sys().(*syscall.Stat_t).Ino == ino {
195                         return uuid + fsPath
196                 }
197         }
198         return giveup("could not find entry in %q matching %q", udir, dev)
199 }
200
201 // Examples implements VolumeWithExamples.
202 func (*UnixVolume) Examples() []Volume {
203         return []Volume{
204                 &UnixVolume{
205                         Root:                 "/mnt/local-disk",
206                         Serialize:            true,
207                         DirectoryReplication: 1,
208                 },
209                 &UnixVolume{
210                         Root:                 "/mnt/network-disk",
211                         Serialize:            false,
212                         DirectoryReplication: 2,
213                 },
214         }
215 }
216
217 // Type implements Volume
218 func (v *UnixVolume) Type() string {
219         return "Directory"
220 }
221
222 // Start implements Volume
223 func (v *UnixVolume) Start(opsCounters, errCounters, ioBytes *prometheus.CounterVec) error {
224         if v.Serialize {
225                 v.locker = &sync.Mutex{}
226         }
227         if !strings.HasPrefix(v.Root, "/") {
228                 return fmt.Errorf("volume root does not start with '/': %q", v.Root)
229         }
230         if v.DirectoryReplication == 0 {
231                 v.DirectoryReplication = 1
232         }
233         // Set up prometheus metrics
234         lbls := prometheus.Labels{"device_id": v.DeviceID()}
235         v.os.stats.opsCounters = opsCounters.MustCurryWith(lbls)
236         v.os.stats.errCounters = errCounters.MustCurryWith(lbls)
237         v.os.stats.ioBytes = ioBytes.MustCurryWith(lbls)
238
239         _, err := v.os.Stat(v.Root)
240
241         return err
242 }
243
244 // Touch sets the timestamp for the given locator to the current time
245 func (v *UnixVolume) Touch(loc string) error {
246         if v.ReadOnly {
247                 return MethodDisabledError
248         }
249         p := v.blockPath(loc)
250         f, err := v.os.OpenFile(p, os.O_RDWR|os.O_APPEND, 0644)
251         if err != nil {
252                 return err
253         }
254         defer f.Close()
255         if err := v.lock(context.TODO()); err != nil {
256                 return err
257         }
258         defer v.unlock()
259         if e := v.lockfile(f); e != nil {
260                 return e
261         }
262         defer v.unlockfile(f)
263         ts := syscall.NsecToTimespec(time.Now().UnixNano())
264         v.os.stats.TickOps("utimes")
265         v.os.stats.Tick(&v.os.stats.UtimesOps)
266         err = syscall.UtimesNano(p, []syscall.Timespec{ts, ts})
267         v.os.stats.TickErr(err)
268         return err
269 }
270
271 // Mtime returns the stored timestamp for the given locator.
272 func (v *UnixVolume) Mtime(loc string) (time.Time, error) {
273         p := v.blockPath(loc)
274         fi, err := v.os.Stat(p)
275         if err != nil {
276                 return time.Time{}, err
277         }
278         return fi.ModTime(), nil
279 }
280
281 // Lock the locker (if one is in use), open the file for reading, and
282 // call the given function if and when the file is ready to read.
283 func (v *UnixVolume) getFunc(ctx context.Context, path string, fn func(io.Reader) error) error {
284         if err := v.lock(ctx); err != nil {
285                 return err
286         }
287         defer v.unlock()
288         f, err := v.os.Open(path)
289         if err != nil {
290                 return err
291         }
292         defer f.Close()
293         return fn(NewCountingReader(ioutil.NopCloser(f), v.os.stats.TickInBytes))
294 }
295
296 // stat is os.Stat() with some extra sanity checks.
297 func (v *UnixVolume) stat(path string) (os.FileInfo, error) {
298         stat, err := v.os.Stat(path)
299         if err == nil {
300                 if stat.Size() < 0 {
301                         err = os.ErrInvalid
302                 } else if stat.Size() > BlockSize {
303                         err = TooLongError
304                 }
305         }
306         return stat, err
307 }
308
309 // Get retrieves a block, copies it to the given slice, and returns
310 // the number of bytes copied.
311 func (v *UnixVolume) Get(ctx context.Context, loc string, buf []byte) (int, error) {
312         return getWithPipe(ctx, loc, buf, v)
313 }
314
315 // ReadBlock implements BlockReader.
316 func (v *UnixVolume) ReadBlock(ctx context.Context, loc string, w io.Writer) error {
317         path := v.blockPath(loc)
318         stat, err := v.stat(path)
319         if err != nil {
320                 return v.translateError(err)
321         }
322         return v.getFunc(ctx, path, func(rdr io.Reader) error {
323                 n, err := io.Copy(w, rdr)
324                 if err == nil && n != stat.Size() {
325                         err = io.ErrUnexpectedEOF
326                 }
327                 return err
328         })
329 }
330
331 // Compare returns nil if Get(loc) would return the same content as
332 // expect. It is functionally equivalent to Get() followed by
333 // bytes.Compare(), but uses less memory.
334 func (v *UnixVolume) Compare(ctx context.Context, loc string, expect []byte) error {
335         path := v.blockPath(loc)
336         if _, err := v.stat(path); err != nil {
337                 return v.translateError(err)
338         }
339         return v.getFunc(ctx, path, func(rdr io.Reader) error {
340                 return compareReaderWithBuf(ctx, rdr, expect, loc[:32])
341         })
342 }
343
344 // Put stores a block of data identified by the locator string
345 // "loc".  It returns nil on success.  If the volume is full, it
346 // returns a FullError.  If the write fails due to some other error,
347 // that error is returned.
348 func (v *UnixVolume) Put(ctx context.Context, loc string, block []byte) error {
349         return putWithPipe(ctx, loc, block, v)
350 }
351
352 // WriteBlock implements BlockWriter.
353 func (v *UnixVolume) WriteBlock(ctx context.Context, loc string, rdr io.Reader) error {
354         if v.ReadOnly {
355                 return MethodDisabledError
356         }
357         if v.IsFull() {
358                 return FullError
359         }
360         bdir := v.blockDir(loc)
361         if err := os.MkdirAll(bdir, 0755); err != nil {
362                 log.Printf("%s: could not create directory %s: %s",
363                         loc, bdir, err)
364                 return err
365         }
366
367         tmpfile, tmperr := v.os.TempFile(bdir, "tmp"+loc)
368         if tmperr != nil {
369                 log.Printf("ioutil.TempFile(%s, tmp%s): %s", bdir, loc, tmperr)
370                 return tmperr
371         }
372
373         bpath := v.blockPath(loc)
374
375         if err := v.lock(ctx); err != nil {
376                 return err
377         }
378         defer v.unlock()
379         n, err := io.Copy(tmpfile, rdr)
380         v.os.stats.TickOutBytes(uint64(n))
381         if err != nil {
382                 log.Printf("%s: writing to %s: %s\n", v, bpath, err)
383                 tmpfile.Close()
384                 v.os.Remove(tmpfile.Name())
385                 return err
386         }
387         if err := tmpfile.Close(); err != nil {
388                 log.Printf("closing %s: %s\n", tmpfile.Name(), err)
389                 v.os.Remove(tmpfile.Name())
390                 return err
391         }
392         if err := v.os.Rename(tmpfile.Name(), bpath); err != nil {
393                 log.Printf("rename %s %s: %s\n", tmpfile.Name(), bpath, err)
394                 return v.os.Remove(tmpfile.Name())
395         }
396         return nil
397 }
398
399 // Status returns a VolumeStatus struct describing the volume's
400 // current state, or nil if an error occurs.
401 //
402 func (v *UnixVolume) Status() *VolumeStatus {
403         fi, err := v.os.Stat(v.Root)
404         if err != nil {
405                 log.Printf("%s: os.Stat: %s\n", v, err)
406                 return nil
407         }
408         devnum := fi.Sys().(*syscall.Stat_t).Dev
409
410         var fs syscall.Statfs_t
411         if err := syscall.Statfs(v.Root, &fs); err != nil {
412                 log.Printf("%s: statfs: %s\n", v, err)
413                 return nil
414         }
415         // These calculations match the way df calculates disk usage:
416         // "free" space is measured by fs.Bavail, but "used" space
417         // uses fs.Blocks - fs.Bfree.
418         free := fs.Bavail * uint64(fs.Bsize)
419         used := (fs.Blocks - fs.Bfree) * uint64(fs.Bsize)
420         return &VolumeStatus{
421                 MountPoint: v.Root,
422                 DeviceNum:  devnum,
423                 BytesFree:  free,
424                 BytesUsed:  used,
425         }
426 }
427
428 var blockDirRe = regexp.MustCompile(`^[0-9a-f]+$`)
429 var blockFileRe = regexp.MustCompile(`^[0-9a-f]{32}$`)
430
431 // IndexTo writes (to the given Writer) a list of blocks found on this
432 // volume which begin with the specified prefix. If the prefix is an
433 // empty string, IndexTo writes a complete list of blocks.
434 //
435 // Each block is given in the format
436 //
437 //     locator+size modification-time {newline}
438 //
439 // e.g.:
440 //
441 //     e4df392f86be161ca6ed3773a962b8f3+67108864 1388894303
442 //     e4d41e6fd68460e0e3fc18cc746959d2+67108864 1377796043
443 //     e4de7a2810f5554cd39b36d8ddb132ff+67108864 1388701136
444 //
445 func (v *UnixVolume) IndexTo(prefix string, w io.Writer) error {
446         var lastErr error
447         rootdir, err := v.os.Open(v.Root)
448         if err != nil {
449                 return err
450         }
451         defer rootdir.Close()
452         v.os.stats.TickOps("readdir")
453         v.os.stats.Tick(&v.os.stats.ReaddirOps)
454         for {
455                 names, err := rootdir.Readdirnames(1)
456                 if err == io.EOF {
457                         return lastErr
458                 } else if err != nil {
459                         return err
460                 }
461                 if !strings.HasPrefix(names[0], prefix) && !strings.HasPrefix(prefix, names[0]) {
462                         // prefix excludes all blocks stored in this dir
463                         continue
464                 }
465                 if !blockDirRe.MatchString(names[0]) {
466                         continue
467                 }
468                 blockdirpath := filepath.Join(v.Root, names[0])
469                 blockdir, err := v.os.Open(blockdirpath)
470                 if err != nil {
471                         log.Print("Error reading ", blockdirpath, ": ", err)
472                         lastErr = err
473                         continue
474                 }
475                 v.os.stats.TickOps("readdir")
476                 v.os.stats.Tick(&v.os.stats.ReaddirOps)
477                 for {
478                         fileInfo, err := blockdir.Readdir(1)
479                         if err == io.EOF {
480                                 break
481                         } else if err != nil {
482                                 log.Print("Error reading ", blockdirpath, ": ", err)
483                                 lastErr = err
484                                 break
485                         }
486                         name := fileInfo[0].Name()
487                         if !strings.HasPrefix(name, prefix) {
488                                 continue
489                         }
490                         if !blockFileRe.MatchString(name) {
491                                 continue
492                         }
493                         _, err = fmt.Fprint(w,
494                                 name,
495                                 "+", fileInfo[0].Size(),
496                                 " ", fileInfo[0].ModTime().UnixNano(),
497                                 "\n")
498                         if err != nil {
499                                 log.Print("Error writing : ", err)
500                                 lastErr = err
501                                 break
502                         }
503                 }
504                 blockdir.Close()
505         }
506 }
507
508 // Trash trashes the block data from the unix storage
509 // If TrashLifetime == 0, the block is deleted
510 // Else, the block is renamed as path/{loc}.trash.{deadline},
511 // where deadline = now + TrashLifetime
512 func (v *UnixVolume) Trash(loc string) error {
513         // Touch() must be called before calling Write() on a block.  Touch()
514         // also uses lockfile().  This avoids a race condition between Write()
515         // and Trash() because either (a) the file will be trashed and Touch()
516         // will signal to the caller that the file is not present (and needs to
517         // be re-written), or (b) Touch() will update the file's timestamp and
518         // Trash() will read the correct up-to-date timestamp and choose not to
519         // trash the file.
520
521         if v.ReadOnly {
522                 return MethodDisabledError
523         }
524         if err := v.lock(context.TODO()); err != nil {
525                 return err
526         }
527         defer v.unlock()
528         p := v.blockPath(loc)
529         f, err := v.os.OpenFile(p, os.O_RDWR|os.O_APPEND, 0644)
530         if err != nil {
531                 return err
532         }
533         defer f.Close()
534         if e := v.lockfile(f); e != nil {
535                 return e
536         }
537         defer v.unlockfile(f)
538
539         // If the block has been PUT in the last blobSignatureTTL
540         // seconds, return success without removing the block. This
541         // protects data from garbage collection until it is no longer
542         // possible for clients to retrieve the unreferenced blocks
543         // anyway (because the permission signatures have expired).
544         if fi, err := v.os.Stat(p); err != nil {
545                 return err
546         } else if time.Since(fi.ModTime()) < time.Duration(theConfig.BlobSignatureTTL) {
547                 return nil
548         }
549
550         if theConfig.TrashLifetime == 0 {
551                 return v.os.Remove(p)
552         }
553         return v.os.Rename(p, fmt.Sprintf("%v.trash.%d", p, time.Now().Add(theConfig.TrashLifetime.Duration()).Unix()))
554 }
555
556 // Untrash moves block from trash back into store
557 // Look for path/{loc}.trash.{deadline} in storage,
558 // and rename the first such file as path/{loc}
559 func (v *UnixVolume) Untrash(loc string) (err error) {
560         if v.ReadOnly {
561                 return MethodDisabledError
562         }
563
564         v.os.stats.TickOps("readdir")
565         v.os.stats.Tick(&v.os.stats.ReaddirOps)
566         files, err := ioutil.ReadDir(v.blockDir(loc))
567         if err != nil {
568                 return err
569         }
570
571         if len(files) == 0 {
572                 return os.ErrNotExist
573         }
574
575         foundTrash := false
576         prefix := fmt.Sprintf("%v.trash.", loc)
577         for _, f := range files {
578                 if strings.HasPrefix(f.Name(), prefix) {
579                         foundTrash = true
580                         err = v.os.Rename(v.blockPath(f.Name()), v.blockPath(loc))
581                         if err == nil {
582                                 break
583                         }
584                 }
585         }
586
587         if foundTrash == false {
588                 return os.ErrNotExist
589         }
590
591         return
592 }
593
594 // blockDir returns the fully qualified directory name for the directory
595 // where loc is (or would be) stored on this volume.
596 func (v *UnixVolume) blockDir(loc string) string {
597         return filepath.Join(v.Root, loc[0:3])
598 }
599
600 // blockPath returns the fully qualified pathname for the path to loc
601 // on this volume.
602 func (v *UnixVolume) blockPath(loc string) string {
603         return filepath.Join(v.blockDir(loc), loc)
604 }
605
606 // IsFull returns true if the free space on the volume is less than
607 // MinFreeKilobytes.
608 //
609 func (v *UnixVolume) IsFull() (isFull bool) {
610         fullSymlink := v.Root + "/full"
611
612         // Check if the volume has been marked as full in the last hour.
613         if link, err := os.Readlink(fullSymlink); err == nil {
614                 if ts, err := strconv.Atoi(link); err == nil {
615                         fulltime := time.Unix(int64(ts), 0)
616                         if time.Since(fulltime).Hours() < 1.0 {
617                                 return true
618                         }
619                 }
620         }
621
622         if avail, err := v.FreeDiskSpace(); err == nil {
623                 isFull = avail < MinFreeKilobytes
624         } else {
625                 log.Printf("%s: FreeDiskSpace: %s\n", v, err)
626                 isFull = false
627         }
628
629         // If the volume is full, timestamp it.
630         if isFull {
631                 now := fmt.Sprintf("%d", time.Now().Unix())
632                 os.Symlink(now, fullSymlink)
633         }
634         return
635 }
636
637 // FreeDiskSpace returns the number of unused 1k blocks available on
638 // the volume.
639 //
640 func (v *UnixVolume) FreeDiskSpace() (free uint64, err error) {
641         var fs syscall.Statfs_t
642         err = syscall.Statfs(v.Root, &fs)
643         if err == nil {
644                 // Statfs output is not guaranteed to measure free
645                 // space in terms of 1K blocks.
646                 free = fs.Bavail * uint64(fs.Bsize) / 1024
647         }
648         return
649 }
650
651 func (v *UnixVolume) String() string {
652         return fmt.Sprintf("[UnixVolume %s]", v.Root)
653 }
654
655 // Writable returns false if all future Put, Mtime, and Delete calls
656 // are expected to fail.
657 func (v *UnixVolume) Writable() bool {
658         return !v.ReadOnly
659 }
660
661 // Replication returns the number of replicas promised by the
662 // underlying device (as specified in configuration).
663 func (v *UnixVolume) Replication() int {
664         return v.DirectoryReplication
665 }
666
667 // GetStorageClasses implements Volume
668 func (v *UnixVolume) GetStorageClasses() []string {
669         return v.StorageClasses
670 }
671
672 // InternalStats returns I/O and filesystem ops counters.
673 func (v *UnixVolume) InternalStats() interface{} {
674         return &v.os.stats
675 }
676
677 // lock acquires the serialize lock, if one is in use. If ctx is done
678 // before the lock is acquired, lock returns ctx.Err() instead of
679 // acquiring the lock.
680 func (v *UnixVolume) lock(ctx context.Context) error {
681         if v.locker == nil {
682                 return nil
683         }
684         locked := make(chan struct{})
685         go func() {
686                 v.locker.Lock()
687                 close(locked)
688         }()
689         select {
690         case <-ctx.Done():
691                 go func() {
692                         <-locked
693                         v.locker.Unlock()
694                 }()
695                 return ctx.Err()
696         case <-locked:
697                 return nil
698         }
699 }
700
701 // unlock releases the serialize lock, if one is in use.
702 func (v *UnixVolume) unlock() {
703         if v.locker == nil {
704                 return
705         }
706         v.locker.Unlock()
707 }
708
709 // lockfile and unlockfile use flock(2) to manage kernel file locks.
710 func (v *UnixVolume) lockfile(f *os.File) error {
711         v.os.stats.TickOps("flock")
712         v.os.stats.Tick(&v.os.stats.FlockOps)
713         err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX)
714         v.os.stats.TickErr(err)
715         return err
716 }
717
718 func (v *UnixVolume) unlockfile(f *os.File) error {
719         err := syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
720         v.os.stats.TickErr(err)
721         return err
722 }
723
724 // Where appropriate, translate a more specific filesystem error to an
725 // error recognized by handlers, like os.ErrNotExist.
726 func (v *UnixVolume) translateError(err error) error {
727         switch err.(type) {
728         case *os.PathError:
729                 // stat() returns a PathError if the parent directory
730                 // (not just the file itself) is missing
731                 return os.ErrNotExist
732         default:
733                 return err
734         }
735 }
736
737 var unixTrashLocRegexp = regexp.MustCompile(`/([0-9a-f]{32})\.trash\.(\d+)$`)
738
739 // EmptyTrash walks hierarchy looking for {hash}.trash.*
740 // and deletes those with deadline < now.
741 func (v *UnixVolume) EmptyTrash() {
742         var bytesDeleted, bytesInTrash int64
743         var blocksDeleted, blocksInTrash int64
744
745         doFile := func(path string, info os.FileInfo) {
746                 if info.Mode().IsDir() {
747                         return
748                 }
749                 matches := unixTrashLocRegexp.FindStringSubmatch(path)
750                 if len(matches) != 3 {
751                         return
752                 }
753                 deadline, err := strconv.ParseInt(matches[2], 10, 64)
754                 if err != nil {
755                         log.Printf("EmptyTrash: %v: ParseInt(%v): %v", path, matches[2], err)
756                         return
757                 }
758                 atomic.AddInt64(&bytesInTrash, info.Size())
759                 atomic.AddInt64(&blocksInTrash, 1)
760                 if deadline > time.Now().Unix() {
761                         return
762                 }
763                 err = v.os.Remove(path)
764                 if err != nil {
765                         log.Printf("EmptyTrash: Remove %v: %v", path, err)
766                         return
767                 }
768                 atomic.AddInt64(&bytesDeleted, info.Size())
769                 atomic.AddInt64(&blocksDeleted, 1)
770         }
771
772         type dirent struct {
773                 path string
774                 info os.FileInfo
775         }
776         var wg sync.WaitGroup
777         todo := make(chan dirent, theConfig.EmptyTrashWorkers)
778         for i := 0; i < 1 || i < theConfig.EmptyTrashWorkers; i++ {
779                 wg.Add(1)
780                 go func() {
781                         defer wg.Done()
782                         for e := range todo {
783                                 doFile(e.path, e.info)
784                         }
785                 }()
786         }
787
788         err := filepath.Walk(v.Root, func(path string, info os.FileInfo, err error) error {
789                 if err != nil {
790                         log.Printf("EmptyTrash: filepath.Walk: %v: %v", path, err)
791                         return nil
792                 }
793                 todo <- dirent{path, info}
794                 return nil
795         })
796         close(todo)
797         wg.Wait()
798
799         if err != nil {
800                 log.Printf("EmptyTrash error for %v: %v", v.String(), err)
801         }
802
803         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)
804 }
805
806 type unixStats struct {
807         statsTicker
808         OpenOps    uint64
809         StatOps    uint64
810         FlockOps   uint64
811         UtimesOps  uint64
812         CreateOps  uint64
813         RenameOps  uint64
814         UnlinkOps  uint64
815         ReaddirOps uint64
816 }
817
818 func (s *unixStats) TickErr(err error) {
819         if err == nil {
820                 return
821         }
822         s.statsTicker.TickErr(err, fmt.Sprintf("%T", err))
823 }
824
825 type osWithStats struct {
826         stats unixStats
827 }
828
829 func (o *osWithStats) Open(name string) (*os.File, error) {
830         o.stats.TickOps("open")
831         o.stats.Tick(&o.stats.OpenOps)
832         f, err := os.Open(name)
833         o.stats.TickErr(err)
834         return f, err
835 }
836
837 func (o *osWithStats) OpenFile(name string, flag int, perm os.FileMode) (*os.File, error) {
838         o.stats.TickOps("open")
839         o.stats.Tick(&o.stats.OpenOps)
840         f, err := os.OpenFile(name, flag, perm)
841         o.stats.TickErr(err)
842         return f, err
843 }
844
845 func (o *osWithStats) Remove(path string) error {
846         o.stats.TickOps("unlink")
847         o.stats.Tick(&o.stats.UnlinkOps)
848         err := os.Remove(path)
849         o.stats.TickErr(err)
850         return err
851 }
852
853 func (o *osWithStats) Rename(a, b string) error {
854         o.stats.TickOps("rename")
855         o.stats.Tick(&o.stats.RenameOps)
856         err := os.Rename(a, b)
857         o.stats.TickErr(err)
858         return err
859 }
860
861 func (o *osWithStats) Stat(path string) (os.FileInfo, error) {
862         o.stats.TickOps("stat")
863         o.stats.Tick(&o.stats.StatOps)
864         fi, err := os.Stat(path)
865         o.stats.TickErr(err)
866         return fi, err
867 }
868
869 func (o *osWithStats) TempFile(dir, base string) (*os.File, error) {
870         o.stats.TickOps("create")
871         o.stats.Tick(&o.stats.CreateOps)
872         f, err := ioutil.TempFile(dir, base)
873         o.stats.TickErr(err)
874         return f, err
875 }