123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136 |
- package main
- import (
- "bytes"
- "flag"
- "fmt"
- "io"
- "io/ioutil"
- "log"
- "math/rand"
- "mime/multipart"
- "net/http"
- "os"
- "strings"
- "sync"
- "time"
- )
- var (
- size = flag.Int("size", 1024, "file size")
- concurrency = flag.Int("c", 4, "concurrent number of uploads")
- times = flag.Int("n", 1024, "repeated number of times")
- fileCount = flag.Int("fileCount", 1, "number of files to write")
- destination = flag.String("to", "http://localhost:8888/", "destination directory on filer")
- statsChan = make(chan stat, 8)
- )
- type stat struct {
- size int64
- }
- func main() {
- flag.Parse()
- data := make([]byte, *size)
- println("data len", len(data))
- var wg sync.WaitGroup
- for x := 0; x < *concurrency; x++ {
- wg.Add(1)
- go func(x int) {
- defer wg.Done()
- client := &http.Client{Transport: &http.Transport{
- MaxConnsPerHost: 1024,
- MaxIdleConnsPerHost: 1024,
- }}
- r := rand.New(rand.NewSource(time.Now().UnixNano() + int64(x)))
- for t := 0; t < *times; t++ {
- for f := 0; f < *fileCount; f++ {
- fn := r.Intn(*fileCount)
- if size, err := uploadFileToFiler(client, data, fmt.Sprintf("file%04d", fn), *destination); err == nil {
- statsChan <- stat{
- size: size,
- }
- } else {
- log.Fatalf("client %d upload %d times: %v", x, t, err)
- }
- }
- }
- }(x)
- }
- go func() {
- ticker := time.NewTicker(1000 * time.Millisecond)
- var lastTime time.Time
- var counter, size int64
- for {
- select {
- case stat := <-statsChan:
- size += stat.size
- counter++
- case x := <-ticker.C:
- if !lastTime.IsZero() {
- elapsed := x.Sub(lastTime).Seconds()
- fmt.Fprintf(os.Stdout, "%.2f files/s, %.2f MB/s\n",
- float64(counter)/elapsed,
- float64(size/1024/1024)/elapsed)
- }
- lastTime = x
- size = 0
- counter = 0
- }
- }
- }()
- wg.Wait()
- }
- func uploadFileToFiler(client *http.Client, data []byte, filename, destination string) (size int64, err error) {
- if !strings.HasSuffix(destination, "/") {
- destination = destination + "/"
- }
- body := &bytes.Buffer{}
- writer := multipart.NewWriter(body)
- part, err := writer.CreateFormFile("file", filename)
- if err != nil {
- return 0, fmt.Errorf("fail to create form %v: %v", filename, err)
- }
- part.Write(data)
- err = writer.Close()
- if err != nil {
- return 0, fmt.Errorf("fail to write part %v: %v", filename, err)
- }
- uri := destination + filename
- request, err := http.NewRequest("POST", uri, body)
- request.Header.Set("Content-Type", writer.FormDataContentType())
- // request.Close = true // can not use this, which do not reuse http connection, impacting filer->volume also.
- resp, err := client.Do(request)
- if err != nil {
- return 0, fmt.Errorf("http POST %s: %v", uri, err)
- } else {
- body := &bytes.Buffer{}
- _, err := body.ReadFrom(resp.Body)
- if err != nil {
- return 0, fmt.Errorf("read http POST %s response: %v", uri, err)
- }
- io.Copy(ioutil.Discard, resp.Body)
- resp.Body.Close()
- }
- return int64(len(data)), nil
- }
|