1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
package topology
import (
"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
)
func (dn *DataNode) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) {
dn.RLock()
for _, ecVolumeInfo := range dn.ecShards {
ret = append(ret, ecVolumeInfo)
}
dn.RUnlock()
return ret
}
func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) (newShards, deletedShards []*erasure_coding.EcVolumeInfo) {
// prepare the new ec shard map
actualEcShardMap := make(map[needle.VolumeId]*erasure_coding.EcVolumeInfo)
for _, ecShards := range actualShards {
actualEcShardMap[ecShards.VolumeId] = ecShards
}
// found out the newShards and deletedShards
dn.ecShardsLock.RLock()
for vid, ecShards := range dn.ecShards {
if actualEcShards, ok := actualEcShardMap[vid]; !ok {
// dn registered ec shards not found in the new set of ec shards
deletedShards = append(deletedShards, ecShards)
} else {
// found, but maybe the actual shard could be missing
a := actualEcShards.Minus(ecShards)
if a.ShardIdCount() > 0 {
newShards = append(newShards, a)
}
d := ecShards.Minus(actualEcShards)
if d.ShardIdCount() > 0 {
deletedShards = append(deletedShards, d)
}
}
}
for _, ecShards := range actualShards {
if _, found := dn.ecShards[ecShards.VolumeId]; !found {
newShards = append(newShards, ecShards)
}
}
dn.ecShardsLock.RUnlock()
if len(newShards) > 0 || len(deletedShards) > 0 {
// if changed, set to the new ec shard map
dn.ecShardsLock.Lock()
dn.ecShards = actualEcShardMap
dn.ecShardsLock.Unlock()
}
return
}
func (dn *DataNode) DeltaUpdateEcShards(newShards, deletedShards []*erasure_coding.EcVolumeInfo) {
for _, newShard := range newShards {
dn.AddOrUpdateEcShard(newShard)
}
for _, deletedShard := range deletedShards {
dn.DeleteEcShard(deletedShard)
}
}
func (dn *DataNode) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) {
dn.ecShardsLock.Lock()
defer dn.ecShardsLock.Unlock()
if existing, ok := dn.ecShards[s.VolumeId]; !ok {
dn.ecShards[s.VolumeId] = s
} else {
existing.ShardBits = existing.ShardBits.Plus(s.ShardBits)
}
}
func (dn *DataNode) DeleteEcShard(s *erasure_coding.EcVolumeInfo) {
dn.ecShardsLock.Lock()
defer dn.ecShardsLock.Unlock()
if existing, ok := dn.ecShards[s.VolumeId]; ok {
existing.ShardBits = existing.ShardBits.Minus(s.ShardBits)
if existing.ShardBits.ShardIdCount() == 0 {
delete(dn.ecShards, s.VolumeId)
}
}
}
func (dn *DataNode) HasVolumesById(id needle.VolumeId) (hasVolumeId bool) {
// check whether normal volumes has this volume id
dn.RLock()
_, ok := dn.volumes[id]
if ok {
hasVolumeId = true
}
dn.RUnlock()
if hasVolumeId {
return
}
// check whether ec shards has this volume id
dn.ecShardsLock.RLock()
_, ok = dn.ecShards[id]
if ok {
hasVolumeId = true
}
dn.ecShardsLock.RUnlock()
return
}
|
