From af32b5272769c269de9acabd0cf02e2cb2ad190b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 18 Mar 2014 23:48:01 -0700 Subject: [PATCH] 1. no locks for all read operations! Switching to pread for all reads. 2. prevent heartbeat lost when vacuuming, by removing locks on Size() function --- go/storage/needle_read_write.go | 16 +++++++------- go/storage/volume.go | 39 ++++++++++----------------------- 2 files changed, 20 insertions(+), 35 deletions(-) diff --git a/go/storage/needle_read_write.go b/go/storage/needle_read_write.go index c414ac299..835d7c270 100644 --- a/go/storage/needle_read_write.go +++ b/go/storage/needle_read_write.go @@ -121,11 +121,11 @@ func (n *Needle) Append(w io.Writer, version Version) (size uint32, err error) { return 0, fmt.Errorf("Unsupported Version! (%d)", version) } -func (n *Needle) Read(r io.Reader, size uint32, version Version) (ret int, err error) { +func (n *Needle) Read(r *os.File, offset int64, size uint32, version Version) (ret int, err error) { switch version { case Version1: bytes := make([]byte, NeedleHeaderSize+size+NeedleChecksumSize) - if ret, err = r.Read(bytes); err != nil { + if ret, err = r.ReadAt(bytes, offset); err != nil { return } n.readNeedleHeader(bytes) @@ -142,7 +142,7 @@ func (n *Needle) Read(r io.Reader, size uint32, version Version) (ret int, err e return 0, nil } bytes := make([]byte, NeedleHeaderSize+size+NeedleChecksumSize) - if ret, err = r.Read(bytes); err != nil { + if ret, err = r.ReadAt(bytes, offset); err != nil { return } if ret != int(NeedleHeaderSize+size+NeedleChecksumSize) { @@ -196,12 +196,12 @@ func (n *Needle) readNeedleDataVersion2(bytes []byte) { } } -func ReadNeedleHeader(r *os.File, version Version) (n *Needle, bodyLength uint32, err error) { +func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bodyLength uint32, err error) { n = new(Needle) if version == Version1 || version == Version2 { bytes := make([]byte, NeedleHeaderSize) var count int - count, err = r.Read(bytes) + count, err = r.ReadAt(bytes, offset) if count <= 0 || err != nil { return nil, 0, err } @@ -214,21 +214,21 @@ 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) (err error) { +func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32) (err error) { if bodyLength <= 0 { return nil } switch version { case Version1: bytes := make([]byte, bodyLength) - if _, err = r.Read(bytes); err != nil { + if _, err = r.ReadAt(bytes, offset); err != nil { return } n.Data = bytes[:n.Size] n.Checksum = NewCRC(n.Data) case Version2: bytes := make([]byte, bodyLength) - if _, err = r.Read(bytes); err != nil { + if _, err = r.ReadAt(bytes, offset); err != nil { return } n.readNeedleDataVersion2(bytes[0:n.Size]) diff --git a/go/storage/volume.go b/go/storage/volume.go index 0301d7968..460fbfccd 100644 --- a/go/storage/volume.go +++ b/go/storage/volume.go @@ -125,8 +125,6 @@ func (v *Volume) Version() Version { return v.SuperBlock.Version } func (v *Volume) Size() int64 { - v.accessLock.Lock() - defer v.accessLock.Unlock() stat, e := v.dataFile.Stat() if e == nil { return stat.Size() @@ -185,11 +183,8 @@ func (v *Volume) NeedToReplicate() bool { func (v *Volume) isFileUnchanged(n *Needle) bool { nv, ok := v.nm.Get(n.Id) if ok && nv.Offset > 0 { - if _, err := v.dataFile.Seek(int64(nv.Offset)*NeedlePaddingSize, 0); err != nil { - return false - } oldNeedle := new(Needle) - oldNeedle.Read(v.dataFile, nv.Size, v.Version()) + oldNeedle.Read(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) if oldNeedle.Checksum == n.Checksum && bytes.Equal(oldNeedle.Data, n.Data) { n.Size = oldNeedle.Size return true @@ -277,14 +272,9 @@ func (v *Volume) delete(n *Needle) (uint32, error) { } func (v *Volume) read(n *Needle) (int, error) { - v.accessLock.Lock() - defer v.accessLock.Unlock() nv, ok := v.nm.Get(n.Id) if ok && nv.Offset > 0 { - if _, err := v.dataFile.Seek(int64(nv.Offset)*NeedlePaddingSize, 0); err != nil { - return -1, err - } - return n.Read(v.dataFile, nv.Size, v.Version()) + return n.Read(v.dataFile, int64(nv.Offset)*NeedlePaddingSize, nv.Size, v.Version()) } return -1, errors.New("Not Found") } @@ -356,21 +346,22 @@ func ScanVolumeFile(dirname string, collection string, id VolumeId, version := v.Version() offset := int64(SuperBlockSize) - n, rest, e := ReadNeedleHeader(v.dataFile, version) + n, rest, e := ReadNeedleHeader(v.dataFile, version, offset) if e != nil { err = fmt.Errorf("cannot read needle header: %s", e) return } for n != nil { - if err = n.ReadNeedleBody(v.dataFile, version, rest); err != nil { + offset += int64(NeedleHeaderSize) + if err = n.ReadNeedleBody(v.dataFile, version, offset, rest); err != nil { err = fmt.Errorf("cannot read needle body: %s", err) return } if err = visitNeedle(n, offset); err != nil { return } - offset += int64(NeedleHeaderSize + rest) - if n, rest, err = ReadNeedleHeader(v.dataFile, version); err != nil { + offset += int64(rest) + if n, rest, err = ReadNeedleHeader(v.dataFile, version, offset); err != nil { if err == io.EOF { return nil } @@ -395,7 +386,6 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string) (err erro } defer idx.Close() - nm := NewNeedleMap(idx) new_offset := int64(SuperBlockSize) err = ScanVolumeFile(v.dir, v.Collection, v.Id, func(superBlock SuperBlock) error { @@ -404,17 +394,12 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string) (err erro }, func(n *Needle, offset int64) error { nv, ok := v.nm.Get(n.Id) //glog.V(0).Infoln("file size is", n.Size, "rest", rest) - if ok && int64(nv.Offset)*NeedlePaddingSize == offset { - if nv.Size > 0 { - if _, err = nm.Put(n.Id, uint32(new_offset/NeedlePaddingSize), n.Size); err != nil { - return fmt.Errorf("cannot put needle: %s", err) - } - if _, err = n.Append(dst, v.Version()); err != nil { - return fmt.Errorf("cannot append needle: %s", err) - } - new_offset += n.DiskSize() - //glog.V(0).Infoln("saving key", n.Id, "volume offset", old_offset, "=>", new_offset, "data_size", n.Size, "rest", rest) + if ok && int64(nv.Offset)*NeedlePaddingSize == offset && nv.Size > 0 { + if _, err = n.Append(dst, v.Version()); err != nil { + return fmt.Errorf("cannot append needle: %s", err) } + new_offset += n.DiskSize() + //glog.V(0).Infoln("saving key", n.Id, "volume offset", old_offset, "=>", new_offset, "data_size", n.Size, "rest", rest) } return nil })