add: lsdb supplement

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
-}
+}

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 {