Merge branch 'thehyve/fix-crunch-documentation' Fix a typo in Crunch Dispatch install...

[arvados.git] / services / keepstore / s3_volume.go

diff --git a/services/keepstore/s3_volume.go b/services/keepstore/s3_volume.go
index 90e8a1b4f6913da10b6789a4111a7ac8aa479721..9d4d8019282ebf01160544d940345b36fe892076 100644 (file)
--- a/services/keepstore/s3_volume.go
+++ b/services/keepstore/s3_volume.go
@@ -18,6 +18,7 @@ import (
        "regexp"
        "strings"
        "sync"
+       "sync/atomic"
        "time"
 
        "git.curoverse.com/arvados.git/sdk/go/arvados"
@@ -152,6 +153,7 @@ type S3Volume struct {
        RaceWindow         arvados.Duration
        ReadOnly           bool
        UnsafeDelete       bool
+       StorageClasses     []string
 
        bucket *s3bucket
 
@@ -686,6 +688,11 @@ func (v *S3Volume) Replication() int {
        return v.S3Replication
 }
 
+// GetStorageClasses implements Volume
+func (v *S3Volume) GetStorageClasses() []string {
+       return v.StorageClasses
+}
+
 var s3KeepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
 
 func (v *S3Volume) isKeepBlock(s string) bool {
@@ -758,26 +765,21 @@ func (v *S3Volume) translateError(err error) error {
 func (v *S3Volume) EmptyTrash() {
        var bytesInTrash, blocksInTrash, bytesDeleted, blocksDeleted int64
 
-       // Use a merge sort to find matching sets of trash/X and recent/X.
-       trashL := s3Lister{
-               Bucket:   v.bucket.Bucket,
-               Prefix:   "trash/",
-               PageSize: v.IndexPageSize,
-       }
        // Define "ready to delete" as "...when EmptyTrash started".
        startT := time.Now()
-       for trash := trashL.First(); trash != nil; trash = trashL.Next() {
+
+       emptyOneKey := func(trash *s3.Key) {
                loc := trash.Key[6:]
                if !v.isKeepBlock(loc) {
-                       continue
+                       return
                }
-               bytesInTrash += trash.Size
-               blocksInTrash++
+               atomic.AddInt64(&bytesInTrash, trash.Size)
+               atomic.AddInt64(&blocksInTrash, 1)
 
                trashT, err := time.Parse(time.RFC3339, trash.LastModified)
                if err != nil {
                        log.Printf("warning: %s: EmptyTrash: %q: parse %q: %s", v, trash.Key, trash.LastModified, err)
-                       continue
+                       return
                }
                recent, err := v.bucket.Head("recent/"+loc, nil)
                if err != nil && os.IsNotExist(v.translateError(err)) {
@@ -786,15 +788,15 @@ func (v *S3Volume) EmptyTrash() {
                        if err != nil {
                                log.Printf("error: %s: EmptyTrash: Untrash(%q): %s", v, loc, err)
                        }
-                       continue
+                       return
                } else if err != nil {
                        log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, "recent/"+loc, err)
-                       continue
+                       return
                }
                recentT, err := v.lastModified(recent)
                if err != nil {
                        log.Printf("warning: %s: EmptyTrash: %q: parse %q: %s", v, "recent/"+loc, recent.Header.Get("Last-Modified"), err)
-                       continue
+                       return
                }
                if trashT.Sub(recentT) < theConfig.BlobSignatureTTL.Duration() {
                        if age := startT.Sub(recentT); age >= theConfig.BlobSignatureTTL.Duration()-time.Duration(v.RaceWindow) {
@@ -809,39 +811,67 @@ func (v *S3Volume) EmptyTrash() {
                                log.Printf("notice: %s: EmptyTrash: detected old race for %q, calling fixRace + Touch", v, loc)
                                v.fixRace(loc)
                                v.Touch(loc)
-                               continue
+                               return
                        }
                        _, err := v.bucket.Head(loc, nil)
                        if os.IsNotExist(err) {
                                log.Printf("notice: %s: EmptyTrash: detected recent race for %q, calling fixRace", v, loc)
                                v.fixRace(loc)
-                               continue
+                               return
                        } else if err != nil {
                                log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, loc, err)
-                               continue
+                               return
                        }
                }
                if startT.Sub(trashT) < theConfig.TrashLifetime.Duration() {
-                       continue
+                       return
                }
                err = v.bucket.Del(trash.Key)
                if err != nil {
                        log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, trash.Key, err)
-                       continue
+                       return
                }
-               bytesDeleted += trash.Size
-               blocksDeleted++
+               atomic.AddInt64(&bytesDeleted, trash.Size)
+               atomic.AddInt64(&blocksDeleted, 1)
 
                _, err = v.bucket.Head(loc, nil)
-               if os.IsNotExist(err) {
-                       err = v.bucket.Del("recent/" + loc)
-                       if err != nil {
-                               log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, "recent/"+loc, err)
-                       }
-               } else if err != nil {
-                       log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, "recent/"+loc, err)
+               if err == nil {
+                       log.Printf("warning: %s: EmptyTrash: HEAD %q succeeded immediately after deleting %q", v, loc, loc)
+                       return
+               }
+               if !os.IsNotExist(v.translateError(err)) {
+                       log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, loc, err)
+                       return
+               }
+               err = v.bucket.Del("recent/" + loc)
+               if err != nil {
+                       log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, "recent/"+loc, err)
                }
        }
+
+       var wg sync.WaitGroup
+       todo := make(chan *s3.Key, theConfig.EmptyTrashWorkers)
+       for i := 0; i < 1 || i < theConfig.EmptyTrashWorkers; i++ {
+               wg.Add(1)
+               go func() {
+                       defer wg.Done()
+                       for key := range todo {
+                               emptyOneKey(key)
+                       }
+               }()
+       }
+
+       trashL := s3Lister{
+               Bucket:   v.bucket.Bucket,
+               Prefix:   "trash/",
+               PageSize: v.IndexPageSize,
+       }
+       for trash := trashL.First(); trash != nil; trash = trashL.Next() {
+               todo <- trash
+       }
+       close(todo)
+       wg.Wait()
+
        if err := trashL.Error(); err != nil {
                log.Printf("error: %s: EmptyTrash: lister: %s", v, err)
        }