mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-10 22:08:48 +01:00
1 parent
d1494ea786
commit
bf9c4ed033
36 files changed
+466
-622
No files matched your search
@@ -175,23 +175,3 @@ func (cm *CompactMap) Peek() {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// iterate over the keys by calling iterate on each key till error is returned
|
||||
func (cm *CompactMap) Walk(pedestrian func(*NeedleValue) error) (err error) {
|
||||
var i int
|
||||
for _, cs := range cm.list {
|
||||
for key := cs.start; key < cs.end; key++ {
|
||||
if i = cs.binarySearchValues(key); i >= 0 {
|
||||
if err = pedestrian(&cs.values[i]); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, val := range cs.overflow {
|
||||
if err = pedestrian(val); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,43 +1,43 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
"pkg/util"
|
||||
"testing"
|
||||
"log"
|
||||
"os"
|
||||
"pkg/util"
|
||||
)
|
||||
|
||||
func TestMemoryUsage(t *testing.T) {
|
||||
|
||||
indexFile, ie := os.OpenFile("sample.idx", os.O_RDWR|os.O_RDONLY, 0644)
|
||||
if ie != nil {
|
||||
log.Fatalln(ie)
|
||||
}
|
||||
LoadNewNeedleMap(indexFile)
|
||||
|
||||
indexFile, ie := os.OpenFile("sample.idx", os.O_RDWR|os.O_RDONLY, 0644)
|
||||
if ie != nil {
|
||||
log.Fatalln(ie)
|
||||
}
|
||||
LoadNewNeedleMap(indexFile)
|
||||
|
||||
}
|
||||
|
||||
func LoadNewNeedleMap(file *os.File) CompactMap {
|
||||
m := NewCompactMap()
|
||||
bytes := make([]byte, 16*1024)
|
||||
count, e := file.Read(bytes)
|
||||
if count > 0 {
|
||||
fstat, _ := file.Stat()
|
||||
log.Println("Loading index file", fstat.Name(), "size", fstat.Size())
|
||||
}
|
||||
for count > 0 && e == nil {
|
||||
for i := 0; i < count; i += 16 {
|
||||
key := util.BytesToUint64(bytes[i : i+8])
|
||||
offset := util.BytesToUint32(bytes[i+8 : i+12])
|
||||
size := util.BytesToUint32(bytes[i+12 : i+16])
|
||||
if offset > 0 {
|
||||
m.Set(Key(key), offset, size)
|
||||
} else {
|
||||
//delete(m, key)
|
||||
}
|
||||
}
|
||||
m := NewCompactMap()
|
||||
bytes := make([]byte, 16*1024)
|
||||
count, e := file.Read(bytes)
|
||||
if count > 0 {
|
||||
fstat, _ := file.Stat()
|
||||
log.Println("Loading index file", fstat.Name(), "size", fstat.Size())
|
||||
}
|
||||
for count > 0 && e == nil {
|
||||
for i := 0; i < count; i += 16 {
|
||||
key := util.BytesToUint64(bytes[i : i+8])
|
||||
offset := util.BytesToUint32(bytes[i+8 : i+12])
|
||||
size := util.BytesToUint32(bytes[i+12 : i+16])
|
||||
if offset > 0 {
|
||||
m.Set(Key(key), offset, size)
|
||||
} else {
|
||||
//delete(m, key)
|
||||
}
|
||||
}
|
||||
|
||||
count, e = file.Read(bytes)
|
||||
}
|
||||
return m
|
||||
count, e = file.Read(bytes)
|
||||
}
|
||||
return m
|
||||
}
|
||||
@@ -18,42 +18,42 @@ func TestXYZ(t *testing.T) {
|
||||
m.Set(Key(i), i+11, i+5)
|
||||
}
|
||||
|
||||
// for i := uint32(0); i < 100; i++ {
|
||||
// if v := m.Get(Key(i)); v != nil {
|
||||
// println(i, "=", v.Key, v.Offset, v.Size)
|
||||
// }
|
||||
// }
|
||||
|
||||
// for i := uint32(0); i < 100; i++ {
|
||||
// if v := m.Get(Key(i)); v != nil {
|
||||
// println(i, "=", v.Key, v.Offset, v.Size)
|
||||
// }
|
||||
// }
|
||||
|
||||
for i := uint32(0); i < 10*batch; i++ {
|
||||
v, ok := m.Get(Key(i))
|
||||
v, ok := m.Get(Key(i))
|
||||
if i%3 == 0 {
|
||||
if !ok {
|
||||
t.Fatal("key", i, "missing!")
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("key", i, "missing!")
|
||||
}
|
||||
if v.Size != i+5 {
|
||||
t.Fatal("key", i, "size", v.Size)
|
||||
}
|
||||
} else if i%37 == 0 {
|
||||
if ok && v.Size > 0 {
|
||||
if ok && v.Size > 0 {
|
||||
t.Fatal("key", i, "should have been deleted needle value", v)
|
||||
}
|
||||
} else if i%2 == 0 {
|
||||
if v.Size != i {
|
||||
if v.Size != i {
|
||||
t.Fatal("key", i, "size", v.Size)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for i := uint32(10 * batch); i < 100*batch; i++ {
|
||||
v, ok := m.Get(Key(i))
|
||||
v, ok := m.Get(Key(i))
|
||||
if i%37 == 0 {
|
||||
if ok && v.Size > 0 {
|
||||
t.Fatal("key", i, "should have been deleted needle value", v)
|
||||
}
|
||||
if ok && v.Size > 0 {
|
||||
t.Fatal("key", i, "should have been deleted needle value", v)
|
||||
}
|
||||
} else if i%2 == 0 {
|
||||
if v == nil {
|
||||
t.Fatal("key", i, "missing")
|
||||
}
|
||||
if v==nil{
|
||||
t.Fatal("key", i, "missing")
|
||||
}
|
||||
if v.Size != i {
|
||||
t.Fatal("key", i, "size", v.Size)
|
||||
}
|
||||
|
||||
@@ -98,8 +98,3 @@ func (nm *NeedleMap) Close() {
|
||||
func (nm *NeedleMap) ContentSize() uint64 {
|
||||
return nm.fileByteCounter
|
||||
}
|
||||
|
||||
// iterate through all needles using the iterator function
|
||||
func (nm *NeedleMap) Walk(pedestrian func(*NeedleValue) error) (err error) {
|
||||
return nm.m.Walk(pedestrian)
|
||||
}
|
||||
@@ -2,10 +2,10 @@ package storage
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"pkg/util"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
func (n *Needle) Append(w io.Writer, version Version) uint32 {
|
||||
@@ -62,8 +62,7 @@ func (n *Needle) Append(w io.Writer, version Version) uint32 {
|
||||
return n.Size
|
||||
}
|
||||
func (n *Needle) Read(r io.Reader, size uint32, version Version) (int, error) {
|
||||
switch version {
|
||||
case Version1:
|
||||
if version == Version1 {
|
||||
bytes := make([]byte, NeedleHeaderSize+size+NeedleChecksumSize)
|
||||
ret, e := r.Read(bytes)
|
||||
n.readNeedleHeader(bytes)
|
||||
@@ -73,7 +72,7 @@ func (n *Needle) Read(r io.Reader, size uint32, version Version) (int, error) {
|
||||
return 0, errors.New("CRC error! Data On Disk Corrupted!")
|
||||
}
|
||||
return ret, e
|
||||
case Version2:
|
||||
} else if version == Version2 {
|
||||
if size == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
@@ -96,7 +95,7 @@ func (n *Needle) Read(r io.Reader, size uint32, version Version) (int, error) {
|
||||
}
|
||||
return ret, e
|
||||
}
|
||||
return 0, fmt.Errorf("Unsupported Version! (%d)", version)
|
||||
return 0, errors.New("Unsupported Version!")
|
||||
}
|
||||
func (n *Needle) readNeedleHeader(bytes []byte) {
|
||||
n.Cookie = util.BytesToUint32(bytes[0:4])
|
||||
|
||||
@@ -1,123 +1,123 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"errors"
|
||||
)
|
||||
|
||||
type ReplicationType string
|
||||
|
||||
const (
|
||||
Copy000 = ReplicationType("000") // single copy
|
||||
Copy001 = ReplicationType("001") // 2 copies, both on the same racks, and same data center
|
||||
Copy010 = ReplicationType("010") // 2 copies, both on different racks, but same data center
|
||||
Copy100 = ReplicationType("100") // 2 copies, each on different data center
|
||||
Copy110 = ReplicationType("110") // 3 copies, 2 on different racks and local data center, 1 on different data center
|
||||
Copy200 = ReplicationType("200") // 3 copies, each on dffereint data center
|
||||
LengthRelicationType = 6
|
||||
CopyNil = ReplicationType(255) // nil value
|
||||
Copy000 = ReplicationType("000") // single copy
|
||||
Copy001 = ReplicationType("001") // 2 copies, both on the same racks, and same data center
|
||||
Copy010 = ReplicationType("010") // 2 copies, both on different racks, but same data center
|
||||
Copy100 = ReplicationType("100") // 2 copies, each on different data center
|
||||
Copy110 = ReplicationType("110") // 3 copies, 2 on different racks and local data center, 1 on different data center
|
||||
Copy200 = ReplicationType("200") // 3 copies, each on dffereint data center
|
||||
LengthRelicationType = 6
|
||||
CopyNil = ReplicationType(255) // nil value
|
||||
)
|
||||
|
||||
func NewReplicationTypeFromString(t string) (ReplicationType, error) {
|
||||
switch t {
|
||||
case "000":
|
||||
return Copy000, nil
|
||||
case "001":
|
||||
return Copy001, nil
|
||||
case "010":
|
||||
return Copy010, nil
|
||||
case "100":
|
||||
return Copy100, nil
|
||||
case "110":
|
||||
return Copy110, nil
|
||||
case "200":
|
||||
return Copy200, nil
|
||||
}
|
||||
return Copy000, errors.New("Unknown Replication Type:" + t)
|
||||
switch t {
|
||||
case "000":
|
||||
return Copy000, nil
|
||||
case "001":
|
||||
return Copy001, nil
|
||||
case "010":
|
||||
return Copy010, nil
|
||||
case "100":
|
||||
return Copy100, nil
|
||||
case "110":
|
||||
return Copy110, nil
|
||||
case "200":
|
||||
return Copy200, nil
|
||||
}
|
||||
return Copy000, errors.New("Unknown Replication Type:"+t)
|
||||
}
|
||||
func NewReplicationTypeFromByte(b byte) (ReplicationType, error) {
|
||||
switch b {
|
||||
case byte(000):
|
||||
return Copy000, nil
|
||||
case byte(001):
|
||||
return Copy001, nil
|
||||
case byte(010):
|
||||
return Copy010, nil
|
||||
case byte(100):
|
||||
return Copy100, nil
|
||||
case byte(110):
|
||||
return Copy110, nil
|
||||
case byte(200):
|
||||
return Copy200, nil
|
||||
}
|
||||
return Copy000, errors.New("Unknown Replication Type:" + string(b))
|
||||
switch b {
|
||||
case byte(000):
|
||||
return Copy000, nil
|
||||
case byte(001):
|
||||
return Copy001, nil
|
||||
case byte(010):
|
||||
return Copy010, nil
|
||||
case byte(100):
|
||||
return Copy100, nil
|
||||
case byte(110):
|
||||
return Copy110, nil
|
||||
case byte(200):
|
||||
return Copy200, nil
|
||||
}
|
||||
return Copy000, errors.New("Unknown Replication Type:"+string(b))
|
||||
}
|
||||
|
||||
func (r *ReplicationType) String() string {
|
||||
switch *r {
|
||||
case Copy000:
|
||||
return "000"
|
||||
case Copy001:
|
||||
return "001"
|
||||
case Copy010:
|
||||
return "010"
|
||||
case Copy100:
|
||||
return "100"
|
||||
case Copy110:
|
||||
return "110"
|
||||
case Copy200:
|
||||
return "200"
|
||||
}
|
||||
return "000"
|
||||
switch *r {
|
||||
case Copy000:
|
||||
return "000"
|
||||
case Copy001:
|
||||
return "001"
|
||||
case Copy010:
|
||||
return "010"
|
||||
case Copy100:
|
||||
return "100"
|
||||
case Copy110:
|
||||
return "110"
|
||||
case Copy200:
|
||||
return "200"
|
||||
}
|
||||
return "000"
|
||||
}
|
||||
func (r *ReplicationType) Byte() byte {
|
||||
switch *r {
|
||||
case Copy000:
|
||||
return byte(000)
|
||||
case Copy001:
|
||||
return byte(001)
|
||||
case Copy010:
|
||||
return byte(010)
|
||||
case Copy100:
|
||||
return byte(100)
|
||||
case Copy110:
|
||||
return byte(110)
|
||||
case Copy200:
|
||||
return byte(200)
|
||||
}
|
||||
return byte(000)
|
||||
switch *r {
|
||||
case Copy000:
|
||||
return byte(000)
|
||||
case Copy001:
|
||||
return byte(001)
|
||||
case Copy010:
|
||||
return byte(010)
|
||||
case Copy100:
|
||||
return byte(100)
|
||||
case Copy110:
|
||||
return byte(110)
|
||||
case Copy200:
|
||||
return byte(200)
|
||||
}
|
||||
return byte(000)
|
||||
}
|
||||
|
||||
func (repType ReplicationType) GetReplicationLevelIndex() int {
|
||||
switch repType {
|
||||
case Copy000:
|
||||
return 0
|
||||
case Copy001:
|
||||
return 1
|
||||
case Copy010:
|
||||
return 2
|
||||
case Copy100:
|
||||
return 3
|
||||
case Copy110:
|
||||
return 4
|
||||
case Copy200:
|
||||
return 5
|
||||
}
|
||||
return -1
|
||||
func (repType ReplicationType)GetReplicationLevelIndex() int {
|
||||
switch repType {
|
||||
case Copy000:
|
||||
return 0
|
||||
case Copy001:
|
||||
return 1
|
||||
case Copy010:
|
||||
return 2
|
||||
case Copy100:
|
||||
return 3
|
||||
case Copy110:
|
||||
return 4
|
||||
case Copy200:
|
||||
return 5
|
||||
}
|
||||
return -1
|
||||
}
|
||||
func (repType ReplicationType) GetCopyCount() int {
|
||||
switch repType {
|
||||
case Copy000:
|
||||
return 1
|
||||
case Copy001:
|
||||
return 2
|
||||
case Copy010:
|
||||
return 2
|
||||
case Copy100:
|
||||
return 2
|
||||
case Copy110:
|
||||
return 3
|
||||
case Copy200:
|
||||
return 3
|
||||
}
|
||||
return 0
|
||||
func (repType ReplicationType)GetCopyCount() int {
|
||||
switch repType {
|
||||
case Copy000:
|
||||
return 1
|
||||
case Copy001:
|
||||
return 2
|
||||
case Copy010:
|
||||
return 2
|
||||
case Copy100:
|
||||
return 2
|
||||
case Copy110:
|
||||
return 3
|
||||
case Copy200:
|
||||
return 3
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -65,13 +65,13 @@ func (s *Store) AddVolume(volumeListString string, replicationType string) error
|
||||
}
|
||||
return e
|
||||
}
|
||||
func (s *Store) addVolume(vid VolumeId, replicationType ReplicationType) (err error) {
|
||||
func (s *Store) addVolume(vid VolumeId, replicationType ReplicationType) error {
|
||||
if s.volumes[vid] != nil {
|
||||
return errors.New("Volume Id " + vid.String() + " already exists!")
|
||||
}
|
||||
log.Println("In dir", s.dir, "adds volume =", vid, ", replicationType =", replicationType)
|
||||
s.volumes[vid], err = NewVolume(s.dir, vid, replicationType)
|
||||
return err
|
||||
s.volumes[vid] = NewVolume(s.dir, vid, replicationType)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) CheckCompactVolume(volumeIdString string, garbageThresholdString string) (error, bool) {
|
||||
@@ -107,10 +107,9 @@ func (s *Store) loadExistingVolumes() {
|
||||
base := name[:len(name)-len(".dat")]
|
||||
if vid, err := NewVolumeId(base); err == nil {
|
||||
if s.volumes[vid] == nil {
|
||||
if v, e := NewVolume(s.dir, vid, CopyNil); e == nil {
|
||||
s.volumes[vid] = v
|
||||
log.Println("In dir", s.dir, "read volume =", vid, "replicationType =", v.replicaType, "version =", v.version, "size =", v.Size())
|
||||
}
|
||||
v := NewVolume(s.dir, vid, CopyNil)
|
||||
s.volumes[vid] = v
|
||||
log.Println("In dir", s.dir, "read volume =", vid, "replicationType =", v.replicaType, "version =", v.version, "size =", v.Size())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,9 +24,9 @@ type Volume struct {
|
||||
accessLock sync.Mutex
|
||||
}
|
||||
|
||||
func NewVolume(dirname string, id VolumeId, replicationType ReplicationType) (v *Volume, e error) {
|
||||
func NewVolume(dirname string, id VolumeId, replicationType ReplicationType) (v *Volume) {
|
||||
v = &Volume{dir: dirname, Id: id, replicaType: replicationType}
|
||||
e = v.load()
|
||||
v.load()
|
||||
return
|
||||
}
|
||||
func (v *Volume) load() error {
|
||||
@@ -43,7 +43,6 @@ func (v *Volume) load() error {
|
||||
} else {
|
||||
v.maybeWriteSuperBlock()
|
||||
}
|
||||
// TODO: if .idx not exists, but .cdb exists, then use (but don't load!) that
|
||||
indexFile, ie := os.OpenFile(fileName+".idx", os.O_RDWR|os.O_CREATE, 0644)
|
||||
if ie != nil {
|
||||
return fmt.Errorf("cannot create Volume Data %s.dat: %s", fileName, e)
|
||||
@@ -80,23 +79,21 @@ func (v *Volume) maybeWriteSuperBlock() {
|
||||
v.dataFile.Write(header)
|
||||
}
|
||||
}
|
||||
func (v *Volume) readSuperBlock() (err error) {
|
||||
func (v *Volume) readSuperBlock() error {
|
||||
v.dataFile.Seek(0, 0)
|
||||
header := make([]byte, SuperBlockSize)
|
||||
if _, e := v.dataFile.Read(header); e != nil {
|
||||
return fmt.Errorf("cannot read superblock: %s", e)
|
||||
}
|
||||
var err error
|
||||
v.version, v.replicaType, err = ParseSuperBlock(header)
|
||||
return err
|
||||
}
|
||||
func ParseSuperBlock(header []byte) (version Version, replicaType ReplicationType, err error) {
|
||||
func ParseSuperBlock(header []byte) (version Version, replicaType ReplicationType, e error) {
|
||||
version = Version(header[0])
|
||||
if version == 0 {
|
||||
err = errors.New("Zero version impossible - bad superblock!")
|
||||
return
|
||||
}
|
||||
var err error
|
||||
if replicaType, err = NewReplicationTypeFromByte(header[1]); err != nil {
|
||||
err = fmt.Errorf("cannot read replica type: %s", err)
|
||||
e = fmt.Errorf("cannot read replica type: %s", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -224,39 +221,3 @@ func (v *Volume) copyDataAndGenerateIndexFile(srcName, dstName, idxName string)
|
||||
func (v *Volume) ContentSize() uint64 {
|
||||
return v.nm.fileByteCounter
|
||||
}
|
||||
|
||||
// Walk over the contained needles (call the function with each NeedleValue till error is returned)
|
||||
func (v *Volume) WalkValues(pedestrian func(*Needle) error) error {
|
||||
pedplus := func(nv *NeedleValue) (err error) {
|
||||
n := new(Needle)
|
||||
if nv.Offset > 0 {
|
||||
v.dataFile.Seek(int64(nv.Offset)*NeedlePaddingSize, 0)
|
||||
if _, err = n.Read(v.dataFile, nv.Size, v.version); err != nil {
|
||||
return
|
||||
}
|
||||
if err = pedestrian(n); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return v.nm.Walk(pedplus)
|
||||
}
|
||||
|
||||
// Walk over the keys
|
||||
func (v *Volume) WalkKeys(pedestrian func(Key) error) error {
|
||||
pedplus := func(nv *NeedleValue) (err error) {
|
||||
if nv.Offset > 0 && nv.Key > 0 {
|
||||
if err = pedestrian(nv.Key); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return v.nm.Walk(pedplus)
|
||||
}
|
||||
|
||||
func (v *Volume) String() string {
|
||||
return fmt.Sprintf("%d@%s:v%d:r%s", v.Id, v.dataFile.Name(),
|
||||
v.Version(), v.replicaType)
|
||||
}
|
||||
@@ -1,18 +1,17 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
type VolumeId uint32
|
||||
|
||||
func NewVolumeId(vid string) (VolumeId, error) {
|
||||
volumeId, err := strconv.ParseUint(vid, 10, 64)
|
||||
return VolumeId(volumeId), err
|
||||
func NewVolumeId(vid string) (VolumeId,error) {
|
||||
volumeId, err := strconv.ParseUint(vid, 10, 64)
|
||||
return VolumeId(volumeId), err
|
||||
}
|
||||
func (vid *VolumeId) String() string {
|
||||
return strconv.FormatUint(uint64(*vid), 10)
|
||||
func (vid *VolumeId) String() string{
|
||||
return strconv.FormatUint(uint64(*vid), 10)
|
||||
}
|
||||
func (vid *VolumeId) Next() VolumeId {
|
||||
return VolumeId(uint32(*vid) + 1)
|
||||
func (vid *VolumeId) Next() VolumeId{
|
||||
return VolumeId(uint32(*vid)+1)
|
||||
}
|
||||
@@ -1,11 +1,12 @@
|
||||
package storage
|
||||
|
||||
import ()
|
||||
import (
|
||||
)
|
||||
|
||||
type Version uint8
|
||||
|
||||
const (
|
||||
Version1 = Version(1)
|
||||
Version2 = Version(2)
|
||||
CurrentVersion = Version2
|
||||
Version1 = Version(1)
|
||||
Version2 = Version(2)
|
||||
CurrentVersion = Version2
|
||||
)
|
||||
Reference in new issue
Block a user