volume_loading.go 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "github.com/syndtr/goleveldb/leveldb/opt"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/stats"
  8. "github.com/seaweedfs/seaweedfs/weed/storage/backend"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/super_block"
  11. "github.com/seaweedfs/seaweedfs/weed/util"
  12. )
  13. func loadVolumeWithoutIndex(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind) (v *Volume, err error) {
  14. v = &Volume{dir: dirname, Collection: collection, Id: id}
  15. v.SuperBlock = super_block.SuperBlock{}
  16. v.needleMapKind = needleMapKind
  17. err = v.load(false, false, needleMapKind, 0)
  18. return
  19. }
  20. func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind NeedleMapKind, preallocate int64) (err error) {
  21. alreadyHasSuperBlock := false
  22. hasLoadedVolume := false
  23. defer func() {
  24. if !hasLoadedVolume {
  25. if v.nm != nil {
  26. v.nm.Close()
  27. v.nm = nil
  28. }
  29. if v.DataBackend != nil {
  30. v.DataBackend.Close()
  31. v.DataBackend = nil
  32. }
  33. }
  34. }()
  35. hasVolumeInfoFile := v.maybeLoadVolumeInfo()
  36. if v.HasRemoteFile() {
  37. v.noWriteCanDelete = true
  38. v.noWriteOrDelete = false
  39. glog.V(0).Infof("loading volume %d from remote %v", v.Id, v.volumeInfo)
  40. v.LoadRemoteFile()
  41. alreadyHasSuperBlock = true
  42. } else if exists, canRead, canWrite, modifiedTime, fileSize := util.CheckFile(v.FileName(".dat")); exists {
  43. // open dat file
  44. if !canRead {
  45. return fmt.Errorf("cannot read Volume Data file %s", v.FileName(".dat"))
  46. }
  47. var dataFile *os.File
  48. if canWrite {
  49. dataFile, err = os.OpenFile(v.FileName(".dat"), os.O_RDWR|os.O_CREATE, 0644)
  50. } else {
  51. glog.V(0).Infof("opening %s in READONLY mode", v.FileName(".dat"))
  52. dataFile, err = os.Open(v.FileName(".dat"))
  53. v.noWriteOrDelete = true
  54. }
  55. v.lastModifiedTsSeconds = uint64(modifiedTime.Unix())
  56. if fileSize >= super_block.SuperBlockSize {
  57. alreadyHasSuperBlock = true
  58. }
  59. v.DataBackend = backend.NewDiskFile(dataFile)
  60. } else {
  61. if createDatIfMissing {
  62. v.DataBackend, err = backend.CreateVolumeFile(v.FileName(".dat"), preallocate, v.MemoryMapMaxSizeMb)
  63. } else {
  64. return fmt.Errorf("volume data file %s does not exist", v.FileName(".dat"))
  65. }
  66. }
  67. if err != nil {
  68. if !os.IsPermission(err) {
  69. return fmt.Errorf("cannot load volume data %s: %v", v.FileName(".dat"), err)
  70. } else {
  71. return fmt.Errorf("load data file %s: %v", v.FileName(".dat"), err)
  72. }
  73. }
  74. if alreadyHasSuperBlock {
  75. err = v.readSuperBlock()
  76. if err == nil {
  77. v.volumeInfo.Version = uint32(v.SuperBlock.Version)
  78. }
  79. glog.V(0).Infof("readSuperBlock volume %d version %v", v.Id, v.SuperBlock.Version)
  80. if v.HasRemoteFile() {
  81. // maybe temporary network problem
  82. glog.Errorf("readSuperBlock remote volume %d: %v", v.Id, err)
  83. err = nil
  84. }
  85. } else {
  86. if !v.SuperBlock.Initialized() {
  87. return fmt.Errorf("volume %s not initialized", v.FileName(".dat"))
  88. }
  89. err = v.maybeWriteSuperBlock()
  90. }
  91. if err == nil && alsoLoadIndex {
  92. // adjust for existing volumes with .idx together with .dat files
  93. if v.dirIdx != v.dir {
  94. if util.FileExists(v.DataFileName() + ".idx") {
  95. v.dirIdx = v.dir
  96. }
  97. }
  98. // check volume idx files
  99. if err := v.checkIdxFile(); err != nil {
  100. glog.Fatalf("check volume idx file %s: %v", v.FileName(".idx"), err)
  101. }
  102. var indexFile *os.File
  103. if v.noWriteOrDelete {
  104. glog.V(0).Infoln("open to read file", v.FileName(".idx"))
  105. if indexFile, err = os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644); err != nil {
  106. return fmt.Errorf("cannot read Volume Index %s: %v", v.FileName(".idx"), err)
  107. }
  108. } else {
  109. glog.V(1).Infoln("open to write file", v.FileName(".idx"))
  110. if indexFile, err = os.OpenFile(v.FileName(".idx"), os.O_RDWR|os.O_CREATE, 0644); err != nil {
  111. return fmt.Errorf("cannot write Volume Index %s: %v", v.FileName(".idx"), err)
  112. }
  113. }
  114. if v.lastAppendAtNs, err = CheckAndFixVolumeDataIntegrity(v, indexFile); err != nil {
  115. v.noWriteOrDelete = true
  116. glog.V(0).Infof("volumeDataIntegrityChecking failed %v", err)
  117. }
  118. if v.noWriteOrDelete || v.noWriteCanDelete {
  119. if v.nm, err = NewSortedFileNeedleMap(v.IndexFileName(), indexFile); err != nil {
  120. glog.V(0).Infof("loading sorted db %s error: %v", v.FileName(".sdx"), err)
  121. }
  122. } else {
  123. switch needleMapKind {
  124. case NeedleMapInMemory:
  125. if v.tmpNm != nil {
  126. glog.V(0).Infof("updating memory compact index %s ", v.FileName(".idx"))
  127. err = v.tmpNm.UpdateNeedleMap(v, indexFile, nil)
  128. } else {
  129. glog.V(0).Infoln("loading memory index", v.FileName(".idx"), "to memory")
  130. if v.nm, err = LoadCompactNeedleMap(indexFile); err != nil {
  131. glog.V(0).Infof("loading index %s to memory error: %v", v.FileName(".idx"), err)
  132. }
  133. }
  134. case NeedleMapLevelDb:
  135. opts := &opt.Options{
  136. BlockCacheCapacity: 2 * 1024 * 1024, // default value is 8MiB
  137. WriteBuffer: 1 * 1024 * 1024, // default value is 4MiB
  138. CompactionTableSizeMultiplier: 10, // default value is 1
  139. }
  140. if v.tmpNm != nil {
  141. glog.V(0).Infoln("updating leveldb index", v.FileName(".ldb"))
  142. err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts)
  143. } else {
  144. glog.V(0).Infoln("loading leveldb index", v.FileName(".ldb"))
  145. if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
  146. glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
  147. }
  148. }
  149. case NeedleMapLevelDbMedium:
  150. opts := &opt.Options{
  151. BlockCacheCapacity: 4 * 1024 * 1024, // default value is 8MiB
  152. WriteBuffer: 2 * 1024 * 1024, // default value is 4MiB
  153. CompactionTableSizeMultiplier: 10, // default value is 1
  154. }
  155. if v.tmpNm != nil {
  156. glog.V(0).Infoln("updating leveldb medium index", v.FileName(".ldb"))
  157. err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts)
  158. } else {
  159. glog.V(0).Infoln("loading leveldb medium index", v.FileName(".ldb"))
  160. if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
  161. glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
  162. }
  163. }
  164. case NeedleMapLevelDbLarge:
  165. opts := &opt.Options{
  166. BlockCacheCapacity: 8 * 1024 * 1024, // default value is 8MiB
  167. WriteBuffer: 4 * 1024 * 1024, // default value is 4MiB
  168. CompactionTableSizeMultiplier: 10, // default value is 1
  169. }
  170. if v.tmpNm != nil {
  171. glog.V(0).Infoln("updating leveldb large index", v.FileName(".ldb"))
  172. err = v.tmpNm.UpdateNeedleMap(v, indexFile, opts)
  173. } else {
  174. glog.V(0).Infoln("loading leveldb large index", v.FileName(".ldb"))
  175. if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts); err != nil {
  176. glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
  177. }
  178. }
  179. }
  180. }
  181. }
  182. if !hasVolumeInfoFile {
  183. v.volumeInfo.Version = uint32(v.SuperBlock.Version)
  184. v.SaveVolumeInfo()
  185. }
  186. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Inc()
  187. if err == nil {
  188. hasLoadedVolume = true
  189. }
  190. return err
  191. }