Aging 26th April Code (2).ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "intialTime 1587888877.643597\n",
      "   task_number  time_of_task\n",
      "0            1          4000\n",
      "1            2          1000\n",
      "2            2          3000\n",
      "3            3          3000\n",
      "4            1           500\n",
      "5            3          1000\n",
      "  flow_size job\n",
      "1     10000   1\n",
      "2      5000   1\n",
      "3     10000   2\n",
      "  flow_size job\n",
      "1      1000   1\n",
      "2     10000   2\n",
      "3      4000   2\n",
      "  flow_size job\n",
      "1     10000   1\n",
      "2      1000   1\n",
      "3      6000   2\n",
      "4      6000   2\n"
     ]
    }
   ],
   "source": [
    "#!/usr/bin/python\n",
    "from random import randint\n",
    "import heapq\n",
    "from datetime import datetime\n",
    "import time\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import os\n",
    "import threading\n",
    "\n",
    "numOfWorkers=100\n",
    "busyWorker=[False]*100   # for showing if the worker is busy working on something\n",
    "flag=True\n",
    "heapEntry=[] # a list of lists. Each entry of this list will be of the form -[priority, date time of creation, task name]\n",
    "allTasksFinished=False\n",
    "totalTime=0\n",
    "#Every datapoint has a time associated with it using which we will schedule the tasks that are given.\n",
    "#So we first read the order of incoming tasks(i.e. the type of task -PageRank, KMeans, SGD) to be picked alongwith their time of pickup from the file and then add all the flows of that particular task to the queue. \n",
    "\n",
    "intialTime=time.time()\n",
    "print(\"intialTime\",intialTime)\n",
    "\n",
    "task_order_path='C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/job_incoming_test.csv'\n",
    "k_means_path='C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/KMeans_test_jb_test.csv'\n",
    "sgd_path='C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/SGD_test_jb_test.csv'\n",
    "page_rank_path='C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/PageRank_test_jb_test.csv'\n",
    "\n",
    "task_order_file=open(task_order_path, 'r')\n",
    "k_means_file=open(k_means_path, 'r')\n",
    "sgd_file=open(sgd_path, 'r')\n",
    "page_rank_file=open(page_rank_path, 'r')\n",
    "'''\n",
    "df_task = pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/job_incoming.txt', sep=\",\", names=['task_number', 'time_of_task'])\n",
    "'''\n",
    "df_task = pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/job_incoming_test.csv', sep=\",\", names=['task_number', 'time_of_task'])\n",
    "print(df_task)\n",
    "\n",
    "df_k_means=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/KMeans_test_jb_test.csv', sep=\",\", names=['flow_size','job'])\n",
    "df_sgd=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/SGD_test_jb_test.csv', sep=\",\", names=['flow_size','job'])\n",
    "df_page_rank=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/PageRank_test_jb_test.csv', sep=\",\", names=['flow_size','job'])\n",
    "df_k_means=df_k_means.tail(3)\n",
    "df_sgd=df_sgd.tail(3)\n",
    "df_page_rank=df_page_rank.tail(4)\n",
    "\n",
    "'''\n",
    "df_k_means=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/KMeans_test_jb.txt', sep=\",\", names=['time','previous_flow_time','agg_net_out','agg_net_in','agg_net_out_per_machine','agg_net_in_per_machine','tcp_queue','machine','cpu','ram','disk_write','memory_read','memory_write','first_call','flow_size','job'])\n",
    "df_sgd=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/SGD_test_jb.csv', sep=\",\", names=['time','first_call','machine','agg_net_out_per_machine','agg_net_in_per_machine','agg_net_out','agg_net_in','flow_frequency','flow_size','job'])\n",
    "df_page_rank=pd.read_csv('C:/Users/bodhv/Desktop/Sem2/Cloud/Project/Data/PageRank_test_jb.csv', sep=\",\", names=['time','flow_frequency','agg_net_out','agg_net_in','agg_net_out_per_machine','agg_net_in_per_machine','machine','flow_size','job'])\n",
    "'''\n",
    "print(df_k_means)\n",
    "print(df_sgd)\n",
    "print(df_page_rank)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6\n",
      "waiting for the first time\n",
      "current_task_index 0\n",
      "Inside first while\n",
      "\n",
      "current task number 1\n",
      "1-curr_wait_time 1.0\n",
      "1-start_wait_time 0.0\n",
      "current_k_means_job 1\n",
      "kmeans curr_df   flow_size job\n",
      "1     10000   1\n",
      "2      5000   1\n",
      "k means job: 1    1\n",
      "2    1\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 2\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "1     10000   1\n",
      "2      5000   1\n",
      "start index 1\n",
      "end index 2\n",
      "x 1\n",
      "curr_df   flow_size job\n",
      "1     10000   1\n",
      "2      5000   1\n",
      "time to completion 10.0\n",
      "listOfTasks[Task1Job1]=[1, '20200426043843', 10.0]\n",
      "x 2\n",
      "curr_df   flow_size job\n",
      "1     10000   1\n",
      "2      5000   1\n",
      "time to completion 5.0\n",
      "listOfTasks[Task1Job2]=[1, '20200426043843', 5.0]\n",
      " final listOfTasks {'Task1Job1': [1, '20200426043843', 10.0], 'Task1Job2': [1, '20200426043843', 5.0]}\n",
      "element added to heap: [1, 20200426043843, 10.0, 'Task1Job1']\n",
      "current heap: [[1, 20200426043843, 10.0, 'Task1Job1']]\n",
      "element added to heap: [1, 20200426043843, 5.0, 'Task1Job2']\n",
      "current heap: [[1, 20200426043843, 5.0, 'Task1Job2'], [1, 20200426043843, 10.0, 'Task1Job1']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 1.0\n",
      "3-start_wait_time 0.0\n",
      "heap length: 2\n",
      "The updated heap: [[1, 20200426043843, 5.0, 'Task1Job2'], [1, 20200426043843, 10.0, 'Task1Job1']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043843, 5.0, 'Task1Job2']\n",
      "worker 1 is free\n",
      "popped entry: [1, 20200426043843, 10.0, 'Task1Job1']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "current_task_index 1\n",
      "Inside first while\n",
      "\n",
      "current task number 2\n",
      "1-curr_wait_time 3.0\n",
      "1-start_wait_time 0.0\n",
      "current_sgd_job 1\n",
      "sgd job: 1    1\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 1\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "1      1000   1\n",
      "start index 1\n",
      "end index 1\n",
      "x 1\n",
      "curr_df   flow_size job\n",
      "1      1000   1\n",
      "time to completion 1.0\n",
      "listOfTasks[Task2Job1]=[1, '20200426043844', 1.0]\n",
      " final listOfTasks {'Task2Job1': [1, '20200426043844', 1.0]}\n",
      "element added to heap: [1, 20200426043844, 1.0, 'Task2Job1']\n",
      "current heap: [[1, 20200426043844, 1.0, 'Task2Job1'], [2, 20200426043843, 9.0, 'Task1Job1'], [2, 20200426043843, 4.0, 'Task1Job2']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 3.0\n",
      "3-start_wait_time 0.0\n",
      "heap length: 3\n",
      "The updated heap: [[1, 20200426043844, 1.0, 'Task2Job1'], [2, 20200426043843, 9.0, 'Task1Job1'], [2, 20200426043843, 4.0, 'Task1Job2']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043844, 1.0, 'Task2Job1']\n",
      "worker 1 is free\n",
      "popped entry: [2, 20200426043843, 4.0, 'Task1Job2']\n",
      "worker 2 is free\n",
      "popped entry: [2, 20200426043843, 9.0, 'Task1Job1']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "3-curr_wait_time 3.0\n",
      "3-start_wait_time 2.009690046310425\n",
      "heap length: 3\n",
      "The updated heap: [[2, 20200426043844, 0.0, 'Task2Job1'], [3, 20200426043843, 8.0, 'Task1Job1'], [3, 20200426043843, 3.0, 'Task1Job2']]\n",
      "worker 0 is free\n",
      "popped entry: [2, 20200426043844, 0.0, 'Task2Job1']\n",
      "temptask finished: [2, 20200426043844, -1.0, 'Task2Job1']\n",
      "worker 1 is free\n",
      "popped entry: [3, 20200426043843, 3.0, 'Task1Job2']\n",
      "worker 2 is free\n",
      "popped entry: [3, 20200426043843, 8.0, 'Task1Job1']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "current_task_index 2\n",
      "Inside first while\n",
      "\n",
      "current task number 2\n",
      "1-curr_wait_time 3.0\n",
      "1-start_wait_time 0.0\n",
      "current_sgd_job 2\n",
      "sgd job: 2    2\n",
      "3    2\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 2\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "2     10000   2\n",
      "3      4000   2\n",
      "start index 2\n",
      "end index 3\n",
      "x 2\n",
      "curr_df   flow_size job\n",
      "2     10000   2\n",
      "3      4000   2\n",
      "time to completion 10.0\n",
      "listOfTasks[Task2Job2]=[1, '20200426043846', 10.0]\n",
      "x 3\n",
      "curr_df   flow_size job\n",
      "2     10000   2\n",
      "3      4000   2\n",
      "time to completion 4.0\n",
      "listOfTasks[Task2Job3]=[1, '20200426043846', 4.0]\n",
      " final listOfTasks {'Task2Job2': [1, '20200426043846', 10.0], 'Task2Job3': [1, '20200426043846', 4.0]}\n",
      "element added to heap: [1, 20200426043846, 10.0, 'Task2Job2']\n",
      "current heap: [[1, 20200426043846, 10.0, 'Task2Job2'], [4, 20200426043843, 7.0, 'Task1Job1'], [4, 20200426043843, 2.0, 'Task1Job2']]\n",
      "element added to heap: [1, 20200426043846, 4.0, 'Task2Job3']\n",
      "current heap: [[1, 20200426043846, 4.0, 'Task2Job3'], [1, 20200426043846, 10.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [4, 20200426043843, 7.0, 'Task1Job1']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 3.0\n",
      "3-start_wait_time 0.0\n",
      "heap length: 4\n",
      "The updated heap: [[1, 20200426043846, 4.0, 'Task2Job3'], [1, 20200426043846, 10.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [4, 20200426043843, 7.0, 'Task1Job1']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043846, 4.0, 'Task2Job3']\n",
      "worker 1 is free\n",
      "popped entry: [1, 20200426043846, 10.0, 'Task2Job2']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "3-curr_wait_time 3.0\n",
      "3-start_wait_time 2.019852876663208\n",
      "heap length: 4\n",
      "The updated heap: [[2, 20200426043846, 3.0, 'Task2Job3'], [2, 20200426043846, 9.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [4, 20200426043843, 7.0, 'Task1Job1']]\n",
      "worker 0 is free\n",
      "popped entry: [2, 20200426043846, 3.0, 'Task2Job3']\n",
      "worker 1 is free\n",
      "popped entry: [2, 20200426043846, 9.0, 'Task2Job2']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "current_task_index 3\n",
      "Inside first while\n",
      "\n",
      "current task number 3\n",
      "1-curr_wait_time 0.5\n",
      "1-start_wait_time 0.0\n",
      "current_page_rank_job 1\n",
      "page rank job 1    1\n",
      "2    1\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 2\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "1     10000   1\n",
      "2      1000   1\n",
      "start index 1\n",
      "end index 2\n",
      "x 1\n",
      "curr_df   flow_size job\n",
      "1     10000   1\n",
      "2      1000   1\n",
      "time to completion 10.0\n",
      "listOfTasks[Task3Job1]=[1, '20200426043848', 10.0]\n",
      "x 2\n",
      "curr_df   flow_size job\n",
      "1     10000   1\n",
      "2      1000   1\n",
      "time to completion 1.0\n",
      "listOfTasks[Task3Job2]=[1, '20200426043848', 1.0]\n",
      " final listOfTasks {'Task3Job1': [1, '20200426043848', 10.0], 'Task3Job2': [1, '20200426043848', 1.0]}\n",
      "element added to heap: [1, 20200426043848, 10.0, 'Task3Job1']\n",
      "current heap: [[1, 20200426043848, 10.0, 'Task3Job1'], [3, 20200426043846, 2.0, 'Task2Job3'], [4, 20200426043843, 2.0, 'Task1Job2'], [4, 20200426043843, 7.0, 'Task1Job1'], [3, 20200426043846, 8.0, 'Task2Job2']]\n",
      "element added to heap: [1, 20200426043848, 1.0, 'Task3Job2']\n",
      "current heap: [[1, 20200426043848, 1.0, 'Task3Job2'], [3, 20200426043846, 2.0, 'Task2Job3'], [1, 20200426043848, 10.0, 'Task3Job1'], [4, 20200426043843, 7.0, 'Task1Job1'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 0.5\n",
      "3-start_wait_time 0.0\n",
      "heap length: 6\n",
      "The updated heap: [[1, 20200426043848, 1.0, 'Task3Job2'], [3, 20200426043846, 2.0, 'Task2Job3'], [1, 20200426043848, 10.0, 'Task3Job1'], [4, 20200426043843, 7.0, 'Task1Job1'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043848, 1.0, 'Task3Job2']\n",
      "worker 1 is free\n",
      "popped entry: [1, 20200426043848, 10.0, 'Task3Job1']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "current_task_index 4\n",
      "Inside first while\n",
      "\n",
      "current task number 1\n",
      "1-curr_wait_time 1.0\n",
      "1-start_wait_time 0.0\n",
      "current_k_means_job 2\n",
      "kmeans curr_df   flow_size job\n",
      "3     10000   2\n",
      "k means job: 3    2\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 1\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "3     10000   2\n",
      "start index 3\n",
      "end index 3\n",
      "x 3\n",
      "curr_df   flow_size job\n",
      "3     10000   2\n",
      "time to completion 10.0\n",
      "listOfTasks[Task1Job3]=[1, '20200426043849', 10.0]\n",
      " final listOfTasks {'Task1Job3': [1, '20200426043849', 10.0]}\n",
      "element added to heap: [1, 20200426043849, 10.0, 'Task1Job3']\n",
      "current heap: [[1, 20200426043849, 10.0, 'Task1Job3'], [3, 20200426043846, 2.0, 'Task2Job3'], [2, 20200426043848, 0.0, 'Task3Job2'], [4, 20200426043843, 7.0, 'Task1Job1'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [2, 20200426043848, 9.0, 'Task3Job1']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 1.0\n",
      "3-start_wait_time 0.0\n",
      "heap length: 7\n",
      "The updated heap: [[1, 20200426043849, 10.0, 'Task1Job3'], [3, 20200426043846, 2.0, 'Task2Job3'], [2, 20200426043848, 0.0, 'Task3Job2'], [4, 20200426043843, 7.0, 'Task1Job1'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [2, 20200426043848, 9.0, 'Task3Job1']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043849, 10.0, 'Task1Job3']\n",
      "worker 1 is free\n",
      "popped entry: [2, 20200426043848, 0.0, 'Task3Job2']\n",
      "temptask finished: [2, 20200426043848, -1.0, 'Task3Job2']\n",
      "worker 2 is free\n",
      "popped entry: [2, 20200426043848, 9.0, 'Task3Job1']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "current_task_index 5\n",
      "Inside first while\n",
      "\n",
      "current task number 3\n",
      "1-curr_wait_time 0.0\n",
      "1-start_wait_time 0.0\n",
      "current_page_rank_job 2\n",
      "page rank job 3    2\n",
      "4    2\n",
      "Name: job, dtype: object\n",
      "number of new tasks: 2\n",
      "initial listOfTasks {}\n",
      "curr_df:   flow_size job\n",
      "3      6000   2\n",
      "4      6000   2\n",
      "start index 3\n",
      "end index 4\n",
      "x 3\n",
      "curr_df   flow_size job\n",
      "3      6000   2\n",
      "4      6000   2\n",
      "time to completion 6.0\n",
      "listOfTasks[Task3Job3]=[1, '20200426043850', 6.0]\n",
      "x 4\n",
      "curr_df   flow_size job\n",
      "3      6000   2\n",
      "4      6000   2\n",
      "time to completion 6.0\n",
      "listOfTasks[Task3Job4]=[1, '20200426043850', 6.0]\n",
      " final listOfTasks {'Task3Job3': [1, '20200426043850', 6.0], 'Task3Job4': [1, '20200426043850', 6.0]}\n",
      "element added to heap: [1, 20200426043850, 6.0, 'Task3Job3']\n",
      "current heap: [[1, 20200426043850, 6.0, 'Task3Job3'], [3, 20200426043846, 8.0, 'Task2Job2'], [2, 20200426043849, 9.0, 'Task1Job3'], [4, 20200426043843, 7.0, 'Task1Job1'], [4, 20200426043843, 2.0, 'Task1Job2'], [3, 20200426043848, 8.0, 'Task3Job1'], [3, 20200426043846, 2.0, 'Task2Job3']]\n",
      "element added to heap: [1, 20200426043850, 6.0, 'Task3Job4']\n",
      "current heap: [[1, 20200426043850, 6.0, 'Task3Job3'], [1, 20200426043850, 6.0, 'Task3Job4'], [2, 20200426043849, 9.0, 'Task1Job3'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [3, 20200426043848, 8.0, 'Task3Job1'], [3, 20200426043846, 2.0, 'Task2Job3'], [4, 20200426043843, 7.0, 'Task1Job1']]\n",
      "inside 2nd while\n",
      "3-curr_wait_time 0.0\n",
      "3-start_wait_time 0.0\n",
      "heap length: 8\n",
      "The updated heap: [[1, 20200426043850, 6.0, 'Task3Job3'], [1, 20200426043850, 6.0, 'Task3Job4'], [2, 20200426043849, 9.0, 'Task1Job3'], [3, 20200426043846, 8.0, 'Task2Job2'], [4, 20200426043843, 2.0, 'Task1Job2'], [3, 20200426043848, 8.0, 'Task3Job1'], [3, 20200426043846, 2.0, 'Task2Job3'], [4, 20200426043843, 7.0, 'Task1Job1']]\n",
      "worker 0 is free\n",
      "popped entry: [1, 20200426043850, 6.0, 'Task3Job3']\n",
      "worker 1 is free\n",
      "popped entry: [1, 20200426043850, 6.0, 'Task3Job4']\n",
      "worker 2 is free\n",
      "popped entry: [2, 20200426043849, 9.0, 'Task1Job3']\n",
      "going to sleep for a sec\n",
      "released all workers from work\n",
      "inside 2nd while\n",
      "Total time: 1454.09330868721\n"
     ]
    }
   ],
   "source": [
    "current_k_means_job=0\n",
    "current_sgd_job=0\n",
    "current_page_rank_job=0\n",
    "current_task_index=-1\n",
    "flagtime=False\n",
    "print(len(df_task.index))\n",
    "while current_task_index<len(df_task.index)-1: #repeats every second\n",
    "#while current_task_index<3:\n",
    "    allTasksFinished=False\n",
    "    current_task_index=current_task_index+1\n",
    "    #initial wait before you start running the first task\n",
    "    if current_task_index==0:\n",
    "        print(\"waiting for the first time\")\n",
    "        tempTime=float(df_task.time_of_task[current_task_index])/1000\n",
    "        time.sleep(tempTime)\n",
    "    flagtime=False    \n",
    "    print('current_task_index',current_task_index)\n",
    "    cur_task_num=df_task.task_number[current_task_index]\n",
    "    print(\"Inside first while\\n\")\n",
    "    print('current task number',cur_task_num)\n",
    "    if current_task_index==len(df_task.index)-1:\n",
    "        curr_wait_time=0.0\n",
    "    else:\n",
    "        curr_wait_time=float(df_task.time_of_task[current_task_index+1])/1000 #time to wait before releasing new entry in secs\n",
    "    #curr_wait_time=10.0\n",
    "    start_wait_time=0.0\n",
    "    starttime=time.time()\n",
    "    print('1-curr_wait_time', curr_wait_time)\n",
    "    print('1-start_wait_time', start_wait_time)\n",
    "    \n",
    "    \n",
    "    \n",
    "\n",
    "    if allTasksFinished==True:\n",
    "        print(\"All tasks finished\")\n",
    "        break\n",
    "        \n",
    "    \n",
    "    listOfTasks = {} \n",
    "    curr_df=None #represents the df which has info of only the current job of the given task\n",
    "\n",
    "    # now read all the tasks that belong to that file(cur_task_num) that is chosen\n",
    "    if cur_task_num==1:\n",
    "        current_k_means_job=current_k_means_job+1\n",
    "        print('current_k_means_job',current_k_means_job)\n",
    "        curr_df=df_k_means.loc[df_k_means['job'] == str(current_k_means_job)].copy()\n",
    "        print(\"kmeans curr_df\",curr_df)\n",
    "        print(\"k means job:\", curr_df.job)\n",
    "    elif cur_task_num==2:\n",
    "        current_sgd_job=current_sgd_job+1\n",
    "        print('current_sgd_job', current_sgd_job)\n",
    "        curr_df=df_sgd.loc[df_sgd['job'] == str(current_sgd_job)].copy()\n",
    "        print(\"sgd job:\", curr_df.job)\n",
    "    elif cur_task_num==3:\n",
    "        current_page_rank_job=current_page_rank_job+1\n",
    "        print('current_page_rank_job',current_page_rank_job)\n",
    "        curr_df=df_page_rank.loc[df_page_rank['job'] == str(current_page_rank_job)].copy()\n",
    "        print('page rank job',curr_df.job)\n",
    "\n",
    "    #print(curr_df)\n",
    "    #curr_df=curr_df.head(5)\n",
    "    curr_df=pd.DataFrame(curr_df)\n",
    "    numOfNewTasks=len(curr_df.index)\n",
    "    currLenDict=len(listOfTasks)\n",
    "    #when new entries enter the queue:\n",
    "    print(\"number of new tasks:\", numOfNewTasks)\n",
    "    print(\"initial listOfTasks\", listOfTasks)\n",
    "    \n",
    "    print(\"curr_df:\",curr_df)\n",
    "    start_index=curr_df.head(1).index[0]\n",
    "    end_index=curr_df.tail(1).index[0]\n",
    "    print('start index', start_index)\n",
    "    print('end index', end_index)\n",
    "    #for x in range(currLenDict+1,currLenDict+numOfNewTasks+1):\n",
    "    for x in range(start_index,numOfNewTasks+start_index):\n",
    "        key=\"Task\"+str(cur_task_num)+\"Job\"+str(x)\n",
    "        val=1 #priority at entry time\n",
    "        now = datetime.now() # time object -- now.strftime(\"%Y%m%d%H%M%S\") \n",
    "        print('x',x)\n",
    "        print('curr_df',curr_df)\n",
    "        timeToCompletion=float(curr_df.flow_size[x])/1000\n",
    "        print('time to completion', timeToCompletion)\n",
    "\n",
    "        listOfTasks[key]=[val,now.strftime(\"%Y%m%d%H%M%S\"),timeToCompletion]\n",
    "        print('listOfTasks['+str(key)+']='+str(listOfTasks[key]))\n",
    "\n",
    "    print(\" final listOfTasks\",listOfTasks)\n",
    "\n",
    "    #create a heap for the first time:\n",
    "    if flag==True:\n",
    "        print(\"creating heap for the first time\")\n",
    "        for keyTaskName in listOfTasks:\n",
    "            tempList=[]\n",
    "            tempList.append(listOfTasks[keyTaskName][0])      #priority\n",
    "            tempList.append(int(listOfTasks[keyTaskName][1])) #time of creation\n",
    "            tempList.append(listOfTasks[keyTaskName][2])      #time required to complete this task\n",
    "            tempList.append(keyTaskName)                      #task name\n",
    "#            print('tempList',tempList)\n",
    "            heapEntry.append(tempList)\n",
    "        print(\"list is:\", heapEntry)\n",
    "        heapq.heapify(heapEntry) \n",
    "        print(\"created heap for the first time\")\n",
    "        print(\"heap is:\", heapEntry)\n",
    "        flag=False\n",
    "    else:\n",
    "#        print(\" adding elements to heap\")\n",
    "        for keyTaskName in listOfTasks:\n",
    "            tempList=[]\n",
    "            tempList.append(listOfTasks[keyTaskName][0])      #priority\n",
    "            tempList.append(int(listOfTasks[keyTaskName][1])) #time of creation\n",
    "            tempList.append(listOfTasks[keyTaskName][2])      #time required to complete this task\n",
    "            tempList.append(keyTaskName)                      #task name\n",
    "            heapq.heappush(heapEntry,tempList)\n",
    "            print('element added to heap:', tempList)\n",
    "            print(\"current heap:\", heapEntry)\n",
    "\n",
    "    while True:\n",
    "        print(\"inside 2nd while\")\n",
    "        if len(heapEntry)>0: # while the heap is not empty i.e. there are tasks yet to finish\n",
    "            if start_wait_time>curr_wait_time:\n",
    "                break\n",
    "            print('3-curr_wait_time', curr_wait_time)\n",
    "            print('3-start_wait_time', start_wait_time)\n",
    "            print(\"heap length:\", len(heapEntry))\n",
    "            print(\"The updated heap:\", heapEntry)\n",
    "            if all(boolean==True for boolean in busyWorker):\n",
    "#                print(\"All workers busy\")\n",
    "                continue\n",
    "            tempSet=set()\n",
    "            for x in range(numOfWorkers):\n",
    "                      \n",
    "                if busyWorker[x]==False and len(heapEntry)>0:\n",
    "                    tempTask=heapq.heappop(heapEntry)\n",
    "                    if tempTask[3] in tempSet:\n",
    "                        heapq.heappush(heapEntry,tempTask)\n",
    "                        break\n",
    "                    else:\n",
    "                        print(\"worker\",x,\"is free\")\n",
    "                        busyWorker[x]=True\n",
    "                        print(\"popped entry:\", tempTask)\n",
    "                        tempTask[2]=tempTask[2]-1 #reduced remaining completion time by 1 sec\n",
    "                        tempSet.add(tempTask[3]) #adding the key in set\n",
    "                        if tempTask[2]<0:\n",
    "                            print(\"temptask finished:\", tempTask)\n",
    "                            continue # will not add that task in the heap again as it is finished\n",
    "                        else:\n",
    "\n",
    "                            tempTask[0]=tempTask[0]+1\n",
    "    #                        print(\"temptask remaining work:\", tempTask)\n",
    "                            heapq.heappush(heapEntry, tempTask)\n",
    "    #                        print(\"pushed\", tempTask, \"back into heap\")\n",
    "    #                        print(\"new heap:\", heapEntry)\n",
    "            print(\"going to sleep for a sec\")\n",
    "#            print(\"time.time()%1.0\", time.time()%1.0)\n",
    "#            print(\"starttime%1.0\", starttime%1.0)\n",
    "#            print(\"sleeping for\", 1.0-(time.time()-starttime)% 1.0)\n",
    "            time.sleep(1.0-(time.time()-starttime)%1.0) #wait for one second\n",
    "            if flagtime==False:\n",
    "                start_wait_time=float(str(time.time()-starttime))\n",
    "                #start_wait_time=float(str(time.time()))\n",
    "                flagtime=True\n",
    "            start_wait_time=start_wait_time+1\n",
    "#            print(\"going out of sleep\")\n",
    "            for x in range(numOfWorkers): #release all workers from work after every second\n",
    "                busyWorker[x]=False\n",
    "            print(\"released all workers from work\")\n",
    "            #Add break inner while loop here to get new entries in list every second.\n",
    "            #break\n",
    "\n",
    "        else:\n",
    "            print(\"Heap Empty: All tasks assigned till now finished\")\n",
    "            allTasksFinished=True\n",
    "            break\n",
    "    if start_wait_time<curr_wait_time:\n",
    "        print('2-curr_wait_time', curr_wait_time)\n",
    "        print('2-start_wait_time', start_wait_time)\n",
    "        print(\"Completed task, now sleeping for\", curr_wait_time-(time.time()-starttime)%curr_wait_time)\n",
    "        time.sleep(curr_wait_time-(time.time()-starttime)%curr_wait_time) #wait for one second\n",
    "#        print(\"Going out of sleep between adding new entries to \")\n",
    "totalTime=time.time()-intialTime\n",
    "print(\"Total time:\",totalTime)\n",
    "                                        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "Missing parentheses in call to 'print'. Did you mean print(\"Starting \" + self.name)? (<ipython-input-55-803c7ba956a3>, line 13)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-55-803c7ba956a3>\"\u001b[1;36m, line \u001b[1;32m13\u001b[0m\n\u001b[1;33m    print \"Starting \" + self.name\u001b[0m\n\u001b[1;37m          ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m Missing parentheses in call to 'print'. Did you mean print(\"Starting \" + self.name)?\n"
     ]
    }
   ],
   "source": [
    "# threading module\n",
    "\n",
    "\n",
    "exitFlag = 0\n",
    "\n",
    "class myThread (threading.Thread):\n",
    "   def __init__(self, threadID, name, counter):\n",
    "      threading.Thread.__init__(self)\n",
    "      self.threadID = threadID\n",
    "      self.name = name\n",
    "      self.counter = counter\n",
    "   def run(self):\n",
    "      print \"Starting \" + self.name\n",
    "      print_time(self.name, 5, self.counter)\n",
    "      print \"Exiting \" + self.name\n",
    "\n",
    "def print_time(threadName, counter, delay):\n",
    "   while counter:\n",
    "      if exitFlag:\n",
    "         threadName.exit()\n",
    "      time.sleep(delay)\n",
    "      print \"%s: %s\" % (threadName, time.ctime(time.time()))\n",
    "      counter -= 1\n",
    "\n",
    "# Create new threads\n",
    "thread1 = myThread(1, \"Thread-1\", 1)\n",
    "thread2 = myThread(2, \"Thread-2\", 2)\n",
    "\n",
    "# Start new Threads\n",
    "thread1.start()\n",
    "thread2.start()\n",
    "\n",
    "print \"Exiting Main Thread\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-15-ff47c9c660ad>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;36m  File \u001b[1;32m\"<ipython-input-15-ff47c9c660ad>\"\u001b[1;36m, line \u001b[1;32m1\u001b[0m\n\u001b[1;33m    pip3 instal xlrd\u001b[0m\n\u001b[1;37m         ^\u001b[0m\n\u001b[1;31mSyntaxError\u001b[0m\u001b[1;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "\n",
    "while True: #repeats every second\n",
    "    current_task_index=current_task_index+1\n",
    "    cur_task_num=df_task.task_number[current_task_index]\n",
    "    print(\"Inside first while\")\n",
    "    starttime=time.time()\n",
    "    if allTasksFinished==True:\n",
    "        totalTime=datetime.now()-initialStartTime\n",
    "        print(\"Total time:\", totalTime.strftime(\"%H-%M-%S\"))\n",
    "        print(\"All tasks finished\")\n",
    "        break\n",
    "\n",
    "    listOfTasks = {} \n",
    "    # now read all the tasks that belong to that file that is chosen\n",
    "\n",
    "    numOfNewTasks=randint(1,5)\n",
    "    currLenDict=len(listOfTasks)\n",
    "    # when new entries enter the queue:\n",
    "    print(\"number of new tasks:\", numOfNewTasks)\n",
    "    print(\" initial listOfTasks\",listOfTasks)\n",
    "\n",
    "    for x in range(currLenDict+1,currLenDict+numOfNewTasks+1):\n",
    "\n",
    "        key=\"Task\"+str(x)\n",
    "        val=1\n",
    "        now = datetime.now() # time object -- now.strftime(\"%Y%m%d%H%M%S\") \n",
    "        timeToCompletion=randint(1,10)\n",
    "        listOfTasks[key]=[val,now.strftime(\"%Y%m%d%H%M%S\"),timeToCompletion]\n",
    "\n",
    "    print(\" final listOfTasks\",listOfTasks)\n",
    "\n",
    "    #create a heap for the first time:\n",
    "    if flag==True:\n",
    "        print(\" creating heap for the fisrt time\")\n",
    "        for keyTaskName in listOfTasks:\n",
    "            tempList=[]\n",
    "            tempList.append(listOfTasks[keyTaskName][0])      #priority\n",
    "            tempList.append(int(listOfTasks[keyTaskName][1])) #time of creation\n",
    "            tempList.append(listOfTasks[keyTaskName][2])\t  #time required to complete this task\n",
    "            tempList.append(keyTaskName)                      #task name\n",
    "            heapEntry.append(tempList)\n",
    "        heapq.heapify(heapEntry) \n",
    "        print(\" created heap for the fisrt time\")\n",
    "        print(\"heap is:\", heapEntry)\n",
    "        flag=False\n",
    "    else:\n",
    "        print(\" adding elements to heap\")\n",
    "        for keyTaskName in listOfTasks:\n",
    "            tempList=[]\n",
    "            tempList.append(listOfTasks[keyTaskName][0])      #priority\n",
    "            tempList.append(int(listOfTasks[keyTaskName][1])) #time of creation\n",
    "            tempList.append(listOfTasks[keyTaskName][2])\t  #time required to complete this task\n",
    "            tempList.append(keyTaskName)                      #task name\n",
    "            heapq.heappush(heapEntry,tempList)\n",
    "            print('element added to heap:', tempList)\n",
    "            print(\"current heap:\", heapEntry)\n",
    "\n",
    "    while True:\n",
    "        print(\"inside 2nd while\")\n",
    "        if len(heapEntry)>0:\n",
    "            print(\"heap length:\", len(heapEntry))\n",
    "            if all(boolean==True for boolean in busyWorker):\n",
    "                print(\"All workers busy\")\n",
    "                continue\n",
    "\n",
    "            for x in range(numOfWorkers):\n",
    "                if busyWorker[x]==False and len(heapEntry)>0:\n",
    "                    print(\"worker\",x,\"is free\")\n",
    "                    busyWorker[x]=True\n",
    "                    tempTask=heapq.heappop(heapEntry)\n",
    "                    print(\"popped entry:\", tempTask)\n",
    "                    tempTask[2]=tempTask[2]-1 #reduced remaining completion time by 1 sec\n",
    "                    if tempTask[2]==0:\n",
    "                        print(\"temptask finished:\", tempTask)\n",
    "                        continue # will not add that task in the heap again as it is finished\n",
    "                    else:\n",
    "\n",
    "                        tempTask[0]=tempTask[0]+1\n",
    "                        print(\"temptask remaining work:\", tempTask)\n",
    "                        heapq.heappush(heapEntry, tempTask)\n",
    "                        print(\"pushed\", tempTask, \"back into heap\")\n",
    "                        print(\"new heap:\", heapEntry)\n",
    "            print(\"going to sleep for a sec\")\n",
    "            print(\"time.time()%1.0\", time.time()%1.0)\n",
    "            print(\"starttime%1.0\", starttime%1.0)\n",
    "            print(\"sleeping for\", 1.0-(time.time()-starttime)% 1.0)\n",
    "            time.sleep(1.0-(time.time()-starttime)%1.0) #wait for one second\n",
    "            print(\"going out of sleep\")\n",
    "            for x in range(numOfWorkers):\t#release all workers from work after every second\n",
    "                busyWorker[x]=False\n",
    "                print(\"released all workers from work\")\n",
    "            #Add break inner while loop here to get new entries in list every second.\n",
    "            #break;\t\n",
    "\n",
    "        else:\n",
    "            print(\"mark all tasks finished tag as true\")\n",
    "            allTasksFinished=True\n",
    "            break\n",
    "\n"
   ]
  },
  {
   "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.8.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}