Communication in Distributed Systems

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// Synchronous message passing pseudocode
// The sender blocks until receiver accepts the message
 
void sender_process() {
    message_t msg;
    prepare_message(&msg, "Request data");
    
    // BLOCKS until receiver calls receive()
    sync_send(RECEIVER_ID, &msg);
    
    // When we reach here, we KNOW receiver got the message
    printf("Message delivered and received\n");
}
 
void receiver_process() {
    message_t msg;
    
    // BLOCKS until sender calls send()
    sync_receive(SENDER_ID, &msg);
    
    // Both parties synchronized at the handoff point
    process_message(&msg);
}
 
// Timeline visualization:
//
// Sender:   ----[send]----X-----block------X----[continue]---->
// Receiver: --------[block]------X----[receive]----[process]--->
//                                ^
//                          Rendezvous point

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// Asynchronous message passing pseudocode
// Sender doesn't wait for receiver
 
void sender_process() {
    message_t msg1, msg2, msg3;
    
    // All sends return immediately - messages queued
    async_send(RECEIVER_ID, &msg1);  // Returns immediately
    async_send(RECEIVER_ID, &msg2);  // Returns immediately
    async_send(RECEIVER_ID, &msg3);  // Returns immediately
    
    // All three messages now "in flight"
    // But we don't know if they've been received yet
    
    do_other_work();  // Can proceed without waiting
}
 
void receiver_process() {
    message_t msg;
    
    // May block if no messages buffered, or return immediately if available
    async_receive(ANY_SENDER, &msg);
    
    process_message(&msg);
}
 
// Timeline visualization (3 messages):
//
// Sender:   --[s1]-[s2]-[s3]--[other_work]---->
//               |    |    |
//               v    v    v   (messages in flight)
// Network:   ===m1===m2===m3==================>
//                              |    |    |
//                              v    v    v  
// Receiver: --------[r1]-[process]-[r2]-[process]-[r3]---->

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

// At-least-once delivery with retry logic
// Messages may be delivered multiple times
 
typedef struct {
    uint64_t message_id;  // Unique identifier for deduplication
    char payload[MAX_PAYLOAD];
} message_t;
 
int send_at_least_once(int socket, message_t *msg, int max_retries) {
    int retries = 0;
    
    while (retries < max_retries) {
        // Send the message
        if (send(socket, msg, sizeof(*msg), 0) < 0) {
            retries++;
            continue;
        }
        
        // Wait for acknowledgment with timeout
        struct timeval timeout = {.tv_sec = 1, .tv_usec = 0};
        setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
        
        ack_t ack;
        ssize_t received = recv(socket, &ack, sizeof(ack), 0);
        
        if (received > 0 && ack.message_id == msg->message_id) {
            return 0;  // Success - acknowledged
        }
        
        // Timeout or wrong ack - retry
        retries++;
        printf("Retry %d for message %lu\n", retries, msg->message_id);
    }
    
    return -1;  // Failed after max retries
}
 
// Receiver side - idempotent processing required!
void receive_at_least_once(int socket) {
    // Track seen message IDs to detect duplicates
    hashset_t seen_messages;
    
    message_t msg;
    while (recv(socket, &msg, sizeof(msg), 0) > 0) {
        // Check for duplicate
        if (hashset_contains(&seen_messages, msg.message_id)) {
            printf("Duplicate message %lu ignored\n", msg.message_id);
        } else {
            // First time seeing this message
            hashset_add(&seen_messages, msg.message_id);
            process_message(&msg);
        }
        
        // Always send acknowledgment (even for duplicates)
        ack_t ack = {.message_id = msg.message_id};
        send(socket, &ack, sizeof(ack), 0);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// FIFO ordering implementation using sequence numbers
// Ensures messages from same sender arrive in order sent
 
typedef struct {
    uint32_t sender_id;
    uint32_t sequence_num;  // Per-sender sequence number
    char payload[MAX_PAYLOAD];
} ordered_message_t;
 
// Sender maintains per-destination sequence counter
typedef struct {
    uint32_t next_sequence[MAX_DESTINATIONS];
} sender_state_t;
 
void fifo_send(sender_state_t *state, int dest_id, const char *payload) {
    ordered_message_t msg = {
        .sender_id = MY_ID,
        .sequence_num = state->next_sequence[dest_id]++,
    };
    strncpy(msg.payload, payload, MAX_PAYLOAD);
    
    network_send(dest_id, &msg, sizeof(msg));
}
 
// Receiver tracks expected sequence number from each sender
typedef struct {
    uint32_t expected_sequence[MAX_SENDERS];
    // Buffer for out-of-order messages
    ordered_message_t buffer[MAX_SENDERS][BUFFER_SIZE];
    int buffer_count[MAX_SENDERS];
} receiver_state_t;
 
void fifo_receive(receiver_state_t *state, ordered_message_t *msg) {
    uint32_t sender = msg->sender_id;
    
    if (msg->sequence_num == state->expected_sequence[sender]) {
        // In-order message - deliver immediately
        deliver_message(msg);
        state->expected_sequence[sender]++;
        
        // Check buffer for now-deliverable messages
        check_buffered_messages(state, sender);
    } else if (msg->sequence_num > state->expected_sequence[sender]) {
        // Out-of-order - buffer for later
        buffer_message(state, sender, msg);
    }
    // Ignore if sequence_num < expected (duplicate or old message)
}
 
void check_buffered_messages(receiver_state_t *state, uint32_t sender) {
    // Deliver any buffered messages that are now in sequence
    while (1) {
        ordered_message_t *buffered = find_in_buffer(
            state, sender, state->expected_sequence[sender]
        );
        
        if (!buffered) break;
        
        deliver_message(buffered);
        remove_from_buffer(state, sender, buffered);
        state->expected_sequence[sender]++;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

// UDP socket-based message passing
// Each sendto()/recvfrom() is a complete message
 
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
 
typedef struct {
    uint32_t type;
    uint32_t sequence;
    char payload[1024];
} app_message_t;
 
// Initialize UDP socket for message passing
int create_message_socket(int port) {
    int sock = socket(AF_INET, SOCK_DGRAM, 0);  // SOCK_DGRAM = UDP
    if (sock < 0) return -1;
    
    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port = htons(port),
        .sin_addr.s_addr = INADDR_ANY,
    };
    
    if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        close(sock);
        return -1;
    }
    
    return sock;
}
 
// Send a message to a specific destination
int send_message(int sock, const char *dest_ip, int dest_port, 
                 app_message_t *msg) {
    struct sockaddr_in dest = {
        .sin_family = AF_INET,
        .sin_port = htons(dest_port),
    };
    inet_pton(AF_INET, dest_ip, &dest.sin_addr);
    
    // sendto() transmits the complete message atomically
    // Either the whole message goes out, or none of it
    ssize_t sent = sendto(sock, msg, sizeof(*msg), 0,
                          (struct sockaddr*)&dest, sizeof(dest));
    
    return (sent == sizeof(*msg)) ? 0 : -1;
}
 
// Receive a message from any sender
int receive_message(int sock, app_message_t *msg, 
                    struct sockaddr_in *sender) {
    socklen_t sender_len = sizeof(*sender);
    
    // recvfrom() returns one complete message
    // Message boundaries are preserved in UDP
    ssize_t received = recvfrom(sock, msg, sizeof(*msg), 0,
                                (struct sockaddr*)sender, &sender_len);
    
    return (received == sizeof(*msg)) ? 0 : -1;
}
 
// Example usage
void example_sender() {
    int sock = create_message_socket(0);  // Ephemeral port
    
    app_message_t msg = {
        .type = MSG_TYPE_REQUEST,
        .sequence = 1,
    };
    strcpy(msg.payload, "Hello, distributed world!");
    
    send_message(sock, "192.168.1.100", 5000, &msg);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// POSIX Message Queue - kernel-managed local message passing
// Provides discrete messages with priority
 
#include <mqueue.h>
#include <fcntl.h>
 
#define QUEUE_NAME "/example_queue"
#define MAX_MSG_SIZE 1024
#define MAX_MSGS 10
 
// Create or open message queue
mqd_t create_message_queue() {
    struct mq_attr attr = {
        .mq_flags = 0,
        .mq_maxmsg = MAX_MSGS,      // Max messages in queue
        .mq_msgsize = MAX_MSG_SIZE, // Max message size
        .mq_curmsgs = 0,
    };
    
    mqd_t mq = mq_open(QUEUE_NAME, 
                       O_CREAT | O_RDWR, 
                       0644, 
                       &attr);
    
    return mq;
}
 
// Send message with priority (0-31, higher = higher priority)
int send_with_priority(mqd_t mq, const char *message, unsigned priority) {
    return mq_send(mq, message, strlen(message) + 1, priority);
}
 
// Receive highest-priority message
int receive_priority(mqd_t mq, char *buffer, size_t bufsize, 
                     unsigned *priority) {
    return mq_receive(mq, buffer, bufsize, priority);
}
 
// Example: multi-process communication
void producer_process() {
    mqd_t mq = create_message_queue();
    
    // Send messages with different priorities
    send_with_priority(mq, "Low priority task", 5);
    send_with_priority(mq, "Critical task", 31);    // Delivered first!
    send_with_priority(mq, "Normal task", 15);
    
    mq_close(mq);
}
 
void consumer_process() {
    mqd_t mq = mq_open(QUEUE_NAME, O_RDONLY);
    
    char buffer[MAX_MSG_SIZE];
    unsigned priority;
    
    while (mq_receive(mq, buffer, sizeof(buffer), &priority) != -1) {
        printf("Received (priority %u): %s\n", priority, buffer);
    }
    
    mq_close(mq);
    mq_unlink(QUEUE_NAME);  // Remove queue
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# Resilient message sending with common patterns
 
import time
import random
from functools import wraps
 
class CircuitBreaker:
    """Prevents overwhelming a failing service"""
    
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = 0
        self.state = "closed"  # closed = normal, open = blocking
    
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
    
    def record_success(self):
        self.failure_count = 0
        self.state = "closed"
    
    def can_proceed(self):
        if self.state == "closed":
            return True
        # Check if reset timeout has elapsed
        if time.time() - self.last_failure_time > self.reset_timeout:
            return True  # Allow one attempt (half-open state)
        return False
 
 
def retry_with_backoff(max_retries=5, base_delay=1.0, max_delay=60.0):
    """Decorator for retry with exponential backoff"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    
                    if attempt < max_retries - 1:
                        # Exponential backoff with jitter
                        delay = min(base_delay * (2 ** attempt), max_delay)
                        jitter = random.uniform(0, delay * 0.1)
                        time.sleep(delay + jitter)
            
            raise last_exception
        
        return wrapper
    return decorator
 
 
class ResilientMessageSender:
    """Message sender with retry, circuit breaker, and idempotency"""
    
    def __init__(self, connection):
        self.connection = connection
        self.circuit_breakers = {}  # Per-destination circuit breakers
    
    def get_circuit_breaker(self, destination):
        if destination not in self.circuit_breakers:
            self.circuit_breakers[destination] = CircuitBreaker()
        return self.circuit_breakers[destination]
    
    @retry_with_backoff(max_retries=5)
    def send_message(self, destination, message, idempotency_key=None):
        """Send with retry, circuit breaker, and idempotency support"""
        
        cb = self.get_circuit_breaker(destination)
        
        if not cb.can_proceed():
            raise CircuitOpenError(f"Circuit open for {destination}")
        
        try:
            # Include idempotency key for receiver-side deduplication
            envelope = {
                "idempotency_key": idempotency_key or generate_uuid(),
                "timestamp": time.time(),
                "payload": message,
            }
            
            result = self.connection.send(destination, envelope)
            cb.record_success()
            return result
            
        except Exception as e:
            cb.record_failure()
            raise

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

# Reliable multicast with acknowledgment tracking
 
from threading import Thread, Lock
import time
 
class ReliableMulticast:
    """
    Implements reliable multicast using sender-based acknowledgment tracking.
    Guarantees: if any correct recipient receives the message, all correct
    recipients eventually receive it.
    """
    
    def __init__(self, group_members, send_func):
        self.members = group_members
        self.send = send_func
        self.pending_acks = {}  # message_id -> set of members not yet acked
        self.lock = Lock()
        self.max_retries = 5
        self.retry_interval = 1.0  # seconds
    
    def multicast(self, message):
        """Send message to all group members reliably"""
        message_id = generate_unique_id()
        
        with self.lock:
            # Track which members need to acknowledge
            self.pending_acks[message_id] = set(self.members)
        
        # Send to all members in parallel
        for member in self.members:
            self._send_to_member(member, message_id, message)
        
        # Start background thread to handle retransmissions
        retransmitter = Thread(
            target=self._retransmission_loop,
            args=(message_id, message)
        )
        retransmitter.daemon = True
        retransmitter.start()
        
        return message_id
    
    def _send_to_member(self, member, message_id, message):
        """Send message to a single member"""
        envelope = {
            'message_id': message_id,
            'payload': message,
            'requires_ack': True,
        }
        self.send(member, envelope)
    
    def receive_ack(self, member, message_id):
        """Process acknowledgment from a member"""
        with self.lock:
            if message_id in self.pending_acks:
                self.pending_acks[message_id].discard(member)
                
                # All members acknowledged?
                if not self.pending_acks[message_id]:
                    del self.pending_acks[message_id]
                    return True  # Fully acknowledged
        return False
    
    def _retransmission_loop(self, message_id, message):
        """Retransmit to members who haven't acknowledged"""
        retries = 0
        
        while retries < self.max_retries:
            time.sleep(self.retry_interval)
            
            with self.lock:
                if message_id not in self.pending_acks:
                    return  # All acknowledged
                
                pending = list(self.pending_acks[message_id])
            
            # Retransmit to non-acknowledging members
            for member in pending:
                self._send_to_member(member, message_id, message)
            
            retries += 1
        
        # Max retries exceeded - log/alert about unreachable members
        with self.lock:
            if message_id in self.pending_acks:
                failed = self.pending_acks[message_id]
                log_delivery_failure(message_id, list(failed))