TrueCloudLab/xk6-frostfs
40
0
Fork 19
Code Issues 35 Pull requests 1 Releases Activity Actions

Allow to pass custom gRPC chunk size for object put #79

Merged
fyrchik merged 1 commit from dstepanov-yadro/xk6-frostfs:feat/put-chunk-size into master 2023-07-26 21:08:04 +00:00
Conversation 0 Commits 1 Files changed 2 +24 -36
2 changed files with 24 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
internal/native/client.go
View file

@ -35,7 +35,6 @@ type (
key ecdsa.PrivateKey
tok session.Object
cli *client.Client
bufsize int
}
PutResponse struct {
@ -69,7 +68,6 @@ type (
vu modules.VU
key ecdsa.PrivateKey
cli *client.Client
bufsize int
hdr object.Object
payload []byte
@ -78,17 +76,6 @@ type (
const defaultBufferSize = 64 * 1024
func (c *Client) SetBufferSize(size int) {
if size < 0 {
panic("buffer size must be positive")
}
if size == 0 {
c.bufsize = defaultBufferSize
} else {
c.bufsize = size
}
}
func (c *Client) Put(containerID string, headers map[string]string, payload goja.ArrayBuffer, chunkSize int) PutResponse {
cliContainerID := parseContainerID(containerID)
@ -116,7 +103,7 @@ func (c *Client) Put(containerID string, headers map[string]string, payload goja
o.SetOwnerID(&owner)
o.SetAttributes(attrs...)
resp, err := put(c.vu, c.bufsize, c.cli, &tok, &o, payload.Bytes(), chunkSize)
resp, err := put(c.vu, c.cli, &tok, &o, payload.Bytes(), chunkSize)
if err != nil {
return PutResponse{Success: false, Error: err.Error()}
}
@ -176,7 +163,7 @@ func (c *Client) Get(containerID, objectID string) GetResponse {
prm.WithinSession(tok)
var objSize = 0
err = get(c.cli, prm, c.vu.Context(), c.bufsize, func(data []byte) {
err = get(c.cli, prm, c.vu.Context(), func(data []byte) {
objSize += len(data)
})
if err != nil {
@ -194,10 +181,9 @@ func get(
cli *client.Client,
prm client.PrmObjectGet,
ctx context.Context,
bufSize int,
onDataChunk func(chunk []byte),
) error {
var buf = make([]byte, bufSize)
var buf = make([]byte, defaultBufferSize)
objectReader, err := cli.ObjectGetInit(ctx, prm)
if err != nil {
@ -245,7 +231,7 @@ func (c *Client) VerifyHash(containerID, objectID, expectedHash string) VerifyHa
prm.WithinSession(tok)
hasher := sha256.New()
err = get(c.cli, prm, c.vu.Context(), c.bufsize, func(data []byte) {
err = get(c.cli, prm, c.vu.Context(), func(data []byte) {
hasher.Write(data)
})
if err != nil {
@ -384,7 +370,6 @@ func (c *Client) Onsite(containerID string, payload goja.ArrayBuffer) PreparedOb
vu: c.vu,
key: c.key,
cli: c.cli,
bufsize: c.bufsize,
hdr: *obj,
payload: data,
@ -413,7 +398,7 @@ func (p PreparedObject) Put(headers map[string]string) PutResponse {
return PutResponse{Success: false, Error: err.Error()}
}
_, err = put(p.vu, p.bufsize, p.cli, nil, &obj, p.payload, 0)
_, err = put(p.vu, p.cli, nil, &obj, p.payload, 0)
if err != nil {
return PutResponse{Success: false, Error: err.Error()}
}
@ -421,8 +406,12 @@ func (p PreparedObject) Put(headers map[string]string) PutResponse {
return PutResponse{Success: true, ObjectID: id.String()}
}
func put(vu modules.VU, bufSize int, cli *client.Client, tok *session.Object,
func put(vu modules.VU, cli *client.Client, tok *session.Object,
hdr *object.Object, payload []byte, chunkSize int) (*client.ResObjectPut, error) {
bufSize := defaultBufferSize
if chunkSize > 0 {
bufSize = chunkSize
}
buf := make([]byte, bufSize)
rdr := bytes.NewReader(payload)
sz := rdr.Size()

1
internal/native/native.go
View file

@ -137,6 +137,5 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
key: pk.PrivateKey,
tok: tok,
cli: &cli,
bufsize: defaultBufferSize,
}, nil
}