diff --git a/Makefile b/Makefile index 6719a7bdd..68bd916be 100644 --- a/Makefile +++ b/Makefile @@ -20,3 +20,6 @@ build: deps linux: deps mkdir -p linux GOOS=linux GOARCH=amd64 go build $(GO_FLAGS) -o linux/$(BINARY) $(SOURCE_DIR) + +imports: + goimports -w $(SOURCE_DIR) \ No newline at end of file diff --git a/go/filer/flat_namespace/flat_namespace_store.go b/go/filer/flat_namespace/flat_namespace_store.go index 832b70e40..068201adf 100644 --- a/go/filer/flat_namespace/flat_namespace_store.go +++ b/go/filer/flat_namespace/flat_namespace_store.go @@ -1,7 +1,5 @@ package flat_namespace -import () - type FlatNamespaceStore interface { Put(fullFileName string, fid string) (err error) Get(fullFileName string) (fid string, err error) diff --git a/go/storage/needle_read_write.go b/go/storage/needle_read_write.go index daab6121f..751c6d797 100644 --- a/go/storage/needle_read_write.go +++ b/go/storage/needle_read_write.go @@ -238,13 +238,25 @@ func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyL } n.Data = bytes[:n.Size] n.Checksum = NewCRC(n.Data) + checksum := binary.BigEndian.Uint32(bytes[n.Size : n.Size+NeedleChecksumSize]) + + if n.Checksum.Value() != checksum { + glog.V(0).Infof("CRC error! Data On Disk Corrupted, needle id = %x", n.Id) + } case Version2: bytes := make([]byte, bodyLength) if _, err = r.ReadAt(bytes, offset); err != nil { return } n.readNeedleDataVersion2(bytes[0:n.Size]) + if n.DataSize == 0 { + return + } n.Checksum = NewCRC(n.Data) + checksum := binary.BigEndian.Uint32(bytes[n.Size : n.Size+NeedleChecksumSize]) + if n.Checksum.Value() != checksum { + glog.V(0).Infof("CRC error! Data On Disk Corrupted, needle id = %x", n.Id) + } default: err = fmt.Errorf("Unsupported Version! (%d)", version) } diff --git a/go/storage/volume_info.go b/go/storage/volume_info.go index a2f139c89..e4979c790 100644 --- a/go/storage/volume_info.go +++ b/go/storage/volume_info.go @@ -2,8 +2,9 @@ package storage import ( "fmt" - "github.com/chrislusf/seaweedfs/go/operation" "sort" + + "github.com/chrislusf/seaweedfs/go/operation" ) type VolumeInfo struct { diff --git a/go/storage/volume_replicate.go b/go/storage/volume_replicate.go new file mode 100644 index 000000000..e85bdc417 --- /dev/null +++ b/go/storage/volume_replicate.go @@ -0,0 +1,212 @@ +package storage + +import ( + "io" + "os" + "sort" + "sync" + + "encoding/binary" +) + +type DirtyData struct { + Offset int64 `comment:"Dirty data start offset"` + Size uint32 `comment:"Size of the dirty data"` +} + +type DirtyDatas []DirtyData + +func (s DirtyDatas) Len() int { return len(s) } +func (s DirtyDatas) Less(i, j int) bool { return s[i].Offset < s[j].Offset } +func (s DirtyDatas) Swap(i, j int) { s[i], s[j] = s[j], s[i] } +func (s DirtyDatas) Sort() { sort.Sort(s) } +func (s DirtyDatas) Search(offset int64) int { + return sort.Search(len(s), func(i int) bool { + v := &s[i] + return v.Offset+int64(v.Size) > offset + }) +} + +type CleanReader struct { + Dirtys DirtyDatas + DataFile *os.File + pr *io.PipeReader + pw *io.PipeWriter + mutex sync.Mutex +} + +func ScanDirtyData(indexFileContent []byte) (dirtys DirtyDatas) { + m := NewCompactMap() + for i := 0; i+16 <= len(indexFileContent); i += 16 { + bytes := indexFileContent[i : i+16] + key := binary.BigEndian.Uint64(bytes[:8]) + offset := binary.BigEndian.Uint32(bytes[8:12]) + size := binary.BigEndian.Uint32(bytes[12:16]) + k := Key(key) + if offset != 0 && size != 0 { + m.Set(k, offset, size) + } else { + if nv, ok := m.Get(k); ok { + //mark old needle data as dirty data + if int64(nv.Size)-NeedleHeaderSize > 0 { + dirtys = append(dirtys, DirtyData{ + Offset: int64(nv.Offset)*8 + NeedleHeaderSize, + Size: nv.Size, + }) + } + } + m.Delete(k) + } + } + dirtys.Sort() + return dirtys +} + +func (cr *CleanReader) Seek(offset int64, whence int) (int64, error) { + oldOff, e := cr.DataFile.Seek(0, 1) + if e != nil { + return 0, e + } + newOff, e := cr.DataFile.Seek(offset, whence) + if e != nil { + return 0, e + } + if oldOff != newOff { + cr.closePipe(true) + } + return newOff, nil +} + +func (cr *CleanReader) Size() (int64, error) { + fi, e := cr.DataFile.Stat() + if e != nil { + return 0, e + } + return fi.Size(), nil +} + +func (cdr *CleanReader) WriteTo(w io.Writer) (written int64, err error) { + off, e := cdr.DataFile.Seek(0, 1) + if e != nil { + return 0, nil + } + const ZeroBufSize = 32 * 1024 + zeroBuf := make([]byte, ZeroBufSize) + dirtyIndex := cdr.Dirtys.Search(off) + var nextDirty *DirtyData + if dirtyIndex < len(cdr.Dirtys) { + nextDirty = &cdr.Dirtys[dirtyIndex] + } + for { + if nextDirty != nil && off >= nextDirty.Offset && off < nextDirty.Offset+int64(nextDirty.Size) { + sz := nextDirty.Offset + int64(nextDirty.Size) - off + for sz > 0 { + mn := int64(ZeroBufSize) + if mn > sz { + mn = sz + } + var n int + if n, e = w.Write(zeroBuf[:mn]); e != nil { + return + } + written += int64(n) + sz -= int64(n) + off += int64(n) + } + dirtyIndex++ + if dirtyIndex < len(cdr.Dirtys) { + nextDirty = &cdr.Dirtys[dirtyIndex] + } else { + nextDirty = nil + } + if _, e = cdr.DataFile.Seek(off, 0); e != nil { + return + } + } else { + var n, sz int64 + if nextDirty != nil { + sz = nextDirty.Offset - off + } + if sz <= 0 { + // copy until eof + n, e = io.Copy(w, cdr.DataFile) + written += n + return + } + if n, e = io.CopyN(w, cdr.DataFile, sz); e != nil { + return + } + off += n + written += n + } + } + return +} + +func (cr *CleanReader) ReadAt(p []byte, off int64) (n int, err error) { + cr.Seek(off, 0) + return cr.Read(p) +} + +func (cr *CleanReader) Read(p []byte) (int, error) { + return cr.getPipeReader().Read(p) +} + +func (cr *CleanReader) Close() (e error) { + cr.closePipe(true) + return cr.DataFile.Close() +} + +func (cr *CleanReader) closePipe(lock bool) (e error) { + if lock { + cr.mutex.Lock() + defer cr.mutex.Unlock() + } + if cr.pr != nil { + if err := cr.pr.Close(); err != nil { + e = err + } + } + cr.pr = nil + if cr.pw != nil { + if err := cr.pw.Close(); err != nil { + e = err + } + } + cr.pw = nil + return e +} + +func (cr *CleanReader) getPipeReader() io.Reader { + cr.mutex.Lock() + defer cr.mutex.Unlock() + if cr.pr != nil && cr.pw != nil { + return cr.pr + } + cr.closePipe(false) + cr.pr, cr.pw = io.Pipe() + go func(pw *io.PipeWriter) { + _, e := cr.WriteTo(pw) + pw.CloseWithError(e) + }(cr.pw) + return cr.pr +} + +func (v *Volume) GetVolumeCleanReader() (cr *CleanReader, err error) { + var dirtys DirtyDatas + if indexData, e := v.nm.IndexFileContent(); e != nil { + return nil, err + } else { + dirtys = ScanDirtyData(indexData) + } + dataFile, e := os.Open(v.FileName() + ".dat") + + if e != nil { + return nil, e + } + cr = &CleanReader{ + Dirtys: dirtys, + DataFile: dataFile, + } + return +} diff --git a/go/storage/volume_replicate_test.go b/go/storage/volume_replicate_test.go new file mode 100644 index 000000000..d1da211c3 --- /dev/null +++ b/go/storage/volume_replicate_test.go @@ -0,0 +1,22 @@ +package storage + +import "testing" + +func TestDirtyDataSearch(t *testing.T) { + testData := DirtyDatas{ + {30, 20}, {106, 200}, {5, 20}, {512, 68}, {412, 50}, + } + testOffset := []int64{ + 0, 150, 480, 1024, + } + testData.Sort() + t.Logf("TestData = %v", testData) + for _, off := range testOffset { + i := testData.Search(off) + if i < testData.Len() { + t.Logf("(%d) nearest chunk[%d]: %v", off, i, testData[i]) + } else { + t.Logf("Search %d return %d ", off, i) + } + } +} diff --git a/go/storage/volume_sync.go b/go/storage/volume_sync.go index 2c72d62f0..01d59d6ae 100644 --- a/go/storage/volume_sync.go +++ b/go/storage/volume_sync.go @@ -202,6 +202,9 @@ func (v *Volume) fetchNeedle(volumeDataContentHandlerUrl string, if err != nil { return fmt.Errorf("Reading from %s error: %v", volumeDataContentHandlerUrl, err) } + if needleValue.Size != uint32(len(b)) { + return fmt.Errorf("Reading from %s error: size incorrect", volumeDataContentHandlerUrl) + } offset, err := v.AppendBlob(b) if err != nil { return fmt.Errorf("Appending volume %d error: %v", v.Id, err) diff --git a/go/weed/weed_server/volume_server.go b/go/weed/weed_server/volume_server.go index 8becdd0f1..b8472235e 100644 --- a/go/weed/weed_server/volume_server.go +++ b/go/weed/weed_server/volume_server.go @@ -57,6 +57,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, adminMux.HandleFunc("/admin/sync/status", vs.guard.WhiteList(vs.getVolumeSyncStatusHandler)) adminMux.HandleFunc("/admin/sync/index", vs.guard.WhiteList(vs.getVolumeIndexContentHandler)) adminMux.HandleFunc("/admin/sync/data", vs.guard.WhiteList(vs.getVolumeDataContentHandler)) + adminMux.HandleFunc("/admin/sync/vol_data", vs.guard.WhiteList(vs.getVolumeCleanDataHandler)) adminMux.HandleFunc("/stats/counter", vs.guard.WhiteList(statsCounterHandler)) adminMux.HandleFunc("/stats/memory", vs.guard.WhiteList(statsMemoryHandler)) adminMux.HandleFunc("/stats/disk", vs.guard.WhiteList(vs.statsDiskHandler)) diff --git a/go/weed/weed_server/volume_server_handlers_replicate.go b/go/weed/weed_server/volume_server_handlers_replicate.go new file mode 100644 index 000000000..411967326 --- /dev/null +++ b/go/weed/weed_server/volume_server_handlers_replicate.go @@ -0,0 +1,66 @@ +package weed_server + +import ( + "fmt" + "io" + "net/http" + "strconv" + + "github.com/chrislusf/seaweedfs/go/glog" + "github.com/pierrec/lz4" +) + +func (vs *VolumeServer) getVolumeCleanDataHandler(w http.ResponseWriter, r *http.Request) { + v, e := vs.getVolume("volume", r) + if v == nil { + http.Error(w, fmt.Sprintf("Not Found volume: %v", e), http.StatusBadRequest) + return + } + cr, e := v.GetVolumeCleanReader() + if e != nil { + http.Error(w, fmt.Sprintf("Get volume clean reader: %v", e), http.StatusInternalServerError) + return + } + totalSize, e := cr.Size() + if e != nil { + http.Error(w, fmt.Sprintf("Get volume size: %v", e), http.StatusInternalServerError) + return + } + w.Header().Set("Accept-Ranges", "bytes") + w.Header().Set("Content-Disposition", fmt.Sprintf(`filename="%d.dat.lz4"`, v.Id)) + + rangeReq := r.Header.Get("Range") + if rangeReq == "" { + w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10)) + w.Header().Set("Content-Encoding", "lz4") + lz4w := lz4.NewWriter(w) + if _, e = io.Copy(lz4w, cr); e != nil { + glog.V(4).Infoln("response write error:", e) + } + lz4w.Close() + return + } + ranges, e := parseRange(rangeReq, totalSize) + if e != nil { + http.Error(w, e.Error(), http.StatusRequestedRangeNotSatisfiable) + return + } + if len(ranges) != 1 { + http.Error(w, "Only support one range", http.StatusNotImplemented) + return + } + ra := ranges[0] + if _, e := cr.Seek(ra.start, 0); e != nil { + http.Error(w, e.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10)) + w.Header().Set("Content-Range", ra.contentRange(totalSize)) + w.Header().Set("Content-Encoding", "lz4") + w.WriteHeader(http.StatusPartialContent) + lz4w := lz4.NewWriter(w) + if _, e = io.CopyN(lz4w, cr, ra.length); e != nil { + glog.V(2).Infoln("response write error:", e) + } + lz4w.Close() +}