-
Notifications
You must be signed in to change notification settings - Fork 38
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
node/object: Serve new replication service
NeoFS protocol has been recently extended with new object replication RPC `ObjectService.Replicate` separated from the general-purpose `ObjectService.Put` one. According to API of the new RPC, all physically stored objects are transmitted in one message. Also, replication request and response formats are much simpler than for other operations. Serve new RPC by the storage node app. Requests are served similar to `ObjectService.Put` ones with TTL=1 (local only). Refs #2317. Signed-off-by: Leonard Lyubich <leonard@morphbits.io>
- Loading branch information
1 parent
9ca1fbd
commit 5fec85b
Showing
9 changed files
with
1,275 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| "github.com/hashicorp/golang-lru/v2/simplelru" | ||
| "github.com/nspcc-dev/neofs-node/pkg/core/container" | ||
| "github.com/nspcc-dev/neofs-node/pkg/core/netmap" | ||
| cid "github.com/nspcc-dev/neofs-sdk-go/container/id" | ||
| netmapsdk "github.com/nspcc-dev/neofs-sdk-go/netmap" | ||
| ) | ||
|
|
||
| // storagePolicyRes combines storage policy application result for particular | ||
| // container and network map in single unit. | ||
| type storagePolicyRes struct { | ||
| ns [][]netmapsdk.NodeInfo | ||
| err error | ||
| } | ||
|
|
||
| // containerNodesAtEpoch is a thread-safe LRU cache mapping containers into | ||
| // storage policy application results for particular network map. | ||
| type containerNodesAtEpoch struct { | ||
| mtx sync.RWMutex | ||
| lru simplelru.LRUCache[cid.ID, storagePolicyRes] | ||
| } | ||
|
|
||
| // containerNodes wraps NeoFS network state to apply container storage policies. | ||
| // | ||
| // Since policy application results are consistent for fixed container and | ||
| // network map, they could be cached. The containerNodes caches up to 1000 | ||
| // results for the latest and the previous epochs. The previous one is required | ||
| // to support data migration after the epoch tick. Older epochs are not cached. | ||
| type containerNodes struct { | ||
| containers container.Source | ||
| network netmap.Source | ||
|
|
||
| lastCurrentEpochMtx sync.Mutex | ||
| lastCurrentEpoch uint64 | ||
|
|
||
| curEpochCache, prevEpochCache *containerNodesAtEpoch | ||
| } | ||
|
|
||
| const cachedContainerNodesPerEpochNum = 1000 | ||
|
|
||
| func newContainerNodes(containers container.Source, network netmap.Source) (*containerNodes, error) { | ||
| lru, err := simplelru.NewLRU[cid.ID, storagePolicyRes](cachedContainerNodesPerEpochNum, nil) | ||
| if err != nil { | ||
| // should never happen | ||
| return nil, fmt.Errorf("create LRU cache for container nodes: %w", err) | ||
| } | ||
| return &containerNodes{ | ||
| containers: containers, | ||
| network: network, | ||
| curEpochCache: &containerNodesAtEpoch{lru: lru}, | ||
| prevEpochCache: new(containerNodesAtEpoch), | ||
| }, nil | ||
| } | ||
|
|
||
| // forEachNodePubKeyInSets passes binary-encoded public key of each node into f. | ||
| // When f returns false, forEachNodePubKeyInSets returns false instantly. | ||
| // Otherwise, true is returned. | ||
| func forEachNodePubKeyInSets(nodeSets [][]netmapsdk.NodeInfo, f func(pubKey []byte) bool) bool { | ||
| for i := range nodeSets { | ||
| for j := range nodeSets[i] { | ||
| if !f(nodeSets[i][j].PublicKey()) { | ||
| return false | ||
| } | ||
| } | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| // forEachContainerNodePublicKeyInLastTwoEpochs passes binary-encoded public key | ||
| // of each node match the referenced container's storage policy at two latest | ||
| // epochs into f. When f returns false, nil is returned instantly. | ||
| func (x *containerNodes) forEachContainerNodePublicKeyInLastTwoEpochs(cnrID cid.ID, f func(pubKey []byte) bool) error { | ||
| curEpoch, err := x.network.Epoch() | ||
| if err != nil { | ||
| return fmt.Errorf("read current NeoFS epoch: %w", err) | ||
| } | ||
|
|
||
| var curEpochCache, prevEpochCache *containerNodesAtEpoch | ||
| x.lastCurrentEpochMtx.Lock() | ||
| switch { | ||
| case curEpoch == x.lastCurrentEpoch-1: | ||
| curEpochCache = x.prevEpochCache | ||
| case curEpoch > x.lastCurrentEpoch: | ||
| curLRU, err := simplelru.NewLRU[cid.ID, storagePolicyRes](cachedContainerNodesPerEpochNum, nil) | ||
| if err != nil { | ||
| // should never happen | ||
| x.lastCurrentEpochMtx.Unlock() | ||
| return fmt.Errorf("create LRU cache for container nodes: %w", err) | ||
| } | ||
|
|
||
| if curEpoch == x.lastCurrentEpoch+1 { | ||
| x.prevEpochCache = x.curEpochCache | ||
| } else { | ||
| prevLRU, err := simplelru.NewLRU[cid.ID, storagePolicyRes](cachedContainerNodesPerEpochNum, nil) | ||
| if err != nil { | ||
| // should never happen | ||
| x.lastCurrentEpochMtx.Unlock() | ||
| return fmt.Errorf("create LRU cache for container nodes: %w", err) | ||
| } | ||
| x.prevEpochCache = &containerNodesAtEpoch{lru: prevLRU} | ||
| } | ||
| x.curEpochCache = &containerNodesAtEpoch{lru: curLRU} | ||
| x.lastCurrentEpoch = curEpoch | ||
| fallthrough | ||
| case curEpoch == x.lastCurrentEpoch: | ||
| curEpochCache = x.curEpochCache | ||
| prevEpochCache = x.prevEpochCache | ||
| } | ||
| x.lastCurrentEpochMtx.Unlock() | ||
|
|
||
| var cnr *container.Container | ||
|
|
||
| processEpoch := func(cache *containerNodesAtEpoch, epoch uint64) (storagePolicyRes, error) { | ||
| var result storagePolicyRes | ||
| var isCached bool | ||
| if cache != nil { | ||
| cache.mtx.Lock() | ||
| defer cache.mtx.Unlock() | ||
| result, isCached = cache.lru.Get(cnrID) | ||
| } | ||
| if !isCached { | ||
| if cnr == nil { | ||
| var err error | ||
| cnr, err = x.containers.Get(cnrID) | ||
| if err != nil { | ||
| // not persistent error => do not cache | ||
| return result, fmt.Errorf("read container by ID: %w", err) | ||
| } | ||
| } | ||
| networkMap, err := x.network.GetNetMapByEpoch(epoch) | ||
| if err != nil { | ||
| // not persistent error => do not cache | ||
| return result, fmt.Errorf("read network map at epoch #%d: %w", epoch, err) | ||
| } | ||
| result.ns, result.err = networkMap.ContainerNodes(cnr.Value.PlacementPolicy(), cnrID) | ||
| if cache != nil { | ||
| cache.lru.Add(cnrID, result) | ||
| } | ||
| } | ||
| return result, nil | ||
| } | ||
|
|
||
| cur, err := processEpoch(curEpochCache, curEpoch) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if cur.err == nil && !forEachNodePubKeyInSets(cur.ns, f) { | ||
| return nil | ||
| } | ||
| if curEpoch == 0 { | ||
| if cur.err != nil { | ||
| return fmt.Errorf("select container nodes for current epoch #%d: %w", curEpoch, cur.err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| prev, err := processEpoch(prevEpochCache, curEpoch-1) | ||
| if err != nil { | ||
| if cur.err != nil { | ||
| return fmt.Errorf("select container nodes for both epochs: (previous=%d) %w, (current=%d) %w", | ||
| curEpoch-1, err, curEpoch, cur.err) | ||
| } | ||
| return err | ||
| } | ||
| if prev.err == nil && !forEachNodePubKeyInSets(prev.ns, f) { | ||
| return nil | ||
| } | ||
|
|
||
| if cur.err != nil { | ||
| if prev.err != nil { | ||
| return fmt.Errorf("select container nodes for both epochs: (previous=%d) %w, (current=%d) %w", | ||
| curEpoch-1, prev.err, curEpoch, cur.err) | ||
| } | ||
| return fmt.Errorf("select container nodes for current epoch #%d: %w", curEpoch, cur.err) | ||
| } | ||
| if prev.err != nil { | ||
| return fmt.Errorf("select container nodes for previous epoch #%d: %w", curEpoch-1, prev.err) | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.