threading_experiments.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "import time\n",
    "from fair_research_login import NativeClient\n",
    "\n",
    "client = NativeClient(client_id='7414f0b4-7d05-4bb6-bb00-076fa3f17cf5')\n",
    "tokens = client.login(\n",
    "    requested_scopes=['urn:globus:auth:scope:transfer.api.globus.org:all',\n",
    "                      \"https://auth.globus.org/scopes/facd7ccc-c5f4-42aa-916b-a0e270e2c2a9/all\", \n",
    "                      'email', 'openid'],\n",
    "    no_local_server=True,\n",
    "    no_browser=True)\n",
    "\n",
    "transfer_token = tokens['transfer.api.globus.org']['access_token']\n",
    "funcx_token = tokens['funcx_service']['access_token']\n",
    "headers = {'Authorization': f\"Bearer {funcx_token}\",'Transfer': transfer_token, 'FuncX': f\"{funcx_token}\"}\n",
    "print(f\"Headers: {headers}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from xtracthub.xcs import XtractConnection\n",
    "xconn = XtractConnection(funcx_token)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Plotting Threads vs. Time for Sigularity and Docker\n",
    "For this experiment I will measure the time that it takes for XCS to build a fixed number of containers for various numbers of threads. I will then scale up the number of containers. The upload speed will be capped in order to prevent too much variance for Docker."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "csv_name = f'thread_time_results_v2.csv'\n",
    "if os.path.exists(\"./\" + csv_name):\n",
    "    print(f\"{csv_name} already exists, do you want to overwrite?\")\n",
    "    if input() == \"no\":\n",
    "        csv_name = None\n",
    "    else:\n",
    "        pass\n",
    "\n",
    "with open(csv_name, mode='w') as f:\n",
    "    csv_writer = csv.writer(f, delimiter=',', quotechar='\"', quoting=csv.QUOTE_MINIMAL)\n",
    "    csv_writer.writerow([\"type\", \"threads\", \"containers\", \"time\", \"fails\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import datetime\n",
    "import time\n",
    "import uuid\n",
    "from IPython.display import clear_output\n",
    "\n",
    "with open(csv_name, mode='a') as f:\n",
    "    csv_writer = csv.writer(f, delimiter=',', quotechar='\"', quoting=csv.QUOTE_MINIMAL)\n",
    "    for containers in range(10, 60, 10):\n",
    "        for threads in range(5, 55, 5):\n",
    "            print(f\"Containers {containers}\")\n",
    "            print(f\"Threads {threads}\")\n",
    "            requests.post(\"http://149.165.168.132/change_thread\", json={\"threads\": threads})\n",
    "            print(\"Starting Singularity\")\n",
    "\n",
    "            definition_ids = []\n",
    "            for i in range(containers):\n",
    "                file_name = \"my_test.def\"\n",
    "                file_path = \"./examples/my_example.txt\"\n",
    "                definition_id = xconn.register_container(file_name, file_path)\n",
    "                definition_ids.append(definition_id)\n",
    "\n",
    "            build_ids = []\n",
    "            start_time = datetime.datetime.now()\n",
    "            for idx, definition_id in enumerate(definition_ids):\n",
    "                build_id = xconn.build(definition_id, \"singularity\", \"my_test_{}.sif\".format(idx))\n",
    "                build_ids.append(build_id)\n",
    "                print(build_id)\n",
    "\n",
    "            keep_printing = True\n",
    "            while keep_printing:\n",
    "                clear_output(True)\n",
    "                is_done = []\n",
    "                statuses = []\n",
    "                finish_times = []\n",
    "                for idx, build_id in enumerate(build_ids):\n",
    "                    status = xconn.get_status(build_id)\n",
    "                    print(status)\n",
    "                    if status[\"build_status\"] == \"success\":\n",
    "                        is_done.append(True)\n",
    "                        finish_times.append(status[\"build_time\"])\n",
    "                        statuses.append(\"success\")\n",
    "                    elif status[\"build_status\"] == \"failed\":\n",
    "                        is_done.append(True)\n",
    "                        statuses.append(\"failed\")\n",
    "                    else:\n",
    "                        is_done.append(False)\n",
    "                if all(is_done):\n",
    "                    keep_printing = False\n",
    "                time.sleep(1)\n",
    "            \n",
    "            finish_times = list(map(lambda x: datetime.datetime.strptime(x, \"%m/%d/%Y, %H:%M:%S\"), finish_times))\n",
    "            finish_times = list(map(lambda x: x - datetime.timedelta(hours=1), finish_times))\n",
    "            total_time = max(list(map(lambda x: (x - start_time).total_seconds(), finish_times)))\n",
    "            \n",
    "            csv_writer.writerow([\"singularity\", threads, containers, total_time, statuses.count(\"failed\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import uuid\n",
    "from IPython.display import clear_output\n",
    "results = []\n",
    "\n",
    "with open(f'thread_time_results.csv', mode='a') as f:\n",
    "    csv_writer = csv.writer(f, delimiter=',', quotechar='\"', quoting=csv.QUOTE_MINIMAL)\n",
    "    \n",
    "    containers=10\n",
    "    threads=5\n",
    "    print(\"Starting Docker\")\n",
    "    requests.post(\"http://149.165.168.132/change_thread\", json={\"threads\": threads})\n",
    "    definition_ids = []\n",
    "    for i in range(containers):\n",
    "        file_name = \"Dockerfile\"\n",
    "        file_path = \"./examples/matio_dockerfile\"\n",
    "        definition_id = xconn.register_container(file_name, file_path)\n",
    "        definition_ids.append(definition_id)\n",
    "\n",
    "    build_ids = []\n",
    "    for idx, definition_id in enumerate(definition_ids):\n",
    "        build_id = xconn.build(definition_id, \"docker\", \"my_test_{}\".format(idx))\n",
    "        build_ids.append(build_id)\n",
    "        print(build_id)\n",
    "\n",
    "    keep_printing = True\n",
    "    start_time = time.time()\n",
    "    while keep_printing:\n",
    "        clear_output(True)\n",
    "        is_done = []\n",
    "        statuses = []\n",
    "        for idx, build_id in enumerate(build_ids):\n",
    "            status = xconn.get_status(build_id)\n",
    "            print(status)\n",
    "            if status[\"build_status\"] == \"success\":\n",
    "                is_done.append(True)\n",
    "                statuses.append(\"success\")\n",
    "            elif status[\"build_status\"] == \"failed\":\n",
    "                is_done.append(True)\n",
    "                statuses.append(\"failed\")\n",
    "            else:\n",
    "                is_done.append(False)\n",
    "        if all(is_done):\n",
    "            keep_printing = False\n",
    "        time.sleep(1)\n",
    "\n",
    "    csv_writer.writerow([\"docker\", threads, containers, time.time() - start_time, statuses.count(\"failed\")])\n",
    "    print(\"Starting Singularity\")\n",
    "\n",
    "    definition_ids = []\n",
    "    for i in range(containers):\n",
    "        file_name = \"my_test.def\"\n",
    "        file_path = \"./examples/my_example.txt\"\n",
    "        definition_id = xconn.register_container(file_name, file_path)\n",
    "        definition_ids.append(definition_id)\n",
    "\n",
    "    build_ids = []\n",
    "    for idx, definition_id in enumerate(definition_ids):\n",
    "        build_id = xconn.build(definition_id, \"singularity\", \"my_test_{}.sif\".format(idx))\n",
    "        build_ids.append(build_id)\n",
    "        print(build_id)\n",
    "\n",
    "    keep_printing = True\n",
    "    start_time = time.time()\n",
    "    while keep_printing:\n",
    "        clear_output(True)\n",
    "        is_done = []\n",
    "        statuses = []\n",
    "        for idx, build_id in enumerate(build_ids):\n",
    "            status = xconn.get_status(build_id)\n",
    "            print(status)\n",
    "            if status[\"build_status\"] == \"success\":\n",
    "                is_done.append(True)\n",
    "                statuses.append(\"success\")\n",
    "            elif status[\"build_status\"] == \"failed\":\n",
    "                is_done.append(True)\n",
    "                statuses.append(\"failed\")\n",
    "            else:\n",
    "                is_done.append(False)\n",
    "        if all(is_done):\n",
    "            keep_printing = False\n",
    "        time.sleep(1)\n",
    "\n",
    "    csv_writer.writerow([\"singularity\", threads, containers, time.time() - start_time, statuses.count(\"failed\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = []\n",
    "with open('thread_time_results_v2.csv') as csv_file:\n",
    "    csv_reader = csv.reader(csv_file, delimiter=',')\n",
    "    line_count = 0\n",
    "    for row in csv_reader:\n",
    "        data.append(row)\n",
    "data.pop(0)\n",
    "data = list(filter(lambda x: x[0] == \"singularity\", data))\n",
    "\n",
    "for i in list(set([x[2] for x in data])):\n",
    "    cont_data = list(filter(lambda x: x[2] == i, data))\n",
    "    x = [i[1] for i in cont_data]\n",
    "    y = [i[3] for i in cont_data]\n",
    "    print(y)\n",
    "    y = list(map(int, list(map(float, y))))\n",
    "    print(y)\n",
    "    plt.scatter(x, y)\n",
    "    plt.title(f\"{i} containers, Singularity\")\n",
    "    plt.xlabel(\"Threads\")\n",
    "    plt.ylabel(\"Time\")\n",
    "    plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "definition_ids = []\n",
    "\n",
    "for i in range(10):\n",
    "    file_name = \"Dockerfile\"\n",
    "    file_path = \"./examples/matio_dockerfile\"\n",
    "    definition_id = xconn.register_container(file_name, open(file_path, \"rb\"))\n",
    "    definition_ids.append(definition_id)\n",
    "    print(definition_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "build_ids = []\n",
    "\n",
    "for idx, definition_id in enumerate(definition_ids):\n",
    "    build_id = xconn.build(definition_id, \"docker\", \"my_test_{}\".format(idx))\n",
    "    build_ids.append(build_id)\n",
    "    print(build_id)\n",
    "\n",
    "\n",
    "print(build_ids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "keep_printing = True\n",
    "import time\n",
    "from IPython.display import clear_output\n",
    "while keep_printing:\n",
    "    clear_output(True)\n",
    "    is_done = []\n",
    "    for idx, build_id in enumerate(build_ids):\n",
    "        status = xconn.get_status(build_id)\n",
    "        print(status)\n",
    "        if status[\"build_status\"] in [\"success\", \"failed\"]:\n",
    "            is_done.append(True)\n",
    "        else:\n",
    "            is_done.append(False)\n",
    "        print(time.time())\n",
    "    if all(is_done):\n",
    "        keep_printing = False\n",
    "    time.sleep(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "t0 = time.time()\n",
    "# Example for pulling a container\n",
    "for build_id in build_ids:\n",
    "    container_path = os.path.join(os.path.abspath(\".\"), \"my_test.tar\")\n",
    "    response = xconn.pull(build_id, container_path)\n",
    "\n",
    "    if os.path.exists(container_path):\n",
    "        print(\"Successfully pulled container to {}\".format(container_path))\n",
    "    else:\n",
    "        print(response)\n",
    "    print(\"Pulled in {}\".format(time.time() - t0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t0 = time.time()\n",
    "build_ids = []\n",
    "# Example for building a Docker container with a git repo\n",
    "for i in range(10):\n",
    "    git_repo = \"https://github.com/rewong03/xtract_file_service\"\n",
    "    container_name = f\"xfs{i}\"\n",
    "    build_id = xconn.repo2docker(container_name, git_repo=git_repo)\n",
    "    build_ids.append(build_id)\n",
    "    print(build_id)\n",
    "    print(\"Response received in {}\".format(time.time() - t0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t0 = time.time()\n",
    "# Example for getting the status of a container\n",
    "status = xconn.get_status(build_id)\n",
    "print(status)\n",
    "print(\"Got status in {}\".format(time.time() - t0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "build_ids = ['26bda2ab-1bd1-4bb1-be04-20f4e243f47b']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}