package replicator import ( "context" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine" putsvc "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/put" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" "go.uber.org/zap" ) // TaskResult is a replication result interface. type TaskResult interface { // SubmitSuccessfulReplication submits the successful object replication // to the given node. SubmitSuccessfulReplication(netmap.NodeInfo) } // HandleTask executes replication task inside invoking goroutine. // Passes all the nodes that accepted the replication to the TaskResult. func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult) { p.metrics.IncInFlightRequest() defer p.metrics.DecInFlightRequest() defer func() { p.log.Debug(logs.ReplicatorFinishWork, zap.Uint32("amount of unfinished replicas", task.quantity), ) }() if task.obj == nil { var err error task.obj, err = engine.Get(ctx, p.localStorage, task.addr) if err != nil { p.log.Error(logs.ReplicatorCouldNotGetObjectFromLocalStorage, zap.Stringer("object", task.addr), zap.Error(err)) return } } prm := new(putsvc.RemotePutPrm). WithObject(task.obj) for i := 0; task.quantity > 0 && i < len(task.nodes); i++ { select { case <-ctx.Done(): return default: } log := p.log.With( zap.String("node", netmap.StringifyPublicKey(task.nodes[i])), zap.Stringer("object", task.addr), ) callCtx, cancel := context.WithTimeout(ctx, p.putTimeout) err := p.remoteSender.PutObject(callCtx, prm.WithNodeInfo(task.nodes[i])) cancel() if err != nil { log.Error(logs.ReplicatorCouldNotReplicateObject, zap.String("error", err.Error()), ) } else { log.Debug(logs.ReplicatorObjectSuccessfullyReplicated) task.quantity-- res.SubmitSuccessfulReplication(task.nodes[i]) p.metrics.IncProcessedObjects() p.metrics.AddPayloadSize(int64(task.obj.PayloadSize())) } } }