diff options
| -rw-r--r-- | weed/storage/erasure_coding/ec_volume.go | 1 | ||||
| -rw-r--r-- | weed/storage/store_ec.go | 28 |
2 files changed, 27 insertions, 2 deletions
diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index 28dbd8c7f..8bd7237f5 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -24,6 +24,7 @@ type EcVolume struct { ShardLocations map[ShardId][]string ShardLocationsRefreshTime time.Time ShardLocationsLock sync.RWMutex + Version needle.Version } func NewEcVolume(dir string, collection string, vid needle.VolumeId) (ev *EcVolume, err error) { diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 92a63da8f..e293ebe5a 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -98,8 +98,13 @@ func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *n for _, location := range s.Locations { if localEcVolume, found := location.FindEcVolume(vid); found { - // TODO need to read the version - version := needle.CurrentVersion + // read the volume version + for localEcVolume.Version == 0 { + err := s.readEcVolumeVersion(ctx, vid, localEcVolume) + time.Sleep(1357 * time.Millisecond) + glog.V(0).Infof("ReadEcShardNeedle vid %d version:%v: %v", vid, localEcVolume.Version, err) + } + version := localEcVolume.Version offset, size, intervals, err := localEcVolume.LocateEcShardNeedle(n, version) if err != nil { @@ -127,6 +132,22 @@ func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *n return 0, fmt.Errorf("ec shard %d not found", vid) } +func (s *Store) readEcVolumeVersion(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume) (err error) { + + interval := erasure_coding.Interval{ + BlockIndex: 0, + InnerBlockOffset: 0, + Size: _SuperBlockSize, + IsLargeBlock: true, // it could be large block, but ok in this place + LargeBlockRowsCount: 0, + } + data, err := s.readEcShardIntervals(ctx, vid, ecVolume, []erasure_coding.Interval{interval}) + if err == nil { + ecVolume.Version = needle.Version(data[0]) + } + return +} + func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, err error) { if err = s.cachedLookupEcShardLocations(ctx, ecVolume); err != nil { @@ -204,6 +225,9 @@ func (s *Store) cachedLookupEcShardLocations(ctx context.Context, ecVolume *eras if err != nil { return fmt.Errorf("lookup ec volume %d: %v", ecVolume.VolumeId, err) } + if len(resp.ShardIdLocations) < erasure_coding.DataShardsCount { + return fmt.Errorf("only %d shards found but %d required", len(resp.ShardIdLocations), erasure_coding.DataShardsCount) + } ecVolume.ShardLocationsLock.Lock() for _, shardIdLocations := range resp.ShardIdLocations { |