ec_shard.go 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
  1. package erasure_coding
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  5. "os"
  6. "path"
  7. "strconv"
  8. "strings"
  9. "github.com/seaweedfs/seaweedfs/weed/stats"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  11. )
  12. type ShardId uint8
  13. type EcVolumeShard struct {
  14. VolumeId needle.VolumeId
  15. ShardId ShardId
  16. Collection string
  17. dir string
  18. ecdFile *os.File
  19. ecdFileSize int64
  20. DiskType types.DiskType
  21. }
  22. func NewEcVolumeShard(diskType types.DiskType, dirname string, collection string, id needle.VolumeId, shardId ShardId) (v *EcVolumeShard, e error) {
  23. v = &EcVolumeShard{dir: dirname, Collection: collection, VolumeId: id, ShardId: shardId, DiskType: diskType}
  24. baseFileName := v.FileName()
  25. // open ecd file
  26. if v.ecdFile, e = os.OpenFile(baseFileName+ToExt(int(shardId)), os.O_RDONLY, 0644); e != nil {
  27. if e == os.ErrNotExist || strings.Contains(e.Error(), "no such file or directory") {
  28. return nil, os.ErrNotExist
  29. }
  30. return nil, fmt.Errorf("cannot read ec volume shard %s%s: %v", baseFileName, ToExt(int(shardId)), e)
  31. }
  32. ecdFi, statErr := v.ecdFile.Stat()
  33. if statErr != nil {
  34. return nil, fmt.Errorf("can not stat ec volume shard %s%s: %v", baseFileName, ToExt(int(shardId)), statErr)
  35. }
  36. v.ecdFileSize = ecdFi.Size()
  37. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "ec_shards").Inc()
  38. return
  39. }
  40. func (shard *EcVolumeShard) Size() int64 {
  41. return shard.ecdFileSize
  42. }
  43. func (shard *EcVolumeShard) String() string {
  44. return fmt.Sprintf("ec shard %v:%v, dir:%s, Collection:%s", shard.VolumeId, shard.ShardId, shard.dir, shard.Collection)
  45. }
  46. func (shard *EcVolumeShard) FileName() (fileName string) {
  47. return EcShardFileName(shard.Collection, shard.dir, int(shard.VolumeId))
  48. }
  49. func EcShardFileName(collection string, dir string, id int) (fileName string) {
  50. idString := strconv.Itoa(id)
  51. if collection == "" {
  52. fileName = path.Join(dir, idString)
  53. } else {
  54. fileName = path.Join(dir, collection+"_"+idString)
  55. }
  56. return
  57. }
  58. func EcShardBaseFileName(collection string, id int) (baseFileName string) {
  59. baseFileName = strconv.Itoa(id)
  60. if collection != "" {
  61. baseFileName = collection + "_" + baseFileName
  62. }
  63. return
  64. }
  65. func (shard *EcVolumeShard) Close() {
  66. if shard.ecdFile != nil {
  67. _ = shard.ecdFile.Close()
  68. shard.ecdFile = nil
  69. }
  70. }
  71. func (shard *EcVolumeShard) Destroy() {
  72. os.Remove(shard.FileName() + ToExt(int(shard.ShardId)))
  73. stats.VolumeServerVolumeCounter.WithLabelValues(shard.Collection, "ec_shards").Dec()
  74. }
  75. func (shard *EcVolumeShard) ReadAt(buf []byte, offset int64) (int, error) {
  76. return shard.ecdFile.ReadAt(buf, offset)
  77. }