diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go index dc409038b..82b5ccead 100644 --- a/weed/server/volume_grpc_client_to_master.go +++ b/weed/server/volume_grpc_client_to_master.go @@ -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 diff --git a/weed/storage/store.go b/weed/storage/store.go index e65f69d90..fbeee62b5 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -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 diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 018f18d72..8c2f8b3de 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -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 } diff --git a/weed/storage/store_ec_disk_type_test.go b/weed/storage/store_ec_disk_type_test.go index 711986519..921db2c33 100644 --- a/weed/storage/store_ec_disk_type_test.go +++ b/weed/storage/store_ec_disk_type_test.go @@ -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. diff --git a/weed/storage/store_ec_hybrid_repro_test.go b/weed/storage/store_ec_hybrid_repro_test.go index db245e8c7..163abd391 100644 --- a/weed/storage/store_ec_hybrid_repro_test.go +++ b/weed/storage/store_ec_hybrid_repro_test.go @@ -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 { diff --git a/weed/storage/store_ec_reconcile.go b/weed/storage/store_ec_reconcile.go index 3c35ba75e..6addef419 100644 --- a/weed/storage/store_ec_reconcile.go +++ b/weed/storage/store_ec_reconcile.go @@ -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 diff --git a/weed/util/slice.go b/weed/util/slice.go new file mode 100644 index 000000000..6b60beec5 --- /dev/null +++ b/weed/util/slice.go @@ -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 + } + } +}