[#845] object/put: Make distributed target to define is node is local

In previous implementation `distributedTarget` didn't check if next node is
local. This check was performed by the handlers (target initializer and
relay func).

Make `distributedTarget` to calculate node's locality. Pass locality flag to
the handlers.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2021-09-24 13:28:58 +03:00 committed by Alex Vanin
parent c96c455125
commit 3b2b6007c6
2 changed files with 26 additions and 23 deletions

View file

@ -1,7 +1,6 @@
package putsvc package putsvc
import ( import (
"errors"
"fmt" "fmt"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -23,15 +22,23 @@ type distributedTarget struct {
chunks [][]byte chunks [][]byte
nodeTargetInitializer func(placement.Node) transformer.ObjectTarget nodeTargetInitializer func(nodeDesc) transformer.ObjectTarget
relay func(placement.Node) error isLocalKey func([]byte) bool
relay func(nodeDesc) error
fmt *object.FormatValidator fmt *object.FormatValidator
log *logger.Logger log *logger.Logger
} }
type nodeDesc struct {
local bool
info placement.Node
}
// errIncompletePut is returned if processing on a container fails. // errIncompletePut is returned if processing on a container fails.
type errIncompletePut struct { type errIncompletePut struct {
singleErr error // error from the last responding node singleErr error // error from the last responding node
@ -81,12 +88,9 @@ func (t *distributedTarget) Close() (*transformer.AccessIdentifiers, error) {
return t.iteratePlacement(t.sendObject) return t.iteratePlacement(t.sendObject)
} }
func (t *distributedTarget) sendObject(node placement.Node) error { func (t *distributedTarget) sendObject(node nodeDesc) error {
if t.relay != nil { if !node.local && t.relay != nil {
err := t.relay(node) return t.relay(node)
if err == nil || !errors.Is(err, errLocalAddress) {
return err
}
} }
target := t.nodeTargetInitializer(node) target := t.nodeTargetInitializer(node)
@ -99,7 +103,7 @@ func (t *distributedTarget) sendObject(node placement.Node) error {
return nil return nil
} }
func (t *distributedTarget) iteratePlacement(f func(placement.Node) error) (*transformer.AccessIdentifiers, error) { func (t *distributedTarget) iteratePlacement(f func(nodeDesc) error) (*transformer.AccessIdentifiers, error) {
traverser, err := placement.NewTraverser( traverser, err := placement.NewTraverser(
append(t.traverseOpts, placement.ForObject(t.obj.ID()))..., append(t.traverseOpts, placement.ForObject(t.obj.ID()))...,
) )
@ -122,10 +126,13 @@ loop:
wg.Add(1) wg.Add(1)
addr := addrs[i] addr := addrs[i]
isLocal := t.isLocalKey(addr.Key())
if err := t.workerPool.Submit(func() { if err := t.workerPool.Submit(func() {
defer wg.Done() defer wg.Done()
if err := f(addr); err != nil { if err := f(nodeDesc{local: isLocal, info: addr}); err != nil {
resErr.Store(err) resErr.Store(err)
svcutil.LogServiceError(t.log, "PUT", addr.Addresses(), err) svcutil.LogServiceError(t.log, "PUT", addr.Addresses(), err)
return return

View file

@ -144,17 +144,11 @@ func (p *Streamer) preparePrm(prm *PutInitPrm) error {
return nil return nil
} }
var errLocalAddress = errors.New("can't relay to local address")
func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget { func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
var relay func(placement.Node) error var relay func(nodeDesc) error
if p.relay != nil { if p.relay != nil {
relay = func(node placement.Node) error { relay = func(node nodeDesc) error {
addr := node.Addresses() addr := node.info.Addresses()
if p.netmapKeys.IsLocalKey(node.Key()) {
return errLocalAddress
}
c, err := p.clientConstructor.Get(addr) c, err := p.clientConstructor.Get(addr)
if err != nil { if err != nil {
@ -168,8 +162,8 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
return &distributedTarget{ return &distributedTarget{
traverseOpts: prm.traverseOpts, traverseOpts: prm.traverseOpts,
workerPool: p.workerPool, workerPool: p.workerPool,
nodeTargetInitializer: func(node placement.Node) transformer.ObjectTarget { nodeTargetInitializer: func(node nodeDesc) transformer.ObjectTarget {
if p.netmapKeys.IsLocalKey(node.Key()) { if node.local {
return &localTarget{ return &localTarget{
storage: p.localStore, storage: p.localStore,
} }
@ -179,13 +173,15 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
ctx: p.ctx, ctx: p.ctx,
keyStorage: p.keyStorage, keyStorage: p.keyStorage,
commonPrm: prm.common, commonPrm: prm.common,
addr: node.Addresses(), addr: node.info.Addresses(),
clientConstructor: p.clientConstructor, clientConstructor: p.clientConstructor,
} }
}, },
relay: relay, relay: relay,
fmt: p.fmtValidator, fmt: p.fmtValidator,
log: p.log, log: p.log,
isLocalKey: p.netmapKeys.IsLocalKey,
} }
} }