From d981eb282fc8c6b2629c86be80297ff2da17703d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 3 Jul 2016 00:10:27 -0700 Subject: [PATCH] refactor volume_read_write.go out of volume.go --- weed/storage/store.go | 4 +- weed/storage/volume.go | 222 ---------------------------- weed/storage/volume_read_write.go | 231 ++++++++++++++++++++++++++++++ weed/storage/volume_sync.go | 2 +- 4 files changed, 234 insertions(+), 225 deletions(-) create mode 100644 weed/storage/volume_read_write.go diff --git a/weed/storage/store.go b/weed/storage/store.go index 485ed437f..d6c7172e7 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -302,7 +302,7 @@ func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) { return } if MaxPossibleVolumeSize >= v.ContentSize()+uint64(size) { - size, err = v.write(n) + size, err = v.writeNeedle(n) } else { err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.volumeSizeLimit, v.ContentSize()) } @@ -320,7 +320,7 @@ func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) { } func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) { if v := s.findVolume(i); v != nil && !v.readOnly { - return v.delete(n) + return v.deleteNeedle(n) } return 0, nil } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 701d88760..801dfe267 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -1,10 +1,7 @@ package storage import ( - "bytes" - "errors" "fmt" - "io" "os" "path" "sync" @@ -76,225 +73,6 @@ func (v *Volume) NeedToReplicate() bool { return v.ReplicaPlacement.GetCopyCount() > 1 } -// isFileUnchanged checks whether this needle to write is same as last one. -// It requires serialized access in the same volume. -func (v *Volume) isFileUnchanged(n *Needle) bool { - if v.Ttl.String() != "" { - return false - } - nv, ok := v.nm.Get(n.Id) - if ok && nv.Offset > 0 { - oldNeedle := new(Needle) - err := oldNeedle.ReadData(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) - if err != nil { - glog.V(0).Infof("Failed to check updated file %v", err) - return false - } - defer oldNeedle.ReleaseMemory() - if oldNeedle.Checksum == n.Checksum && bytes.Equal(oldNeedle.Data, n.Data) { - n.DataSize = oldNeedle.DataSize - return true - } - } - return false -} - -// Destroy removes everything related to this volume -func (v *Volume) Destroy() (err error) { - if v.readOnly { - err = fmt.Errorf("%s is read-only", v.dataFile.Name()) - return - } - v.Close() - err = os.Remove(v.dataFile.Name()) - if err != nil { - return - } - err = v.nm.Destroy() - return -} - -// AppendBlob append a blob to end of the data file, used in replication -func (v *Volume) AppendBlob(b []byte) (offset int64, err error) { - if v.readOnly { - err = fmt.Errorf("%s is read-only", v.dataFile.Name()) - return - } - v.dataFileAccessLock.Lock() - defer v.dataFileAccessLock.Unlock() - if offset, err = v.dataFile.Seek(0, 2); err != nil { - glog.V(0).Infof("failed to seek the end of file: %v", err) - return - } - //ensure file writing starting from aligned positions - if offset%NeedlePaddingSize != 0 { - offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize) - if offset, err = v.dataFile.Seek(offset, 0); err != nil { - glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err) - return - } - } - v.dataFile.Write(b) - return -} - -func (v *Volume) write(n *Needle) (size uint32, err error) { - glog.V(4).Infof("writing needle %s", NewFileIdFromNeedle(v.Id, n).String()) - if v.readOnly { - err = fmt.Errorf("%s is read-only", v.dataFile.Name()) - return - } - v.dataFileAccessLock.Lock() - defer v.dataFileAccessLock.Unlock() - if v.isFileUnchanged(n) { - size = n.DataSize - glog.V(4).Infof("needle is unchanged!") - return - } - var offset int64 - if offset, err = v.dataFile.Seek(0, 2); err != nil { - glog.V(0).Infof("failed to seek the end of file: %v", err) - return - } - - //ensure file writing starting from aligned positions - if offset%NeedlePaddingSize != 0 { - offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize) - if offset, err = v.dataFile.Seek(offset, 0); err != nil { - glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err) - return - } - } - - if size, err = n.Append(v.dataFile, v.Version()); err != nil { - if e := v.dataFile.Truncate(offset); e != nil { - err = fmt.Errorf("%s\ncannot truncate %s: %v", err, v.dataFile.Name(), e) - } - return - } - nv, ok := v.nm.Get(n.Id) - if !ok || int64(nv.Offset)*NeedlePaddingSize < offset { - if err = v.nm.Put(n.Id, uint32(offset/NeedlePaddingSize), n.Size); err != nil { - glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) - } - } - if v.lastModifiedTime < n.LastModified { - v.lastModifiedTime = n.LastModified - } - return -} - -func (v *Volume) delete(n *Needle) (uint32, error) { - glog.V(4).Infof("delete needle %s", NewFileIdFromNeedle(v.Id, n).String()) - if v.readOnly { - return 0, fmt.Errorf("%s is read-only", v.dataFile.Name()) - } - v.dataFileAccessLock.Lock() - defer v.dataFileAccessLock.Unlock() - nv, ok := v.nm.Get(n.Id) - //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) - if ok { - size := nv.Size - if err := v.nm.Delete(n.Id); err != nil { - return size, err - } - if _, err := v.dataFile.Seek(0, 2); err != nil { - return size, err - } - n.Data = nil - _, err := n.Append(v.dataFile, v.Version()) - return size, err - } - return 0, nil -} - -// read fills in Needle content by looking up n.Id from NeedleMapper -func (v *Volume) readNeedle(n *Needle) (int, error) { - nv, ok := v.nm.Get(n.Id) - if !ok || nv.Offset == 0 { - return -1, errors.New("Not Found") - } - err := n.ReadData(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) - if err != nil { - return 0, err - } - bytesRead := len(n.Data) - if !n.HasTtl() { - return bytesRead, nil - } - ttlMinutes := n.Ttl.Minutes() - if ttlMinutes == 0 { - return bytesRead, nil - } - if !n.HasLastModifiedDate() { - return bytesRead, nil - } - if uint64(time.Now().Unix()) < n.LastModified+uint64(ttlMinutes*60) { - return bytesRead, nil - } - n.ReleaseMemory() - return -1, errors.New("Not Found") -} - -func ScanVolumeFile(dirname string, collection string, id VolumeId, - needleMapKind NeedleMapType, - visitSuperBlock func(SuperBlock) error, - readNeedleBody bool, - visitNeedle func(n *Needle, offset int64) error) (err error) { - var v *Volume - if v, err = loadVolumeWithoutIndex(dirname, collection, id, needleMapKind); err != nil { - return fmt.Errorf("Failed to load volume %d: %v", id, err) - } - if err = visitSuperBlock(v.SuperBlock); err != nil { - return fmt.Errorf("Failed to process volume %d super block: %v", id, err) - } - - version := v.Version() - - offset := int64(SuperBlockSize) - n, rest, e := ReadNeedleHeader(v.dataFile, version, offset) - if e != nil { - err = fmt.Errorf("cannot read needle header: %v", e) - return - } - for n != nil { - if readNeedleBody { - if err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest); err != nil { - glog.V(0).Infof("cannot read needle body: %v", err) - //err = fmt.Errorf("cannot read needle body: %v", err) - //return - } - if n.DataSize >= n.Size { - // this should come from a bug reported on #87 and #93 - // fixed in v0.69 - // remove this whole "if" clause later, long after 0.69 - oldRest, oldSize := rest, n.Size - padding := NeedlePaddingSize - ((n.Size + NeedleHeaderSize + NeedleChecksumSize) % NeedlePaddingSize) - n.Size = 0 - rest = n.Size + NeedleChecksumSize + padding - if rest%NeedlePaddingSize != 0 { - rest += (NeedlePaddingSize - rest%NeedlePaddingSize) - } - glog.V(4).Infof("Adjusting n.Size %d=>0 rest:%d=>%d %+v", oldSize, oldRest, rest, n) - } - } - if err = visitNeedle(n, offset); err != nil { - glog.V(0).Infof("visit needle error: %v", err) - } - offset += int64(NeedleHeaderSize) + int64(rest) - glog.V(4).Infof("==> new entry offset %d", offset) - if n, rest, err = ReadNeedleHeader(v.dataFile, version, offset); err != nil { - if err == io.EOF { - return nil - } - return fmt.Errorf("cannot read needle header: %v", err) - } - glog.V(4).Infof("new entry needle size:%d rest:%d", n.Size, rest) - } - - return -} - func (v *Volume) ContentSize() uint64 { return v.nm.ContentSize() } diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go new file mode 100644 index 000000000..7458b4879 --- /dev/null +++ b/weed/storage/volume_read_write.go @@ -0,0 +1,231 @@ +package storage + +import ( + "bytes" + "errors" + "fmt" + "io" + "os" + "time" + + "github.com/chrislusf/seaweedfs/weed/glog" +) + +// isFileUnchanged checks whether this needle to write is same as last one. +// It requires serialized access in the same volume. +func (v *Volume) isFileUnchanged(n *Needle) bool { + if v.Ttl.String() != "" { + return false + } + nv, ok := v.nm.Get(n.Id) + if ok && nv.Offset > 0 { + oldNeedle := new(Needle) + err := oldNeedle.ReadData(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) + if err != nil { + glog.V(0).Infof("Failed to check updated file %v", err) + return false + } + defer oldNeedle.ReleaseMemory() + if oldNeedle.Checksum == n.Checksum && bytes.Equal(oldNeedle.Data, n.Data) { + n.DataSize = oldNeedle.DataSize + return true + } + } + return false +} + +// Destroy removes everything related to this volume +func (v *Volume) Destroy() (err error) { + if v.readOnly { + err = fmt.Errorf("%s is read-only", v.dataFile.Name()) + return + } + v.Close() + err = os.Remove(v.dataFile.Name()) + if err != nil { + return + } + err = v.nm.Destroy() + return +} + +// AppendBlob append a blob to end of the data file, used in replication +func (v *Volume) AppendBlob(b []byte) (offset int64, err error) { + if v.readOnly { + err = fmt.Errorf("%s is read-only", v.dataFile.Name()) + return + } + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if offset, err = v.dataFile.Seek(0, 2); err != nil { + glog.V(0).Infof("failed to seek the end of file: %v", err) + return + } + //ensure file writing starting from aligned positions + if offset%NeedlePaddingSize != 0 { + offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize) + if offset, err = v.dataFile.Seek(offset, 0); err != nil { + glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err) + return + } + } + v.dataFile.Write(b) + return +} + +func (v *Volume) writeNeedle(n *Needle) (size uint32, err error) { + glog.V(4).Infof("writing needle %s", NewFileIdFromNeedle(v.Id, n).String()) + if v.readOnly { + err = fmt.Errorf("%s is read-only", v.dataFile.Name()) + return + } + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if v.isFileUnchanged(n) { + size = n.DataSize + glog.V(4).Infof("needle is unchanged!") + return + } + var offset int64 + if offset, err = v.dataFile.Seek(0, 2); err != nil { + glog.V(0).Infof("failed to seek the end of file: %v", err) + return + } + + //ensure file writing starting from aligned positions + if offset%NeedlePaddingSize != 0 { + offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize) + if offset, err = v.dataFile.Seek(offset, 0); err != nil { + glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err) + return + } + } + + if size, err = n.Append(v.dataFile, v.Version()); err != nil { + if e := v.dataFile.Truncate(offset); e != nil { + err = fmt.Errorf("%s\ncannot truncate %s: %v", err, v.dataFile.Name(), e) + } + return + } + nv, ok := v.nm.Get(n.Id) + if !ok || int64(nv.Offset)*NeedlePaddingSize < offset { + if err = v.nm.Put(n.Id, uint32(offset/NeedlePaddingSize), n.Size); err != nil { + glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) + } + } + if v.lastModifiedTime < n.LastModified { + v.lastModifiedTime = n.LastModified + } + return +} + +func (v *Volume) deleteNeedle(n *Needle) (uint32, error) { + glog.V(4).Infof("delete needle %s", NewFileIdFromNeedle(v.Id, n).String()) + if v.readOnly { + return 0, fmt.Errorf("%s is read-only", v.dataFile.Name()) + } + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + nv, ok := v.nm.Get(n.Id) + //fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size) + if ok { + size := nv.Size + if err := v.nm.Delete(n.Id); err != nil { + return size, err + } + if _, err := v.dataFile.Seek(0, 2); err != nil { + return size, err + } + n.Data = nil + _, err := n.Append(v.dataFile, v.Version()) + return size, err + } + return 0, nil +} + +// read fills in Needle content by looking up n.Id from NeedleMapper +func (v *Volume) readNeedle(n *Needle) (int, error) { + nv, ok := v.nm.Get(n.Id) + if !ok || nv.Offset == 0 { + return -1, errors.New("Not Found") + } + err := n.ReadData(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) + if err != nil { + return 0, err + } + bytesRead := len(n.Data) + if !n.HasTtl() { + return bytesRead, nil + } + ttlMinutes := n.Ttl.Minutes() + if ttlMinutes == 0 { + return bytesRead, nil + } + if !n.HasLastModifiedDate() { + return bytesRead, nil + } + if uint64(time.Now().Unix()) < n.LastModified+uint64(ttlMinutes*60) { + return bytesRead, nil + } + n.ReleaseMemory() + return -1, errors.New("Not Found") +} + +func ScanVolumeFile(dirname string, collection string, id VolumeId, + needleMapKind NeedleMapType, + visitSuperBlock func(SuperBlock) error, + readNeedleBody bool, + visitNeedle func(n *Needle, offset int64) error) (err error) { + var v *Volume + if v, err = loadVolumeWithoutIndex(dirname, collection, id, needleMapKind); err != nil { + return fmt.Errorf("Failed to load volume %d: %v", id, err) + } + if err = visitSuperBlock(v.SuperBlock); err != nil { + return fmt.Errorf("Failed to process volume %d super block: %v", id, err) + } + + version := v.Version() + + offset := int64(SuperBlockSize) + n, rest, e := ReadNeedleHeader(v.dataFile, version, offset) + if e != nil { + err = fmt.Errorf("cannot read needle header: %v", e) + return + } + for n != nil { + if readNeedleBody { + if err = n.ReadNeedleBody(v.dataFile, version, offset+int64(NeedleHeaderSize), rest); err != nil { + glog.V(0).Infof("cannot read needle body: %v", err) + //err = fmt.Errorf("cannot read needle body: %v", err) + //return + } + if n.DataSize >= n.Size { + // this should come from a bug reported on #87 and #93 + // fixed in v0.69 + // remove this whole "if" clause later, long after 0.69 + oldRest, oldSize := rest, n.Size + padding := NeedlePaddingSize - ((n.Size + NeedleHeaderSize + NeedleChecksumSize) % NeedlePaddingSize) + n.Size = 0 + rest = n.Size + NeedleChecksumSize + padding + if rest%NeedlePaddingSize != 0 { + rest += (NeedlePaddingSize - rest%NeedlePaddingSize) + } + glog.V(4).Infof("Adjusting n.Size %d=>0 rest:%d=>%d %+v", oldSize, oldRest, rest, n) + } + } + if err = visitNeedle(n, offset); err != nil { + glog.V(0).Infof("visit needle error: %v", err) + } + offset += int64(NeedleHeaderSize) + int64(rest) + glog.V(4).Infof("==> new entry offset %d", offset) + if n, rest, err = ReadNeedleHeader(v.dataFile, version, offset); err != nil { + if err == io.EOF { + return nil + } + return fmt.Errorf("cannot read needle header: %v", err) + } + glog.V(4).Infof("new entry needle size:%d rest:%d", n.Size, rest) + } + + return +} diff --git a/weed/storage/volume_sync.go b/weed/storage/volume_sync.go index 231ff31c2..7448b856f 100644 --- a/weed/storage/volume_sync.go +++ b/weed/storage/volume_sync.go @@ -181,7 +181,7 @@ func (v *Volume) IndexFileContent() ([]byte, error) { func (v *Volume) removeNeedle(key Key) { n := new(Needle) n.Id = uint64(key) - v.delete(n) + v.deleteNeedle(n) } // fetchNeedle fetches a remote volume needle by vid, id, offset