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.

213 lines
6.8 KiB

  1. package storage
  2. import (
  3. "fmt"
  4. "io"
  5. "io/ioutil"
  6. "net/url"
  7. "os"
  8. "sort"
  9. "strconv"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/operation"
  12. "github.com/chrislusf/seaweedfs/weed/util"
  13. )
  14. // The volume sync with a master volume via 2 steps:
  15. // 1. The slave checks master side to find subscription checkpoint
  16. // to setup the replication.
  17. // 2. The slave receives the updates from master
  18. /*
  19. Assume the slave volume needs to follow the master volume.
  20. The master volume could be compacted, and could be many files ahead of
  21. slave volume.
  22. Step 1:
  23. The slave volume will ask the master volume for a snapshot
  24. of (existing file entries, last offset, number of compacted times).
  25. For each entry x in master existing file entries:
  26. if x does not exist locally:
  27. add x locally
  28. For each entry y in local slave existing file entries:
  29. if y does not exist on master:
  30. delete y locally
  31. Step 2:
  32. After this, use the last offset and number of compacted times to request
  33. the master volume to send a new file, and keep looping. If the number of
  34. compacted times is changed, go back to step 1 (very likely this can be
  35. optimized more later).
  36. */
  37. func (v *Volume) Synchronize(volumeServer string) (err error) {
  38. var lastCompactRevision uint16 = 0
  39. var compactRevision uint16 = 0
  40. var masterMap CompactMap
  41. for i := 0; i < 3; i++ {
  42. if masterMap, _, compactRevision, err = fetchVolumeFileEntries(volumeServer, v.Id); err != nil {
  43. return fmt.Errorf("Failed to sync volume %d entries with %s: %v", v.Id, volumeServer, err)
  44. }
  45. if lastCompactRevision != compactRevision && lastCompactRevision != 0 {
  46. if err = v.Compact(); err != nil {
  47. return fmt.Errorf("Compact Volume before synchronizing %v", err)
  48. }
  49. if err = v.commitCompact(); err != nil {
  50. return fmt.Errorf("Commit Compact before synchronizing %v", err)
  51. }
  52. }
  53. lastCompactRevision = compactRevision
  54. if err = v.trySynchronizing(volumeServer, masterMap, compactRevision); err == nil {
  55. return
  56. }
  57. }
  58. return
  59. }
  60. type ByOffset []NeedleValue
  61. func (a ByOffset) Len() int { return len(a) }
  62. func (a ByOffset) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
  63. func (a ByOffset) Less(i, j int) bool { return a[i].Offset < a[j].Offset }
  64. // trySynchronizing sync with remote volume server incrementally by
  65. // make up the local and remote delta.
  66. func (v *Volume) trySynchronizing(volumeServer string, masterMap CompactMap, compactRevision uint16) error {
  67. slaveIdxFile, err := os.Open(v.nm.IndexFileName())
  68. if err != nil {
  69. return fmt.Errorf("Open volume %d index file: %v", v.Id, err)
  70. }
  71. defer slaveIdxFile.Close()
  72. slaveMap, err := LoadNeedleMap(slaveIdxFile)
  73. if err != nil {
  74. return fmt.Errorf("Load volume %d index file: %v", v.Id, err)
  75. }
  76. var delta []NeedleValue
  77. if err := masterMap.Visit(func(needleValue NeedleValue) error {
  78. if needleValue.Key == 0 {
  79. return nil
  80. }
  81. if _, ok := slaveMap.Get(uint64(needleValue.Key)); ok {
  82. return nil // skip intersection
  83. }
  84. delta = append(delta, needleValue)
  85. return nil
  86. }); err != nil {
  87. return fmt.Errorf("Add master entry: %v", err)
  88. }
  89. if err := slaveMap.m.Visit(func(needleValue NeedleValue) error {
  90. if needleValue.Key == 0 {
  91. return nil
  92. }
  93. if _, ok := masterMap.Get(needleValue.Key); ok {
  94. return nil // skip intersection
  95. }
  96. needleValue.Size = 0
  97. delta = append(delta, needleValue)
  98. return nil
  99. }); err != nil {
  100. return fmt.Errorf("Remove local entry: %v", err)
  101. }
  102. // simulate to same ordering of remote .dat file needle entries
  103. sort.Sort(ByOffset(delta))
  104. // make up the delta
  105. fetchCount := 0
  106. volumeDataContentHandlerUrl := "http://" + volumeServer + "/admin/sync/data"
  107. for _, needleValue := range delta {
  108. if needleValue.Size == 0 {
  109. // remove file entry from local
  110. v.removeNeedle(needleValue.Key)
  111. continue
  112. }
  113. // add master file entry to local data file
  114. if err := v.fetchNeedle(volumeDataContentHandlerUrl, needleValue, compactRevision); err != nil {
  115. glog.V(0).Infof("Fetch needle %v from %s: %v", needleValue, volumeServer, err)
  116. return err
  117. }
  118. fetchCount++
  119. }
  120. glog.V(1).Infof("Fetched %d needles from %s", fetchCount, volumeServer)
  121. return nil
  122. }
  123. func fetchVolumeFileEntries(volumeServer string, vid VolumeId) (m CompactMap, lastOffset uint64, compactRevision uint16, err error) {
  124. m = NewCompactMap()
  125. syncStatus, err := operation.GetVolumeSyncStatus(volumeServer, vid.String())
  126. if err != nil {
  127. return m, 0, 0, err
  128. }
  129. total := 0
  130. err = operation.GetVolumeIdxEntries(volumeServer, vid.String(), func(key uint64, offset, size uint32) {
  131. // println("remote key", key, "offset", offset*NeedlePaddingSize, "size", size)
  132. if offset != 0 && size != 0 {
  133. m.Set(Key(key), offset, size)
  134. } else {
  135. m.Delete(Key(key))
  136. }
  137. total++
  138. })
  139. glog.V(2).Infof("server %s volume %d, entries %d, last offset %d, revision %d", volumeServer, vid, total, syncStatus.TailOffset, syncStatus.CompactRevision)
  140. return m, syncStatus.TailOffset, syncStatus.CompactRevision, err
  141. }
  142. func (v *Volume) GetVolumeSyncStatus() operation.SyncVolumeResponse {
  143. var syncStatus = operation.SyncVolumeResponse{}
  144. if stat, err := v.dataFile.Stat(); err == nil {
  145. syncStatus.TailOffset = uint64(stat.Size())
  146. }
  147. syncStatus.IdxFileSize = v.nm.IndexFileSize()
  148. syncStatus.CompactRevision = v.SuperBlock.CompactRevision
  149. syncStatus.Ttl = v.SuperBlock.Ttl.String()
  150. syncStatus.Replication = v.SuperBlock.ReplicaPlacement.String()
  151. return syncStatus
  152. }
  153. func (v *Volume) IndexFileContent() ([]byte, error) {
  154. return v.nm.IndexFileContent()
  155. }
  156. // removeNeedle removes one needle by needle key
  157. func (v *Volume) removeNeedle(key Key) {
  158. n := new(Needle)
  159. n.Id = uint64(key)
  160. v.deleteNeedle(n)
  161. }
  162. // fetchNeedle fetches a remote volume needle by vid, id, offset
  163. // The compact revision is checked first in case the remote volume
  164. // is compacted and the offset is invalid any more.
  165. func (v *Volume) fetchNeedle(volumeDataContentHandlerUrl string,
  166. needleValue NeedleValue, compactRevision uint16) error {
  167. // add master file entry to local data file
  168. values := make(url.Values)
  169. values.Add("revision", strconv.Itoa(int(compactRevision)))
  170. values.Add("volume", v.Id.String())
  171. values.Add("id", needleValue.Key.String())
  172. values.Add("offset", strconv.FormatUint(uint64(needleValue.Offset), 10))
  173. values.Add("size", strconv.FormatUint(uint64(needleValue.Size), 10))
  174. glog.V(4).Infof("Fetch %+v", needleValue)
  175. return util.GetUrlStream(volumeDataContentHandlerUrl, values, func(r io.Reader) error {
  176. b, err := ioutil.ReadAll(r)
  177. if err != nil {
  178. return fmt.Errorf("Reading from %s error: %v", volumeDataContentHandlerUrl, err)
  179. }
  180. offset, err := v.AppendBlob(b)
  181. if err != nil {
  182. return fmt.Errorf("Appending volume %d error: %v", v.Id, err)
  183. }
  184. // println("add key", needleValue.Key, "offset", offset, "size", needleValue.Size)
  185. v.nm.Put(uint64(needleValue.Key), uint32(offset/NeedlePaddingSize), needleValue.Size)
  186. return nil
  187. })
  188. }