forked from TrueCloudLab/frostfs-node
[#460] services/object: Reduce distibutedTarget
memory footprint
Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
parent
4d48377cec
commit
785d81a68a
2 changed files with 12 additions and 14 deletions
|
@ -25,8 +25,6 @@ type preparedObjectTarget interface {
|
|||
type distributedTarget struct {
|
||||
traversal traversal
|
||||
|
||||
remotePool, localPool util.WorkerPool
|
||||
|
||||
obj *objectSDK.Object
|
||||
objMeta object.ContentMeta
|
||||
|
||||
|
@ -34,7 +32,7 @@ type distributedTarget struct {
|
|||
|
||||
nodeTargetInitializer func(nodeDesc) preparedObjectTarget
|
||||
|
||||
isLocalKey func([]byte) bool
|
||||
getWorkerPool func([]byte) (util.WorkerPool, bool)
|
||||
|
||||
relay func(context.Context, nodeDesc) error
|
||||
|
||||
|
@ -223,13 +221,8 @@ func (t *distributedTarget) iterateAddresses(ctx context.Context, traverser *pla
|
|||
wg.Add(1)
|
||||
|
||||
addr := addrs[i]
|
||||
isLocal := t.isLocalKey(addr.PublicKey())
|
||||
|
||||
workerPool := t.remotePool
|
||||
if isLocal {
|
||||
workerPool = t.localPool
|
||||
}
|
||||
|
||||
workerPool, isLocal := t.getWorkerPool(addr.PublicKey())
|
||||
if err := workerPool.Submit(func() {
|
||||
defer wg.Done()
|
||||
|
||||
|
|
|
@ -10,6 +10,7 @@ import (
|
|||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
||||
pkgutil "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
|
||||
containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/transformer"
|
||||
|
@ -219,9 +220,8 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
|
|||
|
||||
extraBroadcastEnabled: withBroadcast,
|
||||
},
|
||||
payload: getPayload(),
|
||||
remotePool: p.remotePool,
|
||||
localPool: p.localPool,
|
||||
payload: getPayload(),
|
||||
getWorkerPool: p.getWorkerPool,
|
||||
nodeTargetInitializer: func(node nodeDesc) preparedObjectTarget {
|
||||
if node.local {
|
||||
return &localTarget{
|
||||
|
@ -242,8 +242,6 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
|
|||
relay: relay,
|
||||
fmt: p.fmtValidator,
|
||||
log: p.log,
|
||||
|
||||
isLocalKey: p.netmapKeys.IsLocalKey,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -280,3 +278,10 @@ func (p *Streamer) Close(ctx context.Context) (*PutResponse, error) {
|
|||
id: ids.SelfID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *Streamer) getWorkerPool(pub []byte) (pkgutil.WorkerPool, bool) {
|
||||
if p.netmapKeys.IsLocalKey(pub) {
|
||||
return p.localPool, true
|
||||
}
|
||||
return p.remotePool, false
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue