tools/keep-exercise/keep-exercise.go

   1 // Testing tool for Keep services.
   2 //
   3 // keepexercise helps measure throughput and test reliability under
   4 // various usage patterns.
   5 //
   6 // By default, it reads and writes blocks containing 2^26 NUL
   7 // bytes. This generates network traffic without consuming much disk
   8 // space.
   9 //
  10 // For a more realistic test, enable -vary-request. Warning: this will
  11 // fill your storage volumes with random data if you leave it running,
  12 // which can cost you money or leave you with too little room for
  13 // useful data.
  14 //
  15 package main
  16
  17 import (
  18         "crypto/rand"
  19         "encoding/binary"
  20         "flag"
  21         "io"
  22         "io/ioutil"
  23         "log"
  24         "time"
  25
  26         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  27         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  28 )
  29
  30 // Command line config knobs
  31 var (
  32         BlockSize     = flag.Int("block-size", keepclient.BLOCKSIZE, "bytes per read/write op")
  33         ReadThreads   = flag.Int("rthreads", 1, "number of concurrent readers")
  34         WriteThreads  = flag.Int("wthreads", 1, "number of concurrent writers")
  35         VaryRequest   = flag.Bool("vary-request", false, "vary the data for each request: consumes disk space, exercises write behavior")
  36         VaryThread    = flag.Bool("vary-thread", false, "use -wthreads different data blocks")
  37         Replicas      = flag.Int("replicas", 1, "replication level for writing")
  38         StatsInterval = flag.Duration("stats-interval", time.Second, "time interval between IO stats reports, or 0 to disable")
  39 )
  40
  41 func main() {
  42         flag.Parse()
  43
  44         arv, err := arvadosclient.MakeArvadosClient()
  45         if err != nil {
  46                 log.Fatal(err)
  47         }
  48         kc, err := keepclient.MakeKeepClient(&arv)
  49         if err != nil {
  50                 log.Fatal(err)
  51         }
  52         kc.Want_replicas = *Replicas
  53         kc.Client.Timeout = 10 * time.Minute
  54
  55         nextBuf := make(chan []byte, *WriteThreads)
  56         nextLocator := make(chan string, *ReadThreads+*WriteThreads)
  57
  58         go countBeans(nextLocator)
  59         for i := 0; i < *WriteThreads; i++ {
  60                 go makeBufs(nextBuf, i)
  61                 go doWrites(kc, nextBuf, nextLocator)
  62         }
  63         for i := 0; i < *ReadThreads; i++ {
  64                 go doReads(kc, nextLocator)
  65         }
  66         <-make(chan struct{})
  67 }
  68
  69 // Send 1234 to bytesInChan when we receive 1234 bytes from keepstore.
  70 var bytesInChan = make(chan uint64)
  71 var bytesOutChan = make(chan uint64)
  72
  73 // Send struct{}{} to errorsChan when an error happens.
  74 var errorsChan = make(chan struct{})
  75
  76 func countBeans(nextLocator chan string) {
  77         t0 := time.Now()
  78         var tickChan <-chan time.Time
  79         if *StatsInterval > 0 {
  80                 tickChan = time.NewTicker(*StatsInterval).C
  81         }
  82         var bytesIn uint64
  83         var bytesOut uint64
  84         var errors uint64
  85         for {
  86                 select {
  87                 case <-tickChan:
  88                         elapsed := time.Since(t0)
  89                         log.Printf("%v elapsed: read %v bytes (%.1f MiB/s), wrote %v bytes (%.1f MiB/s), errors %d",
  90                                 elapsed,
  91                                 bytesIn, (float64(bytesIn) / elapsed.Seconds() / 1048576),
  92                                 bytesOut, (float64(bytesOut) / elapsed.Seconds() / 1048576),
  93                                 errors,
  94                         )
  95                 case i := <-bytesInChan:
  96                         bytesIn += i
  97                 case o := <-bytesOutChan:
  98                         bytesOut += o
  99                 case <-errorsChan:
 100                         errors++
 101                 }
 102         }
 103 }
 104
 105 func makeBufs(nextBuf chan []byte, threadID int) {
 106         buf := make([]byte, *BlockSize)
 107         if *VaryThread {
 108                 binary.PutVarint(buf, int64(threadID))
 109         }
 110         for {
 111                 if *VaryRequest {
 112                         if _, err := io.ReadFull(rand.Reader, buf); err != nil {
 113                                 log.Fatal(err)
 114                         }
 115                 }
 116                 nextBuf <- buf
 117         }
 118 }
 119
 120 func doWrites(kc *keepclient.KeepClient, nextBuf chan []byte, nextLocator chan string) {
 121         for buf := range nextBuf {
 122                 locator, _, err := kc.PutB(buf)
 123                 if err != nil {
 124                         log.Print(err)
 125                         errorsChan <- struct{}{}
 126                         continue
 127                 }
 128                 bytesOutChan <- uint64(len(buf))
 129                 for cap(nextLocator) > len(nextLocator)+*WriteThreads {
 130                         // Give the readers something to do, unless
 131                         // they have lots queued up already.
 132                         nextLocator <- locator
 133                 }
 134         }
 135 }
 136
 137 func doReads(kc *keepclient.KeepClient, nextLocator chan string) {
 138         for locator := range nextLocator {
 139                 rdr, size, url, err := kc.Get(locator)
 140                 if err != nil {
 141                         log.Print(err)
 142                         errorsChan <- struct{}{}
 143                         continue
 144                 }
 145                 n, err := io.Copy(ioutil.Discard, rdr)
 146                 rdr.Close()
 147                 if n != size || err != nil {
 148                         log.Printf("Got %d bytes (expected %d) from %s: %v", n, size, url, err)
 149                         errorsChan <- struct{}{}
 150                         continue
 151                         // Note we don't count the bytes received in
 152                         // partial/corrupt responses: we are measuring
 153                         // throughput, not resource consumption.
 154                 }
 155                 bytesInChan <- uint64(n)
 156         }
 157 }