oyd
/
minio
6
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

New donut frame implemented

Browse Source
master
Frederick F. Kautz IV 10 years ago
parent 9744efc725
commit 58a04ee831
3 changed files with 217 additions and 239 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 214
      pkg/storage/donut/v1/donut.go
  2. 120
      pkg/storage/donut/v1/donut_gen_v1/donut_gen.go
  3. 120
      pkg/storage/donut/v1/donut_test.go

214
pkg/storage/donut/v1/donut.go
Unescape View File

@ -19,15 +19,10 @@ package v1
import ( import (
"bytes" "bytes"
"encoding/binary" "encoding/binary"
"encoding/gob"
"errors"
"io" "io"
"io/ioutil"
"os"
"sync"
"github.com/minio-io/minio/pkg/storage/erasure"
"github.com/minio-io/minio/pkg/utils/checksum/crc32c" "github.com/minio-io/minio/pkg/utils/checksum/crc32c"
"github.com/minio-io/minio/pkg/utils/crypto/sha512"
) )
/* /*
@ -55,178 +50,95 @@ var (
MagicINIM = binary.LittleEndian.Uint32([]byte{'I', 'N', 'I', 'M'}) MagicINIM = binary.LittleEndian.Uint32([]byte{'I', 'N', 'I', 'M'})
) )
type DonutFormat struct { type DonutFrameHeader struct {
BlockStart uint32 // Magic="MINI"=1229867341 MagicMINI uint32
VersionMajor uint16 VersionMajor uint16
VersionMinor uint16 VersionMinor uint16
VersionPatch uint16 VersionPatch uint16
VersionReserved uint16 VersionReserved uint16
Reserved uint64 Reserved uint64
GobHeaderLen uint32 DataLength uint64
GobHeader []byte
HeaderCrc32c uint32
BlockData uint32 // Magic="DATA"=1096040772
Data io.Reader
FooterCrc uint32
BlockLen uint64
BlockEnd uint32
}
type DonutFooter struct {
BlockLen uint64
BlockEnd uint32 // Magic="INIM"=1229867341
}
type Donut struct {
file io.ReadWriteSeeker
mutex *sync.RWMutex
} }
type Crc32c uint32
type Sha512 [sha512.Size]byte
type GobHeader struct { type DonutFrameFooter struct {
Blocks []EncodedChunk DataSha512 Sha512
Md5sum []byte OffsetToMINI uint64
EncoderParams erasure.EncoderParams MagicINIM uint32
} }
type EncodedChunk struct { type Data bytes.Buffer
Crc uint32
Length int
Offset int
}
func New(file io.ReadWriteSeeker) *Donut { func Write(target io.Writer, reader io.Reader, length uint64) error {
donut := Donut{} // write header
donut.mutex = new(sync.RWMutex) header := DonutFrameHeader{
donut.file = file MagicMINI: MagicMINI,
return &donut VersionMajor: 1,
VersionMinor: 0,
VersionPatch: 0,
VersionReserved: 0,
Reserved: 0,
DataLength: length,
} }
var headerBytes bytes.Buffer
binary.Write(&headerBytes, binary.LittleEndian, header)
headerCrc := crc32c.Sum32(headerBytes.Bytes())
func (donut *Donut) WriteGob(gobHeader GobHeader) (bytes.Buffer, error) { binary.Write(&headerBytes, binary.LittleEndian, headerCrc)
var gobBuffer bytes.Buffer binary.Write(&headerBytes, binary.LittleEndian, MagicDATA)
encoder := gob.NewEncoder(&gobBuffer) // write header
err := encoder.Encode(gobHeader) headerLen, err := io.Copy(target, &headerBytes)
if err != nil { if err != nil {
return bytes.Buffer{}, err
}
return gobBuffer, nil
}
func (donut *Donut) WriteEnd(target io.Writer, donutFormat DonutFormat) error {
var tempBuffer bytes.Buffer
if err := binary.Write(&tempBuffer, binary.LittleEndian, donutFormat.BlockLen); err != nil {
return err
}
if err := binary.Write(&tempBuffer, binary.LittleEndian, donutFormat.BlockEnd); err != nil {
return err
}
crc := crc32c.Sum32(tempBuffer.Bytes())
if err := binary.Write(target, binary.LittleEndian, crc); err != nil {
return err
}
if _, err := io.Copy(target, &tempBuffer); err != nil {
return err
}
return nil
}
func (donut *Donut) WriteData(target io.Writer, donutFormat DonutFormat) error {
var b bytes.Buffer
if count, err := io.Copy(&b, donutFormat.Data); uint64(count) != donutFormat.BlockLen || err != nil {
if err == nil {
return binary.Write(target, binary.LittleEndian, b.Bytes())
}
return errors.New("Copy failed, count incorrect.")
}
return nil
}
func (donut *Donut) WriteBegin(target io.Writer, donutFormat DonutFormat) error {
var headerBytes bytes.Buffer
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.BlockStart); err != nil {
return err
}
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.VersionMajor); err != nil {
return err return err
} }
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.VersionMinor); err != nil { // write DATA
return err // create sha512 tee
} sumReader, sumWriter := io.Pipe()
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.VersionPatch); err != nil { defer sumWriter.Close()
return err checksumChannel := make(chan checksumValue)
} go generateChecksum(sumReader, checksumChannel)
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.VersionReserved); err != nil { teeReader := io.TeeReader(reader, sumWriter)
return err _, err = io.Copy(target, teeReader)
} if err != nil {
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.Reserved); err != nil {
return err
}
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.GobHeaderLen); err != nil {
return err
}
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.GobHeader); err != nil {
return err return err
} }
crc := crc32c.Sum32(headerBytes.Bytes()) sumWriter.Close()
if err := binary.Write(&headerBytes, binary.LittleEndian, crc); err != nil { dataChecksum := <-checksumChannel
return err if dataChecksum.err != nil {
return dataChecksum.err
} }
if err := binary.Write(&headerBytes, binary.LittleEndian, donutFormat.BlockData); err != nil { // generate footer
return err frameFooter := DonutFrameFooter{
DataSha512: dataChecksum.checksum,
OffsetToMINI: length + uint64(headerLen) + uint64(80), /*footer size*/
MagicINIM: MagicINIM,
} }
io.Copy(target, &headerBytes) var frameFooterBytes bytes.Buffer
return nil binary.Write(&frameFooterBytes, binary.LittleEndian, frameFooter)
} // write footer crc
footerChecksum := crc32c.Sum32(frameFooterBytes.Bytes())
func (donut *Donut) Write(gobHeader GobHeader, object io.Reader) error { if err := binary.Write(target, binary.LittleEndian, footerChecksum); err != nil {
donut.mutex.Lock()
defer donut.mutex.Unlock()
gobBytes, err := donut.WriteGob(gobHeader)
if err != nil {
return err return err
} }
// write write footer
// Create bytes buffer representing the new object _, err = io.Copy(target, &frameFooterBytes)
donutFormat := DonutFormat{
BlockStart: MagicMINI,
VersionMajor: 1,
VersionMinor: 0,
VersionPatch: 0,
VersionReserved: 0,
Reserved: 0,
GobHeaderLen: uint32(gobBytes.Len()),
GobHeader: gobBytes.Bytes(),
BlockData: MagicDATA,
Data: object,
BlockLen: 0,
BlockEnd: MagicINIM,
}
tempBuffer, err := ioutil.TempFile(os.TempDir(), "minio-staging")
if err != nil { if err != nil {
return err return err
} }
defer os.Remove(tempBuffer.Name()) return nil
// write header
if err := donut.WriteBegin(tempBuffer, donutFormat); err != nil {
return err
} }
// write data type checksumValue struct {
if err := donut.WriteData(tempBuffer, donutFormat); err != nil { checksum Sha512
return err err error
} }
// write footer crc func generateChecksum(reader io.Reader, c chan<- checksumValue) {
if err := donut.WriteEnd(tempBuffer, donutFormat); err != nil { checksum, err := sha512.SumStream(reader)
return err result := checksumValue{
checksum: checksum,
err: err,
} }
c <- result
// write footer
donut.file.Seek(0, 2)
tempBuffer.Seek(0, 0)
io.Copy(donut.file, tempBuffer)
return nil
} }

120
pkg/storage/donut/v1/donut_gen_v1/donut_gen.go
Unescape View File

@ -1,61 +1,65 @@
package main package main
import ( // +build ignore
"bytes" func main() {}
"fmt"
"os"
"reflect"
"github.com/minio-io/minio/pkg/storage/donut/v1" //
) //import (
// "bytes"
func main() { // "fmt"
fmt.Println("--start") // "os"
// "reflect"
file, err := os.OpenFile("hello", os.O_WRONLY|os.O_CREATE, 0666) //
if err != nil { // "github.com/minio-io/minio/pkg/storage/donut/v1"
panic(err) //)
} //
donut := v1.New(file) //func main() {
// fmt.Println("--start")
gobHeader := v1.GobHeader{} //
data := []byte("Lorem Ipsum is simply dummy text of the printing and typesetting industry. Lorem Ipsum has been the industry's standard dummy text ever since the 1500s, when an unknown printer took a galley of type and scrambled it to make a type specimen book. It has survived not only five centuries, but also the leap into electronic typesetting, remaining essentially unchanged. It was popularised in the 1960s with the release of Letraset sheets containing Lorem Ipsum passages, and more recently with desktop publishing software like Aldus PageMaker including versions of Lorem Ipsum.") // file, err := os.OpenFile("newfile", os.O_WRONLY|os.O_CREATE, 0666)
// if err != nil {
dataBuffer := bytes.NewBuffer(data) // panic(err)
err = donut.Write(gobHeader, dataBuffer) // }
if err != nil { // donut := v1.Write(file)
panic(err) //
} // gobHeader := v1.GobHeader{}
file.Close() // data := []byte("Lorem Ipsum is simply dummy text of the printing and typesetting industry. Lorem Ipsum has been the industry's standard dummy text ever since the 1500s, when an unknown printer took a galley of type and scrambled it to make a type specimen book. It has survived not only five centuries, but also the leap into electronic typesetting, remaining essentially unchanged. It was popularised in the 1960s with the release of Letraset sheets containing Lorem Ipsum passages, and more recently with desktop publishing software like Aldus PageMaker including versions of Lorem Ipsum.")
fmt.Println("--closed") //
// dataBuffer := bytes.NewBuffer(data)
fmt.Println("--verify") // err = donut.Write(gobHeader, dataBuffer)
stat, _ := os.Stat("hello") // if err != nil {
fileSize := stat.Size() // panic(err)
// }
rfile, _ := os.OpenFile("hello", os.O_RDONLY, 0666) // file.Close()
blockStart := make([]byte, 4) // fmt.Println("--closed")
blockStartCheck := []byte{'M', 'I', 'N', 'I'} //
// fmt.Println("--verify")
_, err = rfile.Read(blockStart) // stat, _ := os.Stat("newfile")
if err != nil { // fileSize := stat.Size()
panic(err) //
} // rfile, _ := os.OpenFile("newfile", os.O_RDONLY, 0666)
// blockStart := make([]byte, 4)
blockEnd := make([]byte, 4) // blockStartCheck := []byte{'M', 'I', 'N', 'I'}
start := fileSize - 4 //
blockEndCheck := []byte{'I', 'N', 'I', 'M'} // _, err = rfile.Read(blockStart)
rfile.ReadAt(blockEnd, start) // if err != nil {
rfile.Close() // panic(err)
// }
if !reflect.DeepEqual(blockStart, blockStartCheck) { //
panic("Corrupted donut file") // blockEnd := make([]byte, 4)
} // start := fileSize - 4
// blockEndCheck := []byte{'I', 'N', 'I', 'M'}
if !reflect.DeepEqual(blockEnd, blockEndCheck) { // rfile.ReadAt(blockEnd, start)
panic("Corrupted donut file") // rfile.Close()
} //
// if !reflect.DeepEqual(blockStart, blockStartCheck) {
fmt.Println("--verified") // panic("Corrupted donut file")
fmt.Println("--end") // }
} //
// if !reflect.DeepEqual(blockEnd, blockEndCheck) {
// panic("Corrupted donut file")
// }
//
// fmt.Println("--verified")
// fmt.Println("--end")
//}

120
pkg/storage/donut/v1/donut_test.go
Unescape View File

@ -18,10 +18,13 @@ package v1
import ( import (
"bytes" "bytes"
. "gopkg.in/check.v1" "crypto/sha512"
"io/ioutil" "encoding/binary"
"os"
"testing" "testing"
"github.com/minio-io/minio/pkg/utils/checksum/crc32c"
. "gopkg.in/check.v1"
) )
func Test(t *testing.T) { TestingT(t) } func Test(t *testing.T) { TestingT(t) }
@ -32,33 +35,92 @@ var _ = Suite(&MySuite{})
func (s *MySuite) TestSingleWrite(c *C) { func (s *MySuite) TestSingleWrite(c *C) {
//var b io.ReadWriteSeeker //var b io.ReadWriteSeeker
var o bytes.Buffer var testBuffer bytes.Buffer
b, err := ioutil.TempFile(os.TempDir(), "minio-donut-test") testData := "Hello, World"
defer os.RemoveAll(b.Name()) testLength := uint64(len(testData))
err := Write(&testBuffer, bytes.NewBufferString(testData), testLength)
c.Assert(err, IsNil) c.Assert(err, IsNil)
donut := New(b) testBufferLength := uint64(testBuffer.Len())
gobheader := GobHeader{}
err = donut.Write(gobheader, &o) // we test our crc here too
c.Assert(err, IsNil) headerBytes := testBuffer.Bytes()[0:28]
blockStart := make([]byte, 4) expectedCrc := crc32c.Sum32(headerBytes)
//n, _ := b.Read(blockStart) // magic mini
// b.Next(b.Len() - n) // jump ahead magicMini := make([]byte, 4)
// b.Read(blockEnd) testBuffer.Read(magicMini)
// read start c.Assert(magicMini, DeepEquals, []byte{'M', 'I', 'N', 'I'})
b.Seek(0, 0) // jump ahead
b.Read(blockStart) // major version
blockStartCheck := []byte{'M', 'I', 'N', 'I'} majorVersion := make([]byte, 2)
c.Assert(blockStart, DeepEquals, blockStartCheck) testBuffer.Read(majorVersion)
c.Assert(binary.LittleEndian.Uint16(majorVersion), DeepEquals, uint16(1))
// read block
// minor version
// read end minorVersion := make([]byte, 2)
blockEnd := make([]byte, 4) testBuffer.Read(minorVersion)
b.Seek(-int64(len(blockEnd)), 2) // jump ahead c.Assert(binary.LittleEndian.Uint16(minorVersion), DeepEquals, uint16(0))
b.Read(blockEnd)
blockEndCheck := []byte{'I', 'N', 'I', 'M'} // patch version
c.Assert(blockEnd, DeepEquals, blockEndCheck) patchVersion := make([]byte, 2)
testBuffer.Read(patchVersion)
c.Assert(binary.LittleEndian.Uint16(patchVersion), DeepEquals, uint16(0))
// reserved version
reservedVersion := make([]byte, 2)
testBuffer.Read(reservedVersion)
c.Assert(binary.LittleEndian.Uint16(reservedVersion), DeepEquals, uint16(0))
// reserved
reserved := make([]byte, 8)
testBuffer.Read(reserved)
c.Assert(binary.LittleEndian.Uint64(reserved), DeepEquals, uint64(0))
// data length
length := make([]byte, 8)
testBuffer.Read(length)
c.Assert(binary.LittleEndian.Uint64(length), DeepEquals, testLength)
// test crc
bufCrc := make([]byte, 4)
testBuffer.Read(bufCrc)
c.Assert(binary.LittleEndian.Uint32(bufCrc), DeepEquals, expectedCrc)
// magic DATA
magicData := make([]byte, 4)
testBuffer.Read(magicData)
c.Assert(magicData, DeepEquals, []byte{'D', 'A', 'T', 'A'})
// data
actualData := make([]byte, int32(testLength))
testBuffer.Read(actualData)
c.Assert(string(actualData), DeepEquals, testData)
// extract footer crc32c
actualFooterCrc := make([]byte, 4)
testBuffer.Read(actualFooterCrc)
remainingBytes := testBuffer.Bytes()
remainingSum := crc32c.Sum32(remainingBytes)
c.Assert(binary.LittleEndian.Uint32(actualFooterCrc), DeepEquals, remainingSum)
// sha512
expectedSha512 := sha512.Sum512([]byte(testData))
actualSha512 := make([]byte, 64)
testBuffer.Read(actualSha512)
c.Assert(actualSha512, DeepEquals, expectedSha512[:])
// length
actualLength := make([]byte, 8)
testBuffer.Read(actualLength)
c.Assert(testBufferLength, DeepEquals, binary.LittleEndian.Uint64(actualLength))
// magic INIM
magicInim := make([]byte, 4)
testBuffer.Read(magicInim)
c.Assert(magicInim, DeepEquals, []byte{'I', 'N', 'I', 'M'})
// ensure no extra data is in the file
c.Assert(testBuffer.Len(), Equals, 0)
} }

Write Preview
Loading…
Cancel
Save