Practical Code Examples
TCP Client with Timeout
import socket
def tcp_client_with_timeout(host, port, timeout=5):
try:
## Create socket with timeout
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.settimeout(timeout)
## Attempt connection
client_socket.connect((host, port))
client_socket.send(b'Hello Server')
## Receive data with timeout
response = client_socket.recv(1024)
print(f"Server response: {response.decode()}")
except socket.timeout:
print("Connection timed out")
except ConnectionRefusedError:
print("Connection refused")
finally:
client_socket.close()
UDP Server with Timeout
import socket
def udp_server_with_timeout(host='localhost', port=12345, timeout=10):
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind((host, port))
server_socket.settimeout(timeout)
try:
while True:
try:
data, client_address = server_socket.recvfrom(1024)
print(f"Received: {data.decode()} from {client_address}")
except socket.timeout:
print("No data received within timeout")
except KeyboardInterrupt:
print("Server stopped")
finally:
server_socket.close()
Timeout Scenarios Comparison
Scenario |
Timeout Strategy |
Recommended Approach |
Network Request |
Short Timeout |
3-5 seconds |
Large Data Transfer |
Longer Timeout |
30-60 seconds |
Critical Services |
Configurable Timeout |
Dynamic adjustment |
Asynchronous Timeout Handling
graph TD
A[Network Operation] --> B{Timeout Configured}
B -->|Yes| C[Start Async Timer]
C --> D{Operation Complete}
D -->|No| E{Timeout Reached}
E -->|Yes| F[Cancel Operation]
E -->|No| D
D -->|Yes| G[Process Result]
Advanced Timeout Example: Multi-Service Monitoring
import socket
import concurrent.futures
def check_service(host, port, timeout=5):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
return f"{host}:{port} - {'Open' if result == 0 else 'Closed'}"
except socket.timeout:
return f"{host}:{port} - Timeout"
finally:
sock.close()
def multi_service_check(services):
with concurrent.futures.ThreadPoolExecutor() as executor:
results = list(executor.map(lambda s: check_service(*s), services))
return results
LabEx Learning Tip
LabEx recommends practicing these timeout techniques in simulated network environments to build robust networking skills.