Fixed internal communications problems including missing removal of adapted newline symbol. Also made InternalSend reliable towards reconnections.

Author: numbleroot

comm/connection.go

@@ -13,7 +13,7 @@ import (
 // ReliableConnect attempts to connect to defined remote node
 // as longs as the error from previous attempts is possible
 // to be dealt with.
-func ReliableConnect(remoteName string, remoteAddr string, tlsConfig *tls.Config, retry int) (*tls.Conn, error) {
+func ReliableConnect(remoteAddr string, tlsConfig *tls.Config, retry int) (*tls.Conn, error) {
 
 	var err error
 	var c *tls.Conn
@@ -33,30 +33,32 @@ func ReliableConnect(remoteName string, remoteAddr string, tlsConfig *tls.Config
 			if err.Error() == okError {
 				time.Sleep(time.Duration(retry) * time.Millisecond)
 			} else {
-				return nil, fmt.Errorf("Could not connect to port of node '%s' because of: %s\n", remoteName, err.Error())
+				return nil, fmt.Errorf("could not connect to port of node '%s' because of: %s", remoteAddr, err.Error())
 			}
 		}
 	}
 
-	log.Printf("Successfully connected to worker node '%s'.\n", remoteName)
-
 	return c, nil
 }
 
-// ReliableSend sends text to other node specified and
-// tries to reconnect in case of simple disconnects.
-func ReliableSend(conn *tls.Conn, text string, remoteName string, remoteAddr string, tlsConfig *tls.Config, timeout int, retry int) error {
+// ReliableSend attempts to transmit a message inbetween
+// pluto nodes. If the first attempt fails, the node will
+// try to reconnect and resend the message until successfully
+// transmitted.
+func ReliableSend(conn *tls.Conn, text string, remoteAddr string, tlsConfig *tls.Config, timeout int, retry int) error {
 
-	var err error
-	var replacedConn *tls.Conn
+	// TODO: Make this routine first check for closed
+	//       pipe and after that attempting 'ping' test
+	//       with exponential (?) backoff for write deadline
+	//       up to timeout.
 
 	// Set configured timeout on waiting for response.
 	conn.SetWriteDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
 
 	// Test long-lived connection.
-	_, err = conn.Write([]byte("> ping <\r\n"))
+	_, err := conn.Write([]byte("> ping <\r\n"))
 	if err != nil {
-		return fmt.Errorf("sending ping to node '%s' failed with: %s\n", remoteName, err.Error())
+		return fmt.Errorf("sending ping to node '%s' failed with: %s\n", remoteAddr, err.Error())
 	}
 
 	// Wait for configured time to pass.
@@ -69,23 +71,25 @@ func ReliableSend(conn *tls.Conn, text string, remoteName string, remoteAddr str
 	_, err = fmt.Fprintf(conn, "%s\r\n", text)
 	for err != nil {
 
-		log.Printf("[comm.ReliableSend] Sending to node '%s' failed, trying to recover...\n", remoteName)
+		log.Printf("[comm.ReliableSend] Sending to node '%s' failed, trying to recover...\n", remoteAddr)
 
 		// Define an error we can deal with.
-		okError := fmt.Sprintf("write tcp %s->%s: write: broken pipe", conn.LocalAddr(), conn.RemoteAddr())
+		okError := fmt.Sprintf("write tcp %s->%s: write: broken pipe", conn.LocalAddr().String(), remoteAddr)
 
 		if err.Error() == okError {
 
 			// Connection was lost. Reconnect.
-			replacedConn, err = ReliableConnect(remoteName, remoteAddr, tlsConfig, retry)
+			conn, err = ReliableConnect(remoteAddr, tlsConfig, retry)
 			if err != nil {
-				return fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
+				return fmt.Errorf("could not reestablish connection with '%s': %s", remoteAddr, err.Error())
 			}
 
-			// Retry transfer.
-			_, err = fmt.Fprintf(replacedConn, "%s\r\n", text)
+			log.Printf("[comm.ReliableSend] Reconnected to '%s'.\n", remoteAddr)
+
+			// Resend message.
+			_, err = fmt.Fprintf(conn, "%s\r\n", text)
 		} else {
-			return fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
+			return fmt.Errorf("could not reestablish connection with '%s': %s", remoteAddr, err.Error())
 		}
 	}
 

comm/receiver.go

@@ -217,7 +217,7 @@ func (recv *Receiver) Shutdown(downRecv chan struct{}) {
 	// Wait for signal.
 	<-downRecv
 
-	log.Printf("[comm.Shutdown] receiver: shutting down...\n")
+	log.Printf("[comm.Shutdown] Receiver: shutting down...\n")
 
 	// Instruct other goroutines to shutdown.
 	recv.shutdown <- struct{}{}
@@ -237,7 +237,7 @@ func (recv *Receiver) Shutdown(downRecv chan struct{}) {
 	recv.socket.Close()
 	recv.lock.Unlock()
 
-	log.Printf("[comm.Shutdown] receiver: done!\n")
+	log.Printf("[comm.Shutdown] Receiver: done!\n")
 }
 
 // IncVClockEntry waits for an incoming name of a node on
@@ -309,25 +309,13 @@ func (recv *Receiver) AcceptIncMsgs() error {
 	// routine on exiting this function.
 	defer recv.wg.Done()
 
-	// Initilize channel to signal end to
-	// storing goroutines further below.
-	downStoring := make(chan struct{})
-
-	// Count how many storing routines were spawned.
-	downNumber := 0
-
 	for {
 
 		select {
 
 		// Check if a shutdown signal was sent.
 		case <-recv.shutdown:
 
-			// Send shutdown signal to each storing routine.
-			for i := 0; i < downNumber; i++ {
-				downStoring <- struct{}{}
-			}
-
 			// Close file descriptor.
 			recv.lock.Lock()
 			recv.writeLog.Close()
@@ -343,104 +331,64 @@ func (recv *Receiver) AcceptIncMsgs() error {
 				return fmt.Errorf("[comm.AcceptIncMsgs] Accepting incoming sync messages at %s failed with: %s\n", recv.name, err.Error())
 			}
 
-			// Dispatch into own goroutine.
-			downNumber += 1
-			go recv.StoreIncMsgs(conn, downStoring)
+			go recv.StoreIncMsgs(conn)
 		}
 	}
 }
 
 // StoreIncMsgs takes received message string and saves
 // it into incoming CRDT message log file.
-func (recv *Receiver) StoreIncMsgs(conn net.Conn, downStoring chan struct{}) {
+func (recv *Receiver) StoreIncMsgs(conn net.Conn) {
 
 	var err error
 
 	// Initial value for received message in order
 	// to skip past the mandatory ping message.
-	msgRaw := "> ping <\r\n"
+	msgRaw := "> ping <"
 
 	// Create new buffered reader for connection.
 	r := bufio.NewReader(conn)
 
-	for msgRaw == "> ping <\r\n" {
+	for msgRaw == "> ping <" {
 
 		// Read string until newline character is received.
 		msgRaw, err = r.ReadString('\n')
 		if err != nil {
 
 			if err.Error() == "EOF" {
-
-				// Error caused by disconnect. Do not crash.
-				log.Printf("[comm.StoreIncMsgs] Node at %s disconnected...\n", conn.RemoteAddr())
-
-				// Simply end this function.
-				msgRaw = "> done <\r\n"
+				log.Printf("[comm.StoreIncMsgs] Reading from closed connection. Ignoring.\n")
+				return
 			} else {
 				log.Fatalf("[comm.StoreIncMsgs] Error while reading sync message: %s\n", err.Error())
 			}
 		}
-	}
-
-	// Unless we do not receive the signal that continuous CRDT
-	// message transmission is done, we accept new messages.
-	for msgRaw != "> done <\r\n" {
 
-		// Lock mutex.
-		recv.lock.Lock()
-
-		// Write it to message log file.
-		_, err = recv.writeLog.WriteString(msgRaw)
-		if err != nil {
-			log.Fatalf("[comm.StoreIncMsgs] Writing to CRDT log file failed with: %s\n", err.Error())
-		}
-
-		// Save to stable storage.
-		err = recv.writeLog.Sync()
-		if err != nil {
-			log.Fatalf("[comm.StoreIncMsgs] Syncing CRDT log file to stable storage failed with: %s\n", err.Error())
-		}
-
-		// Unlock mutex.
-		recv.lock.Unlock()
-
-		// Indicate to applying routine that a new message
-		// is available to process.
-		if len(recv.msgInLog) < 1 {
-			recv.msgInLog <- struct{}{}
-		}
-
-		select {
-
-		case <-downStoring:
-
-			// Break from inifinte loop.
-			msgRaw = "> done <\r\n"
-
-		default:
-
-			// Reset message to expected ping.
-			msgRaw = "> ping <\r\n"
+		// Remove trailing characters denoting line end.
+		msgRaw = strings.TrimRight(msgRaw, "\r\n")
+	}
 
-			for msgRaw == "> ping <\r\n" {
+	// Lock mutex.
+	recv.lock.Lock()
 
-				// Read next CRDT message until newline character is received.
-				msgRaw, err = r.ReadString('\n')
-				if err != nil {
+	// Write it to message log file.
+	_, err = recv.writeLog.WriteString(msgRaw)
+	if err != nil {
+		log.Fatalf("[comm.StoreIncMsgs] Writing to CRDT log file failed with: %s\n", err.Error())
+	}
 
-					if err.Error() == "EOF" {
+	// Save to stable storage.
+	err = recv.writeLog.Sync()
+	if err != nil {
+		log.Fatalf("[comm.StoreIncMsgs] Syncing CRDT log file to stable storage failed with: %s\n", err.Error())
+	}
 
-						// Error caused by disconnect. Do not crash.
-						log.Printf("[comm.StoreIncMsgs] Node at %s disconnected...\n", conn.RemoteAddr())
+	// Unlock mutex.
+	recv.lock.Unlock()
 
-						// Simply end this function.
-						msgRaw = "> done <\r\n"
-					} else {
-						log.Fatalf("[comm.StoreIncMsgs] Error while reading sync message: %s\n", err.Error())
-					}
-				}
-			}
-		}
+	// Indicate to applying routine that a new message
+	// is available to process.
+	if len(recv.msgInLog) < 1 {
+		recv.msgInLog <- struct{}{}
 	}
 }
 

comm/sender.go

@@ -282,16 +282,16 @@ func (sender *Sender) SendMsgs() {
 				// Unlock mutex.
 				sender.lock.Unlock()
 
-				for nodeName, nodeAddr := range sender.nodes {
+				for _, nodeAddr := range sender.nodes {
 
 					// Connect to node.
-					conn, err := ReliableConnect(nodeName, nodeAddr, sender.tlsConfig, sender.intlConnRetry)
+					conn, err := ReliableConnect(nodeAddr, sender.tlsConfig, sender.intlConnRetry)
 					if err != nil {
 						log.Fatalf("[comm.SendMsgs] Failed to connect to %s: %s\n", err.Error())
 					}
 
 					// Send payload reliably to other nodes.
-					err = ReliableSend(conn, marshalledMsg, nodeName, nodeAddr, sender.tlsConfig, sender.intlConnTimeout, sender.intlConnRetry)
+					err = ReliableSend(conn, marshalledMsg, nodeAddr, sender.tlsConfig, sender.intlConnTimeout, sender.intlConnRetry)
 					if err != nil {
 						log.Fatalf("[comm.SendMsgs] Failed to send: %s\n", err.Error())
 					}

imap/connection.go

@@ -8,6 +8,8 @@ import (
 
 	"crypto/tls"
 	"path/filepath"
+
+	"github.com/numbleroot/pluto/comm"
 )
 
 // Constants
@@ -35,14 +37,15 @@ type IMAPState int
 // a pluto node that only authenticates and proxies
 // IMAP connections.
 type Connection struct {
-	IncConn   *tls.Conn
-	IncReader *bufio.Reader
-	OutConn   *tls.Conn
-	OutReader *bufio.Reader
-	OutIP     string
-	OutPort   string
-	ClientID  string
-	UserName  string
+	IncConn       *tls.Conn
+	IncReader     *bufio.Reader
+	OutConn       *tls.Conn
+	OutReader     *bufio.Reader
+	IntlTLSConfig *tls.Config
+	IntlConnRetry int
+	OutAddr       string
+	ClientID      string
+	UserName      string
 }
 
 // IMAPConnection contains additional elements needed
@@ -83,8 +86,8 @@ func (c *Connection) Send(inc bool, text string) error {
 
 // InternalSend is used by nodes of the pluto system to
 // successfully transmit a message to another node or fail
-// definitely. This prevents further handler advancement
-// in case a link failed.
+// definitely if no reconnection is possible. This prevents
+// further handler advancement in case a link failed.
 func (c *Connection) InternalSend(inc bool, text string) error {
 
 	// Check which attached connection should be used.
@@ -96,13 +99,51 @@ func (c *Connection) InternalSend(inc bool, text string) error {
 	// Test if connection is still healthy.
 	_, err := conn.Write([]byte("> ping <\r\n"))
 	if err != nil {
-		return fmt.Errorf("sending ping to node '%s' failed: %s\n", conn.RemoteAddr().String(), err.Error())
+		return fmt.Errorf("sending ping to node '%s' failed: %s", conn.RemoteAddr().String(), err.Error())
 	}
 
 	// Write message to TLS connections.
 	_, err = fmt.Fprintf(conn, "%s\r\n", text)
 	for err != nil {
-		return fmt.Errorf("sending message to node '%s' failed: %s\n", conn.RemoteAddr().String(), err.Error())
+
+		log.Printf("[imap.InternalSend] Sending to node '%s' failed, trying to recover...\n", conn.RemoteAddr())
+
+		// Define what IP and port of remote node look like.
+		remoteAddr := conn.RemoteAddr().String()
+
+		// Define an error we can deal with.
+		okError := fmt.Sprintf("write tcp %s->%s: write: broken pipe", conn.LocalAddr().String(), remoteAddr)
+
+		if err.Error() == okError {
+
+			// Reestablish TLS connection to remote node.
+			conn, err = comm.ReliableConnect(remoteAddr, c.IntlTLSConfig, c.IntlConnRetry)
+			if err != nil {
+				return fmt.Errorf("failed to reestablish connection with '%s': %s", remoteAddr, err.Error())
+			}
+
+			// Save context to connection.
+			if inc {
+				c.IncConn = conn
+				c.IncReader = bufio.NewReader(conn)
+			} else {
+				c.OutConn = conn
+				c.OutReader = bufio.NewReader(conn)
+
+				// Inform remote node about which session was active.
+				err = c.SignalSessionStart(false)
+				if err != nil {
+					return fmt.Errorf("signalling session to remote node failed with: %s", err.Error())
+				}
+			}
+
+			log.Printf("[imap.InternalSend] Reconnected to '%s'.\n", remoteAddr)
+
+			// Resend message to remote node.
+			_, err = fmt.Fprintf(conn, "%s\r\n", text)
+		} else {
+			return fmt.Errorf("failed to send message to remote node '%s': %s", remoteAddr, err.Error())
+		}
 	}
 
 	return nil
@@ -205,7 +246,7 @@ func (c *Connection) SignalSessionStart(inc bool) error {
 func (c *Connection) SignalSessionStartFailover(inc bool, recvClientID string, origWorker string) error {
 
 	// Text to send.
-	msg := strings.Replace(recvClientID, " <", fmt.Sprintf("%s <", origWorker), -1)
+	msg := strings.Replace(recvClientID, " <", fmt.Sprintf(" %s <", origWorker), -1)
 
 	// Send session information internally.
 	err := c.InternalSend(inc, msg)
@@ -279,7 +320,7 @@ func (c *Connection) Terminate() error {
 func (c *Connection) Error(msg string, err error) {
 
 	// Log error.
-	log.Printf("%s: %s. Terminating connection to %s\n", msg, err.Error(), c.IncConn.RemoteAddr().String())
+	log.Printf("%s: %s. Terminating connection to %s.\n", msg, err.Error(), c.IncConn.RemoteAddr().String())
 
 	// Terminate connection.
 	err = c.Terminate()