diff --git a/MP3/consistency.go b/MP3/consistency.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/consistency.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/eventual.go b/MP3/eventual.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/eventual.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/main.go b/MP3/main.go new file mode 100644 index 0000000..e560ff0 --- /dev/null +++ b/MP3/main.go @@ -0,0 +1,35 @@ +package main + +import ( + // "fmt" + "net" + "os" + "time" + // "net" + "context" + "log/slog" +) + +type Config struct{ + introducer_ip string + port int + my_ip *net.IP + gossip_heartbeat_interval time.Duration + join_retry_interval time.Duration + gossip_fanout int + gossip_failure_timeout time.Duration +} + +func main(){ + ctx := context.TODO() + + logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) + + //dial up trick to get my_ip + conn, _ := net.Dial("udp", "8.8.8.8:80"); defer conn.Close(); my_ip := conn.LocalAddr().(*net.UDPAddr).IP + + config := Config{"172.22.156.3", 8989, &my_ip, 250 * time.Millisecond, 1 * time.Second, 2, 6 * time.Second} + + start_membership(ctx, config, logger) + +} \ No newline at end of file diff --git a/MP3/membership.go b/MP3/membership.go new file mode 100644 index 0000000..3b7408a --- /dev/null +++ b/MP3/membership.go @@ -0,0 +1,284 @@ +package main + +import ( + "context" + "log/slog" + "net" + "sync" + "time" + "math/rand" + "encoding/json" +) + +// this file handles gossip protocol from MP2 +// thread 1 - handles receiving gossip +// thread 2 - handles sending gossips and updating heart beat +// thread 3 - handles failure detection + +type Member_ID struct { + 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 +} + +type MembershipService struct { + members_list map[Member_ID]MemberInfo + mutex sync.RWMutex +} + +type GossipMessageType int + + +type GossipMessage struct { + Type string + Senders_id Member_ID + Members map[Member_ID]MemberInfo +} + +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 { + 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 { + + } + 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} + m.mutex.Lock() + defer m.mutex.Unlock() + m.members_list[my_id] = members_info +} + +func (m *MembershipService) handle_join(new_member Member_ID, logger *slog.Logger, socket *net.UDPConn) { + + m.mutex.Lock() + m.members_list[new_member] = MemberInfo{0, "alive", 0, nil, nil} + m.mutex.Unlock() + + //ack payload with teh current membership list + gossip_msg := GossipMessage{ + Type: "gossip", + Senders_id: Member_ID{}, + Members: m.get_current_members(), + } + data, err := json.Marshal(gossip_msg) + if err != nil { + logger.Error("Error marshaling gossip message", "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 + } +} + + +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 + last_join_time := time.Time{} + + for { + if ctx.Err() != nil { + return + } + + if !joined && time.Since(last_join_time) >= config.join_retry_interval { + // Send join message 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 + } + 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 + } + last_join_time = time.Now() + } + + socket.SetReadDeadline(time.Now().Add(1 * time.Second)) + 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 + } + + var gossip_msg GossipMessage + err = json.Unmarshal(buffer[:n], &gossip_msg) + if err != nil { + logger.Error("Error unmarshaling gossip message", "error", err) + return + } + + 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") + } + default: + logger.Error("Unknown message type received", "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) + } + + member_ids := get_all_member_ids(current_members) + 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) { + defer wg.Done() + ticker := time.NewTicker(config.gossip_heartbeat_interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + //increment heart beat 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 + } + } + m.mutex.Unlock() + + current_members := m.get_current_members() + selected_peers := select_peers(current_members, config.gossip_fanout) + + 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, + } + _, err = socket.WriteToUDP(data, peer_addr) + if err != nil { + logger.Error("Error sending gossip message", "error", err) + return + } + } + } + } +} + +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() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // TODO: Implement failure detection logic + } + } +} + +func start_membership(ctx context.Context, config Config, logger *slog.Logger) { + logger.Info("Starting Membership") + + var membership_service MembershipService + membership_service.members_list = make(map[Member_ID]MemberInfo) + 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) + + // 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) + + wg.Wait() +} diff --git a/MP3/peer.go b/MP3/peer.go deleted file mode 100644 index 73dafbb..0000000 --- a/MP3/peer.go +++ /dev/null @@ -1,11 +0,0 @@ -package main - -import ( - "fmt" - "os" - "net" -) - -func main(){ - -} \ No newline at end of file diff --git a/MP3/replair.go b/MP3/replair.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/replair.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/replication.go b/MP3/replication.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/replication.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/storage.go b/MP3/storage.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/storage.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/strong.go b/MP3/strong.go new file mode 100644 index 0000000..85f0393 --- /dev/null +++ b/MP3/strong.go @@ -0,0 +1 @@ +package main \ No newline at end of file diff --git a/MP3/utils.go b/MP3/utils.go new file mode 100644 index 0000000..c9ecbf5 --- /dev/null +++ b/MP3/utils.go @@ -0,0 +1,2 @@ +package main +