dstepanov-yadro/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#501] object/put: refactor distributed target

Browse source 
Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
Evgenii Stratonikov 2021-05-28 10:34:31 +03:00 committed by Leonard Lyubich
parent 6ca7f4511c
commit a422f42ca9
1 changed files with 24 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
pkg/services/object/put/distributed.go
View file

@ -45,13 +45,6 @@ func (t *distributedTarget) Write(p []byte) (n int, err error) {
} }
func (t *distributedTarget) Close() (*transformer.AccessIdentifiers, error) { func (t *distributedTarget) Close() (*transformer.AccessIdentifiers, error) {
traverser, err := placement.NewTraverser(
append(t.traverseOpts, placement.ForObject(t.obj.ID()))...,
)
if err != nil {
return nil, fmt.Errorf("(%T) could not create object placement traverser: %w", t, err)
}
sz := 0 sz := 0
for i := range t.chunks { for i := range t.chunks {
@ -70,6 +63,28 @@ func (t *distributedTarget) Close() (*transformer.AccessIdentifiers, error) {
return nil, fmt.Errorf("(%T) could not validate payload content: %w", t, err) return nil, fmt.Errorf("(%T) could not validate payload content: %w", t, err)
} }
return t.iteratePlacement(t.sendObject)
}
func (t *distributedTarget) sendObject(addr *network.Address) error {
target := t.nodeTargetInitializer(addr)
if err := target.WriteHeader(t.obj); err != nil {
return fmt.Errorf("could not write header: %w", err)
} else if _, err := target.Close(); err != nil {
return fmt.Errorf("could not close object stream: %w", err)
}
return nil
}
func (t *distributedTarget) iteratePlacement(f func(*network.Address) error) (*transformer.AccessIdentifiers, error) {
traverser, err := placement.NewTraverser(
append(t.traverseOpts, placement.ForObject(t.obj.ID()))...,
)
if err != nil {
return nil, fmt.Errorf("(%T) could not create object placement traverser: %w", t, err)
}
loop: loop:
for { for {
addrs := traverser.Next() addrs := traverser.Next()
@ -83,21 +98,11 @@ loop:
wg.Add(1) wg.Add(1)
addr := addrs[i] addr := addrs[i]
if err := t.workerPool.Submit(func() { if err := t.workerPool.Submit(func() {
defer wg.Done() defer wg.Done()
target := t.nodeTargetInitializer(addr) if err := f(addr); err != nil {
svcutil.LogServiceError(t.log, "PUT", addr, err)
if err := target.WriteHeader(t.obj); err != nil {
svcutil.LogServiceError(t.log, "PUT", addr,
fmt.Errorf("could not write header: %w", err))
return
} else if _, err := target.Close(); err != nil {
svcutil.LogServiceError(t.log, "PUT", addr,
fmt.Errorf("could not close object stream: %w", err))
return return
} }