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.

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