Add runner changes to the grid workflow

Author: rajeeja

workflows/p1b1_grid/data/settings.json

@@ -1,9 +1,9 @@
 {
     "parameters":
           { 
-          	"epochs": [1],
-            "batch_size": [30],
-            "N1": [1000], 
-            "NE": [400]
+          	"epochs": [2, 4, 8 ],
+            "batch_size": [20, 40],
+            "N1": [1000, 2000], 
+            "NE": [500]
           }
 }

workflows/p1b1_grid/python/evaluateOne.py

@@ -1,7 +1,7 @@
 import sys
 import p1b1_runner
 import json, os
-
+import socket
 
 if (len(sys.argv) < 3):
 	print('requires arg1=param and arg2=filename')
@@ -12,7 +12,7 @@
 
 # print (parameterString)
 print ("filename is " + filename)
-
+print (socket.gethostname())
 
 integs = [int(x) for x in parameterString.split(',')]
 print (integs)
@@ -21,16 +21,25 @@
 hyper_parameter_map['framework'] = 'keras'
 hyper_parameter_map['batch_size'] = integs[1]
 hyper_parameter_map['dense'] = [integs[2], integs[3]] 
-hyper_parameter_map['save'] = os.environ['TURBINE_OUTPUT']+ "/"+'./output'+parameterString
-
+hyper_parameter_map['run_id'] = parameterString
+# hyper_parameter_map['instance_directory'] = os.environ['TURBINE_OUTPUT'] 
+hyper_parameter_map['save'] = os.environ['TURBINE_OUTPUT']+ "/output-"+os.environ['PMI_RANK']
+sys.argv = ['p1b1_runner']
 val_loss = p1b1_runner.run(hyper_parameter_map)
 print (val_loss)
+
+sfn = os.environ['TURBINE_OUTPUT']+ "/output-"+os.environ['PMI_RANK'] + "/procname-" + parameterString
+with open(sfn, 'w') as sfile:
+    sfile.write(socket.getfqdn())
+    proc_id = "-"+ str(os.getpid())
+    sfile.write(proc_id)
+
 # works around this error:
 # https://github.com/tensorflow/tensorflow/issues/3388
 from keras import backend as K
 K.clear_session()
 
-# writing the val loss to the output file
+# writing the val loss to the output file (result-*)
 with open(filename, 'w') as the_file:
     the_file.write(repr(val_loss))
 

workflows/p1b1_grid/python/p1b1_runner.py

@@ -7,39 +7,7 @@
 import json
 import os
 import p1b1
-import numpy as np
-
-DATA_TYPES = {type(np.float16): 'f16', type(np.float32): 'f32', type(np.float64): 'f64'}
-
-def write_params(params, hyper_parameter_map):
-    parent_dir =  hyper_parameter_map['instance_directory'] if 'instance_directory' in hyper_parameter_map else '.'
-    f = "{}/parameters_p1b1.txt".format(parent_dir)
-    with open(f, "w") as f_out:
-        f_out.write("[parameters]\n")
-        for k,v in params.items():
-            if type(v) in DATA_TYPES:
-                v = DATA_TYPES[type(v)]
-            if isinstance(v, basestring):
-                v = "'{}'".format(v)
-            f_out.write("{}={}\n".format(k, v))
-
-def is_numeric(val):
-    try:
-        float(val)
-        return True
-    except ValueError:
-        return False
-
-def format_params(hyper_parameter_map):
-    for k,v in hyper_parameter_map.items():
-        vals = str(v).split(" ")
-        if len(vals) > 1 and is_numeric(vals[0]):
-            # assume this should be a list
-            if "." in vals[0]:
-                hyper_parameter_map[k] = [float(x) for x in vals]
-            else:
-                hyper_parameter_map[k] = [int(x) for x in vals]
-
+import runner_utils
 
 def run(hyper_parameter_map):
     framework = hyper_parameter_map['framework']
@@ -56,17 +24,15 @@ def run(hyper_parameter_map):
         raise ValueError("Invalid framework: {}".format(framework))
 
     # params is python dictionary
-    sys.argv = ['fail here', '--epochs', '54321']
     params = pkg.initialize_parameters()
-    format_params(hyper_parameter_map)
+    runner_utils.format_params(hyper_parameter_map)
 
     for k,v in hyper_parameter_map.items():
         #if not k in params:
         #    raise Exception("Parameter '{}' not found in set of valid arguments".format(k))
         params[k] = v
 
     print(params)
-    write_params(params, hyper_parameter_map)
     history = pkg.run(params)
 
     if framework is 'keras':
@@ -81,27 +47,3 @@ def run(hyper_parameter_map):
     # use the last validation_loss as the value to minimize
     val_loss = history.history['val_loss']
     return val_loss[-1]
-
-def write_output(result, instance_directory):
-    with open('{}/result.txt'.format(instance_directory), 'w') as f_out:
-        f_out.write("{}\n".format(result))
-
-def init(param_file, instance_directory):
-    with open(param_file) as f_in:
-        hyper_parameter_map = json.load(f_in)
-
-    hyper_parameter_map['framework'] = 'keras'
-    hyper_parameter_map['save'] = '{}/output'.format(instance_directory)
-    hyper_parameter_map['instance_directory'] = instance_directory
-    
-    return hyper_parameter_map
-
-if __name__ == '__main__':
-    print('p1b1_runner main ' + str(argv))
-    param_file = sys.argv[1]
-    instance_directory = sys.argv[2]
-    hyper_parameter_map = init(param_file, instance_directory)
-    # clear sys.argv so that argparse doesn't object
-    sys.argv = ['p1b1_runner']
-    result = run(hyper_parameter_map)
-    write_output(result, instance_directory)

workflows/p1b1_grid/swift/run

@@ -18,7 +18,7 @@ export APP_HOME=$THIS
 
 PROJECT_ROOT=$APP_HOME/..
 
-export PYTHONPATH=$PYTHONPATH:$PROJECT_ROOT/python:$P1B1_DIR:$PYTHONPATH
+export PYTHONPATH=$PYTHONPATH:$PROJECT_ROOT/python:$P1B1_DIR:$PROJECT_ROOT/../common/python:$PYTHONPATH
 
 export EXPID=$1
 export TURBINE_OUTPUT=$APP_HOME/../experiments/$EXPID
@@ -29,13 +29,17 @@ export TURBINE_OUTPUT=$APP_HOME/../experiments/$EXPID
 # be ignored if MACHINE flag (see below) is not set
 export QUEUE=batch
 export WALLTIME=00:10:00
-export PPN=16
+export PPN=1
 export TURBINE_JOBNAME="${EXPID}_job"
 
+# Resident task workers and ranks
+export TURBINE_RESIDENT_WORK_WORKERS=1
+export RESIDENT_WORK_RANKS=$(( PROCS - 2 ))
+
 echo $PYTHONPATH
 
 ### set the desired number of processors
-PROCS=2
+PROCS=3
 ###
 
 # remove -l option for removing printing processors ranks