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.

220 lines
5.0 KiB

7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
6 years ago
7 years ago
7 years ago
6 years ago
6 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
6 years ago
6 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
6 years ago
6 years ago
7 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
7 years ago
  1. package filer2
  2. import (
  3. "fmt"
  4. "hash/fnv"
  5. "sort"
  6. "sync"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. )
  9. func TotalSize(chunks []*filer_pb.FileChunk) (size uint64) {
  10. for _, c := range chunks {
  11. t := uint64(c.Offset + int64(c.Size))
  12. if size < t {
  13. size = t
  14. }
  15. }
  16. return
  17. }
  18. func ETag(chunks []*filer_pb.FileChunk) (etag string) {
  19. if len(chunks) == 1 {
  20. return chunks[0].ETag
  21. }
  22. h := fnv.New32a()
  23. for _, c := range chunks {
  24. h.Write([]byte(c.ETag))
  25. }
  26. return fmt.Sprintf("%x", h.Sum32())
  27. }
  28. func CompactFileChunks(chunks []*filer_pb.FileChunk) (compacted, garbage []*filer_pb.FileChunk) {
  29. visibles := NonOverlappingVisibleIntervals(chunks)
  30. fileIds := make(map[string]bool)
  31. for _, interval := range visibles {
  32. fileIds[interval.fileId] = true
  33. }
  34. for _, chunk := range chunks {
  35. if _, found := fileIds[chunk.GetFileIdString()]; found {
  36. compacted = append(compacted, chunk)
  37. } else {
  38. garbage = append(garbage, chunk)
  39. }
  40. }
  41. return
  42. }
  43. func MinusChunks(as, bs []*filer_pb.FileChunk) (delta []*filer_pb.FileChunk) {
  44. fileIds := make(map[string]bool)
  45. for _, interval := range bs {
  46. fileIds[interval.GetFileIdString()] = true
  47. }
  48. for _, chunk := range as {
  49. if _, found := fileIds[chunk.GetFileIdString()]; !found {
  50. delta = append(delta, chunk)
  51. }
  52. }
  53. return
  54. }
  55. type ChunkView struct {
  56. FileId string
  57. Offset int64
  58. Size uint64
  59. LogicOffset int64
  60. IsFullChunk bool
  61. CipherKey []byte
  62. isGzipped bool
  63. }
  64. func ViewFromChunks(chunks []*filer_pb.FileChunk, offset int64, size int) (views []*ChunkView) {
  65. visibles := NonOverlappingVisibleIntervals(chunks)
  66. return ViewFromVisibleIntervals(visibles, offset, size)
  67. }
  68. func ViewFromVisibleIntervals(visibles []VisibleInterval, offset int64, size int) (views []*ChunkView) {
  69. stop := offset + int64(size)
  70. for _, chunk := range visibles {
  71. if chunk.start <= offset && offset < chunk.stop && offset < stop {
  72. isFullChunk := chunk.isFullChunk && chunk.start == offset && chunk.stop <= stop
  73. views = append(views, &ChunkView{
  74. FileId: chunk.fileId,
  75. Offset: offset - chunk.start, // offset is the data starting location in this file id
  76. Size: uint64(min(chunk.stop, stop) - offset),
  77. LogicOffset: offset,
  78. IsFullChunk: isFullChunk,
  79. CipherKey: chunk.cipherKey,
  80. isGzipped: chunk.isGzipped,
  81. })
  82. offset = min(chunk.stop, stop)
  83. }
  84. }
  85. return views
  86. }
  87. func logPrintf(name string, visibles []VisibleInterval) {
  88. /*
  89. log.Printf("%s len %d", name, len(visibles))
  90. for _, v := range visibles {
  91. log.Printf("%s: => %+v", name, v)
  92. }
  93. */
  94. }
  95. var bufPool = sync.Pool{
  96. New: func() interface{} {
  97. return new(VisibleInterval)
  98. },
  99. }
  100. func MergeIntoVisibles(visibles, newVisibles []VisibleInterval, chunk *filer_pb.FileChunk) []VisibleInterval {
  101. newV := newVisibleInterval(chunk.Offset, chunk.Offset+int64(chunk.Size), chunk.GetFileIdString(), chunk.Mtime, true, chunk.CipherKey, chunk.IsGzipped)
  102. length := len(visibles)
  103. if length == 0 {
  104. return append(visibles, newV)
  105. }
  106. last := visibles[length-1]
  107. if last.stop <= chunk.Offset {
  108. return append(visibles, newV)
  109. }
  110. logPrintf(" before", visibles)
  111. for _, v := range visibles {
  112. if v.start < chunk.Offset && chunk.Offset < v.stop {
  113. newVisibles = append(newVisibles, newVisibleInterval(v.start, chunk.Offset, v.fileId, v.modifiedTime, false, v.cipherKey, v.isGzipped))
  114. }
  115. chunkStop := chunk.Offset + int64(chunk.Size)
  116. if v.start < chunkStop && chunkStop < v.stop {
  117. newVisibles = append(newVisibles, newVisibleInterval(chunkStop, v.stop, v.fileId, v.modifiedTime, false, v.cipherKey, v.isGzipped))
  118. }
  119. if chunkStop <= v.start || v.stop <= chunk.Offset {
  120. newVisibles = append(newVisibles, v)
  121. }
  122. }
  123. newVisibles = append(newVisibles, newV)
  124. logPrintf(" append", newVisibles)
  125. for i := len(newVisibles) - 1; i >= 0; i-- {
  126. if i > 0 && newV.start < newVisibles[i-1].start {
  127. newVisibles[i] = newVisibles[i-1]
  128. } else {
  129. newVisibles[i] = newV
  130. break
  131. }
  132. }
  133. logPrintf(" sorted", newVisibles)
  134. return newVisibles
  135. }
  136. func NonOverlappingVisibleIntervals(chunks []*filer_pb.FileChunk) (visibles []VisibleInterval) {
  137. sort.Slice(chunks, func(i, j int) bool {
  138. return chunks[i].Mtime < chunks[j].Mtime
  139. })
  140. var newVisibles []VisibleInterval
  141. for _, chunk := range chunks {
  142. newVisibles = MergeIntoVisibles(visibles, newVisibles, chunk)
  143. t := visibles[:0]
  144. visibles = newVisibles
  145. newVisibles = t
  146. logPrintf("add", visibles)
  147. }
  148. return
  149. }
  150. // find non-overlapping visible intervals
  151. // visible interval map to one file chunk
  152. type VisibleInterval struct {
  153. start int64
  154. stop int64
  155. modifiedTime int64
  156. fileId string
  157. isFullChunk bool
  158. cipherKey []byte
  159. isGzipped bool
  160. }
  161. func newVisibleInterval(start, stop int64, fileId string, modifiedTime int64, isFullChunk bool, cipherKey []byte, isGzipped bool) VisibleInterval {
  162. return VisibleInterval{
  163. start: start,
  164. stop: stop,
  165. fileId: fileId,
  166. modifiedTime: modifiedTime,
  167. isFullChunk: isFullChunk,
  168. cipherKey: cipherKey,
  169. isGzipped: isGzipped,
  170. }
  171. }
  172. func min(x, y int64) int64 {
  173. if x <= y {
  174. return x
  175. }
  176. return y
  177. }