mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-05-31 15:51:39 +02:00
Code Issues Projects Releases Wiki Activity

refactoring to typed Size

Browse source 
Go is amazing with refactoring!
This commit is contained in:
Chris Lu 2020-08-18 17:04:28 -07:00
parent 618b2f6829
commit 6a92f0bc7a
34 changed files with 118 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
unmaintained/diff_volume_servers/diff_volume_servers.go
View file

@ -154,7 +154,7 @@ func getVolumeFiles(v uint32, addr string) (map[types.NeedleId]needleState, int6
var maxOffset int64
files := map[types.NeedleId]needleState{}
err = idx.WalkIndexFile(idxFile, func(key types.NeedleId, offset types.Offset, size uint32) error {
err = idx.WalkIndexFile(idxFile, func(key types.NeedleId, offset types.Offset, size Size) error {
if offset.IsZero() || size == types.TombstoneFileSize {
files[key] = needleState{
state: stateDeleted,

2
unmaintained/see_idx/see_idx.go
View file

@ -36,7 +36,7 @@ func main() {
}
defer indexFile.Close()
idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size uint32) error {
idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size Size) error {
fmt.Printf("key:%v offset:%v size:%v(%v)\n", key, offset, size, util.BytesToHumanReadable(uint64(size)))
return nil
})

4
weed/command/export.go
View file

@ -72,9 +72,9 @@ var (
func printNeedle(vid needle.VolumeId, n *needle.Needle, version needle.Version, deleted bool) {
key := needle.NewFileIdFromNeedle(vid, n).String()
size := n.DataSize
size := int32(n.DataSize)
if version == needle.Version1 {
size = n.Size
size = int32(n.Size)
}
fmt.Printf("%s\t%s\t%d\t%t\t%s\t%s\t%s\t%t\n",
key,

2
weed/server/volume_grpc_admin.go
View file

@ -199,7 +199,7 @@ func (vs *VolumeServer) VolumeNeedleStatus(ctx context.Context, req *volume_serv
resp.NeedleId = uint64(n.Id)
resp.Cookie = uint32(n.Cookie)
resp.Size = n.Size
resp.Size = uint32(n.Size)
resp.LastModified = n.LastModified
resp.Crc = n.Checksum.Value()
if n.HasTtl() {

2
weed/server/volume_grpc_batch_delete.go
View file

@ -79,7 +79,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B
resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{
FileId: fid,
Status: http.StatusAccepted,
Size: size},
Size: uint32(size)},
)
}
}

4
weed/storage/erasure_coding/ec_decoder.go
View file

@ -52,7 +52,7 @@ func FindDatFileSize(baseFileName string) (datSize int64, err error) {
return 0, fmt.Errorf("read ec volume %s version: %v", baseFileName, err)
}
err = iterateEcxFile(baseFileName, func(key types.NeedleId, offset types.Offset, size uint32) error {
err = iterateEcxFile(baseFileName, func(key types.NeedleId, offset types.Offset, size types.Size) error {
if size == types.TombstoneFileSize {
return nil
@ -88,7 +88,7 @@ func readEcVolumeVersion(baseFileName string) (version needle.Version, err error
}
func iterateEcxFile(baseFileName string, processNeedleFn func(key types.NeedleId, offset types.Offset, size uint32) error) error {
func iterateEcxFile(baseFileName string, processNeedleFn func(key types.NeedleId, offset types.Offset, size types.Size) error) error {
ecxFile, openErr := os.OpenFile(baseFileName+".ecx", os.O_RDONLY, 0644)
if openErr != nil {
return fmt.Errorf("cannot open ec index %s.ecx: %v", baseFileName, openErr)

2
weed/storage/erasure_coding/ec_encoder.go
View file

@ -294,7 +294,7 @@ func readNeedleMap(baseFileName string) (*needle_map.MemDb, error) {
defer indexFile.Close()
cm := needle_map.NewMemDb()
err = idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size uint32) error {
err = idx.WalkIndexFile(indexFile, func(key types.NeedleId, offset types.Offset, size types.Size) error {
if !offset.IsZero() && size != types.TombstoneFileSize {
cm.Set(key, offset, size)
} else {

10
weed/storage/erasure_coding/ec_locate.go
View file

@ -1,14 +1,18 @@
package erasure_coding
import (
"github.com/chrislusf/seaweedfs/weed/storage/types"
)
type Interval struct {
BlockIndex int
InnerBlockOffset int64
Size uint32
Size types.Size
IsLargeBlock bool
LargeBlockRowsCount int
}
func LocateData(largeBlockLength, smallBlockLength int64, datSize int64, offset int64, size uint32) (intervals []Interval) {
func LocateData(largeBlockLength, smallBlockLength int64, datSize int64, offset int64, size types.Size) (intervals []Interval) {
blockIndex, isLargeBlock, innerBlockOffset := locateOffset(largeBlockLength, smallBlockLength, datSize, offset)
// adding DataShardsCount*smallBlockLength to ensure we can derive the number of large block size from a shard size
@ -32,7 +36,7 @@ func LocateData(largeBlockLength, smallBlockLength int64, datSize int64, offset
intervals = append(intervals, interval)
return
}
interval.Size = uint32(blockRemaining)
interval.Size = types.Size(blockRemaining)
intervals = append(intervals, interval)
size -= interval.Size

8
weed/storage/erasure_coding/ec_test.go
View file

@ -71,7 +71,7 @@ func validateFiles(baseFileName string) error {
return nil
}
func assertSame(datFile *os.File, datSize int64, ecFiles []*os.File, offset types.Offset, size uint32) error {
func assertSame(datFile *os.File, datSize int64, ecFiles []*os.File, offset types.Offset, size types.Size) error {
data, err := readDatFile(datFile, offset, size)
if err != nil {
@ -90,7 +90,7 @@ func assertSame(datFile *os.File, datSize int64, ecFiles []*os.File, offset type
return nil
}
func readDatFile(datFile *os.File, offset types.Offset, size uint32) ([]byte, error) {
func readDatFile(datFile *os.File, offset types.Offset, size types.Size) ([]byte, error) {
data := make([]byte, size)
n, err := datFile.ReadAt(data, offset.ToAcutalOffset())
@ -103,7 +103,7 @@ func readDatFile(datFile *os.File, offset types.Offset, size uint32) ([]byte, er
return data, nil
}
func readEcFile(datSize int64, ecFiles []*os.File, offset types.Offset, size uint32) (data []byte, err error) {
func readEcFile(datSize int64, ecFiles []*os.File, offset types.Offset, size types.Size) (data []byte, err error) {
intervals := LocateData(largeBlockSize, smallBlockSize, datSize, offset.ToAcutalOffset(), size)
@ -140,7 +140,7 @@ func readOneInterval(interval Interval, ecFiles []*os.File) (data []byte, err er
return
}
func readFromOtherEcFiles(ecFiles []*os.File, ecFileIndex int, ecFileOffset int64, size uint32) (data []byte, err error) {
func readFromOtherEcFiles(ecFiles []*os.File, ecFileIndex int, ecFileOffset int64, size types.Size) (data []byte, err error) {
enc, err := reedsolomon.New(DataShardsCount, ParityShardsCount)
if err != nil {
return nil, fmt.Errorf("failed to create encoder: %v", err)

8
weed/storage/erasure_coding/ec_volume.go
View file

@ -187,7 +187,7 @@ func (ev *EcVolume) ToVolumeEcShardInformationMessage() (messages []*master_pb.V
return
}
func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.Version) (offset types.Offset, size uint32, intervals []Interval, err error) {
func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.Version) (offset types.Offset, size types.Size, intervals []Interval, err error) {
// find the needle from ecx file
offset, size, err = ev.FindNeedleFromEcx(needleId)
@ -198,16 +198,16 @@ func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.
shard := ev.Shards[0]
// calculate the locations in the ec shards
intervals = LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, DataShardsCount*shard.ecdFileSize, offset.ToAcutalOffset(), uint32(needle.GetActualSize(size, version)))
intervals = LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, DataShardsCount*shard.ecdFileSize, offset.ToAcutalOffset(), types.Size(needle.GetActualSize(size, version)))
return
}
func (ev *EcVolume) FindNeedleFromEcx(needleId types.NeedleId) (offset types.Offset, size uint32, err error) {
func (ev *EcVolume) FindNeedleFromEcx(needleId types.NeedleId) (offset types.Offset, size types.Size, err error) {
return SearchNeedleFromSortedIndex(ev.ecxFile, ev.ecxFileSize, needleId, nil)
}
func SearchNeedleFromSortedIndex(ecxFile *os.File, ecxFileSize int64, needleId types.NeedleId, processNeedleFn func(file *os.File, offset int64) error) (offset types.Offset, size uint32, err error) {
func SearchNeedleFromSortedIndex(ecxFile *os.File, ecxFileSize int64, needleId types.NeedleId, processNeedleFn func(file *os.File, offset int64) error) (offset types.Offset, size types.Size, err error) {
var key types.NeedleId
buf := make([]byte, types.NeedleMapEntrySize)
l, h := int64(0), ecxFileSize/types.NeedleMapEntrySize

2
weed/storage/erasure_coding/ec_volume_test.go
View file

@ -44,7 +44,7 @@ func TestPositioning(t *testing.T) {
fmt.Printf("offset: %d size: %d\n", offset.ToAcutalOffset(), size)
var shardEcdFileSize int64 = 1118830592 // 1024*1024*1024*3
intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, DataShardsCount*shardEcdFileSize, offset.ToAcutalOffset(), uint32(needle.GetActualSize(size, needle.CurrentVersion)))
intervals := LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, DataShardsCount*shardEcdFileSize, offset.ToAcutalOffset(), types.Size(needle.GetActualSize(size, needle.CurrentVersion)))
for _, interval := range intervals {
shardId, shardOffset := interval.ToShardIdAndOffset(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize)

9
weed/storage/idx/walk.go
View file

@ -5,12 +5,11 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/chrislusf/seaweedfs/weed/util"
)
// walks through the index file, calls fn function with each key, offset, size
// stops with the error returned by the fn function
func WalkIndexFile(r io.ReaderAt, fn func(key types.NeedleId, offset types.Offset, size uint32) error) error {
func WalkIndexFile(r io.ReaderAt, fn func(key types.NeedleId, offset types.Offset, size types.Size) error) error {
var readerOffset int64
bytes := make([]byte, types.NeedleMapEntrySize*RowsToRead)
count, e := r.ReadAt(bytes, readerOffset)
@ -22,7 +21,7 @@ func WalkIndexFile(r io.ReaderAt, fn func(key types.NeedleId, offset types.Offse
var (
key types.NeedleId
offset types.Offset
size uint32
size types.Size
i int
)
@ -43,10 +42,10 @@ func WalkIndexFile(r io.ReaderAt, fn func(key types.NeedleId, offset types.Offse
return e
}
func IdxFileEntry(bytes []byte) (key types.NeedleId, offset types.Offset, size uint32) {
func IdxFileEntry(bytes []byte) (key types.NeedleId, offset types.Offset, size types.Size) {
key = types.BytesToNeedleId(bytes[:types.NeedleIdSize])
offset = types.BytesToOffset(bytes[types.NeedleIdSize : types.NeedleIdSize+types.OffsetSize])
size = util.BytesToUint32(bytes[types.NeedleIdSize+types.OffsetSize : types.NeedleIdSize+types.OffsetSize+types.SizeSize])
size = types.BytesToSize(bytes[types.NeedleIdSize+types.OffsetSize : types.NeedleIdSize+types.OffsetSize+types.SizeSize])
return
}

2
weed/storage/needle/needle.go
View file

@ -24,7 +24,7 @@ const (
type Needle struct {
Cookie Cookie `comment:"random number to mitigate brute force lookups"`
Id NeedleId `comment:"needle id"`
Size uint32 `comment:"sum of DataSize,Data,NameSize,Name,MimeSize,Mime"`
Size Size `comment:"sum of DataSize,Data,NameSize,Name,MimeSize,Mime"`
DataSize uint32 `comment:"Data size"` //version2
Data []byte `comment:"The actual file data"`

34
weed/storage/needle/needle_read_write.go
View file

@ -28,7 +28,7 @@ func (n *Needle) DiskSize(version Version) int64 {
return GetActualSize(n.Size, version)
}
func (n *Needle) prepareWriteBuffer(version Version) ([]byte, uint32, int64, error) {
func (n *Needle) prepareWriteBuffer(version Version) ([]byte, Size, int64, error) {
writeBytes := make([]byte, 0)
@ -37,8 +37,8 @@ func (n *Needle) prepareWriteBuffer(version Version) ([]byte, uint32, int64, err
header := make([]byte, NeedleHeaderSize)
CookieToBytes(header[0:CookieSize], n.Cookie)
NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
n.Size = uint32(len(n.Data))
util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
n.Size = Size(len(n.Data))
SizeToBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
size := n.Size
actualSize := NeedleHeaderSize + int64(n.Size)
writeBytes = append(writeBytes, header...)
@ -58,12 +58,12 @@ func (n *Needle) prepareWriteBuffer(version Version) ([]byte, uint32, int64, err
}
n.DataSize, n.MimeSize = uint32(len(n.Data)), uint8(len(n.Mime))
if n.DataSize > 0 {
n.Size = 4 + n.DataSize + 1
n.Size = 4 + Size(n.DataSize) + 1
if n.HasName() {
n.Size = n.Size + 1 + uint32(n.NameSize)
n.Size = n.Size + 1 + Size(n.NameSize)
}
if n.HasMime() {
n.Size = n.Size + 1 + uint32(n.MimeSize)
n.Size = n.Size + 1 + Size(n.MimeSize)
}
if n.HasLastModifiedDate() {
n.Size = n.Size + LastModifiedBytesLength
@ -72,12 +72,12 @@ func (n *Needle) prepareWriteBuffer(version Version) ([]byte, uint32, int64, err
n.Size = n.Size + TtlBytesLength
}
if n.HasPairs() {
n.Size += 2 + uint32(n.PairsSize)
n.Size += 2 + Size(n.PairsSize)
}
} else {
n.Size = 0
}
util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
SizeToBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
writeBytes = append(writeBytes, header[0:NeedleHeaderSize]...)
if n.DataSize > 0 {
util.Uint32toBytes(header[0:4], n.DataSize)
@ -119,13 +119,13 @@ func (n *Needle) prepareWriteBuffer(version Version) ([]byte, uint32, int64, err
writeBytes = append(writeBytes, header[0:NeedleChecksumSize+TimestampSize+padding]...)
}
return writeBytes, n.DataSize, GetActualSize(n.Size, version), nil
return writeBytes, Size(n.DataSize), GetActualSize(n.Size, version), nil
}
return writeBytes, 0, 0, fmt.Errorf("Unsupported Version! (%d)", version)
}
func (n *Needle) Append(w backend.BackendStorageFile, version Version) (offset uint64, size uint32, actualSize int64, err error) {
func (n *Needle) Append(w backend.BackendStorageFile, version Version) (offset uint64, size Size, actualSize int64, err error) {
if end, _, e := w.GetStat(); e == nil {
defer func(w backend.BackendStorageFile, off int64) {
@ -154,7 +154,7 @@ func (n *Needle) Append(w backend.BackendStorageFile, version Version) (offset u
return offset, size, actualSize, err
}
func ReadNeedleBlob(r backend.BackendStorageFile, offset int64, size uint32, version Version) (dataSlice []byte, err error) {
func ReadNeedleBlob(r backend.BackendStorageFile, offset int64, size Size, version Version) (dataSlice []byte, err error) {
dataSize := GetActualSize(size, version)
dataSlice = make([]byte, int(dataSize))
@ -165,7 +165,7 @@ func ReadNeedleBlob(r backend.BackendStorageFile, offset int64, size uint32, ver
}
// ReadBytes hydrates the needle from the bytes buffer, with only n.Id is set.
func (n *Needle) ReadBytes(bytes []byte, offset int64, size uint32, version Version) (err error) {
func (n *Needle) ReadBytes(bytes []byte, offset int64, size Size, version Version) (err error) {
n.ParseNeedleHeader(bytes)
if n.Size != size {
return fmt.Errorf("entry not found: offset %d found id %x size %d, expected size %d", offset, n.Id, n.Size, size)
@ -195,7 +195,7 @@ func (n *Needle) ReadBytes(bytes []byte, offset int64, size uint32, version Vers
}
// ReadData hydrates the needle from the file, with only n.Id is set.
func (n *Needle) ReadData(r backend.BackendStorageFile, offset int64, size uint32, version Version) (err error) {
func (n *Needle) ReadData(r backend.BackendStorageFile, offset int64, size Size, version Version) (err error) {
bytes, err := ReadNeedleBlob(r, offset, size, version)
if err != nil {
return err
@ -206,7 +206,7 @@ func (n *Needle) ReadData(r backend.BackendStorageFile, offset int64, size uint3
func (n *Needle) ParseNeedleHeader(bytes []byte) {
n.Cookie = BytesToCookie(bytes[0:CookieSize])
n.Id = BytesToNeedleId(bytes[CookieSize : CookieSize+NeedleIdSize])
n.Size = util.BytesToUint32(bytes[CookieSize+NeedleIdSize : NeedleHeaderSize])
n.Size = BytesToSize(bytes[CookieSize+NeedleIdSize : NeedleHeaderSize])
}
func (n *Needle) readNeedleDataVersion2(bytes []byte) (err error) {
@ -288,7 +288,7 @@ func ReadNeedleHeader(r backend.BackendStorageFile, version Version, offset int6
return
}
func PaddingLength(needleSize uint32, version Version) uint32 {
func PaddingLength(needleSize Size, version Version) Size {
if version == Version3 {
// this is same value as version2, but just listed here for clarity
return NeedlePaddingSize - ((NeedleHeaderSize + needleSize + NeedleChecksumSize + TimestampSize) % NeedlePaddingSize)
@ -296,7 +296,7 @@ func PaddingLength(needleSize uint32, version Version) uint32 {
return NeedlePaddingSize - ((NeedleHeaderSize + needleSize + NeedleChecksumSize) % NeedlePaddingSize)
}
func NeedleBodyLength(needleSize uint32, version Version) int64 {
func NeedleBodyLength(needleSize Size, version Version) int64 {
if version == Version3 {
return int64(needleSize) + NeedleChecksumSize + TimestampSize + int64(PaddingLength(needleSize, version))
}
@ -390,6 +390,6 @@ func (n *Needle) SetHasPairs() {
n.Flags = n.Flags | FlagHasPairs
}
func GetActualSize(size uint32, version Version) int64 {
func GetActualSize(size Size, version Version) int64 {
return NeedleHeaderSize + NeedleBodyLength(size, version)
}

4
weed/storage/needle_map.go
View file

@ -19,7 +19,7 @@ const (
)
type NeedleMapper interface {
Put(key NeedleId, offset Offset, size uint32) error
Put(key NeedleId, offset Offset, size Size) error
Get(key NeedleId) (element *needle_map.NeedleValue, ok bool)
Delete(key NeedleId, offset Offset) error
Close()
@ -48,7 +48,7 @@ func (nm *baseNeedleMapper) IndexFileSize() uint64 {
return 0
}
func (nm *baseNeedleMapper) appendToIndexFile(key NeedleId, offset Offset, size uint32) error {
func (nm *baseNeedleMapper) appendToIndexFile(key NeedleId, offset Offset, size Size) error {
bytes := needle_map.ToBytes(key, offset, size)
nm.indexFileAccessLock.Lock()

16
weed/storage/needle_map/compact_map.go
View file

@ -18,7 +18,7 @@ const SectionalNeedleIdLimit = 1<<32 - 1
type SectionalNeedleValue struct {
Key SectionalNeedleId
OffsetLower OffsetLower `comment:"Volume offset"` //since aligned to 8 bytes, range is 4G*8=32G
Size uint32 `comment:"Size of the data portion"`
Size Size `comment:"Size of the data portion"`
}
type SectionalNeedleValueExtra struct {
@ -50,7 +50,7 @@ func NewCompactSection(start NeedleId) *CompactSection {
}
//return old entry size
func (cs *CompactSection) Set(key NeedleId, offset Offset, size uint32) (oldOffset Offset, oldSize uint32) {
func (cs *CompactSection) Set(key NeedleId, offset Offset, size Size) (oldOffset Offset, oldSize Size) {
cs.Lock()
if key > cs.end {
cs.end = key
@ -80,7 +80,7 @@ func (cs *CompactSection) Set(key NeedleId, offset Offset, size uint32) (oldOffs
return
}
func (cs *CompactSection) setOverflowEntry(skey SectionalNeedleId, offset Offset, size uint32) {
func (cs *CompactSection) setOverflowEntry(skey SectionalNeedleId, offset Offset, size Size) {
needleValue := SectionalNeedleValue{Key: skey, OffsetLower: offset.OffsetLower, Size: size}
needleValueExtra := SectionalNeedleValueExtra{OffsetHigher: offset.OffsetHigher}
insertCandidate := sort.Search(len(cs.overflow), func(i int) bool {
@ -125,10 +125,10 @@ func (cs *CompactSection) deleteOverflowEntry(key SectionalNeedleId) {
}
//return old entry size
func (cs *CompactSection) Delete(key NeedleId) uint32 {
func (cs *CompactSection) Delete(key NeedleId) Size {
skey := SectionalNeedleId(key - cs.start)
cs.Lock()
ret := uint32(0)
ret := Size(0)
if i := cs.binarySearchValues(skey); i >= 0 {
if cs.values[i].Size > 0 && cs.values[i].Size != TombstoneFileSize {
ret = cs.values[i].Size
@ -181,7 +181,7 @@ func NewCompactMap() *CompactMap {
return &CompactMap{}
}
func (cm *CompactMap) Set(key NeedleId, offset Offset, size uint32) (oldOffset Offset, oldSize uint32) {
func (cm *CompactMap) Set(key NeedleId, offset Offset, size Size) (oldOffset Offset, oldSize Size) {
x := cm.binarySearchCompactSection(key)
if x < 0 || (key-cm.list[x].start) > SectionalNeedleIdLimit {
// println(x, "adding to existing", len(cm.list), "sections, starting", key)
@ -204,10 +204,10 @@ func (cm *CompactMap) Set(key NeedleId, offset Offset, size uint32) (oldOffset O
// println(key, "set to section[", x, "].start", cm.list[x].start)
return cm.list[x].Set(key, offset, size)
}
func (cm *CompactMap) Delete(key NeedleId) uint32 {
func (cm *CompactMap) Delete(key NeedleId) Size {
x := cm.binarySearchCompactSection(key)
if x < 0 {
return uint32(0)
return Size(0)
}
return cm.list[x].Delete(key)
}

3
weed/storage/needle_map/compact_map_perf_test.go
View file

@ -9,7 +9,6 @@ import (
"time"
. "github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/chrislusf/seaweedfs/weed/util"
)
/*
@ -60,7 +59,7 @@ func loadNewNeedleMap(file *os.File) (*CompactMap, uint64) {
rowCount++
key := BytesToNeedleId(bytes[i : i+NeedleIdSize])
offset := BytesToOffset(bytes[i+NeedleIdSize : i+NeedleIdSize+OffsetSize])
size := util.BytesToUint32(bytes[i+NeedleIdSize+OffsetSize : i+NeedleIdSize+OffsetSize+SizeSize])
size := BytesToSize(bytes[i+NeedleIdSize+OffsetSize : i+NeedleIdSize+OffsetSize+SizeSize])
if !offset.IsZero() {
m.Set(NeedleId(key), offset, size)

10
weed/storage/needle_map/compact_map_test.go
View file

@ -49,7 +49,7 @@ func TestIssue52(t *testing.T) {
func TestCompactMap(t *testing.T) {
m := NewCompactMap()
for i := uint32(0); i < 100*batch; i += 2 {
m.Set(NeedleId(i), ToOffset(int64(i)), i)
m.Set(NeedleId(i), ToOffset(int64(i)), Size(i))
}
for i := uint32(0); i < 100*batch; i += 37 {
@ -57,7 +57,7 @@ func TestCompactMap(t *testing.T) {
}
for i := uint32(0); i < 10*batch; i += 3 {
m.Set(NeedleId(i), ToOffset(int64(i+11)), i+5)
m.Set(NeedleId(i), ToOffset(int64(i+11)), Size(i+5))
}
// for i := uint32(0); i < 100; i++ {
@ -72,7 +72,7 @@ func TestCompactMap(t *testing.T) {
if !ok {
t.Fatal("key", i, "missing!")
}
if v.Size != i+5 {
if v.Size != Size(i+5) {
t.Fatal("key", i, "size", v.Size)
}
} else if i%37 == 0 {
@ -80,7 +80,7 @@ func TestCompactMap(t *testing.T) {
t.Fatal("key", i, "should have been deleted needle value", v)
}
} else if i%2 == 0 {
if v.Size != i {
if v.Size != Size(i) {
t.Fatal("key", i, "size", v.Size)
}
}
@ -96,7 +96,7 @@ func TestCompactMap(t *testing.T) {
if v == nil {
t.Fatal("key", i, "missing")
}
if v.Size != i {
if v.Size != Size(i) {
t.Fatal("key", i, "size", v.Size)
}
}

9
weed/storage/needle_map/memdb.go
View file

@ -11,7 +11,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/storage/idx"
. "github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/chrislusf/seaweedfs/weed/util"
)
//This map uses in memory level db
@ -32,7 +31,7 @@ func NewMemDb() *MemDb {
return t
}
func (cm *MemDb) Set(key NeedleId, offset Offset, size uint32) error {
func (cm *MemDb) Set(key NeedleId, offset Offset, size Size) error {
bytes := ToBytes(key, offset, size)
@ -56,7 +55,7 @@ func (cm *MemDb) Get(key NeedleId) (*NeedleValue, bool) {
return nil, false
}
offset := BytesToOffset(data[0:OffsetSize])
size := util.BytesToUint32(data[OffsetSize : OffsetSize+SizeSize])
size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
return &NeedleValue{Key: key, Offset: offset, Size: size}, true
}
@ -67,7 +66,7 @@ func (cm *MemDb) AscendingVisit(visit func(NeedleValue) error) (ret error) {
key := BytesToNeedleId(iter.Key())
data := iter.Value()
offset := BytesToOffset(data[0:OffsetSize])
size := util.BytesToUint32(data[OffsetSize : OffsetSize+SizeSize])
size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
needle := NeedleValue{Key: key, Offset: offset, Size: size}
ret = visit(needle)
@ -105,7 +104,7 @@ func (cm *MemDb) LoadFromIdx(idxName string) (ret error) {
}
defer idxFile.Close()
return idx.WalkIndexFile(idxFile, func(key NeedleId, offset Offset, size uint32) error {
return idx.WalkIndexFile(idxFile, func(key NeedleId, offset Offset, size Size) error {
if offset.IsZero() || size == TombstoneFileSize {
return cm.Delete(key)
}

6
weed/storage/needle_map/needle_value.go
View file

@ -9,7 +9,7 @@ import (
type NeedleValue struct {
Key NeedleId
Offset Offset `comment:"Volume offset"` //since aligned to 8 bytes, range is 4G*8=32G
Size uint32 `comment:"Size of the data portion"`
Size Size `comment:"Size of the data portion"`
}
func (this NeedleValue) Less(than btree.Item) bool {
@ -21,10 +21,10 @@ func (nv NeedleValue) ToBytes() []byte {
return ToBytes(nv.Key, nv.Offset, nv.Size)
}
func ToBytes(key NeedleId, offset Offset, size uint32) []byte {
func ToBytes(key NeedleId, offset Offset, size Size) []byte {
bytes := make([]byte, NeedleIdSize+OffsetSize+SizeSize)
NeedleIdToBytes(bytes[0:NeedleIdSize], key)
OffsetToBytes(bytes[NeedleIdSize:NeedleIdSize+OffsetSize], offset)
util.Uint32toBytes(bytes[NeedleIdSize+OffsetSize:NeedleIdSize+OffsetSize+SizeSize], size)
util.Uint32toBytes(bytes[NeedleIdSize+OffsetSize:NeedleIdSize+OffsetSize+SizeSize], uint32(size))
return bytes
}

4
weed/storage/needle_map/needle_value_map.go
View file

@ -5,8 +5,8 @@ import (
)
type NeedleValueMap interface {
Set(key NeedleId, offset Offset, size uint32) (oldOffset Offset, oldSize uint32)
Delete(key NeedleId) uint32
Set(key NeedleId, offset Offset, size Size) (oldOffset Offset, oldSize Size)
Delete(key NeedleId) Size
Get(key NeedleId) (*NeedleValue, bool)
AscendingVisit(visit func(NeedleValue) error) error
}

11
weed/storage/needle_map_leveldb.go
View file

@ -15,7 +15,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/storage/needle_map"
. "github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/chrislusf/seaweedfs/weed/util"
)
type LevelDbNeedleMap struct {
@ -74,7 +73,7 @@ func generateLevelDbFile(dbFileName string, indexFile *os.File) error {
return err
}
defer db.Close()
return idx.WalkIndexFile(indexFile, func(key NeedleId, offset Offset, size uint32) error {
return idx.WalkIndexFile(indexFile, func(key NeedleId, offset Offset, size Size) error {
if !offset.IsZero() && size != TombstoneFileSize {
levelDbWrite(db, key, offset, size)
} else {
@ -92,12 +91,12 @@ func (m *LevelDbNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue, o
return nil, false
}
offset := BytesToOffset(data[0:OffsetSize])
size := util.BytesToUint32(data[OffsetSize : OffsetSize+SizeSize])
size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
return &needle_map.NeedleValue{Key: key, Offset: offset, Size: size}, true
}
func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size uint32) error {
var oldSize uint32
func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
var oldSize Size
if oldNeedle, ok := m.Get(key); ok {
oldSize = oldNeedle.Size
}
@ -109,7 +108,7 @@ func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size uint32) error {
return levelDbWrite(m.db, key, offset, size)
}
func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size uint32) error {
func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size Size) error {
bytes := needle_map.ToBytes(key, offset, size)

4
weed/storage/needle_map_memory.go
View file

@ -28,7 +28,7 @@ func LoadCompactNeedleMap(file *os.File) (*NeedleMap, error) {
}
func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
e := idx.WalkIndexFile(file, func(key NeedleId, offset Offset, size uint32) error {
e := idx.WalkIndexFile(file, func(key NeedleId, offset Offset, size Size) error {
nm.MaybeSetMaxFileKey(key)
if !offset.IsZero() && size != TombstoneFileSize {
nm.FileCounter++
@ -49,7 +49,7 @@ func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
return nm, e
}
func (nm *NeedleMap) Put(key NeedleId, offset Offset, size uint32) error {
func (nm *NeedleMap) Put(key NeedleId, offset Offset, size Size) error {
_, oldSize := nm.m.Set(NeedleId(key), offset, size)
nm.logPut(key, oldSize, size)
return nm.appendToIndexFile(key, offset, size)

12
weed/storage/needle_map_metric.go
View file

@ -18,14 +18,14 @@ type mapMetric struct {
MaximumFileKey uint64 `json:"MaxFileKey"`
}
func (mm *mapMetric) logDelete(deletedByteCount uint32) {
func (mm *mapMetric) logDelete(deletedByteCount Size) {
if mm == nil {
return
}
mm.LogDeletionCounter(deletedByteCount)
}
func (mm *mapMetric) logPut(key NeedleId, oldSize uint32, newSize uint32) {
func (mm *mapMetric) logPut(key NeedleId, oldSize Size, newSize Size) {
if mm == nil {
return
}
@ -35,14 +35,14 @@ func (mm *mapMetric) logPut(key NeedleId, oldSize uint32, newSize uint32) {
mm.LogDeletionCounter(oldSize)
}
}
func (mm *mapMetric) LogFileCounter(newSize uint32) {
func (mm *mapMetric) LogFileCounter(newSize Size) {
if mm == nil {
return
}
atomic.AddUint32(&mm.FileCounter, 1)
atomic.AddUint64(&mm.FileByteCounter, uint64(newSize))
}
func (mm *mapMetric) LogDeletionCounter(oldSize uint32) {
func (mm *mapMetric) LogDeletionCounter(oldSize Size) {
if mm == nil {
return
}
@ -97,7 +97,7 @@ func newNeedleMapMetricFromIndexFile(r *os.File) (mm *mapMetric, err error) {
buf := make([]byte, NeedleIdSize)
err = reverseWalkIndexFile(r, func(entryCount int64) {
bf = bloom.NewWithEstimates(uint(entryCount), 0.001)
}, func(key NeedleId, offset Offset, size uint32) error {
}, func(key NeedleId, offset Offset, size Size) error {
mm.MaybeSetMaxFileKey(key)
NeedleIdToBytes(buf, key)
@ -121,7 +121,7 @@ func newNeedleMapMetricFromIndexFile(r *os.File) (mm *mapMetric, err error) {
return
}
func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key NeedleId, offset Offset, size uint32) error) error {
func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key NeedleId, offset Offset, size Size) error) error {
fi, err := r.Stat()
if err != nil {
return fmt.Errorf("file %s stat error: %v", r.Name(), err)

2
weed/storage/needle_map_metric_test.go
View file

@ -15,7 +15,7 @@ func TestFastLoadingNeedleMapMetrics(t *testing.T) {
nm := NewCompactNeedleMap(idxFile)
for i := 0; i < 10000; i++ {
nm.Put(Uint64ToNeedleId(uint64(i+1)), Uint32ToOffset(uint32(0)), uint32(1))
nm.Put(Uint64ToNeedleId(uint64(i+1)), Uint32ToOffset(uint32(0)), Size(1))
if rand.Float32() < 0.2 {
nm.Delete(Uint64ToNeedleId(uint64(rand.Int63n(int64(i))+1)), Uint32ToOffset(uint32(0)))
}

2
weed/storage/needle_map_sorted_file.go
View file

@ -65,7 +65,7 @@ func (m *SortedFileNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue
}
func (m *SortedFileNeedleMap) Put(key NeedleId, offset Offset, size uint32) error {
func (m *SortedFileNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
return os.ErrInvalid
}

2
weed/storage/store.go
View file

@ -273,7 +273,7 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, fsync boo
return
}
func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (uint32, error) {
func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) {
if v := s.findVolume(i); v != nil {
if v.noWriteOrDelete {
return 0, fmt.Errorf("volume %d is read only", i)

13
weed/storage/types/needle_types.go
View file

@ -2,9 +2,10 @@ package types
import (
"fmt"
"github.com/chrislusf/seaweedfs/weed/util"
"math"
"strconv"
"github.com/chrislusf/seaweedfs/weed/util"
)
type Offset struct {
@ -12,6 +13,8 @@ type Offset struct {
OffsetLower
}
type Size uint32
type OffsetLower struct {
b3 byte
b2 byte
@ -49,3 +52,11 @@ func ParseCookie(cookieString string) (Cookie, error) {
}
return Cookie(cookie), nil
}
func BytesToSize(bytes []byte) Size {
return Size(util.BytesToUint32(bytes))
}
func SizeToBytes(bytes []byte, size Size) {
util.Uint32toBytes(bytes, uint32(size))
}

2
weed/storage/volume_checking.go
View file

@ -55,7 +55,7 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err
return
}
func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, offset int64, key NeedleId, size uint32) (lastAppendAtNs uint64, err error) {
func verifyNeedleIntegrity(datFile backend.BackendStorageFile, v needle.Version, offset int64, key NeedleId, size Size) (lastAppendAtNs uint64, err error) {
n := new(needle.Needle)
if err = n.ReadData(datFile, offset, size, v); err != nil {
return n.AppendAtNs, fmt.Errorf("read data [%d,%d) : %v", offset, offset+int64(size), err)

22
weed/storage/volume_read_write.go
View file

@ -68,9 +68,9 @@ func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) {
v.asyncRequestsChan <- request
}
func (v *Volume) syncWrite(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) {
func (v *Volume) syncWrite(n *needle.Needle) (offset uint64, size Size, isUnchanged bool, err error) {
// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
actualSize := needle.GetActualSize(uint32(len(n.Data)), v.Version())
actualSize := needle.GetActualSize(Size(len(n.Data)), v.Version())
v.dataFileAccessLock.Lock()
defer v.dataFileAccessLock.Unlock()
@ -80,7 +80,7 @@ func (v *Volume) syncWrite(n *needle.Needle) (offset uint64, size uint32, isUnch
return
}
if v.isFileUnchanged(n) {
size = n.DataSize
size = Size(n.DataSize)
isUnchanged = true
return
}
@ -120,7 +120,7 @@ func (v *Volume) syncWrite(n *needle.Needle) (offset uint64, size uint32, isUnch
return
}
func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size uint32, isUnchanged bool, err error) {
func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL {
n.SetHasTtl()
@ -132,7 +132,7 @@ func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size
} else {
asyncRequest := needle.NewAsyncRequest(n, true)
// using len(n.Data) here instead of n.Size before n.Size is populated in n.Append()
asyncRequest.ActualSize = needle.GetActualSize(uint32(len(n.Data)), v.Version())
asyncRequest.ActualSize = needle.GetActualSize(Size(len(n.Data)), v.Version())
v.asyncRequestAppend(asyncRequest)
offset, _, isUnchanged, err = asyncRequest.WaitComplete()
@ -141,10 +141,10 @@ func (v *Volume) writeNeedle2(n *needle.Needle, fsync bool) (offset uint64, size
}
}
func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, isUnchanged bool, err error) {
func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size Size, isUnchanged bool, err error) {
// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
if v.isFileUnchanged(n) {
size = n.DataSize
size = Size(n.DataSize)
isUnchanged = true
return
}
@ -183,7 +183,7 @@ func (v *Volume) doWriteRequest(n *needle.Needle) (offset uint64, size uint32, i
return
}
func (v *Volume) syncDelete(n *needle.Needle) (uint32, error) {
func (v *Volume) syncDelete(n *needle.Needle) (Size, error) {
glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
actualSize := needle.GetActualSize(0, v.Version())
v.dataFileAccessLock.Lock()
@ -213,7 +213,7 @@ func (v *Volume) syncDelete(n *needle.Needle) (uint32, error) {
return 0, nil
}
func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) {
func (v *Volume) deleteNeedle2(n *needle.Needle) (Size, error) {
// todo: delete info is always appended no fsync, it may need fsync in future
fsync := false
@ -226,11 +226,11 @@ func (v *Volume) deleteNeedle2(n *needle.Needle) (uint32, error) {
v.asyncRequestAppend(asyncRequest)
_, size, _, err := asyncRequest.WaitComplete()
return uint32(size), err
return Size(size), err
}
}
func (v *Volume) doDeleteRequest(n *needle.Needle) (uint32, error) {
func (v *Volume) doDeleteRequest(n *needle.Needle) (Size, error) {
glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
nv, ok := v.nm.Get(n.Id)
//fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size)

2
weed/storage/volume_vacuum.go
View file

@ -207,7 +207,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
type keyField struct {
offset Offset
size uint32
size Size
}
incrementedHasUpdatedIndexEntry := make(map[NeedleId]keyField)

4
weed/storage/volume_vacuum_test.go
View file

@ -117,7 +117,7 @@ func TestCompaction(t *testing.T) {
if err != nil {
t.Fatalf("read file %d: %v", i, err)
}
if infos[i-1].size != uint32(size) {
if infos[i-1].size != types.Size(size) {
t.Fatalf("read file %d size mismatch expected %d found %d", i, infos[i-1].size, size)
}
if infos[i-1].crc != n.Checksum {
@ -151,7 +151,7 @@ func doSomeWritesDeletes(i int, v *Volume, t *testing.T, infos []*needleInfo) {
}
type needleInfo struct {
size uint32
size types.Size
crc needle.CRC
}

3
weed/topology/store_replicate.go
View file

@ -14,6 +14,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/storage"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
"github.com/chrislusf/seaweedfs/weed/storage/types"
"github.com/chrislusf/seaweedfs/weed/util"
)
@ -92,7 +93,7 @@ func ReplicatedWrite(masterNode string, s *storage.Store, volumeId needle.Volume
func ReplicatedDelete(masterNode string, store *storage.Store,
volumeId needle.VolumeId, n *needle.Needle,
r *http.Request) (size uint32, err error) {
r *http.Request) (size types.Size, err error) {
//check JWT
jwt := security.GetJwt(r)

2
weed/util/chunk_cache/chunk_cache_on_disk.go
View file

@ -137,7 +137,7 @@ func (v *ChunkCacheVolume) WriteNeedle(key types.NeedleId, data []byte) error {
v.fileSize += int64(types.NeedlePaddingSize - extraSize)
}
if err := v.nm.Put(key, types.ToOffset(offset), uint32(len(data))); err != nil {
if err := v.nm.Put(key, types.ToOffset(offset), types.Size(len(data))); err != nil {
return err
}