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.

222 lines
4.6 KiB

7 years ago
6 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 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
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
6 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
  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.FileId]; found {
  36. compacted = append(compacted, chunk)
  37. } else {
  38. garbage = append(garbage, chunk)
  39. }
  40. }
  41. return
  42. }
  43. func FindUnusedFileChunks(oldChunks, newChunks []*filer_pb.FileChunk) (unused []*filer_pb.FileChunk) {
  44. fileIds := make(map[string]bool)
  45. for _, interval := range newChunks {
  46. fileIds[interval.FileId] = true
  47. }
  48. for _, chunk := range oldChunks {
  49. if found := fileIds[chunk.FileId]; !found {
  50. unused = append(unused, 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. }
  62. func ViewFromChunks(chunks []*filer_pb.FileChunk, offset int64, size int) (views []*ChunkView) {
  63. visibles := nonOverlappingVisibleIntervals(chunks)
  64. stop := offset + int64(size)
  65. for _, chunk := range visibles {
  66. if chunk.start <= offset && offset < chunk.stop && offset < stop {
  67. isFullChunk := chunk.isFullChunk && chunk.start == offset && chunk.stop <= stop
  68. views = append(views, &ChunkView{
  69. FileId: chunk.fileId,
  70. Offset: offset - chunk.start, // offset is the data starting location in this file id
  71. Size: uint64(min(chunk.stop, stop) - offset),
  72. LogicOffset: offset,
  73. IsFullChunk: isFullChunk,
  74. })
  75. offset = min(chunk.stop, stop)
  76. }
  77. }
  78. return views
  79. }
  80. func logPrintf(name string, visibles []*visibleInterval) {
  81. /*
  82. log.Printf("%s len %d", name, len(visibles))
  83. for _, v := range visibles {
  84. log.Printf("%s: => %+v", name, v)
  85. }
  86. */
  87. }
  88. var bufPool = sync.Pool{
  89. New: func() interface{} {
  90. return new(visibleInterval)
  91. },
  92. }
  93. func mergeIntoVisibles(visibles, newVisibles []*visibleInterval, chunk *filer_pb.FileChunk) []*visibleInterval {
  94. newV := newVisibleInterval(
  95. chunk.Offset,
  96. chunk.Offset+int64(chunk.Size),
  97. chunk.FileId,
  98. chunk.Mtime,
  99. true,
  100. )
  101. length := len(visibles)
  102. if length == 0 {
  103. return append(visibles, newV)
  104. }
  105. last := visibles[length-1]
  106. if last.stop <= chunk.Offset {
  107. return append(visibles, newV)
  108. }
  109. logPrintf(" before", visibles)
  110. for _, v := range visibles {
  111. if v.start < chunk.Offset && chunk.Offset < v.stop {
  112. newVisibles = append(newVisibles, newVisibleInterval(
  113. v.start,
  114. chunk.Offset,
  115. v.fileId,
  116. v.modifiedTime,
  117. false,
  118. ))
  119. }
  120. chunkStop := chunk.Offset + int64(chunk.Size)
  121. if v.start < chunkStop && chunkStop < v.stop {
  122. newVisibles = append(newVisibles, newVisibleInterval(
  123. chunkStop,
  124. v.stop,
  125. v.fileId,
  126. v.modifiedTime,
  127. false,
  128. ))
  129. }
  130. if chunkStop <= v.start || v.stop <= chunk.Offset {
  131. newVisibles = append(newVisibles, v)
  132. }
  133. }
  134. newVisibles = append(newVisibles, newV)
  135. logPrintf(" append", newVisibles)
  136. for i := len(newVisibles) - 1; i >= 0; i-- {
  137. if i > 0 && newV.start < newVisibles[i-1].start {
  138. newVisibles[i] = newVisibles[i-1]
  139. } else {
  140. newVisibles[i] = newV
  141. break
  142. }
  143. }
  144. logPrintf(" sorted", newVisibles)
  145. return newVisibles
  146. }
  147. func nonOverlappingVisibleIntervals(chunks []*filer_pb.FileChunk) (visibles []*visibleInterval) {
  148. sort.Slice(chunks, func(i, j int) bool {
  149. return chunks[i].Mtime < chunks[j].Mtime
  150. })
  151. var newVislbles []*visibleInterval
  152. for _, chunk := range chunks {
  153. newVislbles = mergeIntoVisibles(visibles, newVislbles, chunk)
  154. t := visibles[:0]
  155. visibles = newVislbles
  156. newVislbles = t
  157. logPrintf("add", visibles)
  158. }
  159. return
  160. }
  161. // find non-overlapping visible intervals
  162. // visible interval map to one file chunk
  163. type visibleInterval struct {
  164. start int64
  165. stop int64
  166. modifiedTime int64
  167. fileId string
  168. isFullChunk bool
  169. }
  170. func newVisibleInterval(start, stop int64, fileId string, modifiedTime int64, isFullChunk bool) *visibleInterval {
  171. return &visibleInterval{
  172. start: start,
  173. stop: stop,
  174. fileId: fileId,
  175. modifiedTime: modifiedTime,
  176. isFullChunk: isFullChunk,
  177. }
  178. }
  179. func min(x, y int64) int64 {
  180. if x <= y {
  181. return x
  182. }
  183. return y
  184. }