frostfs-node/pkg/services/replicator/process.go
Alex Vanin 65be09d3db [#155] Update neofs-api-go with refactored pkg/netmap
Refactored pkg/netmap package provides JSON converters for
NodeInfo and PlacementPolicy structures, that has been used
by client applications.

It also updates Node structure itself so it is a part of
grpc <-> v2 <-> pkg conversion chain.

Signed-off-by: Alex Vanin <alexey@nspcc.ru>
2020-11-06 09:55:05 +03:00

95 lines
1.7 KiB
Go

package replicator
import (
"context"
"github.com/nspcc-dev/neofs-node/pkg/network"
putsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/put"
"go.uber.org/zap"
)
func (p *Replicator) Run(ctx context.Context) {
defer func() {
close(p.ch)
p.log.Info("routine stopped")
}()
p.ch = make(chan *Task, p.taskCap)
p.log.Info("process routine",
zap.Uint32("task queue capacity", p.taskCap),
zap.Duration("put timeout", p.putTimeout),
)
for {
select {
case <-ctx.Done():
p.log.Warn("context is done",
zap.String("error", ctx.Err().Error()),
)
return
case task, ok := <-p.ch:
if !ok {
p.log.Warn("trigger channel is closed")
return
}
p.handleTask(ctx, task)
}
}
}
func (p *Replicator) handleTask(ctx context.Context, task *Task) {
defer func() {
p.log.Info("finish work",
zap.Uint32("amount of unfinished replicas", task.quantity),
)
}()
obj, err := p.localStorage.Get(task.addr)
if err != nil {
p.log.Error("could not get object from local storage")
return
}
prm := new(putsvc.RemotePutPrm).
WithObject(obj)
for i := 0; task.quantity > 0 && i < len(task.nodes); i++ {
select {
case <-ctx.Done():
return
default:
}
netAddr := task.nodes[i].Address()
log := p.log.With(zap.String("node", netAddr))
node, err := network.AddressFromString(netAddr)
if err != nil {
log.Error("could not parse network address")
continue
}
callCtx, cancel := context.WithTimeout(ctx, p.putTimeout)
err = p.remoteSender.PutObject(callCtx, prm.WithNodeAddress(node))
cancel()
if err != nil {
log.Error("could not replicate object",
zap.String("error", err.Error()),
)
} else {
log.Info("object successfully replicated")
task.quantity--
}
}
}