兄弟連Go語言+區(qū)塊鏈技術(shù)培訓(xùn)以太坊源碼分析(30)eth-bloombits和filter源碼分析

求E兄弟連教育兄弟連Go語言+區(qū)塊鏈技術(shù)培訓(xùn)以太坊源碼分析(30)eth-bloombits和filter源碼分析##以太坊的布隆過濾器以太坊的區(qū)塊頭中包含了一個叫做logsBloom的區(qū)域。這個區(qū)域存儲了當前區(qū)塊中所有的收據(jù)的日志的布隆過濾器，一共是2048個bit。也就是256個字節(jié)。而我們的一個交易的收據(jù)包含了很多的日志記錄。每個日志記錄包含了合約的地址，多個Topic。而在我們的收據(jù)中也存在一個布隆過濾器，這個布隆過濾器記錄了所有的日志記錄的信息。如果我們看黃皮書里面對日志記錄的形式化定義。O代表我們的日志記錄，Oa代表logger的地址，Oto,Ot1代表日志的Topics，Od代表時間。Oa是20個字節(jié)，Ot是32個字節(jié)，Od是很多字節(jié)我們定義了一個布隆過濾器函數(shù)M,用來把一個日志對象轉(zhuǎn)換成256字節(jié)的hashM3:2045是一個特別的函數(shù)，用來設(shè)置2048個bit位中的三位為1。對于任意的輸入值，首先求他的KEC輸出，然后通過取KEC輸出的[0,1][2,3],[4,5]這幾位的值對2048取模，得到三個值，這三個值就是輸出的2048中需要置位的下標。也就是說對于任何一個輸入，如果它對應(yīng)的三個下標的值不都為1，那么它肯定不在這個區(qū)塊中。當如如果對應(yīng)的三位都為1，也不能說明一定在這個區(qū)塊中。這就是布隆過濾器的特性。收據(jù)中的布隆過濾器就是所有的日志的布隆過濾器輸出的并集。同時區(qū)塊頭中的logBloom，就是所有的收據(jù)的布隆過濾器的并集。##ChainIndexer和BloomIndexer最開始看到ChainIndexer，不是很明白是什么功能。其實從名字中可以看到，是Chain的索引。在eth中我們有看到BloomIndexer,這個就是布隆過濾器的索引。在我們的協(xié)議中提供了查找指定Log的功能。用戶可以通過傳遞下面的參數(shù)來查找指定的Log,開始的區(qū)塊號，結(jié)束的區(qū)塊號，根據(jù)合約Addresses指定的地址過濾，根據(jù)指定的Topics來過濾。//FilterCriteriarepresentsarequesttocreateanewfilter.typeFilterCriteriastruct{FromBlock*big.IntToBlock*big.IntAddresses[]common.AddressTopics[][]common.Hash}如果開始和結(jié)束之間間隔很大，那么如果直接依次檢索每個區(qū)塊頭的logBloom區(qū)域是比較低效的。因為每個區(qū)塊頭都是分開存儲的，可能需要非常多的磁盤隨機訪問。所以以太坊協(xié)議在本地維護了一套索引，用來加速這個過程。大致原理是。每4096個區(qū)塊稱為一個Section,—個Section里面的logBloom會存儲在一起。對于每個Section,用一個二維數(shù)據(jù)，A[2048][4096]來存儲。第一維2048代表了bloom過濾器的長度2048個字節(jié)。第二維4096代表了一個Section里面的所有區(qū)塊，每一個位置按照順序代表了其中的一個區(qū)塊。A[0][0]=blockchain[section*4096+0].logBloom[0],A[0][1]=blockchain[section*4096+1].logBloom[0],A[0][4096]=blockchain[section*4096+1].logBloom[0],A[1][0]=blockchain[section*4096+0].logBloom[1],A[1][1024]=blockchain[section*4096+1024].logBloom[1],A[2047][1]=blockchain[section*4096+1].logBloom[2047],如果Section填充完畢，那么會寫成2048個KV。##bloombit.go代碼分析這個代碼相對不是很獨立，如果單獨看這個代碼，有點摸不著頭腦的感覺，因為它只是實現(xiàn)了一些接口，具體的處理邏輯并不在這里，而是在core里面。不過這里我先結(jié)合之前講到的信息分析一下。后續(xù)更詳細的邏輯在分析core的代碼的時候再詳細分析。服務(wù)線程startBloomHandlers,這個方法是為了響應(yīng)具體的查詢請求，給定指定的Section和bit來從levelDB里面查詢?nèi)缓蠓祷爻鋈?。單獨看這里有點摸不著頭腦。這個方法的調(diào)用比較復(fù)雜。涉及到core里面的很多邏輯。這里先不細說了。直到有這個方法就行了。typeRetrievalstruct{Bituint //Bit的取值0-2047代表了想要獲取哪一位的值Sections[]uint64//那些SectionBitsets[][]byte//返回值查詢出來的結(jié)果。}//startBloomHandlersstartsabatchofgoroutinestoacceptbloombitdatabase//retrievalsfrompossiblyarangeoffiltersandservingthedatatosatisfy.func(eth*Ethereum)startBloomHandlers(){fori:=0;i<bloomServiceThreads;i++{gofunc(){for{select{case<-eth.shutdownChan:returncaserequest:=<-eth.bloomRequests://request是一個通道task:=<-request//從通道里面獲取一個tasktask.Bitsets=make([][]byte,len(task.Sections))fori,section:=rangetask.Sections{head:=core.GetCanonicalHash(eth.chainDb,(section+1)*params.BloomBitsBlocks-1)blob,err:=bitutil.DecompressBytes(core.GetBloomBits(eth.chainDb,task.Bit,section,head),int(params.BloomBitsBlocks)/8)iferr!=nil{

panic(err)}task.Bitsets[i]=blob}request<-task//通過request通道返回結(jié)果}}}()}}###數(shù)據(jù)結(jié)構(gòu)Bloomlndexer對象主要用戶構(gòu)建索引的過程，是core.ChainIndexer的一個接口實現(xiàn),所以只實現(xiàn)了一些必須的接口。對于創(chuàng)建索引的邏輯還在core.ChainIndexer里面。//BloomIndexerimplementsacore.ChainIndexer,buildinguparotatedbloombitsindex//fortheEthereumheaderbloomfilters,permittingblazingfastfiltering.typeBloomIndexerstruct{sizeuint64//sectionsizetogeneratebloombitsfordbethdb.Database//databaseinstancetowriteindexdataandmetadataintogen*bloombits.Generator//generatortorotatethebloombitscratingthebloomindexsectionuint64//Sectionisthesectionnumberbeingprocessedcurrently當前的sectionheadcommon.Hash//Headisthehashofthelastheaderprocessed}//NewBloomIndexerreturnsachainindexerthatgeneratesbloombitsdataforthe//canonicalchainforfastlogsfiltering.funcNewBloomIndexer(dbethdb.Database,sizeuint64)*core.ChainIndexer{backend:=&BloomIndexer{db:db,size:size,}table:=ethdb.NewTable(db,string(core.BloomBitsIndexPrefix))returncore.NewChainIndexer(db,table,backend,size,bloomConfirms,bloomThrottling,"bloombits")}Reset實現(xiàn)了ChainIndexerBackend的方法，啟動一個新的section//Resetimplementscore.ChainIndexerBackend,startinganewbloombitsindex//section.func(b*BloomIndexer)Reset(sectionuint64){gen,err:=bloombits.NewGenerator(uint(b.size))iferr!=nil{panic(err)}b.gen,b.section,b.head=gen,section,common.Hash{}}Process實現(xiàn)了ChainIndexerBackend，增加一個新的區(qū)塊頭到index//Processimplementscore.ChainIndexerBackend,addinganewheader'sbloominto//theindex.func(b*BloomIndexer)Process(header*types.Header){b.gen.AddBloom(uint(header.Number.Uint64()-b.section*b.size),header.Bloom)b.head=header.Hash()}Commit方法實現(xiàn)了ChainIndexerBackend，持久化并寫入數(shù)據(jù)庫。//Commitimplementscore.ChainIndexerBackend,finalizingthebloomsectionand//writingitoutintothedatabase.func(b*BloomIndexer)Commit()error{batch:=b.db.NewBatch()fori:=0;i<types.BloomBitLength;i++{bits,err:=b.gen.Bitset(uint(i))iferr!=nil{returnerr}core.WriteBloomBits(batch,uint(i),b.section,b.head,bitutil.CompressBytes(bits))}returnbatch.Write()}##filter/api.go源碼分析eth/filter包包含了給用戶提供過濾的功能，用戶可以通過調(diào)用對交易或者區(qū)塊進行過濾，然后持續(xù)的獲取結(jié)果，如果5分鐘沒有操作，這個過濾器會被刪除。過濾器的結(jié)構(gòu)。var(deadline=5*time.Minute//considerafilterinactiveifithasnotbeenpolledforwithindeadline)//filterisahelperstructthatholdsmetainformationoverthefiltertype//andassociatedsubscriptionintheeventsystem.typefilterstruct{typType //過濾器的類型，過濾什么類型的數(shù)據(jù)deadline*time.Timer//filterisinactivwhendeadlinetriggers當計時器響起的時候，會觸發(fā)定時器。hashes[]common.Hash//過濾出來的hash結(jié)果critFilterCriteria//過濾條件logs[]*types.Log//過濾出來的Log信息s*Subscription//associatedsubscriptionineventsystem事件系統(tǒng)中的訂閱器。}構(gòu)造方法//PublicFilterAPIofferssupporttocreateandmanagefilters.Thiswillallowexternalclientstoretrievevarious//informationrelatedtotheEthereumprotocolsuchalsblocks,transactionsandlogs.//PublicFilterAPI用來創(chuàng)建和管理過濾器。允許外部的客戶端獲取以太坊協(xié)議的一些信息，比如區(qū)塊信息，交易信息和日志信息。typePublicFilterAPIstruct{backendBackendmux*event.TypeMuxquitchanstruct{}chainDbethdb.Databaseevents*EventSystemfiltersMusync.Mutexfiltersmap[rpc.ID]*filter}//NewPublicFilterAPIreturnsanewPublicFilterAPIinstance.funcNewPublicFilterAPI(backendBackend,lightModebool)*PublicFilterAPI{api:=&PublicFilterAPI{backend:backend,mux:backend.EventMux(),chainDb:backend.ChainDb(),events:NewEventSystem(backend.EventMux(),backend,lightMode),filters:make(map[rpc.ID]*filter),}goapi.timeoutLoop()returnapi}###超時檢查//timeoutLooprunsevery5minutesanddeletesfiltersthathavenotbeenrecentlyused.//Ttisstartedwhentheapiiscreated.//每隔5分鐘檢查一下。如果過期的過濾器，刪除。func(api*PublicFilterAPI)timeoutLoop(){ticker:=time.NewTicker(5*time.Minute)for{<-ticker.Capi.filtersMu.Lock()forid,f:=rangeapi.filters{select{case<-f.deadline.C:f.s.Unsubscribe()delete(api.filters,id)default:continue}}api.filtersMu.Unlock()}}NewPendingTransactionFilter,用來創(chuàng)建一個PendingTransactionFilter。這種方式是用來給那種無法創(chuàng)建長連接的通道使用的（比如HTTP）,如果對于可以建立長鏈接的通道（比如WebSocket）可以使用rpc提供的發(fā)送訂閱模式來處理，就不用持續(xù)的輪詢了//NewPendingTransactionFiltercreatesafilterthatfetchespendingtransactionhashes//astransactionsenterthependingstate.////Itispartofthefilterpackagebecausethisfiltercanbeusedthrougthe//'eth_getFilterChanges'pollingmethodthatisalsousedforlogfilters./////ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilterfunc(api*PublicFilterAPI)NewPendingTransactionFilter()rpc.ID{var(pendingTxs=make(chancommon.Hash)//在事件系統(tǒng)訂閱這種消息pendingTxSub=api.events.SubscribePendingTxEvents(pendingTxs))api.filtersMu.Lock()api.filters[pendingTxSub.ID]=&filter{typ:PendingTransactionsSubscription,deadline:time.NewTimer(deadline),hashes:make([]common.Hash,0),s:pendingTxSub}api.filtersMu.Unlock()gofunc(){for{select{caseph:=<-pendingTxs://接收至UpendingTxs，存儲在過濾器的hashes容器里面。api.filtersMu.Lock()iff,found:=api.filters[pendingTxSub.ID];found{f.hashes=append(f.hashes,ph)}api.filtersMu.Unlock()case<-pendingTxSub.Err():api.filtersMu.Lock()delete(api.filters,pendingTxSub.ID)api.filtersMu.Unlock()return}}}()returnpendingTxSub.ID}輪詢:GetFilterChanges//GetFilterChangesreturnsthelogsforthefilterwiththegivenidsince//lasttimeitwascalled.Thiscanbeusedforpolling.//GetFilterChanges用來返回從上次調(diào)用到現(xiàn)在的所有的指定id的所有過濾信息。這個可以用來輪詢。//Forpendingtransactionandblockfilterstheresultis[]common.Hash.//(pending)Logfiltersreturn[]Log.//對于pendingtransaction和block的過濾器，返回結(jié)果類型是[]common.Hash.對于pendingLog過濾器，返回的是[]Log///ethereum/wiki/wiki/JSON-RPC#eth_getfilterchangesfunc(api*PublicFilterAPI)GetFilterChanges(idrpc.ID)(interface{},error){api.filtersMu.Lock()deferapi.filtersMu.Unlock()iff,found:=api.filters[id];found{if!f.deadline.Stop(){//如果定時器已經(jīng)觸發(fā)，但是filter還沒有移除，那么我們先接收定時器的值，然后重置定時器//timerexpiredbutfilterisnotyetremovedintimeoutloop//receivetimervalueandresettimer<-f.deadline.C}f.deadline.Reset(deadline)switchf.typ{casePendingTransactionsSubscription,BlocksSubscription:hashes:=f.hashesf.hashes=nilreturnreturnHashes(hashes),nilcaseLogsSubscription:logs:=f.logsf.logs=nilreturnreturnLogs(logs),nil}return[]interface{}{},fmt.Errorf("filternotfound")}對于可以建立長連接的通道，可以直接使用rpc的發(fā)送訂閱模式，這樣客戶端就可以直接接收到過濾信息，不用調(diào)用輪詢的方式了?？梢钥吹竭@種模式下面并沒有添加到filters這個容器，也沒有超時管理了。也就是說支持兩種模式。//NewPendingTransactionscreatesasubscriptionthatistriggeredeachtimeatransaction//entersthetransactionpoolandwassignedfromoneofthetransactionsthisnodesmanages.func(api*PublicFilterAPI)NewPendingTransactions(ctxcontext.Context)(*rpc.Subscription,error){notifier,supported:=rpc.NotifierFromContext(ctx)if!supported{return&rpc.Subscription{},rpc.ErrNotificationsUnsupported}rpcSub:=notifier.CreateSubscription()gofunc(){txHashes:=make(chancommon.Hash)pendingTxSub:=api.events.SubscribePendingTxEvents(txHashes)for{select{caseh:=<-txHashes:notifier.Notify(rpcSub.ID,h)case<-rpcSub.Err():pendingTxSub.Unsubscribe()returncase<-notifier.Closed():pendingTxSub.Unsubscribe()return}}}()returnrpcSub,nil日志過濾功能，根據(jù)FilterCriteria指定的參數(shù)，來對日志進行過濾，開始區(qū)塊，結(jié)束區(qū)塊，地址和Topics，這里面引入了一個新的對象filter//FilterCriteriarepresentsarequesttocreateanewfilter.typeFilterCriteriastruct{FromBlock*big.IntToBlock*big.IntAddresses[]common.AddressTopics[][]common.Hash}//GetLogsreturnslogsmatchingthegivenargumentthatarestoredwithinthestate./////ethereum/wiki/wiki/JSON-RPC#eth_getlogsfunc(api*PublicFilterAPI)GetLogs(ctxcontext.Context,critFilterCriteria)([]*types.Log,error){//ConverttheRPCblocknumbersintointernalrepresentationsifcrit.FromBlock==nil{crit.FromBlock=big.NewInt(rpc.LatestBlockNumber.Int64())}ifcrit.ToBlock==nil{crit.ToBlock=big.NewInt(rpc.LatestBlockNumber.Int64())}//Createandrunthefiltertogetallthelogs//創(chuàng)建了一個Filter對象然后調(diào)用filter.Logsfilter:=New(api.backend,crit.FromBlock.Int64(),crit.ToBlock.Int64(),crit.Addresses,crit.Topics)logs,err:=filter.Logs(ctx)iferr!=nil{returnnil,err}returnreturnLogs(logs),err}##filter.gofiter.go里面定義了一個Filter對象。這個對象主要用來根據(jù)區(qū)塊的BloomIndexer和布隆過濾器等來執(zhí)行日志的過濾功能。###數(shù)據(jù)結(jié)構(gòu)//后端，這個后端其實是在core里面實現(xiàn)的。布隆過濾器的主要算法在core里面實現(xiàn)了。typeBackendinterface{ChainDb()ethdb.DatabaseEventMux()*event.TypeMuxHeaderByNumber(ctxcontext.Context,blockNrrpc.BlockNumber)(*types.Header,error)GetReceipts(ctxcontext.Context,blockHashcommon.Hash)(types.Receipts,error)SubscribeTxPreEvent(chan<-core.TxPreEvent)event.SubscriptionSubscribeChainEvent(chchan<-core.ChainEvent)event.SubscriptionSubscribeRemovedLogsEvent(chchan<-core.RemovedLogsEvent)event.SubscriptionSubscribeLogsEvent(chchan<-[]*types.Log)event.SubscriptionBloomStatus()(uint64,uint64)ServiceFilter(ctxcontext.Context,session*bloombits.MatcherSession)}//Filtercanbeusedtoretrieveandfilterlogs.typeFilterstruct{backendBackend //后端dbethdb.Database//數(shù)據(jù)庫begin,endint64 //開始結(jié)束區(qū)塊addresses[]common.Address//篩選地址topics[][]common.Hash//篩選主題matcher*bloombits.Matcher//布隆過濾器的匹配器}構(gòu)造函數(shù)把address和topic都加入到filters容器。然后構(gòu)建了一個bloombits.NewMatcher(size,filters)。這個函數(shù)在core里面實現(xiàn)，暫時不會講解。//Newcreatesanewfilterwhichusesabloomfilteronblockstofigureoutwhether//aparticularblockisinterestingornot.funcNew(backendBackend,begin,endint64,addresses[]common.Address,topics[][]common.Hash)*Filter{//Flattentheaddressandtopicfilterclausesintoasinglebloombitsfilter//system.Sincethebloombitsarenotpositional,niltopicsarepermitted,//whichgetflattenedintoanilbyteslice.varfilters[][][]byteiflen(addresses)>0{filter:=make([][]byte,len(addresses))fori,address:=rangeaddresses{filter[i]=address.Bytes()}filters=append(filters,filter)}for_,topicList:=rangetopics{filter:=make([][]byte,len(topicList))fori,topic:=rangetopicList{filter[i]=topic.Bytes()}filters=append(filters,filter)}//Assembleandreturnthefiltersize,_:=backend.BloomStatus()return&Filter{backend:backend,begin:begin,end:end,addresses:addresses,topics:topics,db:backend.ChainDb(),matcher:bloombits.NewMatcher(size,filters),}}Logs執(zhí)行過濾//Logssearchestheblockchainformatchinglogentries,returningallfromthe//firstblockthatcontainsmatches,updatingthestartofthefilteraccordingly.func(f*Filter)Logs(ctxcontext.Context)([]*types.Log,error){//Figureoutthelimitsofthefilterrangeheader,_:=f.backend.HeaderByNumber(ctx,rpc.LatestBlockNumber)ifheader==nil{returnnil,nil}head:=header.Number.Uint64()iff.begin==-1{f.begin=int64(head)}end:=uint64(f.end)iff.end==-1{end=head}//Gatherallindexedlogs,andfinishwithnonindexedonesvar(logs[]*types.Logerrerror)size,sections:=f.backend.BloomStatus()//indexed是指創(chuàng)建了索引的區(qū)塊的最大值。如果過濾的范圍落在了創(chuàng)建了索引的部分。//那么執(zhí)行索引搜索。ifindexed:=sections*size;indexed>uint64(f.begin){ifindexed>end{logs,err=f.indexedLogs(ctx,end)}else{logs,err=f.indexedLogs(ctx,indexed-1)}iferr!=nil{returnlogs,err}}//對于剩下的部分執(zhí)行非索引的搜索。rest,err:=f.unindexedLogs(ctx,end)logs=append(logs,rest...)returnlogs,err}索引搜索//indexedLogsreturnsthelogsmatchingthefiltercriteriabasedonthebloom//bitsindexedavailablelocallyorviathenetwork.func(f*Filter)indexedLogs(ctxcontext.Context,enduint64)([]*types.Log,error){//Createamatchersessionandrequestservicingfromthebackendmatches:=make(chanuint64,64)//啟動matchersession,err:=f.matcher.Start(uint64(f.begin),end,matches)iferr!=nil{returnnil,err}defersession.Close(time.Second)//進行過濾服務(wù)。這些都在core里面。后續(xù)分析core的代碼會進行分析。f.backend.ServiceFilter(ctx,session)//Iterateoverthematchesuntilexhaustedorcontextclosedvarlogs[]*types.Logfor{select{casenumber,ok:=<-matches://Abortifallmatcheshavebeenfulfilledif!ok{//沒有接收到值并且channel已經(jīng)被關(guān)閉f.begin=int64(end)+1//更新begin。以便于下面的非索引搜索returnlogs,nil}//Retrievethesuggestedblockandpullanytrulymatchinglogsheader,err:=f.backend.HeaderByNumber(ctx,rpc.BlockNumber(number))ifheader==nil||err!=nil{returnlogs,err}found,err:=f.checkMatches(ctx,header)//查找匹配的值iferr!=nil{returnlogs,err}logs=append(logs,found...)case<-ctx.Done():returnlogs,ctx.Err()}}}checkMatches,拿到所有的收據(jù)，并從收據(jù)中拿到所有的日志。執(zhí)行filterLogs方法。//checkMatcheschecksifthereceiptsbelongingtothegivenheadercontainanylogeventsthat//matchthefiltercriteria.Thisfunctioniscalledwhenthebloomfiltersignalsapotentialmatch.func(f*Filter)checkMatches(ctxcontext.Context,header*types.Header)(logs[]*types.Log,errerror){//Getthelogsoftheblockreceipts,err:=f.backend.GetReceipts(ctx,header.Hash())iferr!=nil{returnnil,err}varunfiltered[]*types.Logfor_,receipt:=rangereceipts{unfiltered=append(unfiltered,([]*types.Log)(receipt.Logs)...)}logs=filterLogs(