realloc/frostfs-s3-gw
1
0
Fork 0
forked from TrueCloudLab/frostfs-s3-gw
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#165] Add list session cache

Browse source 
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
Denis Kirillov 2023-10-16 09:27:21 +03:00
parent 29ac91dfd5
commit 6e8960b2ab
13 changed files with 196 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

41
api/layer/object.go
View file

@ -17,7 +17,6 @@ import (
"strconv"
"strings"
"sync"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/auth"
@ -652,41 +651,39 @@ func (n *layer) getLatestObjectsVersionsV2(ctx context.Context, p allObjectParam
return nil, nil, nil
}
testKey := p.Prefix + p.Delimiter + p.ContinuationToken
nodeVersionsStreamValue, ok := n.cache.testCache[testKey]
if ok {
delete(n.cache.testCache, testKey)
owner := n.BearerOwner(ctx)
cacheKey := cache.CreateListSessionCacheKey(p.Bucket.CID, p.Prefix, p.Delimiter)
session := n.cache.GetListSession(owner, cacheKey)
if session != nil {
// after reading next object from stream in session
// the current cache value already doesn't match with next token in cache key
n.cache.DeleteListSession(owner, cacheKey)
} else {
ctx2, cancel2 := context.WithCancel(context.Background())
go func() {
<-time.After(10 * time.Second)
cancel2()
}()
session = &data.ListSession{NamesMap: make(map[string]struct{})}
session.Context, session.Cancel = context.WithCancel(context.Background())
if bd, err := middleware.GetBoxData(ctx); err == nil {
ctx2 = middleware.SetBoxData(ctx2, bd)
session.Context = middleware.SetBoxData(session.Context, bd)
}
nodeVersionsStreamValue.Stream, err = n.treeService.GetLatestVersionsByPrefixStream(ctx2, p.Bucket, p.Prefix)
session.Stream, err = n.treeService.GetLatestVersionsByPrefixStream(session.Context, p.Bucket, p.Prefix)
if err != nil {
return nil, nil, err
}
nodeVersionsStreamValue.NamesMap = map[string]struct{}{}
}
poolCtx, cancel := context.WithCancel(ctx)
defer cancel()
generator, errorCh := nodesGeneratorStream(poolCtx, p, nodeVersionsStreamValue)
generator, errorCh := nodesGeneratorStream(poolCtx, p, session)
objOutCh, err := n.initWorkerPoolStream(poolCtx, 2, p, generator)
if err != nil {
return nil, nil, fmt.Errorf("failed to init worker pool: %w", err)
}
objects = make([]*data.ObjectInfo, 0, p.MaxKeys+1)
if nodeVersionsStreamValue.Next != nil {
objects = append(objects, nodeVersionsStreamValue.Next)
if session.Next != nil {
objects = append(objects, session.Next)
}
for obj := range objOutCh {
@ -694,9 +691,7 @@ func (n *layer) getLatestObjectsVersionsV2(ctx context.Context, p allObjectParam
}
if err = <-errorCh; err != nil {
fmt.Println(len(objects))
fmt.Println(objects[len(objects)-1].Name)
return nil, nil, fmt.Errorf("failed to get object from tree: %w", err)
return nil, nil, fmt.Errorf("failed to get next object from stream: %w", err)
}
sort.Slice(objects, func(i, j int) bool {
@ -709,8 +704,8 @@ func (n *layer) getLatestObjectsVersionsV2(ctx context.Context, p allObjectParam
}
if next != nil {
nodeVersionsStreamValue.Next = next
n.cache.testCache[p.Prefix+p.Delimiter+next.VersionID()] = nodeVersionsStreamValue
session.Next = next
n.cache.PutListSession(owner, cache.CreateListSessionCacheKey(p.Bucket.CID, p.Prefix, next.VersionID()), session)
}
return
@ -772,7 +767,7 @@ func nodesGeneratorVersions(ctx context.Context, p allObjectParams, nodeVersions
return nodeCh
}
func nodesGeneratorStream(ctx context.Context, p allObjectParams, stream TestCacheValue) (<-chan *data.NodeVersion, <-chan error) {
func nodesGeneratorStream(ctx context.Context, p allObjectParams, stream *data.ListSession) (<-chan *data.NodeVersion, <-chan error) {
nodeCh := make(chan *data.NodeVersion)
errCh := make(chan error, 1)
//existed := make(map[string]struct{}, p.MaxKeys) // to squash the same directories