IPFS(三)源碼解讀之-add

2023年12月9日發(作者：四季的問候)

-

IPFS（三）源碼解讀之-add

Add 所作的事其實就是將文件傳到IPFS上，通過塊的方式存到本地blockstore中。

在ipfs的安裝目錄的blocks目錄下保存了當前本地節點所存儲的所有的塊數據，具體有沒有對數據加密，我也沒有仔細去看

Ps:我去看了一下，并沒有加密，原文存儲，這里需要批評一下…

首先，add的入口在core/commands/文件，這是一個命令行工具，主要作用是提供交互以及命令的一下定義和相關配置對應的不

同功能的解析，這里真的只是解析，然后保存到AddedObject這個對象中，這個對象的作用就是當上傳是的關于文件的一下配置信息，和

一些操作。

type AddedObject struct {

Name string

Hash string `json:",omitempty"`

Bytes int64 `json:",omitempty"`

Size string `json:",omitempty"`

VID string `json:",omitempty"`

VersionInfo *nInfo

}

然后，通過下面這種看起來很奇怪的方式去上傳文件的數據量，具體后面的塊生成部分我們不去探討，這里只看是怎么從本地讀取到節點，

并將這個流送入塊中的

其實下面做的事非常簡單，先定義好addAllAndPin這個方法，這個方法最主要的作用就是對文件路徑進行遍歷，也就是我們在命令行輸入

的路徑，讀取文件內容，通過e(file)將文件寫入到下一步

而下面的協程用于監聽上傳是否完成，是否有錯誤。并將錯誤信息丟入errCh管道，并且關閉output這個管道，

作用在于這兩個管道被用于向控制臺輸出。output是輸出上傳情況，上傳完成后的塊hash等，errCh就是錯誤信息

addAllAndPin := func(f ) error {

// Iterate over each top-level file and add individually. Otherwi the

// single f is treated as a directory, affecting hidden file

// mantics.

for {

file, err := le()

if err == {

// Finished the list of files.

break

} el if err != nil {

return err

}

if err := e(file); err != nil {

return err

}

}

// copy intermediary nodes from editor to our actual dagrvice

_, err := ze()

if err != nil {

return err

}

if hash {

return nil

}

return t()

}

errCh := make(chan error)

go func() {

var err error

defer func() { errCh <- err }()

defer clo(outChan)

err = addAllAndPin()

}()

defer ()

err = (outChan)

if err != nil {

(err)

return

}

err = <-errCh

if err != nil {

or(err, mal)

}

下面進入具體上傳工作的函數，也就是e(file)，fileAddrer是上面生成一個AddedObject這個結構體的對象，它有一些工

具方法，AddFile就是用于上傳的對外接口，在core/coreunix/文件中

func (adder *Adder) AddFile(file ) error {

if {

er = k()

}

defer func() {

if er != nil {

()

}

}()

return e(file, fal, nil)

}

主要就是幾個鎖的設置，繼續調用內部的addFile方法，到這里其實就以及開始上傳了，后面的代碼就不分析了，有興趣的小伙伴可以自己

去看一下

下面是命令行入口的全部內容

package commands

import (

“errors”

“fmt”

“io”

“os”

“strings”

//塊服務提供的接口

blockrvice “”

//核心api

core “”

//add的一些工具方法和結構

“”

//文件存儲接口

filestore “”

//dag服務接口

dag “”

//提供一個新的線程安全的dag

dagtest “”

//一個可變IPFS文件系統的內存模型

mfs “”

//文件系統格式

ft “”

//控制臺入口工具包 以下都是工具包

cmds “gx/ipfs/QmNueRyPRQiV7PUEpnP4GgGLuK1rKQLaRW7sfPvUetYig1/go-ipfs-cmds”

mh “gx/ipfs/QmPnFwZ2JXKnXgMw8CdBPxn7FWh6LLdjUjxV1fKHuJnkr8/go-multihash”

pb “gx/ipfs/QmPtj12fdwuAqj9sBSTNUxBNu8kCGNp8b3o8yUzMm5GHpq/pb”

offline “gx/ipfs/QmS6mo1dPpHdYsVkm27BRZDLxpKBCiJKUH8fHX15XFfMez/go-ipfs-exchange-offline”

bstore “gx/ipfs/QmadMhXJLHMFjpRmh85XjpmVDkEtQpNYEZNRpWRvYVLrvb/go-ipfs-blockstore”

cmdkit “gx/ipfs/QmdE4gMduCKCGAcczM2F5ioYDfdeKuPix138wrES1YSr7f/go-ipfs-cmdkit”

files “gx/ipfs/QmdE4gMduCKCGAcczM2F5ioYDfdeKuPix138wrES1YSr7f/go-ipfs-cmdkit/files”

)

//限制深度 深度達到上限

// ErrDepthLimitExceeded indicates that the max depth has been exceeded.

var ErrDepthLimitExceeded = (“depth limit exceeded”)

//構建命令選項參數常量

const (

quietOptionName = “quiet”

quieterOptionName = “quieter”

silentOptionName = “silent”

progressOptionName = “progress”

trickleOptionName = “trickle”

wrapOptionName = “wrap-with-directory”

hiddenOptionName = “hidden”

onlyHashOptionName = “only-hash”

chunkerOptionName = “chunker”

pinOptionName = “pin”

rawLeavesOptionName = “raw-leaves”

noCopyOptionName = “nocopy”

fstoreCacheOptionName = “fscache”

cidVersionOptionName = “cid-version”

hashOptionName = “hash”

)

//管道上限

const adderOutChanSize = 8

//構建一個命令

var AddCmd = &d{

//命令對應的幫助信息

Helptext: xt{

Tagline: “Add a file or directory to ipfs.”,

ShortDescription:

Adds contents of to ipfs. U -r to add directories (recursively).

,

LongDescription: `

Adds contents of to ipfs. U -r to add directories.

Note that directories are added recursively, to form the ipfs

MerkleDAG.

The wrap option, ‘-w’, wraps the file (or files, if using the

recursive option) in a directory. This directory contains only

the files which have been added, and means that the file retains

its filename. For example:

ipfs add

added QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH

ipfs add -w

added QmbFMke1KXqnYyBBWxB74N4c5SBnJMVAiMNRcGu6x1AwQH

added QmaG4FuMqEBnQNn3C8XJ5bpW8kLs7zq2ZXgHptJHbKDDVx

You can now refer to the added file in a gateway, like so:

/ipfs/QmaG4FuMqEBnQNn3C8XJ5bpW8kLs7zq2ZXgHptJHbKDDVx/

The chunker option, ‘-s’, specifies the chunking strategy that dictates

how to break files into blocks. Blocks with same content can

be deduplicated. The default is a fixed block size of

256 * 1024 bytes, ‘size-262144’. Alternatively, you can u the

rabin chunker for content defined chunking by specifying

rabin-[min]-[avg]-[max] (where min/avg/max refer to the resulting

chunk sizes). Using other chunking strategies will produce

different hashes for the same file.

ipfs add --chunker=size-2048

added QmafrLBfzRLV4XSH1XcaMMeaXEUhDJjmtDfsYU95TrWG87

ipfs add --chunker=rabin-512-1024-2048

added Qmf1hDN65tR55Ubh2RN1FPxr69xq3giVBz1KApsresY8Gn

You can now check what blocks have been created by:

ipfs object links QmafrLBfzRLV4XSH1XcaMMeaXEUhDJjmtDfsYU95TrWG87

QmY6yj1GrmExDXoosVE3aSPxdMNYr6aKuw3nA8LoWPRS 2059

Qmf7ZQeSxq2fJVJbCmgTrLLVN9tDR9Wy5k75DxQKuz5Gyt 1195

ipfs object links Qmf1hDN65tR55Ubh2RN1FPxr69xq3giVBz1KApsresY8Gn

QmY6yj1GrmExDXoosVE3aSPxdMNYr6aKuw3nA8LoWPRS 2059

QmerURi9k4XzKCaaPbsK6BL5pMEjF7PGphjDvkkjDtsVf3 868

QmQB28iwSriSUSMqG2nXDTLtdPHgWb4rebBrU7Q1j4vxPv 338

`,

},

//命令對應參數格式

Arguments: []nt{

g(“path”, true, true, “The path to a file to be added to ipfs.”).EnableRecursive().EnableStdin(),

},

//命令參數可選項設置

Options: []{

//注意所有帶有experimental的命令選項都是實驗的部分需要在配置文件中啟用如果需要使用測試的話

RecursivePath, // a builtin option that allows recursive paths (-r, --recursive)

tion(quietOptionName, “q”, “Write minimal output.”),

tion(quieterOptionName, “Q”, “Write only final hash.”),

tion(silentOptionName, “Write no output.”),

tion(progressOptionName, “p”, “Stream progress data.”),

tion(trickleOptionName, “t”, “U trickle-dag format for dag generation.”),

tion(onlyHashOptionName, “n”, “Only chunk and hash - do not write to disk.”),

tion(wrapOptionName, “w”, “Wrap files with a directory object.”),

tion(hiddenOptionName, “H”, “Include files that are hidden. Only takes effect on recursive add.”),

Option(chunkerOptionName, “s”, “Chunking algorithm, size-[bytes] or rabin-[min]-[avg]-

[max]”).WithDefault(“size-262144”),

tion(pinOptionName, “Pin this object when adding.”).WithDefault(true),

tion(rawLeavesOptionName, “U raw blocks for leaf nodes. (experimental)”),

tion(noCopyOptionName, “Add the file using filestore. Implies raw-leaves. (experimental)”),

tion(fstoreCacheOptionName, “Check the filestore for pre-existing blocks. (experimental)”),

ion(cidVersionOptionName, “CID version. Defaults to 0 unless an option that depends on CIDv1 is

pasd. (experimental)”),

Option(hashOptionName, “Hash function to u. Implies CIDv1 if not sha2-256.

(experimental)”).WithDefault(“sha2-256”),

},

//設置命令默認選項的默認值，節點啟動時運行

PreRun: func(req *t, env nment) error {

quiet, _ := s[quietOptionName].(bool)

quieter, _ := s[quieterOptionName].(bool)

quiet = quiet || quieter

silent, _ := s[silentOptionName].(bool)

if quiet || silent {

return nil

}

// ipfs cli progress bar defaults to true unless quiet or silent is ud

_, found := s[progressOptionName].(bool)

if !found {

s[progressOptionName] = true

}

return nil

},

//在控制臺命令時調用run

Run: func(req *t, res Emitter, env nment) {

//獲取IPFSNode的配置信息

n, err := GetNode(env)

if err != nil {

or(err, mal)

return

}

//獲取IPFS全局配置文件配置信息

cfg, err := ()

if err != nil {

or(err, mal)

return

}

// check if repo will exceed storage limit if added

// TODO: this doesn’t handle the ca if the hashed file is already in blocks (deduplicated)

// TODO: conditional GC is disabled due to it is somehow not possible to pass the size to the daemon

//if err := ionalGC(t(), n, uint64(size)); err != nil {

// or(err, mal)

// return

//}

//將所有的命令參數對應的值強轉bool 以用來驗證該命令參數有沒有被使用

progress, _ := s[progressOptionName].(bool)

trickle, _ := s[trickleOptionName].(bool)

wrap, _ := s[wrapOptionName].(bool)

hash, _ := s[onlyHashOptionName].(bool)

hidden, _ := s[hiddenOptionName].(bool)

silent, _ := s[silentOptionName].(bool)

chunker, _ := s[chunkerOptionName].(string)

dopin, _ := s[pinOptionName].(bool)

rawblks, rbt := s[rawLeavesOptionName].(bool)

nocopy, _ := s[noCopyOptionName].(bool)

fscache, _ := s[fstoreCacheOptionName].(bool)

cidVer, cidVerSet := s[cidVersionOptionName].(int)

hashFunStr, _ := s[hashOptionName].(string)

// The arguments are subject to the following constraints.

//

// nocopy -> filestoreEnabled

// nocopy -> rawblocks

// (hash != sha2-256) -> cidv1

// NOTE: 'rawblocks -> cidv1' is missing. Legacy reasons.

// nocopy -> filestoreEnabled

//實驗方法，具體可以自行實驗

if nocopy && !oreEnabled {

or(estoreNotEnabled, ent)

return

}

//實驗方法，具體可以自行實驗

// nocopy -> rawblocks

if nocopy && !rawblks {

// fixed?

if rbt {

or(

("nocopy option requires '--raw-leaves' to be enabled as well"),

mal,

)

return

return

}

// No, satisfy mandatory constraint.

rawblks = true

}

//實驗方法，具體可以自行實驗

// (hash != "sha2-256") -> CIDv1

if hashFunStr != "sha2-256" && cidVer == 0 {

if cidVerSet {

or(

("CIDv0 only supports sha2-256"),

ent,

)

return

}

cidVer = 1

}

//實驗方法，具體可以自行實驗

// cidV1 -> raw blocks (by default)

if cidVer > 0 && !rbt {

rawblks = true

}

//實驗方法，具體可以自行實驗

prefix, err := ForCidVersion(cidVer)

if err != nil {

or(err, mal)

return

}

hashFunCode, ok := [r(hashFunStr)]

if !ok {

or(("unrecognized hash function: %s", r(hashFunStr)), mal)

return

}

= hashFunCode

th = -1

//如果使用 -n 命令參數 只寫入塊hash，不寫入磁盤

if hash {

nilnode, err := e(t(), &fg{

//TODO: need this to be true or all files

// hashed will be stored in memory!

NilRepo: true,

})

if err != nil {

or(err, mal)

return

}

n = nilnode

}

//一個可以回收的塊存儲

addblockstore := tore

//如果true 就構建一個新的可以回收的塊

if !(fscache || nocopy) {

addblockstore = lockstore(ocks, er)

}

//基本上不會被執行，可能是版本跟新代碼沒有刪除干凈

exch := ge

local, _ := s["local"].(bool)

if local {

exch = ge(addblockstore)

}

//通過塊服務構建一個新的塊啟用塊的交換策略

brv := (addblockstore, exch) // hash curity 001

//將塊交給dag服務管理

drv := Service(brv)

drv := Service(brv)

//新建輸出管道 設置長度

outChan := make(chan interface{}, adderOutChanSize)

//返回于文件添加操作的新文件對象

fileAdder, err := er(t, g, tore, drv)

if err != nil {

or(err, mal)

return

}

//為文件對象設置屬性

= outChan

r = chunker

ss = progress

= hidden

e = trickle

= wrap

= dopin

= silent

ves = rawblks

= nocopy

= &prefix

//如果使用 -n 命令參數 只寫入塊hash，不寫入磁盤

if hash {

//獲取一個新的線程安全的dag

md := ()

emptyDirNode := irNode()

// U the same prefix for the "empty" MFS root as for the file adder.

= *

mr, err := t(t, md, emptyDirNode, nil)

if err != nil {

or(err, mal)

return

}

Root(mr)

}

//構建文件上傳io

addAllAndPin := func(f ) error {

// Iterate over each top-level file and add individually. Otherwi the

// single f is treated as a directory, affecting hidden file

// mantics.

//每次讀取一個文件保存到新的文件對象fileAdder中

for {

file, err := le()

if err == {

// Finished the list of files.

break

} el if err != nil {

return err

}

if err := e(file); err != nil {

return err

}

}

// copy intermediary nodes from editor to our actual dagrvice

// Finalize方法刷新mfs根目錄并返回mfs根節點。

_, err := ze()

if err != nil {

return err

}

if hash {

return nil

}

//遞歸新的我那件對象和根節點

//遞歸新的我那件對象和根節點

//將pin節點狀態寫入后臺數據存儲。

return t()

}

errCh := make(chan error)

//開啟協程進行文件上傳

go func() {

//一個錯誤變量

var err error

//defer一個管道接收err變量 存放文件傳輸過程中出現的錯誤

defer func() { errCh <- err }()

defer clo(outChan)

//傳輸文件并返回錯誤信息

err = addAllAndPin()

}()

//關閉鏈接

defer ()

//res 錯誤

err = (outChan)

if err != nil {

(err)

return

}

//傳輸錯誤

err = < -errCh

if err != nil {

or(err, mal)

}

},

//返回執行結果到命令行

PostRun: nMap{

//實習接口方法

: func(req *t, re Emitter) Emitter {

//新建一個輸出通道標準格式

reNext, res := nResponPair(req)

//add 命令行返回值 文件hash信息 存儲管道

outChan := make(chan interface{})

//add 命令行方悔之 文件大小 存儲管道

sizeChan := make(chan int64, 1)

//通過文件對象獲取文件大小對象

sizeFile, ok := .(le)

//如果獲取文件對象成功

if ok {

// Could be slow.

go func() {

//通過文件對象獲取大小

size, err := ()

if err != nil {

gf(“error getting files size: %s”, err)

// e comment above

return

}

//將文件大小存到文件大小管道中

sizeChan <- size

}()

} el {

//不能獲得上傳文件的大小

// we don’t need to error, the progress bar just

// won’t know how big the files are

g(“cannot determine size of input file”)

}

//進度條

progressBar := func(wait chan struct{}) {

defer clo(wait)

quiet, _ := s[quietOptionName].(bool)

quieter, _ := s[quieterOptionName].(bool)

quiet = quiet || quieter

progress, _ := s[progressOptionName].(bool)

var bar *ssBar

if progress {

bar = 64(0).SetUnits(pb.U_BYTES)

Update = true

meLeft = fal

rcent = fal

=

()

}

lastFile := ""

lastHash := ""

var totalProgress, prevFiles, lastBytes int64

LOOP:

for {

for {

lect {

ca out, ok := <-outChan:

if !ok {

if quieter {

ln(, lastHash)

}

break LOOP

}

output := out.(*bject)

if len() > 0 {

lastHash =

if quieter {

continue

}

if progress {

// clear progress bar line before we print "added x" output

f(, "033[2Kr")

}

if quiet {

f(, "%sn", )

} el {

f(, "added %s %sn", , )

}

} el {

if !progress {

continue

}

if len(lastFile) == 0 {

lastFile =

}

if != lastFile || < lastBytes {

prevFiles += lastBytes

lastFile =

}

lastBytes =

delta := prevFiles + lastBytes - totalProgress

totalProgress = 64(delta)

}

if progress {

()

}

ca size := <-sizeChan:

if progress {

= size

rcent = true

r = true

meLeft = true

}

ca <-():

// don't t or print error here, that happens in the goroutine below

return

}

}

}

//控制文件上傳時的制度條顯示

go func() {

// defer order important! First clo outChan, then wait for output to finish, then clo re

defer ()

if e := (); e != nil {

if e := (); e != nil {

defer clo(outChan)

or(e, )

return

}

wait := make(chan struct{})

go progressBar(wait)

defer func() { <-wait }()

defer clo(outChan)

for {

v, err := ()

if !Error(err, res, re) {

break

}

lect {

ca outChan <- v:

ca <-():

or((), mal)

return

}

}

}()

return reNext

},

},

//添加一個object對象

Type: bject{},

}

-