benchmark.go 16 KB

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