RabbitMQ, Docker deployment

Author: alexmojaki

.dockerignore

@@ -0,0 +1,2 @@
+frontend/node_modules/
+backend/db.sqlite3

Dockerfile

@@ -0,0 +1,47 @@
+# Stage 1 - Build frontend
+FROM node:12 as build-frontend
+WORKDIR /usr/src/app
+COPY frontend/package-lock.json frontend/package.json ./frontend/
+
+WORKDIR frontend
+RUN npm ci
+COPY frontend .
+RUN npm run build
+
+# Stage 2 - Setup server
+FROM ubuntu:18.04
+
+RUN apt-get update \
+    && DEBIAN_FRONTEND=noninteractive apt-get install -y \
+    python3.8-dev \
+    unzip \
+    nginx \
+    curl \
+    vim \
+    python3.8-distutils \
+    gcc
+
+RUN ln -s /usr/bin/python3.8 /usr/local/bin/python
+RUN python --version
+RUN curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py
+RUN python get-pip.py
+RUN pip --version
+RUN pip install poetry
+
+#RUN curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.deb.sh | bash
+
+WORKDIR /usr/src/app
+COPY poetry.lock pyproject.toml ./
+RUN poetry config virtualenvs.create false
+RUN LC_ALL=C.UTF-8 LANG=C.UTF-8 poetry install --extras "production" --no-dev
+
+COPY --from=build-frontend /usr/src/app/frontend/build ./frontend/build
+COPY ./backend ./backend
+WORKDIR /usr/src/app/backend
+RUN ./manage.py collectstatic --noinput
+
+COPY ./backend/nginx.conf /etc/nginx
+
+EXPOSE 3000
+
+CMD ["bash", "./entrypoint.sh"]

backend/book/settings.py

@@ -29,6 +29,9 @@
 
 GITHUB_TOKEN = os.environ.get('GITHUB_TOKEN')
 
+RABBITMQ_HOST = os.environ.get('RABBITMQ_HOST', '')
+SEPARATE_WORKER_PROCESS = os.environ.get('SEPARATE_WORKER_PROCESS', 'False')[0].upper() == 'T'
+
 ALLOWED_HOSTS = [
     'alexmojaki.pythonanywhere.com',
     'localhost',

backend/entrypoint.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+set -ex
+
+nginx
+
+#rabbitmq-server -detached
+
+# We probably don't want this to be automatic but it makes life a lot easier
+# For setting up the cloud
+python manage.py migrate
+python manage.py init_db
+
+#python -m main.worker &
+
+gunicorn -c gunicorn_config.py book.wsgi:application

backend/gunicorn_config.py

@@ -0,0 +1,25 @@
+# user = 'root'
+bind = '0.0.0.0:8000'
+# workers = 1
+worker_class = 'gevent'
+# worker_connections = 1024
+# The number of seconds to wait for requests on a Keep-Alive connection.
+# Set higher than default to deal with latency.
+keepalive = 90
+# How max time worker can handle request after got restart signal.
+# If the time is up worker will be force killed.
+graceful_timeout = 10
+# If a worker does not notify the master process in this number of seconds it is
+# killed and a new worker is spawned to replace it.
+#
+# For the non sync workers it just means that the worker  process is still
+# communicating and is not tied to the length of time required to handle a single
+# request.
+timeout = 600
+loglevel = 'debug'
+log_file = '-'
+# "-" means log to stdout
+accesslog = '-'
+errorlog = '-'
+# Sets the process title
+proc_name = 'gunicorn'

backend/main/utils.py

@@ -1,5 +1,7 @@
 import functools
+import threading
 import traceback
+from functools import lru_cache
 from io import StringIO
 import re
 import sys
@@ -64,3 +66,18 @@ def row_to_dict(row):
 
 def rows_to_dicts(rows):
     return [row_to_dict(row) for row in rows]
+
+
+def thread_separate_lru_cache(*cache_args, **cache_kwargs):
+    def decorator(func):
+        @lru_cache(*cache_args, **cache_kwargs)
+        def cached(_thread_id, *args, **kwargs):
+            return func(*args, **kwargs)
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            return cached(threading.get_ident(), *args, **kwargs)
+
+        return wrapper
+
+    return decorator

backend/main/views.py

@@ -22,7 +22,7 @@
 from main.chapters.c06_lists import UnderstandingProgramsWithPythonTutor
 from main.models import CodeEntry
 from main.text import Page, page_slugs_list, pages, ExerciseStep, clean_program
-from main.worker import worker_connection
+from main.worker import worker_result
 
 log = logging.getLogger(__name__)
 
@@ -91,16 +91,12 @@ def run_code(self, code, source):
             source=source,
             page_slug=self.user.page_slug,
             step_name=self.user.step_name,
+            user_id=self.user.id,
         )
 
-        entry = CodeEntry.objects.create(
-            **entry_dict,
-            user=self.user,
-        )
+        entry = CodeEntry.objects.create(**entry_dict)
 
-        connection = worker_connection(self.user.id)
-        connection.send(entry_dict)
-        result = connection.recv()
+        result = worker_result(entry_dict)
 
         entry.output = result["output"]
         entry.save()

backend/main/worker.py

@@ -9,13 +9,13 @@
 import resource
 import sys
 from code import InteractiveConsole
+from collections import defaultdict
 from datetime import datetime
 from functools import lru_cache
 from importlib import import_module
-from multiprocessing import Pipe
-from multiprocessing.connection import Connection
 from multiprocessing.context import Process
 from threading import Thread
+from time import sleep
 
 import birdseye.bird
 import snoop
@@ -26,6 +26,7 @@
 
 from main.text import pages
 from main.utils import format_exception_string, rows_to_dicts
+from main.workers.communications import AbstractCommunications, ThreadCommunications
 
 log = logging.getLogger(__name__)
 
@@ -288,57 +289,53 @@ def readline():
     )
 
 
-def consumer(connection: Connection):
-    manager = multiprocessing.Manager()
-    task_queue = manager.Queue()
-    input_queue = manager.Queue()
-    result_queue = manager.Queue()
-    process = None
-
-    def start_process():
-        nonlocal process
-        process = Process(
+class UserProcess:
+    def __init__(self, manager):
+        self.task_queue = manager.Queue()
+        self.input_queue = manager.Queue()
+        self.result_queue = manager.Queue()
+        self.awaiting_input = False
+        self.process = None
+        self.start_process()
+
+        @atexit.register
+        def cleanup():
+            if self.process:
+                self.process.terminate()
+
+    def start_process(self):
+        self.process = Process(
             target=worker_loop_in_thread,
-            args=(task_queue, input_queue, result_queue),
+            args=(self.task_queue, self.input_queue, self.result_queue),
             daemon=True,
         )
-        process.start()
-
-    start_process()
-
-    def cleanup():
-        process.terminate()
-
-    atexit.register(cleanup)
+        self.process.start()
 
-    awaiting_input = False
-
-    def run():
-        task_queue.put(entry)
-
-    while True:
-        entry = connection.recv()
+    def handle_entry(self, entry):
         if entry["source"] == "shell":
-            if awaiting_input:
-                input_queue.put(entry["input"])
+            if self.awaiting_input:
+                self.input_queue.put(entry["input"])
             else:
-                run()
+                self.task_queue.put(entry)
         else:
-            if not TESTING and awaiting_input:
-                process.terminate()
-                start_process()
+            if not TESTING and self.awaiting_input:
+                self.process.terminate()
+                self.start_process()
 
-            run()
+            self.task_queue.put(entry)
+
+    def await_result(self, callback):
+        # TODO cancel if result was cancelled by a newer handle_entry
         result = None
         while result is None:
             try:
-                result = result_queue.get(timeout=3)
+                result = self.result_queue.get(timeout=3)
             except queue.Empty:
-                alive = process.is_alive()
+                alive = self.process.is_alive()
                 print(f"Process {alive=}")
                 if alive:
-                    process.terminate()
-                start_process()
+                    self.process.terminate()
+                self.start_process()
                 result = dict(
                     output_parts=[
                         dict(color='red', text='The process died.\n'),
@@ -351,13 +348,62 @@ def run():
                     awaiting_input=False,
                     birdseye_objects=None,
                 )
-        awaiting_input = result["awaiting_input"]
-        connection.send(result)
+        self.awaiting_input = result["awaiting_input"]
+        callback(result)
+
+
+def master_consumer_loop(comms: AbstractCommunications):
+    comms = comms.make_master_side_communications()
+    manager = multiprocessing.Manager()
+    user_processes = defaultdict(lambda: UserProcess(manager))
+
+    while True:
+        entry = comms.recv_entry()
+        user_id = str(entry["user_id"])
+        user_process = user_processes[user_id]
+        user_process.handle_entry(entry)
+
+        def callback(result):
+            comms.send_result(user_id, result)
+
+        Thread(
+            target=user_process.await_result,
+            args=[callback],
+        ).start()
+
+
+@lru_cache()
+def master_communications() -> AbstractCommunications:
+    from django.conf import settings
+    if settings.RABBITMQ_HOST:
+        from .workers.pika import PikaCommunications
+        comms = PikaCommunications()
+    else:
+        comms = ThreadCommunications()
+
+    if not settings.SEPARATE_WORKER_PROCESS:
+        Thread(
+            target=master_consumer_loop,
+            args=[comms],
+            daemon=True,
+            name=master_consumer_loop.__name__,
+        ).start()
+
+    return comms
+
+
+def worker_result(entry):
+    comms: AbstractCommunications = master_communications()
+    comms.send_entry(entry)
+    user_id = str(entry["user_id"])
+    return comms.recv_result(user_id)
+
+
+def main():
+    from main.workers.pika import PikaCommunications
+    comms = PikaCommunications()
+    master_consumer_loop(comms)
 
 
-@lru_cache(maxsize=None)
-def worker_connection(_user_id):
-    parent_connection, child_connection = Pipe()
-    p = Thread(target=consumer, args=(child_connection,), daemon=True)
-    p.start()
-    return parent_connection
+if __name__ == '__main__':
+    main()

backend/main/workers/__init__.py

@@ -0,0 +1,41 @@
+from abc import ABC, abstractmethod
+
+
+class AbstractCommunications(ABC):
+    @abstractmethod
+    def send_entry(self, entry): ...
+
+    @abstractmethod
+    def recv_entry(self): ...
+
+    @abstractmethod
+    def send_result(self, queue_name, result): ...
+
+    @abstractmethod
+    def recv_result(self, queue_name): ...
+
+    @abstractmethod
+    def make_master_side_communications(self) -> "AbstractCommunications": ...
+
+
+class ThreadCommunications(AbstractCommunications):
+    def __init__(self):
+        from multiprocessing.dummy import Pipe
+
+        self.server_connection, self.child_connection = Pipe()
+
+    def make_master_side_communications(self):
+        return self
+
+    def send_entry(self, entry):
+        self.server_connection.send(entry)
+
+    def recv_entry(self):
+        return self.child_connection.recv()
+
+    # TODO doesn't handle multiple users (queues)
+    def send_result(self, queue_name, result):
+        self.child_connection.send(result)
+
+    def recv_result(self, queue_name):
+        return self.server_connection.recv()