- gevent is a concurrency library based around libev. It provides a clean API for a variety of concurrency and network related tasks. + gevent is a concurrency library based around libev or + his new powerful descendant libuv. + It provides a clean API for a variety of concurrency and network related tasks.diff --git a/build b/build index b593022..9349798 100755 --- a/build +++ b/build @@ -1,14 +1,13 @@ #!/usr/bin/env python -import re +import codecs import os +import re import sys -import codecs +from io import StringIO + import markdown -from cStringIO import StringIO from cogapp import Cog -from jinja2 import Template - from jinja2 import Environment, FileSystemLoader OUTPUT = 'index.html' @@ -35,8 +34,8 @@ start_code = """
"""
end_code = """- gevent is a concurrency library based around libev. It provides a clean API for a variety of concurrency and network related tasks. + gevent is a concurrency library based around libev or + his new powerful descendant libuv. + It provides a clean API for a variety of concurrency and network related tasks.@@ -78,6 +80,7 @@
In chronological order of contribution: -Stephen Diehl -Jérémy Bethmont -sww -Bruno Bigras -David Ripton -Travis Cline -Boris Feld -youngsterxyf -Eddie Hebert -Alexis Metaireau -Daniel Velkov -Sean Wang -Inada Naoki -Balthazar Rouberol -Glen Baker -Jan-Philip Gehrcke -Matthijs van der Vleuten -Simon Hayward -Alexander James Phillips -Ramiro Morales -Philip Damra -Francisco José Marques Vieira -David Xia -satoru -James Summerfield -Adam Szkoda -Roy Smith -Jianbin Wei -Anton Larkin -Matias Herranz -Pietro Bertera
+Stephen Diehl, +Jérémy Bethmont, +sww, +Bruno Bigras, +David Ripton, +Travis Cline, +Boris Feld, +youngsterxyf, +Eddie Hebert, +Alexis Metaireau, +Daniel Velkov, +Sean Wang, +Inada Naoki, +Balthazar Rouberol, +Glen Baker, +Jan-Philip Gehrcke, +Matthijs van der Vleuten, +Simon Hayward, +Alexander James Phillips, +Ramiro Morales, +Philip Damra, +Francisco José Marques Vieira, +David Xia, +satoru, +James Summerfield, +Adam Szkoda, +Roy Smith, +Jianbin Wei, +Anton Larkin, +Matias Herranz, +Pietro Bertera, +Kevin TewoudaAlso thanks to Denis Bilenko for writing gevent and guidance in constructing this tutorial.
This is a collaborative document published under MIT license. @@ -244,11 +248,11 @@
In the synchronous case all the tasks are run sequentially, which results in the main programming blocking ( @@ -308,12 +312,13 @@
import gevent.monkey
gevent.monkey.patch_socket()
+import json
+from urllib.request import urlopen
+
import gevent
-import urllib2
-import simplejson as json
def fetch(pid):
- response = urllib2.urlopen('http://jsontime.herokuapp.com/')
+ response = urlopen('http://jsontime.herokuapp.com/')
result = response.read()
json_result = json.loads(result)
datetime = json_result['datetime']
@@ -347,34 +352,32 @@ Determinism
import time
+from multiprocessing.pool import Pool
+
+from gevent.pool import Pool as GPool
def echo(i):
time.sleep(0.001)
return i
-# Non Deterministic Process Pool
-
-from multiprocessing.pool import Pool
-
-p = Pool(10)
-run1 = [a for a in p.imap_unordered(echo, xrange(10))]
-run2 = [a for a in p.imap_unordered(echo, xrange(10))]
-run3 = [a for a in p.imap_unordered(echo, xrange(10))]
-run4 = [a for a in p.imap_unordered(echo, xrange(10))]
-
-print(run1 == run2 == run3 == run4)
-
-# Deterministic Gevent Pool
-
-from gevent.pool import Pool
-
-p = Pool(10)
-run1 = [a for a in p.imap_unordered(echo, xrange(10))]
-run2 = [a for a in p.imap_unordered(echo, xrange(10))]
-run3 = [a for a in p.imap_unordered(echo, xrange(10))]
-run4 = [a for a in p.imap_unordered(echo, xrange(10))]
-
-print(run1 == run2 == run3 == run4)
+if __name__ == '__main__':
+ # Non Deterministic Process Pool
+ with Pool(processes=10) as pool:
+ run1 = [a for a in pool.imap_unordered(echo, range(10))]
+ run2 = [a for a in pool.imap_unordered(echo, range(10))]
+ run3 = [a for a in pool.imap_unordered(echo, range(10))]
+ run4 = [a for a in pool.imap_unordered(echo, range(10))]
+
+ print(run1 == run2 == run3 == run4)
+
+ # Deterministic Gevent Pool
+ p = GPool(10)
+ run1 = [a for a in p.imap_unordered(echo, range(10))]
+ run2 = [a for a in p.imap_unordered(echo, range(10))]
+ run3 = [a for a in p.imap_unordered(echo, range(10))]
+ run4 = [a for a in p.imap_unordered(echo, range(10))]
+
+ print(run1 == run2 == run3 == run4)
@@ -532,10 +535,10 @@ Greenlet State
Greenlets that fail to yield when the main program receives a -SIGQUIT may hold the program's execution longer than expected. +SIGINT may hold the program's execution longer than expected. This results in so called "zombie processes" which need to be killed from outside of the Python interpreter.
-A common pattern is to listen SIGQUIT events on the main program +
A common pattern is to listen SIGINT events on the main program
and to invoke gevent.kill or gevent.killall before exit.
import gevent
@@ -546,7 +549,7 @@ Program Shutdown
if __name__ == '__main__':
thread = gevent.spawn(run_forever)
- gevent.signal(signal.SIGQUIT, gevent.kill, thread)
+ gevent.signal_handler(signal.SIGINT, gevent.kill, thread)
thread.join()
@@ -721,7 +724,8 @@
Size of group 3
-Hello from Greenlet 4405439216
+Hello from Greenlet 71908208
Size of group 3
-Hello from Greenlet 4405439056
+Hello from Greenlet 71908352
Size of group 3
-Hello from Greenlet 4405440336
+Hello from Greenlet 71908496
Ordered
('task', 0)
('task', 1)
@@ -1024,7 +1028,7 @@ Groups and Pools
def hello_from(n):
print('Size of pool %s' % len(pool))
-pool.map(hello_from, xrange(3))
+pool.map(hello_from, range(3))
@@ -1090,8 +1094,8 @@
@@ -1186,7 +1190,7 @@
+
Flask's system is a bit more sophisticated than this example, but the @@ -1211,6 +1215,7 @@
@@ -1221,7 +1226,7 @@Subprocess
cron cron Linux -+
Many people also want to use gevent and multiprocessing together. One of
@@ -1244,17 +1249,17 @@
import gevent
+from gevent import Greenlet
from gevent.queue import Queue
-class Actor(gevent.Greenlet):
+class Actor(Greenlet):
def __init__(self):
self.inbox = Queue()
@@ -1320,14 +1326,7 @@ Actors
message = self.inbox.get()
self.receive(message)
-
-
-
-In a use case:
-
-import gevent
-from gevent.queue import Queue
-from gevent import Greenlet
+# use case
class Pinger(Actor):
def receive(self, message):
@@ -1348,7 +1347,7 @@ Actors
pong.start()
ping.inbox.put('start')
-gevent.joinall([ping, pong])
+gevent.joinall([ping, pong], timeout=2)
@@ -1378,7 +1377,7 @@ But regardless, performance on Gevent servers is phenomenal -compared to other Python servers. libev is a very vetted technology +compared to other Python servers. libev/libuv is a very vetted technology and its derivative servers are known to perform well at scale.
-To benchmark, try Apache Benchmark ab or see this
-Benchmark of Python WSGI Servers
-for comparison with other servers.
To benchmark, try Apache Benchmark ab or locust for comparison with other servers.
$ ab -n 10000 -c 100 http://127.0.0.1:8000/@@ -1530,9 +1527,8 @@Streaming Servers
Long Polling
import gevent -from gevent.queue import Queue, Empty from gevent.pywsgi import WSGIServer -import simplejson as json +from gevent.queue import Queue, Empty data_source = Queue() @@ -1552,7 +1548,7 @@Long Polling
while True: try: datum = data_source.get(timeout=5) - yield json.dumps(datum) + '\n' + yield '{}\n'.format(json.dumps(datum)).encode() except Empty: pass @@ -1626,21 +1622,20 @@Websockets
</html>Chat Server
-The final motivating example, a realtime chat room. This example -requires Flask ( but not necessarily so, you could use Django, +
In this example we will build a realtime chat room. It requires +requires Flask ( but not necessarily so, you could use Django, Pyramid, etc ). The corresponding Javascript and HTML files can be found here.
+ +# Micro gevent chatroom. # ---------------------- -from flask import Flask, render_template, request +import json +from flask import Flask, render_template, request from gevent import queue from gevent.pywsgi import WSGIServer -import simplejson as json - app = Flask(__name__) app.debug = True @@ -1724,6 +1719,202 @@Chat Server
http.serve_forever()HTTP2 Server
+In this last example, we will create an HTTP/2 server, +HTTP/2 is the next generation of HTTP protocol, which will serve static files from a browser. +The files are served from a specific folder if specified if not, they will be serve from the +current folder where you run the server. To test this example, you will need python3.6+, +the excellent sans-io library h2. +To generate a private key
+localhost.keyand a self-signed certificatelocalhost.crt+used in the following code you will need the OpenSSH library. If you don't know how to use it, +you can check this +tutorial+import mimetypes +import sys +from functools import partial +from pathlib import Path +from typing import Tuple, Dict + +from gevent import socket, ssl +from gevent.event import Event +from gevent.server import StreamServer +from h2 import events +from h2.config import H2Configuration +from h2.connection import H2Connection + +def get_http2_tls_context() -> ssl.SSLContext: + ctx = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH) + # RFC 7540 Section 9.2: Implementations of HTTP/2 MUST use TLS version 1.2 + # or higher. Disable TLS 1.1 and lower. + ctx.options |= ( + ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 + ) + # RFC 7540 Section 9.2.1: A deployment of HTTP/2 over TLS 1.2 MUST disable + # compression. + ctx.options |= ssl.OP_NO_COMPRESSION + ctx.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20') + ctx.load_cert_chain(certfile='localhost.crt', keyfile='localhost.key') + ctx.set_alpn_protocols(['h2']) + try: + ctx.set_npn_protocols(['h2']) + except NotImplementedError: + pass + + return ctx + +class H2Worker: + + def __init__(self, sock: socket, address: Tuple[str, str], source_dir: str = None): + self._sock = sock + self._address = address + self._flow_control_events: Dict[int, Event] = {} + self._server_name = 'gevent-h2' + self._connection: H2Connection = None + # The maximum amount of a file we'll send in a single DATA frame + self._read_chunk_size = 8192 + + self._check_sources_dir(source_dir) + self._sources_dir = source_dir + + self._run() + + def _initiate_connection(self): + h2_config = H2Configuration(client_side=False, header_encoding='utf-8') + self._connection = H2Connection(h2_config) + self._connection.initiate_connection() + self._sock.sendall(self._connection.data_to_send()) + + @staticmethod + def _check_sources_dir(sources_dir: str) -> None: + p = Path(sources_dir) + if not p.is_dir(): + raise NotADirectoryError(f'{sources_dir} does not exists') + + def _send_error_response(self, status_code: str, + event: events.RequestReceived) -> None: + self._connection.send_headers( + stream_id=event.stream_id, + headers=[ + (':status', status_code), + ('content-length', '0'), + ('server', self._server_name), + ], + end_stream=True + ) + self._sock.sendall(self._connection.data_to_send()) + + def _handle_request(self, event: events.RequestReceived) -> None: + headers = dict(event.headers) + if headers[':method'] != 'GET': + self._send_error_response('405', event) + return + + file_path = Path(self._sources_dir) / headers[':path'].lstrip('/') + if not file_path.is_file(): + self._send_error_response('404', event) + return + + self._send_file(file_path, event.stream_id) + + def _send_file(self, file_path: Path, stream_id: int) -> None: + """ + Send a file, obeying the rules of HTTP/2 flow control. + """ + file_size = file_path.stat().st_size + content_type, content_encoding = mimetypes.guess_type(str(file_path)) + response_headers = [ + (':status', '200'), + ('content-length', str(file_size)), + ('server', self._server_name) + ] + if content_type: + response_headers.append(('content-type', content_type)) + if content_encoding: + response_headers.append(('content-encoding', content_encoding)) + + self._connection.send_headers(stream_id, response_headers) + self._sock.sendall(self._connection.data_to_send()) + + with file_path.open(mode='rb', buffering=0) as f: + self._send_file_data(f, stream_id) + + def _send_file_data(self, file_obj, stream_id: int) -> None: + """ + Send the data portion of a file. Handles flow control rules. + """ + while True: + while self._connection.local_flow_control_window(stream_id) < 1: + self._wait_for_flow_control(stream_id) + + chunk_size = min( + self._connection.local_flow_control_window(stream_id), self._read_chunk_size + ) + data = file_obj.read(chunk_size) + keep_reading = (len(data) == chunk_size) + + self._connection.send_data(stream_id, data, not keep_reading) + self._sock.sendall(self._connection.data_to_send()) + + if not keep_reading: + break + + def _wait_for_flow_control(self, stream_id: int) -> None: + """ + Blocks until the flow control window for a given stream is opened. + """ + event = Event() + self._flow_control_events[stream_id] = event + event.wait() + + def _handle_window_update(self, event: events.WindowUpdated) -> None: + """ + Unblock streams waiting on flow control, if needed. + """ + stream_id = event.stream_id + + if stream_id and stream_id in self._flow_control_events: + g_event = self._flow_control_events.pop(stream_id) + g_event.set() + elif not stream_id: + # Need to keep a real list here to use only the events present at this time. + blocked_streams = list(self._flow_control_events.keys()) + for stream_id in blocked_streams: + g_event = self._flow_control_events.pop(stream_id) + g_event.set() + + def _run(self) -> None: + self._initiate_connection() + + while True: + data = self._sock.recv(65535) + if not data: + break + + h2_events = self._connection.receive_data(data) + for event in h2_events: + if isinstance(event, events.RequestReceived): + self._handle_request(event) + elif isinstance(event, events.DataReceived): + self._connection.reset_stream(event.stream_id) + elif isinstance(event, events.WindowUpdated): + self._handle_window_update(event) + + data_to_send = self._connection.data_to_send() + if data_to_send: + self._sock.sendall(data_to_send) + +if __name__ == '__main__': + files_dir = sys.argv[1] if len(sys.argv) > 1 else f'{Path().cwd()}' + server = StreamServer(('127.0.0.1', 8080), partial(H2Worker, source_dir=files_dir), + ssl_context=get_http2_tls_context()) + try: + server.serve_forever() + except KeyboardInterrupt: + server.close() ++