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
This commit is contained in:
Chris Lu 2014-03-18 23:48:01 -07:00
parent 2841d59fb0
commit af32b52727
2 changed files with 20 additions and 35 deletions

View file

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

View file

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