115 lines
2.4 KiB
Go
115 lines
2.4 KiB
Go
package searchsvc
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func (exec *execCtx) executeOnContainer(ctx context.Context) error {
|
|
lookupDepth := exec.netmapLookupDepth()
|
|
|
|
exec.log.Debug(logs.TryingToExecuteInContainer,
|
|
zap.Uint64("netmap lookup depth", lookupDepth),
|
|
)
|
|
|
|
// initialize epoch number
|
|
if err := exec.initEpoch(); err != nil {
|
|
return fmt.Errorf("%s: %w", logs.CouldNotGetCurrentEpochNumber, err)
|
|
}
|
|
|
|
for {
|
|
if err := exec.processCurrentEpoch(ctx); err != nil {
|
|
break
|
|
}
|
|
|
|
// check the maximum depth has been reached
|
|
if lookupDepth == 0 {
|
|
break
|
|
}
|
|
|
|
lookupDepth--
|
|
|
|
// go to the previous epoch
|
|
exec.curProcEpoch--
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (exec *execCtx) processCurrentEpoch(ctx context.Context) error {
|
|
exec.log.Debug(logs.ProcessEpoch,
|
|
zap.Uint64("number", exec.curProcEpoch),
|
|
)
|
|
|
|
traverser, err := exec.svc.traverserGenerator.GenerateTraverser(exec.containerID(), nil, exec.curProcEpoch)
|
|
if err != nil {
|
|
return fmt.Errorf("%s: %w", logs.SearchCouldNotGenerateContainerTraverser, err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
|
|
for {
|
|
addrs := traverser.Next()
|
|
if len(addrs) == 0 {
|
|
exec.log.Debug(logs.NoMoreNodesAbortPlacementIteration)
|
|
break
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
var mtx sync.Mutex
|
|
|
|
for i := range addrs {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
select {
|
|
case <-ctx.Done():
|
|
exec.log.Debug(logs.InterruptPlacementIterationByContext,
|
|
zap.String("error", ctx.Err().Error()))
|
|
return
|
|
default:
|
|
}
|
|
|
|
var info client.NodeInfo
|
|
|
|
client.NodeInfoFromNetmapElement(&info, addrs[i])
|
|
|
|
exec.log.Debug(logs.ProcessingNode, zap.String("key", hex.EncodeToString(addrs[i].PublicKey())))
|
|
|
|
c, err := exec.svc.clientConstructor.get(info)
|
|
if err != nil {
|
|
exec.log.Debug(logs.SearchCouldNotConstructRemoteNodeClient, zap.String("error", err.Error()))
|
|
return
|
|
}
|
|
|
|
ids, err := c.searchObjects(ctx, exec, info)
|
|
if err != nil {
|
|
exec.log.Debug(logs.SearchRemoteOperationFailed,
|
|
zap.String("error", err.Error()))
|
|
|
|
return
|
|
}
|
|
|
|
mtx.Lock()
|
|
err = exec.writeIDList(ids)
|
|
mtx.Unlock()
|
|
if err != nil {
|
|
exec.log.Debug(logs.SearchCouldNotWriteObjectIdentifiers, zap.String("error", err.Error()))
|
|
return
|
|
}
|
|
}(i)
|
|
}
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
return nil
|
|
}
|