[WIP]: Network Simulation For AI Applications

.gitignore

@@ -7,6 +7,7 @@ id_rsa*
 *.tf
 .tmp
 logs/
+jupyter_notebooks/*
 *package-lock.json
 *node_modules
 scripts/*.csv

application/text_translation/launch_benchmark_kubernetes.yml

@@ -0,0 +1,57 @@
+---
+- hosts: cloudcontroller
+  become: true
+  tasks:
+    - name: Create job file
+      shell: |
+        cat > "/home/{{ username }}/job-template.yaml" <<EOF
+        apiVersion: batch/v1
+        kind: Job
+        metadata:
+          name: {{ app_name }}-%ITEM
+        spec:
+          template:
+            metadata:
+              name: {{ app_name }}
+            spec:
+              containers:
+              - name: {{ app_name }}
+                image: {{ image }}
+                ports:
+                - containerPort: 1883
+                imagePullPolicy: Always
+                resources:
+                  requests:
+                    ephemeral-storage: "12Gi"
+                    memory: "{{ memory_req }}Mi"
+                    cpu: {{ cpu_req }}
+                env:
+                - name: MQTT_LOCAL_IP
+                  valueFrom:
+                    fieldRef:
+                      fieldPath: status.hostIP
+                - name: MQTT_LOGS
+                  value: "{{ mqtt_logs }}"
+                - name: CPU_THREADS
+                  value: "{{ cpu_threads }}"
+                - name: ENDPOINT_CONNECTED
+                  value: "{{ endpoint_connected }}"
+              restartPolicy: Never
+        EOF
+
+    - name: Create job directory
+      file:
+        path: /home/{{ username }}/jobs
+        state: directory
+
+    - name: Create job multiple times
+      shell: |
+        for i in `seq 1 {{ replicas }}`
+        do
+          cat "/home/{{ username }}/job-template.yaml" | sed "s/\%ITEM/$i/" \
+          > "/home/{{ username }}/jobs/job-$i.yaml"
+        done
+
+    - name: Launch jobs
+      command: >
+        kubectl apply -f "/home/{{ username }}/jobs"

application/text_translation/src/publisher/Dockerfile

@@ -0,0 +1,23 @@
+# Define custom function directory
+ARG FUNCTION_DIR="/function"
+
+#------------------------------------------------------------------------
+FROM python:3.8.7-slim-buster
+
+# Include global arg in this stage of the build
+ARG FUNCTION_DIR
+
+# Copy function source code
+RUN mkdir -p ${FUNCTION_DIR}
+COPY src/ ${FUNCTION_DIR}/
+
+# Install app specific packages
+RUN pip3 install -r ${FUNCTION_DIR}/requirements.txt
+
+# Set working directory to function root directory
+WORKDIR ${FUNCTION_DIR}
+
+# Open port to the MQTT broker
+EXPOSE 1883
+
+CMD [ "python3", "-u", "publisher.py"]

application/text_translation/src/publisher/docker.sh

@@ -0,0 +1 @@
+docker buildx build --platform linux/amd64,linux/arm64 -t fzovpec2/text_translation:text_translation_publisher --push .

application/text_translation/src/publisher/src/crime_and_punishment.txt

@@ -0,0 +1,53 @@
+On an exceptionally hot evening early in July a young man came out of the garret in which he lodged in S. 
+Place and walked slowly, as though in hesitation, towards K. bridge.
+He had successfully avoided meeting his landlady on the staircase. His garret was under the roof of a high, 
+five-storied house and was more like a cupboard than a room. The landlady who provided him with garret, dinners, 
+and attendance, lived on the floor below, and every time he went out he was obliged to pass her kitchen, the 
+door of which invariably stood open. And each time he passed, the young man had a sick, frightened feeling,
+which made him scowl and feel ashamed. He was hopelessly in debt to his landlady, and was afraid of meeting her.
+This was not because he was cowardly and abject, quite the contrary; but for some time past he had been in an 
+overstrained irritable condition, verging on hypochondria. He had become so completely absorbed in himself,
+and isolated from his fellows that he dreaded meeting, not only his landlady, but anyone at all. He was 
+crushed by poverty, but the anxieties of his position had of late ceased to weigh upon him. He had given 
+up attending to matters of practical importance; he had lost all desire to do so. Nothing that any landlady 
+could do had a real terror for him. But to be stopped on the stairs, to be forced to listen to her trivial, 
+irrelevant gossip, to pestering demands for payment, threats and complaints, and to rack his brains for excuses, 
+to prevaricate, to lie—no, rather than that, he would creep down the stairs like a cat and slip out unseen.
+This evening, however, on coming out into the street, he became acutely aware of his fears.
+“I want to attempt a thing like that and am frightened by these trifles,” he thought, with an odd 
+smile. “Hm... yes, all is in a man’s hands and he lets it all slip from cowardice, that’s an axiom. It would 
+be interesting to know what it is men are most afraid of. Taking a new step, uttering a new word is what they 
+fear most.... But I am talking too much. It’s because I chatter that I do nothing. Or perhaps it is that I 
+chatter because I do nothing. I’ve learned to chatter this last month, lying for days together in my den 
+thinking... of Jack the Giant-killer. Why am I going there now? Am I capable of that? Is that serious? It is 
+not serious at all. It’s simply a fantasy to amuse myself; a plaything! Yes, maybe it is a plaything.”
+The heat in the street was terrible: and the airlessness, the bustle and the plaster, scaffolding, bricks, 
+and dust all about him, and that special Petersburg stench, so familiar to all who are unable to get out of 
+town in summer—all worked painfully upon the young man’s already overwrought nerves. The insufferable stench 
+from the pot-houses, which are particularly numerous in that part of the town, and the drunken men whom he met 
+continually, although it was a working day, completed the revolting misery of the picture. An expression of the 
+profoundest disgust gleamed for a moment in the young man’s refined face. He was, by the way, exceptionally handsome, 
+above the average in height, slim, well-built, with beautiful dark eyes and dark brown hair. Soon he sank into 
+deep thought, or more accurately speaking into a complete blankness of mind; he walked along not observing what 
+was about him and not caring to observe it. From time to time, he would mutter something, from the habit of talking 
+to himself, to which he had just confessed. At these moments he would become conscious that his ideas were sometimes 
+in a tangle and that he was very weak; for two days he had scarcely tasted food.
+He was so badly dressed that even a man accustomed to shabbiness would have been ashamed to be seen in the street in 
+such rags. In that quarter of the town, however, scarcely any shortcoming in dress would have created surprise. Owing 
+to the proximity of the Hay Market, the number of establishments of bad character, the preponderance of the trading 
+and working class population crowded in these streets and alleys in the heart of Petersburg, types so various were 
+to be seen in the streets that no figure, however queer, would have caused surprise. But there was such accumulated 
+bitterness and contempt in the young man’s heart, that, in spite of all the fastidiousness of youth, he minded 
+his rags least of all in the street. It was a different matter when he met with acquaintances or with former 
+fellow students, whom, indeed, he disliked meeting at any time. And yet when a drunken man who, for some 
+unknown reason, was being taken somewhere in a huge waggon dragged by a heavy dray horse, suddenly shouted at him 
+as he drove past: “Hey there, German hatter” bawling at the top of his voice and pointing at him—the young man 
+stopped suddenly and clutched tremulously at his hat. It was a tall round hat from Zimmerman’s, but completely worn 
+out, rusty with age, all torn and bespattered, brimless and bent on one side in a most unseemly fashion. Not shame, 
+however, but quite another feeling akin to terror had overtaken him.
+“I knew it,” he muttered in confusion, “I thought so! That’s the worst of all! Why, a stupid thing like this, the 
+most trivial detail might spoil the whole plan. Yes, my hat is too noticeable.... It looks absurd and that makes 
+it noticeable.... With my rags I ought to wear a cap, any sort of old pancake, but not this grotesque thing. Nobody 
+wears such a hat, it would be noticed a mile off, it would be remembered.... What matters is that people would 
+remember it, and that would give them a clue. For this business one should be as little conspicuous as possible....
+Trifles, trifles are what matter! Why, it’s just such trifles that always ruin everything....”

application/text_translation/src/publisher/src/publisher.py

@@ -0,0 +1,187 @@
+"""\
+This is a publisher, sending local images over MQTT to a subscriber for further processing.
+"""
+
+import time
+import os
+import paho.mqtt.client as mqtt
+
+MQTT_LOCAL_IP = os.environ["MQTT_LOCAL_IP"]
+MQTT_REMOTE_IP = os.environ["MQTT_REMOTE_IP"]
+MQTT_LOGS = os.environ["MQTT_LOGS"]
+FREQUENCY = float(os.environ["FREQUENCY"]) / 10
+MQTT_TOPIC_PUB = "text-translation-pub"
+MQTT_TOPIC_SUB = "text-translation-sub"
+
+if "DURATION" in os.environ:
+    DURATION = int(os.environ["DURATION"])
+else:
+    DURATION = 300
+
+# Set how many imgs to send, and how often
+SEC_PER_FRAME = float(1 / FREQUENCY)
+MAX_TXTS = int(FREQUENCY * DURATION)
+
+RECEIVED = 0
+
+
+def on_connect(local_client, _userdata, _flags, rc):
+    """Execute when connecting to MQTT broker
+
+    Args:
+        local_client (object): Client object
+        _userdata (_type_): _description_
+        _flags (_type_): _description_
+        rc (str): Result code
+    """
+    print("Connected with result code " + str(rc) + "\n", end="")
+    local_client.subscribe(MQTT_TOPIC_PUB)
+
+
+def on_subscribe(_mqttc, _obj, _mid, _granted_qos):
+    """Execute when subscribing to a topic on a MQTT broker
+
+    Args:
+        _mqttc (_type_): _description_
+        _obj (_type_): _description_
+        _mid (_type_): _description_
+        _granted_qos (_type_): _description_
+    """
+    print("Subscribed to topic\n", end="")
+
+
+def on_log(_client, _userdata, level, buff):
+    """Execute MQTT log on every MQTT event
+
+    Args:
+        _client (_type_): _description_
+        _userdata (_type_): _description_
+        level (str): Log level (error, warning, info, etc)
+        buff (str): Log message
+    """
+    print("[ %s ] %s\n" % (str(level), buff), end="")
+
+
+def on_message(_client, _userdata, msg):
+    """Execute when receiving a message on a topic you are subscribed to
+
+    Args:
+        _client (_type_): _description_
+        _userdata (_type_): _description_
+        msg (str): Received message
+    """
+    t_now = time.time_ns()
+
+    print(msg.payload)
+
+    t_old_bytes = msg.payload[-20:]
+    print(len(t_old_bytes))
+    t_old = int(t_old_bytes.decode("utf-8"))
+
+    print("Latency (ns): %i" % (t_now - t_old))
+    global RECEIVED
+    RECEIVED += 1
+
+
+def on_publish(_mqttc, _obj, _mid):
+    """Execute when publishing / sending data
+
+    Args:
+        _mqttc (_type_): _description_
+        _obj (_type_): _description_
+        _mid (_type_): _description_
+    """
+    print("Published data")
+
+
+def connect():
+    """Execute when connecting to a MQTT broker"""
+    print("Start connecting to the local MQTT broker")
+    print("Broker ip: " + str(MQTT_LOCAL_IP))
+    print("Topic: " + str(MQTT_TOPIC_PUB))
+
+    local_client = mqtt.Client()
+    local_client.on_connect = on_connect
+    local_client.on_message = on_message
+    local_client.on_subscribe = on_subscribe
+
+    if MQTT_LOGS == "True":
+        local_client.on_log = on_log
+
+    local_client.connect(MQTT_LOCAL_IP, port=1883, keepalive=300)
+    local_client.loop_start()
+
+
+def send():
+    """Loop over local images, and send them one by one to a remote MQTT broker"""
+    # Loop over the dataset of 60 images
+    files = []
+    print("Start connecting to the remote MQTT broker")
+    print("Broker ip: " + str(MQTT_REMOTE_IP))
+    print("Topic: " + str(MQTT_TOPIC_SUB))
+
+    remote_client = mqtt.Client()
+    remote_client.on_publish = on_publish
+
+    remote_client.connect(MQTT_REMOTE_IP, port=1883, keepalive=120)
+    print("Connected with the broker")
+
+    with open('crime_and_punishment.txt', 'r') as file:
+        lines = file.readlines()
+    print("Read %i lines from the file" % len(lines))
+
+    # Send all frames over MQTT, one by one
+    for i in range(MAX_TXTS):
+        start_time = time.time_ns()
+
+        print("The start time (ns): %i" % (start_time))
+
+        text = lines[i % len(lines)].strip()
+        print("Sending text: %s" % text)
+        byte_arr = bytearray(text.encode("utf-8"))
+
+        # Prepend 0's to the time to get a fixed length string
+        t = time.time_ns()
+        t = (20 - len(str(t))) * "0" + str(t)
+        byte_arr.extend(t.encode("utf-8"))
+
+        # Append local IP address so edge or cloud knows who to send a reply to
+        ip_bytes = (15 - len(MQTT_LOCAL_IP)) * "-" + MQTT_LOCAL_IP
+        byte_arr.extend(ip_bytes.encode("utf-8"))
+
+        print("Sending data (bytes): %i" % (len(byte_arr)))
+        _ = remote_client.publish(MQTT_TOPIC_SUB, byte_arr, qos=0)
+
+        # Try to keep a frame rate of X
+        sec_frame = time.time_ns() - start_time
+        print("Preparation and preprocessing (ns): %i" % (sec_frame))
+        sec_frame = float(sec_frame) / 10**9
+
+        if sec_frame < SEC_PER_FRAME:
+            # Wait until next frame should happen
+            frame = 0.1 * (SEC_PER_FRAME - sec_frame)
+            while sec_frame < SEC_PER_FRAME:
+                time.sleep(frame)
+                sec_frame = float(time.time_ns() - start_time) / 10**9
+        else:
+            print("Can't keep up with %f seconds per frame: Took %f" % (SEC_PER_FRAME, sec_frame))
+
+    # Make sure the finish message arrives
+    remote_client.loop_start()
+    remote_client.publish(MQTT_TOPIC_SUB, "1", qos=2)
+    remote_client.loop_stop()
+
+    remote_client.disconnect()
+    print("Finished, sent %i texts" % (MAX_TXTS))
+
+
+if __name__ == "__main__":
+    connect()
+    send()
+
+    print("Wait for all texts to be received back")
+    while RECEIVED != MAX_TXTS:
+        print("Waiting progress: %i / %i" % (RECEIVED, MAX_TXTS))
+        time.sleep(10)
+
+    print("All %i texts have been received back" % (MAX_TXTS))

application/text_translation/src/publisher/src/requirements.txt

@@ -0,0 +1 @@
+paho-mqtt==1.5.1

application/text_translation/src/subscriber/Dockerfile

@@ -0,0 +1,56 @@
+# Define custom function directory
+ARG FUNCTION_DIR="/function"
+
+#------------------------------------------------------------------------
+FROM python:3.8.7-slim-buster as build_image
+
+# Install packages needed for TFlite
+RUN apt update
+RUN apt install -y \
+    swig \
+    libjpeg-dev \
+    zlib1g-dev \
+    python3-dev \
+    python3-numpy \
+    python3-pip \
+    git \
+    curl \ 
+    unzip \
+    wget \ 
+    tar
+
+# Install python specific packages
+RUN pip3 install numpy pybind11
+
+# Install TF lite from github
+RUN wget https://github.com/tensorflow/tensorflow/archive/refs/tags/v2.3.4.tar.gz
+RUN tar xvf v2.3.4.tar.gz
+RUN bash ./tensorflow-2.3.4/tensorflow/lite/tools/make/download_dependencies.sh
+RUN bash ./tensorflow-2.3.4/tensorflow/lite/tools/pip_package/build_pip_package.sh
+
+
+#------------------------------------------------------------------------
+FROM python:3.8.7-slim-buster as runtime_image
+
+# Include global arg in this stage of the build
+ARG FUNCTION_DIR
+
+# Copy function source code
+RUN mkdir -p ${FUNCTION_DIR}
+COPY src/* ${FUNCTION_DIR}/
+
+# Copy TFlite wheel
+# NOTE: IF TENSORFLOW UPDATES, THIS DOCKERFILE MAY CRASH HERE. UPDATE THE NAMES IN THAT CASE
+# COPY --from=build_image tensorflow-2.3.4/tensorflow/lite/tools/pip_package/gen/tflite_pip/python3/dist/*.whl ./
+# RUN pip3 install *.whl
+
+# Install app specific packages
+RUN pip3 install -r ${FUNCTION_DIR}/requirements.txt
+
+# Set working directory to function root directory
+WORKDIR ${FUNCTION_DIR}
+
+# Open port to the MQTT broker
+EXPOSE 1883
+
+CMD [ "python3", "-u", "subscriber.py"]