aboutsummaryrefslogtreecommitdiff
path: root/go/weed/benchmark.go
diff options
context:
space:
mode:
Diffstat (limited to 'go/weed/benchmark.go')
-rw-r--r--go/weed/benchmark.go277
1 files changed, 149 insertions, 128 deletions
diff --git a/go/weed/benchmark.go b/go/weed/benchmark.go
index fec8472e5..f4f0b1874 100644
--- a/go/weed/benchmark.go
+++ b/go/weed/benchmark.go
@@ -2,9 +2,6 @@ package main
import (
"bufio"
- "github.com/chrislusf/weed-fs/go/glog"
- "github.com/chrislusf/weed-fs/go/operation"
- "github.com/chrislusf/weed-fs/go/util"
"fmt"
"io"
"math"
@@ -16,6 +13,10 @@ import (
"strings"
"sync"
"time"
+
+ "github.com/chrislusf/weed-fs/go/glog"
+ "github.com/chrislusf/weed-fs/go/operation"
+ "github.com/chrislusf/weed-fs/go/util"
)
type BenchmarkOptions struct {
@@ -30,11 +31,14 @@ type BenchmarkOptions struct {
sequentialRead *bool
collection *string
cpuprofile *string
+ maxCpu *int
vid2server map[string]string //cache for vid locations
+
}
var (
- b BenchmarkOptions
+ b BenchmarkOptions
+ sharedBytes []byte
)
func init() {
@@ -50,33 +54,35 @@ func init() {
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.cpuprofile = cmdBenchmark.Flag.String("cpuprofile", "", "cpu profile output file")
+ b.maxCpu = cmdBenchmark.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
b.vid2server = make(map[string]string)
+ sharedBytes = make([]byte, 1024)
}
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.
-
+
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.
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
+ 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
@@ -87,18 +93,17 @@ var cmdBenchmark = &Command{
}
var (
- wait sync.WaitGroup
- writeStats *stats
- readStats *stats
- serverLimitChan map[string]chan bool
+ wait sync.WaitGroup
+ writeStats *stats
+ readStats *stats
)
-func init() {
- serverLimitChan = make(map[string]chan bool)
-}
-
func runbenchmark(cmd *Command, args []string) bool {
fmt.Printf("This is Seaweed File System version %s %s %s\n", util.VERSION, runtime.GOOS, runtime.GOARCH)
+ if *b.maxCpu < 1 {
+ *b.maxCpu = runtime.NumCPU()
+ }
+ runtime.GOMAXPROCS(*b.maxCpu)
if *b.cpuprofile != "" {
f, err := os.Create(*b.cpuprofile)
if err != nil {
@@ -122,12 +127,12 @@ func runbenchmark(cmd *Command, args []string) bool {
func bench_write() {
fileIdLineChan := make(chan string)
finishChan := make(chan bool)
- writeStats = newStats()
+ writeStats = newStats(*b.concurrency)
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)
+ wait.Add(1)
+ go writeFiles(idChan, fileIdLineChan, &writeStats.localStats[i])
}
writeStats.start = time.Now()
writeStats.total = *b.numberOfFiles
@@ -138,28 +143,30 @@ func bench_write() {
close(idChan)
wait.Wait()
writeStats.end = time.Now()
- wait.Add(1)
+ wait.Add(2)
finishChan <- true
finishChan <- true
- close(finishChan)
wait.Wait()
+ close(finishChan)
writeStats.printStats()
}
func bench_read() {
fileIdLineChan := make(chan string)
finishChan := make(chan bool)
- readStats = newStats()
- wait.Add(*b.concurrency)
+ readStats = newStats(*b.concurrency)
go readFileIds(*b.idListFile, fileIdLineChan)
readStats.start = time.Now()
readStats.total = *b.numberOfFiles
go readStats.checkProgress("Randomly Reading Benchmark", finishChan)
for i := 0; i < *b.concurrency; i++ {
- go readFiles(fileIdLineChan, readStats)
+ wait.Add(1)
+ go readFiles(fileIdLineChan, &readStats.localStats[i])
}
wait.Wait()
+ wait.Add(1)
finishChan <- true
+ wait.Wait()
close(finishChan)
readStats.end = time.Now()
readStats.printStats()
@@ -170,126 +177,102 @@ type delayedFile struct {
fp *operation.FilePart
}
-func writeFiles(idChan chan int, fileIdLineChan chan string, s *stats) {
+func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
+ defer wait.Done()
delayedDeleteChan := make(chan *delayedFile, 100)
var waitForDeletions sync.WaitGroup
for i := 0; i < 7; i++ {
+ waitForDeletions.Add(1)
go func() {
- waitForDeletions.Add(1)
+ defer waitForDeletions.Done()
for df := range delayedDeleteChan {
- if df == nil {
- break
- }
if df.enterTime.After(time.Now()) {
time.Sleep(df.enterTime.Sub(time.Now()))
}
- fp := df.fp
- serverLimitChan[fp.Server] <- true
- if e := util.Delete("http://" + fp.Server + "/" + fp.Fid); e == nil {
+ if e := util.Delete("http://" + df.fp.Server + "/" + df.fp.Fid); e == nil {
s.completed++
} else {
s.failed++
}
- <-serverLimitChan[fp.Server]
}
- waitForDeletions.Done()
}()
}
- for {
- if id, ok := <-idChan; ok {
- start := time.Now()
- fileSize := int64(*b.fileSize + rand.Intn(64))
- fp := &operation.FilePart{Reader: &FakeReader{id: uint64(id), size: fileSize}, FileSize: 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 {
- if rand.Intn(100) < *b.deletePercentage {
- s.total++
- delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}
- } else {
- fileIdLineChan <- fp.Fid
- }
- s.completed++
- s.transferred += fileSize
+ for id := range idChan {
+ start := time.Now()
+ fileSize := int64(*b.fileSize + rand.Intn(64))
+ fp := &operation.FilePart{Reader: &FakeReader{id: uint64(id), size: fileSize}, FileSize: 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 _, err := fp.Upload(0, *b.server); err == nil {
+ if rand.Intn(100) < *b.deletePercentage {
+ s.total++
+ delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}
} else {
- s.failed++
- }
- writeStats.addSample(time.Now().Sub(start))
- <-serverLimitChan[fp.Server]
- if *cmdBenchmark.IsDebug {
- fmt.Printf("writing %d file %s\n", id, fp.Fid)
+ fileIdLineChan <- fp.Fid
}
+ s.completed++
+ s.transferred += fileSize
} else {
s.failed++
- println("writing file error:", err.Error())
+ fmt.Printf("Failed to write with error:%v\n", err)
+ }
+ writeStats.addSample(time.Now().Sub(start))
+ if *cmdBenchmark.IsDebug {
+ fmt.Printf("writing %d file %s\n", id, fp.Fid)
}
} else {
- break
+ s.failed++
+ println("writing file error:", err.Error())
}
}
close(delayedDeleteChan)
waitForDeletions.Wait()
- wait.Done()
}
-func readFiles(fileIdLineChan chan string, s *stats) {
- serverLimitChan := make(map[string]chan bool)
+func readFiles(fileIdLineChan chan string, s *stat) {
+ defer wait.Done()
masterLimitChan := make(chan bool, 1)
- 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
- if _, now_ok := b.vid2server[vid]; !now_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
- }
+ for fid := range fileIdLineChan {
+ 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
+ if _, now_ok := b.vid2server[vid]; !now_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
}
}
- <-masterLimitChan
}
- if server, ok := b.vid2server[vid]; ok {
- if _, ok := serverLimitChan[server]; !ok {
- serverLimitChan[server] = make(chan bool, 7)
- }
- serverLimitChan[server] <- true
- url := "http://" + server + "/" + fid
- if bytesRead, err := util.Get(url); err == nil {
- s.completed++
- s.transferred += int64(len(bytesRead))
- readStats.addSample(time.Now().Sub(start))
- } else {
- s.failed++
- println("!!!! Failed to read from ", url, " !!!!!")
- }
- <-serverLimitChan[server]
+ <-masterLimitChan
+ }
+ if server, ok := b.vid2server[vid]; ok {
+ url := "http://" + server + "/" + fid
+ if bytesRead, err := util.Get(url); err == nil {
+ s.completed++
+ s.transferred += int64(len(bytesRead))
+ readStats.addSample(time.Now().Sub(start))
} else {
s.failed++
- println("!!!! volume id ", vid, " location not found!!!!!")
+ fmt.Printf("Failed to read %s error:%v\n", url, err)
}
} else {
- break
+ s.failed++
+ println("!!!! volume id ", vid, " location not found!!!!!")
}
}
- wait.Done()
}
func writeFileIds(fileName string, fileIdLineChan chan string, finishChan chan bool) {
@@ -353,20 +336,28 @@ const (
// An efficient statics collecting and rendering
type stats struct {
- data []int
- overflow []int
+ data []int
+ overflow []int
+ localStats []stat
+ start time.Time
+ end time.Time
+ total int
+}
+type stat struct {
completed int
failed int
total int
transferred int64
- start time.Time
- end time.Time
}
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)}
+func newStats(n int) *stats {
+ return &stats{
+ data: make([]int, benchResolution),
+ overflow: make([]int, 0),
+ localStats: make([]stat, n),
+ }
}
func (s *stats) addSample(d time.Duration) {
@@ -387,28 +378,41 @@ func (s *stats) checkProgress(testName string, finishChan chan bool) {
for {
select {
case <-finishChan:
+ wait.Done()
return
case t := <-ticker:
- completed, transferred, taken := s.completed-lastCompleted, s.transferred-lastTransferred, t.Sub(lastTime)
+ completed, transferred, taken, total := 0, int64(0), t.Sub(lastTime), s.total
+ for _, localStat := range s.localStats {
+ completed += localStat.completed
+ transferred += localStat.transferred
+ total += localStat.total
+ }
fmt.Printf("Completed %d of %d requests, %3.1f%% %3.1f/s %3.1fMB/s\n",
- s.completed, s.total, float64(s.completed)*100/float64(s.total),
- float64(completed)*float64(int64(time.Second))/float64(int64(taken)),
- float64(transferred)*float64(int64(time.Second))/float64(int64(taken))/float64(1024*1024),
+ completed, total, float64(completed)*100/float64(total),
+ float64(completed-lastCompleted)*float64(int64(time.Second))/float64(int64(taken)),
+ float64(transferred-lastTransferred)*float64(int64(time.Second))/float64(int64(taken))/float64(1024*1024),
)
- lastCompleted, lastTransferred, lastTime = s.completed, s.transferred, t
+ lastCompleted, lastTransferred, lastTime = completed, transferred, t
}
}
}
func (s *stats) printStats() {
+ completed, failed, transferred, total := 0, 0, int64(0), s.total
+ for _, localStat := range s.localStats {
+ completed += localStat.completed
+ failed += localStat.failed
+ transferred += localStat.transferred
+ total += localStat.total
+ }
timeTaken := float64(int64(s.end.Sub(s.start))) / 1000000000
fmt.Printf("\nConcurrency Level: %d\n", *b.concurrency)
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)
+ fmt.Printf("Complete requests: %d\n", completed)
+ fmt.Printf("Failed requests: %d\n", failed)
+ fmt.Printf("Total transferred: %d bytes\n", transferred)
+ fmt.Printf("Requests per second: %.2f [#/sec]\n", float64(completed)/timeTaken)
+ fmt.Printf("Transfer rate: %.2f [Kbytes/sec]\n", float64(transferred)/1024/timeTaken)
n, sum := 0, 0
min, max := 10000000, 0
for i := 0; i < len(s.data); i++ {
@@ -496,15 +500,32 @@ func (l *FakeReader) Read(p []byte) (n int, err error) {
} 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))
+ if n >= 8 {
+ for i := 0; i < 8; i++ {
+ p[i] = byte(l.id >> uint(i*8))
}
}
l.size -= int64(n)
return
}
+func (l *FakeReader) WriteTo(w io.Writer) (n int64, err error) {
+ size := int(l.size)
+ bufferSize := len(sharedBytes)
+ for size > 0 {
+ tempBuffer := sharedBytes
+ if size < bufferSize {
+ tempBuffer = sharedBytes[0:size]
+ }
+ count, e := w.Write(tempBuffer)
+ if e != nil {
+ return int64(size), e
+ }
+ size -= count
+ }
+ return l.size, nil
+}
+
func Readln(r *bufio.Reader) ([]byte, error) {
var (
isPrefix bool = true
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-22 21:40:14 +0000