Begun work on the case study for the server

python · cjrh · Oct 7, 2018 · Oct 14, 2018 · Oct 21, 2018 · Oct 21, 2018
commit 50a901eba6b359d0181c1cf31663448cb9769d66
diff --git a/...brary/asyncio-tutorial/case-study-cli.rst → ...o-tutorial/case-study-chat-client-cli.rst b/...brary/asyncio-tutorial/case-study-cli.rst → ...o-tutorial/case-study-chat-client-cli.rst
diff --git a/...brary/asyncio-tutorial/case-study-gui.rst → ...o-tutorial/case-study-chat-client-gui.rst b/...brary/asyncio-tutorial/case-study-gui.rst → ...o-tutorial/case-study-chat-client-gui.rst
diff --git a/Doc/library/asyncio-tutorial/case-study-chat-server.rst b/Doc/library/asyncio-tutorial/case-study-chat-server.rst
@@ -0,0 +1,154 @@
+Asyncio Case Study: Chat Application (Server)
+=============================================
+
+We're going to build a chat application. Users will be able to
+connect to a central server and send messages to "chat rooms", which
+all the other users in those rooms will be able to see. This case study
+gives us an opportunity to show how to use the various features of
+``asyncio`` in a "real world" application.
+
+This will be a client-server application.  The server application
+will run on a central host, and the chat application will run on each
+user's computer or device. The server is the easier of the two, for
+interesting reasons that we'll get into later.
+
+We're going to start with a basic application layout and build up from
+there.
+
+Starting Code Layout
+--------------------
+
+The code below shows the basic starting template for an asyncio
+service. In these kinds of applications, the service will typically
+run for a long time, serving clients that may connect,
+disconnect, and then later reconnect multiple times.
+
+.. literalinclude:: server01.py
+    :language: python
+
+As explained earlier, ``main`` itself is a *coroutine function*, and
+when evaluated, i.e., ``main()``, it returns a *coroutine* object
+which the ``asyncio.run()`` function knows how to execute.
+
+.. note::
+    The ``asyncio.run()`` function waits for ``main()`` to complete.
+    When ``main()`` returns, the ``run()`` function will then cancel
+    all tasks that are still around. This means, precisely, that the
+    ``asyncio.CancelledError`` exception will get raised in all such
+    pending tasks. This gives you a way to deal with an "application
+    shutdown" scenario: you just need to handle the ``CancelledError``
+    exception in places where you need a controlled way of terminating
+    tasks.
+
+There isn't much more to say about the basic template, so let's add
+the actual server.
+
+Server
+------
+
+We can use the *Streams API* (ref:TODO) to create a TCP server very
+easily:
+
+.. literalinclude:: server02.py
+    :language: python
+    :linenos:
+
+We've added the ``start_server()`` call on line 5, and this call takes
+not only the ``host`` and ``port`` parameters you'd expect, but also a
+*callback* function that will be called for each new connection. This
+is coroutine function ``client_connected()``, on line 13.
+
+The callback is provided with ``reader`` and ``writer`` parameters.
+These give access to two streams for this new connection. It is here
+that data will be received from, and sent to clients.
+
+Printing out "New client connected!" is obviously going to be quite
+useless. We're going to want to receive chat messages from a client,
+and we also want to send these messages to all the other clients in the
+same "chat room". We don't yet have the concept of "rooms" defined
+anywhere yet, but that's ok. Let's first focus on what must be sent
+and received between server and client.
+
+Let's sketch out a basic design of the communication pattern:
+
+#. Client connects to server
+#. Client sends a message to server to announce themselves, and join
+   a room
+#. Client sends a message to a room
+#. Server relays that message to all other clients in the same room
+#. Eventually, client disconnects.
+
+These actions suggest a few different kinds of information that need to
+be sent between server and client. We need to create a *protocol*
+that both server and client can use to communicate.
+
+How about we use JSON messages? Here is an example of the payload a
+client needs to provide immediately after connection:
+
+.. code-block:: json
+    :caption: Client payload after connection
+
+    {
+        "action": "connect",
+        "username": "<something>"
+    }
+
+Here are example messages for joining and leaving rooms:
+
+.. code-block:: json
+    :caption: Client payload to join a room
+
+    {
+        "action": "joinroom",
+        "room": "<room_name>"
+    }
+
+.. code-block:: json
+    :caption: Client payload to leave a room
+
+    {
+        "action": "leaveroom",
+        "room": "<room_name>"
+    }
+
+And here's an example of a client payload for sending a chat
+message to a room:
+
+.. code-block:: json
+    :caption: Client payload to send a message to a room
+
+    {
+        "action": "chat",
+        "room": "<room_name>",
+        "message": "I'm reading the asyncio tutorial!"
+    }
+
+All of the JSON examples above are for payloads that will be received
+from clients, but remember that the server must also send messages
+to all clients in a room. That message might look something like this:
+
+.. code-block:: json
+    :caption: Server payload to update all clients in a room
+    :linenos:
+
+    {
+        "action": "chat",
+        "room": "<room_name>",
+        "message": "I'm reading the asyncio tutorial!",
+        "from": "<username>"
+    }
+
+The message is similar to the one received by a client, but on line 5
+we now need to indicate from whom the message was sent.
+
+
+
+TODO
+
+Notes:
+
+- using the streams API
+- first show the client.
+- will have to explain a message protocol. (But that's easy)
+- then show the server
+- spend some time on clean shutdown.
diff --git a/Doc/library/asyncio-tutorial/index.rst b/Doc/library/asyncio-tutorial/index.rst
@@ -20,5 +20,6 @@ primarily on the "high-level" API, as described in the
    async-functions.rst
    running-async-functions.rst
    asyncio-cookbook.rst
-   case-study-cli.rst
-   case-study-gui.rst
+   case-study-chat-server.rst
+   case-study-chat-client-cli.rst
+   case-study-chat-client-gui.rst
diff --git a/Doc/library/asyncio-tutorial/running-async-functions.rst b/Doc/library/asyncio-tutorial/running-async-functions.rst
@@ -1,4 +1,4 @@
-Executing async functions
+Executing Async Functions
 =========================
 
 TODO

diff --git a/Doc/library/asyncio-tutorial/server01.py b/Doc/library/asyncio-tutorial/server01.py
@@ -0,0 +1,7 @@
+import asyncio
+
+async def main():
+    ...
+
+if __name__ == '__main__':
+    asyncio.run(main())
diff --git a/Doc/library/asyncio-tutorial/server02.py b/Doc/library/asyncio-tutorial/server02.py
@@ -0,0 +1,17 @@
+import asyncio
+from asyncio import StreamReader, StreamWriter
+
+async def main():
+    server = await asyncio.start_server(
+        client_connected_cb=client_connected,
+        host='localhost',
+        port='9011',
+    )
+    async with server:
+        await server.serve_forever()
+
+async def client_connected(reader: StreamReader, writer: StreamWriter):
+    print('New client connected!')
+
+if __name__ == '__main__':
+    asyncio.run(main())
diff --git a/Doc/library/asyncio-tutorial/server20.py b/Doc/library/asyncio-tutorial/server20.py
@@ -0,0 +1,73 @@
+import asyncio
+from collections import defaultdict
+from weakref import WeakValueDictionary
+import json
+import utils
+import ssl
+
+
+WRITERS = WeakValueDictionary()
+ROOMS = defaultdict(WeakValueDictionary)
+
+
+async def sender(addr, writer, room, msg):
+    try:
+        await utils.send_message(
+            writer,
+            json.dumps(dict(room=room, msg=msg)).encode()
+        )
+    except (ConnectionAbortedError, ConnectionResetError):
+        """ Connection is dead, remove it."""
+        if addr in WRITERS:
+            del WRITERS[addr]
+        if addr in ROOMS[room]:
+            del ROOMS[room][addr]
+
+
+def send_to_room(from_addr, room: str, msg: str):
+    """Send the message to all clients in the room."""
+    for addr, writer in ROOMS[room].items():
+        print(f'Sending message to {addr} in room {room}: {msg}')
+        asyncio.create_task(sender(addr, writer, room, msg))
+
+
+async def client_connected_cb(reader, writer):
+    addr = writer.get_extra_info('peername')
+    print(f'New connection from {addr}')
+    WRITERS[addr] = writer
+    async for msg in utils.messages(reader):
+        print(f'Received bytes: {msg}')
+        d = json.loads(msg)
+        if d.get('action') == 'join':
+            ROOMS[d['room']][addr] = writer
+        elif d.get('action') == 'leave':
+            del ROOMS[d['room']][addr]
+        else:
+            d['from'] = addr
+            send_to_room(addr, d['room'], d['msg'])
+
+
+async def main():
+    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    ctx.check_hostname = False
+    ctx.load_cert_chain('chat.crt', 'chat.key')
+    server = await asyncio.start_server(
+        client_connected_cb=client_connected_cb,
+        host='localhost',
+        port='9011',
+        ssl=ctx,
+    )
+    shutdown = asyncio.Future()
+    utils.install_signal_handling(shutdown)
+    print('listening...')
+    async with server:
+        done, pending = await asyncio.wait(
+            [server.serve_forever(), shutdown],
+            return_when=asyncio.FIRST_COMPLETED
+        )
+        if shutdown.done():
+            return
+
+
+if __name__ == '__main__':
+    asyncio.run(main())
diff --git a/Doc/library/asyncio-tutorial/utils20.py b/Doc/library/asyncio-tutorial/utils20.py
@@ -0,0 +1,59 @@
+import sys
+from asyncio import (StreamReader, StreamWriter, IncompleteReadError, Future,
+    get_running_loop)
+
+if sys.platform == 'win32':
+    from signal import signal, SIGBREAK, SIGTERM, SIGINT
+else:
+    SIGBREAK = None
+    from signal import signal, SIGTERM, SIGINT
+
+from typing import AsyncGenerator
+
+
+async def messages(reader: StreamReader) -> AsyncGenerator[bytes, None]:
+    """Async generator to return messages as they come in."""
+    try:
+        while True:
+            size_prefix = await reader.readexactly(4)
+            size = int.from_bytes(size_prefix, byteorder='little')
+            message = await reader.readexactly(size)
+            yield message
+    except (IncompleteReadError, ConnectionAbortedError, ConnectionResetError):
+        return
+
+
+async def send_message(writer: StreamWriter, message: bytes):
+    """To close the connection, use an empty message."""
+    if not message:
+        writer.close()
+        await writer.wait_closed()
+        return
+    size_prefix = len(message).to_bytes(4, byteorder='little')
+    writer.write(size_prefix)
+    writer.write(message)
+    await writer.drain()
+
+
+def install_signal_handling(fut: Future):
+    """Given future will be set a signal is received. This
+    can be used to control the shutdown sequence."""
+    if sys.platform == 'win32':
+        sigs = SIGBREAK, SIGINT
+        loop = get_running_loop()
+
+        def busyloop():
+            """Required to handle CTRL-C quickly on Windows
+            https://bugs.python.org/issue23057 """
+            loop.call_later(0.1, busyloop)
+
+        loop.call_later(0.1, busyloop)
+    else:
+        sigs = SIGTERM, SIGINT
+
+    # Signal handlers. Windows is a bit tricky
+    for s in sigs:
+        signal(
+            s,
+            lambda *args: loop.call_soon_threadsafe(fut.set_result, None)
+        )
diff --git a/Doc/library/asyncio-tutorial/what-asyncio.rst b/Doc/library/asyncio-tutorial/what-asyncio.rst
@@ -1,5 +1,5 @@
-What Does Async Mean?
-=====================
+What Does "Async" Mean?
+=======================
 
 Let's make a function that communicates over the network:
 
@@ -54,6 +54,9 @@ code is also going to **wait** sequentially. This is, quite literally,
 a waste of time. What we would really like to do here is wait for the
 all the replies *concurrently*, i.e., at the same time.
 
+Preemptive Concurrency
+----------------------
+
 Operating systems like Windows, Mac and Linux, and others, understand
 this problem deeply. If you're reading this on a computer or even your
 mobile, there will be tens or hundreds of processes running at the same
@@ -68,22 +71,34 @@ from the operating system.  But more than that, there is another tricky
 problem about *how* the operating system knows when to allocate
 execution time between each process. The answer: it doesn't! This means
 that the operating system can decide when to give processor time to each
-process.
-
-And this means that you will never be sure of when each of your processes
-is actually running, relative to each other. This is quite safe because
+process. Your code, and therefore you, will not know when these switches
+occur. This is called "preemption". From
+`Wikipedia <https://en.wikipedia.org/wiki/Preemption_(computing)>`_:
+*In computing, preemption is the act of temporarily interrupting a
+task being carried out by a computer system, without requiring
+its cooperation, and with the intention of resuming the task
+at a later time*.
+
+This means that you will never be sure of when each of your processes
+is *actually* executing on a CPU. This is quite safe because
 processes are isolated from each other; however, **threads** are not
 isolated from each other. In fact, the primary feature of threads over
 processes is that multiple threads within a single process can
-access the same memory. And this is where all the problems appear.
+access the same memory. And this is where all the problems begin.
 
-So: we can also run the ``greet()`` function in multiple threads, and then
+Jumping back to our code sample further up: we may also choose to run the
+``greet()`` function in multiple threads; and then
 they will also wait for replies concurrently. However, now you have
-two threads that is allowed to access the same objects, with no control over
-how execution will be transferred between the two threads. This
+two threads that are allowed to access the same objects in memory,
+with no control over
+how execution will be transferred between the two threads (unless you
+use the synchronization primitives in the ``threading`` module) . This
 situation can result in *race conditions* in how objects are modified,
 and these bugs can be very difficult to debug.
 
+Cooperative Concurrency
+-----------------------
+
 This is where "async" programming comes in. It provides a way to manage
 multiple socket connections all in a single thread; and the best part
 is that you get to control *when* execution is allowed to switch between