From 62f63d4bf0bb48e942804fcabde76fbf46fbabfc Mon Sep 17 00:00:00 2001 From: zhengqi <709030194@qq.com> Date: Fri, 26 Aug 2022 17:10:57 +0800 Subject: [PATCH] add: lsdb supplement --- lsa/CoordinateLsa.go | 10 +- lsdb/Lsdb.go | 261 ++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 251 insertions(+), 20 deletions(-) diff --git a/lsa/CoordinateLsa.go b/lsa/CoordinateLsa.go index 2fb5738..6e984be 100644 --- a/lsa/CoordinateLsa.go +++ b/lsa/CoordinateLsa.go @@ -10,8 +10,8 @@ package lsa import "minlib/encoding" -type HyperbolicRadius float64 // 双曲半径 -type HyperbolicAngles []float64 // 双曲角度 +type HyperbolicRadius float64 // 双曲半径 +type HyperbolicAngles []float64 // 双曲角度 // // CoordinateLsa @@ -37,12 +37,12 @@ func (c *CoordinateLsa) ToString() string { return "" } -func (c *CoordinateLsa) WireEncode(encoder *encoding.Encoder) (int,error) { +func (c *CoordinateLsa) WireEncode(encoder *encoding.Encoder) (int, error) { // todo - return 0,nil + return 0, nil } func (c *CoordinateLsa) WireDecode(block *encoding.Block) error { // todo return nil -} \ No newline at end of file +} diff --git a/lsdb/Lsdb.go b/lsdb/Lsdb.go index 18fa8f4..f6663ad 100644 --- a/lsdb/Lsdb.go +++ b/lsdb/Lsdb.go @@ -1,5 +1,5 @@ // Package lsdb -// @Author: Wang Feng +// @Author: Wang Feng/zhengqi Wu // @Description: // Set的用法：https://www.jianshu.com/p/fee89fbb5f27 // @Version: 0.1.0 @@ -11,7 +11,6 @@ package lsdb import ( "MINSync" - "bytes" common2 "minlib/common" "minlib/component" mencoding "minlib/encoding" @@ -23,6 +22,8 @@ import ( "mlsr/lsa" "mlsr/lsdb/LsaContainer" "mlsr/utils" + "strconv" + "sync" "time" "github.com/liyue201/gostl/ds/set" @@ -66,7 +67,7 @@ type Lsdb struct { // 将LSA的名称从sync映射到其已知的最高序列号； // 用于阻止MLSR尝试获取过时的LSA - highestSeqNo map[string]uint64 + highestSeqNo sync.Map // 不同类型lsa的序列号管理器 sequencingManager *SequencingManager @@ -107,7 +108,7 @@ func (l *Lsdb) Init(mc *common.MlsrConfig, ms *MlsrScheduler, } // highestSeqNo - l.highestSeqNo = make(map[string]uint64) + // l.highestSeqNo = make(map[string]uint64) // 构造sync dummyClientLogicFace := logicface.NewDummyClientLogicFaceByOptions("dummy://face0", "", @@ -125,6 +126,11 @@ func (l *Lsdb) Init(mc *common.MlsrConfig, ms *MlsrScheduler, // m_onNewLsaConnection l.m_onNewLsaConnection = l.sync.OnNewLsa.Connect(func(args ...interface{}) { + if args == nil || len(args) < 4 { + common2.LogError("invalid arg list") + return + } + var updateName, originRouter *component.Identifier var seq, incomingLogicFaceId uint64 var ok bool @@ -769,7 +775,7 @@ func (l *Lsdb) ExpireOrRefreshLsa(lsa lsa.ILsa) { // @param deadline // func (l *Lsdb) ExpressInterest(interestName *component.Identifier, - timeOut, incomingFaceId uint32, deadline time.Time) { + timeOutCount, incomingFaceId uint32, deadline time.Time) { deadline = deadline.Add(LSA_REFRESH_TIME_MAX * time.Second) lsaName, err := interestName.GetSubIdentifier(0, interestName.Size()-1) @@ -787,10 +793,10 @@ func (l *Lsdb) ExpressInterest(interestName *component.Identifier, seqNo := seqNoComponent.GetNonNegativeInteger() lsaNameUri := lsaName.ToUri() - if v, ok := l.highestSeqNo[lsaNameUri]; !ok { - l.highestSeqNo[lsaNameUri] = seqNo - } else if seqNo > v { - l.highestSeqNo[lsaNameUri] = seqNo + if v, ok := l.highestSeqNo.Load(lsaNameUri); !ok { + l.highestSeqNo.Store(lsaNameUri, seqNo) + } else if seqNo > v.(uint64) { + l.highestSeqNo.Store(lsaNameUri, seqNo) } else { // Otherwise, its an old/invalid LSA common2.LogWarn("old/invalid LSA ", lsaNameUri, " seq: ", seqNo) @@ -810,16 +816,106 @@ func (l *Lsdb) ExpressInterest(interestName *component.Identifier, } l.m_fetchers.Insert(fetcher) - fetcher.AfterFragmentValidated.Connect(func(i ...interface{}) { + fetcher.AfterFragmentValidated.Connect(func(args ...interface{}) { + if args == nil || len(args) != 1 { + common2.LogError("invalid arg list") + return + } + // 解析data参数 + var data *packet.Data + var ok bool + if args[0] != nil { + data, ok = args[0].(*packet.Data) + if !ok { + common2.LogError("data conversion failed.") + return + } + } + // 插入内存数据库 + l.m_lsaStorage.Insert(data, MINSync.INFINITEWINDOW) + // 计划删除该分片 + fragmentName := data.GetName() + job, err := l.scheduler.ScheduleTaskAfterDuration(LSA_REFRESH_TIME_DEFAULT*time.Second, 1).Do(func(fragmentName *component.Identifier) { + l.m_lsaStorage.Erase(fragmentName, false) + }, fragmentName) + if err != nil { + common2.LogError("deletion scheduling error: ", err.Error()) + return + } + job.LimitRunsTo(1) + // 开始调度 + l.scheduler.StartAsync() }) - fetcher.OnComplete.Connect(func(i ...interface{}) { + fetcher.OnComplete.Connect(func(args ...interface{}) { + if args == nil || len(args) != 1 { + common2.LogError("invalid arg list") + return + } + // 解析buffer参数 + var buffer []byte + var ok bool + if args[0] != nil { + buffer, ok = args[0].([]byte) + if !ok { + common2.LogError("buffer conversion failed.") + return + } + } + // 从内存数据库中移除该lsa + lsaNameBlockForCopy, err := lsaName.SelfWireEncode(lsaName) + if err != nil { + common2.LogError("lsaName encode failed: ", err.Error()) + return + } + lsaNameCopy, err := component.NewIdentifierByBlock(lsaNameBlockForCopy) + if err != nil { + common2.LogError("new identifier failed: ", err.Error()) + return + } + lsaNameCopy.Append(component.CreateIdentifierComponentByNonNegativeInteger(seqNo - 1)) + l.m_lsaStorage.Erase(lsaNameCopy, false) + + // 从fetchers中移除该fetcher + l.m_fetchers.Erase(fetcher) + + // 将收到的数据传递给afterFetchLsa + l.afterFetchLsa(buffer, interestName) }) - fetcher.OnError.Connect(func(i ...interface{}) { + fetcher.OnError.Connect(func(args ...interface{}) { + if args == nil || len(args) != 2 { + common2.LogError("invalid arg list") + return + } + // 解析参数 + var errCode MINSync.ErrorCode + var msg string + var ok bool + + if args[0] != nil { + errCode, ok = args[0].(MINSync.ErrorCode) + if !ok { + common2.LogError("ErrorCode conversion failed.") + return + } + } + + if args[1] != nil { + msg, ok = args[1].(string) + if !ok { + common2.LogError("msg conversion failed.") + return + } + } + + // 调用onFetchLsaError + l.onFetchLsaError(errCode, msg, interestName, timeOutCount, deadline, lsaName, seqNo) + // 从fetchers中移除该fetcher + l.m_fetchers.Erase(fetcher) }) } @@ -835,9 +931,26 @@ func (l *Lsdb) ExpressInterest(interestName *component.Identifier, // @param lsaName // @param seqNo // -func (l *Lsdb) onFetchLsaError(errorCode uint32, msg string, +func (l *Lsdb) onFetchLsaError(errorCode MINSync.ErrorCode, msg string, interestName *component.Identifier, retransmitNo uint32, - deadline *time.Time, lsaName *component.Identifier, seqNo uint64) { + deadline time.Time, lsaName *component.Identifier, seqNo uint64) { + common2.LogDebug("Failed to fetch LSA: ", lsaName, ", Error Code: ", errorCode, ", Message: ", msg) + + if time.Now().Before(deadline) { + delay := l.mlsrConfig.GetLsaInterestLifetime() + if errorCode == MINSync.INTEREST_TIMEOUT { + delay = 0 * time.Second + } + // 调度LSA兴趣包重传任务 + job, err := l.scheduler.ScheduleTaskAfterDuration(delay, 1).Do(l.ExpressInterest, interestName, retransmitNo+1, 0, deadline) + if err != nil { + common2.LogError("retransimission scheduling error: ", err.Error()) + return + } + job.LimitRunsTo(1) + // 开始调度 + l.scheduler.StartAsync() + } } // @@ -849,7 +962,125 @@ func (l *Lsdb) onFetchLsaError(errorCode uint32, msg string, // @param buffer // @param interestName // -func (l *Lsdb) afterFetchLsa(buffer bytes.Buffer, interestName *component.Identifier) { +func (l *Lsdb) afterFetchLsa(buffer []byte, interestName *component.Identifier) { + common2.LogDebug("Received data for LSA interest: ", interestName.ToUri()) + + // 解析lsaName + lsaName, err := interestName.GetSubIdentifier(0, interestName.Size()-1) + if err != nil { + common2.LogError("lsaName error: ", err.Error()) + return + } + + // 解析seqNo + seqNoComponent, err := interestName.Get(-1) + if err != nil { + common2.LogError("seqNo error: ", err.Error()) + return + } + seqNo := seqNoComponent.GetNonNegativeInteger() + + // 更新highestSeqNo + lsaNameUri := lsaName.ToUri() + if v, ok := l.highestSeqNo.Load(lsaNameUri); !ok { + l.highestSeqNo.Store(lsaNameUri, seqNo) + } else if seqNo > v.(uint64) { + l.highestSeqNo.Store(lsaNameUri, seqNo) + } else { + // Otherwise, its an old/invalid LSA + common2.LogWarn("old/invalid LSA ", lsaNameUri, " seq: ", seqNo) + return + } + + chkString := "LSA" + lsaPosition := utils.GetNameComponentPosition(interestName, chkString) + + // 提取LSA类型并Install + if lsaPosition >= 0 { + // 复制originRouter + originRouter := l.mlsrConfig.GetNetwork() + originRouterBlockForCopy, err := originRouter.SelfWireEncode(originRouter) + if err != nil { + common2.LogError("originRouter encoding failed.") + return + } + originRouterCopy, err := component.NewIdentifierByBlock(originRouterBlockForCopy) + if err != nil { + common2.LogError("new identifier failed: ", err.Error()) + return + } + + // 复制subInterestName + interestNameBlockForCopy, err := interestName.SelfWireEncode(interestName) + if err != nil { + common2.LogError("interestName encoding failed: ", err.Error()) + return + } + interestNameCopy, err := component.NewIdentifierByBlock(interestNameBlockForCopy) + subInterestName, err := interestNameCopy.GetSubIdentifier(lsaPosition+1, interestNameCopy.Size()-lsaPosition-3) + if err != nil { + common2.LogError("get sub identifier failed: ", err.Error()) + return + } + + // 将subInterestName 附加到 originRouter + for _, comp := range subInterestName.GetComponents() { + originRouterCopy.Append(comp) + } + + // 获取LsaType组件 + lsTypeComponent, err := interestName.Get(-2) + if err != nil { + common2.LogError("get lsa type component failed: ", err.Error()) + return + } + + // 解析LsaType + interestLsTypeInInt, err := strconv.Atoi(lsTypeComponent.ToUri()) + if err != nil { + common2.LogError("Atoi failed: ", err.Error()) + return + } + + interestLsType := lsa.LsaType(interestLsTypeInInt) + + if interestLsType == lsa.LsaBaseType { + common2.LogError("Received Unsupported LSA Type: ", interestLsType) + return + } + + block, err := mencoding.CreateBlockByBuffer(buffer, true) + if err != nil { + common2.LogError("Create Block failed: ", err.Error()) + return + } + + if interestLsType == lsa.LsaNAMEType { + if l.isLsaNew(originRouterCopy, interestLsType, seqNo) { + nameLsa := new(lsa.NameLsa) + err = nameLsa.WireDecode(block) + if err != nil { + common2.LogError("name lsa decode failed: ", err.Error()) + return + } + l.InstallLsa(nameLsa) + } + } else if interestLsType == lsa.LsaADJACENCYType { + if l.isLsaNew(originRouterCopy, interestLsType, seqNo) { + adjLsa := new(lsa.AdjLsa) + err = adjLsa.WireDecode(block) + if err != nil { + common2.LogError("adj lsa decode failed: ", err.Error()) + return + } + l.InstallLsa(adjLsa) + } + } else if interestLsType == lsa.LsaCOORDINATEType { + if l.isLsaNew(originRouterCopy, interestLsType, seqNo) { + panic("implement me") + } + } + } } func (l *Lsdb) getLsaExpirationTimePoint() uint64 {