Networking Basics

Building Connected Game Experiences

Understanding Game Networking

Interactive Network Simulator

Networking Implementation in Python

import socket
import threading
import json
import time
import struct
from typing import Dict, List, Optional, Any, Callable, Set
from dataclasses import dataclass, asdict
from enum import Enum
import queue

class PacketType(Enum):
    CONNECT = "connect"
    DISCONNECT = "disconnect"
    STATE = "state"
    INPUT = "input"
    EVENT = "event"
    PING = "ping"
    PONG = "pong"
    ACK = "ack"

@dataclass
class Packet:
    """Network packet structure"""
    type: PacketType
    sequence: int
    timestamp: float
    data: Dict[str, Any]
    reliable: bool = False
    
    def serialize(self) -> bytes:
        """Serialize packet to bytes"""
        json_data = json.dumps({
            'type': self.type.value,
            'seq': self.sequence,
            'time': self.timestamp,
            'data': self.data,
            'rel': self.reliable
        })
        
        # Add header with packet size
        data = json_data.encode('utf-8')
        header = struct.pack('!I', len(data))
        return header + data
    
    @staticmethod
    def deserialize(data: bytes) -> 'Packet':
        """Deserialize bytes to packet"""
        json_data = json.loads(data.decode('utf-8'))
        return Packet(
            type=PacketType(json_data['type']),
            sequence=json_data['seq'],
            timestamp=json_data['time'],
            data=json_data['data'],
            reliable=json_data.get('rel', False)
        )

class NetworkManager:
    """Base network manager for client and server"""
    
    def __init__(self, host: str = 'localhost', port: int = 5555) -> None:
        self.host: str = host
        self.port: int = port
        self.socket: Optional[socket.socket] = None
        self.running: bool = False
        
        self.sequence_number: int = 0
        self.pending_acks: Dict[int, Packet] = {}
        self.received_sequences: Set[int] = set()
        
        self.send_queue: queue.Queue = queue.Queue()
        self.receive_queue: queue.Queue = queue.Queue()
        
        self.latency: float = 0
        self.packet_loss: float = 0
        
        self.callbacks: Dict[PacketType, List[Callable]] = {
            packet_type: [] for packet_type in PacketType
        }
    
    def register_callback(self, packet_type: PacketType, callback: Callable) -> None:
        """Register callback for packet type"""
        self.callbacks[packet_type].append(callback)
    
    def send_packet(self, packet_type: PacketType, data: Dict[str, Any], 
                   reliable: bool = False) -> None:
        """Queue packet for sending"""
        packet = Packet(
            type=packet_type,
            sequence=self.sequence_number,
            timestamp=time.time(),
            data=data,
            reliable=reliable
        )
        
        self.sequence_number += 1
        
        if reliable:
            self.pending_acks[packet.sequence] = packet
        
        self.send_queue.put(packet)
    
    def process_packet(self, packet: Packet) -> None:
        """Process received packet"""
        # Check for duplicates
        if packet.sequence in self.received_sequences:
            return
        
        self.received_sequences.add(packet.sequence)
        
        # Send ACK for reliable packets
        if packet.reliable:
            self.send_packet(PacketType.ACK, {'ack_seq': packet.sequence})
        
        # Handle ACK
        if packet.type == PacketType.ACK:
            ack_seq = packet.data.get('ack_seq')
            if ack_seq in self.pending_acks:
                del self.pending_acks[ack_seq]
            return
        
        # Call registered callbacks
        for callback in self.callbacks[packet.type]:
            callback(packet)
    
    def calculate_latency(self, ping_time: float) -> float:
        """Calculate round-trip latency"""
        return (time.time() - ping_time) * 1000  # Convert to ms

class GameServer(NetworkManager):
    """Game server implementation"""
    
    def __init__(self, host: str = '0.0.0.0', port: int = 5555) -> None:
        super().__init__(host, port)
        self.clients: Dict[str, Dict[str, Any]] = {}
        self.game_state: Dict[str, Dict[str, Any]] = {}
        self.tick_rate: int = 60
        
    def start(self) -> None:
        """Start server"""
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind((self.host, self.port))
        self.socket.listen(5)
        self.running = True
        
        print(f"Server listening on {self.host}:{self.port}")
        
        # Start threads
        threading.Thread(target=self.accept_clients, daemon=True).start()
        threading.Thread(target=self.game_loop, daemon=True).start()
    
    def accept_clients(self) -> None:
        """Accept new client connections"""
        while self.running:
            try:
                client_socket, address = self.socket.accept()
                client_id = f"{address[0]}:{address[1]}"
                
                self.clients[client_id] = {
                    'socket': client_socket,
                    'address': address,
                    'state': {},
                    'last_input': None,
                    'ping': 0
                }
                
                print(f"Client connected: {client_id}")
                
                # Start client handler
                threading.Thread(
                    target=self.handle_client,
                    args=(client_id,),
                    daemon=True
                ).start()
                
            except Exception as e:
                print(f"Accept error: {e}")
    
    def handle_client(self, client_id: str) -> None:
        """Handle client communication"""
        client = self.clients[client_id]
        buffer = b''
        
        while self.running and client_id in self.clients:
            try:
                # Receive data
                data = client['socket'].recv(4096)
                if not data:
                    break
                
                buffer += data
                
                # Process complete packets
                while len(buffer) >= 4:
                    # Read packet size
                    size = struct.unpack('!I', buffer[:4])[0]
                    
                    if len(buffer) < 4 + size:
                        break
                    
                    # Extract packet
                    packet_data = buffer[4:4+size]
                    buffer = buffer[4+size:]
                    
                    # Process packet
                    packet = Packet.deserialize(packet_data)
                    self.process_client_packet(client_id, packet)
                    
            except Exception as e:
                print(f"Client error {client_id}: {e}")
                break
        
        # Client disconnected
        self.disconnect_client(client_id)
    
    def process_client_packet(self, client_id: str, packet: Packet) -> None:
        """Process packet from client"""
        client = self.clients.get(client_id)
        if not client:
            return
        
        if packet.type == PacketType.INPUT:
            client['last_input'] = packet.data
            
        elif packet.type == PacketType.PING:
            # Send pong
            self.send_to_client(client_id, PacketType.PONG, 
                              {'ping_time': packet.data['time']})
            
        elif packet.type == PacketType.STATE:
            client['state'].update(packet.data)
        
        # Process with base class
        self.process_packet(packet)
    
    def send_to_client(self, client_id: str, packet_type: PacketType, 
                      data: Dict[str, Any], reliable: bool = False) -> None:
        """Send packet to specific client"""
        client = self.clients.get(client_id)
        if not client:
            return
        
        packet = Packet(
            type=packet_type,
            sequence=self.sequence_number,
            timestamp=time.time(),
            data=data,
            reliable=reliable
        )
        
        self.sequence_number += 1
        
        try:
            client['socket'].send(packet.serialize())
        except:
            self.disconnect_client(client_id)
    
    def broadcast(self, packet_type: PacketType, data: Dict[str, Any], 
                 exclude: Optional[str] = None) -> None:
        """Broadcast packet to all clients"""
        for client_id in list(self.clients.keys()):
            if client_id != exclude:
                self.send_to_client(client_id, packet_type, data)
    
    def disconnect_client(self, client_id: str) -> None:
        """Disconnect and remove client"""
        if client_id in self.clients:
            try:
                self.clients[client_id]['socket'].close()
            except:
                pass
            
            del self.clients[client_id]
            print(f"Client disconnected: {client_id}")
            
            # Notify other clients
            self.broadcast(PacketType.EVENT, {
                'event': 'player_disconnected',
                'player_id': client_id
            })
    
    def game_loop(self) -> None:
        """Main game loop"""
        tick_interval = 1.0 / self.tick_rate
        last_tick = time.time()
        
        while self.running:
            current_time = time.time()
            delta_time = current_time - last_tick
            
            if delta_time >= tick_interval:
                self.update_game_state(delta_time)
                self.send_state_updates()
                last_tick = current_time
            
            time.sleep(0.001)
    
    def update_game_state(self, delta_time: float) -> None:
        """Update game state"""
        # Process client inputs
        for client_id, client in self.clients.items():
            if client['last_input']:
                # Apply input to game state
                self.apply_input(client_id, client['last_input'], delta_time)
                client['last_input'] = None
    
    def apply_input(self, client_id: str, input_data: Dict[str, Any], 
                    delta_time: float) -> None:
        """Apply client input to game state"""
        # Example: Move player based on input
        if client_id not in self.game_state:
            self.game_state[client_id] = {
                'position': {'x': 0, 'y': 0},
                'velocity': {'x': 0, 'y': 0}
            }
        
        player = self.game_state[client_id]
        
        # Apply movement
        if 'move' in input_data:
            player['velocity']['x'] = input_data['move'].get('x', 0) * 100
            player['velocity']['y'] = input_data['move'].get('y', 0) * 100
        
        # Update position
        player['position']['x'] += player['velocity']['x'] * delta_time
        player['position']['y'] += player['velocity']['y'] * delta_time
    
    def send_state_updates(self) -> None:
        """Send game state to all clients"""
        state_data = {
            'timestamp': time.time(),
            'players': self.game_state
        }
        
        self.broadcast(PacketType.STATE, state_data)

class GameClient(NetworkManager):
    """Game client implementation"""
    
    def __init__(self) -> None:
        super().__init__()
        self.connected: bool = False
        self.local_state: Dict[str, Any] = {}
        self.server_state: Dict[str, Any] = {}
        self.input_buffer: List[Dict[str, Any]] = []
        
    def connect(self, host: str, port: int) -> bool:
        """Connect to server"""
        self.host = host
        self.port = port
        
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.connect((host, port))
            self.connected = True
            self.running = True
            
            print(f"Connected to server {host}:{port}")
            
            # Start receive thread
            threading.Thread(target=self.receive_loop, daemon=True).start()
            
            # Send connect packet
            self.send_packet(PacketType.CONNECT, {'name': 'Player'})
            
            return True
            
        except Exception as e:
            print(f"Connection failed: {e}")
            return False
    
    def receive_loop(self) -> None:
        """Receive packets from server"""
        buffer = b''
        
        while self.running and self.connected:
            try:
                data = self.socket.recv(4096)
                if not data:
                    break
                
                buffer += data
                
                # Process complete packets
                while len(buffer) >= 4:
                    size = struct.unpack('!I', buffer[:4])[0]
                    
                    if len(buffer) < 4 + size:
                        break
                    
                    packet_data = buffer[4:4+size]
                    buffer = buffer[4+size:]
                    
                    packet = Packet.deserialize(packet_data)
                    self.receive_queue.put(packet)
                    
            except Exception as e:
                print(f"Receive error: {e}")
                break
        
        self.disconnect()
    
    def send_input(self, input_data: Dict[str, Any]) -> None:
        """Send input to server"""
        self.input_buffer.append(input_data)
        self.send_packet(PacketType.INPUT, input_data)
    
    def update(self, delta_time: float) -> None:
        """Update client"""
        # Process received packets
        while not self.receive_queue.empty():
            packet = self.receive_queue.get()
            self.process_packet(packet)
            
            if packet.type == PacketType.STATE:
                self.server_state = packet.data
                self.reconcile_state()
        
        # Send queued packets
        while not self.send_queue.empty():
            packet = self.send_queue.get()
            try:
                self.socket.send(packet.serialize())
            except:
                self.disconnect()
    
    def reconcile_state(self) -> None:
        """Reconcile local and server state"""
        # Client-side prediction reconciliation
        server_time = self.server_state.get('timestamp', 0)
        
        # Remove old inputs
        self.input_buffer = [
            inp for inp in self.input_buffer 
            if inp.get('timestamp', 0) > server_time
        ]
        
        # Re-apply unacknowledged inputs
        for input_data in self.input_buffer:
            self.apply_local_input(input_data)
    
    def apply_local_input(self, input_data: Dict[str, Any]) -> None:
        """Apply input locally for prediction"""
        # Update local state immediately
        if 'move' in input_data:
            if 'position' not in self.local_state:
                self.local_state['position'] = {'x': 0, 'y': 0}
            
            self.local_state['position']['x'] += input_data['move'].get('x', 0)
            self.local_state['position']['y'] += input_data['move'].get('y', 0)
    
    def disconnect(self) -> None:
        """Disconnect from server"""
        if self.connected:
            self.send_packet(PacketType.DISCONNECT, {})
            self.connected = False
            self.running = False
            
            try:
                self.socket.close()
            except:
                pass
            
            print("Disconnected from server")

Best Practices

Key Takeaways

import math
import pygame

pygame.init()
W, H = 800, 480
screen = pygame.display.set_mode((W, H))
pygame.display.set_caption('Connection Scaling - CS vs Full Mesh')
font = pygame.font.SysFont('arial', 16)
small = pygame.font.SysFont('arial', 12)
clock = pygame.time.Clock()

n = 4  # player count, range [2, 12]

def cs_layout(n):
    server = (200, 80)
    clients = []
    for i in range(n):
        a = (math.pi * 2 / n) * i - math.pi / 2
        clients.append((200 + 110 * math.cos(a), 280 + 90 * math.sin(a)))
    return server, clients

def mesh_layout(n):
    cx, cy = 600, 240
    return [(cx + 110 * math.cos((math.pi * 2 / n) * i - math.pi / 2),
             cy + 110 * math.sin((math.pi * 2 / n) * i - math.pi / 2))
            for i in range(n)]

packets_cs = []   # [client_idx, prog]
packets_mesh = [] # [i, j, prog]
last_burst = -1.5
t = 0.0
running = True

while running:
    dt = clock.tick(60) / 1000.0
    t += dt
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False
        elif event.type == pygame.KEYDOWN:
            if event.key in (pygame.K_MINUS, pygame.K_KP_MINUS):
                n = max(2, n - 1); packets_cs.clear(); packets_mesh.clear()
            elif event.key in (pygame.K_EQUALS, pygame.K_PLUS, pygame.K_KP_PLUS):
                n = min(12, n + 1); packets_cs.clear(); packets_mesh.clear()
            elif event.key == pygame.K_ESCAPE:
                running = False

    if t - last_burst >= 1.5:
        last_burst = t
        for i in range(n):
            packets_cs.append([i, 0.0])
            for j in range(i + 1, n):
                packets_mesh.append([i, j, 0.0])

    for p in packets_cs: p[1] += dt
    for p in packets_mesh: p[2] += dt
    packets_cs[:] = [p for p in packets_cs if p[1] < 1.0]
    packets_mesh[:] = [p for p in packets_mesh if p[2] < 1.0]

    server, clients = cs_layout(n)
    peers = mesh_layout(n)
    cs_count = n
    mesh_count = n * (n - 1) // 2
    ratio = (n - 1) / 2.0

    screen.fill((15, 15, 25))
    screen.blit(font.render('Client-Server  O(N)', True, (200, 220, 255)), (110, 30))
    screen.blit(font.render('Full Mesh  O(N**2)', True, (255, 200, 220)), (510, 30))

    for c in clients:
        pygame.draw.line(screen, (80, 160, 240), server, c, 2)
    pygame.draw.circle(screen, (240, 80, 80), server, 14)
    screen.blit(small.render('SERVER', True, (255, 255, 255)),
                (server[0] - 22, server[1] + 18))
    for c in clients:
        pygame.draw.circle(screen, (80, 160, 240), (int(c[0]), int(c[1])), 9)

    for idx, prog in packets_cs:
        if idx < len(clients):
            c = clients[idx]
            x = c[0] + (server[0] - c[0]) * prog
            y = c[1] + (server[1] - c[1]) * prog
            pygame.draw.circle(screen, (255, 255, 100), (int(x), int(y)), 4)

    for i in range(n):
        for j in range(i + 1, n):
            pygame.draw.line(screen, (220, 100, 200), peers[i], peers[j], 1)
    for p in peers:
        pygame.draw.circle(screen, (220, 100, 200), (int(p[0]), int(p[1])), 9)

    for i, j, prog in packets_mesh:
        if i < len(peers) and j < len(peers):
            x = peers[i][0] + (peers[j][0] - peers[i][0]) * prog
            y = peers[i][1] + (peers[j][1] - peers[i][1]) * prog
            pygame.draw.circle(screen, (255, 255, 100), (int(x), int(y)), 4)

    hud = [
        f'N = {n}    (- / + to change, range [2, 12])',
        f'Client-Server connections: {cs_count}    LINEAR  O(N)',
        f'Full-Mesh connections:     {mesh_count}    QUADRATIC  O(N**2)',
        f'Ratio mesh/CS = (N-1)/2 = {ratio:.1f}x',
        f'In-flight packets:  CS={len(packets_cs)}    Mesh={len(packets_mesh)}',
    ]
    for i, line in enumerate(hud):
        screen.blit(font.render(line, True, (220, 220, 220)),
                    (20, H - 105 + i * 18))

    pygame.display.flip()
pygame.quit()

What's Next?

Now that you understand networking basics, next we'll implement a complete multiplayer game with real-time synchronization!