TrueCloudLab/frostfs-node
21
1
Fork 27
Code Issues 104 Pull requests 4 Projects Releases 2 Activity Actions

Fix EC put when some node is off #1233

Merged
fyrchik merged 2 commits from dstepanov-yadro/frostfs-node:fix/ec_put_node_off into master 2024-07-09 07:54:31 +00:00
Conversation 0 Commits 2 Files changed 2 +80 -11
1 changed files with 80 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 870c6254b8 - Show all commits

91
pkg/services/object/put/ec.go
View file

@ -3,8 +3,10 @@ package putsvc
import (
"context"
"crypto/ecdsa"
"encoding/hex"
"errors"
"fmt"
"sync/atomic"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
@ -160,7 +162,7 @@ func (e *ecWriter) writeECPart(ctx context.Context, obj *objectSDK.Object) error
}
fyrchik commented 2024-07-08 14:31:03 +00:00
Outdated
Review
Copy link

Irrelevant to the commit.

Irrelevant to the commit.
dstepanov-yadro commented 2024-07-09 05:43:07 +00:00
Review
Copy link

fixed

fixed
eg.Go(func() error {
return e.writePart(egCtx, obj, int(obj.ECHeader().Index()), nodes)
return e.writePart(egCtx, obj, int(obj.ECHeader().Index()), nodes, make([]atomic.Bool, len(nodes)))
})
t.SubmitSuccess()
}
@ -195,10 +197,15 @@ func (e *ecWriter) writeRawObject(ctx context.Context, obj *objectSDK.Object) er
break
}
visited := make([]atomic.Bool, len(nodes))
for idx := range parts {
fyrchik commented 2024-07-08 14:41:23 +00:00
Outdated
Review
Copy link

What does this line achieve? If we have 2 loops in writeECPart, we will still iterate over all nodes.

What does this line achieve? If we have 2 loops in `writeECPart`, we will still iterate over all nodes.
dstepanov-yadro commented 2024-07-08 14:46:50 +00:00
Outdated
Review
Copy link

If there are 3 parts and 4 nodes, and part 3 fails, so part 3 should try to save to node 4, not node 1 or node 2 (if node 1 or node 2 goroutines are not started yet).

If there are 3 parts and 4 nodes, and part 3 fails, so part 3 should try to save to node 4, not node 1 or node 2 (if node 1 or node 2 goroutines are not started yet).
👍 1
visited[idx%len(nodes)].Store(true)
}
for idx := range parts {
idx := idx
eg.Go(func() error {
return e.writePart(egCtx, parts[idx], idx, nodes)
return e.writePart(egCtx, parts[idx], idx, nodes, visited)
})
t.SubmitSuccess()
}
@ -211,19 +218,81 @@ func (e *ecWriter) writeRawObject(ctx context.Context, obj *objectSDK.Object) er
return nil
}
func (e *ecWriter) writePart(ctx context.Context, obj *objectSDK.Object, partIdx int, nodes []placement.Node) error {
var err error
node := nodes[partIdx%len(nodes)]
if e.cfg.netmapKeys.IsLocalKey(node.PublicKey()) {
err = e.writePartLocal(ctx, obj)
} else {
err = e.writePartRemote(ctx, obj, node)
func (e *ecWriter) writePart(ctx context.Context, obj *objectSDK.Object, partIdx int, nodes []placement.Node, visited []atomic.Bool) error {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
// try to save to node for current part index
node := nodes[partIdx%len(nodes)]
err := e.putECPartToNode(ctx, obj, node)
if err == nil {
return nil
}
e.cfg.log.Warn(logs.ECFailedToSaveECPart, zap.Stringer("parent_object", object.AddressOf(obj)), zap.Error(err))
return err
e.cfg.log.Warn(logs.ECFailedToSaveECPart, zap.Stringer("part_address", object.AddressOf(obj)),
zap.Stringer("parent_address", obj.ECHeader().Parent()), zap.Int("part_index", partIdx),
zap.String("node", hex.EncodeToString(node.PublicKey())), zap.Error(err))
partVisited := make([]bool, len(nodes))
partVisited[partIdx%len(nodes)] = true
// try to save to any node not visited by any of other parts
for i := 1; i < len(nodes); i++ {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
idx := (partIdx + i) % len(nodes)
if !visited[idx].CompareAndSwap(false, true) {
continue
}
node = nodes[idx]
err := e.putECPartToNode(ctx, obj, node)
if err == nil {
return nil
}
e.cfg.log.Warn(logs.ECFailedToSaveECPart, zap.Stringer("part_address", object.AddressOf(obj)),
zap.Stringer("parent_address", obj.ECHeader().Parent()), zap.Int("part_index", partIdx),
zap.String("node", hex.EncodeToString(node.PublicKey())),
zap.Error(err))
partVisited[idx] = true
}
// try to save to any node not visited by current part
fyrchik commented 2024-07-08 14:38:58 +00:00
Outdated
Review
Copy link

If we are here, it means that either partVisited[idx] == true or we skipped some iteration from the previous loop in if !visited[idx].CompareAndSwap(false, true). If partVisited[idx] == true, we skip the node, so this loop iterates over those visited[idx] which it skipped in the previous loop.

The question is: why do we need 2 loops?

If we are here, it means that either `partVisited[idx] == true` or we skipped some iteration from the previous loop in `if !visited[idx].CompareAndSwap(false, true)`. If `partVisited[idx] == true`, we skip the node, so this loop iterates over those `visited[idx]` which it skipped in the previous loop. The question is: why do we need 2 loops?
dstepanov-yadro commented 2024-07-08 14:53:09 +00:00
Outdated
Review
Copy link

It there are 3 parts and 4 nodes and part with index 2 fails on nodes with index 2 and 3, then state after first iteration will be:

visited[0] = true // visited by part 0
visited[1] = true // visited by part 1
visited[2] = true // visited by part 2 and failed
visited[3] = true // visited by part 2 and failed

So after first iteration part 3 will try to save to node 0 and 1.

It there are 3 parts and 4 nodes and part with index 2 fails on nodes with index 2 and 3, then state after first iteration will be: ``` visited[0] = true // visited by part 0 visited[1] = true // visited by part 1 visited[2] = true // visited by part 2 and failed visited[3] = true // visited by part 2 and failed ``` So after first iteration part 3 will try to save to node 0 and 1.
for i := 0; i < len(nodes); i++ {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
if partVisited[i] {
continue
}
node = nodes[i]
err := e.putECPartToNode(ctx, obj, node)
if err == nil {
return nil
}
e.cfg.log.Warn(logs.ECFailedToSaveECPart, zap.Stringer("part_address", object.AddressOf(obj)),
zap.Stringer("parent_address", obj.ECHeader().Parent()), zap.Int("part_index", partIdx),
zap.String("node", hex.EncodeToString(node.PublicKey())),
zap.Error(err))
}
return fmt.Errorf("failed to save EC chunk %s to any node", object.AddressOf(obj))
}
func (e *ecWriter) putECPartToNode(ctx context.Context, obj *objectSDK.Object, node placement.Node) error {
if e.cfg.netmapKeys.IsLocalKey(node.PublicKey()) {
return e.writePartLocal(ctx, obj)
}
return e.writePartRemote(ctx, obj, node)
}
func (e *ecWriter) writePartLocal(ctx context.Context, obj *objectSDK.Object) error {