mirror of
https://gitee.com/willfree/mlsr.git
synced 2026-10-04 19:42:02 +08:00
391 lines
11 KiB
Go
391 lines
11 KiB
Go
// Package mlsrcmd
|
|
// @Author: Wang Feng
|
|
// @Description:
|
|
// @Version: 0.1.0
|
|
// @Date: 2022/7/21 21:11
|
|
// @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
|
|
//
|
|
|
|
package mlsrcmd
|
|
|
|
import (
|
|
"encoding/json"
|
|
common2 "minlib/common"
|
|
"minlib/component"
|
|
"minlib/logicface"
|
|
"minlib/mgmt"
|
|
"minlib/security"
|
|
utils2 "minlib/utils"
|
|
"mlsr/common"
|
|
"mlsr/hello"
|
|
"mlsr/lsa"
|
|
"mlsr/lsdb"
|
|
"mlsr/mir"
|
|
"mlsr/route"
|
|
"mlsr/utils"
|
|
"time"
|
|
)
|
|
|
|
const unixPath = "/tmp/mir.sock"
|
|
|
|
type Mlsr struct {
|
|
LOCALHOST_PREFIX *component.Identifier
|
|
m_face *logicface.DummyClientLogicFace
|
|
m_scheduler *lsdb.MlsrScheduler
|
|
m_confParam *common.MlsrConfig
|
|
m_adjacencyList *lsa.AdjLsaAdjacenctList
|
|
m_namePrefixList *lsa.NameLsaNamePrefixList
|
|
m_fib *route.Fib
|
|
m_lsdb *lsdb.Lsdb
|
|
m_routingTable *route.RoutingTable
|
|
m_namePrefixTable *route.NamePrefixTable
|
|
m_helloProtocol *hello.HelloProtocol
|
|
|
|
// 用于异步事件触发的信号量接收器
|
|
m_onNewLsaConnection *utils2.Connection
|
|
m_onPrefixRegistrationSuccess *utils.SignalConnection
|
|
m_onInitialHelloDataValidated *utils.SignalConnection
|
|
|
|
// 用于与MIR进行交互的控制器
|
|
m_controller *mgmt.MIRController
|
|
m_faceDatasetController *mgmt.MIRController
|
|
|
|
// todo face通知流
|
|
}
|
|
|
|
// Init
|
|
// @Description: Mlsr初始化
|
|
// @receiver m
|
|
// @param face face
|
|
// @param scheduler
|
|
// @param passwd mir密码
|
|
// @param chain
|
|
// @param confParam
|
|
//
|
|
func (m *Mlsr) Init(face *logicface.DummyClientLogicFace, scheduler *lsdb.MlsrScheduler,
|
|
passwd string,
|
|
chain *security.KeyChain, confParam *common.MlsrConfig) {
|
|
common2.LogDebug("Initializing Mlsr")
|
|
// MLSR特有的本地前缀
|
|
nameIdentifier, err := component.CreateIdentifierByString("/localhost/nlsr")
|
|
if err != nil {
|
|
common2.LogFatal("mlsr identifier \"/localhost/nlsr\" create error, ", err)
|
|
}
|
|
|
|
// mir controller 初始化
|
|
controller := mgmt.CreateMIRController(func() (mgmt.IMgmtLogicFace, error) {
|
|
face := new(logicface.LogicFace)
|
|
// 建立unix连接
|
|
if err := face.InitWithUnixSocket(unixPath); err != nil {
|
|
return nil, err
|
|
}
|
|
return face, nil
|
|
}, true, chain)
|
|
|
|
// conf parameters 初始化
|
|
conf := new(common.MlsrConfigParameters)
|
|
conf.Init(confParam)
|
|
|
|
// fib 初始化
|
|
fib, err := route.NewFib(scheduler, controller, confParam.GetAdjacencys(), conf)
|
|
if err != nil {
|
|
common2.LogFatal("Fib Init Failed.")
|
|
}
|
|
|
|
m.LOCALHOST_PREFIX = nameIdentifier
|
|
|
|
m.m_face = face
|
|
m.m_scheduler = scheduler
|
|
m.m_confParam = confParam
|
|
m.m_adjacencyList = confParam.GetAdjacencys()
|
|
m.m_namePrefixList = confParam.GetNamePrefixs()
|
|
m.m_fib = fib
|
|
|
|
// lsdb
|
|
m.m_lsdb = new(lsdb.Lsdb)
|
|
m.m_lsdb.Init(m.m_confParam, m.m_scheduler, m.m_face)
|
|
|
|
// 路由表
|
|
m.m_routingTable = new(route.RoutingTable)
|
|
m.m_routingTable.Init(m.m_confParam, m.m_scheduler, m.m_lsdb)
|
|
|
|
// 名称前缀表
|
|
m.m_namePrefixTable = new(route.NamePrefixTable)
|
|
m.m_namePrefixTable.Init(m.m_confParam.GetRouterPrefix(), m.m_routingTable,
|
|
m.m_routingTable.AfterRoutingChange, m.m_lsdb.OnLsdbModified, fib)
|
|
|
|
// hello协议
|
|
m.m_helloProtocol = new(hello.HelloProtocol)
|
|
m.m_helloProtocol.Init(m.m_face, m.m_scheduler, chain, m.m_confParam, m.m_routingTable, m.m_lsdb)
|
|
|
|
// 三个信号量接收器的触发
|
|
// hello信号量接收器
|
|
bindHelloFunc := func(args ...interface{}) {
|
|
common2.LogInfo("Mlsr.bindHelloFunc")
|
|
// Hello协议收到有效Data包，解析其参数为邻居
|
|
neighbor, ok := args[0].(*component.Identifier)
|
|
if !ok || neighbor == nil {
|
|
common2.LogError("Invalid Identifier")
|
|
return
|
|
}
|
|
// 查看本机是否已知道此路由器
|
|
adjInfo := m.m_adjacencyList.FindAdjacencyByRouterIdentifier(neighbor)
|
|
if adjInfo != nil {
|
|
// 在fib中注册该前缀
|
|
// m_fib.registerPrefix(m_confParam.getSyncPrefix(), it->getFaceUri(), it->getLinkCost(),
|
|
// ndn::time::milliseconds::max(), ndn::nfd::ROUTE_FLAG_CAPTURE, 0);
|
|
if err := fib.RegisterPrefix(confParam.GetSyncPrefix(), adjInfo.LogicFaceUri(), adjInfo.LinkCost()); err != nil {
|
|
common2.LogError(err.Error())
|
|
}
|
|
}
|
|
}
|
|
m.m_onInitialHelloDataValidated = m.m_helloProtocol.OnInitialHelloDataValidated.Connect(bindHelloFunc)
|
|
|
|
// NewLSAConnection 信号量接收器
|
|
sync := m.m_lsdb.GetSync()
|
|
m.m_onNewLsaConnection = sync.OnNewLsa.Connect(func(i ...interface{}) {
|
|
if i == nil || len(i) < 4 {
|
|
common2.LogError("invalid arg list")
|
|
return
|
|
}
|
|
|
|
var originRouter *component.Identifier
|
|
var ok bool
|
|
if i[2] != nil {
|
|
originRouter, ok = i[2].(*component.Identifier)
|
|
if !ok {
|
|
common2.LogError("originRouter conversion failed.")
|
|
return
|
|
}
|
|
common2.LogDebug("originRouter: ", originRouter.ToUri())
|
|
}
|
|
|
|
m.RegisterStrategyForCerts(originRouter)
|
|
})
|
|
|
|
// onPrefixRegistrationSuccess 信号量接收器
|
|
m.m_onPrefixRegistrationSuccess = m.m_fib.OnPrefixRegistrationSuccess.Connect(func(i ...interface{}) {
|
|
if i == nil || len(i) < 1 {
|
|
common2.LogError("invalid arg list")
|
|
return
|
|
}
|
|
|
|
var name *component.Identifier
|
|
var ok bool
|
|
|
|
if i[0] != nil {
|
|
name, ok = i[0].(*component.Identifier)
|
|
if !ok {
|
|
common2.LogError("originRouter conversion failed.")
|
|
return
|
|
}
|
|
common2.LogDebug("originRouter: ", name.ToUri())
|
|
}
|
|
m.m_helloProtocol.SendHelloInterest(name)
|
|
})
|
|
|
|
// 设置组播策略
|
|
if err = m.m_fib.SetStrategy(confParam.GetLsaPrefix(), fib.MULTICAST_STRATEGY, 0); err != nil {
|
|
common2.LogFatal("设置组播策略失败：", err)
|
|
}
|
|
if err = m.m_fib.SetStrategy(confParam.GetSyncPrefix(), fib.MULTICAST_STRATEGY, 0); err != nil {
|
|
common2.LogFatal("设置组播策略失败：", err)
|
|
}
|
|
|
|
// 管理用 Controller
|
|
m.m_controller = mgmt.CreateMIRController(func() (mgmt.IMgmtLogicFace, error) {
|
|
face := new(logicface.LogicFace)
|
|
// 建立unix连接
|
|
if err := face.InitWithUnixSocket(unixPath); err != nil {
|
|
return nil, err
|
|
}
|
|
return face, nil
|
|
}, true, chain)
|
|
|
|
m.m_faceDatasetController = mgmt.CreateMIRController(func() (mgmt.IMgmtLogicFace, error) {
|
|
face := new(logicface.LogicFace)
|
|
// 建立unix连接
|
|
if err := face.InitWithUnixSocket(unixPath); err != nil {
|
|
return nil, err
|
|
}
|
|
return face, nil
|
|
}, true, chain)
|
|
|
|
// 拉取Face列表并初始化
|
|
m.InitializeFaces(m.processFaceDataset, m.onFaceDatasetFetchTimeout)
|
|
|
|
// todo: face通知流
|
|
|
|
}
|
|
|
|
// RegisterStrategyForCerts
|
|
// @Description: 为某个前缀设置转发策略
|
|
// @receiver m
|
|
// @param originRouter
|
|
//
|
|
func (m *Mlsr) RegisterStrategyForCerts(originRouter *component.Identifier) {
|
|
// NLSR中为命名空间routerKey 和 instanceKey 都设置了BEST_ROUTE策略
|
|
// 然而MIR中默认策略即为BEST_ROUTE策略，所以该函数不用实现
|
|
}
|
|
|
|
// AddDispatcherTopPrefix
|
|
// @Description: 注册TopPrefix
|
|
// @receiver m
|
|
//
|
|
func (m *Mlsr) AddDispatcherTopPrefix() {
|
|
// Dispatcher 用于处理rib的请求
|
|
// 暂时不需要实现
|
|
}
|
|
|
|
// RegisterPrefix
|
|
// @Description: 使用MLSR的LogicFace注册一个前缀
|
|
// @receiver m
|
|
// @param prefix
|
|
//
|
|
func (m *Mlsr) RegisterPrefix(prefix *component.Identifier) {
|
|
// Dispatcher 用于处理rib的请求
|
|
// 暂时不需要实现
|
|
}
|
|
|
|
// InitializeFaces
|
|
// @Description: 通过MIR的face列表初始化邻居节点
|
|
// @receiver m
|
|
// @param notification
|
|
//
|
|
func (m *Mlsr) onFaceEventNotification(notification interface{}) {
|
|
panic("Implement me")
|
|
}
|
|
|
|
// InitializeFaces
|
|
//
|
|
// @Description: 通过MIR的face列表初始化邻居节点
|
|
//
|
|
// @receiver m
|
|
//
|
|
// @param onFetchSuccess 拉取成功的回调函数
|
|
//
|
|
// @param onFetchFailure 拉取失败的回调函数
|
|
//
|
|
func (m *Mlsr) InitializeFaces(onFetchSuccess func([]mgmt.FaceInfo), onFetchFailure func(int, string)) {
|
|
// 参数校验
|
|
if onFetchFailure == nil || onFetchSuccess == nil {
|
|
common2.LogFatal("InitializeFaces: Invalid Arguments")
|
|
}
|
|
|
|
// 开始拉取Face列表
|
|
commandExecutor, err := m.m_faceDatasetController.PrepareCommandExecutor(mgmt.CreateLogicFaceListCommand(mir.TopPrefix))
|
|
if err != nil {
|
|
common2.LogFatal(err.Error())
|
|
}
|
|
commandExecutor.SetAutoShutdown(true)
|
|
|
|
// 执行命令拉取结果
|
|
response, err := commandExecutor.Start()
|
|
if err != nil {
|
|
// 拉取失败，调用onFetchFailure回调
|
|
onFetchFailure(response.Code, err.Error())
|
|
}
|
|
|
|
// 反序列化，得到Face列表
|
|
var faceInfoList []mgmt.FaceInfo
|
|
err = json.Unmarshal(response.GetBytes(), &faceInfoList)
|
|
if err != nil {
|
|
common2.LogFatal(err.Error())
|
|
}
|
|
|
|
// 拉取成功，调用onFetchSuccess回调
|
|
onFetchSuccess(faceInfoList)
|
|
}
|
|
|
|
// processFaceDataset(onFetchSuccess)
|
|
//
|
|
// @Description: 处理向MIR拉取到的Face列表
|
|
//
|
|
// @receiver m
|
|
//
|
|
// @param faces 拉取到的Face信息列表
|
|
//
|
|
func (m *Mlsr) processFaceDataset(faces []mgmt.FaceInfo) {
|
|
// 遍历nlsr.conf中指定的所有邻居节点
|
|
for _, adjacent := range m.m_adjacencyList.GetAdjList() {
|
|
faceUriString := adjacent.LogicFaceUri()
|
|
// 遍历 face 列表，查看有无匹配项
|
|
for _, face := range faces {
|
|
if adjacent.LogicFaceId() == 0 && faceUriString == face.RemoteUri {
|
|
// 更新邻居节点的LogicFaceId
|
|
adjacent.SetLogicFaceId(face.LogicFaceId)
|
|
m.registerAdjacencyPrefixes(adjacent)
|
|
}
|
|
}
|
|
|
|
if adjacent.LogicFaceId() == 0 {
|
|
common2.LogWarn("The adjacency ", adjacent.GetNeighborRouterIdentifier().ToUri(), " has no Face information in this dataset.")
|
|
}
|
|
}
|
|
|
|
// 继续调度FaceDataset的拉取任务
|
|
m.scheduleDatasetFetch()
|
|
}
|
|
|
|
// registerAdjacencyPrefixes
|
|
//
|
|
// @Description: 向MIR注册邻接节点的前缀
|
|
//
|
|
// @receiver m
|
|
//
|
|
// @param adj 邻接节点信息
|
|
//
|
|
func (m *Mlsr) registerAdjacencyPrefixes(adj *lsa.AdjLsaAdjacencyInfo) {
|
|
faceUri := adj.LogicFaceUri()
|
|
linkCost := adj.LinkCost()
|
|
adjName := adj.GetNeighborRouterIdentifier()
|
|
|
|
// 注册前缀AdjName
|
|
if err := m.m_fib.RegisterPrefix(adjName, faceUri, linkCost); err != nil {
|
|
common2.LogError(err.Error())
|
|
}
|
|
|
|
// 注册前缀LsaPrefix
|
|
if err := m.m_fib.RegisterPrefix(m.m_confParam.GetLsaPrefix(), faceUri, linkCost); err != nil {
|
|
common2.LogError(err.Error())
|
|
}
|
|
}
|
|
|
|
// scheduleDatasetFetch
|
|
//
|
|
// @Description: 调度向MIR拉取Face列表的任务
|
|
//
|
|
// @receiver m
|
|
//
|
|
func (m *Mlsr) scheduleDatasetFetch() {
|
|
// 生成调度任务
|
|
job, err := m.m_scheduler.ScheduleTaskAtTime(time.Now().
|
|
Add(time.Duration(m.m_confParam.LogicFaceDatasetFetchInterval)*time.Second), 1).
|
|
Do(func() {
|
|
m.InitializeFaces(m.processFaceDataset, m.onFaceDatasetFetchTimeout)
|
|
})
|
|
if err != nil {
|
|
// 生成任务异常，暂停调度
|
|
common2.LogError(err.Error())
|
|
return
|
|
}
|
|
job.LimitRunsTo(1)
|
|
// 开始调度任务
|
|
m.m_scheduler.StartAsync()
|
|
}
|
|
|
|
// onFaceDatasetFetchTimeout
|
|
//
|
|
// @Description: 向MIR拉取Face列表失败时的回调函数
|
|
//
|
|
// @receiver m
|
|
//
|
|
// @param code 错误代码
|
|
//
|
|
// @param msg 错误信息
|
|
//
|
|
func (m *Mlsr) onFaceDatasetFetchTimeout(code int, msg string) {
|
|
common2.LogError("Failed to fetch dataset: ", msg, "not trying again this time.")
|
|
m.scheduleDatasetFetch()
|
|
}
|