Achieving concurrency within a single thread presents a method to improve efficiency by eliminating the overhead associated with creating and managing processes or threads. The core principle involves switching between tasks and preserving their execution state. In standard multi-threading, the operating system controls task switching, primarily when a task blocks on I/O or after prolonged execution. While switching due to long computation doesn't enhance performance, switching during I/O operations allows the CPU to utilize otherwise idle time.
Pure computational task switching using yield demonstrates this inefficiency.
# Sequential execution for performance baseline
import time
def data_processor(result_data):
pass # Simulate data processing
def data_generator():
data_collection = []
for num in range(10000000):
data_collection.append(num)
return data_collection
start_time = time.time()
result = data_generator()
data_processor(result)
end_time = time.time()
print(f'Sequential duration: {end_time - start_time}')
# Concurrent execution simulation using yield
import time
def process_task():
while True:
received_value = yield
def generate_task():
task = process_task()
next(task)
for num in range(10000000):
task.send(num)
start_time = time.time()
generate_task()
end_time = time.time()
print(f'Yield-based duration: {end_time - start_time}')
The yield statement cannot automatically detect and switch on I/O operations.
import time
def process_task():
while True:
val = yield
def generate_task():
task = process_task()
next(task)
for i in range(5):
task.send(i)
time.sleep(2) # I/O simulation, but yield does not switch tasks here
start_time = time.time()
generate_task()
end_time = time.time()
print(f'Duration with blocking I/O: {end_time - start_time}')
Coroutines address this by enabling user-level control over task switching, particular during I/O waits, making a single thread appear highly active to the operating system. An effective coroutine implementation requires:
- Controlled switching between tasks with state preservation.
- Automatic detection of I/O operations to trigger switches.
Coroutines, or micro-threads, are user-space lightweight threads managed by the application, not the OS kernel.
Advantages include reduced switching overhead and efficient CPU use within a single thread. Limitations are the inability to leverage multiple CPU cores directly and potential thread blockage if a coroutine blocks. Key characteristics are single-threaded concurrency, no need for locks on shared data, user-managed context stacks, and automatic I/O-based switching.
The greenlet module offers explicit, manual switching between tasks.
from greenlet import greenlet
def task_one(name):
print(f'{name} executes task_one part 1')
gr_two.switch('Alex')
print(f'{name} executes task_one part 2')
gr_two.switch()
def task_two(name):
print(f'{name} executes task_two part 1')
gr_one.switch()
print(f'{name} executes task_two part 2')
gr_one = greenlet(task_one)
gr_two = greenlet(task_two)
gr_one.switch('Alex')
Manual switching for pure computation reduces performance.
# Sequential execution
import time
def calculate_sum():
total = 1
for i in range(100000000):
total += i
def calculate_product():
product = 1
for i in range(100000000):
product *= i
start = time.time()
calculate_sum()
calculate_product()
end = time.time()
print(f'Sequential time: {end - start}')
# Switching execution with greenlet
from greenlet import greenlet
import time
def calc_sum():
total = 1
for i in range(100000000):
total += i
gr2.switch()
def calc_prod():
product = 1
for i in range(100000000):
product *= i
gr1.switch()
start = time.time()
gr1 = greenlet(calc_sum)
gr2 = greenlet(calc_prod)
gr1.switch()
end = time.time()
print(f'Greenlet switching time: {end - start}')
greenlet does not solve automatic I/O switching. The gevent library provides this by patching standard I/O functions.
import gevent
def activity_one(name):
print(f'{name} starts activity_one')
gevent.sleep(2) # Gevent-recognized I/O
print(f'{name} finishes activity_one')
def activity_two(name):
print(f'{name} starts activity_two')
gevent.sleep(1)
print(f'{name} finishes activity_two')
coroutine1 = gevent.spawn(activity_one, 'Task1')
coroutine2 = gevent.spawn(activity_two, 'Task2')
gevent.joinall([coroutine1, coroutine2])
print('Main thread continues')
To make gevent recognize standard libray I/O like time.sleep(), apply a monkey patch at the start.
from gevent import monkey
monkey.patch_all() # Must be called before other imports
import gevent
import time
def job_one():
print('Job one phase 1')
time.sleep(2)
print('Job one phase 2')
def job_two():
print('Job two phase 1')
time.sleep(1)
print('Job two phase 2')
g1 = gevent.spawn(job_one)
g2 = gevent.spawn(job_two)
gevent.joinall([g1, g2])
gevent enables asynchronous execution patterns.
from gevent import spawn, joinall, monkey
monkey.patch_all()
import time
def work(task_id):
time.sleep(0.5)
print(f'Completed task {task_id}')
def run_sync():
for i in range(10):
work(i)
def run_async():
coroutines = [spawn(work, i) for i in range(10)]
joinall(coroutines)
print('All async tasks done')
print('Synchronous execution:')
run_sync()
print('\nAsynchronous execution:')
run_async()
A practical use case is concurrent web requests.
from gevent import monkey; monkey.patch_all()
import gevent
import requests
import time
def fetch_url(url):
print(f'Fetching: {url}')
resp = requests.get(url)
if resp.status_code == 200:
print(f'Received {len(resp.text)} bytes from {url}')
start = time.time()
gevent.joinall([
gevent.spawn(fetch_url, 'https://www.python.org/'),
gevent.spawn(fetch_url, 'https://www.github.com/'),
gevent.spawn(fetch_url, 'https://www.example.com/'),
])
end = time.time()
print(f'Total fetch time: {end - start}')
gevent can also manage concurrent socket connections in a single thread.
# Server
from gevent import monkey; monkey.patch_all()
from socket import *
import gevent
def start_server(host, port):
server_sock = socket(AF_INET, SOCK_STREAM)
server_sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
server_sock.bind((host, port))
server_sock.listen(5)
while True:
client_conn, client_addr = server_sock.accept()
gevent.spawn(handle_client, client_conn, client_addr)
def handle_client(connection, address):
try:
while True:
data = connection.recv(1024)
if not data:
break
print(f'Msg from {address}: {data.decode()}')
connection.send(data.upper())
except Exception as e:
print(f'Error: {e}')
finally:
connection.close()
if __name__ == '__main__':
start_server('127.0.0.1', 8080)
# Client
from socket import *
client_socket = socket(AF_INET, SOCK_STREAM)
client_socket.connect(('127.0.0.1', 8080))
while True:
message = input('Enter message: ').strip()
if message:
client_socket.send(message.encode('utf-8'))
response = client_socket.recv(1024)
print(f'Server reply: {response.decode("utf-8")}')