Merge pull request #697 from chrislusf/add_version3

Add version3

8 files changed, 55 insertions, 40 deletions

diff --git a/weed/storage/needle.go b/weed/storage/needle.go
index 31bada091..f67556c72 100644
--- a/weed/storage/needle.go
+++ b/weed/storage/needle.go
@@ -39,8 +39,9 @@ type Needle struct {
 	LastModified uint64 //only store LastModifiedBytesLength bytes, which is 5 bytes to disk
 	Ttl          *TTL
 
-	Checksum CRC    `comment:"CRC32 to check integrity"`
-	Padding  []byte `comment:"Aligned to 8 bytes"`
+	Checksum   CRC    `comment:"CRC32 to check integrity"`
+	AppendAtNs uint64 `comment:"append timestamp in nano seconds"` //version3
+	Padding    []byte `comment:"Aligned to 8 bytes"`
 }
 
 func (n *Needle) String() (str string) {
diff --git a/weed/storage/needle_read_write.go b/weed/storage/needle_read_write.go
index bfd325475..aa5974842 100644
--- a/weed/storage/needle_read_write.go
+++ b/weed/storage/needle_read_write.go
@@ -23,8 +23,8 @@ const (
 	TtlBytesLength          = 2
 )
 
-func (n *Needle) DiskSize() int64 {
-	return getActualSize(n.Size)
+func (n *Needle) DiskSize(version Version) int64 {
+	return getActualSize(n.Size, version)
 }
 
 func (n *Needle) Append(w io.Writer, version Version) (size uint32, actualSize int64, err error) {
@@ -57,12 +57,12 @@ func (n *Needle) Append(w io.Writer, version Version) (size uint32, actualSize i
 			return
 		}
 		actualSize = NeedleEntrySize + int64(n.Size)
-		padding := NeedlePaddingSize - ((NeedleEntrySize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
+		padding := PaddingLength(n.Size, version)
 		util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
 		_, err = w.Write(header[0 : NeedleChecksumSize+padding])
 		return
-	case Version2:
-		header := make([]byte, NeedleEntrySize)
+	case Version2, Version3:
+		header := make([]byte, NeedleEntrySize+TimestampSize) // adding timestamp to reuse it and avoid extra allocation
 		CookieToBytes(header[0:CookieSize], n.Cookie)
 		NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
 		n.DataSize, n.NameSize, n.MimeSize = uint32(len(n.Data)), uint8(len(n.Name)), uint8(len(n.Mime))
@@ -88,7 +88,7 @@ func (n *Needle) Append(w io.Writer, version Version) (size uint32, actualSize i
 		}
 		size = n.DataSize
 		util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
-		if _, err = w.Write(header); err != nil {
+		if _, err = w.Write(header[0:NeedleEntrySize]); err != nil {
 			return
 		}
 		if n.DataSize > 0 {
@@ -143,23 +143,29 @@ func (n *Needle) Append(w io.Writer, version Version) (size uint32, actualSize i
 				}
 			}
 		}
-		padding := NeedlePaddingSize - ((NeedleEntrySize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
+		padding := PaddingLength(n.Size, version)
 		util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
-		_, err = w.Write(header[0 : NeedleChecksumSize+padding])
+		if version == Version2 {
+			_, err = w.Write(header[0 : NeedleChecksumSize+padding])
+		} else {
+			// version3
+			util.Uint64toBytes(header[NeedleChecksumSize:NeedleChecksumSize+TimestampSize], n.AppendAtNs)
+			_, err = w.Write(header[0 : NeedleChecksumSize+TimestampSize+padding])
+		}
 
-		return n.DataSize, getActualSize(n.Size), err
+		return n.DataSize, getActualSize(n.Size, version), err
 	}
 	return 0, 0, fmt.Errorf("Unsupported Version! (%d)", version)
 }
 
-func ReadNeedleBlob(r *os.File, offset int64, size uint32) (dataSlice []byte, err error) {
-	dataSlice = make([]byte, int(getActualSize(size)))
+func ReadNeedleBlob(r *os.File, offset int64, size uint32, version Version) (dataSlice []byte, err error) {
+	dataSlice = make([]byte, int(getActualSize(size, version)))
 	_, err = r.ReadAt(dataSlice, offset)
 	return dataSlice, err
 }
 
 func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version) (err error) {
-	bytes, err := ReadNeedleBlob(r, offset, size)
+	bytes, err := ReadNeedleBlob(r, offset, size, version)
 	if err != nil {
 		return err
 	}
@@ -170,7 +176,7 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version
 	switch version {
 	case Version1:
 		n.Data = bytes[NeedleEntrySize : NeedleEntrySize+size]
-	case Version2:
+	case Version2, Version3:
 		n.readNeedleDataVersion2(bytes[NeedleEntrySize : NeedleEntrySize+int(n.Size)])
 	}
 	if size == 0 {
@@ -182,6 +188,10 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version
 		return errors.New("CRC error! Data On Disk Corrupted")
 	}
 	n.Checksum = newChecksum
+	if version == Version3 {
+		tsOffset := NeedleEntrySize + size + NeedleChecksumSize
+		n.AppendAtNs = util.BytesToUint64(bytes[tsOffset : tsOffset+TimestampSize])
+	}
 	return nil
 }
 
@@ -237,7 +247,7 @@ func (n *Needle) readNeedleDataVersion2(bytes []byte) {
 
 func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bodyLength int64, err error) {
 	n = new(Needle)
-	if version == Version1 || version == Version2 {
+	if version == Version1 || version == Version2 || version == Version3 {
 		bytes := make([]byte, NeedleEntrySize)
 		var count int
 		count, err = r.ReadAt(bytes, offset)
@@ -245,12 +255,26 @@ func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bod
 			return nil, 0, err
 		}
 		n.ParseNeedleHeader(bytes)
-		padding := NeedlePaddingSize - ((n.Size + NeedleEntrySize + NeedleChecksumSize) % NeedlePaddingSize)
-		bodyLength = int64(n.Size) + NeedleChecksumSize + int64(padding)
+		bodyLength = NeedleBodyLength(n.Size, version)
 	}
 	return
 }
 
+func PaddingLength(needleSize uint32, version Version) uint32 {
+	if version == Version3 {
+		// this is same value as version2, but just listed here for clarity
+		return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize + TimestampSize) % NeedlePaddingSize)
+	}
+	return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize) % NeedlePaddingSize)
+}
+
+func NeedleBodyLength(needleSize uint32, version Version) int64 {
+	if version == Version3 {
+		return int64(needleSize) + NeedleChecksumSize + TimestampSize + int64(PaddingLength(needleSize, version))
+	}
+	return int64(needleSize) + NeedleChecksumSize + int64(PaddingLength(needleSize, version))
+}
+
 //n should be a needle already read the header
 //the input stream will read until next file entry
 func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength int64) (err error) {
@@ -265,7 +289,7 @@ func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyL
 		}
 		n.Data = bytes[:n.Size]
 		n.Checksum = NewCRC(n.Data)
-	case Version2:
+	case Version2, Version3:
 		bytes := make([]byte, bodyLength)
 		if _, err = r.ReadAt(bytes, offset); err != nil {
 			return
diff --git a/weed/storage/types/needle_types.go b/weed/storage/types/needle_types.go
index 8a2054fc5..ce4e601e4 100644
--- a/weed/storage/types/needle_types.go
+++ b/weed/storage/types/needle_types.go
@@ -14,6 +14,7 @@ const (
 	OffsetSize            = 4
 	SizeSize              = 4 // uint32 size
 	NeedleEntrySize       = NeedleIdSize + OffsetSize + SizeSize
+	TimestampSize         = 8 // int64 size
 	NeedlePaddingSize     = 8
 	MaxPossibleVolumeSize = 4 * 1024 * 1024 * 1024 * 8
 	TombstoneFileSize     = math.MaxUint32
diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go
index c928ae9a2..12c282be9 100644
--- a/weed/storage/volume_checking.go
+++ b/weed/storage/volume_checking.go
@@ -8,9 +8,8 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func getActualSize(size uint32) int64 {
-	padding := NeedlePaddingSize - ((NeedleEntrySize + size + NeedleChecksumSize) % NeedlePaddingSize)
-	return NeedleEntrySize + int64(size) + NeedleChecksumSize + int64(padding)
+func getActualSize(size uint32, version Version) int64 {
+	return NeedleEntrySize + NeedleBodyLength(size, version)
 }
 
 func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) error {
diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go
index 3e36bfb2e..3e711708e 100644
--- a/weed/storage/volume_read_write.go
+++ b/weed/storage/volume_read_write.go
@@ -203,19 +203,6 @@ func ScanVolumeFile(dirname string, collection string, id VolumeId,
 				//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 + NeedleEntrySize + NeedleChecksumSize) % NeedlePaddingSize)
-				n.Size = 0
-				rest = int64(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)
-			}
 		}
 		err = visitNeedle(n, offset)
 		if err == io.EOF {
diff --git a/weed/storage/volume_super_block.go b/weed/storage/volume_super_block.go
index 1b81788d3..6435a051f 100644
--- a/weed/storage/volume_super_block.go
+++ b/weed/storage/volume_super_block.go
@@ -33,7 +33,7 @@ type SuperBlock struct {
 
 func (s *SuperBlock) BlockSize() int {
 	switch s.version {
-	case Version2:
+	case Version2, Version3:
 		return _SuperBlockSize + int(s.extraSize)
 	}
 	return _SuperBlockSize
diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go
index 58ecc73cb..fea5e7d6e 100644
--- a/weed/storage/volume_vacuum.go
+++ b/weed/storage/volume_vacuum.go
@@ -196,7 +196,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
 			//even the needle cache in memory is hit, the need_bytes is correct
 			glog.V(4).Infof("file %d offset %d size %d", key, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
 			var needle_bytes []byte
-			needle_bytes, err = ReadNeedleBlob(oldDatFile, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
+			needle_bytes, err = ReadNeedleBlob(oldDatFile, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size, v.Version())
 			if err != nil {
 				return fmt.Errorf("ReadNeedleBlob %s key %d offset %d size %d failed: %v", oldDatFile.Name(), key, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size, err)
 			}
@@ -243,8 +243,10 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, prealloca
 
 	now := uint64(time.Now().Unix())
 
+	var version Version
 	err = ScanVolumeFile(v.dir, v.Collection, v.Id, v.needleMapKind,
 		func(superBlock SuperBlock) error {
+			version = superBlock.Version()
 			superBlock.CompactRevision++
 			_, err = dst.Write(superBlock.Bytes())
 			new_offset = int64(superBlock.BlockSize())
@@ -262,7 +264,7 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, prealloca
 				if _, _, err := n.Append(dst, v.Version()); err != nil {
 					return fmt.Errorf("cannot append needle: %s", err)
 				}
-				new_offset += n.DiskSize()
+				new_offset += n.DiskSize(version)
 				glog.V(3).Infoln("saving key", n.Id, "volume offset", offset, "=>", new_offset, "data_size", n.Size)
 			}
 			return nil
@@ -322,7 +324,7 @@ func (v *Volume) copyDataBasedOnIndexFile(dstName, idxName string) (err error) {
 			if _, _, err = n.Append(dst, v.Version()); err != nil {
 				return fmt.Errorf("cannot append needle: %s", err)
 			}
-			new_offset += n.DiskSize()
+			new_offset += n.DiskSize(v.Version())
 			glog.V(3).Infoln("saving key", n.Id, "volume offset", offset, "=>", new_offset, "data_size", n.Size)
 		}
 		return nil
diff --git a/weed/storage/volume_version.go b/weed/storage/volume_version.go
index 2e9f58aa2..fc0270c03 100644
--- a/weed/storage/volume_version.go
+++ b/weed/storage/volume_version.go
@@ -5,5 +5,6 @@ type Version uint8
 const (
 	Version1       = Version(1)
 	Version2       = Version(2)
-	CurrentVersion = Version2
+	Version3       = Version(3)
+	CurrentVersion = Version3
 )