lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-09-21 02:17:57 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

use tsMemory to determine whether read from disk or memory

Browse Source 
remove lastFlushTime
This commit is contained in:
Chris Lu
2021-10-04 16:02:56 -07:00
parent 6a030547a2
commit 4ed2994555
1 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
weed/util/log_buffer/log_buffer.go
View File

@@ -29,7 +29,6 @@ type LogBuffer struct {
pos int pos int
startTime time.Time startTime time.Time
stopTime time.Time stopTime time.Time
lastFlushTime time.Time
sizeBuf []byte sizeBuf []byte
flushInterval time.Duration flushInterval time.Duration
flushFn func(startTime, stopTime time.Time, buf []byte) flushFn func(startTime, stopTime time.Time, buf []byte)
@@ -133,8 +132,6 @@ func (m *LogBuffer) loopFlush() {
// glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes())) // glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes()))
m.flushFn(d.startTime, d.stopTime, d.data.Bytes()) m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
d.releaseMemory() d.releaseMemory()
// local logbuffer is different from aggregate logbuffer here
m.lastFlushTime = d.stopTime
} }
} }
} }
@@ -165,9 +162,6 @@ func (m *LogBuffer) copyToFlush() *dataToFlush {
data: copiedBytes(m.buf[:m.pos]), data: copiedBytes(m.buf[:m.pos]),
} }
// glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime) // glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
} else {
// glog.V(4).Infof("%s removed from memory [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
m.lastFlushTime = m.stopTime
} }
m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos) m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos)
m.startTime = time.Unix(0, 0) m.startTime = time.Unix(0, 0)
@@ -188,20 +182,31 @@ func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Bu
m.RLock() m.RLock()
defer m.RUnlock() defer m.RUnlock()
if !m.lastFlushTime.IsZero() && m.lastFlushTime.After(lastReadTime) { // Read from disk and memory
if time.Now().Sub(m.lastFlushTime) < m.flushInterval*2 { // 1. read from disk, last time is = td
diff := m.lastFlushTime.Sub(lastReadTime) // 2. in memory, the earliest time = tm
glog.V(4).Infof("lastFlush:%v lastRead:%v diff:%v", m.lastFlushTime, lastReadTime, diff) // if tm <= td, case 2.1
return nil, ResumeFromDiskError // read from memory
// if tm is empty, case 2.2
// read from memory
// if td < tm, case 2.3
// read from disk again
var tsMemory time.Time
if !m.startTime.IsZero() {
tsMemory = m.startTime
}
for _, prevBuf := range m.prevBuffers.buffers {
if !prevBuf.startTime.IsZero() && prevBuf.startTime.Before(tsMemory) {
tsMemory = prevBuf.startTime
} }
} }
if tsMemory.IsZero() { // case 2.2
return nil, nil
} else if lastReadTime.Before(tsMemory) { // case 2.3
return nil, ResumeFromDiskError
}
/* // the following is case 2.1
fmt.Printf("read buffer %p: %v last stop time: [%v,%v], pos %d, entries:%d, prevBufs:%d\n", m, lastReadTime, m.startTime, m.stopTime, m.pos, len(m.idx), len(m.prevBuffers.buffers))
for i, prevBuf := range m.prevBuffers.buffers {
fmt.Printf(" prev %d : %s\n", i, prevBuf.String())
}
*/
if lastReadTime.Equal(m.stopTime) { if lastReadTime.Equal(m.stopTime) {
return nil, nil return nil, nil