Examining the Code¶

This page describes in detail the contents of each used code file.

Utilities¶

There are a few things that both the workers and the client need. These are factored out into a common module, utils.py. You can download the entire file from here.

In the first half of it, there are the imports it will use, and a QuietBytes helper class. We use this in place of its superclass, the built-in bytes type. It only changes the string representation of the base type, to make it shorter than the actual contents. There is a technical reason for using this: It reduces the amount of data transferred to, and stored in, the Redis database.

import io
import zipfile
import os


class QuietBytes(bytes):
    def __str__(self):
        return str(len(self)) + " bytes"

    def __repr__(self):
        return str(len(self)) + " bytes"

There are also two functions in it to handle ZIP archives in memory. The first is for extracting, and the second is for compressing, with the option to exclude some directories.

    with io.BytesIO(zip_bytes) as bytestream:
        with zipfile.ZipFile(bytestream, "r") as zipf:

            zipf.extractall(".")


def zip_directory(directory, exclude_dirs=[]):
    with io.BytesIO() as bytestream:
        with zipfile.ZipFile(bytestream, "w") as zipf:

            for root, dirs, files in os.walk(directory):
                dirs[:] = [d for d in dirs if d not in exclude_dirs]

                for f in files:
                    zipf.write(os.path.join(root, f))

        bytestream.seek(0)
        zipped = QuietBytes(bytestream.read())

    return zipped

Worker code¶

And the code for the jobs, worker.py:

import subprocess
import os
import shutil
import utils


MODEL_DIR = "/tmp/model/"

With the actual job function:

    # housekeeping
    shutil.rmtree(MODEL_DIR, ignore_errors=True)
    os.makedirs(MODEL_DIR)
    os.chdir(MODEL_DIR)

    # unzip source_zip
    utils.unzip_bytes(source_zip)

    # run make, first clean just to be sure
    subprocess.call(["make", "clean"])
    subprocess.call(["make", "MODE=release"])

    # execute binary with args
    subprocess.call([executable] + arguments)

    # zip results
    results_zip = utils.zip_directory("results")

    # cleaning up
    shutil.rmtree(MODEL_DIR, ignore_errors=True)

    # return zip
    return results_zip

The comments make its operation pretty straightforward.

The model needs to be cleaned, then rebuilt inside the container, because the version of some basic system libraries might not match that of those present on the host system, which would lead to incompatibility problems, possibly preventing the simulation from starting.

Dockerfile¶

We select the base image to be ubuntu:16.04, then we install Python, pip, the dependencies of OMNeT++, and wget.

FROM ubuntu:22.04


RUN apt-get update -y && apt-get install -y python3 python3-pip \
      build-essential bison flex libxml2-dev zlib1g-dev wget

We upgrade pip using itself, then install RQ with it. It will also install the Redis client module as a dependency. Then a few environment variables need to be set, to make RQ use the right character encoding.

      build-essential bison flex libxml2-dev zlib1g-dev wget


RUN pip3 install --upgrade pip && pip3 install rq

# These are necessary to make Click in rq happy.
# See: http://click.pocoo.org/5/python3/#python-3-surrogate-handling
ENV LC_ALL C.UTF-8
ENV LANG C.UTF-8

Next, we copy the worker source code into the image, and set the working directory.

ENV LANG C.UTF-8


COPY utils.py /opt/
COPY worker.py /opt/


WORKDIR /opt/

Downloading the OMNeT++ 5.1.1 Core release archive from the official website, extracting it, then deleting it. The referer URL has to be passed to wget, otherwise the server denies access. The --progress flag is there just to reduce the amount of textual output, which would overly pollute the build log.

RUN wget https://github.com/omnetpp/omnetpp/releases/download/omnetpp-5.6.3/omnetpp-5.6.3-src-core.tgz --progress=dot:giga

RUN tar xf omnetpp-5.6.3-src-core.tgz && rm omnetpp-5.6.3-src-core.tgz

The bin directory added to the PATH environment variable (which would be done by setenv normally). Finally the standard building procedure is performed by running ./configure and make. Both graphical runtime environments and the support for 3D rendering are disabled. The -j $(nproc) arguments to make enable it to use all your local CPU cores when building OMNeT++ itself.

Installing ccache to make subsequent builds of the same model sources faster:

And finally setting up the entry point to launch the rq worker, asking it to keep the results only for one minute. This will be enough, because the client will start downloading them right away, and it will reduce the amount of data stored in the Redis database on average.

Later, when we run containers from the image, we will be able to append additional arguments to the entrypoint.

Client Software¶

The following file, client.py, implements the command-line application for submitting jobs and getting the results.

First, the usual imports:

import time
import argparse
import subprocess
import re

from rq import Queue
from redis import Redis

import utils
import worker

Then a helper function to resolve the run filter in a given configuration by invoking the opp_run tool locally, in “query” mode. This is not strictly necessary, since using a run filter is optional, but it’s a nice addition.

import worker

def get_runs_from_filter(configuration, runfilter):
    runs = []

    output = subprocess.check_output(["opp_run", "-q", "runnumbers",
                                      "-c", configuration, "-r", runfilter]).decode("utf-8")
    match = re.search(r"Run numbers: ([\d ]+)", output)
    runs = match.group(1).split()

    return runs

Defining the arguments of the tool and parsing their values:

parser = argparse.ArgumentParser()

parser.add_argument('executable', type=str, help='the Simulation program')
parser.add_argument('-c', metavar='configuration', dest='configuration',
                    type=str, required=True, help='the Configuration to run')
parser.add_argument('-r', metavar='runfilter', dest='runfilter', type=str,
                    required=False, default='', help='the Run Filter selecting the runs')
parser.add_argument('--redis-host', metavar='addr', dest='redis_host',
                    type=str, required=False, default="localhost",
                    help="""the address of the Redis server to use
                            (default: localhost)""")

args = parser.parse_args()

Setting up the connection to the job queue, then using the helper function to get the actual list of run numbers. Finally pack the model source into a compressed archive (ZIP) in memory. A few directories are excluded from this archive, because they are not needed by the workers, and are usually very large.

print("Connecting to Redis at '" + args.redis_host + "'...")

redis_conn = Redis(host=args.redis_host) # Tell RQ what Redis connection to use
q = Queue(connection=redis_conn)  # no args implies the default queue

runs = get_runs_from_filter(args.configuration, args.runfilter)
print("Matched runs: " + ", ".join(runs))

model_source_zip = utils.zip_directory(".", exclude_dirs=["results", "frames", "out"])
print("Size of sources: " + str(len(model_source_zip)) + "B")

Submitting a job into the queue for each run, storing the jobs in a list. The run number is also written into the meta field of each job, but that is necessary only so we know later which run was performed by a particular job, and we can print it when it is done. The job function itself doesn’t use this, only its parameters.

And finally poll for the results of each job, downloading and unpacking the output (the results) of completed jobs, and removing them from the list:

And exit when all jobs are completed.