store.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602
  1. package storage
  2. import (
  3. "fmt"
  4. "io"
  5. "path/filepath"
  6. "strings"
  7. "sync"
  8. "sync/atomic"
  9. "github.com/seaweedfs/seaweedfs/weed/pb"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
  11. "github.com/seaweedfs/seaweedfs/weed/util"
  12. "google.golang.org/grpc"
  13. "github.com/seaweedfs/seaweedfs/weed/glog"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/stats"
  16. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  17. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  18. "github.com/seaweedfs/seaweedfs/weed/storage/super_block"
  19. . "github.com/seaweedfs/seaweedfs/weed/storage/types"
  20. )
  21. const (
  22. MAX_TTL_VOLUME_REMOVAL_DELAY = 10 // 10 minutes
  23. )
  24. type ReadOption struct {
  25. // request
  26. ReadDeleted bool
  27. AttemptMetaOnly bool
  28. MustMetaOnly bool
  29. // response
  30. IsMetaOnly bool // read status
  31. VolumeRevision uint16
  32. IsOutOfRange bool // whether read over MaxPossibleVolumeSize
  33. // If HasSlowRead is set to true:
  34. // * read requests and write requests compete for the lock.
  35. // * large file read P99 latency on busy sites will go up, due to the need to get locks multiple times.
  36. // * write requests will see lower latency.
  37. // If HasSlowRead is set to false:
  38. // * read requests should complete asap, not blocking other requests.
  39. // * write requests may see high latency when downloading large files.
  40. HasSlowRead bool
  41. // increasing ReadBufferSize can reduce the number of get locks times and shorten read P99 latency.
  42. // but will increase memory usage a bit. Use with hasSlowRead normally.
  43. ReadBufferSize int
  44. }
  45. /*
  46. * A VolumeServer contains one Store
  47. */
  48. type Store struct {
  49. MasterAddress pb.ServerAddress
  50. grpcDialOption grpc.DialOption
  51. volumeSizeLimit uint64 // read from the master
  52. Ip string
  53. Port int
  54. GrpcPort int
  55. PublicUrl string
  56. Locations []*DiskLocation
  57. dataCenter string // optional information, overwriting master setting if exists
  58. rack string // optional information, overwriting master setting if exists
  59. connected bool
  60. NeedleMapKind NeedleMapKind
  61. NewVolumesChan chan master_pb.VolumeShortInformationMessage
  62. DeletedVolumesChan chan master_pb.VolumeShortInformationMessage
  63. NewEcShardsChan chan master_pb.VolumeEcShardInformationMessage
  64. DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage
  65. isStopping bool
  66. }
  67. func (s *Store) String() (str string) {
  68. str = fmt.Sprintf("Ip:%s, Port:%d, GrpcPort:%d PublicUrl:%s, dataCenter:%s, rack:%s, connected:%v, volumeSizeLimit:%d", s.Ip, s.Port, s.GrpcPort, s.PublicUrl, s.dataCenter, s.rack, s.connected, s.GetVolumeSizeLimit())
  69. return
  70. }
  71. func NewStore(grpcDialOption grpc.DialOption, ip string, port int, grpcPort int, publicUrl string, dirnames []string, maxVolumeCounts []int32,
  72. minFreeSpaces []util.MinFreeSpace, idxFolder string, needleMapKind NeedleMapKind, diskTypes []DiskType, ldbTimeout int64) (s *Store) {
  73. s = &Store{grpcDialOption: grpcDialOption, Port: port, Ip: ip, GrpcPort: grpcPort, PublicUrl: publicUrl, NeedleMapKind: needleMapKind}
  74. s.Locations = make([]*DiskLocation, 0)
  75. var wg sync.WaitGroup
  76. for i := 0; i < len(dirnames); i++ {
  77. location := NewDiskLocation(dirnames[i], int32(maxVolumeCounts[i]), minFreeSpaces[i], idxFolder, diskTypes[i])
  78. s.Locations = append(s.Locations, location)
  79. stats.VolumeServerMaxVolumeCounter.Add(float64(maxVolumeCounts[i]))
  80. wg.Add(1)
  81. go func() {
  82. defer wg.Done()
  83. location.loadExistingVolumes(needleMapKind, ldbTimeout)
  84. }()
  85. }
  86. wg.Wait()
  87. s.NewVolumesChan = make(chan master_pb.VolumeShortInformationMessage, 3)
  88. s.DeletedVolumesChan = make(chan master_pb.VolumeShortInformationMessage, 3)
  89. s.NewEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
  90. s.DeletedEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
  91. return
  92. }
  93. func (s *Store) AddVolume(volumeId needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement string, ttlString string, preallocate int64, MemoryMapMaxSizeMb uint32, diskType DiskType, ldbTimeout int64) error {
  94. rt, e := super_block.NewReplicaPlacementFromString(replicaPlacement)
  95. if e != nil {
  96. return e
  97. }
  98. ttl, e := needle.ReadTTL(ttlString)
  99. if e != nil {
  100. return e
  101. }
  102. e = s.addVolume(volumeId, collection, needleMapKind, rt, ttl, preallocate, MemoryMapMaxSizeMb, diskType, ldbTimeout)
  103. return e
  104. }
  105. func (s *Store) DeleteCollection(collection string) (e error) {
  106. for _, location := range s.Locations {
  107. e = location.DeleteCollectionFromDiskLocation(collection)
  108. if e != nil {
  109. return
  110. }
  111. stats.DeleteCollectionMetrics(collection)
  112. // let the heartbeat send the list of volumes, instead of sending the deleted volume ids to DeletedVolumesChan
  113. }
  114. return
  115. }
  116. func (s *Store) findVolume(vid needle.VolumeId) *Volume {
  117. for _, location := range s.Locations {
  118. if v, found := location.FindVolume(vid); found {
  119. return v
  120. }
  121. }
  122. return nil
  123. }
  124. func (s *Store) FindFreeLocation(diskType DiskType) (ret *DiskLocation) {
  125. max := int32(0)
  126. for _, location := range s.Locations {
  127. if diskType != location.DiskType {
  128. continue
  129. }
  130. if location.isDiskSpaceLow {
  131. continue
  132. }
  133. currentFreeCount := location.MaxVolumeCount - int32(location.VolumesLen())
  134. currentFreeCount *= erasure_coding.DataShardsCount
  135. currentFreeCount -= int32(location.EcVolumesLen())
  136. currentFreeCount /= erasure_coding.DataShardsCount
  137. if currentFreeCount > max {
  138. max = currentFreeCount
  139. ret = location
  140. }
  141. }
  142. return ret
  143. }
  144. func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32, diskType DiskType, ldbTimeout int64) error {
  145. if s.findVolume(vid) != nil {
  146. return fmt.Errorf("Volume Id %d already exists!", vid)
  147. }
  148. if location := s.FindFreeLocation(diskType); location != nil {
  149. glog.V(0).Infof("In dir %s adds volume:%v collection:%s replicaPlacement:%v ttl:%v",
  150. location.Directory, vid, collection, replicaPlacement, ttl)
  151. if volume, err := NewVolume(location.Directory, location.IdxDirectory, collection, vid, needleMapKind, replicaPlacement, ttl, preallocate, memoryMapMaxSizeMb, ldbTimeout); err == nil {
  152. location.SetVolume(vid, volume)
  153. glog.V(0).Infof("add volume %d", vid)
  154. s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
  155. Id: uint32(vid),
  156. Collection: collection,
  157. ReplicaPlacement: uint32(replicaPlacement.Byte()),
  158. Version: uint32(volume.Version()),
  159. Ttl: ttl.ToUint32(),
  160. DiskType: string(diskType),
  161. }
  162. return nil
  163. } else {
  164. return err
  165. }
  166. }
  167. return fmt.Errorf("No more free space left")
  168. }
  169. func (s *Store) VolumeInfos() (allStats []*VolumeInfo) {
  170. for _, location := range s.Locations {
  171. stats := collectStatsForOneLocation(location)
  172. allStats = append(allStats, stats...)
  173. }
  174. sortVolumeInfos(allStats)
  175. return allStats
  176. }
  177. func collectStatsForOneLocation(location *DiskLocation) (stats []*VolumeInfo) {
  178. location.volumesLock.RLock()
  179. defer location.volumesLock.RUnlock()
  180. for k, v := range location.volumes {
  181. s := collectStatForOneVolume(k, v)
  182. stats = append(stats, s)
  183. }
  184. return stats
  185. }
  186. func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) {
  187. s = &VolumeInfo{
  188. Id: vid,
  189. Collection: v.Collection,
  190. ReplicaPlacement: v.ReplicaPlacement,
  191. Version: v.Version(),
  192. ReadOnly: v.IsReadOnly(),
  193. Ttl: v.Ttl,
  194. CompactRevision: uint32(v.CompactionRevision),
  195. DiskType: v.DiskType().String(),
  196. }
  197. s.RemoteStorageName, s.RemoteStorageKey = v.RemoteStorageNameKey()
  198. v.dataFileAccessLock.RLock()
  199. defer v.dataFileAccessLock.RUnlock()
  200. if v.nm == nil {
  201. return
  202. }
  203. s.FileCount = v.nm.FileCount()
  204. s.DeleteCount = v.nm.DeletedCount()
  205. s.DeletedByteCount = v.nm.DeletedSize()
  206. s.Size = v.nm.ContentSize()
  207. return
  208. }
  209. func (s *Store) SetDataCenter(dataCenter string) {
  210. s.dataCenter = dataCenter
  211. }
  212. func (s *Store) SetRack(rack string) {
  213. s.rack = rack
  214. }
  215. func (s *Store) GetDataCenter() string {
  216. return s.dataCenter
  217. }
  218. func (s *Store) GetRack() string {
  219. return s.rack
  220. }
  221. func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
  222. var volumeMessages []*master_pb.VolumeInformationMessage
  223. maxVolumeCounts := make(map[string]uint32)
  224. var maxFileKey NeedleId
  225. collectionVolumeSize := make(map[string]int64)
  226. collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
  227. for _, location := range s.Locations {
  228. var deleteVids []needle.VolumeId
  229. maxVolumeCounts[string(location.DiskType)] += uint32(location.MaxVolumeCount)
  230. location.volumesLock.RLock()
  231. for _, v := range location.volumes {
  232. curMaxFileKey, volumeMessage := v.ToVolumeInformationMessage()
  233. if volumeMessage == nil {
  234. continue
  235. }
  236. if maxFileKey < curMaxFileKey {
  237. maxFileKey = curMaxFileKey
  238. }
  239. shouldDeleteVolume := false
  240. if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
  241. volumeMessages = append(volumeMessages, volumeMessage)
  242. } else {
  243. if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
  244. deleteVids = append(deleteVids, v.Id)
  245. shouldDeleteVolume = true
  246. } else {
  247. glog.V(0).Infof("volume %d is expired", v.Id)
  248. }
  249. if v.lastIoError != nil {
  250. deleteVids = append(deleteVids, v.Id)
  251. shouldDeleteVolume = true
  252. glog.Warningf("volume %d has IO error: %v", v.Id, v.lastIoError)
  253. }
  254. }
  255. if _, exist := collectionVolumeSize[v.Collection]; !exist {
  256. collectionVolumeSize[v.Collection] = 0
  257. }
  258. if !shouldDeleteVolume {
  259. collectionVolumeSize[v.Collection] += int64(volumeMessage.Size)
  260. } else {
  261. collectionVolumeSize[v.Collection] -= int64(volumeMessage.Size)
  262. if collectionVolumeSize[v.Collection] <= 0 {
  263. delete(collectionVolumeSize, v.Collection)
  264. }
  265. }
  266. if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
  267. collectionVolumeReadOnlyCount[v.Collection] = map[string]uint8{
  268. stats.IsReadOnly: 0,
  269. stats.NoWriteOrDelete: 0,
  270. stats.NoWriteCanDelete: 0,
  271. stats.IsDiskSpaceLow: 0,
  272. }
  273. }
  274. if !shouldDeleteVolume && v.IsReadOnly() {
  275. collectionVolumeReadOnlyCount[v.Collection][stats.IsReadOnly] += 1
  276. if v.noWriteOrDelete {
  277. collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteOrDelete] += 1
  278. }
  279. if v.noWriteCanDelete {
  280. collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteCanDelete] += 1
  281. }
  282. if v.location.isDiskSpaceLow {
  283. collectionVolumeReadOnlyCount[v.Collection][stats.IsDiskSpaceLow] += 1
  284. }
  285. }
  286. }
  287. location.volumesLock.RUnlock()
  288. if len(deleteVids) > 0 {
  289. // delete expired volumes.
  290. location.volumesLock.Lock()
  291. for _, vid := range deleteVids {
  292. found, err := location.deleteVolumeById(vid, false)
  293. if err == nil {
  294. if found {
  295. glog.V(0).Infof("volume %d is deleted", vid)
  296. }
  297. } else {
  298. glog.Warningf("delete volume %d: %v", vid, err)
  299. }
  300. }
  301. location.volumesLock.Unlock()
  302. }
  303. }
  304. var uuidList []string
  305. for _, loc := range s.Locations {
  306. uuidList = append(uuidList, loc.DirectoryUuid)
  307. }
  308. for col, size := range collectionVolumeSize {
  309. stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "normal").Set(float64(size))
  310. }
  311. for col, types := range collectionVolumeReadOnlyCount {
  312. for t, count := range types {
  313. stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues(col, t).Set(float64(count))
  314. }
  315. }
  316. return &master_pb.Heartbeat{
  317. Ip: s.Ip,
  318. Port: uint32(s.Port),
  319. GrpcPort: uint32(s.GrpcPort),
  320. PublicUrl: s.PublicUrl,
  321. MaxVolumeCounts: maxVolumeCounts,
  322. MaxFileKey: NeedleIdToUint64(maxFileKey),
  323. DataCenter: s.dataCenter,
  324. Rack: s.rack,
  325. Volumes: volumeMessages,
  326. HasNoVolumes: len(volumeMessages) == 0,
  327. LocationUuids: uuidList,
  328. }
  329. }
  330. func (s *Store) SetStopping() {
  331. s.isStopping = true
  332. for _, location := range s.Locations {
  333. location.SetStopping()
  334. }
  335. }
  336. func (s *Store) LoadNewVolumes() {
  337. for _, location := range s.Locations {
  338. location.loadExistingVolumes(s.NeedleMapKind, 0)
  339. }
  340. }
  341. func (s *Store) Close() {
  342. for _, location := range s.Locations {
  343. location.Close()
  344. }
  345. }
  346. func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCookie bool, fsync bool) (isUnchanged bool, err error) {
  347. if v := s.findVolume(i); v != nil {
  348. if v.IsReadOnly() {
  349. err = fmt.Errorf("volume %d is read only", i)
  350. return
  351. }
  352. _, _, isUnchanged, err = v.writeNeedle2(n, checkCookie, fsync && s.isStopping)
  353. return
  354. }
  355. glog.V(0).Infoln("volume", i, "not found!")
  356. err = fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port)
  357. return
  358. }
  359. func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) {
  360. if v := s.findVolume(i); v != nil {
  361. if v.noWriteOrDelete {
  362. return 0, fmt.Errorf("volume %d is read only", i)
  363. }
  364. return v.deleteNeedle2(n)
  365. }
  366. return 0, fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port)
  367. }
  368. func (s *Store) ReadVolumeNeedle(i needle.VolumeId, n *needle.Needle, readOption *ReadOption, onReadSizeFn func(size Size)) (int, error) {
  369. if v := s.findVolume(i); v != nil {
  370. return v.readNeedle(n, readOption, onReadSizeFn)
  371. }
  372. return 0, fmt.Errorf("volume %d not found", i)
  373. }
  374. func (s *Store) ReadVolumeNeedleMetaAt(i needle.VolumeId, n *needle.Needle, offset int64, size int32) error {
  375. if v := s.findVolume(i); v != nil {
  376. return v.readNeedleMetaAt(n, offset, size)
  377. }
  378. return fmt.Errorf("volume %d not found", i)
  379. }
  380. func (s *Store) ReadVolumeNeedleDataInto(i needle.VolumeId, n *needle.Needle, readOption *ReadOption, writer io.Writer, offset int64, size int64) error {
  381. if v := s.findVolume(i); v != nil {
  382. return v.readNeedleDataInto(n, readOption, writer, offset, size)
  383. }
  384. return fmt.Errorf("volume %d not found", i)
  385. }
  386. func (s *Store) GetVolume(i needle.VolumeId) *Volume {
  387. return s.findVolume(i)
  388. }
  389. func (s *Store) HasVolume(i needle.VolumeId) bool {
  390. v := s.findVolume(i)
  391. return v != nil
  392. }
  393. func (s *Store) MarkVolumeReadonly(i needle.VolumeId) error {
  394. v := s.findVolume(i)
  395. if v == nil {
  396. return fmt.Errorf("volume %d not found", i)
  397. }
  398. v.noWriteLock.Lock()
  399. v.noWriteOrDelete = true
  400. v.noWriteLock.Unlock()
  401. return nil
  402. }
  403. func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
  404. v := s.findVolume(i)
  405. if v == nil {
  406. return fmt.Errorf("volume %d not found", i)
  407. }
  408. v.noWriteLock.Lock()
  409. v.noWriteOrDelete = false
  410. v.noWriteLock.Unlock()
  411. return nil
  412. }
  413. func (s *Store) MountVolume(i needle.VolumeId) error {
  414. for _, location := range s.Locations {
  415. if found := location.LoadVolume(i, s.NeedleMapKind); found == true {
  416. glog.V(0).Infof("mount volume %d", i)
  417. v := s.findVolume(i)
  418. s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
  419. Id: uint32(v.Id),
  420. Collection: v.Collection,
  421. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  422. Version: uint32(v.Version()),
  423. Ttl: v.Ttl.ToUint32(),
  424. DiskType: string(v.location.DiskType),
  425. }
  426. return nil
  427. }
  428. }
  429. return fmt.Errorf("volume %d not found on disk", i)
  430. }
  431. func (s *Store) UnmountVolume(i needle.VolumeId) error {
  432. v := s.findVolume(i)
  433. if v == nil {
  434. return nil
  435. }
  436. message := master_pb.VolumeShortInformationMessage{
  437. Id: uint32(v.Id),
  438. Collection: v.Collection,
  439. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  440. Version: uint32(v.Version()),
  441. Ttl: v.Ttl.ToUint32(),
  442. DiskType: string(v.location.DiskType),
  443. }
  444. for _, location := range s.Locations {
  445. err := location.UnloadVolume(i)
  446. if err == nil {
  447. glog.V(0).Infof("UnmountVolume %d", i)
  448. stats.DeleteCollectionMetrics(v.Collection)
  449. s.DeletedVolumesChan <- message
  450. return nil
  451. } else if err == ErrVolumeNotFound {
  452. continue
  453. }
  454. }
  455. return fmt.Errorf("volume %d not found on disk", i)
  456. }
  457. func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool) error {
  458. v := s.findVolume(i)
  459. if v == nil {
  460. return fmt.Errorf("delete volume %d not found on disk", i)
  461. }
  462. message := master_pb.VolumeShortInformationMessage{
  463. Id: uint32(v.Id),
  464. Collection: v.Collection,
  465. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  466. Version: uint32(v.Version()),
  467. Ttl: v.Ttl.ToUint32(),
  468. DiskType: string(v.location.DiskType),
  469. }
  470. for _, location := range s.Locations {
  471. err := location.DeleteVolume(i, onlyEmpty)
  472. if err == nil {
  473. glog.V(0).Infof("DeleteVolume %d", i)
  474. s.DeletedVolumesChan <- message
  475. return nil
  476. } else if err == ErrVolumeNotFound {
  477. continue
  478. } else if err == ErrVolumeNotEmpty {
  479. return fmt.Errorf("DeleteVolume %d: %v", i, err)
  480. } else {
  481. glog.Errorf("DeleteVolume %d: %v", i, err)
  482. }
  483. }
  484. return fmt.Errorf("volume %d not found on disk", i)
  485. }
  486. func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
  487. for _, location := range s.Locations {
  488. fileInfo, found := location.LocateVolume(i)
  489. if !found {
  490. continue
  491. }
  492. // load, modify, save
  493. baseFileName := strings.TrimSuffix(fileInfo.Name(), filepath.Ext(fileInfo.Name()))
  494. vifFile := filepath.Join(location.Directory, baseFileName+".vif")
  495. volumeInfo, _, _, err := volume_info.MaybeLoadVolumeInfo(vifFile)
  496. if err != nil {
  497. return fmt.Errorf("volume %d failed to load vif: %v", i, err)
  498. }
  499. volumeInfo.Replication = replication
  500. err = volume_info.SaveVolumeInfo(vifFile, volumeInfo)
  501. if err != nil {
  502. return fmt.Errorf("volume %d failed to save vif: %v", i, err)
  503. }
  504. return nil
  505. }
  506. return fmt.Errorf("volume %d not found on disk", i)
  507. }
  508. func (s *Store) SetVolumeSizeLimit(x uint64) {
  509. atomic.StoreUint64(&s.volumeSizeLimit, x)
  510. }
  511. func (s *Store) GetVolumeSizeLimit() uint64 {
  512. return atomic.LoadUint64(&s.volumeSizeLimit)
  513. }
  514. func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
  515. volumeSizeLimit := s.GetVolumeSizeLimit()
  516. if volumeSizeLimit == 0 {
  517. return
  518. }
  519. var newMaxVolumeCount int32
  520. for _, diskLocation := range s.Locations {
  521. if diskLocation.OriginalMaxVolumeCount == 0 {
  522. currentMaxVolumeCount := atomic.LoadInt32(&diskLocation.MaxVolumeCount)
  523. diskStatus := stats.NewDiskStatus(diskLocation.Directory)
  524. unusedSpace := diskLocation.UnUsedSpace(volumeSizeLimit)
  525. unclaimedSpaces := int64(diskStatus.Free) - int64(unusedSpace)
  526. volCount := diskLocation.VolumesLen()
  527. maxVolumeCount := int32(volCount)
  528. if unclaimedSpaces > int64(volumeSizeLimit) {
  529. maxVolumeCount += int32(uint64(unclaimedSpaces)/volumeSizeLimit) - 1
  530. }
  531. newMaxVolumeCount = newMaxVolumeCount + maxVolumeCount
  532. atomic.StoreInt32(&diskLocation.MaxVolumeCount, maxVolumeCount)
  533. glog.V(4).Infof("disk %s max %d unclaimedSpace:%dMB, unused:%dMB volumeSizeLimit:%dMB",
  534. diskLocation.Directory, maxVolumeCount, unclaimedSpaces/1024/1024, unusedSpace/1024/1024, volumeSizeLimit/1024/1024)
  535. hasChanges = hasChanges || currentMaxVolumeCount != atomic.LoadInt32(&diskLocation.MaxVolumeCount)
  536. } else {
  537. newMaxVolumeCount = newMaxVolumeCount + diskLocation.OriginalMaxVolumeCount
  538. }
  539. }
  540. stats.VolumeServerMaxVolumeCounter.Set(float64(newMaxVolumeCount))
  541. return
  542. }