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.

212 lines
4.6 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
  1. package queue
  2. import (
  3. "sync"
  4. "time"
  5. "github.com/golang/protobuf/proto"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. )
  10. type dataToFlush struct {
  11. startTime time.Time
  12. stopTime time.Time
  13. data []byte
  14. }
  15. type LogBuffer struct {
  16. buf []byte
  17. idx []int
  18. pos int
  19. startTime time.Time
  20. stopTime time.Time
  21. sizeBuf []byte
  22. flushInterval time.Duration
  23. flushFn func(startTime, stopTime time.Time, buf []byte)
  24. notifyFn func()
  25. isStopping bool
  26. flushChan chan *dataToFlush
  27. sync.RWMutex
  28. }
  29. func NewLogBuffer(flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer {
  30. lb := &LogBuffer{
  31. buf: make([]byte, 4*1024*1024),
  32. sizeBuf: make([]byte, 4),
  33. flushInterval: flushInterval,
  34. flushFn: flushFn,
  35. notifyFn: notifyFn,
  36. flushChan: make(chan *dataToFlush, 256),
  37. }
  38. go lb.loopFlush()
  39. go lb.loopInterval()
  40. return lb
  41. }
  42. func (m *LogBuffer) AddToBuffer(ts time.Time, key, data []byte) {
  43. logEntry := &filer_pb.LogEntry{
  44. TsNs: ts.UnixNano(),
  45. PartitionKeyHash: util.HashToInt32(key),
  46. Data: data,
  47. }
  48. logEntryData, _ := proto.Marshal(logEntry)
  49. size := len(logEntryData)
  50. m.Lock()
  51. defer func() {
  52. m.Unlock()
  53. if m.notifyFn != nil {
  54. m.notifyFn()
  55. }
  56. }()
  57. if m.pos == 0 {
  58. m.startTime = ts
  59. }
  60. if m.startTime.Add(m.flushInterval).Before(ts) || len(m.buf)-m.pos < size+4 {
  61. m.flushChan <- m.copyToFlush()
  62. m.startTime = ts
  63. if len(m.buf) < size+4 {
  64. m.buf = make([]byte, 2*size+4)
  65. }
  66. }
  67. m.stopTime = ts
  68. m.idx = append(m.idx, m.pos)
  69. util.Uint32toBytes(m.sizeBuf, uint32(size))
  70. copy(m.buf[m.pos:m.pos+4], m.sizeBuf)
  71. copy(m.buf[m.pos+4:m.pos+4+size], logEntryData)
  72. m.pos += size + 4
  73. }
  74. func (m *LogBuffer) Shutdown() {
  75. if m.isStopping {
  76. return
  77. }
  78. m.isStopping = true
  79. m.Lock()
  80. toFlush := m.copyToFlush()
  81. m.Unlock()
  82. m.flushChan <- toFlush
  83. close(m.flushChan)
  84. }
  85. func (m *LogBuffer) loopFlush() {
  86. for d := range m.flushChan {
  87. if d != nil {
  88. m.flushFn(d.startTime, d.stopTime, d.data)
  89. }
  90. }
  91. }
  92. func (m *LogBuffer) loopInterval() {
  93. for !m.isStopping {
  94. m.Lock()
  95. toFlush := m.copyToFlush()
  96. m.Unlock()
  97. m.flushChan <- toFlush
  98. time.Sleep(m.flushInterval)
  99. }
  100. }
  101. func (m *LogBuffer) copyToFlush() *dataToFlush {
  102. if m.flushFn != nil && m.pos > 0 {
  103. // fmt.Printf("flush buffer %d pos %d empty space %d\n", len(m.buf), m.pos, len(m.buf)-m.pos)
  104. d := &dataToFlush{
  105. startTime: m.startTime,
  106. stopTime: m.stopTime,
  107. data: copiedBytes(m.buf[:m.pos]),
  108. }
  109. m.pos = 0
  110. m.idx = m.idx[:0]
  111. return d
  112. }
  113. return nil
  114. }
  115. func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (ts time.Time, bufferCopy []byte) {
  116. m.RLock()
  117. defer m.RUnlock()
  118. // fmt.Printf("read from buffer: %v\n", lastReadTime)
  119. if lastReadTime.Equal(m.stopTime) {
  120. return lastReadTime, nil
  121. }
  122. if lastReadTime.After(m.stopTime) {
  123. // glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadTime, m.stopTime)
  124. return lastReadTime, nil
  125. }
  126. if lastReadTime.Before(m.startTime) {
  127. return m.stopTime, copiedBytes(m.buf[:m.pos])
  128. }
  129. lastTs := lastReadTime.UnixNano()
  130. l, h := 0, len(m.idx)-1
  131. /*
  132. for i, pos := range m.idx {
  133. logEntry, ts := readTs(m.buf, pos)
  134. event := &filer_pb.FullEventNotification{}
  135. proto.Unmarshal(logEntry.Data, event)
  136. entry := event.EventNotification.OldEntry
  137. if entry == nil {
  138. entry = event.EventNotification.NewEntry
  139. }
  140. fmt.Printf("entry %d ts: %v offset:%d dir:%s name:%s\n", i, time.Unix(0, ts), pos, event.Directory, entry.Name)
  141. }
  142. fmt.Printf("l=%d, h=%d\n", l, h)
  143. */
  144. for l <= h {
  145. mid := (l + h) / 2
  146. pos := m.idx[mid]
  147. _, t := readTs(m.buf, m.idx[mid])
  148. if t <= lastTs {
  149. l = mid + 1
  150. } else if lastTs < t {
  151. var prevT int64
  152. if mid > 0 {
  153. _, prevT = readTs(m.buf, m.idx[mid-1])
  154. }
  155. if prevT <= lastTs {
  156. // println("found mid = ", mid)
  157. return time.Unix(0, t), copiedBytes(m.buf[pos:m.pos])
  158. }
  159. h = mid - 1
  160. }
  161. // fmt.Printf("l=%d, h=%d\n", l, h)
  162. }
  163. // FIXME: this could be that the buffer has been flushed already
  164. // println("not found")
  165. return lastReadTime, nil
  166. }
  167. func copiedBytes(buf []byte) (copied []byte) {
  168. copied = make([]byte, len(buf))
  169. copy(copied, buf)
  170. return
  171. }
  172. func readTs(buf []byte, pos int) (*filer_pb.LogEntry, int64) {
  173. size := util.BytesToUint32(buf[pos : pos+4])
  174. entryData := buf[pos+4 : pos+4+int(size)]
  175. logEntry := &filer_pb.LogEntry{}
  176. err := proto.Unmarshal(entryData, logEntry)
  177. if err != nil {
  178. glog.Fatalf("unexpected unmarshal filer_pb.LogEntry: %v", err)
  179. }
  180. return logEntry, logEntry.TsNs
  181. }