mirror of
https://github.com/chrislusf/seaweedfs
synced 2024-08-31 06:00:38 +02:00
Merge branch 'master' of https://github.com/chrislusf/seaweedfs
This commit is contained in:
commit
655b449270
|
@ -106,8 +106,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *storage.Needl
|
|||
|
||||
fmt.Printf("key: %d offsetFromIndex %d n.Size %d sizeFromIndex:%d\n", key, offsetFromIndex, n.Size, sizeFromIndex)
|
||||
|
||||
padding := types.NeedlePaddingSize - ((sizeFromIndex + types.NeedleEntrySize + storage.NeedleChecksumSize) % types.NeedlePaddingSize)
|
||||
rest = int64(sizeFromIndex + storage.NeedleChecksumSize + padding)
|
||||
rest = storage.NeedleBodyLength(sizeFromIndex, version)
|
||||
|
||||
func() {
|
||||
defer func() {
|
||||
|
|
|
@ -161,7 +161,7 @@ func runExport(cmd *Command, args []string) bool {
|
|||
}, true, func(n *storage.Needle, offset int64) error {
|
||||
nv, ok := needleMap.Get(n.Id)
|
||||
glog.V(3).Infof("key %d offset %d size %d disk_size %d gzip %v ok %v nv %+v",
|
||||
n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped(), ok, nv)
|
||||
n.Id, offset, n.Size, n.DiskSize(version), n.IsGzipped(), ok, nv)
|
||||
if ok && nv.Size > 0 && int64(nv.Offset)*types.NeedlePaddingSize == offset {
|
||||
if newerThanUnix >= 0 && n.HasLastModifiedDate() && n.LastModified < uint64(newerThanUnix) {
|
||||
glog.V(3).Infof("Skipping this file, as it's old enough: LastModified %d vs %d",
|
||||
|
|
|
@ -47,13 +47,15 @@ func runFix(cmd *Command, args []string) bool {
|
|||
nm := storage.NewBtreeNeedleMap(indexFile)
|
||||
defer nm.Close()
|
||||
|
||||
var version storage.Version
|
||||
vid := storage.VolumeId(*fixVolumeId)
|
||||
err = storage.ScanVolumeFile(*fixVolumePath, *fixVolumeCollection, vid,
|
||||
storage.NeedleMapInMemory,
|
||||
func(superBlock storage.SuperBlock) error {
|
||||
version = superBlock.Version()
|
||||
return nil
|
||||
}, false, func(n *storage.Needle, offset int64) error {
|
||||
glog.V(2).Infof("key %d offset %d size %d disk_size %d gzip %v", n.Id, offset, n.Size, n.DiskSize(), n.IsGzipped())
|
||||
glog.V(2).Infof("key %d offset %d size %d disk_size %d gzip %v", n.Id, offset, n.Size, n.DiskSize(version), n.IsGzipped())
|
||||
if n.Size > 0 {
|
||||
pe := nm.Put(n.Id, types.Offset(offset/types.NeedlePaddingSize), n.Size)
|
||||
glog.V(2).Infof("saved %d with error %v", n.Size, pe)
|
||||
|
|
|
@ -3,12 +3,13 @@ package s3api
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/gorilla/mux"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/gorilla/mux"
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
|
@ -3,17 +3,18 @@ package s3api
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/gorilla/mux"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"time"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/chrislusf/seaweedfs/weed/filer2"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/gorilla/mux"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
|
@ -51,7 +51,7 @@ func (vs *VolumeServer) getVolumeDataContentHandler(w http.ResponseWriter, r *ht
|
|||
}
|
||||
offset := uint32(util.ParseUint64(r.FormValue("offset"), 0))
|
||||
size := uint32(util.ParseUint64(r.FormValue("size"), 0))
|
||||
content, err := storage.ReadNeedleBlob(v.DataFile(), int64(offset)*types.NeedlePaddingSize, size)
|
||||
content, err := storage.ReadNeedleBlob(v.DataFile(), int64(offset)*types.NeedlePaddingSize, size, v.Version())
|
||||
if err != nil {
|
||||
writeJsonError(w, r, http.StatusInternalServerError, err)
|
||||
return
|
||||
|
|
|
@ -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) {
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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 {
|
||||
|
|
|
@ -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 {
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -5,5 +5,6 @@ type Version uint8
|
|||
const (
|
||||
Version1 = Version(1)
|
||||
Version2 = Version(2)
|
||||
CurrentVersion = Version2
|
||||
Version3 = Version(3)
|
||||
CurrentVersion = Version3
|
||||
)
|
||||
|
|
Loading…
Reference in a new issue