Full scheduler refactor (eager strace-based dependency checking, loop support, updated benchmark infra)

.github/workflows/tests.yaml

@@ -31,6 +31,10 @@ jobs:
       if: github.event.pull_request.draft == false
       steps:
       - uses: actions/checkout@v2 
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.11'
       - name: Running Correctness Tests
         run: |
           cd ..

.gitignore

@@ -1,3 +1,6 @@
 scripts/vars.sh
 *__pycache__*
 .DS_Store
+*~
+\#*\#
+.\#*

.gitmodules

@@ -1,7 +1,3 @@
-[submodule "deps/riker"]
-	path = deps/riker
-	url = https://github.com/angelhof/riker.git
-	branch = eric-custom-db-store
 [submodule "deps/pash"]
 	path = deps/pash
 	url = https://github.com/binpash/pash.git

parallel-orch/executor.py

@@ -8,31 +8,33 @@
 # and traces them with Riker. 
 # All commands are run inside an overlay sandbox.
 
-def async_run_and_trace_command_return_trace(command, node_id, latest_env_file, speculate_mode=False):
+def async_run_and_trace_command_return_trace(command, concrete_node_id, execution_id, pre_execution_env_file, speculate_mode=False):
     trace_file = util.ptempfile()
     stdout_file = util.ptempfile()
     stderr_file = util.ptempfile()
     post_execution_env_file = util.ptempfile()
-    logging.debug(f'Scheduler: Stdout file for: {node_id} is: {stdout_file}')
-    logging.debug(f'Scheduler: Stderr file for: {node_id} is: {stderr_file}')
-    logging.debug(f'Scheduler: Trace file for: {node_id}: {trace_file}')
-    process = async_run_and_trace_command_return_trace_in_sandbox(command, trace_file, node_id, stdout_file, stderr_file, latest_env_file, post_execution_env_file, speculate_mode)
-    return process, trace_file, stdout_file, stderr_file, post_execution_env_file
+    sandbox_dir, tmp_dir = util.create_sandbox()
+    logging.debug(f'Scheduler: Stdout file for: {concrete_node_id} is: {stdout_file}')
+    logging.debug(f'Scheduler: Stderr file for: {concrete_node_id} is: {stderr_file}')
+    logging.debug(f'Scheduler: Trace file for: {concrete_node_id}: {trace_file}')
+    process = async_run_and_trace_command_return_trace_in_sandbox(command, execution_id, trace_file, concrete_node_id, stdout_file, stderr_file, pre_execution_env_file, post_execution_env_file, sandbox_dir, tmp_dir, speculate_mode)
+    return process, trace_file, stdout_file, stderr_file, pre_execution_env_file, post_execution_env_file, sandbox_dir
 
-def async_run_and_trace_command_return_trace_in_sandbox_speculate(command, node_id, latest_env_file):
-    process, trace_file, stdout_file, stderr_file, post_execution_env_file = async_run_and_trace_command_return_trace(command, node_id, latest_env_file, speculate_mode=True)
-    return process, trace_file, stdout_file, stderr_file, post_execution_env_file
+def async_run_and_trace_command_return_trace_in_sandbox_speculate(command, execution_id, concrete_node_id, pre_execution_env_file):
+    process, trace_file, stdout_file, stderr_file, post_execution_env_file, sandbox_dir = async_run_and_trace_command_return_trace(command, execution_id, concrete_node_id, pre_execution_env_file, speculate_mode=True)
+    return process, trace_file, stdout_file, stderr_file, post_execution_env_file, sandbox_dir
 
-def async_run_and_trace_command_return_trace_in_sandbox(command, trace_file, node_id, stdout_file, stderr_file, latest_env_file, post_execution_env_file, speculate_mode=False):
+def async_run_and_trace_command_return_trace_in_sandbox(command, execution_id, trace_file, concrete_node_id, stdout_file, stderr_file, pre_execution_env_file, post_execution_env_file, sandbox_dir, tmp_dir, speculate_mode=False):
     ## Call Riker to execute the command
     run_script = f'{config.PASH_SPEC_TOP}/parallel-orch/run_command.sh'
-    args = ["/bin/bash", run_script, command, trace_file, stdout_file, latest_env_file]
+    args = ["/bin/bash", run_script, command, trace_file, stdout_file, pre_execution_env_file, sandbox_dir, tmp_dir]
     if speculate_mode:
         args.append("speculate")
     else:
         args.append("standard")
-    args.append(str(node_id))
+    args.append(str(concrete_node_id))
     args.append(post_execution_env_file)
+    args.append(str(execution_id))
     # Save output to temporary files to not saturate the memory
     logging.debug(args)
     process = subprocess.Popen(args, stdout=None, stderr=None)
@@ -56,7 +58,7 @@ def read_trace(sandbox_dir, trace_file):
         path = f"{sandbox_dir}/upperdir/{trace_file}"
     logging.debug(f'Reading trace from: {path}')
     with open(path) as f:
-        return f.readlines()
+        return f.read().split('\n')[:-1]
 
 def read_env_file(env_file, sandbox_dir=None):
     if sandbox_dir is None: