[ARVADOS] created: dc6888a23d9af8d87dcb11806c93424dff6c1753

[git at public.curoverse.com]

        at  dc6888a23d9af8d87dcb11806c93424dff6c1753 (commit)


commit dc6888a23d9af8d87dcb11806c93424dff6c1753
Author: Tom Clegg <tom at curoverse.com>
Date:   Fri Oct 2 19:49:18 2015 -0400

    7410: Add keepexercise

diff --git a/tools/keepexercise/keepexercise.go b/tools/keepexercise/keepexercise.go
new file mode 100644
index 0000000..c475f6c
--- /dev/null
+++ b/tools/keepexercise/keepexercise.go
@@ -0,0 +1,144 @@
+// Testing tool for Keep services.
+//
+// keepexercise helps measure throughput and test reliability under
+// various usage patterns.
+//
+// By default, it reads and writes blocks containing 2^26 NUL
+// bytes. This generates network traffic without consuming much disk
+// space.
+//
+// For a more realistic test, enable -vary-data. Warning: this will
+// fill your storage volumes with random data if you leave it running,
+// which can cost you money or leave you with too little room for
+// useful data.
+//
+package main
+
+import (
+	"crypto/rand"
+	"flag"
+	"io"
+	"io/ioutil"
+	"log"
+	"time"
+
+	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
+	"git.curoverse.com/arvados.git/sdk/go/keepclient"
+)
+
+var BlockSize = flag.Int("block-size", keepclient.BLOCKSIZE, "bytes per read/write op")
+var ReadThreads = flag.Int("rthreads", 1, "number of concurrent readers")
+var WriteThreads = flag.Int("wthreads", 1, "number of concurrent writers")
+var VaryData = flag.Bool("vary-data", false, "vary the data for each request: consumes disk space, exercises write behavior")
+var Replicas = flag.Int("replicas", 1, "replication level for writing")
+var StatsInterval = flag.Duration("stats-interval", time.Second, "time interval between IO stats reports, or 0 to disable")
+
+func main() {
+	flag.Parse()
+
+	arv, err := arvadosclient.MakeArvadosClient()
+	if err != nil {
+		log.Fatal(err)
+	}
+	kc, err := keepclient.MakeKeepClient(&arv)
+	if err != nil {
+		log.Fatal(err)
+	}
+	kc.Want_replicas = *Replicas
+
+	nextBuf := make(chan []byte, *WriteThreads)
+	nextLocator := make(chan string, *ReadThreads + *WriteThreads)
+
+	go countBeans(nextLocator)
+	for i := 0; i < *WriteThreads; i++ {
+		go makeBufs(nextBuf)
+		go doWrites(kc, nextBuf, nextLocator)
+	}
+	for i := 0; i < *ReadThreads; i++ {
+		go doReads(kc, nextLocator)
+	}
+	<-make(chan struct{})
+}
+
+// Send 1234 to bytesInChan when we receive 1234 bytes from keepstore.
+var bytesInChan = make(chan uint64)
+var bytesOutChan = make(chan uint64)
+
+// Send struct{}{} to errorsChan when an error happens.
+var errorsChan = make(chan struct{})
+
+func countBeans(nextLocator chan string) {
+	t0 := time.Now()
+	var tickChan <-chan time.Time
+	if *StatsInterval > 0 {
+		tickChan = time.NewTicker(*StatsInterval).C
+	}
+	var bytesIn uint64
+	var bytesOut uint64
+	var errors uint64
+	for {
+		select {
+		case <-tickChan:
+			elapsed := time.Since(t0)
+			log.Printf("%v elapsed: read %v bytes (%.1f MiB/s), wrote %v bytes (%.1f MiB/s), errors %d",
+				elapsed,
+				bytesIn, (float64(bytesIn) / elapsed.Seconds() / 1048576),
+				bytesOut, (float64(bytesOut) / elapsed.Seconds() / 1048576),
+				errors,
+			)
+		case i := <-bytesInChan:
+			bytesIn += i
+		case o := <-bytesOutChan:
+			bytesOut += o
+		case <-errorsChan:
+			errors++
+		}
+	}
+}
+
+func makeBufs(nextBuf chan []byte) {
+	buf := make([]byte, *BlockSize)
+	if *VaryData {
+		if _, err := io.ReadFull(rand.Reader, buf); err != nil {
+			log.Fatal(err)
+		}
+	}
+	for {
+		nextBuf <- buf
+	}
+}
+
+func doWrites(kc *keepclient.KeepClient, nextBuf chan []byte, nextLocator chan string) {
+	for buf := range nextBuf {
+		locator, _, err := kc.PutB(buf)
+		if err != nil {
+			log.Print(err)
+			errorsChan <- struct{}{}
+			continue
+		}
+		bytesOutChan <- uint64(len(buf))
+		for cap(nextLocator) > len(nextLocator) + *WriteThreads {
+			// Give the readers something to do, unless
+			// they have lots queued up already.
+			nextLocator <- locator
+		}
+	}
+}
+
+func doReads(kc *keepclient.KeepClient, nextLocator chan string) {
+	for locator := range nextLocator {
+		rdr, size, url, err := kc.Get(locator)
+		if err != nil {
+			log.Print(err)
+			errorsChan <- struct{}{}
+			continue
+		}
+		n, err := io.Copy(ioutil.Discard, rdr)
+		bytesInChan <- uint64(n)
+		if n != size {
+			log.Printf("Got %d bytes (expected %d) from %s", n, size, url)
+			errorsChan <- struct{}{}
+			continue
+		}
+	}
+}

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


hooks/post-receive
--