You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: master_grpc_server.go
    path of file: ./repos/seaweedfs/weed/server
    
    the code of the file until where you have to start completion: package weed_server

import (
	"context"
	"errors"
	"fmt"
	"github.com/seaweedfs/seaweedfs/weed/cluster"
	"net"
	"sort"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/stats"
	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
	"github.com/seaweedfs/seaweedfs/weed/util"

	"github.com/seaweedfs/raft"
	"google.golang.org/grpc/peer"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/topology"
)

func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) {
	ms.Topo.UuidAccessLock.Lock()
	defer ms.Topo.UuidAccessLock.Unlock()
	key := fmt.Sprintf("%s:%d", heartbeat.Ip, heartbeat.Port)
	if ms.Topo.UuidMap == nil {
		ms.Topo.UuidMap = make(map[string][]string)
	}
	// find whether new uuid exists
	for k, v := range ms.Topo.UuidMap {
		sort.Strings(v)
		for _, id := range heartbeat.LocationUuids {
			index := sort.SearchStrings(v, id)
			if index < len(v) && v[index] == id {
				duplicated_uuids = append(duplicated_uuids, id)
				glog.Errorf("directory of %s on %s has been loaded", id, k)
			}
		}
	}
	if len(duplicated_uuids) > 0 {
		return duplicated_uuids, errors.New("volume: Duplicated volume directories were loaded")
	}

	ms.Topo.UuidMap[key] = heartbeat.LocationUuids
	glog.V(0).Infof("found new uuid:%v %v , %v", key, heartbeat.LocationUuids, ms.Topo.UuidMap)
	return nil, nil
}

func (ms *MasterServer) UnRegisterUuids(ip string, port int) {
	ms.Topo.UuidAccessLock.Lock()
	defer ms.Topo.UuidAccessLock.Unlock()
	key := fmt.Sprintf("%s:%d", ip, port)
	delete(ms.Topo.UuidMap, key)
	glog.V(0).Infof("remove volume server %v, online volume server: %v", key, ms.Topo.UuidMap)
}

func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
	var dn *topology.DataNode

	defer func() {
		if dn != nil {
			dn.Counter--
			if dn.Counter > 0 {
				glog.V(0).Infof("disconnect phantom volume server %s:%d remaining %d", dn.Ip, dn.Port, dn.Counter)
				return
			}

			message := &master_pb.VolumeLocation{
				DataCenter: dn.GetDataCenterId(),
				Url:        dn.Url(),
				PublicUrl:  dn.PublicUrl,
			}
			for _, v := range dn.GetVolumes() {
				message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
			}
			for _, s := range dn.GetEcShards() {
				message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
			}

			// if the volume server disconnects and reconnects quickly
			//  the unregister and register can race with each other
			ms.Topo.UnRegisterDataNode(dn)
			glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
			ms.UnRegisterUuids(dn.Ip, dn.Port)

			if len(message.DeletedVids) > 0 {
				ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
			}
		}
	}()

	for {
		heartbeat, err := stream.Recv()
		if err != nil {
			if dn != nil {
				glog.Warningf("SendHeartbeat.Recv server %s:%d : %v", dn.Ip, dn.Port, err)
			} else {
				glog.Warningf("SendHeartbeat.Recv: %v", err)
			}
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("error").Inc()
			return err
		}

		if !ms.Topo.IsLeader() {
			// tell the volume servers about the leader
			newLeader, err := ms.Topo.Leader()
			if err != nil {
				glog.Warningf("SendHeartbeat find leader: %v", err)
				return err
			}
			if err := stream.Send(&master_pb.HeartbeatResponse{
				Leader: string(newLeader),
			}); err != nil {
				if dn != nil {
					glog.Warningf("SendHeartbeat.Send response to %s:%d %v", dn.Ip, dn.Port, err)
				} else {
					glog.Warningf("SendHeartbeat.Send response %v", err)
				}
				return err
			}
			continue
		}

		ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
		if dn == nil {
			// Skip delta heartbeat for volume server versions  better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630
			if heartbeat.Ip == "" {
				continue
			} // ToDo must be removed after update major version
			dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
			dc := ms.Topo.GetOrCreateDataCenter(dcName)
			rack := dc.GetOrCreateRack(rackName)
			dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
			glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
			uuidlist, err := ms.RegisterUuids(heartbeat)
			if err != nil {
				if stream_err := stream.Send(&master_pb.HeartbeatResponse{
					DuplicatedUuids: uuidlist,
				}); stream_err != nil {
					glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
					return stream_err
				}
				return err
			}

			if err := stream.Send(&master_pb.HeartbeatResponse{
				VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
			}); err != nil {
				glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
				return err
			}
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc()
			dn.Counter++
		}

		dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)

		glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
		stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()

		message := &master_pb.VolumeLocation{
			Url:        dn.Url(),
			PublicUrl:  dn.PublicUrl,
			DataCenter: dn.GetDataCenterId(),
		}
		if len(heartbeat.NewVolumes) > 0 {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("newVolumes").Inc()
		}
		if len(heartbeat.DeletedVolumes) > 0 {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("deletedVolumes").Inc()
		}
		if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
			// process delta volume ids if exists for fast volume id updates
			for _, volInfo := range heartbeat.NewVolumes {
				message.NewVids = append(message.NewVids, volInfo.Id)
			}
			for _, volInfo := range heartbeat.DeletedVolumes {
				message.DeletedVids = append(message.DeletedVids, volInfo.Id)
			}
			// update master internal volume layouts
			ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
		}

		if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
			i
    
func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
	var dn *topology.DataNode

	defer func() {
		if dn != nil {
			dn.Counter--
			if dn.Counter > 0 {
				glog.V(0).Infof("disconnect phantom volume server %s:%d remaining %d", dn.Ip, dn.Port, dn.Counter)
				return
			}

			message := &master_pb.VolumeLocation{
				DataCenter: dn.GetDataCenterId(),
				Url:        dn.Url(),
				PublicUrl:  dn.PublicUrl,
			}
			for _, v := range dn.GetVolumes() {
				message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
			}
			for _, s := range dn.GetEcShards() {
				message.DeletedVids = append(message.DeletedVids, uint32(s.VolumeId))
			}

			// if the volume server disconnects and reconnects quickly
			//  the unregister and register can race with each other
			ms.Topo.UnRegisterDataNode(dn)
			glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
			ms.UnRegisterUuids(dn.Ip, dn.Port)

			if len(message.DeletedVids) > 0 {
				ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
			}
		}
	}()

	for {
		heartbeat, err := stream.Recv()
		if err != nil {
			if dn != nil {
				glog.Warningf("SendHeartbeat.Recv server %s:%d : %v", dn.Ip, dn.Port, err)
			} else {
				glog.Warningf("SendHeartbeat.Recv: %v", err)
			}
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("error").Inc()
			return err
		}

		if !ms.Topo.IsLeader() {
			// tell the volume servers about the leader
			newLeader, err := ms.Topo.Leader()
			if err != nil {
				glog.Warningf("SendHeartbeat find leader: %v", err)
				return err
			}
			if err := stream.Send(&master_pb.HeartbeatResponse{
				Leader: string(newLeader),
			}); err != nil {
				if dn != nil {
					glog.Warningf("SendHeartbeat.Send response to %s:%d %v", dn.Ip, dn.Port, err)
				} else {
					glog.Warningf("SendHeartbeat.Send response %v", err)
				}
				return err
			}
			continue
		}

		ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey)
		if dn == nil {
			// Skip delta heartbeat for volume server versions  better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630
			if heartbeat.Ip == "" {
				continue
			} // ToDo must be removed after update major version
			dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
			dc := ms.Topo.GetOrCreateDataCenter(dcName)
			rack := dc.GetOrCreateRack(rackName)
			dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.MaxVolumeCounts)
			glog.V(0).Infof("added volume server %d: %v:%d %v", dn.Counter, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids)
			uuidlist, err := ms.RegisterUuids(heartbeat)
			if err != nil {
				if stream_err := stream.Send(&master_pb.HeartbeatResponse{
					DuplicatedUuids: uuidlist,
				}); stream_err != nil {
					glog.Warningf("SendHeartbeat.Send DuplicatedDirectory response to %s:%d %v", dn.Ip, dn.Port, stream_err)
					return stream_err
				}
				return err
			}

			if err := stream.Send(&master_pb.HeartbeatResponse{
				VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
			}); err != nil {
				glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
				return err
			}
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("dataNode").Inc()
			dn.Counter++
		}

		dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)

		glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
		stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()

		message := &master_pb.VolumeLocation{
			Url:        dn.Url(),
			PublicUrl:  dn.PublicUrl,
			DataCenter: dn.GetDataCenterId(),
		}
		if len(heartbeat.NewVolumes) > 0 {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("newVolumes").Inc()
		}
		if len(heartbeat.DeletedVolumes) > 0 {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("deletedVolumes").Inc()
		}
		if len(heartbeat.NewVolumes) > 0 || len(heartbeat.DeletedVolumes) > 0 {
			// process delta volume ids if exists for fast volume id updates
			for _, volInfo := range heartbeat.NewVolumes {
				message.NewVids = append(message.NewVids, volInfo.Id)
			}
			for _, volInfo := range heartbeat.DeletedVolumes {
				message.DeletedVids = append(message.DeletedVids, volInfo.Id)
			}
			// update master internal volume layouts
			ms.Topo.IncrementalSyncDataNodeRegistration(heartbeat.NewVolumes, heartbeat.DeletedVolumes, dn)
		}

		if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
			if heartbeat.Ip != "" {
				dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
				ms.Topo.DataNodeRegistration(dcName, rackName, dn)
			}

			// process heartbeat.Volumes
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("Volumes").Inc()
			newVolumes, deletedVolumes := ms.Topo.SyncDataNodeRegistration(heartbeat.Volumes, dn)

			for _, v := range newVolumes {
				glog.V(0).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
				message.NewVids = append(message.NewVids, uint32(v.Id))
			}
			for _, v := range deletedVolumes {
				glog.V(0).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
				message.DeletedVids = append(message.DeletedVids, uint32(v.Id))
			}
		}

		if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc()
			// update master internal volume layouts
			ms.Topo.IncrementalSyncDataNodeEcShards(heartbeat.NewEcShards, heartbeat.DeletedEcShards, dn)

			for _, s := range heartbeat.NewEcShards {
				message.NewEcVids = append(message.NewEcVids, s.Id)
			}
			for _, s := range heartbeat.DeletedEcShards {
				if dn.HasEcShards(needle.VolumeId(s.Id)) {
					continue
				}
				message.DeletedEcVids = append(message.DeletedEcVids, s.Id)
			}

		}

		if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
			stats.MasterReceivedHeartbeatCounter.WithLabelValues("ecShards").Inc()
			glog.V(4).Infof("master received ec shards from %s: %+v", dn.Url(), heartbeat.EcShards)
			newShards, deletedShards := ms.Topo.SyncDataNodeEcShards(heartbeat.EcShards, dn)

			// broadcast the ec vid changes to master clients
			for _, s := range newShards {
				message.NewEcVids = append(message.NewEcVids, uint32(s.VolumeId))
			}
			for _, s := range deletedShards {
				if dn.HasVolumesById(s.VolumeId) {
					continue
				}
				message.DeletedEcVids = append(message.DeletedEcVids, uint32(s.VolumeId))
			}

		}
		if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
			ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
		}
	}
}