正文

精通 Filecoin：Lotus真实数据处理之Provider初始化

乔疯2020-08-03 06:50:03

因为


      StorageProvider

对象被存储矿工 API 对象所依赖，所以在启动存储矿工的过程中，DI 容器会调用


      StorageProvider

函数（node/modules/storageminer.go）来创建它。


      StorageProvider

函数流程如下：

调用 NewFromLibp2pHost 函数，生成 StorageMarketNetwork 对象。
```
net := smnet.NewFromLibp2pHost(h)
```

调用


        NewLocalFileStore

函数，生成


        FileStore

存储对象。

store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))


        NewLocalFileStore

函数(go-fil-markets 类库 filestore/filestore.go)流程如下：

base := filepath.Clean(string(basedirectory))
info, err := os.Stat(string(base))

if !info.IsDir() { return nil, fmt.Errorf("%s is not a directory", base) }

return &fileStore{string(base)}, nil NewLocalFileStore 函数使用的路径为仓库目录。即碎片的临时目录就是仓库目录。

调用


     CustomDealDecisionLogic

函数，返回一个函数对象。在函数对象中调用我们提供的回调函数，进行自定义交易逻辑判断。

opt := storageimpl.CustomDealDecisionLogic(func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {})

生成并返回


     StorageProvider

对象。

p, err := storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), ibs, store, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk, opt)return p, nil


     NewProvider

函数处理如下：

生成


       PieceIOWithStore

对象。

carIO := cario.NewCarIO()
pio := pieceio.NewPieceIOWithStore(carIO, fs, bs)

生成


       Provider

对象。

h := &Provider{
    net:                  net,
    proofType:            rt,
    spn:                  spn,
    fs:                   fs,
    pio:                  pio,
    pieceStore:           pieceStore,
    conns:                connmanager.NewConnManager(),
    storedAsk:            storedAsk,
    actor:                minerAddress,
    dataTransfer:         dataTransfer,
    dealAcceptanceBuffer: DefaultDealAcceptanceBuffer,
    pubSub:               pubsub.New(providerDispatcher),
}

生成 fsm 状态组对象。

deals, err := NewProviderStateMachine(
    ds,
    &providerDealEnvironment{h},
    h.dispatch,
)h.deals = deals

fsm 状态组对象使用的配置参数如下：

return fsm.New(ds, fsm.Parameters{
    Environment:     env,
    StateType:       storagemarket.MinerDeal{},
    StateKeyField:   "State",
    Events:          providerstates.ProviderEvents,
    StateEntryFuncs: providerstates.ProviderStateEntryFuncs,
    FinalityStates:  providerstates.ProviderFinalityStates,
    Notifier:        notifier,
})

环境对象为 providerDealEnvironment 。
状态对象为 MinerDeal 。
状态字段为 State 。
事件集合为 ProviderEvents ，参考 storagemarket/impl/providerstates/provider_fsm.go 文件。
状态处理函数集合为 ProviderStateEntryFuncs ，状态机的状态处理器根据对应的状态获取到指定的函数进行处理。
终止状态集合为 ProviderFinalityStates 。
通知对象为 Provider 对象的 dispatch 方法。

使用配置选项，配置 Provider 对象。
```
h.Configure(options...)
```
设置数据传输监听对象。
```
dataTransfer.SubscribeToEvents(dtutils.ProviderDataTransferSubscriber(deals))
```
当开始数据传输、传输结束、传输错误时会发送 ProviderEventDataTransferInitiated 、 ProviderEventDataTransferCompleted 、 ProviderEventDataTransferFailed 等事件到 fsm 状态组。
返回 Provider 对象。

在存储矿工启动过程自动调用 HandleDeals 函数（node/modules/storageminer.go）。在这个函数中，调用 StorageProvider 对象的 Start 方法，从而启动这个对象。

Start 方法执行过程如下：

调用


      StorageMarketNetwork

网络对象的


      SetDelegate

设置代理/委托为自身。

err := p.net.SetDelegate(p)

网络对象的实现为


      libp2pStorageMarketNetwork

结构体（storagemarket/network/libp2p_impl.go）。它的


      SetDelegate

方法内容如下：

impl.receiver = r
impl.host.SetStreamHandler(storagemarket.DealProtocolID, impl.handleNewDealStream)
impl.host.SetStreamHandler(storagemarket.AskProtocolID, impl.handleNewAskStream)
return nil

上面分别设置网络对象的


      handleNewDealStream

方法处理


      DealProtocolID

协议，表示存储；


      handleNewAskStream

方法处理


      AskProtocolID

协议，表示 ask。

handleNewDealStream 方法内容如下：

// 客户端 peer id
remotePID := s.Conn().RemotePeer()  buffered := bufio.NewReaderSize(s, 16)
// 对流进行包装
ds := &dealStream{remotePID, impl.host, s, buffered}
// 调用 StorageProvider 对象的 HandleDealStream 方法，处理客户端存储请求
impl.receiver.HandleDealStream(ds)

在协程中调用 StorageProvider 对象的 restartDeals 方法，重新进行交易处理。 restartDeals 方法流程如下：
- 从 fsm 状态组对象中获取所有的交易对象。
```
var deals []storagemarket.MinerDeal
err := c.deals.List(&deals)
```
- 遍历所有的交易对象，进行下面的处理：
  - 如果当前交易对象已经终止，则进行下一个处理。
  - 如果当前交易对象的连接已经关闭，则进行下一个处理。
  - 发送初始交易事件给 fsm 状态组。
```
err = c.deals.Send(deal.ProposalCid, storagemarket.ProviderEventRestart)
```
    交易提案的 Cid 表示了状态机的名称/编号。
返回空值。

免责声明：本文版权归原作者所有，不代表MyToken(www.mytokencap.com)观点和立场；如有关于内容、版权等问题，请与我们联系。

关于MyToken：https://www.mytokencap.com/aboutus本文链接：https://www.mytokencap.com/news/152901.html

上一篇:交易量再创新高！Uniswap 远超各大交易所的原因竟是因为它

下一篇:挑战百度网盘主网即将上线 IPFS到底是个啥？

精通 Filecoin：Lotus 真实数据处理之 Client 初始化

Lotus是怎么处理真实数据的？...

精通Filecoin：Filecoin 源码之 Hello 协议

当调用 Filecoin 全节点的 start 方法启动全节点时，调用 hello 协议的 New 方法，这个方法的处理如下：...

精通IPFS：系统启动篇

当用户以守护进程模式启动 filecoin 时，最终执行的命令位于 commands/daemon.go 文件，它的 Run 方法直接调用同文件中的 daemonRun 函数进行处理。...