seaweedfs/go/weed/benchmark.go

459 lines
12 KiB
Go
Raw Normal View History

2014-03-10 02:42:50 +00:00
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"
"io"
2014-03-10 06:12:05 +00:00
"math"
"math/rand"
2014-03-10 02:42:50 +00:00
"os"
"runtime"
"runtime/pprof"
"sort"
2014-03-10 02:42:50 +00:00
"strings"
"sync"
"time"
)
type BenchmarkOptions struct {
server *string
concurrency *int
numberOfFiles *int
fileSize *int
idListFile *string
write *bool
read *bool
sequentialRead *bool
collection *string
cpuprofile *string
vid2server map[string]string //cache for vid locations
2014-03-10 02:42:50 +00:00
}
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", 64, "number of concurrent write or read processes")
2014-03-10 02:42:50 +00:00
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")
b.sequentialRead = cmdBenchmark.Flag.Bool("readSequentially", false, "randomly read by ids from \"-list\" specified file")
b.collection = cmdBenchmark.Flag.String("collection", "benchmark", "write data to this collection")
b.cpuprofile = cmdBenchmark.Flag.String("cpuprofile", "", "write cpu profile to file")
b.vid2server = make(map[string]string)
2014-03-10 02:42:50 +00:00
}
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.
2014-03-13 19:11:26 +00:00
2014-03-10 02:42:50 +00:00
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.
2014-03-13 19:11:26 +00:00
Usually your network or the hard drive is the real bottleneck.
Another thing to watch is whether the volumes are evenly distributed
to each volume server. Because the 7 more benchmark volumes are randomly distributed
to servers with free slots, it's highly possible some servers have uneven amount of
benchmark volumes. To remedy this, you can use this to grow the benchmark volumes
before starting the benchmark command:
http://localhost:9333/vol/grow?collection=benchmark&count=5
2014-03-10 02:42:50 +00:00
`,
}
var (
2014-03-10 06:12:05 +00:00
wait sync.WaitGroup
writeStats *stats
readStats *stats
2014-03-10 02:42:50 +00:00
)
func runbenchmark(cmd *Command, args []string) bool {
fmt.Printf("This is Weed File System version %s %s %s\n", VERSION, runtime.GOOS, runtime.GOARCH)
if *b.cpuprofile != "" {
f, err := os.Create(*b.cpuprofile)
if err != nil {
glog.Fatal(err)
}
pprof.StartCPUProfile(f)
defer pprof.StopCPUProfile()
}
2014-03-10 02:42:50 +00:00
if *b.write {
bench_write()
2014-03-10 02:42:50 +00:00
}
if *b.read {
bench_read()
2014-03-10 02:42:50 +00:00
}
return true
}
func bench_write() {
fileIdLineChan := make(chan string)
finishChan := make(chan bool)
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, writeStats)
}
writeStats.start = time.Now()
go writeStats.checkProgress("Writing Benchmark", finishChan)
for i := 0; i < *b.numberOfFiles; i++ {
idChan <- i
}
close(idChan)
wait.Wait()
writeStats.end = time.Now()
wait.Add(1)
finishChan <- true
finishChan <- true
close(finishChan)
wait.Wait()
writeStats.printStats()
}
func bench_read() {
fileIdLineChan := make(chan string)
finishChan := make(chan bool)
readStats = newStats()
wait.Add(*b.concurrency)
go readFileIds(*b.idListFile, fileIdLineChan)
readStats.start = time.Now()
go readStats.checkProgress("Randomly Reading Benchmark", finishChan)
for i := 0; i < *b.concurrency; i++ {
go readFiles(fileIdLineChan, readStats)
}
wait.Wait()
finishChan <- true
close(finishChan)
readStats.end = time.Now()
readStats.printStats()
}
2014-03-10 06:12:05 +00:00
func writeFiles(idChan chan int, fileIdLineChan chan string, s *stats) {
serverLimitChan := make(map[string]chan bool)
2014-03-10 02:42:50 +00:00
for {
if id, ok := <-idChan; ok {
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, "", *b.collection); err == nil {
fp.Server, fp.Fid, fp.Collection = assignResult.PublicUrl, assignResult.Fid, *b.collection
if _, ok := serverLimitChan[fp.Server]; !ok {
serverLimitChan[fp.Server] = make(chan bool, 7)
}
serverLimitChan[fp.Server] <- true
if _, err := fp.Upload(0, *b.server); err == nil {
fileIdLineChan <- fp.Fid
s.completed++
s.transferred += int64(*b.fileSize)
} else {
s.failed++
}
2014-03-10 02:42:50 +00:00
writeStats.addSample(time.Now().Sub(start))
<-serverLimitChan[fp.Server]
2014-03-10 02:42:50 +00:00
if *cmdBenchmark.IsDebug {
fmt.Printf("writing %d file %s\n", id, fp.Fid)
}
} else {
2014-03-10 06:12:05 +00:00
s.failed++
2014-03-10 02:42:50 +00:00
println("writing file error:", err.Error())
}
} else {
break
}
}
wait.Done()
}
2014-03-10 06:12:05 +00:00
func readFiles(fileIdLineChan chan string, s *stats) {
serverLimitChan := make(map[string]chan bool)
masterLimitChan := make(chan bool, 7)
2014-03-10 02:42:50 +00:00
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 {
masterLimitChan <- true
2014-03-10 02:42:50 +00:00
if ret, err := operation.Lookup(*b.server, vid); err == nil {
if len(ret.Locations) > 0 {
server = ret.Locations[0].PublicUrl
b.vid2server[vid] = server
}
}
<-masterLimitChan
2014-03-10 02:42:50 +00:00
}
if server, ok := b.vid2server[vid]; ok {
if _, ok := serverLimitChan[server]; !ok {
serverLimitChan[server] = make(chan bool, 7)
}
serverLimitChan[server] <- true
2014-03-10 02:42:50 +00:00
url := "http://" + server + "/" + fid
if bytesRead, err := util.Get(url); err == nil {
2014-03-10 06:12:05 +00:00
s.completed++
s.transferred += int64(len(bytesRead))
2014-03-10 02:42:50 +00:00
readStats.addSample(time.Now().Sub(start))
} else {
2014-03-10 06:12:05 +00:00
s.failed++
2014-03-10 02:42:50 +00:00
println("!!!! Failed to read from ", url, " !!!!!")
}
<-serverLimitChan[server]
2014-03-10 02:42:50 +00:00
} else {
2014-03-10 06:12:05 +00:00
s.failed++
2014-03-10 02:42:50 +00:00
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)
if *b.sequentialRead {
for {
if line, err := Readln(r); err == nil {
fileIdLineChan <- string(line)
} else {
break
}
}
} else {
lines := make([]string, 0, *b.numberOfFiles)
for {
if line, err := Readln(r); err == nil {
lines = append(lines, string(line))
} else {
break
}
}
for i := 0; i < *b.numberOfFiles; i++ {
fileIdLineChan <- lines[rand.Intn(len(lines))]
2014-03-10 02:42:50 +00:00
}
}
2014-03-10 02:42:50 +00:00
close(fileIdLineChan)
}
2014-03-10 06:12:05 +00:00
const (
benchResolution = 10000 //0.1 microsecond
benchBucket = 1000000000 / benchResolution
)
// An efficient statics collecting and rendering
2014-03-10 02:42:50 +00:00
type stats struct {
2014-03-10 06:12:05 +00:00
data []int
overflow []int
2014-03-10 06:12:05 +00:00
completed int
failed int
transferred int64
start time.Time
end time.Time
2014-03-10 02:42:50 +00:00
}
2014-03-10 06:12:05 +00:00
var percentages = []int{50, 66, 75, 80, 90, 95, 98, 99, 100}
func newStats() *stats {
return &stats{data: make([]int, benchResolution), overflow: make([]int, 0)}
2014-03-10 02:42:50 +00:00
}
func (s *stats) addSample(d time.Duration) {
index := int(d / benchBucket)
if index < 0 {
fmt.Printf("This request takes %3.1f seconds, skipping!\n", float64(index)/10000)
} else if index < len(s.data) {
s.data[int(d/benchBucket)]++
} else {
s.overflow = append(s.overflow, index)
}
2014-03-10 02:42:50 +00:00
}
2014-03-10 06:12:05 +00:00
func (s *stats) checkProgress(testName string, finishChan chan bool) {
2014-03-10 06:12:05 +00:00
fmt.Printf("\n------------ %s ----------\n", testName)
ticker := time.Tick(time.Second)
for {
select {
case <-finishChan:
return
case <-ticker:
fmt.Printf("Completed %d of %d requests, %3.1f%%\n", s.completed, *b.numberOfFiles, float64(s.completed)*100/float64(*b.numberOfFiles))
}
}
}
func (s *stats) printStats() {
2014-03-10 06:12:05 +00:00
timeTaken := float64(int64(s.end.Sub(s.start))) / 1000000000
2014-03-10 20:22:08 +00:00
fmt.Printf("\nConcurrency Level: %d\n", *b.concurrency)
2014-03-10 06:12:05 +00:00
fmt.Printf("Time taken for tests: %.3f seconds\n", timeTaken)
fmt.Printf("Complete requests: %d\n", s.completed)
fmt.Printf("Failed requests: %d\n", s.failed)
fmt.Printf("Total transferred: %d bytes\n", s.transferred)
fmt.Printf("Requests per second: %.2f [#/sec]\n", float64(s.completed)/timeTaken)
fmt.Printf("Transfer rate: %.2f [Kbytes/sec]\n", float64(s.transferred)/1024/timeTaken)
n, sum := 0, 0
min, max := 10000000, 0
for i := 0; i < len(s.data); i++ {
n += s.data[i]
sum += s.data[i] * i
if s.data[i] > 0 {
if min > i {
min = i
}
if max < i {
max = i
}
}
}
n += len(s.overflow)
for i := 0; i < len(s.overflow); i++ {
sum += s.overflow[i]
if min > s.overflow[i] {
min = s.overflow[i]
}
if max < s.overflow[i] {
max = s.overflow[i]
}
}
2014-03-10 06:12:05 +00:00
avg := float64(sum) / float64(n)
varianceSum := 0.0
for i := 0; i < len(s.data); i++ {
if s.data[i] > 0 {
d := float64(i) - avg
varianceSum += d * d * float64(s.data[i])
}
}
for i := 0; i < len(s.overflow); i++ {
d := float64(s.overflow[i]) - avg
varianceSum += d * d
}
2014-03-10 06:12:05 +00:00
std := math.Sqrt(varianceSum / float64(n))
fmt.Printf("\nConnection Times (ms)\n")
fmt.Printf(" min avg max std\n")
fmt.Printf("Total: %2.1f %3.1f %3.1f %3.1f\n", float32(min)/10, float32(avg)/10, float32(max)/10, std/10)
//printing percentiles
fmt.Printf("\nPercentage of the requests served within a certain time (ms)\n")
percentiles := make([]int, len(percentages))
for i := 0; i < len(percentages); i++ {
percentiles[i] = n * percentages[i] / 100
}
percentiles[len(percentiles)-1] = n
percentileIndex := 0
currentSum := 0
for i := 0; i < len(s.data); i++ {
currentSum += s.data[i]
if s.data[i] > 0 && percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
fmt.Printf(" %3d%% %5.1f ms\n", percentages[percentileIndex], float32(i)/10.0)
percentileIndex++
for percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
percentileIndex++
}
2014-03-10 02:42:50 +00:00
}
}
sort.Ints(s.overflow)
for i := 0; i < len(s.overflow); i++ {
currentSum++
if percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
fmt.Printf(" %3d%% %5.1f ms\n", percentages[percentileIndex], float32(s.overflow[i])/10.0)
percentileIndex++
for percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
percentileIndex++
}
}
}
2014-03-10 02:42:50 +00:00
}
// 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
}