WorkQueueExample.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d297a417",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "488e09ffed284ca7b7f5357c3b43ec55",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(Text(value='Initializing workers', description='Workqueue Status', disabled=True, style=Descrip…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "46ba443ff967425d998e0afcb8f8f70e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(IntProgress(value=0, description='Percent of tasks complete:', max=98, style=ProgressStyle(desc…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cd7087169c5841c8b87f55920cc58020",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox(children=(HBox(children=(ToggleButton(value=False, description='2', layout=Layout(height='20px', width='2…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a17751fb4f9a44c59dcea097820dbd9a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox(children=(Figure(fig_margin={'top': 60, 'bottom': 60, 'left': 60, 'right': 60}, marks=[Pie(colors=['tomat…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#!/usr/bin/env python\n",
    "\n",
    "# Copyright (c) 2021- The University of Notre Dame.\n",
    "# This software is distributed under the GNU General Public License.\n",
    "# See the file COPYING for details.\n",
    "\n",
    "# This program is a demonstration of using Work Queue with PyTask\n",
    "# PyTask creates a Work Queue task using a python function and its argument and returns the output\n",
    "# This example program finds prime numbers in a range\n",
    "# Finding the primality of each number is created as a seperate task\n",
    "# The tasks for all numbers in the given range are sent out to workers and then computed and returned\n",
    "\n",
    "# How to run this program:\n",
    "# Run this cell to execute the entire work queue program, from creating workers, generating tasks, and then recieving the output\n",
    "# It is recommended that the default values are left for the first run through\n",
    "# Real time output, as well as some informative diagnostic graphs, will be shown at the very bottom of the notebook\n",
    "# Grey cells are tasks that have been submitted, yellow cells are tasks that are currently being worked on,\n",
    "# Green tasks are completed tasks returned by workers. Light green represents that the number is prime, dark green is composite\n",
    "# After the Workqueue Status textbox displays \"All tasks complete!\", the program is finished! \n",
    "# Afterwards, the four variables below (min_workers, max_workers, starting_task, ending_task) can be adjusted to test out different combinations\n",
    "# Then feel free to play around with the number range and number of workers\n",
    "\n",
    "import work_queue as wq\n",
    "import ipywidgets as widgets\n",
    "import time\n",
    "import widget_control \n",
    "\n",
    "display = widget_control.Display() # display class to handle all the fancy output widgets\n",
    "display.min_workers = 1 # controls the minimum number of workers allowed\n",
    "display.max_workers = 30 # controls the maximum number of workers allowed to connect\n",
    "display.starting_task = 2 # controls the first number to determine the primality of\n",
    "display.ending_task = 100 # controls the last number to determine the primality of\n",
    "\n",
    "# function to create the widget display at the bottom of the notebook\n",
    "display.create_output_widgets()\n",
    "\n",
    "# The function below takes in a number as an input and returns 0 if the number is composite or 1 if the number is prime\n",
    "# PyTask uses this function to create the tasks given to workers to determine the primality of numbers in a range\n",
    "def is_prime(number): \n",
    "    import math # PyTask requires that all libraries used in the function are included inside the function\n",
    "    for i in range(2, int(math.sqrt(number)) + 1):\n",
    "            if (number % i == 0):\n",
    "                return 0 # if the number has a divisor, it is composite\n",
    "    return 1 # otherwise it is prime\n",
    "\n",
    "# create the Work Queue queue and specify the number of workers as well as the properties of those workers\n",
    "q = wq.WorkQueue(0) # specify the port to create the workers on. 0 selects a random unused port\n",
    "workers = wq.Factory('local', manager_host_port=f'localhost:{q.port}') # create the worker factory which handles the workers\n",
    "# specify the resources of the workers as well as the maximum and minimum amount of them\n",
    "workers.cores = 1\n",
    "workers.memory = 100\n",
    "workers.disk = 100\n",
    "workers.min_workers = 1\n",
    "workers.max_workers = display.max_workers\n",
    "\n",
    "# the line below is a fault tolerant method of instantiating the Work Queue workers as well as cleaning them up when finished\n",
    "with workers: \n",
    "    # loop through the range of numbers desired and use PyTask to create tasks to determine their primality\n",
    "    for i in range(display.starting_task, display.ending_task + 1):\n",
    "        p_task = wq.PythonTask(is_prime, i) # create the task using PyTask, supplying the function and its argument(s)\n",
    "        q.submit(p_task) # submit the tasks into the queue\n",
    "        \n",
    "    while not q.empty(): # continue until all the tasks have been completed\n",
    "        t = q.wait(5) # give control to Work Queue so that it can talk to workers\n",
    "        if t:\n",
    "            if t.return_status != 0: # The task failed\n",
    "                continue\n",
    "            display.update_output_widgets(q, t) # update output widgets "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.10"
  },
  "metadata": {
   "interpreter": {
    "hash": "202893b2fb2d862276ef4222df57aaac0db00b652ff02bbc43a600d07b7c5397"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}