ec_shard.go 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. package erasure_coding
  2. import (
  3. "fmt"
  4. "os"
  5. "path"
  6. "strconv"
  7. "strings"
  8. "github.com/seaweedfs/seaweedfs/weed/stats"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  10. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  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. _ = v.ecdFile.Close()
  35. return nil, fmt.Errorf("can not stat ec volume shard %s%s: %v", baseFileName, ToExt(int(shardId)), statErr)
  36. }
  37. v.ecdFileSize = ecdFi.Size()
  38. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "ec_shards").Inc()
  39. return
  40. }
  41. func (shard *EcVolumeShard) Size() int64 {
  42. return shard.ecdFileSize
  43. }
  44. func (shard *EcVolumeShard) String() string {
  45. return fmt.Sprintf("ec shard %v:%v, dir:%s, Collection:%s", shard.VolumeId, shard.ShardId, shard.dir, shard.Collection)
  46. }
  47. func (shard *EcVolumeShard) FileName() (fileName string) {
  48. return EcShardFileName(shard.Collection, shard.dir, int(shard.VolumeId))
  49. }
  50. func EcShardFileName(collection string, dir string, id int) (fileName string) {
  51. idString := strconv.Itoa(id)
  52. if collection == "" {
  53. fileName = path.Join(dir, idString)
  54. } else {
  55. fileName = path.Join(dir, collection+"_"+idString)
  56. }
  57. return
  58. }
  59. func EcShardBaseFileName(collection string, id int) (baseFileName string) {
  60. baseFileName = strconv.Itoa(id)
  61. if collection != "" {
  62. baseFileName = collection + "_" + baseFileName
  63. }
  64. return
  65. }
  66. func (shard *EcVolumeShard) Close() {
  67. if shard.ecdFile != nil {
  68. _ = shard.ecdFile.Close()
  69. shard.ecdFile = nil
  70. }
  71. }
  72. func (shard *EcVolumeShard) Destroy() {
  73. os.Remove(shard.FileName() + ToExt(int(shard.ShardId)))
  74. stats.VolumeServerVolumeCounter.WithLabelValues(shard.Collection, "ec_shards").Dec()
  75. }
  76. func (shard *EcVolumeShard) ReadAt(buf []byte, offset int64) (int, error) {
  77. return shard.ecdFile.ReadAt(buf, offset)
  78. }