filecoin技术架构分析之五：filecoin源码分析之协议层心跳协议

《filecoin技术架构分析之五：filecoin源码分析之协议层心跳协议》由会员分享，可在线阅读，更多相关《filecoin技术架构分析之五：filecoin源码分析之协议层心跳协议（9页珍藏版）》请在装配图网上搜索。

1、目录5filecoin源码协议层分析之心跳协议o5.1源码信息o5.2源码分析5.2.1数据结构5.2.2方法5.2.3函数5.2.4实例化及业务逻辑5.1源码信息 versionomaster分支619b0eb1（2019年3月2日） packageometrics locationometrics/heartbeat.goonode/node.go5.2源码分析5.2.1 数据结构 定义心跳协议名称以及连接超时时间/HeartbeatProtocolisthelibp2pprotocolusedfortheheartbeatserviceconst（HeartbeatProtocol=fi

2、l/heartbeat/1.0.0/MinutestowaitbeforeloggingconnectionfailureatERRORlevelconnectionFailureErrorLogPeriodMinutes=10*time.Minute） 定义心跳信息结构o节点的区块头o节点的区块高度o节点的昵称o是否在区块同步中（TODO）/Heartbeatcontainstheinformationrequiredtodeterminethecurrentstateofanode./Heartbeatsareusedforaggregatinginformationaboutnodesi

3、nalogaggregator/tosupportalertinganddevnetvisualization.typeHeartbeatstruct/HeadrepresentstheheaviesttipsetthenodesisminingonHeadstring/HeightrepresentsthecurrentheightoftheTipsetHeightuint64/NicknameisthenicknamegiventothefilecoinnodebytheuserNicknamestring/TODO:addwhenimplemented/Syncingistrueifft

4、henodeiscurrentlysyncingitschainwiththenetwork./Syncingbool/Addressofthisnodesactiveminer.Canbeempty-willreturnthezeroaddressMinerAddressaddress.Address心跳服务结构体o主机结构体：对应libp2p主机o心跳配置o区块头获取o挖矿地址获取ostream锁ostream/HeartbeatServiceisresponsibleforsendingheartbeats.typeHeartbeatServicestructHosthost.HostC

5、onfig*config.HeartbeatConfig/AfunctionthatreturnstheheaviesttipsetHeadGetterfunc()types.TipSet/AfunctionthatreturnstheminersaddressMinerAddressGetterfunc()address.AddressstreamMusync.Mutexstreamnet.Stream定义心跳服务Option函数o函数入参为心跳服务结构体,主要用于对心跳服务结构体传参或者解析/HeartbeatServiceOptionisthetypeoftheheartbeatserv

6、icesfunctionaloptions.typeHeartbeatServiceOptionfunc(service*HeartbeatService)5.2.2 方法 获取心跳服务的stream实例/StreamreturnstheHeartbeatServicestream.Safeforconcurrentaccess./Streamisalibp2pconnectionthatheartbeatmessagesaresentovertoanaggregator.func(hbs*HeartbeatService)Stream()net.Streamhbs.streamMu.Lock

7、()deferhbs.streamMu.Unlock()returnhbs.stream 设置心跳服务的stream实例/SetStreamsetsthestreamontheHeartbeatService.Safeforconcurrentaccess.func(hbs*HeartbeatService)SetStream(snet.Stream)hbs.streamMu.Lock()deferhbs.streamMu.Unlock()hbs.stream=s 定时确认连接性，并调用运行心跳服务/Startstartstheheartbeatserviceby,startingthecon

8、nectionloop.Theconnection/loopwillattempttoconnectedtotheaggregatorservice,onceasuccessful/connectionismadewiththeaggregatorservicehearbeatswillbesenttoit./Iftheconnectionisbrokentheheartbeatservicewillattempttoreconnectvia/theconnectionloop.StartwillnotreturnuntilcontextctxisDone.func(hbs*Heartbeat

9、Service)Start(ctxcontext.Context)log.Debug(startingheartbeatservice)rd,err:=time.ParseDuration(hbs.Config.ReconnectPeriod)iferr!=nillog.Errorf(invalidheartbeatreconnectPeriod:%s,return/启动重连定时器reconTicker:=time.NewTicker(rd)deferreconTicker.Stop()/Timestampofthefirstconnectionfailuresincethelastsucce

10、ssfulconnection./Zeroinitiallyandwhileconnected.varfailedAttime.Time/TimestampofthelastERRORlog(oroffailure,beforethefirstERRORlog).varerroredAttime.Timeforselectcase-ctx.Done():returncaseconnectionFailureErrorLogPeriodMinuteslogfnlo=g.ErrorferroredAt=n/o/wResetthetimerfailureDuration:=now.Sub(faile

11、dAt)logfnH(eartbeatservicefailedtoconnectfor%s:%sfailureDuration,err)/failedtoconnect,continuereconnectloopcontinuefailedAt=time.Time/weconnected,sendheartbeats!/Runwillblockuntilitfailstosendaheartbeat./如果连接成功，运行心跳服务iferr:=hbs.Run(ctx);err!=nillog.Warning(disconnectingfromaggregator,failedtosendhea

12、rtbeat)continue运行心跳服务/Runiscalledoncetheheartbeatserviceconnectstotheaggregator.Run/sendtheactualheartbeat.RunwillblockuntilctxisDone.Anerrorwill/bereturnedifRunencountersanerrorwhensendingtheheartbeatandtheconnection/totheaggregatorwillbeclosed.func(hbs*HeartbeatService)Run(ctxcontext.Context)error

13、bd,err:=time.ParseDuration(hbs.Config.BeatPeriod)iferr!=nillog.Errorf(invalidheartbeatbeatPeriod:%s,err)returnerr/启动心跳定时器beatTicker:=time.NewTicker(bd)deferbeatTicker.Stop()/通过encoder进行流写入/TODOusecborinsteadofjsonencoder:=json.NewEncoder(hbs.stream)forselectcase-ctx.Done():returnnilcase-beatTicker.C

14、:/心跳定时周期到，调用Beat方法获取心跳参数hb:=hbs.Beat()/写入流，发起心跳iferr:=encoder.Encode(hb);err!=nil/发生错误会关闭流连接hbs.stream.Conn().Close(/)/nolint:errcheckreturnerr获取心跳参数/Beatwillcreateaheartbeat.func(hbs*HeartbeatService)Beat()Heartbeatnick:=hbs.Config.Nicknamets:=hbs.HeadGetter()tipset:=ts.ToSortedCidSet().String()hei

15、ght,err:=ts.Height()iferr!=nillog.Warningf(heartbeatservicefailedtogetchainheight:%s,err)addr:=hbs.MinerAddressGetter()returnHeartbeatHead:tipset,Height:height,Nickname:nick,MinerAddress:addr,心跳流连接/Connectwillconnectstohbs.Config.BeatTargetorreturnsanerrorfunc(hbs*HeartbeatService)Connect(ctxcontext

16、.Context)errorlog.Debugf(Heartbeatserviceattemptingtoconnect,targetAddress:%s,hbs.Config.BeatTarget)targetMaddr,err:=ma.NewMultiaddr(hbs.Config.BeatTarget)iferr!=nilreturnerrpid,err:=targetMaddr.ValueForProtocol(ma.P_P2P)iferr!=nilreturnerrpeerid,err:=peer.IDB58Decode(pid)iferr!=nilreturnerr/Decapsu

17、latethe/p2p/partfromthetarget/ip4/p2p/becomes/ip4/targetPeerAddr,_:=ma.NewMultiaddr(fmt.Sprintf(/p2p/%s,peer.IDB58Encode(peerid)targetAddr:=targetMaddr.Decapsulate(targetPeerAddr)hbs.Host.Peerstore().AddAddr(peerid,targetAddr,pstore.PermanentAddrTTL)/建立心跳服务流s,err:=hbs.Host.NewStream(ctx,peerid,Heart

18、beatProtocol)iferr!=nillog.Debugf(failedtoopenstream,peerID:%s,targetAddr:%s%s,peerid,targetAddr,err)returnerrlog.Infof(successfullytoopenstream,peerID:%s,targetAddr:%s,peerid,targetAddr)/设置流函数hbs.SetStream(s)returnnil5.2.3 函数向心跳服务结构体传参,用于设置获取矿工地址函数/WithMinerAddressGetterreturnsanoptionthatcanbeused

19、tosetthemineraddressgetter.funcWithMinerAddressGetter(agfunc()address.Address)HeartbeatServiceOptionreturnfunc(service*HeartbeatService)service.MinerAddressGetter=ag获取默认的矿工地址funcdefaultMinerAddressGetter()address.Addressreturnaddress.Address实例化心跳服务，具体的实例化在node包中实现。/NewHeartbeatServicereturnsaHeartbe

20、atServicefuncNewHeartbeatService(hhost.Host,hbc*config.HeartbeatConfig,hgfunc()types.TipSet,options.HeartbeatServiceOption)*HeartbeatServicesrv:=&HeartbeatServiceHost:h,Config:hbc,HeadGetter:hg,MinerAddressGetter:defaultMinerAddressGetter,/设置心跳服务的获取矿工属性,这会覆盖到上面设置的默认矿工地址for_,option:=rangeoptionsoptio

21、n(srv)returnsrv5.2.4 实例化及业务逻辑主要由node调用,location:node/node.go,主要逻辑如下在node的启动方法中，调用node.setupHeartbeatServices方法，建立心跳服务/Startbootsupthenode.func(node*Node)Start(ctxcontext.Context)erroriferr:=node.setupHeartbeatServices(ctx);err!=nilreturnerrors.Wrap(err,failedtostartheartbeatservices)returnnil建立心跳服务,

22、具体见如下注释func(node*Node)setupHeartbeatServices(ctxcontext.Context)error/设置“矿工地址获取函数”mag:=func()address.Addressaddr,err:=node.miningAddress()/theonlyerrorminingAddress()returnsisErrNoMinerAddress./ifthereisnoconfiguredmineraddress,simplysendazero/addressacrossthewire.iferr!=nilreturnaddress.Addressretu

23、rnaddr/存在心跳目标的时候，实例化心跳服务实例/starttheprimaryheartbeatserviceiflen(node.Repo.Config().Heartbeat.BeatTarget)0/调用metrics包中的建立心跳服务实例、以及启动心跳服务实例方法hbs:=metrics.NewHeartbeatService(node.Host(),node.Repo.Config().Heartbeat,node.ChainReader.Head,metrics.WithMinerAddressGetter(mag)gohbs.Start(ctx)/确认是否用户有通过环境变量

24、配置额外的心跳告警服务(自定义指向其他节点)，根据用户配置的数目，拉起对应的多线程心跳服务。/checkifwewanttoconnecttoanalertservice.Analertingserviceisaheartbeat/servicethatcantriggeralertsbasedonthecontentsofheatbeats.ifalertTarget:=os.Getenv(FIL_HEARTBEAT_ALERTS);len(alertTarget)0ahbs:=metrics.NewHeartbeatService(node.Host(),&config.HeartbeatConfig10s,rBeatTarget:alertTarget,BeatPeriod:ReconnectPeriod:10s,Nickname:node.Repo.Config().Heartbeat.Nickname,node.ChainReader.Head,metrics.WithMinerAddressGetter(mag)goahbs.Start(ctx)returnnil作者：先河CTO杨尉任何形式的转载都请联系作者获得授权并注明出处。