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.

230 lines
7.4 KiB

6 years ago
6 years ago
4 years ago
6 years ago
3 years ago
3 years ago
6 years ago
2 years ago
6 years ago
  1. package topology
  2. import (
  3. "context"
  4. "github.com/seaweedfs/seaweedfs/weed/pb"
  5. "io"
  6. "sync/atomic"
  7. "time"
  8. "google.golang.org/grpc"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  10. "github.com/seaweedfs/seaweedfs/weed/glog"
  11. "github.com/seaweedfs/seaweedfs/weed/operation"
  12. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  13. )
  14. func (t *Topology) batchVacuumVolumeCheck(grpcDialOption grpc.DialOption, vid needle.VolumeId,
  15. locationlist *VolumeLocationList, garbageThreshold float64) (*VolumeLocationList, bool) {
  16. ch := make(chan int, locationlist.Length())
  17. errCount := int32(0)
  18. for index, dn := range locationlist.list {
  19. go func(index int, url pb.ServerAddress, vid needle.VolumeId) {
  20. err := operation.WithVolumeServerClient(false, url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  21. resp, err := volumeServerClient.VacuumVolumeCheck(context.Background(), &volume_server_pb.VacuumVolumeCheckRequest{
  22. VolumeId: uint32(vid),
  23. })
  24. if err != nil {
  25. atomic.AddInt32(&errCount, 1)
  26. ch <- -1
  27. return err
  28. }
  29. if resp.GarbageRatio >= garbageThreshold {
  30. ch <- index
  31. } else {
  32. ch <- -1
  33. }
  34. return nil
  35. })
  36. if err != nil {
  37. glog.V(0).Infof("Checking vacuuming %d on %s: %v", vid, url, err)
  38. }
  39. }(index, dn.ServerAddress(), vid)
  40. }
  41. vacuumLocationList := NewVolumeLocationList()
  42. waitTimeout := time.NewTimer(time.Minute * time.Duration(t.volumeSizeLimit/1024/1024/1000+1))
  43. defer waitTimeout.Stop()
  44. for range locationlist.list {
  45. select {
  46. case index := <-ch:
  47. if index != -1 {
  48. vacuumLocationList.list = append(vacuumLocationList.list, locationlist.list[index])
  49. }
  50. case <-waitTimeout.C:
  51. return vacuumLocationList, false
  52. }
  53. }
  54. return vacuumLocationList, errCount == 0 && len(vacuumLocationList.list) > 0
  55. }
  56. func (t *Topology) batchVacuumVolumeCompact(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid needle.VolumeId,
  57. locationlist *VolumeLocationList, preallocate int64) bool {
  58. vl.accessLock.Lock()
  59. vl.removeFromWritable(vid)
  60. vl.accessLock.Unlock()
  61. ch := make(chan bool, locationlist.Length())
  62. for index, dn := range locationlist.list {
  63. go func(index int, url pb.ServerAddress, vid needle.VolumeId) {
  64. glog.V(0).Infoln(index, "Start vacuuming", vid, "on", url)
  65. err := operation.WithVolumeServerClient(true, url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  66. stream, err := volumeServerClient.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
  67. VolumeId: uint32(vid),
  68. Preallocate: preallocate,
  69. })
  70. if err != nil {
  71. return err
  72. }
  73. for {
  74. resp, recvErr := stream.Recv()
  75. if recvErr != nil {
  76. if recvErr == io.EOF {
  77. break
  78. } else {
  79. return recvErr
  80. }
  81. }
  82. glog.V(0).Infof("%d vacuum %d on %s processed %d bytes, loadAvg %.02f%%",
  83. index, vid, url, resp.ProcessedBytes, resp.LoadAvg_1M*100)
  84. }
  85. return nil
  86. })
  87. if err != nil {
  88. glog.Errorf("Error when vacuuming %d on %s: %v", vid, url, err)
  89. ch <- false
  90. } else {
  91. glog.V(0).Infof("Complete vacuuming %d on %s", vid, url)
  92. ch <- true
  93. }
  94. }(index, dn.ServerAddress(), vid)
  95. }
  96. isVacuumSuccess := true
  97. waitTimeout := time.NewTimer(3 * time.Minute * time.Duration(t.volumeSizeLimit/1024/1024/1000+1))
  98. defer waitTimeout.Stop()
  99. for range locationlist.list {
  100. select {
  101. case canCommit := <-ch:
  102. isVacuumSuccess = isVacuumSuccess && canCommit
  103. case <-waitTimeout.C:
  104. return false
  105. }
  106. }
  107. return isVacuumSuccess
  108. }
  109. func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid needle.VolumeId, locationlist *VolumeLocationList) bool {
  110. isCommitSuccess := true
  111. isReadOnly := false
  112. for _, dn := range locationlist.list {
  113. glog.V(0).Infoln("Start Committing vacuum", vid, "on", dn.Url())
  114. err := operation.WithVolumeServerClient(false, dn.ServerAddress(), grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  115. resp, err := volumeServerClient.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{
  116. VolumeId: uint32(vid),
  117. })
  118. if resp != nil && resp.IsReadOnly {
  119. isReadOnly = true
  120. }
  121. return err
  122. })
  123. if err != nil {
  124. glog.Errorf("Error when committing vacuum %d on %s: %v", vid, dn.Url(), err)
  125. isCommitSuccess = false
  126. } else {
  127. glog.V(0).Infof("Complete Committing vacuum %d on %s", vid, dn.Url())
  128. }
  129. }
  130. if isCommitSuccess {
  131. for _, dn := range locationlist.list {
  132. vl.SetVolumeAvailable(dn, vid, isReadOnly)
  133. }
  134. }
  135. return isCommitSuccess
  136. }
  137. func (t *Topology) batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid needle.VolumeId, locationlist *VolumeLocationList) {
  138. for _, dn := range locationlist.list {
  139. glog.V(0).Infoln("Start cleaning up", vid, "on", dn.Url())
  140. err := operation.WithVolumeServerClient(false, dn.ServerAddress(), grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  141. _, err := volumeServerClient.VacuumVolumeCleanup(context.Background(), &volume_server_pb.VacuumVolumeCleanupRequest{
  142. VolumeId: uint32(vid),
  143. })
  144. return err
  145. })
  146. if err != nil {
  147. glog.Errorf("Error when cleaning up vacuum %d on %s: %v", vid, dn.Url(), err)
  148. } else {
  149. glog.V(0).Infof("Complete cleaning up vacuum %d on %s", vid, dn.Url())
  150. }
  151. }
  152. }
  153. func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, volumeId uint32, collection string, preallocate int64) {
  154. // if there is vacuum going on, return immediately
  155. swapped := atomic.CompareAndSwapInt64(&t.vacuumLockCounter, 0, 1)
  156. if !swapped {
  157. return
  158. }
  159. defer atomic.StoreInt64(&t.vacuumLockCounter, 0)
  160. // now only one vacuum process going on
  161. glog.V(1).Infof("Start vacuum on demand with threshold: %f collection: %s volumeId: %d",
  162. garbageThreshold, collection, volumeId)
  163. for _, col := range t.collectionMap.Items() {
  164. c := col.(*Collection)
  165. if collection != "" && collection != c.Name {
  166. continue
  167. }
  168. for _, vl := range c.storageType2VolumeLayout.Items() {
  169. if vl != nil {
  170. volumeLayout := vl.(*VolumeLayout)
  171. if volumeId > 0 {
  172. if volumeLayout.Lookup(needle.VolumeId(volumeId)) != nil {
  173. t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
  174. }
  175. } else {
  176. t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
  177. }
  178. }
  179. }
  180. }
  181. }
  182. func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, preallocate int64) {
  183. volumeLayout.accessLock.RLock()
  184. tmpMap := make(map[needle.VolumeId]*VolumeLocationList)
  185. for vid, locationList := range volumeLayout.vid2location {
  186. tmpMap[vid] = locationList.Copy()
  187. }
  188. volumeLayout.accessLock.RUnlock()
  189. for vid, locationList := range tmpMap {
  190. volumeLayout.accessLock.RLock()
  191. isReadOnly := volumeLayout.readonlyVolumes.IsTrue(vid)
  192. volumeLayout.accessLock.RUnlock()
  193. if isReadOnly {
  194. continue
  195. }
  196. glog.V(2).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
  197. if vacuumLocationList, needVacuum := t.batchVacuumVolumeCheck(
  198. grpcDialOption, vid, locationList, garbageThreshold); needVacuum {
  199. if t.batchVacuumVolumeCompact(grpcDialOption, volumeLayout, vid, vacuumLocationList, preallocate) {
  200. t.batchVacuumVolumeCommit(grpcDialOption, volumeLayout, vid, vacuumLocationList)
  201. } else {
  202. t.batchVacuumVolumeCleanup(grpcDialOption, volumeLayout, vid, vacuumLocationList)
  203. }
  204. }
  205. }
  206. }