50 execute frontier in overlay

.github/workflows/tests.yaml

@@ -19,7 +19,6 @@ on:
     - test/**
     - scripts/**
   workflow_dispatch:
-
 # Jobs section
 jobs:
     PaSh-Spec-Tests:
@@ -32,20 +31,21 @@ jobs:
       if: github.event.pull_request.draft == false
       steps:
       - uses: actions/checkout@v2 
-        with:
-          ref: ${{ github.event.pull_request.head.sha }}
       - name: Running Correctness Tests
         run: |
+          cd ..
+          cp -r dynamic-parallelizer ~
+          cd ~/dynamic-parallelizer
           sudo touch /.githubenv
           # install the system deps and pash the environment
           sudo -E bash scripts/install_deps_ubuntu20.sh
-          export PASH_TOP="$PWD/deps/pash"
-          export PATH=$PATH:$PASH_TOP
-          file ./deps/riker/debug/bin/rkr
           uname -a
           # run all the tests
           export DEBUG=100
-          bash ./test/test_orch.sh  
+          export PASH_TOP="$PWD/deps/pash"
+          export PATH=$PATH:$PASH_TOP
+          file ~/dynamic-parallelizer/deps/riker/debug/bin/rkr
+          bash ~/dynamic-parallelizer/test/test_orch.sh  
           # get the timer
           timer=$(LANG=en_us_88591; date)
           echo "VERSION<<EOF" >> $GITHUB_ENV
@@ -74,4 +74,4 @@ jobs:
       - name: Exit Code
         run: |
           # check if everything executed without errors
-          cd test && bash exit_code.sh
+          cd ~/dynamic-parallelizer/test && bash exit_code.sh

.gitmodules

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

parallel-orch/config.py

@@ -34,3 +34,5 @@ def log_root(msg, *args, **kwargs):
 SOCKET_BUF_SIZE = 8192
 
 SCHEDULER_SOCKET = os.getenv("PASH_SPEC_SCHEDULER_SOCKET")
+
+MAX_KILL_ATTEMPTS = 10  # Define a maximum number of kill attempts for each process in the partial program order

parallel-orch/executor.py

@@ -6,38 +6,37 @@
 
 # This module executes a sequence of commands 
 # and traces them with Riker. 
-# Commands [1:N] are run inside an overlay sandbox.
+# All commands are run inside an overlay sandbox.
 
-def async_run_and_trace_command_return_trace(command, node_id, sandbox_mode=False):
+def async_run_and_trace_command_return_trace(command, node_id, speculate_mode=False):
     trace_file = util.ptempfile()
-    ## KK 2023-04-24: @giorgo Is there a reason you used tempfile.NamedTemporaryFile and not util.ptempfile()?
-    stdout_file = tempfile.NamedTemporaryFile(dir=config.PASH_SPEC_TMP_PREFIX)
     stdout_file = util.ptempfile()
-    stderr_file = tempfile.NamedTemporaryFile(dir=config.PASH_SPEC_TMP_PREFIX)
+    stderr_file = util.ptempfile()
     variable_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: Output variable file for: {node_id} is: {variable_file}')
-    process = async_run_and_trace_command(command, trace_file, node_id, stdout_file, stderr_file, variable_file, sandbox_mode)
+    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, variable_file, speculate_mode)
     return process, trace_file, stdout_file, stderr_file, variable_file
 
-def async_run_and_trace_command_return_trace_in_sandbox(command, node_id):
-    process, trace_file, stdout_file, stderr_file, variable_file = async_run_and_trace_command_return_trace(command, node_id, sandbox_mode=True)
+def async_run_and_trace_command_return_trace_in_sandbox_speculate(command, node_id):
+    process, trace_file, stdout_file, stderr_file, variable_file = async_run_and_trace_command_return_trace(command, node_id, speculate_mode=True)
     return process, trace_file, stdout_file, stderr_file, variable_file
 
-def async_run_and_trace_command(command, trace_file, node_id, stdout_file, stderr_file, variable_file, sandbox_mode=False):
+def async_run_and_trace_command_return_trace_in_sandbox(command, trace_file, node_id, stdout_file, stderr_file, variable_file, 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, variable_file]
-    if sandbox_mode:
-        # print(" -- Sandbox mode")
-        args.append("sandbox")
+    if speculate_mode:
+        args.append("speculate")
     else:
-        # print(" -- Standard mode")
         args.append("standard")
     args.append(str(node_id))
     # Save output to temporary files to not saturate the memory
+    logging.debug(args)
     process = subprocess.Popen(args, stdout=None, stderr=None)
+
     # For debugging
     # process = subprocess.Popen(args)
     return process
@@ -54,8 +53,7 @@ def read_trace(sandbox_dir, trace_file):
     if sandbox_dir == "":
         path = trace_file
     else:
-        path = f"{sandbox_dir}upperdir/{trace_file}"
-
+        path = f"{sandbox_dir}/upperdir/{trace_file}"
     logging.debug(f'Reading trace from: {path}')
     with open(path) as f:
         return f.readlines()