forked from seaweedfs/seaweedfs
/
bench_filer_upload.go
136 lines (112 loc) · 2.92 KB
/
bench_filer_upload.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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
}