frostfs-node/pkg/services/reputation/eigentrust/calculator/calls.go

291 lines
5.9 KiB
Go
Raw Normal View History

package eigentrustcalc
import (
"context"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/reputation"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/reputation/eigentrust"
apireputation "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/reputation"
"go.uber.org/zap"
)
type CalculatePrm struct {
last bool
ei eigentrust.EpochIteration
}
func (p *CalculatePrm) SetLast(last bool) {
p.last = last
}
func (p *CalculatePrm) SetEpochIteration(ei eigentrust.EpochIteration) {
p.ei = ei
}
func (c *Calculator) Calculate(prm CalculatePrm) {
alpha, err := c.prm.AlphaProvider.EigenTrustAlpha()
if err != nil {
c.opts.log.Debug(
"failed to get alpha param",
zap.Error(err),
)
return
}
c.alpha = reputation.TrustValueFromFloat64(alpha)
c.beta = reputation.TrustValueFromFloat64(1 - alpha)
ctx := eigentrust.IterContext{
Context: context.Background(),
EpochIteration: prm.ei,
}
iter := ctx.I()
log := c.opts.log.With(
zap.Uint64("epoch", ctx.Epoch()),
zap.Uint32("iteration", iter),
)
if iter == 0 {
c.sendInitialValues(ctx)
return
}
// decrement iteration number to select the values collected
// on the previous stage
ctx.SetI(iter - 1)
consumersIter, err := c.prm.DaughterTrustSource.InitConsumersIterator(ctx)
if err != nil {
log.Debug("consumers trust iterator's init failure",
zap.String("error", err.Error()),
)
return
}
// continue with initial iteration number
ctx.SetI(iter)
err = consumersIter.Iterate(func(daughter apireputation.PeerID, iter TrustIterator) error {
err := c.prm.WorkerPool.Submit(func() {
c.iterateDaughter(iterDaughterPrm{
lastIter: prm.last,
ctx: ctx,
id: daughter,
consumersIter: iter,
})
})
if err != nil {
log.Debug("worker pool submit failure",
zap.String("error", err.Error()),
)
}
// don't stop trying
return nil
})
if err != nil {
log.Debug("iterate daughter's consumers failed",
zap.String("error", err.Error()),
)
}
}
type iterDaughterPrm struct {
lastIter bool
ctx Context
id apireputation.PeerID
consumersIter TrustIterator
}
// nolint: funlen
func (c *Calculator) iterateDaughter(p iterDaughterPrm) {
initTrust, err := c.prm.InitialTrustSource.InitialTrust(p.id)
if err != nil {
c.opts.log.Debug("get initial trust failure",
zap.Stringer("daughter", p.id),
zap.String("error", err.Error()),
)
return
}
daughterIter, err := c.prm.DaughterTrustSource.InitDaughterIterator(p.ctx, p.id)
if err != nil {
c.opts.log.Debug("daughter trust iterator's init failure",
zap.String("error", err.Error()),
)
return
}
sum := reputation.TrustZero
err = p.consumersIter.Iterate(func(trust reputation.Trust) error {
if !p.lastIter {
select {
case <-p.ctx.Done():
return p.ctx.Err()
default:
}
}
sum.Add(trust.Value())
return nil
})
if err != nil {
c.opts.log.Debug("iterate over daughter's trusts failure",
zap.String("error", err.Error()),
)
return
}
// Alpha * Pd
initTrust.Mul(c.alpha)
sum.Mul(c.beta)
sum.Add(initTrust)
var intermediateTrust eigentrust.IterationTrust
intermediateTrust.SetEpoch(p.ctx.Epoch())
intermediateTrust.SetPeer(p.id)
intermediateTrust.SetI(p.ctx.I())
if p.lastIter {
finalWriter, err := c.prm.FinalResultTarget.InitIntermediateWriter(p.ctx)
if err != nil {
c.opts.log.Debug("init writer failure",
zap.String("error", err.Error()),
)
return
}
intermediateTrust.SetValue(sum)
err = finalWriter.WriteIntermediateTrust(intermediateTrust)
if err != nil {
c.opts.log.Debug("write final result failure",
zap.String("error", err.Error()),
)
return
}
} else {
intermediateWriter, err := c.prm.IntermediateValueTarget.InitWriter(p.ctx)
if err != nil {
c.opts.log.Debug("init writer failure",
zap.String("error", err.Error()),
)
return
}
err = daughterIter.Iterate(func(trust reputation.Trust) error {
select {
case <-p.ctx.Done():
return p.ctx.Err()
default:
}
val := trust.Value()
val.Mul(sum)
trust.SetValue(val)
err := intermediateWriter.Write(trust)
if err != nil {
c.opts.log.Debug("write value failure",
zap.String("error", err.Error()),
)
}
return nil
})
if err != nil {
c.opts.log.Debug("iterate daughter trusts failure",
zap.String("error", err.Error()),
)
}
err = intermediateWriter.Close()
if err != nil {
c.opts.log.Error(
"could not close writer",
zap.String("error", err.Error()),
)
}
}
}
func (c *Calculator) sendInitialValues(ctx Context) {
daughterIter, err := c.prm.DaughterTrustSource.InitAllDaughtersIterator(ctx)
if err != nil {
c.opts.log.Debug("all daughters trust iterator's init failure",
zap.String("error", err.Error()),
)
return
}
intermediateWriter, err := c.prm.IntermediateValueTarget.InitWriter(ctx)
if err != nil {
c.opts.log.Debug("init writer failure",
zap.String("error", err.Error()),
)
return
}
err = daughterIter.Iterate(func(daughter apireputation.PeerID, iterator TrustIterator) error {
return iterator.Iterate(func(trust reputation.Trust) error {
trusted := trust.Peer()
initTrust, err := c.prm.InitialTrustSource.InitialTrust(trusted)
if err != nil {
c.opts.log.Debug("get initial trust failure",
zap.Stringer("peer", trusted),
zap.String("error", err.Error()),
)
// don't stop on single failure
return nil
}
initTrust.Mul(trust.Value())
trust.SetValue(initTrust)
err = intermediateWriter.Write(trust)
if err != nil {
c.opts.log.Debug("write value failure",
zap.String("error", err.Error()),
)
// don't stop on single failure
}
return nil
})
})
if err != nil {
c.opts.log.Debug("iterate over all daughters failure",
zap.String("error", err.Error()),
)
}
err = intermediateWriter.Close()
if err != nil {
c.opts.log.Debug("could not close writer",
zap.String("error", err.Error()),
)
}
}