X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/6e10831be039fb540fc252d35e0ffe673ea8d398..8781021a9ab265b225f0f1eaef6170abde9ce3c3:/services/keepstore/keepstore.go?ds=sidebyside

diff --git a/services/keepstore/keepstore.go b/services/keepstore/keepstore.go
index c6cb00db0f..53cf7be6ca 100644
--- a/services/keepstore/keepstore.go
+++ b/services/keepstore/keepstore.go
@@ -14,6 +14,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 )
@@ -54,7 +55,10 @@ var data_manager_token string
 
 // never_delete can be used to prevent the DELETE handler from
 // actually deleting anything.
-var never_delete = false
+var never_delete = true
+
+var maxBuffers = 128
+var bufs *bufferPool
 
 // ==========
 // Error types.
@@ -129,7 +133,15 @@ func (vs *volumeSet) Set(value string) error {
 	if _, err := os.Stat(value); err != nil {
 		return err
 	}
-	*vs = append(*vs, MakeUnixVolume(value, flagSerializeIO, flagReadonly))
+	var locker sync.Locker
+	if flagSerializeIO {
+		locker = &sync.Mutex{}
+	}
+	*vs = append(*vs, &UnixVolume{
+		root:     value,
+		locker:   locker,
+		readonly: flagReadonly,
+	})
 	return nil
 }
 
@@ -195,7 +207,8 @@ func (vs *volumeSet) Discover() int {
 // permission arguments).
 
 func main() {
-	log.Println("Keep started: pid", os.Getpid())
+	log.Println("keepstore starting, pid", os.Getpid())
+	defer log.Println("keepstore exiting, pid", os.Getpid())
 
 	var (
 		data_manager_token_file string
@@ -224,7 +237,7 @@ func main() {
 	flag.BoolVar(
 		&never_delete,
 		"never-delete",
-		false,
+		true,
 		"If set, nothing will be deleted. HTTP 405 will be returned "+
 			"for valid DELETE requests.")
 	flag.StringVar(
@@ -271,10 +284,49 @@ func main() {
 		&pidfile,
 		"pid",
 		"",
-		"Path to write pid file")
+		"Path to write pid file during startup. This file is kept open and locked with LOCK_EX until keepstore exits, so `fuser -k pidfile` is one way to shut down. Exit immediately if there is an error opening, locking, or writing the pid file.")
+	flag.IntVar(
+		&maxBuffers,
+		"max-buffers",
+		maxBuffers,
+		fmt.Sprintf("Maximum RAM to use for data buffers, given in multiples of block size (%d MiB). When this limit is reached, HTTP requests requiring buffers (like GET and PUT) will wait for buffer space to be released.", BLOCKSIZE>>20))
 
 	flag.Parse()
 
+	if never_delete != true {
+		log.Fatal("never_delete must be true, see #6221")
+	}
+
+	if maxBuffers < 0 {
+		log.Fatal("-max-buffers must be greater than zero.")
+	}
+	bufs = newBufferPool(maxBuffers, BLOCKSIZE)
+
+	if pidfile != "" {
+		f, err := os.OpenFile(pidfile, os.O_RDWR|os.O_CREATE, 0777)
+		if err != nil {
+			log.Fatalf("open pidfile (%s): %s", pidfile, err)
+		}
+		err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+		if err != nil {
+			log.Fatalf("flock pidfile (%s): %s", pidfile, err)
+		}
+		err = f.Truncate(0)
+		if err != nil {
+			log.Fatalf("truncate pidfile (%s): %s", pidfile, err)
+		}
+		_, err = fmt.Fprint(f, os.Getpid())
+		if err != nil {
+			log.Fatalf("write pidfile (%s): %s", pidfile, err)
+		}
+		err = f.Sync()
+		if err != nil {
+			log.Fatalf("sync pidfile (%s): %s", pidfile, err)
+		}
+		defer f.Close()
+		defer os.Remove(pidfile)
+	}
+
 	if len(volumes) == 0 {
 		if volumes.Discover() == 0 {
 			log.Fatal("No volumes found.")
@@ -312,7 +364,7 @@ func main() {
 			log.Println("Running without a PermissionSecret. Block locators " +
 				"returned by this server will not be signed, and will be rejected " +
 				"by a server that enforces permissions.")
-			log.Println("To fix this, use the -permission-key-file flag " +
+			log.Println("To fix this, use the -blob-signing-key-file flag " +
 				"to specify the file containing the permission key.")
 		}
 	}
@@ -357,24 +409,9 @@ func main() {
 		listener.Close()
 	}(term)
 	signal.Notify(term, syscall.SIGTERM)
+	signal.Notify(term, syscall.SIGINT)
 
-	if pidfile != "" {
-		f, err := os.Create(pidfile)
-		if err == nil {
-			fmt.Fprint(f, os.Getpid())
-			f.Close()
-		} else {
-			log.Printf("Error writing pid file (%s): %s", pidfile, err.Error())
-		}
-	}
-
-	// Start listening for requests.
+	log.Println("listening at", listen)
 	srv := &http.Server{Addr: listen}
 	srv.Serve(listener)
-
-	log.Println("shutting down")
-
-	if pidfile != "" {
-		os.Remove(pidfile)
-	}
 }