mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-17 04:36:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

batch drain delta heartbeat messages (#9914)

Browse Source
This commit is contained in:
Bruce Zou
2026-06-10 13:33:45 -07:00
committed by GitHub
parent 6b4d20a6f3
commit 1dd292fb84
7 changed files with 79 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/server/volume_grpc_client_to_master.go
+38 -34
View File
@@ -243,64 +243,68 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
glog.V(0).Infof("Volume Server Failed to update state to master %s: %v", masterAddress, err)
return "", err
}
case volumeMessage := <-vs.store.NewVolumesChan:
case first := <-vs.store.NewVolumesChan:
volumes := util.DrainChannel(vs.store.NewVolumesChan, first)
deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
NewVolumes: []*master_pb.VolumeShortInformationMessage{
&volumeMessage, // volumeMessage is already a copy from the channel receive
},
NewVolumes: volumes,
}
for _, v := range volumes {
glog.V(0).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, v.Id)
}
glog.V(0).Infof("volume server %s:%d adds volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
return "", err
}
case ecShardMessage := <-vs.store.NewEcShardsChan:
case first := <-vs.store.NewEcShardsChan:
shards := util.DrainChannel(vs.store.NewEcShardsChan, first)
deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
&ecShardMessage, // ecShardMessage is already a copy from the channel receive
},
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
NewEcShards: shards,
}
for _, s := range shards {
si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(s)
glog.V(0).Infof("volume server %s:%d adds ec shards to %d [%s]", vs.store.Ip, vs.store.Port, s.Id, si.String())
}
si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(&ecShardMessage)
glog.V(0).Infof("volume server %s:%d adds ec shards to %d [%s]", vs.store.Ip, vs.store.Port, ecShardMessage.Id, si.String())
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
return "", err
}
case volumeMessage := <-vs.store.DeletedVolumesChan:
case first := <-vs.store.DeletedVolumesChan:
volumes := util.DrainChannel(vs.store.DeletedVolumesChan, first)
deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedVolumes: []*master_pb.VolumeShortInformationMessage{
&volumeMessage, // volumeMessage is already a copy from the channel receive
},
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedVolumes: volumes,
}
for _, v := range volumes {
glog.V(0).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, v.Id)
}
glog.V(0).Infof("volume server %s:%d deletes volume %d", vs.store.Ip, vs.store.Port, volumeMessage.Id)
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
return "", err
}
case ecShardMessage := <-vs.store.DeletedEcShardsChan:
case first := <-vs.store.DeletedEcShardsChan:
shards := util.DrainChannel(vs.store.DeletedEcShardsChan, first)
deltaBeat := &master_pb.Heartbeat{
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{
&ecShardMessage, // ecShardMessage is already a copy from the channel receive
},
Ip: ip,
Port: port,
DataCenter: dataCenter,
Rack: rack,
DeletedEcShards: shards,
}
for _, s := range shards {
glog.V(0).Infof("volume server %s:%d deletes ec shard %d:%s", vs.store.Ip, vs.store.Port, s.Id,
erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(s).String())
}
si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(&ecShardMessage)
glog.V(0).Infof("volume server %s:%d deletes ec shards from %d disk_id:%d [%s]", vs.store.Ip, vs.store.Port, ecShardMessage.Id, ecShardMessage.DiskId, si.String())
if err = stream.Send(deltaBeat); err != nil {
glog.V(0).Infof("Volume Server Failed to update to master %s: %v", masterAddress, err)
return "", err
weed/storage/store.go
+13 -13
View File
@@ -73,10 +73,10 @@ type Store struct {
NeedleMapKind NeedleMapKind
State *State
StateUpdateChan chan *volume_server_pb.VolumeServerState
NewVolumesChan chan master_pb.VolumeShortInformationMessage
DeletedVolumesChan chan master_pb.VolumeShortInformationMessage
NewEcShardsChan chan master_pb.VolumeEcShardInformationMessage
DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage
NewVolumesChan chan *master_pb.VolumeShortInformationMessage
DeletedVolumesChan chan *master_pb.VolumeShortInformationMessage
NewEcShardsChan chan *master_pb.VolumeEcShardInformationMessage
DeletedEcShardsChan chan *master_pb.VolumeEcShardInformationMessage
isStopping bool
}
@@ -107,10 +107,10 @@ func NewStore(
Locations: make([]*DiskLocation, 0),
StateUpdateChan: make(chan *volume_server_pb.VolumeServerState, HEARTBEAT_CHAN_SIZE),
NewVolumesChan: make(chan master_pb.VolumeShortInformationMessage, HEARTBEAT_CHAN_SIZE),
DeletedVolumesChan: make(chan master_pb.VolumeShortInformationMessage, HEARTBEAT_CHAN_SIZE),
NewEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, HEARTBEAT_CHAN_SIZE),
DeletedEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, HEARTBEAT_CHAN_SIZE),
NewVolumesChan: make(chan *master_pb.VolumeShortInformationMessage, HEARTBEAT_CHAN_SIZE),
DeletedVolumesChan: make(chan *master_pb.VolumeShortInformationMessage, HEARTBEAT_CHAN_SIZE),
NewEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, HEARTBEAT_CHAN_SIZE),
DeletedEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, HEARTBEAT_CHAN_SIZE),
}
var wg sync.WaitGroup
@@ -132,7 +132,7 @@ func NewStore(
// Use non-blocking send during startup to avoid deadlock
// The channel reader only starts after connecting to master, but we're loading during startup
select {
case s.NewEcShardsChan <- master_pb.VolumeEcShardInformationMessage{
case s.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{
Id: uint32(vid),
Collection: collection,
EcIndexBits: si.Bitmap(),
@@ -303,7 +303,7 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind
volume.diskId = diskId // Set the disk ID
location.SetVolume(vid, volume)
glog.V(0).Infof("add volume %d on disk ID %d", vid, diskId)
s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
Id: uint32(vid),
Collection: collection,
ReplicaPlacement: uint32(replicaPlacement.Byte()),
@@ -733,7 +733,7 @@ func (s *Store) MountVolume(i needle.VolumeId) error {
glog.V(0).Infof("mount volume %d", i)
v := s.findVolume(i)
v.diskId = uint32(diskId) // Set disk ID when mounting
s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
Id: uint32(v.Id),
Collection: v.Collection,
ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
@@ -768,7 +768,7 @@ func (s *Store) UnmountVolume(i needle.VolumeId) error {
err := location.UnloadVolume(i)
if err == nil {
glog.V(0).Infof("UnmountVolume %d", i)
s.DeletedVolumesChan <- message
s.DeletedVolumesChan <- &message
return nil
} else if err == ErrVolumeNotFound {
continue
@@ -796,7 +796,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool, keepRemoteData b
err := location.DeleteVolume(i, onlyEmpty, keepRemoteData)
if err == nil {
glog.V(0).Infof("DeleteVolume %d", i)
s.DeletedVolumesChan <- message
s.DeletedVolumesChan <- &message
return nil
} else if err == ErrVolumeNotFound {
continue
weed/storage/store_ec.go
+2 -2
View File
@@ -210,7 +210,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er
si := erasure_coding.NewShardsInfo()
si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize())))
s.NewEcShardsChan <- master_pb.VolumeEcShardInformationMessage{
s.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{
Id: uint32(vid),
Collection: collection,
EcIndexBits: uint32(si.Bitmap()),
@@ -272,7 +272,7 @@ func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.Shar
if deleted := location.UnloadEcShard(vid, shardId); deleted {
glog.V(0).Infof("UnmountEcShards %d.%d disk_id:%d", vid, shardId, diskId)
s.DeletedEcShardsChan <- message
s.DeletedEcShardsChan <- &message
return nil
}
weed/storage/store_ec_disk_type_test.go
+4 -4
View File
@@ -26,7 +26,7 @@ import (
// and would otherwise pollute the captured slice). The returned slice
// is appended to only by the goroutine; tests should not race against
// it directly — use waitForShardMsg instead.
func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[]master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId, collection string, plant func()) {
func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[]*master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId, collection string, plant func()) {
t.Helper()
tempDir := t.TempDir()
hddDir := filepath.Join(tempDir, "hdd")
@@ -56,7 +56,7 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[
diskIOProbeConfig,
)
captured := []master_pb.VolumeEcShardInformationMessage{}
captured := []*master_pb.VolumeEcShardInformationMessage{}
drainedShardMsgs = &captured
done := make(chan struct{})
go func() {
@@ -116,7 +116,7 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[
// waitForShardMsg returns the first captured mount message for vid, polling
// briefly because the goroutine reads asynchronously from MountEcShards.
func waitForShardMsg(t *testing.T, captured *[]master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId) master_pb.VolumeEcShardInformationMessage {
func waitForShardMsg(t *testing.T, captured *[]*master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId) *master_pb.VolumeEcShardInformationMessage {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
@@ -128,7 +128,7 @@ func waitForShardMsg(t *testing.T, captured *[]master_pb.VolumeEcShardInformatio
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("no NewEcShardsChan message captured for volume %d within 2s", vid)
return master_pb.VolumeEcShardInformationMessage{}
return nil
}
// findHeartbeatShard returns the EC-shard heartbeat entry for vid.
weed/storage/store_ec_hybrid_repro_test.go
+5 -5
View File
@@ -112,11 +112,11 @@ func TestIssue9478_PartialEcOnSiblingDiskOfHealthyDat(t *testing.T) {
// consistent state.
store := &Store{
Locations: []*DiskLocation{datLoc, ecLoc},
NewEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
DeletedEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
NewEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, 16),
DeletedEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, 16),
}
ecLoc.ecShardNotifyHandler = func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume) {
store.NewEcShardsChan <- master_pb.VolumeEcShardInformationMessage{
store.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{
Id: uint32(vid),
Collection: collection,
}
@@ -231,8 +231,8 @@ func TestIssue9478_ZeroByteSiblingDatKeepsPartialEc(t *testing.T) {
store := &Store{
Locations: []*DiskLocation{datLoc, ecLoc},
NewEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
DeletedEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
NewEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, 16),
DeletedEcShardsChan: make(chan *master_pb.VolumeEcShardInformationMessage, 16),
}
if err := ecLoc.loadAllEcShards(nil); err != nil {
weed/storage/store_ec_reconcile.go
+1 -1
View File
@@ -297,7 +297,7 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() {
loc.removeEcVolumeFiles(v.collection, v.vid)
for _, msg := range v.messages {
select {
case s.DeletedEcShardsChan <- *msg:
case s.DeletedEcShardsChan <- msg:
default:
// Channel full during startup is fine — the next
// periodic heartbeat reports the full ecVolumes
weed/util/slice.go
+16
View File
@@ -0,0 +1,16 @@
package util
// DrainChannel performs a non-blocking drain of a channel after receiving
// the first message. Returns all drained messages including the first one.
// The first message is passed in explicitly (already received by the caller).
func DrainChannel[T any](ch chan T, first T) []T {
result := []T{first}
for {
select {
case v := <-ch:
result = append(result, v)
default:
return result
}
}
}