Fixes and container killer script

common/communication/UDPClient.go → common/communication/udp_client.go

@@ -8,24 +8,39 @@ import (
 )
 
 type UdpClient struct {
-	conn net.Conn
+	conn    net.Conn
+	address string
 }
 
 const UdpReadTimeout = 400
 
 func NewUdpClient(address string) (*UdpClient, error) {
+	conn, err := connectUDP(address)
+	return &UdpClient{
+			conn:    conn,
+			address: address,
+		},
+		err
+}
+
+func connectUDP(address string) (net.Conn, error) {
 	conn, err := net.Dial("udp", address)
 	if err != nil {
 		log.Errorf("UdpClient | Error trying to create | %v", err)
 		return nil, err
 	}
-	return &UdpClient{
-			conn: conn,
-		},
-		nil
+	return conn, nil
 }
 
 func (u *UdpClient) Receive(sizeToRecv uint) ([]byte, *net.UDPAddr, error) {
+	if u.conn == nil {
+		conn, err := connectUDP(u.address)
+		if err != nil {
+			log.Errorf("UdpClient | Error trying to create | %v", err)
+			return []byte{}, nil, err
+		}
+		u.conn = conn
+	}
 	err := u.conn.SetReadDeadline(time.Now().Add(UdpReadTimeout * time.Millisecond))
 	if err != nil {
 		log.Errorf("UdpClient | Error setting read timeout | %v", err)
@@ -46,6 +61,14 @@ func (u *UdpClient) Receive(sizeToRecv uint) ([]byte, *net.UDPAddr, error) {
 }
 
 func (u *UdpClient) Send(message []byte, _ *net.UDPAddr) (int, error) {
+	if u.conn == nil {
+		conn, err := connectUDP(u.address)
+		if err != nil {
+			log.Errorf("UdpClient | Error trying to create | %v", err)
+			return 0, err
+		}
+		u.conn = conn
+	}
 	sizeSent, err := u.conn.Write(message)
 	if err != nil {
 		log.Errorf("UdpServer | Error trying to Write | %v", err)

common/leader/leader_election_service.go → common/leader/leaderelectionservice.go

@@ -7,6 +7,7 @@ import (
 	"github.com/brunograssano/Distribuidos-TP1/common/protocol/sockets"
 	log "github.com/sirupsen/logrus"
 	"net"
+	"strconv"
 )
 
 type ElectionService interface {
@@ -18,22 +19,26 @@ type ElectionService interface {
 type LeaderElectionService struct {
 	currentState     BullyState
 	id               uint8
-	address          *net.UDPAddr
+	address          []string
 	leaderID         uint8
 	listener         *sockets.UdpProtocolhandler
-	netAddresses     map[uint8]*net.UDPAddr
+	netAddresses     map[uint8][]string
 	netClientSockets map[uint8]*sockets.UdpProtocolhandler
 	leaderDown       chan bool
 }
 
-func NewLeaderElectionService(id uint8, networkNodes map[uint8]*net.UDPAddr, myAddr *net.UDPAddr) *LeaderElectionService {
+func NewLeaderElectionService(id uint8, networkNodes map[uint8][]string, myAddr []string) *LeaderElectionService {
 	netClientSockets := make(map[uint8]*sockets.UdpProtocolhandler)
-	listenerSocketUdp, err := communication.NewUdpServer(myAddr.IP.String(), myAddr.Port)
+	port, err := strconv.Atoi(myAddr[1])
+	if err != nil {
+		log.Errorf("LeaderElectionService | Error converting port from string to int | %v", err)
+	}
+	listenerSocketUdp, err := communication.NewUdpServer(myAddr[0], port)
 	if err != nil {
 		log.Fatalf("LeaderElectionService %v | Error trying to create UDP Server Socket | %v", id, err)
 	}
 	for idNode, udpAddr := range networkNodes {
-		udpCli, err := communication.NewUdpClient(fmt.Sprintf("%v:%v", udpAddr.IP.String(), udpAddr.Port))
+		udpCli, err := communication.NewUdpClient(fmt.Sprintf("%v:%v", udpAddr[0], udpAddr[1]))
 		if err != nil {
 			log.Errorf("LeaderElectionService | Error trying to create UDP Client Socket | %v", err)
 		}

common/leader/leaderelectionservice_test.go

@@ -0,0 +1,87 @@
+package leader
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestReceivingElectionFromMinorIDShouldRetransmitToGreaterID(t *testing.T) {
+	addressMap := make(map[uint8][]string)
+	addressMap[2] = []string{"127.0.0.1", "50012"}
+	addressMap[3] = []string{"127.0.0.1", "50013"}
+	leaderService1 := NewLeaderElectionService(1, addressMap, []string{"127.0.0.1", "50011"})
+	defer leaderService1.Close()
+
+	addressMap2 := make(map[uint8][]string)
+	addressMap2[1] = []string{"127.0.0.1", "50011"}
+	addressMap2[3] = []string{"127.0.0.1", "50013"}
+	leaderService2 := NewLeaderElectionService(2, addressMap2, []string{"127.0.0.1", "50012"})
+	defer leaderService2.Close()
+
+	addressMap3 := make(map[uint8][]string)
+	addressMap3[1] = []string{"127.0.0.1", "50011"}
+	addressMap3[2] = []string{"127.0.0.1", "50012"}
+	leaderService3 := NewLeaderElectionService(3, addressMap3, []string{"127.0.0.1", "50013"})
+	defer leaderService3.Close()
+
+	go leaderService1.ReceiveNetMessages()
+	go leaderService2.ReceiveNetMessages()
+	go leaderService3.ReceiveNetMessages()
+
+	time.Sleep(2 * time.Second)
+
+	assert.Falsef(t, leaderService1.AmILeader(), "LeaderService1 should not be leader")
+	assert.Falsef(t, leaderService2.AmILeader(), "LeaderService2 should not be leader")
+	assert.Truef(t, leaderService3.AmILeader(), "LeaderService3 should be leader")
+}
+
+func TestShouldProclaimItselfAsTheLeaderIfItIsTheOnlyNode(t *testing.T) {
+	addressMap := make(map[uint8][]string)
+	addressMap[2] = []string{"127.0.0.1", "50022"}
+	addressMap[3] = []string{"127.0.0.1", "50023"}
+	leaderService1 := NewLeaderElectionService(1, addressMap, []string{"127.0.0.1", "50021"})
+	defer leaderService1.Close()
+
+	go leaderService1.ReceiveNetMessages()
+
+	time.Sleep(5 * time.Second)
+
+	assert.Truef(t, leaderService1.AmILeader(), "LeaderService1 should be leader")
+}
+
+func TestLeaderFallsDownAndNewLeaderIsElected(t *testing.T) {
+	addressMap := make(map[uint8][]string)
+	addressMap[2] = []string{"127.0.0.1", "50042"}
+	addressMap[3] = []string{"127.0.0.1", "50043"}
+	leaderService1 := NewLeaderElectionService(1, addressMap, []string{"127.0.0.1", "50041"})
+	defer leaderService1.Close()
+
+	addressMap2 := make(map[uint8][]string)
+	addressMap2[1] = []string{"127.0.0.1", "50041"}
+	addressMap2[3] = []string{"127.0.0.1", "50043"}
+	leaderService2 := NewLeaderElectionService(2, addressMap2, []string{"127.0.0.1", "50042"})
+	defer leaderService2.Close()
+
+	addressMap3 := make(map[uint8][]string)
+	addressMap3[1] = []string{"127.0.0.1", "50041"}
+	addressMap3[2] = []string{"127.0.0.1", "50042"}
+	leaderService3 := NewLeaderElectionService(3, addressMap3, []string{"127.0.0.1", "50043"})
+
+	go leaderService1.ReceiveNetMessages()
+	go leaderService2.ReceiveNetMessages()
+	go leaderService3.ReceiveNetMessages()
+
+	time.Sleep(2 * time.Second)
+
+	assert.Falsef(t, leaderService1.AmILeader(), "LeaderService1 should not be leader")
+	assert.Falsef(t, leaderService2.AmILeader(), "LeaderService2 should not be leader")
+	assert.Truef(t, leaderService3.AmILeader(), "LeaderService3 should be leader")
+	leaderService3.Close()
+
+	time.Sleep(5 * time.Second)
+
+	assert.Falsef(t, leaderService1.AmILeader(), "LeaderService1 should not be the new leader")
+	assert.Truef(t, leaderService2.AmILeader(), "LeaderService2 should be the new leader")
+
+}

common/leader/replica_state.go

@@ -19,8 +19,8 @@ type ReplicaState struct {
 	leaderDown    chan bool
 }
 
-func NewReplicaState(leaderAddress *net.UDPAddr, nodeId uint8, leaderDown chan bool) (*ReplicaState, error) {
-	udpCli, err := communication.NewUdpClient(fmt.Sprintf("%v:%v", leaderAddress.IP.String(), leaderAddress.Port))
+func NewReplicaState(leaderAddress []string, nodeId uint8, leaderDown chan bool) (*ReplicaState, error) {
+	udpCli, err := communication.NewUdpClient(fmt.Sprintf("%v:%v", leaderAddress[0], leaderAddress[1]))
 	if err != nil {
 		log.Errorf("ReplicaState | Error instantiating udp Client | %v", err)
 		return nil, err