qsub test added

Author: stephenoffer

.idea/workspace.xml

@@ -14,16 +14,20 @@
 from dkeras.utils.sys_functions import get_port, get_addr
 
 
+def _cmd(cmd):
+    return subprocess.check_output(cmd.split(' ')).decode()
+
+
 def _which_qsub():
-    return subprocess.check_output(['which', 'qsub']).decode().replace('\n', ' ')
+    return subprocess.check_output(['which', 'qsub']).decode().replace('\n', '')
 
 
 def _which_qstat():
-    return subprocess.check_output(['which', 'qstat']).decode().replace('\n', ' ')
+    return subprocess.check_output(['which', 'qstat']).decode().replace('\n', '')
 
 
 def _which_qdel():
-    return subprocess.check_output(['which', 'qdel']).decode().replace('\n', ' ')
+    return subprocess.check_output(['which', 'qdel']).decode().replace('\n', '')
 
 
 create_worker_script = """
@@ -42,19 +46,19 @@ def wait_for_workers(n_workers, timeout=300):
     print("Waiting for {} workers".format(n_workers))
     while True:
         n_nodes = ray.get(_get_n_nodes.remote())
-        if len(n_nodes) >= n_workers:
+        if n_nodes >= n_workers:
             return True
         if (start_time - time.time() >= timeout):
             return False
 
 
 def rm_existing_workers(qstat_path='qstat', qdel_path='qdel'):
-    cmd = "{} | grep worker_script | cut - d ' ' - f1 | xargs {}".format(
+    cmd = "{} | grep worker_script | cut - d ' ' -f1 | xargs {}".format(
         qstat_path, qdel_path)
     os.system(cmd)
 
 
-def init_pbs_ray(n_workers=3, rm_existing=True, iface_name='eno1', worker_time=3600):
+def init_pbs_ray(n_workers=3, rm_existing=True, iface_name='eno1', worker_time=3600, verbose=True):
     """
 
     :param n_workers:
@@ -65,25 +69,28 @@ def init_pbs_ray(n_workers=3, rm_existing=True, iface_name='eno1', worker_time=3
     """
     if ray.is_initialized():
         if rm_existing:
-            ray.shutdown()
+            _cmd('ray stop')
 
     qsub_path = _which_qsub()
     qstat_path = _which_qstat()
     qdel_path = _which_qdel()
 
-    rm_existing_workers(qstat_path=qstat_path, qdel_path=qdel_path)
+    # rm_existing_workers(qstat_path=qstat_path, qdel_path=qdel_path)
+    rm_existing_workers()
 
     addresses = get_addr('eno1')
     addr = addresses[0]
 
-    print("Address: ", addr)
+    if verbose:
+        print("Address: ", addr)
     if addr == 'No IP addr':
         raise Exception("Address not found for {}".format(iface_name))
 
     port = get_port()[1]
-    print("Port ", port)
-    print("ray start --head --redis-port={}".format(port))
-    os.system('ray start --head --redis-port={}'.format(port))
+    if verbose:
+        print("Port ", port)
+        print("ray start --head --redis-port={}".format(port))
+    _cmd('ray start --head --redis-port={}'.format(port))
 
     temp_dir = 'temp_{}'.format('_'.join(str(time.time()).split('.')))
     if not os.path.exists(temp_dir):
@@ -93,28 +100,29 @@ def init_pbs_ray(n_workers=3, rm_existing=True, iface_name='eno1', worker_time=3
     worker_script = create_worker_script.format(addr, port, worker_time)
     with open(worker_file, 'w') as f:
         f.write(worker_script)
-    print("Worker file ", worker_file)
+
+    if verbose:
+        print("Worker file ", worker_file)
 
     qsub_pids = []
     for i in range(n_workers):
-        print("{} {}".format(qsub_path, worker_file))
-        print(list(qsub_path))
-        print(list(worker_file))
+        if verbose:
+            print("{} {}".format(qsub_path, worker_file))
+            print(list(qsub_path))
+            print(list(worker_file))
 
-        qsub_pid = subprocess.check_output([qsub_path, '-l', 'nodes=1:ppn=2', worker_file])
+        qsub_pid = subprocess.check_output([qsub_path, '-lselect=1', '-lplace=excl', worker_file])
         qsub_pid = qsub_pid.decode()[:-1].split('.')[0]
         qsub_pids.append(qsub_pid)
 
-        # os.system('{} -l nodes=1:ppn=2 {}'.format(qsub_path, worker_file))
-    # print("{} {}".format(qsub_path, worker_file))
-    print("{}:{}".format(addr, port))
     ray.init(redis_address='{}:{}'.format(addr, port))
     print("Ray initialized")
     return wait_for_workers(n_workers + 1), qsub_pids
 
 
 def main():
     init_pbs_ray()
+    print(ray.nodes())
 
 
 if __name__ == "__main__":

dkeras/utils/qsub_functions.py

@@ -0,0 +1,23 @@
+from tensorflow.keras.applications import ResNet50
+from dkeras import dKeras
+from dkeras.utils.qsub_functions import init_pbs_ray
+import numpy as np
+import time
+import ray
+
+init_pbs_ray()
+print(ray.nodes())
+
+data = np.random.uniform(-1, 1, (10000, 224, 224, 3))
+
+start_time = time.time()
+model = dKeras(ResNet50, init_ray=False, wait_for_workers=True, n_workers=500)
+elapsed = time.time() - start_time
+
+print("Workers initialized after {}".format(elapsed))
+
+start_time = time.time()
+preds = model.predict(data)
+elapsed = time.time() - start_time
+
+print("Preds after {}".format(elapsed))