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 {