From 465bfb0ea7c37c7dfff5cfe2f94862d023ac79a3 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Mon, 26 Oct 2026 21:52:28 -0500 Subject: [PATCH] Preserve local working files during Gitea migration --- MP3/membership.go | 572 +++++++++++++++++++++++++++++++++++---------- MP3/membership2.go | 9 + 2 files changed, 457 insertions(+), 124 deletions(-) create mode 100644 MP3/membership2.go diff --git a/MP3/membership.go b/MP3/membership.go index ff217a5..fc8f03d 100644 --- a/MP3/membership.go +++ b/MP3/membership.go @@ -1,13 +1,15 @@ + package main import ( + "bytes" "context" + "encoding/gob" "log/slog" + "math/rand" "net" "sync" "time" - "math/rand" - "encoding/json" ) // this file handles gossip protocol from MP2 @@ -16,93 +18,200 @@ import ( // thread 3 - handles failure detection type Member_ID struct { - ip string - port int - time time.Time + IP string + Port int + Time time.Time } type MemberInfo struct { - heart_beat int - state string - incarnation_counter int - local_failed_since *time.Time - local_last_seen *time.Time + HeartBeat int + State string + local_failed_since *time.Time + local_last_seen *time.Time } type MembershipService struct { - members_list map[Member_ID]MemberInfo + MembersList map[Member_ID]MemberInfo + SelfID Member_ID + deadSessions map[Member_ID]struct{} mutex sync.RWMutex } -type GossipMessageType int - - type GossipMessage struct { - Type string + Type string Senders_id Member_ID - Members map[Member_ID]MemberInfo + Members map[Member_ID]MemberInfo +} + +// One helper for gob encoding and UDP sending +func send_gossip_message( + socket *net.UDPConn, + addr *net.UDPAddr, + gossip_msg GossipMessage, +) error { + var buffer bytes.Buffer + + if err := gob.NewEncoder(&buffer).Encode(gossip_msg); err != nil { + return err + } + + _, err := socket.WriteToUDP(buffer.Bytes(), addr) + return err } func (m *MembershipService) get_current_members() map[Member_ID]MemberInfo { m.mutex.RLock() defer m.mutex.RUnlock() - + // Create a copy of the members_list to avoid race conditions current_members := make(map[Member_ID]MemberInfo) - for id, info := range m.members_list { + + for id, info := range m.MembersList { current_members[id] = info } + return current_members } -func get_all_member_ids(members map[Member_ID]MemberInfo) []Member_ID { - member_ids := make([]Member_ID, 0, len(members) - 1) // Exclude self - for id := range members { - +func (m *MembershipService) get_self_id() Member_ID { + m.mutex.RLock() + defer m.mutex.RUnlock() + + return m.SelfID +} + +func get_all_member_ids( + members map[Member_ID]MemberInfo, + self Member_ID, +) []Member_ID { + member_ids := make([]Member_ID, 0, len(members)) + + for id, _ := range members { + // Exclude self and DOWN nodes + if id != self { + member_ids = append(member_ids, id) + } } + return member_ids } func (m *MembershipService) member_init(ip string, port int) { - my_id := Member_ID{ip, port, time.Now()} - members_info := MemberInfo{0, "alive", 0, nil, nil} + now := time.Now() + + // Round(0) removes monotonic clock data for gob map keys + my_id := Member_ID{ip, port, now.UTC().Round(0)} + members_info := MemberInfo{0, "alive", nil, &now} + m.mutex.Lock() defer m.mutex.Unlock() - m.members_list[my_id] = members_info + + m.SelfID = my_id + m.MembersList[my_id] = members_info } -func (m *MembershipService) handle_join(new_member Member_ID, logger *slog.Logger, socket *net.UDPConn) { +// Add a new process session. +// Caller must hold m.mutex.Lock(). +func (m *MembershipService) add_new_member( + id Member_ID, + info MemberInfo, + now time.Time, +) bool { + // Never accept a session already declared dead + if _, dead := m.deadSessions[id]; dead { + return false + } + + // Check whether we already know a newer session + for old_id := range m.MembersList { + if old_id.IP == id.IP && old_id.Port == id.Port { + if !id.Time.After(old_id.Time) { + return false + } + } + } + + // New process lifetime: retire older sessions + for old_id := range m.MembersList { + if old_id.IP == id.IP && old_id.Port == id.Port { + m.deadSessions[old_id] = struct{}{} + delete(m.MembersList, old_id) + } + } + + info.local_last_seen = &now + info.local_failed_since = nil + + if info.State == "down" { + info.local_failed_since = &now + m.deadSessions[id] = struct{}{} + } + + m.MembersList[id] = info + return true +} + +func (m *MembershipService) handle_join( + new_member Member_ID, + logger *slog.Logger, + socket *net.UDPConn, +) { + if net.ParseIP(new_member.IP) == nil { + logger.Warn("Invalid JOIN member", "member", new_member) + return + } m.mutex.Lock() - m.members_list[new_member] = MemberInfo{0, "alive", 0, nil, nil} + + // Add only if this is a new process session. + // Repeated JOIN does not reset heartbeat tracking. + now := time.Now() + m.add_new_member( + new_member, + MemberInfo{0, "alive", nil, nil}, + now, + ) + m.mutex.Unlock() - //ack payload with teh current membership list + // ACK with current membership list gossip_msg := GossipMessage{ - Type: "gossip", - Senders_id: Member_ID{}, - Members: m.get_current_members(), + Type: "join_ack", + Senders_id: m.get_self_id(), + Members: m.get_current_members(), } - data, err := json.Marshal(gossip_msg) + + err := send_gossip_message( + socket, + &net.UDPAddr{ + IP: net.ParseIP(new_member.IP), + Port: new_member.Port, + }, + gossip_msg, + ) + if err != nil { - logger.Error("Error marshaling gossip message", "error", err) + logger.Warn("Error sending JOIN acknowledgment", "error", err) return } - _, err = socket.WriteToUDP(data, &net.UDPAddr{IP: net.ParseIP(new_member.ip), Port: new_member.port}) - if err != nil { - logger.Error("Error sending gossip message", "error", err) - return - } + logger.Info("JOIN acknowledgment sent", "member", new_member) } - -func (m *MembershipService) handle_gossip_receive(ctx context.Context, socket *net.UDPConn, logger *slog.Logger, config Config, wg *sync.WaitGroup) { +func (m *MembershipService) handle_gossip_receive( + ctx context.Context, + socket *net.UDPConn, + logger *slog.Logger, + config Config, + wg *sync.WaitGroup, +) { defer wg.Done() buffer := make([]byte, 65535) - //send the join message before processing the gossip messages - joined := (config.introducer_ip == config.my_ip.String()) // Introducer node doesn't need to join + + // Introducer doesn't need to JOIN + is_introducer := config.introducer_ip == config.my_ip.String() + joined := is_introducer last_join_time := time.Time{} for { @@ -111,83 +220,130 @@ func (m *MembershipService) handle_gossip_receive(ctx context.Context, socket *n } if !joined && time.Since(last_join_time) >= config.join_retry_interval { - // Send join message to introducer + // Send JOIN to introducer join_msg := GossipMessage{ - Type: "join", - Senders_id: Member_ID{ip: config.my_ip.String(), port: config.port, time: time.Now()}, - Members: map[Member_ID]MemberInfo{}, - } - data, err := json.Marshal(join_msg) - if err != nil { - logger.Error("Error marshaling join message", "error", err) - return + Type: "join", + Senders_id: m.get_self_id(), + Members: map[Member_ID]MemberInfo{}, } + introducer_addr := &net.UDPAddr{ IP: net.ParseIP(config.introducer_ip), Port: config.port, } - _, err = socket.WriteToUDP(data, introducer_addr) - if err != nil { - logger.Error("Error sending join message", "error", err) - return + + if err := send_gossip_message(socket, introducer_addr, join_msg); err != nil { + logger.Warn("Error sending JOIN", "error", err) } + last_join_time = time.Now() } - socket.SetReadDeadline(time.Now().Add(1 * time.Second)) + socket.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + n, _, err := socket.ReadFromUDP(buffer) + if err != nil { if ctx.Err() != nil { return } + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - // Timeout occurred, continue to the next iteration to check for context cancellation continue } - logger.Error("Error reading from UDP socket", "error", err) - return + + logger.Warn("Error reading UDP socket", "error", err) + continue } - + var gossip_msg GossipMessage - err = json.Unmarshal(buffer[:n], &gossip_msg) + + // Decode gob message + err = gob.NewDecoder( + bytes.NewReader(buffer[:n]), + ).Decode(&gossip_msg) + if err != nil { - logger.Error("Error unmarshaling gossip message", "error", err) - return + logger.Warn("Invalid gossip packet", "error", err) + continue } switch gossip_msg.Type { + case "gossip": - m.handle_gossip(gossip_msg.Members, logger) - case "join": - if config.introducer_ip == config.my_ip.String() { - m.handle_join(gossip_msg.Senders_id, logger, socket) - joined = true - } else { - logger.Error("Received join message but not the introducer") + // Merge incoming membership records + // and rejoin if our own session was falsely marked DOWN + if m.handle_gossip(gossip_msg.Members, logger) && !is_introducer { + joined = false + last_join_time = time.Time{} } + + case "join": + if is_introducer { + m.handle_join(gossip_msg.Senders_id, logger, socket) + } + + case "join_ack": + // Ignore delayed ACKs for an older local session + self_id := m.get_self_id() + if _, exists := gossip_msg.Members[self_id]; !exists { + continue + } + + restarted := m.handle_gossip(gossip_msg.Members, logger) + + if restarted { + joined = false + last_join_time = time.Time{} + continue + } + + if !joined { + logger.Info("Successfully joined membership group") + } + + joined = true + default: - logger.Error("Unknown message type received", "type", gossip_msg.Type) + logger.Warn("Unknown gossip message", "type", gossip_msg.Type) } } } -func select_peers(current_members map[Member_ID]MemberInfo, fanout int) []Member_ID { - if len(current_members) <= fanout { - return get_all_member_ids(current_members) +func select_peers( + current_members map[Member_ID]MemberInfo, + fanout int, + self Member_ID, +) []Member_ID { + member_ids := get_all_member_ids(current_members, self) + + if fanout <= 0 { + return nil } - - member_ids := get_all_member_ids(current_members) + + if len(member_ids) <= fanout { + return member_ids + } + selected := make([]Member_ID, 0, fanout) perm := rand.Perm(len(member_ids)) + for i := 0; i < fanout; i++ { selected = append(selected, member_ids[perm[i]]) } + return selected } - -func (m *MembershipService) handle_gossip_send(ctx context.Context, socket *net.UDPConn, logger *slog.Logger, config Config, wg *sync.WaitGroup) { +func (m *MembershipService) handle_gossip_send( + ctx context.Context, + socket *net.UDPConn, + logger *slog.Logger, + config Config, + wg *sync.WaitGroup, +) { defer wg.Done() + ticker := time.NewTicker(config.gossip_heartbeat_interval) defer ticker.Stop() @@ -195,48 +351,162 @@ func (m *MembershipService) handle_gossip_send(ctx context.Context, socket *net. select { case <-ctx.Done(): return + case <-ticker.C: - //increment heart beat for self + // Increment heartbeat for self m.mutex.Lock() - for id, info := range m.members_list { - if id.ip == config.my_ip.String() && id.port == config.port { - info.heart_beat++ - m.members_list[id] = info - break - } - } + + self_id := m.SelfID + info := m.MembersList[self_id] + + info.HeartBeat++ + m.MembersList[self_id] = info + m.mutex.Unlock() current_members := m.get_current_members() - selected_peers := select_peers(current_members, config.gossip_fanout) + + selected_peers := select_peers( + current_members, + config.gossip_fanout, + m.get_self_id(), + ) + + gossip_msg := GossipMessage{ + Type: "gossip", + Senders_id: m.get_self_id(), + Members: current_members, + } for _, peer := range selected_peers { - gossip_msg := GossipMessage{ - Type: "gossip", - Members: current_members, - } - data, err := json.Marshal(gossip_msg) - if err != nil { - logger.Error("Error marshaling gossip message", "error", err) - return - } - // Send the gossip message to the selected peer peer_addr := &net.UDPAddr{ - IP: net.ParseIP(peer.ip), - Port: peer.port, + IP: net.ParseIP(peer.IP), + Port: peer.Port, } - _, err = socket.WriteToUDP(data, peer_addr) - if err != nil { - logger.Error("Error sending gossip message", "error", err) - return + + if err := send_gossip_message(socket, peer_addr, gossip_msg); err != nil { + logger.Warn( + "Error sending gossip", + "peer", peer, + "error", err, + ) + continue } } } } } -func (m *MembershipService) handle_failure_detection(ctx context.Context, socket *net.UDPConn, logger *slog.Logger, config Config, wg *sync.WaitGroup) { +// Merge incoming gossip into local membership. +// Returns true if our own process session had to be replaced. +func (m *MembershipService) handle_gossip( + incoming_members map[Member_ID]MemberInfo, + logger *slog.Logger, +) bool { + m.mutex.Lock() + defer m.mutex.Unlock() + + now := time.Now() + restarted := false + + for id, incoming := range incoming_members { + if incoming.State != "alive" && incoming.State != "down" { + continue + } + + // Handle information about our own machine + if id.IP == m.SelfID.IP && id.Port == m.SelfID.Port { + + // Ignore reports about older local sessions + if id != m.SelfID { + continue + } + + // Someone marked our current session DOWN. + // Create a NEW session timestamp to refute it. + if incoming.State == "down" { + old_id := m.SelfID + m.deadSessions[old_id] = struct{}{} + delete(m.MembersList, old_id) + + new_time := now.UTC().Round(0) + if !new_time.After(old_id.Time) { + new_time = old_id.Time.Add(time.Nanosecond) + } + + new_id := Member_ID{ + IP: old_id.IP, + Port: old_id.Port, + Time: new_time, + } + + m.SelfID = new_id + m.MembersList[new_id] = MemberInfo{ + HeartBeat: 0, + State: "alive", + local_last_seen: &now, + } + + restarted = true + logger.Info("Refuted DOWN report with new session") + } + + continue + } + + // Never resurrect an already dead process session + if _, dead := m.deadSessions[id]; dead { + continue + } + + current, exists := m.MembersList[id] + + // New session discovered through gossip + if !exists { + m.add_new_member(id, incoming, now) + continue + } + + // Same session: DOWN cannot be overridden by ALIVE + if current.State == "down" { + continue + } + + // Incoming DOWN report + if incoming.State == "down" { + if incoming.HeartBeat >= current.HeartBeat { + current.State = "down" + current.local_failed_since = &now + + m.MembersList[id] = current + m.deadSessions[id] = struct{}{} + + logger.Info("Member marked DOWN through gossip", "member", id) + } + continue + } + + // Same session + ALIVE: accept newer heartbeat + if incoming.HeartBeat > current.HeartBeat { + current.HeartBeat = incoming.HeartBeat + current.local_last_seen = &now + + m.MembersList[id] = current + } + } + + return restarted +} + +func (m *MembershipService) handle_failure_detection( + ctx context.Context, + socket *net.UDPConn, + logger *slog.Logger, + config Config, + wg *sync.WaitGroup, +) { defer wg.Done() + ticker := time.NewTicker(500 * time.Millisecond) defer ticker.Stop() @@ -244,28 +514,32 @@ func (m *MembershipService) handle_failure_detection(ctx context.Context, socket select { case <-ctx.Done(): return + case <-ticker.C: m.mutex.Lock() now := time.Now() - for id, info := range m.members_list { - // Don't check ourselves or already failed nodes - if id == m.self_id || info.State == "down" { + for id, info := range m.MembersList { + // Don't check self or already DOWN nodes + if id == m.SelfID || info.State == "down" { continue } - if info.LocalLastSeen == nil { - info.LocalLastSeen = &now - m.members_list[id] = info + if info.local_last_seen == nil { + info.local_last_seen = &now + m.MembersList[id] = info continue } - // No heartbeat advancement for 2.5 seconds - if now.Sub(*info.LocalLastSeen) > 2500*time.Millisecond { + // No heartbeat advancement beyond timeout + if now.Sub(*info.local_last_seen) > config.gossip_failure_timeout { info.State = "down" - info.LocalFailedSince = &now + info.local_failed_since = &now - m.members_list[id] = info + m.MembersList[id] = info + + // Remember this specific process session + m.deadSessions[id] = struct{}{} logger.Warn("Member marked DOWN", "member", id) } @@ -276,35 +550,85 @@ func (m *MembershipService) handle_failure_detection(ctx context.Context, socket } } -func start_membership(ctx context.Context, config Config, logger *slog.Logger) { +func (m *MembershipService) print_membership(logger *slog.Logger) { + ticker := time.NewTicker(150 * time.Millisecond) + defer ticker.Stop() + + for range ticker.C { + current_members := m.get_current_members() + logger.Info("Current Membership List:") + for id, info := range current_members { + logger.Info("Member", "id", id, "state", info.State) + } + } +} + +func start_membership( + ctx context.Context, + config Config, + logger *slog.Logger, +) { logger.Info("Starting Membership") + if config.my_ip == nil || + config.gossip_heartbeat_interval <= 0 || + config.gossip_failure_timeout <= 0 || + config.join_retry_interval <= 0 { + logger.Error("Invalid membership configuration") + return + } + var membership_service MembershipService - membership_service.members_list = make(map[Member_ID]MemberInfo) - membership_service.member_init(config.my_ip.String(), config.port) + + membership_service.MembersList = make(map[Member_ID]MemberInfo) + membership_service.deadSessions = make(map[Member_ID]struct{}) + + membership_service.member_init( + config.my_ip.String(), + config.port, + ) - // 2. Check the error on socket creation socket, err := net.ListenUDP("udp", &net.UDPAddr{ IP: *config.my_ip, Port: config.port, }) + if err != nil { logger.Error("Failed to create UDP listener", "error", err) return } - // 5. Keep defer socket.Close() since the function remains blocking defer socket.Close() - logger.Info("Membership Service Listening on", "IP", config.my_ip.String(), "Port", config.port) + logger.Info( + "Membership Service Listening on", + "IP", config.my_ip.String(), + "Port", config.port, + ) - // 6. Goroutine tracking with WaitGroup var wg sync.WaitGroup wg.Add(3) - go membership_service.handle_gossip_receive(ctx, socket, logger, config, &wg) - go membership_service.handle_gossip_send(ctx, socket, logger, config, &wg) - go membership_service.handle_failure_detection(ctx, socket, logger, config, &wg) + go membership_service.handle_gossip_receive( + ctx, socket, logger, config, &wg, + ) + go membership_service.handle_gossip_send( + ctx, socket, logger, config, &wg, + ) + + go membership_service.handle_failure_detection( + ctx, socket, logger, config, &wg, + ) + + // Wait for cancellation + <-ctx.Done() + + // Close socket to unblock receiver + socket.Close() + + // Wait for all three goroutines to finish wg.Wait() + + logger.Info("Membership Service Stopped") } diff --git a/MP3/membership2.go b/MP3/membership2.go new file mode 100644 index 0000000..0f30331 --- /dev/null +++ b/MP3/membership2.go @@ -0,0 +1,9 @@ +package main + +import ( + "fmt" +) + +func main() { + +} \ No newline at end of file