gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

ask for the ec volume version

Browse source
This commit is contained in:
Chris Lu 2019-05-31 00:58:51 -07:00
parent 689930f092
commit 47f1901843
2 changed files with 27 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
weed/storage/erasure_coding/ec_volume.go
View file

@ -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) {

28
weed/storage/store_ec.go
View file

@ -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 {