TrueCloudLab/xk6-frostfs
42
1
Fork 20
Code Issues 33 Pull requests 1 Releases Activity Actions

Do not store all payload in memory #115

Merged
fyrchik merged 5 commits from fyrchik/xk6-frostfs:streaming_payload into master 2024-09-04 19:51:18 +00:00
Conversation 0 Commits 5 Files changed 22 +18 -15
5 changed files with 18 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 636a1e9290 - Show all commits

2
internal/datagen/generator.go
View file

@ -84,7 +84,7 @@ func (g *Generator) fillBuffer() {
} }
g.buf = b.Bytes() g.buf = b.Bytes()
default: default:
rand.Read(g.buf) // Per docs, err is always nil here g.rand.Read(g.buf) // Per docs, err is always nil here
} }
} }

6
internal/local/local.go
View file

@ -248,7 +248,7 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool, l Limiter) ([]
var shOpts [][]shard.Option var shOpts [][]shard.Option
engineconfig.IterateShards(c, false, func(sc *shardconfig.Config) error { err := engineconfig.IterateShards(c, false, func(sc *shardconfig.Config) error {
opts := []shard.Option{ opts := []shard.Option{
shard.WithRefillMetabase(sc.RefillMetabase()), shard.WithRefillMetabase(sc.RefillMetabase()),
shard.WithMode(sc.Mode()), shard.WithMode(sc.Mode()),
@ -375,7 +375,9 @@ func storageEngineOptionsFromConfig(c *config.Config, debug bool, l Limiter) ([]
return nil return nil
}) })
if err != nil {
return nil, nil, fmt.Errorf("iterate shards: %w", err)
}
return ngOpts, shOpts, nil return ngOpts, shOpts, nil
} }

14
internal/native/client.go
View file

@ -309,10 +309,9 @@ func (c *Client) PutContainer(params map[string]string) PutContainerResponse {
} }
start := time.Now() start := time.Now()
var prm client.PrmContainerPut res, err := c.cli.ContainerPut(c.vu.Context(), client.PrmContainerPut{
prm.SetContainer(cnr) Container: &cnr,
})
res, err := c.cli.ContainerPut(c.vu.Context(), prm)
if err != nil { if err != nil {
return c.putCnrErrorResponse(err) return c.putCnrErrorResponse(err)
} }
@ -500,10 +499,9 @@ func (x *waitParams) setDefaults() {
func (c *Client) waitForContainerPresence(ctx context.Context, cnrID cid.ID, wp *waitParams) error { func (c *Client) waitForContainerPresence(ctx context.Context, cnrID cid.ID, wp *waitParams) error {
return waitFor(ctx, wp, func(ctx context.Context) bool { return waitFor(ctx, wp, func(ctx context.Context) bool {
var prm client.PrmContainerGet _, err := c.cli.ContainerGet(ctx, client.PrmContainerGet{
prm.SetContainer(cnrID) ContainerID: &cnrID,
})
_, err := c.cli.ContainerGet(ctx, prm)
return err == nil return err == nil
}) })
} }

6
internal/native/native.go
View file

@ -89,9 +89,9 @@ func (n *Native) Connect(endpoint, hexPrivateKey string, dialTimeout, streamTime
// generate session token // generate session token
exp := uint64(math.MaxUint64) exp := uint64(math.MaxUint64)
var prmSessionCreate client.PrmSessionCreate sessionResp, err := cli.SessionCreate(n.vu.Context(), client.PrmSessionCreate{
prmSessionCreate.SetExp(exp) Expiration: exp,
sessionResp, err := cli.SessionCreate(n.vu.Context(), prmSessionCreate) })
if err != nil { if err != nil {
return nil, fmt.Errorf("dial endpoint: %s %w", endpoint, err) return nil, fmt.Errorf("dial endpoint: %s %w", endpoint, err)
} }

5
internal/s3local/local.go
View file

@ -151,7 +151,10 @@ func (s *Local) Connect(configFile string, configDir string, params map[string]s
} }
l := layer.NewLayer(zap.L(), &frostfs{rc}, cfg) l := layer.NewLayer(zap.L(), &frostfs{rc}, cfg)
l.Initialize(s.l.VU().Context(), nopEventListener{}) err = l.Initialize(s.l.VU().Context(), nopEventListener{})
if err != nil {
return nil, fmt.Errorf("initialize: %w", err)
}
return &Client{ return &Client{
vu: s.l.VU(), vu: s.l.VU(),