forked from TrueCloudLab/frostfs-node
[#1731] services/replicator: Unify Task interface with other parameters
Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>
This commit is contained in:
parent
4e043a801c
commit
898689ec14
4 changed files with 24 additions and 40 deletions
|
@ -86,11 +86,11 @@ func (s *Server) replicate(addr oid.Address, obj *objectSDK.Object) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
var res replicatorResult
|
var res replicatorResult
|
||||||
task := new(replicator.Task).
|
var task replicator.Task
|
||||||
WithObject(obj).
|
task.SetObject(obj)
|
||||||
WithObjectAddress(addr).
|
task.SetObjectAddress(addr)
|
||||||
WithCopiesNumber(1).
|
task.SetCopiesNumber(1)
|
||||||
WithNodes(nodes)
|
task.SetNodes(nodes)
|
||||||
s.replicator.HandleTask(context.TODO(), task, &res)
|
s.replicator.HandleTask(context.TODO(), task, &res)
|
||||||
|
|
||||||
if res.count == 0 {
|
if res.count == 0 {
|
||||||
|
|
|
@ -154,11 +154,11 @@ func (p *Policer) processNodes(ctx *processPlacementContext, addr oid.Address,
|
||||||
zap.Uint32("shortage", shortage),
|
zap.Uint32("shortage", shortage),
|
||||||
)
|
)
|
||||||
|
|
||||||
p.replicator.HandleTask(ctx, new(replicator.Task).
|
var task replicator.Task
|
||||||
WithObjectAddress(addr).
|
task.SetObjectAddress(addr)
|
||||||
WithNodes(nodes).
|
task.SetNodes(nodes)
|
||||||
WithCopiesNumber(shortage),
|
task.SetCopiesNumber(shortage)
|
||||||
checkedNodes,
|
|
||||||
)
|
p.replicator.HandleTask(ctx, task, checkedNodes)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -19,7 +19,7 @@ type TaskResult interface {
|
||||||
|
|
||||||
// HandleTask executes replication task inside invoking goroutine.
|
// HandleTask executes replication task inside invoking goroutine.
|
||||||
// Passes all the nodes that accepted the replication to the TaskResult.
|
// Passes all the nodes that accepted the replication to the TaskResult.
|
||||||
func (p *Replicator) HandleTask(ctx context.Context, task *Task, res TaskResult) {
|
func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult) {
|
||||||
defer func() {
|
defer func() {
|
||||||
p.log.Debug("finish work",
|
p.log.Debug("finish work",
|
||||||
zap.Uint32("amount of unfinished replicas", task.quantity),
|
zap.Uint32("amount of unfinished replicas", task.quantity),
|
||||||
|
|
|
@ -17,38 +17,22 @@ type Task struct {
|
||||||
nodes []netmap.NodeInfo
|
nodes []netmap.NodeInfo
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithCopiesNumber sets number of copies to replicate.
|
// SetCopiesNumber sets number of copies to replicate.
|
||||||
func (t *Task) WithCopiesNumber(v uint32) *Task {
|
func (t *Task) SetCopiesNumber(v uint32) {
|
||||||
if t != nil {
|
t.quantity = v
|
||||||
t.quantity = v
|
|
||||||
}
|
|
||||||
|
|
||||||
return t
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithObjectAddress sets address of local object.
|
// SetObjectAddress sets address of local object.
|
||||||
func (t *Task) WithObjectAddress(v oid.Address) *Task {
|
func (t *Task) SetObjectAddress(v oid.Address) {
|
||||||
if t != nil {
|
t.addr = v
|
||||||
t.addr = v
|
|
||||||
}
|
|
||||||
|
|
||||||
return t
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithObject sets object to avoid fetching it from the local storage.
|
// SetObject sets object to avoid fetching it from the local storage.
|
||||||
func (t *Task) WithObject(obj *objectSDK.Object) *Task {
|
func (t *Task) SetObject(obj *objectSDK.Object) {
|
||||||
if t != nil {
|
t.obj = obj
|
||||||
t.obj = obj
|
|
||||||
}
|
|
||||||
|
|
||||||
return t
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithNodes sets a list of potential object holders.
|
// SetNodes sets a list of potential object holders.
|
||||||
func (t *Task) WithNodes(v []netmap.NodeInfo) *Task {
|
func (t *Task) SetNodes(v []netmap.NodeInfo) {
|
||||||
if t != nil {
|
t.nodes = v
|
||||||
t.nodes = v
|
|
||||||
}
|
|
||||||
|
|
||||||
return t
|
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue