You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

306 lines
7.5 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
3 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
5 years ago
  1. package log_buffer
  2. import (
  3. "bytes"
  4. "sync"
  5. "time"
  6. "github.com/golang/protobuf/proto"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. )
  11. const BufferSize = 4 * 1024 * 1024
  12. const PreviousBufferCount = 3
  13. type dataToFlush struct {
  14. startTime time.Time
  15. stopTime time.Time
  16. data *bytes.Buffer
  17. }
  18. type LogBuffer struct {
  19. name string
  20. prevBuffers *SealedBuffers
  21. buf []byte
  22. idx []int
  23. pos int
  24. startTime time.Time
  25. stopTime time.Time
  26. sizeBuf []byte
  27. flushInterval time.Duration
  28. flushFn func(startTime, stopTime time.Time, buf []byte)
  29. notifyFn func()
  30. isStopping bool
  31. flushChan chan *dataToFlush
  32. lastTsNs int64
  33. sync.RWMutex
  34. }
  35. func NewLogBuffer(name string, flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer {
  36. lb := &LogBuffer{
  37. name: name,
  38. prevBuffers: newSealedBuffers(PreviousBufferCount),
  39. buf: make([]byte, BufferSize),
  40. sizeBuf: make([]byte, 4),
  41. flushInterval: flushInterval,
  42. flushFn: flushFn,
  43. notifyFn: notifyFn,
  44. flushChan: make(chan *dataToFlush, 256),
  45. }
  46. go lb.loopFlush()
  47. go lb.loopInterval()
  48. return lb
  49. }
  50. func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, processingTsNs int64) {
  51. m.Lock()
  52. defer func() {
  53. m.Unlock()
  54. if m.notifyFn != nil {
  55. m.notifyFn()
  56. }
  57. }()
  58. // need to put the timestamp inside the lock
  59. var ts time.Time
  60. if processingTsNs == 0 {
  61. ts = time.Now()
  62. processingTsNs = ts.UnixNano()
  63. } else {
  64. ts = time.Unix(0, processingTsNs)
  65. }
  66. if m.lastTsNs >= processingTsNs {
  67. // this is unlikely to happen, but just in case
  68. processingTsNs = m.lastTsNs + 1
  69. ts = time.Unix(0, processingTsNs)
  70. }
  71. m.lastTsNs = processingTsNs
  72. logEntry := &filer_pb.LogEntry{
  73. TsNs: processingTsNs,
  74. PartitionKeyHash: util.HashToInt32(partitionKey),
  75. Data: data,
  76. }
  77. logEntryData, _ := proto.Marshal(logEntry)
  78. size := len(logEntryData)
  79. if m.pos == 0 {
  80. m.startTime = ts
  81. }
  82. if m.startTime.Add(m.flushInterval).Before(ts) || len(m.buf)-m.pos < size+4 {
  83. // glog.V(4).Infof("%s copyToFlush1 start time %v, ts %v, remaining %d bytes", m.name, m.startTime, ts, len(m.buf)-m.pos)
  84. m.flushChan <- m.copyToFlush()
  85. m.startTime = ts
  86. if len(m.buf) < size+4 {
  87. m.buf = make([]byte, 2*size+4)
  88. }
  89. }
  90. m.stopTime = ts
  91. m.idx = append(m.idx, m.pos)
  92. util.Uint32toBytes(m.sizeBuf, uint32(size))
  93. copy(m.buf[m.pos:m.pos+4], m.sizeBuf)
  94. copy(m.buf[m.pos+4:m.pos+4+size], logEntryData)
  95. m.pos += size + 4
  96. // fmt.Printf("entry size %d total %d count %d, buffer:%p\n", size, m.pos, len(m.idx), m)
  97. }
  98. func (m *LogBuffer) Shutdown() {
  99. m.Lock()
  100. defer m.Unlock()
  101. if m.isStopping {
  102. return
  103. }
  104. m.isStopping = true
  105. toFlush := m.copyToFlush()
  106. m.flushChan <- toFlush
  107. close(m.flushChan)
  108. }
  109. func (m *LogBuffer) loopFlush() {
  110. for d := range m.flushChan {
  111. if d != nil {
  112. // glog.V(4).Infof("%s flush [%v, %v] size %d", m.name, d.startTime, d.stopTime, len(d.data.Bytes()))
  113. m.flushFn(d.startTime, d.stopTime, d.data.Bytes())
  114. d.releaseMemory()
  115. }
  116. }
  117. }
  118. func (m *LogBuffer) loopInterval() {
  119. for !m.isStopping {
  120. time.Sleep(m.flushInterval)
  121. m.Lock()
  122. if m.isStopping {
  123. m.Unlock()
  124. return
  125. }
  126. toFlush := m.copyToFlush()
  127. m.flushChan <- toFlush
  128. m.Unlock()
  129. }
  130. }
  131. func (m *LogBuffer) copyToFlush() *dataToFlush {
  132. if m.pos > 0 {
  133. // fmt.Printf("flush buffer %d pos %d empty space %d\n", len(m.buf), m.pos, len(m.buf)-m.pos)
  134. var d *dataToFlush
  135. if m.flushFn != nil {
  136. d = &dataToFlush{
  137. startTime: m.startTime,
  138. stopTime: m.stopTime,
  139. data: copiedBytes(m.buf[:m.pos]),
  140. }
  141. // glog.V(4).Infof("%s flushing [0,%d) with %d entries [%v, %v]", m.name, m.pos, len(m.idx), m.startTime, m.stopTime)
  142. }
  143. m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos)
  144. m.startTime = time.Unix(0, 0)
  145. m.stopTime = time.Unix(0, 0)
  146. m.pos = 0
  147. m.idx = m.idx[:0]
  148. return d
  149. }
  150. return nil
  151. }
  152. func (d *dataToFlush) releaseMemory() {
  153. d.data.Reset()
  154. bufferPool.Put(d.data)
  155. }
  156. func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Buffer, err error) {
  157. m.RLock()
  158. defer m.RUnlock()
  159. // Read from disk and memory
  160. // 1. read from disk, last time is = td
  161. // 2. in memory, the earliest time = tm
  162. // if tm <= td, case 2.1
  163. // read from memory
  164. // if tm is empty, case 2.2
  165. // read from memory
  166. // if td < tm, case 2.3
  167. // read from disk again
  168. var tsMemory time.Time
  169. if !m.startTime.IsZero() {
  170. tsMemory = m.startTime
  171. }
  172. for _, prevBuf := range m.prevBuffers.buffers {
  173. if !prevBuf.startTime.IsZero() && prevBuf.startTime.Before(tsMemory) {
  174. tsMemory = prevBuf.startTime
  175. }
  176. }
  177. if tsMemory.IsZero() { // case 2.2
  178. return nil, nil
  179. } else if lastReadTime.Before(tsMemory) { // case 2.3
  180. return nil, ResumeFromDiskError
  181. }
  182. // the following is case 2.1
  183. if lastReadTime.Equal(m.stopTime) {
  184. return nil, nil
  185. }
  186. if lastReadTime.After(m.stopTime) {
  187. // glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadTime, m.stopTime)
  188. return nil, nil
  189. }
  190. if lastReadTime.Before(m.startTime) {
  191. // println("checking ", lastReadTime.UnixNano())
  192. for _, buf := range m.prevBuffers.buffers {
  193. if buf.startTime.After(lastReadTime) {
  194. // glog.V(4).Infof("%s return the %d sealed buffer %v", m.name, i, buf.startTime)
  195. // println("return the", i, "th in memory", buf.startTime.UnixNano())
  196. return copiedBytes(buf.buf[:buf.size]), nil
  197. }
  198. if !buf.startTime.After(lastReadTime) && buf.stopTime.After(lastReadTime) {
  199. pos := buf.locateByTs(lastReadTime)
  200. // fmt.Printf("locate buffer[%d] pos %d\n", i, pos)
  201. return copiedBytes(buf.buf[pos:buf.size]), nil
  202. }
  203. }
  204. // glog.V(4).Infof("%s return the current buf %v", m.name, lastReadTime)
  205. return copiedBytes(m.buf[:m.pos]), nil
  206. }
  207. lastTs := lastReadTime.UnixNano()
  208. l, h := 0, len(m.idx)-1
  209. /*
  210. for i, pos := range m.idx {
  211. logEntry, ts := readTs(m.buf, pos)
  212. event := &filer_pb.SubscribeMetadataResponse{}
  213. proto.Unmarshal(logEntry.Data, event)
  214. entry := event.EventNotification.OldEntry
  215. if entry == nil {
  216. entry = event.EventNotification.NewEntry
  217. }
  218. fmt.Printf("entry %d ts: %v offset:%d dir:%s name:%s\n", i, time.Unix(0, ts), pos, event.Directory, entry.Name)
  219. }
  220. fmt.Printf("l=%d, h=%d\n", l, h)
  221. */
  222. for l <= h {
  223. mid := (l + h) / 2
  224. pos := m.idx[mid]
  225. _, t := readTs(m.buf, pos)
  226. if t <= lastTs {
  227. l = mid + 1
  228. } else if lastTs < t {
  229. var prevT int64
  230. if mid > 0 {
  231. _, prevT = readTs(m.buf, m.idx[mid-1])
  232. }
  233. if prevT <= lastTs {
  234. // fmt.Printf("found l=%d, m-1=%d(ts=%d), m=%d(ts=%d), h=%d [%d, %d) \n", l, mid-1, prevT, mid, t, h, pos, m.pos)
  235. return copiedBytes(m.buf[pos:m.pos]), nil
  236. }
  237. h = mid
  238. }
  239. // fmt.Printf("l=%d, h=%d\n", l, h)
  240. }
  241. // FIXME: this could be that the buffer has been flushed already
  242. return nil, nil
  243. }
  244. func (m *LogBuffer) ReleaseMemory(b *bytes.Buffer) {
  245. bufferPool.Put(b)
  246. }
  247. var bufferPool = sync.Pool{
  248. New: func() interface{} {
  249. return new(bytes.Buffer)
  250. },
  251. }
  252. func copiedBytes(buf []byte) (copied *bytes.Buffer) {
  253. copied = bufferPool.Get().(*bytes.Buffer)
  254. copied.Reset()
  255. copied.Write(buf)
  256. return
  257. }
  258. func readTs(buf []byte, pos int) (size int, ts int64) {
  259. size = int(util.BytesToUint32(buf[pos : pos+4]))
  260. entryData := buf[pos+4 : pos+4+size]
  261. logEntry := &filer_pb.LogEntry{}
  262. err := proto.Unmarshal(entryData, logEntry)
  263. if err != nil {
  264. glog.Fatalf("unexpected unmarshal filer_pb.LogEntry: %v", err)
  265. }
  266. return size, logEntry.TsNs
  267. }