自作オブジェクトストレージ日記.part4

part4の概要

前回は1MBごとにファイルを分割し、ハッシュ値を計算してフッターとして書き込みました。今回はリード・ソロモン符号を用いて、データの冗長性を持たせ、復元可能にします。

必要なパッケージのインストール

go get -u github.com/klauspost/reedsolomon

リード・ソロモン符号の設定

データシャードとパリティシャードの数を設定します。

const (
    DataShard = 6  //リード・ソロモン符号のパラメータ データの分割数
    Parity    = 2  //リード・ソロモン符号のパラメータ　パリティーの数
)

var ErasureSetNum = DataShard + Parity //リード・ソロモン符号のパラメータ

データの分割とエンコード

データを分割し、リード・ソロモン符号を用いてパリティシャードを生成します。
datashrdsには この場合[8][1024*1024] のバイト配列を入力します。

func encode(dataShards [][]byte) ([][]byte, error) {
    enc, err := reedsolomon.New(DataShard, Parity)
    if err != nil {
        return nil, err
    }

    err = enc.Encode(dataShards)
    if err != nil {
        return nil, err
    }

    parityShards := dataShards[DataShard:]
    return parityShards, nil
}

データの書き込み

データシャードとパリティシャードをファイルに書き込みます。

for j := 0; j < ErasureSetNum; j++ {
    // outputfileを作成
    outputfile := fmt.Sprintf("%s%d_%d", outputdir, i/(1024*1024*DataShard), j)
    output, err := os.Create(outputfile)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer output.Close()

    //ヘッダーを作成
    header := Header{
        FileSize:  uint32(restSize),
        I_Counter: uint32(i / (1024 * 1024)),
        J_Counter: uint32(j),
    }

    //ヘッダーを書き込む
    err = binary.Write(output, binary.BigEndian, header)
    if err != nil {
        fmt.Println(err)
        return
    }

    var data []byte
    if j < DataShard {
        data = dataShards[j]
    } else {
        data = parityShards[j-DataShard]
    }

    //バイナリデータを書き込む
    _, err = output.Write(data)
    if err != nil {
        fmt.Println(err)
        return
    }

    //ハッシュ値を計算
    key := make([]byte, 32) // Use a proper key for HighwayHash
    highwayHash, err := highwayhash.New(key)
    if err != nil {
        fmt.Println(err)
        return
    }
    highwayHash.Write(data)
    hash := highwayHash.Sum(nil)
    if len(hash) != HashSize {
        fmt.Println("hash size is invalid")
        return
    }

    //フッターを作成
    footer := Footer{
        HashType: 0,
        Hash:     *(*[32]byte)(hash),
    }

    //フッターを書き込む
    err = binary.Write(output, binary.BigEndian, footer)
    if err != nil {
        fmt.Println(err)
        return
    }
}

データの復元

リード・ソロモン符号を用いてデータを復元します。

func main() {
    // メタデータファイルを読み込む
    metafile, err := os.Open(metafilename)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer metafile.Close()

    var objectInformation ObjectInfo
    err = binary.Read(metafile, binary.BigEndian, &objectInformation)
    if err != nil {
        fmt.Println(err)
        return
    }
    extractFileSize := objectInformation.FileSize

    buf := make([]byte, objectInformation.FileNameSize)
    _, err = metafile.Read(buf)
    if err != nil {
        fmt.Println(err)
        return
    }
    outputfilename := string(buf)

    iCounter := 0
    outputfile := outputdir + outputfilename
    output, err := os.Create(outputfile)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer output.Close()

    for {
        binaries := make([][]byte, ErasureSetNum)
        fragmentsize := -1
        for j := 0; j < ErasureSetNum; j++ {
            inputfile := fmt.Sprintf("binary/fragments/%d_%d", iCounter, j)
            input, err := os.Open(inputfile)
            if err != nil {
                binaries[j] = nil
                continue
            }
            defer input.Close()

            header := Header{}
            err = binary.Read(input, binary.BigEndian, &header)
            fragmentsize = int(header.FileSize)
            if err != nil {
                binaries[j] = nil
                continue
            }

            buf := make([]byte, 1024*1024)
            _, err = input.Read(buf)
            if err != nil {
                binaries[j] = nil
                continue
            }

            footer := Footer{}
            err = binary.Read(input, binary.BigEndian, &footer)
            if err != nil {
                binaries[j] = nil
                continue
            }

            key := make([]byte, 32)
            highwayHash, err := highwayhash.New(key)
            if err != nil {
                binaries[j] = nil
                continue
            }
            highwayHash.Write(buf)
            hash := highwayHash.Sum(nil)
            if len(hash) != HashSize {
                binaries[j] = nil
                continue
            }

            if footer.Hash != *(*[32]byte)(hash) {
                binaries[j] = nil
                continue
            }
            binaries[j] = buf
        }

        enc, err := reedsolomon.New(DataShard, Parity)
        if err != nil {
            fmt.Println(err)
            return
        }
        err = enc.Reconstruct(binaries)
        if err != nil {
            fmt.Println(err, iCounter)
            return
        }

        dataShards := make([]byte, 0)
        for j := 0; j < DataShard; j++ {
            dataShards = append(dataShards, binaries[j]...)
        }
        _, err = output.Write(dataShards[:fragmentsize])
        if err != nil {
            fmt.Println(err)
            return
        }

        iCounter++
        extractFileSize -= uint64(fragmentsize)
        if extractFileSize <= 0 {
            break
        }
    }
    fmt.Println("復元完了")
}

メタデータというものに気がついたでしょうか？前回までは必要ありませんでしたが、ファイル全体のサイズを格納するため必要になったため、新たに作りました。

type ObjectInfo struct {
	Version      [6]byte //設定ファイルのバージョン情報
	FileNameSize uint32  //ファイル名のサイズ
	FileSize     uint64  //ファイルサイズ
	Created      int64   // Unix timestamp
	UpDated      int64   // Unix timestamp
	//可変長のファイル名があとに来る
}

実行結果

Hello World!
fileSize:  5689922849
5689922849 / 5689922849 [-------------------------------------------------------------] 100.00%
分割完了
エンターキーを押して復元します

outputfilename:  output2.mp4
5689922849 / 5689922849 [-------------------------------------------------------------] 100.00%
復元完了

real    0m17.803s
user    0m15.742s
sys     0m12.701s

キー入力による待機を追加し、待機中に分割されたファイルをいくつか削除しました。削除したあとも無事復元が行えることが確認できました。あとはこれらを複数のHDDに適切に配置できればコア部分は完了です。

そろそろ行数が多くなって分かりにくくなってきたため次回はリファクタリングを行います。
part5へ