Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

better error handling, check in for Gulácsi Tamás

pull/2/head
Chris Lu 12 years ago
parent
768eaff4b8
commit
866d15023d
6 changed files with 159 additions and 80 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 16
      weed-fs/src/cmd/weed/fix.go
  2. 12
      weed-fs/src/cmd/weed/volume.go
  3. 109
      weed-fs/src/pkg/storage/needle_read_write.go
  4. 14
      weed-fs/src/pkg/storage/store.go
  5. 84
      weed-fs/src/pkg/storage/volume.go
  6. 4
      weed-fs/src/pkg/topology/volume_layout.go

16
weed-fs/src/cmd/weed/fix.go
View File

@ -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

12
weed-fs/src/cmd/weed/volume.go
View File

@ -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 {

109
weed-fs/src/pkg/storage/needle_read_write.go
View File

@ -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
}

14
weed-fs/src/pkg/storage/store.go
View File

@ -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 {

84
weed-fs/src/pkg/storage/volume.go
View File

@ -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

4
weed-fs/src/pkg/topology/volume_layout.go
View File

@ -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 {

Write Preview
Loading…
Cancel
Save