Browse Source

ensure only compatible volume versions are writable

pull/2/head
Chris Lu 12 years ago
parent
commit
6c8810e4d2
  1. 1
      weed-fs/src/pkg/storage/volume_info.go
  2. 4
      weed-fs/src/pkg/topology/topology_event_handling.go
  3. 6
      weed-fs/src/pkg/topology/volume_layout.go

1
weed-fs/src/pkg/storage/volume_info.go

@ -6,6 +6,7 @@ type VolumeInfo struct {
Id VolumeId
Size uint64
RepType ReplicationType
Version Version
FileCount int
DeleteCount int
DeletedByteCount uint64

4
weed-fs/src/pkg/topology/topology_event_handling.go

@ -59,8 +59,8 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) {
}
func (t *Topology) RegisterRecoveredDataNode(dn *DataNode) {
for _, v := range dn.volumes {
if uint64(v.Size) < t.volumeSizeLimit {
vl := t.GetVolumeLayout(v.RepType)
vl := t.GetVolumeLayout(v.RepType)
if vl.isWritable(&v) {
vl.SetVolumeAvailable(dn, v.Id)
}
}

6
weed-fs/src/pkg/topology/volume_layout.go

@ -31,13 +31,17 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
}
if vl.vid2location[v.Id].Add(dn) {
if len(vl.vid2location[v.Id].list) == v.RepType.GetCopyCount() {
if uint64(v.Size) < vl.volumeSizeLimit {
if vl.isWritable(v) {
vl.writables = append(vl.writables, v.Id)
}
}
}
}
func (vl *VolumeLayout) isWritable(v *storage.VolumeInfo) bool{
return uint64(v.Size) < vl.volumeSizeLimit && v.Version == storage.CurrentVersion
}
func (vl *VolumeLayout) Lookup(vid storage.VolumeId) *[]*DataNode {
return &vl.vid2location[vid].list
}

Loading…
Cancel
Save