benchmark.go 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579
  1. package command
  2. import (
  3. "bufio"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/pb"
  6. "io"
  7. "math"
  8. "math/rand"
  9. "os"
  10. "runtime"
  11. "runtime/pprof"
  12. "sort"
  13. "sync"
  14. "time"
  15. "google.golang.org/grpc"
  16. "github.com/seaweedfs/seaweedfs/weed/glog"
  17. "github.com/seaweedfs/seaweedfs/weed/operation"
  18. "github.com/seaweedfs/seaweedfs/weed/security"
  19. "github.com/seaweedfs/seaweedfs/weed/util"
  20. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  21. )
  22. type BenchmarkOptions struct {
  23. masters *string
  24. concurrency *int
  25. numberOfFiles *int
  26. fileSize *int
  27. idListFile *string
  28. write *bool
  29. deletePercentage *int
  30. read *bool
  31. sequentialRead *bool
  32. collection *string
  33. replication *string
  34. diskType *string
  35. cpuprofile *string
  36. maxCpu *int
  37. grpcDialOption grpc.DialOption
  38. masterClient *wdclient.MasterClient
  39. fsync *bool
  40. }
  41. var (
  42. b BenchmarkOptions
  43. sharedBytes []byte
  44. isSecure bool
  45. )
  46. func init() {
  47. cmdBenchmark.Run = runBenchmark // break init cycle
  48. cmdBenchmark.IsDebug = cmdBenchmark.Flag.Bool("debug", false, "verbose debug information")
  49. b.masters = cmdBenchmark.Flag.String("master", "localhost:9333", "SeaweedFS master location")
  50. b.concurrency = cmdBenchmark.Flag.Int("c", 16, "number of concurrent write or read processes")
  51. b.fileSize = cmdBenchmark.Flag.Int("size", 1024, "simulated file size in bytes, with random(0~63) bytes padding")
  52. b.numberOfFiles = cmdBenchmark.Flag.Int("n", 1024*1024, "number of files to write for each thread")
  53. b.idListFile = cmdBenchmark.Flag.String("list", os.TempDir()+"/benchmark_list.txt", "list of uploaded file ids")
  54. b.write = cmdBenchmark.Flag.Bool("write", true, "enable write")
  55. b.deletePercentage = cmdBenchmark.Flag.Int("deletePercent", 0, "the percent of writes that are deletes")
  56. b.read = cmdBenchmark.Flag.Bool("read", true, "enable read")
  57. b.sequentialRead = cmdBenchmark.Flag.Bool("readSequentially", false, "randomly read by ids from \"-list\" specified file")
  58. b.collection = cmdBenchmark.Flag.String("collection", "benchmark", "write data to this collection")
  59. b.replication = cmdBenchmark.Flag.String("replication", "000", "replication type")
  60. b.diskType = cmdBenchmark.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
  61. b.cpuprofile = cmdBenchmark.Flag.String("cpuprofile", "", "cpu profile output file")
  62. b.maxCpu = cmdBenchmark.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
  63. b.fsync = cmdBenchmark.Flag.Bool("fsync", false, "flush data to disk after write")
  64. sharedBytes = make([]byte, 1024)
  65. }
  66. var cmdBenchmark = &Command{
  67. UsageLine: "benchmark -master=localhost:9333 -c=10 -n=100000",
  68. Short: "benchmark by writing millions of files and reading them out",
  69. Long: `benchmark on an empty SeaweedFS file system.
  70. Two tests during benchmark:
  71. 1) write lots of small files to the system
  72. 2) read the files out
  73. The file content is mostly zeros, but no compression is done.
  74. You can choose to only benchmark read or write.
  75. During write, the list of uploaded file ids is stored in "-list" specified file.
  76. You can also use your own list of file ids to run read test.
  77. Write speed and read speed will be collected.
  78. The numbers are used to get a sense of the system.
  79. Usually your network or the hard drive is the real bottleneck.
  80. Another thing to watch is whether the volumes are evenly distributed
  81. to each volume server. Because the 7 more benchmark volumes are randomly distributed
  82. to servers with free slots, it's highly possible some servers have uneven amount of
  83. benchmark volumes. To remedy this, you can use this to grow the benchmark volumes
  84. before starting the benchmark command:
  85. http://localhost:9333/vol/grow?collection=benchmark&count=5
  86. After benchmarking, you can clean up the written data by deleting the benchmark collection
  87. http://localhost:9333/col/delete?collection=benchmark
  88. `,
  89. }
  90. var (
  91. wait sync.WaitGroup
  92. writeStats *stats
  93. readStats *stats
  94. )
  95. func runBenchmark(cmd *Command, args []string) bool {
  96. util.LoadConfiguration("security", false)
  97. b.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
  98. fmt.Printf("This is SeaweedFS version %s %s %s\n", util.Version(), runtime.GOOS, runtime.GOARCH)
  99. if *b.maxCpu < 1 {
  100. *b.maxCpu = runtime.NumCPU()
  101. }
  102. runtime.GOMAXPROCS(*b.maxCpu)
  103. if *b.cpuprofile != "" {
  104. f, err := os.Create(*b.cpuprofile)
  105. if err != nil {
  106. glog.Fatal(err)
  107. }
  108. pprof.StartCPUProfile(f)
  109. defer pprof.StopCPUProfile()
  110. }
  111. b.masterClient = wdclient.NewMasterClient(b.grpcDialOption, "", "client", "", "", "", pb.ServerAddresses(*b.masters).ToAddressMap())
  112. go b.masterClient.KeepConnectedToMaster()
  113. b.masterClient.WaitUntilConnected()
  114. if *b.write {
  115. benchWrite()
  116. }
  117. if *b.read {
  118. benchRead()
  119. }
  120. return true
  121. }
  122. func benchWrite() {
  123. fileIdLineChan := make(chan string)
  124. finishChan := make(chan bool)
  125. writeStats = newStats(*b.concurrency)
  126. idChan := make(chan int)
  127. go writeFileIds(*b.idListFile, fileIdLineChan, finishChan)
  128. for i := 0; i < *b.concurrency; i++ {
  129. wait.Add(1)
  130. go writeFiles(idChan, fileIdLineChan, &writeStats.localStats[i])
  131. }
  132. writeStats.start = time.Now()
  133. writeStats.total = *b.numberOfFiles
  134. go writeStats.checkProgress("Writing Benchmark", finishChan)
  135. for i := 0; i < *b.numberOfFiles; i++ {
  136. idChan <- i
  137. }
  138. close(idChan)
  139. wait.Wait()
  140. writeStats.end = time.Now()
  141. wait.Add(2)
  142. finishChan <- true
  143. finishChan <- true
  144. wait.Wait()
  145. close(finishChan)
  146. writeStats.printStats()
  147. }
  148. func benchRead() {
  149. fileIdLineChan := make(chan string)
  150. finishChan := make(chan bool)
  151. readStats = newStats(*b.concurrency)
  152. go readFileIds(*b.idListFile, fileIdLineChan)
  153. readStats.start = time.Now()
  154. readStats.total = *b.numberOfFiles
  155. go readStats.checkProgress("Randomly Reading Benchmark", finishChan)
  156. for i := 0; i < *b.concurrency; i++ {
  157. wait.Add(1)
  158. go readFiles(fileIdLineChan, &readStats.localStats[i])
  159. }
  160. wait.Wait()
  161. wait.Add(1)
  162. finishChan <- true
  163. wait.Wait()
  164. close(finishChan)
  165. readStats.end = time.Now()
  166. readStats.printStats()
  167. }
  168. type delayedFile struct {
  169. enterTime time.Time
  170. fp *operation.FilePart
  171. }
  172. func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
  173. defer wait.Done()
  174. delayedDeleteChan := make(chan *delayedFile, 100)
  175. var waitForDeletions sync.WaitGroup
  176. for i := 0; i < 7; i++ {
  177. waitForDeletions.Add(1)
  178. go func() {
  179. defer waitForDeletions.Done()
  180. for df := range delayedDeleteChan {
  181. if df.enterTime.After(time.Now()) {
  182. time.Sleep(df.enterTime.Sub(time.Now()))
  183. }
  184. var jwtAuthorization security.EncodedJwt
  185. if isSecure {
  186. jwtAuthorization = operation.LookupJwt(b.masterClient.GetMaster(), b.grpcDialOption, df.fp.Fid)
  187. }
  188. if e := util.Delete(fmt.Sprintf("http://%s/%s", df.fp.Server, df.fp.Fid), string(jwtAuthorization)); e == nil {
  189. s.completed++
  190. } else {
  191. s.failed++
  192. }
  193. }
  194. }()
  195. }
  196. random := rand.New(rand.NewSource(time.Now().UnixNano()))
  197. for id := range idChan {
  198. start := time.Now()
  199. fileSize := int64(*b.fileSize + random.Intn(64))
  200. fp := &operation.FilePart{
  201. Reader: &FakeReader{id: uint64(id), size: fileSize, random: random},
  202. FileSize: fileSize,
  203. MimeType: "image/bench", // prevent gzip benchmark content
  204. Fsync: *b.fsync,
  205. }
  206. ar := &operation.VolumeAssignRequest{
  207. Count: 1,
  208. Collection: *b.collection,
  209. Replication: *b.replication,
  210. DiskType: *b.diskType,
  211. }
  212. if assignResult, err := operation.Assign(b.masterClient.GetMaster, b.grpcDialOption, ar); err == nil {
  213. fp.Server, fp.Fid, fp.Collection = assignResult.Url, assignResult.Fid, *b.collection
  214. if !isSecure && assignResult.Auth != "" {
  215. isSecure = true
  216. }
  217. if _, err := fp.Upload(0, b.masterClient.GetMaster, false, assignResult.Auth, b.grpcDialOption); err == nil {
  218. if random.Intn(100) < *b.deletePercentage {
  219. s.total++
  220. delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}
  221. } else {
  222. fileIdLineChan <- fp.Fid
  223. }
  224. s.completed++
  225. s.transferred += fileSize
  226. } else {
  227. s.failed++
  228. fmt.Printf("Failed to write with error:%v\n", err)
  229. }
  230. writeStats.addSample(time.Now().Sub(start))
  231. if *cmdBenchmark.IsDebug {
  232. fmt.Printf("writing %d file %s\n", id, fp.Fid)
  233. }
  234. } else {
  235. s.failed++
  236. println("writing file error:", err.Error())
  237. }
  238. }
  239. close(delayedDeleteChan)
  240. waitForDeletions.Wait()
  241. }
  242. func readFiles(fileIdLineChan chan string, s *stat) {
  243. defer wait.Done()
  244. for fid := range fileIdLineChan {
  245. if len(fid) == 0 {
  246. continue
  247. }
  248. if fid[0] == '#' {
  249. continue
  250. }
  251. if *cmdBenchmark.IsDebug {
  252. fmt.Printf("reading file %s\n", fid)
  253. }
  254. start := time.Now()
  255. var bytesRead int
  256. var err error
  257. urls, err := b.masterClient.LookupFileId(fid)
  258. if err != nil {
  259. s.failed++
  260. println("!!!! ", fid, " location not found!!!!!")
  261. continue
  262. }
  263. var bytes []byte
  264. for _, url := range urls {
  265. bytes, _, err = util.Get(url)
  266. if err == nil {
  267. break
  268. }
  269. }
  270. bytesRead = len(bytes)
  271. if err == nil {
  272. s.completed++
  273. s.transferred += int64(bytesRead)
  274. readStats.addSample(time.Now().Sub(start))
  275. } else {
  276. s.failed++
  277. fmt.Printf("Failed to read %s error:%v\n", fid, err)
  278. }
  279. }
  280. }
  281. func writeFileIds(fileName string, fileIdLineChan chan string, finishChan chan bool) {
  282. file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  283. if err != nil {
  284. glog.Fatalf("File to create file %s: %s\n", fileName, err)
  285. }
  286. defer file.Close()
  287. for {
  288. select {
  289. case <-finishChan:
  290. wait.Done()
  291. return
  292. case line := <-fileIdLineChan:
  293. file.Write([]byte(line))
  294. file.Write([]byte("\n"))
  295. }
  296. }
  297. }
  298. func readFileIds(fileName string, fileIdLineChan chan string) {
  299. file, err := os.Open(fileName) // For read access.
  300. if err != nil {
  301. glog.Fatalf("File to read file %s: %s\n", fileName, err)
  302. }
  303. defer file.Close()
  304. random := rand.New(rand.NewSource(time.Now().UnixNano()))
  305. r := bufio.NewReader(file)
  306. if *b.sequentialRead {
  307. for {
  308. if line, err := Readln(r); err == nil {
  309. fileIdLineChan <- string(line)
  310. } else {
  311. break
  312. }
  313. }
  314. } else {
  315. lines := make([]string, 0, readStats.total)
  316. for {
  317. if line, err := Readln(r); err == nil {
  318. lines = append(lines, string(line))
  319. } else {
  320. break
  321. }
  322. }
  323. if len(lines) > 0 {
  324. for i := 0; i < readStats.total; i++ {
  325. fileIdLineChan <- lines[random.Intn(len(lines))]
  326. }
  327. }
  328. }
  329. close(fileIdLineChan)
  330. }
  331. const (
  332. benchResolution = 10000 // 0.1 microsecond
  333. benchBucket = 1000000000 / benchResolution
  334. )
  335. // An efficient statics collecting and rendering
  336. type stats struct {
  337. data []int
  338. overflow []int
  339. localStats []stat
  340. start time.Time
  341. end time.Time
  342. total int
  343. }
  344. type stat struct {
  345. completed int
  346. failed int
  347. total int
  348. transferred int64
  349. }
  350. var percentages = []int{50, 66, 75, 80, 90, 95, 98, 99, 100}
  351. func newStats(n int) *stats {
  352. return &stats{
  353. data: make([]int, benchResolution),
  354. overflow: make([]int, 0),
  355. localStats: make([]stat, n),
  356. }
  357. }
  358. func (s *stats) addSample(d time.Duration) {
  359. index := int(d / benchBucket)
  360. if index < 0 {
  361. fmt.Printf("This request takes %3.1f seconds, skipping!\n", float64(index)/10000)
  362. } else if index < len(s.data) {
  363. s.data[int(d/benchBucket)]++
  364. } else {
  365. s.overflow = append(s.overflow, index)
  366. }
  367. }
  368. func (s *stats) checkProgress(testName string, finishChan chan bool) {
  369. fmt.Printf("\n------------ %s ----------\n", testName)
  370. ticker := time.Tick(time.Second)
  371. lastCompleted, lastTransferred, lastTime := 0, int64(0), time.Now()
  372. for {
  373. select {
  374. case <-finishChan:
  375. wait.Done()
  376. return
  377. case t := <-ticker:
  378. completed, transferred, taken, total := 0, int64(0), t.Sub(lastTime), s.total
  379. for _, localStat := range s.localStats {
  380. completed += localStat.completed
  381. transferred += localStat.transferred
  382. total += localStat.total
  383. }
  384. fmt.Printf("Completed %d of %d requests, %3.1f%% %3.1f/s %3.1fMB/s\n",
  385. completed, total, float64(completed)*100/float64(total),
  386. float64(completed-lastCompleted)*float64(int64(time.Second))/float64(int64(taken)),
  387. float64(transferred-lastTransferred)*float64(int64(time.Second))/float64(int64(taken))/float64(1024*1024),
  388. )
  389. lastCompleted, lastTransferred, lastTime = completed, transferred, t
  390. }
  391. }
  392. }
  393. func (s *stats) printStats() {
  394. completed, failed, transferred, total := 0, 0, int64(0), s.total
  395. for _, localStat := range s.localStats {
  396. completed += localStat.completed
  397. failed += localStat.failed
  398. transferred += localStat.transferred
  399. total += localStat.total
  400. }
  401. timeTaken := float64(int64(s.end.Sub(s.start))) / 1000000000
  402. fmt.Printf("\nConcurrency Level: %d\n", *b.concurrency)
  403. fmt.Printf("Time taken for tests: %.3f seconds\n", timeTaken)
  404. fmt.Printf("Completed requests: %d\n", completed)
  405. fmt.Printf("Failed requests: %d\n", failed)
  406. fmt.Printf("Total transferred: %d bytes\n", transferred)
  407. fmt.Printf("Requests per second: %.2f [#/sec]\n", float64(completed)/timeTaken)
  408. fmt.Printf("Transfer rate: %.2f [Kbytes/sec]\n", float64(transferred)/1024/timeTaken)
  409. n, sum := 0, 0
  410. min, max := 10000000, 0
  411. for i := 0; i < len(s.data); i++ {
  412. n += s.data[i]
  413. sum += s.data[i] * i
  414. if s.data[i] > 0 {
  415. if min > i {
  416. min = i
  417. }
  418. if max < i {
  419. max = i
  420. }
  421. }
  422. }
  423. n += len(s.overflow)
  424. for i := 0; i < len(s.overflow); i++ {
  425. sum += s.overflow[i]
  426. if min > s.overflow[i] {
  427. min = s.overflow[i]
  428. }
  429. if max < s.overflow[i] {
  430. max = s.overflow[i]
  431. }
  432. }
  433. avg := float64(sum) / float64(n)
  434. varianceSum := 0.0
  435. for i := 0; i < len(s.data); i++ {
  436. if s.data[i] > 0 {
  437. d := float64(i) - avg
  438. varianceSum += d * d * float64(s.data[i])
  439. }
  440. }
  441. for i := 0; i < len(s.overflow); i++ {
  442. d := float64(s.overflow[i]) - avg
  443. varianceSum += d * d
  444. }
  445. std := math.Sqrt(varianceSum / float64(n))
  446. fmt.Printf("\nConnection Times (ms)\n")
  447. fmt.Printf(" min avg max std\n")
  448. fmt.Printf("Total: %2.1f %3.1f %3.1f %3.1f\n", float32(min)/10, float32(avg)/10, float32(max)/10, std/10)
  449. // printing percentiles
  450. fmt.Printf("\nPercentage of the requests served within a certain time (ms)\n")
  451. percentiles := make([]int, len(percentages))
  452. for i := 0; i < len(percentages); i++ {
  453. percentiles[i] = n * percentages[i] / 100
  454. }
  455. percentiles[len(percentiles)-1] = n
  456. percentileIndex := 0
  457. currentSum := 0
  458. for i := 0; i < len(s.data); i++ {
  459. currentSum += s.data[i]
  460. if s.data[i] > 0 && percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
  461. fmt.Printf(" %3d%% %5.1f ms\n", percentages[percentileIndex], float32(i)/10.0)
  462. percentileIndex++
  463. for percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
  464. percentileIndex++
  465. }
  466. }
  467. }
  468. sort.Ints(s.overflow)
  469. for i := 0; i < len(s.overflow); i++ {
  470. currentSum++
  471. if percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
  472. fmt.Printf(" %3d%% %5.1f ms\n", percentages[percentileIndex], float32(s.overflow[i])/10.0)
  473. percentileIndex++
  474. for percentileIndex < len(percentiles) && currentSum >= percentiles[percentileIndex] {
  475. percentileIndex++
  476. }
  477. }
  478. }
  479. }
  480. // a fake reader to generate content to upload
  481. type FakeReader struct {
  482. id uint64 // an id number
  483. size int64 // max bytes
  484. random *rand.Rand
  485. }
  486. func (l *FakeReader) Read(p []byte) (n int, err error) {
  487. if l.size <= 0 {
  488. return 0, io.EOF
  489. }
  490. if int64(len(p)) > l.size {
  491. n = int(l.size)
  492. } else {
  493. n = len(p)
  494. }
  495. if n >= 8 {
  496. for i := 0; i < 8; i++ {
  497. p[i] = byte(l.id >> uint(i*8))
  498. }
  499. l.random.Read(p[8:])
  500. }
  501. l.size -= int64(n)
  502. return
  503. }
  504. func (l *FakeReader) WriteTo(w io.Writer) (n int64, err error) {
  505. size := int(l.size)
  506. bufferSize := len(sharedBytes)
  507. for size > 0 {
  508. tempBuffer := sharedBytes
  509. if size < bufferSize {
  510. tempBuffer = sharedBytes[0:size]
  511. }
  512. count, e := w.Write(tempBuffer)
  513. if e != nil {
  514. return int64(size), e
  515. }
  516. size -= count
  517. }
  518. return l.size, nil
  519. }
  520. func Readln(r *bufio.Reader) ([]byte, error) {
  521. var (
  522. isPrefix = true
  523. err error
  524. line, ln []byte
  525. )
  526. for isPrefix && err == nil {
  527. line, isPrefix, err = r.ReadLine()
  528. ln = append(ln, line...)
  529. }
  530. return ln, err
  531. }