[ARVADOS] created: 1.1.4-147-g2e16af6

[Git user]

        at  2e16af6f5e76b3f6cba1601c429517e4525ef106 (commit)


commit 2e16af6f5e76b3f6cba1601c429517e4525ef106
Author: Tom Clegg <tclegg at veritasgenetics.com>
Date:   Tue Apr 24 16:32:24 2018 -0400

    13335: Add PullWorkers, TrashWorkers, EmptyTrashWorkers configs.
    
    EmptyTrashWorkers only applies to S3 volumes.
    
    Arvados-DCO-1.1-Signed-off-by: Tom Clegg <tclegg at veritasgenetics.com>

diff --git a/services/keepstore/config.go b/services/keepstore/config.go
index 17d6acd..19dc7f6 100644
--- a/services/keepstore/config.go
+++ b/services/keepstore/config.go
@@ -40,6 +40,9 @@ type Config struct {
 	EnableDelete        bool
 	TrashLifetime       arvados.Duration
 	TrashCheckInterval  arvados.Duration
+	PullWorkers         int
+	TrashWorkers        int
+	EmptyTrashWorkers   int
 
 	Volumes VolumeList
 
diff --git a/services/keepstore/keepstore.go b/services/keepstore/keepstore.go
index 03eef7e..c742752 100644
--- a/services/keepstore/keepstore.go
+++ b/services/keepstore/keepstore.go
@@ -165,19 +165,23 @@ func main() {
 		log.Fatal(err)
 	}
 
-	// Initialize Pull queue and worker
+	// Initialize keepclient for pull workers
 	keepClient := &keepclient.KeepClient{
 		Arvados:       &arvadosclient.ArvadosClient{},
 		Want_replicas: 1,
 	}
 
-	// Initialize the pullq and worker
+	// Initialize the pullq and workers
 	pullq = NewWorkQueue()
-	go RunPullWorker(pullq, keepClient)
+	for i := 0; i < 1 || i < theConfig.PullWorkers; i++ {
+		go RunPullWorker(pullq, keepClient)
+	}
 
-	// Initialize the trashq and worker
+	// Initialize the trashq and workers
 	trashq = NewWorkQueue()
-	go RunTrashWorker(trashq)
+	for i := 0; i < 1 || i < theConfig.TrashWorkers; i++ {
+		go RunTrashWorker(trashq)
+	}
 
 	// Start emptyTrash goroutine
 	doneEmptyingTrash := make(chan bool)
diff --git a/services/keepstore/s3_volume.go b/services/keepstore/s3_volume.go
index a60b2fc..532a082 100644
--- 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"
@@ -764,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)) {
@@ -792,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) {
@@ -815,28 +811,28 @@ 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) {
@@ -848,6 +844,30 @@ func (v *S3Volume) EmptyTrash() {
 			log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, "recent/"+loc, err)
 		}
 	}
+
+	var wg sync.WaitGroup
+	todo := make(chan *s3.Key)
+	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)
 	}

-----------------------------------------------------------------------


hooks/post-receive
--