aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--weed-fs/src/cmd/weed/fix.go16
-rw-r--r--weed-fs/src/cmd/weed/volume.go12
-rw-r--r--weed-fs/src/pkg/storage/needle_read_write.go109
-rw-r--r--weed-fs/src/pkg/storage/store.go14
-rw-r--r--weed-fs/src/pkg/storage/volume.go84
-rw-r--r--weed-fs/src/pkg/topology/volume_layout.go4
6 files changed, 159 insertions, 80 deletions
diff --git a/weed-fs/src/cmd/weed/fix.go b/weed-fs/src/cmd/weed/fix.go
index d794374cb..deee9a81f 100644
--- a/weed-fs/src/cmd/weed/fix.go
+++ b/weed-fs/src/cmd/weed/fix.go
@@ -50,20 +50,28 @@ func runFix(cmd *Command, args []string) bool {
log.Fatalf("cannot read superblock: %s", e)
}
- ver, _, _ := storage.ParseSuperBlock(header)
+ ver, _, e := storage.ParseSuperBlock(header)
+ if e != nil {
+ log.Fatalf("error parsing superblock: %s", e)
+ }
- n, rest := storage.ReadNeedleHeader(dataFile, ver)
+ n, rest, e := storage.ReadNeedleHeader(dataFile, ver)
+ if e != nil {
+ log.Fatalf("error reading needle header: %s", e)
+ }
dataFile.Seek(int64(rest), 1)
nm := storage.NewNeedleMap(indexFile)
offset := uint32(storage.SuperBlockSize)
for n != nil {
debug("key", n.Id, "volume offset", offset, "data_size", n.Size, "rest", rest)
if n.Size > 0 {
- count, pe := nm.Put(n.Id, offset/8, n.Size)
+ count, pe := nm.Put(n.Id, offset/storage.NeedlePaddingSize, n.Size)
debug("saved", count, "with error", pe)
}
offset += rest + 16
- n, rest = storage.ReadNeedleHeader(dataFile, ver)
+ if n, rest, e = storage.ReadNeedleHeader(dataFile, ver); e != nil {
+ log.Fatalf("error reading needle header: %s", e)
+ }
dataFile.Seek(int64(rest), 1)
}
return true
diff --git a/weed-fs/src/cmd/weed/volume.go b/weed-fs/src/cmd/weed/volume.go
index 4ee8c31a5..ee569843f 100644
--- a/weed-fs/src/cmd/weed/volume.go
+++ b/weed-fs/src/cmd/weed/volume.go
@@ -176,10 +176,12 @@ func PostHandler(w http.ResponseWriter, r *http.Request) {
if ne != nil {
writeJson(w, r, ne)
} else {
- ret := store.Write(volumeId, needle)
+ ret, err := store.Write(volumeId, needle)
errorStatus := ""
needToReplicate := !store.HasVolume(volumeId)
- if ret > 0 {
+ if err != nil {
+ errorStatus = "Failed to write to local disk (" + err.Error() + ")"
+ } else if ret > 0 {
needToReplicate = needToReplicate || store.GetVolume(volumeId).NeedToReplicate()
} else {
errorStatus = "Failed to write to local disk"
@@ -238,7 +240,11 @@ func DeleteHandler(w http.ResponseWriter, r *http.Request) {
}
n.Size = 0
- ret := store.Delete(volumeId, n)
+ ret, err := store.Delete(volumeId, n)
+ if err != nil {
+ log.Println("delete error: %s", err)
+ return
+ }
needToReplicate := !store.HasVolume(volumeId)
if !needToReplicate && ret > 0 {
diff --git a/weed-fs/src/pkg/storage/needle_read_write.go b/weed-fs/src/pkg/storage/needle_read_write.go
index 74b520fb8..62d267d24 100644
--- a/weed-fs/src/pkg/storage/needle_read_write.go
+++ b/weed-fs/src/pkg/storage/needle_read_write.go
@@ -8,20 +8,31 @@ import (
"pkg/util"
)
-func (n *Needle) Append(w io.Writer, version Version) uint32 {
+const (
+ FlagGzip = 0x01
+ FlagHasName = 0x02
+ FlagHasMime = 0x04
+)
+
+func (n *Needle) Append(w io.Writer, version Version) (size uint32, err error) {
switch version {
case Version1:
header := make([]byte, NeedleHeaderSize)
util.Uint32toBytes(header[0:4], n.Cookie)
util.Uint64toBytes(header[4:12], n.Id)
n.Size = uint32(len(n.Data))
+ size = n.Size
util.Uint32toBytes(header[12:16], n.Size)
- w.Write(header)
- w.Write(n.Data)
+ if _, err = w.Write(header); err != nil {
+ return
+ }
+ if _, err = w.Write(n.Data); err != nil {
+ return
+ }
padding := NeedlePaddingSize - ((NeedleHeaderSize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
- w.Write(header[0 : NeedleChecksumSize+padding])
- return n.Size
+ _, err = w.Write(header[0 : NeedleChecksumSize+padding])
+ return
case Version2:
header := make([]byte, NeedleHeaderSize)
util.Uint32toBytes(header[0:4], n.Cookie)
@@ -36,52 +47,71 @@ func (n *Needle) Append(w io.Writer, version Version) uint32 {
n.Size = n.Size + 1 + uint32(n.MimeSize)
}
}
+ size = n.DataSize
util.Uint32toBytes(header[12:16], n.Size)
- w.Write(header)
+ if _, err = w.Write(header); err != nil {
+ return
+ }
if n.DataSize > 0 {
util.Uint32toBytes(header[0:4], n.DataSize)
- w.Write(header[0:4])
- w.Write(n.Data)
+ if _, err = w.Write(header[0:4]); err != nil {
+ return
+ }
+ if _, err = w.Write(n.Data); err != nil {
+ return
+ }
util.Uint8toBytes(header[0:1], n.Flags)
- w.Write(header[0:1])
+ if _, err = w.Write(header[0:1]); err != nil {
+ return
+ }
}
if n.HasName() {
util.Uint8toBytes(header[0:1], n.NameSize)
- w.Write(header[0:1])
- w.Write(n.Name)
+ if _, err = w.Write(header[0:1]); err != nil {
+ return
+ }
+ if _, err = w.Write(n.Name); err != nil {
+ return
+ }
}
if n.HasMime() {
util.Uint8toBytes(header[0:1], n.MimeSize)
- w.Write(header[0:1])
- w.Write(n.Mime)
+ if _, err = w.Write(header[0:1]); err != nil {
+ return
+ }
+ if _, err = w.Write(n.Mime); err != nil {
+ return
+ }
}
padding := NeedlePaddingSize - ((NeedleHeaderSize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
- w.Write(header[0 : NeedleChecksumSize+padding])
- return n.DataSize
+ _, err = w.Write(header[0 : NeedleChecksumSize+padding])
+ return n.DataSize, err
}
- return n.Size
+ return 0, fmt.Errorf("Unsupported Version! (%d)", version)
}
-func (n *Needle) Read(r io.Reader, size uint32, version Version) (int, error) {
+
+func (n *Needle) Read(r io.Reader, size uint32, version Version) (ret int, err error) {
switch version {
case Version1:
bytes := make([]byte, NeedleHeaderSize+size+NeedleChecksumSize)
- ret, e := r.Read(bytes)
+ if ret, err = r.Read(bytes); err != nil {
+ return
+ }
n.readNeedleHeader(bytes)
n.Data = bytes[NeedleHeaderSize : NeedleHeaderSize+size]
checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize])
if checksum != NewCRC(n.Data).Value() {
return 0, errors.New("CRC error! Data On Disk Corrupted!")
}
- return ret, e
+ return
case Version2:
if size == 0 {
return 0, nil
}
bytes := make([]byte, NeedleHeaderSize+size+NeedleChecksumSize)
- ret, e := r.Read(bytes)
- if e != nil {
- return 0, e
+ if ret, err = r.Read(bytes); err != nil {
+ return
}
if ret != int(NeedleHeaderSize+size+NeedleChecksumSize) {
return 0, errors.New("File Entry Not Found!")
@@ -95,7 +125,7 @@ func (n *Needle) Read(r io.Reader, size uint32, version Version) (int, error) {
if checksum != NewCRC(n.Data).Value() {
return 0, errors.New("CRC error! Data On Disk Corrupted!")
}
- return ret, e
+ return
}
return 0, fmt.Errorf("Unsupported Version! (%d)", version)
}
@@ -126,13 +156,14 @@ func (n *Needle) readNeedleDataVersion2(bytes []byte) {
n.Mime = bytes[index : index+int(n.MimeSize)]
}
}
-func ReadNeedleHeader(r *os.File, version Version) (n *Needle, bodyLength uint32) {
+func ReadNeedleHeader(r *os.File, version Version) (n *Needle, bodyLength uint32, err error) {
n = new(Needle)
if version == Version1 || version == Version2 {
bytes := make([]byte, NeedleHeaderSize)
- count, e := r.Read(bytes)
- if count <= 0 || e != nil {
- return nil, 0
+ var count int
+ count, err = r.Read(bytes)
+ if count <= 0 || err != nil {
+ return nil, 0, err
}
n.readNeedleHeader(bytes)
padding := NeedlePaddingSize - ((n.Size + NeedleHeaderSize + NeedleChecksumSize) % NeedlePaddingSize)
@@ -143,37 +174,43 @@ func ReadNeedleHeader(r *os.File, version Version) (n *Needle, bodyLength uint32
//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, bodyLength uint32) {
+func (n *Needle) ReadNeedleBody(r *os.File, version Version, bodyLength uint32) (err error) {
switch version {
case Version1:
bytes := make([]byte, bodyLength)
- r.Read(bytes)
+ if _, err = r.Read(bytes); err != nil {
+ return
+ }
n.Data = bytes[:n.Size]
n.Checksum = NewCRC(n.Data)
case Version2:
bytes := make([]byte, bodyLength)
- r.Read(bytes)
+ if _, err = r.Read(bytes); err != nil {
+ return
+ }
n.readNeedleDataVersion2(bytes[0:n.Size])
n.Checksum = NewCRC(n.Data)
+ default:
+ err = fmt.Errorf("Unsupported Version! (%d)", version)
}
return
}
func (n *Needle) IsGzipped() bool {
- return n.Flags&0x01 == 0x01
+ return n.Flags&FlagGzip > 0
}
func (n *Needle) SetGzipped() {
- n.Flags = n.Flags | 0x01
+ n.Flags = n.Flags | FlagGzip
}
func (n *Needle) HasName() bool {
- return n.Flags&0x02 == 0x02
+ return n.Flags&FlagHasName > 0
}
func (n *Needle) SetHasName() {
- n.Flags = n.Flags | 0x02
+ n.Flags = n.Flags | FlagHasName
}
func (n *Needle) HasMime() bool {
- return n.Flags&0x04 == 0x04
+ return n.Flags&FlagHasMime > 0
}
func (n *Needle) SetHasMime() {
- n.Flags = n.Flags | 0x04
+ n.Flags = n.Flags | FlagHasMime
}
diff --git a/weed-fs/src/pkg/storage/store.go b/weed-fs/src/pkg/storage/store.go
index 79cf65e28..18af19aff 100644
--- a/weed-fs/src/pkg/storage/store.go
+++ b/weed-fs/src/pkg/storage/store.go
@@ -168,23 +168,23 @@ func (s *Store) Close() {
v.Close()
}
}
-func (s *Store) Write(i VolumeId, n *Needle) uint32 {
+func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
if v := s.volumes[i]; v != nil {
- size := v.write(n)
- if s.volumeSizeLimit < v.ContentSize()+uint64(size) && s.volumeSizeLimit >= v.ContentSize() {
+ size, err = v.write(n)
+ if err != nil && s.volumeSizeLimit < v.ContentSize()+uint64(size) && s.volumeSizeLimit >= v.ContentSize() {
log.Println("volume", i, "size is", v.ContentSize(), "close to", s.volumeSizeLimit)
s.Join()
}
- return size
+ return
}
log.Println("volume", i, "not found!")
- return 0
+ return
}
-func (s *Store) Delete(i VolumeId, n *Needle) uint32 {
+func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
if v := s.volumes[i]; v != nil {
return v.delete(n)
}
- return 0
+ return 0, nil
}
func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
if v := s.volumes[i]; v != nil {
diff --git a/weed-fs/src/pkg/storage/volume.go b/weed-fs/src/pkg/storage/volume.go
index 6da991f60..933a84060 100644
--- a/weed-fs/src/pkg/storage/volume.go
+++ b/weed-fs/src/pkg/storage/volume.go
@@ -70,7 +70,11 @@ func (v *Volume) Close() {
v.dataFile.Close()
}
func (v *Volume) maybeWriteSuperBlock() {
- stat, _ := v.dataFile.Stat()
+ stat, e := v.dataFile.Stat()
+ if e != nil {
+ fmt.Printf("failed to stat datafile %s: %s", v.dataFile, e)
+ return
+ }
if stat.Size() == 0 {
v.version = CurrentVersion
header := make([]byte, SuperBlockSize)
@@ -99,18 +103,23 @@ func (v *Volume) NeedToReplicate() bool {
return v.replicaType.GetCopyCount() > 1
}
-func (v *Volume) write(n *Needle) uint32 {
+func (v *Volume) write(n *Needle) (size uint32, err error) {
v.accessLock.Lock()
defer v.accessLock.Unlock()
- offset, _ := v.dataFile.Seek(0, 2)
- ret := n.Append(v.dataFile, v.version)
+ var offset int64
+ if offset, err = v.dataFile.Seek(0, 2); err != nil {
+ return
+ }
+ if size, err = n.Append(v.dataFile, v.version); err != nil {
+ return
+ }
nv, ok := v.nm.Get(n.Id)
- if !ok || int64(nv.Offset)*8 < offset {
- v.nm.Put(n.Id, uint32(offset/8), n.Size)
+ if !ok || int64(nv.Offset)*NeedlePaddingSize < offset {
+ _, err = v.nm.Put(n.Id, uint32(offset/NeedlePaddingSize), n.Size)
}
- return ret
+ return
}
-func (v *Volume) delete(n *Needle) uint32 {
+func (v *Volume) delete(n *Needle) (uint32, error) {
v.accessLock.Lock()
defer v.accessLock.Unlock()
nv, ok := v.nm.Get(n.Id)
@@ -118,10 +127,10 @@ func (v *Volume) delete(n *Needle) uint32 {
if ok {
v.nm.Delete(n.Id)
v.dataFile.Seek(int64(nv.Offset*NeedlePaddingSize), 0)
- n.Append(v.dataFile, v.version)
- return nv.Size
+ _, err := n.Append(v.dataFile, v.version)
+ return nv.Size, err
}
- return 0
+ return 0, nil
}
func (v *Volume) read(n *Needle) (int, error) {
@@ -164,33 +173,44 @@ func (v *Volume) commitCompact() error {
}
func (v *Volume) copyDataAndGenerateIndexFile(srcName, dstName, idxName string) (err error) {
- src, err := os.OpenFile(srcName, os.O_RDONLY, 0644)
- if err != nil {
- return err
+ var (
+ src, dst, idx *os.File
+ )
+ if src, err = os.OpenFile(srcName, os.O_RDONLY, 0644); err != nil {
+ return
}
defer src.Close()
- dst, err := os.OpenFile(dstName, os.O_WRONLY|os.O_CREATE, 0644)
- if err != nil {
- return err
+ if dst, err = os.OpenFile(dstName, os.O_WRONLY|os.O_CREATE, 0644); err != nil {
+ return
}
defer dst.Close()
- idx, err := os.OpenFile(idxName, os.O_WRONLY|os.O_CREATE, 0644)
- if err != nil {
- return err
+ if idx, err = os.OpenFile(idxName, os.O_WRONLY|os.O_CREATE, 0644); err != nil {
+ return
}
defer idx.Close()
src.Seek(0, 0)
header := make([]byte, SuperBlockSize)
- if _, error := src.Read(header); error == nil {
- dst.Write(header)
+ if _, err = src.Read(header); err == nil {
+ _, err = dst.Write(header)
+ }
+ if err != nil {
+ return err
}
- version, _, _ := ParseSuperBlock(header)
+ version, _, e := ParseSuperBlock(header)
+ if e != nil {
+ err = fmt.Errorf("cannot parse superblock: %s", e)
+ return
+ }
- n, rest := ReadNeedleHeader(src, version)
+ n, rest, e := ReadNeedleHeader(src, version)
+ if e != nil {
+ err = fmt.Errorf("cannot read needle header: %s", e)
+ return
+ }
nm := NewNeedleMap(idx)
old_offset := uint32(SuperBlockSize)
new_offset := uint32(SuperBlockSize)
@@ -201,9 +221,15 @@ func (v *Volume) copyDataAndGenerateIndexFile(srcName, dstName, idxName string)
src.Seek(int64(rest), 1)
} else {
if nv.Size > 0 {
- nm.Put(n.Id, new_offset/NeedlePaddingSize, n.Size)
- n.ReadNeedleBody(src, version, rest)
- n.Append(dst, v.version)
+ if _, err = nm.Put(n.Id, new_offset/NeedlePaddingSize, n.Size); err != nil {
+ return fmt.Errorf("cannot put needle: %s", err)
+ }
+ if err = n.ReadNeedleBody(src, version, rest); err != nil {
+ return fmt.Errorf("cannot read needle body: %s", err)
+ }
+ if _, err = n.Append(dst, v.version); err != nil {
+ return fmt.Errorf("cannot append needle: %s", err)
+ }
new_offset += rest + NeedleHeaderSize
//log.Println("saving key", n.Id, "volume offset", old_offset, "=>", new_offset, "data_size", n.Size, "rest", rest)
} else {
@@ -211,7 +237,9 @@ func (v *Volume) copyDataAndGenerateIndexFile(srcName, dstName, idxName string)
}
}
old_offset += rest + NeedleHeaderSize
- n, rest = ReadNeedleHeader(src, version)
+ if n, rest, err = ReadNeedleHeader(src, version); err != nil {
+ return fmt.Errorf("cannot read needle header: %s", err)
+ }
}
return nil
diff --git a/weed-fs/src/pkg/topology/volume_layout.go b/weed-fs/src/pkg/topology/volume_layout.go
index 314aca69f..6d62128d9 100644
--- a/weed-fs/src/pkg/topology/volume_layout.go
+++ b/weed-fs/src/pkg/topology/volume_layout.go
@@ -38,8 +38,8 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
}
}
-func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool{
- return uint64(v.Size) < vl.volumeSizeLimit && v.Version == storage.CurrentVersion
+func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool {
+ return uint64(v.Size) < vl.volumeSizeLimit && v.Version == storage.CurrentVersion
}
func (vl *VolumeLayout) Lookup(vid storage.VolumeId) []*DataNode {
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-23 20:00:01 +0000