Merge pull request #202 from ICRAR/LIU-303

Liu 303 - Correcting Scheduler Algorithms

daliuge-translator/dlg/dropmake/pg_generator.py

@@ -28,38 +28,11 @@
 if __name__ == "__main__":
     __package__ = "dlg.dropmake"
 
-import collections
-import datetime
 import json
 import logging
 import string
 import time
-from collections import defaultdict
-from itertools import product
-from typing import ValuesView
-
-import networkx as nx
-import numpy as np
-
-from .scheduler import MySarkarScheduler, DAGUtil, MinNumPartsScheduler, PSOScheduler
-from .utils.bash_parameter import BashCommand
-from ..common import dropdict
-from ..common import Categories, DropType
-from .dm_utils import (
-    LG_APPREF,
-    getNodesKeyDict,
-    get_lg_ver_type,
-    convert_construct,
-    convert_fields,
-    convert_mkn,
-    getAppRefInputs,
-    LG_VER_EAGLE,
-    LG_VER_EAGLE_CONVERTED,
-)
-from .utils.bash_parameter import BashCommand
-from ..common import Categories
-from ..common import STORAGE_TYPES, APP_DROP_TYPES
-from ..common import dropdict
+
 from dlg.dropmake.lg import LG, GraphException
 from dlg.dropmake.pgt import PGT
 from dlg.dropmake.pgtp import MetisPGTP, MySarkarPGTP, MinNumPartsPGTP, PSOPGTP

daliuge-translator/dlg/dropmake/pgt.py

@@ -45,6 +45,10 @@ class GPGTNoNeedMergeException(GraphException):
     pass
 
 
+class GPGTException(GraphException):
+    pass
+
+
 class PGT(object):
     """
     A DROP representation of Physical Graph Template

daliuge-translator/dlg/dropmake/pgtp.py

@@ -26,7 +26,7 @@
 
 import networkx as nx
 
-from dlg.dropmake.pgt import PGT, GraphException
+from dlg.dropmake.pgt import PGT, GPGTException
 from dlg.dropmake.scheduler import (
     MySarkarScheduler,
     DAGUtil,
@@ -38,10 +38,6 @@
 logger = logging.getLogger(__name__)
 
 
-class GPGTException(GraphException):
-    pass
-
-
 class MetisPGTP(PGT):
     """
     DROP and GOJS representations of Physical Graph Template with Partitions

daliuge-translator/dlg/dropmake/scheduler.py

@@ -23,7 +23,6 @@
 import logging
 import os
 import platform
-import random
 import time
 from collections import defaultdict
 
@@ -235,8 +234,8 @@ def can_add(self, u, v, gu, gv):
         if len(self._dag.nodes()) == 0:
             return (True, False, False)
 
-        unew = u not in self._dag.node
-        vnew = v not in self._dag.node
+        unew = u not in self._dag.nodes
+        vnew = v not in self._dag.nodes
 
         if DEBUG:
             slow_max = DAGUtil.get_max_antichains(self._dag)
@@ -268,7 +267,7 @@ def can_add(self, u, v, gu, gv):
                             mydop, mydop_slow, err_msg
                         )
                     )
-        ret = False if mydop > self._ask_max_dop else True
+        ret = False if mydop > self._ask_max_dop.get('num_cpus', 1) else True
         if unew:
             self.remove(u)
         if vnew:
@@ -284,8 +283,8 @@ def add(self, u, v, gu, gv, sequential=False, global_dag=None):
         #     logger.debug("u = ", u, ", v = ", v, ", partition = ", self.partition_id)
         uw = gu["weight"]
         vw = gv["weight"]
-        unew = u not in self._dag.node
-        vnew = v not in self._dag.node
+        unew = u not in self._dag.nodes
+        vnew = v not in self._dag.nodes
         self._dag.add_node(u, weight=uw, num_cpus=gu["num_cpus"])
         self._dag.add_node(v, weight=vw, num_cpus=gv["num_cpus"])
         self._dag.add_edge(u, v)

daliuge-translator/dlg/dropmake/utils/antichains.py

@@ -81,7 +81,6 @@ def _get_pi_solution(split_graph):
     for ed in split_graph.edges(data=True):
         Cxy = int(ed[2].get("capacity", sys.maxsize))
         Axy = int(ed[2]["weight"])
-        logger.debug(f"Found capacity and weight: {Axy}, {Cxy}")
         if Axy == 0 and Cxy > 0:
             H.add_edge(ed[0], ed[1], capacity=Cxy, weight=Axy)
 

daliuge-translator/dlg/dropmake/web/lg_web.py

@@ -60,6 +60,7 @@
 from dlg.dropmake.lg import GraphException, load_lg
 from dlg.dropmake.pg_generator import unroll, partition
 from dlg.dropmake.pg_manager import PGManager
+from dlg.dropmake.pgt import GPGTException
 from dlg.dropmake.scheduler import SchedulerException
 from jsonschema import validate, ValidationError
 
@@ -524,7 +525,11 @@ def gen_pgt():
     try:
         lgt = prepare_lgt(lg_path(lg_name), rmode)
         # LG -> PGT
-        pgt = unroll_and_partition_with_params(lgt, query)
+        try:
+            pgt = unroll_and_partition_with_params(lgt, query)
+        except GPGTException as GE:
+            response.status = 500
+            return "{0}: logical graph {1} cannot be unrolled {2}".format(err_prefix, lg_name, GE)
 
         num_partitions = 0  # pgt._num_parts;
 
@@ -582,7 +587,12 @@ def gen_pgt_post():
 
         logical_graph = prepare_lgt(logical_graph, rmode)
         # LG -> PGT
-        pgt = unroll_and_partition_with_params(logical_graph, reqform)
+        try:
+            pgt = unroll_and_partition_with_params(logical_graph, reqform)
+        except GPGTException as GE:
+            response.status = 500
+            return "{0}: logical graph {1} cannot be unrolled {2}".format(err_prefix, lg_name, GE)
+
         par_algo = reqform.get("algo", "none")
         pgt_id = pg_mgr.add_pgt(pgt, lg_name)
 

daliuge-translator/test/dropmake/test_lgweb.py

@@ -146,7 +146,7 @@ def test_get_pgtjson(self):
         # good!
         c._get_json("/pgt_jsonbody?pgt_name=logical_graphs/chiles_simple1_pgt.graph")
 
-    def test_get_pgt_post(self):
+    def test_get_pgt_post(self, algo="metis"):
 
         c = RestClient("localhost", lgweb_port, timeout=10)
 
@@ -160,7 +160,7 @@ def test_get_pgt_post(self):
 
         # add 'correct' data to the form
         form_data = {
-            "algo": "metis",
+            "algo": algo,
             "lg_name": "metis.graph",
             "json_data": json_data,
             "num_islands": 0,
@@ -246,6 +246,26 @@ def test_loop_pgt_post(self):
         except RestClientException as e:
             self.fail(e)
 
+    def _test_translate_alg(self, algorithm):
+        self.test_get_pgt_post(algo=algorithm)
+
+    @unittest.skip("None translation is not an option in EAGLE and does not work.")
+    def test_none_translation(self):
+        self._test_translate_alg(algorithm='none')
+
+    def test_metis_translation(self):
+        self._test_translate_alg(algorithm='metis')
+
+    def test_sarkar_translation(self):
+        self._test_translate_alg(algorithm='mysarkar')
+
+    def test_min_num_parts_translation(self):
+        self._test_translate_alg(algorithm='min_num_parts')
+
+    def test_pso_translation(self):
+        self._test_translate_alg(algorithm='pso')
+
+
     def test_pg_viewerer(self):
 
         c = RestClient("localhost", lgweb_port, timeout=10)