From 86b17e8972ffafbb68fce51521ed7e053f908cef Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 9 Mar 2014 19:42:50 -0700 Subject: [PATCH] a kind of working benchmark --- go/weed/benchmark.go | 288 +++++++++++++++++++++++++++++++++++++++++++ go/weed/weed.go | 1 + 2 files changed, 289 insertions(+) create mode 100644 go/weed/benchmark.go diff --git a/go/weed/benchmark.go b/go/weed/benchmark.go new file mode 100644 index 000000000..dfba52b06 --- /dev/null +++ b/go/weed/benchmark.go @@ -0,0 +1,288 @@ +package main + +import ( + "bufio" + "code.google.com/p/weed-fs/go/glog" + "code.google.com/p/weed-fs/go/operation" + "code.google.com/p/weed-fs/go/util" + "fmt" + "github.com/patrick-higgins/summstat" + "io" + "os" + "strings" + "sync" + "time" +) + +type BenchmarkOptions struct { + server *string + concurrency *int + numberOfFiles *int + fileSize *int + idListFile *string + write *bool + read *bool + vid2server map[string]string //cache for vid locations +} + +var ( + b BenchmarkOptions +) + +func init() { + cmdBenchmark.Run = runbenchmark // break init cycle + cmdBenchmark.IsDebug = cmdBenchmark.Flag.Bool("debug", false, "verbose debug information") + b.server = cmdBenchmark.Flag.String("server", "localhost:9333", "weedfs master location") + b.concurrency = cmdBenchmark.Flag.Int("c", 7, "number of concurrent write or read processes") + b.fileSize = cmdBenchmark.Flag.Int("size", 1024, "simulated file size in bytes") + b.numberOfFiles = cmdBenchmark.Flag.Int("n", 1024*1024, "number of files to write for each thread") + b.idListFile = cmdBenchmark.Flag.String("list", os.TempDir()+"/benchmark_list.txt", "list of uploaded file ids") + b.write = cmdBenchmark.Flag.Bool("write", true, "enable write") + b.read = cmdBenchmark.Flag.Bool("read", true, "enable read") +} + +var cmdBenchmark = &Command{ + UsageLine: "benchmark -server=localhost:9333 -c=10 -n=100000", + Short: "benchmark on writing millions of files and read out", + Long: `benchmark on an empty weed file system. + + Two tests during benchmark: + 1) write lots of small files to the system + 2) read the files out + + The file content is mostly zero, but no compression is done. + + By default, write 1 million files of 1KB each with 7 concurrent threads, + and randomly read them out with 7 concurrent threads. + + You can choose to only benchmark read or write. + During write, the list of uploaded file ids is stored in "-list" specified file. + You can also use your own list of file ids to run read test. + + Write speed and read speed will be collected. + The numbers are used to get a sense of the system. + But usually your network or the hard drive is + the real bottleneck. + + `, +} + +var ( + countWritten int64 + sizeWritten int64 + countRead int64 + sizeRead int64 + wait sync.WaitGroup + writeStats stats + readStats stats +) + +func runbenchmark(cmd *Command, args []string) bool { + finishChan := make(chan bool) + fileIdLineChan := make(chan string) + b.vid2server = make(map[string]string) + + if *b.write { + writeStats = newStats() + idChan := make(chan int) + wait.Add(*b.concurrency) + go writeFileIds(*b.idListFile, fileIdLineChan, finishChan) + for i := 0; i < *b.concurrency; i++ { + go writeFiles(idChan, fileIdLineChan) + } + for i := 0; i < *b.numberOfFiles; i++ { + idChan <- i + } + close(idChan) + wait.Wait() + wait.Add(1) + finishChan <- true + wait.Wait() + println("completed", countWritten, "write requests, bytes:", sizeWritten) + writeStats.printStats() + } + + if *b.read { + readStats = newStats() + wait.Add(*b.concurrency) + go readFileIds(*b.idListFile, fileIdLineChan) + for i := 0; i < *b.concurrency; i++ { + go readFiles(fileIdLineChan) + } + wait.Wait() + println("completed", countRead, "read requests, bytes:", sizeRead) + readStats.printStats() + } + + return true +} + +func writeFiles(idChan chan int, fileIdLineChan chan string) { + for { + if id, ok := <-idChan; ok { + countWritten++ + start := time.Now() + fp := &operation.FilePart{Reader: &FakeReader{id: uint64(id), size: int64(*b.fileSize)}, FileSize: int64(*b.fileSize)} + if assignResult, err := operation.Assign(*b.server, 1, ""); err == nil { + fp.Server, fp.Fid = assignResult.PublicUrl, assignResult.Fid + fp.Upload(0, *b.server, "") + writeStats.addSample(time.Now().Sub(start)) + fileIdLineChan <- fp.Fid + sizeWritten += int64(*b.fileSize) + if *cmdBenchmark.IsDebug { + fmt.Printf("writing %d file %s\n", id, fp.Fid) + } + } else { + println("writing file error:", err.Error()) + } + } else { + break + } + } + wait.Done() +} + +func readFiles(fileIdLineChan chan string) { + for { + if fid, ok := <-fileIdLineChan; ok { + if len(fid) == 0 { + continue + } + if fid[0] == '#' { + continue + } + if *cmdBenchmark.IsDebug { + fmt.Printf("reading file %s\n", fid) + } + parts := strings.SplitN(fid, ",", 2) + vid := parts[0] + start := time.Now() + if server, ok := b.vid2server[vid]; !ok { + if ret, err := operation.Lookup(*b.server, vid); err == nil { + if len(ret.Locations) > 0 { + server = ret.Locations[0].PublicUrl + b.vid2server[vid] = server + } + } + } + if server, ok := b.vid2server[vid]; ok { + url := "http://" + server + "/" + fid + if bytesRead, err := util.Get(url); err == nil { + countRead++ + sizeRead += int64(len(bytesRead)) + readStats.addSample(time.Now().Sub(start)) + } else { + println("!!!! Failed to read from ", url, " !!!!!") + } + } else { + println("!!!! volume id ", vid, " location not found!!!!!") + } + } else { + break + } + } + wait.Done() +} + +func writeFileIds(fileName string, fileIdLineChan chan string, finishChan chan bool) { + file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + glog.Fatalf("File to create file %s: %s\n", fileName, err) + } + defer file.Close() + + for { + select { + case <-finishChan: + wait.Done() + return + case line := <-fileIdLineChan: + file.Write([]byte(line)) + file.Write([]byte("\n")) + } + } +} + +func readFileIds(fileName string, fileIdLineChan chan string) { + file, err := os.Open(fileName) // For read access. + if err != nil { + glog.Fatalf("File to read file %s: %s\n", fileName, err) + } + defer file.Close() + + r := bufio.NewReader(file) + for { + if line, err := Readln(r); err == nil { + fileIdLineChan <- string(line) + } else { + break + } + } + close(fileIdLineChan) +} + +type stats struct { + stats *summstat.Stats +} + +func newStats() stats { + s := stats{stats: summstat.NewStats()} + s.stats.CreateBins(21, 1000, 20000) + return s +} + +func (s stats) addSample(d time.Duration) { + //fmt.Printf("adding %d μs\n", int(d / 1000)) + s.stats.AddSample(summstat.Sample(int(d / 1000))) +} +func (s stats) printStats() { + fmt.Printf("\n%s Count:%d ", time.Now().Local(), s.stats.Count()) + fmt.Printf("Min:%.2fms ", s.stats.Min()/1000) + fmt.Printf("Max:%.2fms ", s.stats.Max()/1000) + fmt.Printf("Stddev:%.2fms\n", s.stats.Stddev()/1000) + for b := 0; b < s.stats.NBins(); b++ { + count, _, _ := s.stats.Bin(b) + if b+1 != s.stats.NBins() { + fmt.Printf("%2d ~ %2d ms: %2.4f%% %d\n", b, (b + 1), float64(count)*100.0/float64(s.stats.Count()), count) + } else { + fmt.Printf("%2d ~ ms: %2.4f%% %d\n", b, float64(count)*100.0/float64(s.stats.Count()), count) + } + } +} + +// a fake reader to generate content to upload +type FakeReader struct { + id uint64 // an id number + size int64 // max bytes +} + +func (l *FakeReader) Read(p []byte) (n int, err error) { + if l.size <= 0 { + return 0, io.EOF + } + if int64(len(p)) > l.size { + n = int(l.size) + } else { + n = len(p) + } + for i := 0; i < n-8; i += 8 { + for s := uint(0); s < 8; s++ { + p[i] = byte(l.id >> (s * 8)) + } + } + l.size -= int64(n) + return +} + +func Readln(r *bufio.Reader) ([]byte, error) { + var ( + isPrefix bool = true + err error = nil + line, ln []byte + ) + for isPrefix && err == nil { + line, isPrefix, err = r.ReadLine() + ln = append(ln, line...) + } + return ln, err +} diff --git a/go/weed/weed.go b/go/weed/weed.go index b02b1f6ff..2a5a040c1 100644 --- a/go/weed/weed.go +++ b/go/weed/weed.go @@ -19,6 +19,7 @@ var IsDebug *bool var server *string var commands = []*Command{ + cmdBenchmark, cmdCompact, cmdFix, cmdServer,