Merge pull request #46 from jhunkeler/split-delivery-code

Add multiprocessing / Split delivery code

7 files changed, 161 insertions, 9 deletions

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 62f58a9..f4380e0 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -5,8 +5,8 @@ include_directories(
 find_program(BASH_PROGRAM bash)
 set(EXECUTABLE_OUTPUT_PATH ${PROJECT_BINARY_DIR}/tests)
 set(CTEST_BINARY_DIRECTORY ${PROJECT_BINARY_DIR}/tests)
-set(nix_gnu_cflags -Wno-error -Wno-unused-parameter -Wno-discarded-qualifiers)
-set(nix_clang_cflags -Wno-unused-parameter -Wno-incompatible-pointer-types-discards-qualifiers)
+set(nix_gnu_cflags -Wno-format-truncation -Wno-error -Wno-unused-parameter -Wno-unused-result -Wno-discarded-qualifiers)
+set(nix_clang_cflags -Wno-format-truncation -Wno-unused-parameter -Wno-unused-result -Wno-incompatible-pointer-types-discards-qualifiers)
 set(win_msvc_cflags /Wall)
 
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/data/generic.ini ${CMAKE_CURRENT_BINARY_DIR} COPYONLY)
diff --git a/tests/data/generic.ini b/tests/data/generic.ini
index c1e5c9c..7d77edd 100644
--- a/tests/data/generic.ini
+++ b/tests/data/generic.ini
@@ -17,6 +17,7 @@ installer_baseurl = https://github.com/conda-forge/miniforge/releases/download/{
 ;conda_packages =
 pip_packages =
     firewatch==0.0.4
+    tweakwcs==0.8.8
 
 
 [runtime]
@@ -25,8 +26,21 @@ PYTHONUNBUFFERED = 1
 
 [test:firewatch]
 repository = https://github.com/astroconda/firewatch
-script =
+script_setup =
     pip install -e '.'
+script =
+    firewatch -c conda-forge -p ${STASIS_CONDA_PLATFORM_SUBDIR} | grep -E ' python-[0-9]'
+
+
+[test:tweakwcs]
+repository = https://github.com/spacetelescope/tweakwcs
+script_setup =
+    pip install -e '.[test]'
+script =
+    pytest \
+        -r fEsx \
+        --basetemp="{{ func:basetemp_dir() }}" \
+        --junitxml="{{ func:junitxml_file() }}"
 
 
 [deploy:artifactory:delivery]
diff --git a/tests/rt_generic.sh b/tests/rt_generic.sh
index 6da953d..6e4454c 100644
--- a/tests/rt_generic.sh
+++ b/tests/rt_generic.sh
@@ -6,10 +6,16 @@ if [ -n "$GITHUB_TOKEN" ] && [ -z "$STASIS_GH_TOKEN"]; then
 else
     export STASIS_GH_TOKEN="anonymous"
 fi
+python_versions=(
+    3.10
+    3.11
+    3.12
+)
 
 topdir=$(pwd)
 
 ws="rt_workspace"
+rm -rf "$ws"
 mkdir -p "$ws"
 ws="$(realpath $ws)"
 
@@ -28,9 +34,12 @@ popd
 pushd "$ws"
     type -P stasis
     type -P stasis_indexer
+    retcode=0
 
-    stasis --no-docker --no-artifactory --unbuffered -v "$topdir"/generic.ini
-    retcode=$?
+    for py_version in "${python_versions[@]}"; do
+        stasis --python "$py_version" --no-docker --no-artifactory --unbuffered -v "$topdir"/generic.ini
+        retcode+=$?
+    done
 
     set +x
 
@@ -54,7 +63,7 @@ pushd "$ws"
         for cond in "${fail_on_main[@]}"; do
             if grep --color -H -n "$cond" "$x" >&2; then
                 echo "ERROR DETECTED IN $x!" >&2
-                retcode=2
+                retcode+=1
             fi
         done
     done
@@ -94,6 +103,8 @@ pushd "$ws"
     done
 popd
 
-rm -rf "$ws"
+if [ -z "$RT_KEEP_WORKSPACE" ]; then
+    rm -rf "$ws"
+fi
 
 exit $retcode
\ No newline at end of file
diff --git a/tests/test_conda.c b/tests/test_conda.c
index 72217fc..fc762e9 100644
--- a/tests/test_conda.c
+++ b/tests/test_conda.c
@@ -38,8 +38,8 @@ struct Delivery ctx;
 
 void test_conda_installation() {
     char *install_url = calloc(255, sizeof(install_url));
-    delivery_get_installer_url(&ctx, install_url);
-    delivery_get_installer(&ctx, install_url);
+    delivery_get_conda_installer_url(&ctx, install_url);
+    delivery_get_conda_installer(&ctx, install_url);
     delivery_install_conda(ctx.conda.installer_path, ctx.storage.conda_install_prefix);
     STASIS_ASSERT_FATAL(access(ctx.storage.conda_install_prefix, F_OK) == 0, "conda was not installed correctly");
     STASIS_ASSERT_FATAL(conda_activate(ctx.storage.conda_install_prefix, "base") == 0, "unable to activate base environment");
diff --git a/tests/test_ini.c b/tests/test_ini.c
index 2579e21..e4a7808 100644
--- a/tests/test_ini.c
+++ b/tests/test_ini.c
@@ -86,11 +86,13 @@ void test_ini_setval_getval() {
     STASIS_ASSERT(ini_getval(ini, "default", "a", INIVAL_TYPE_STR, render_mode, &val) == 0, "failed to get value");
     STASIS_ASSERT(strcmp(val.as_char_p, "a") != 0, "unexpected value loaded from modified variable");
     STASIS_ASSERT(strcmp(val.as_char_p, "changed") == 0, "unexpected value loaded from modified variable");
+    guard_free(val.as_char_p);
 
     STASIS_ASSERT(ini_setval(&ini, INI_SETVAL_APPEND, "default", "a", " twice") == 0, "failed to set value");
     STASIS_ASSERT(ini_getval(ini, "default", "a", INIVAL_TYPE_STR, render_mode, &val) == 0, "failed to get value");
     STASIS_ASSERT(strcmp(val.as_char_p, "changed") != 0, "unexpected value loaded from modified variable");
     STASIS_ASSERT(strcmp(val.as_char_p, "changed twice") == 0, "unexpected value loaded from modified variable");
+    guard_free(val.as_char_p);
     ini_free(&ini);
     remove(filename);
 }
diff --git a/tests/test_multiprocessing.c b/tests/test_multiprocessing.c
new file mode 100644
index 0000000..cd037d0
--- /dev/null
+++ b/tests/test_multiprocessing.c
@@ -0,0 +1,123 @@
+#include "testing.h"
+
+static struct MultiProcessingPool *pool;
+char *commands[] = {
+        "true",
+        "uname -a",
+        "/bin/echo hello world",
+};
+
+void test_mp_pool_init() {
+    STASIS_ASSERT((pool = mp_pool_init("mypool", "mplogs")) != NULL, "Pool initialization failed");
+    STASIS_ASSERT_FATAL(pool != NULL, "Should not be NULL");
+    STASIS_ASSERT(pool->num_alloc == MP_POOL_TASK_MAX, "Wrong number of default records");
+    STASIS_ASSERT(pool->num_used == 0, "Wrong number of used records");
+    STASIS_ASSERT(strcmp(pool->log_root, "mplogs") == 0, "Wrong log root directory");
+    STASIS_ASSERT(strcmp(pool->ident, "mypool") == 0, "Wrong identity");
+
+    int data_bad_total = 0;
+    for (size_t i = 0; i < pool->num_alloc; i++) {
+        int data_bad = 0;
+        struct MultiProcessingTask *task = &pool->task[i];
+
+        data_bad += task->status == 0 ? 0 : 1;
+        data_bad += task->pid == 0 ? 0 : 1;
+        data_bad += task->parent_pid == 0 ? 0 : 1;
+        data_bad += task->signaled_by == 0 ? 0 : 1;
+        data_bad += task->time_data.t_start.tv_nsec == 0 ? 0 : 1;
+        data_bad += task->time_data.t_start.tv_sec == 0 ? 0 : 1;
+        data_bad += task->time_data.t_stop.tv_nsec == 0 ? 0 : 1;
+        data_bad += task->time_data.t_stop.tv_sec == 0 ? 0 : 1;
+        data_bad += (int) strlen(task->ident) == 0 ? 0 : 1;
+        data_bad += (int) strlen(task->parent_script) == 0 ? 0 : 1;
+        if (data_bad) {
+            SYSERROR("%s.task[%zu] has garbage values!", pool->ident, i);
+            SYSERROR("  ident: %s", task->ident);
+            SYSERROR("  status: %d", task->status);
+            SYSERROR("  pid: %d", task->pid);
+            SYSERROR("  parent_pid: %d", task->parent_pid);
+            SYSERROR("  signaled_by: %d", task->signaled_by);
+            SYSERROR("  t_start.tv_nsec: %ld", task->time_data.t_start.tv_nsec);
+            SYSERROR("  t_start.tv_sec: %ld", task->time_data.t_start.tv_sec);
+            SYSERROR("  t_stop.tv_nsec: %ld", task->time_data.t_stop.tv_nsec);
+            SYSERROR("  t_stop.tv_sec: %ld", task->time_data.t_stop.tv_sec);
+            data_bad_total++;
+        }
+    }
+    STASIS_ASSERT(data_bad_total == 0, "Task array is not pristine");
+    mp_pool_free(&pool);
+}
+
+void test_mp_task() {
+    pool = mp_pool_init("mypool", "mplogs");
+
+    if (pool) {
+        for (size_t i = 0; i < sizeof(commands) / sizeof(*commands); i++) {
+            struct MultiProcessingTask *task;
+            char task_name[100] = {0};
+            sprintf(task_name, "mytask%zu", i);
+            STASIS_ASSERT_FATAL((task = mp_pool_task(pool, task_name, NULL, commands[i])) != NULL, "Task should not be NULL");
+            STASIS_ASSERT(task->pid == MP_POOL_PID_UNUSED, "PID should be non-zero at this point");
+            STASIS_ASSERT(task->parent_pid == MP_POOL_PID_UNUSED, "Parent PID should be non-zero");
+            STASIS_ASSERT(task->status == -1, "Status should be -1 (not started yet)");
+            STASIS_ASSERT(strcmp(task->ident, task_name) == 0, "Wrong task identity");
+            STASIS_ASSERT(strstr(task->log_file, pool->log_root) != NULL, "Log file path must be in log_root");
+        }
+    }
+}
+
+void test_mp_pool_join() {
+    STASIS_ASSERT(mp_pool_join(pool, get_cpu_count(), 0) == 0, "Pool tasks should have not have failed");
+    for (size_t i = 0; i < pool->num_used; i++) {
+        struct MultiProcessingTask *task = &pool->task[i];
+        STASIS_ASSERT(task->pid == MP_POOL_PID_UNUSED, "Task should be marked as unused");
+        STASIS_ASSERT(task->status == 0, "Task status should be zero (success)");
+    }
+}
+
+void test_mp_pool_free() {
+    mp_pool_free(&pool);
+    STASIS_ASSERT(pool == NULL, "Should be NULL");
+}
+
+void test_mp_pool_workflow() {
+    struct testcase {
+        const char *input_cmd;
+        int input_join_flags;
+        int expected_result;
+        int expected_status;
+        int expected_signal;
+    };
+    struct testcase tc[] = {
+        {.input_cmd = "true && kill $$", .input_join_flags = 0, .expected_result = 1, .expected_status = 0, .expected_signal = SIGTERM},
+        {.input_cmd = "false || kill $$", .input_join_flags = 0, .expected_result = 1, .expected_status = 0, .expected_signal = SIGTERM},
+        {.input_cmd = "true", .input_join_flags = 0,.expected_result = 0, .expected_status = 0, .expected_signal = 0},
+        {.input_cmd = "false", .input_join_flags = 0, .expected_result = 1, .expected_status = 1, .expected_signal = 0},
+    };
+    for (size_t i = 0; i < sizeof(tc) / sizeof(*tc); i++) {
+        struct testcase *test = &tc[i];
+        struct MultiProcessingPool *p;
+        struct MultiProcessingTask *task;
+        STASIS_ASSERT((p = mp_pool_init("workflow", "mplogs")) != NULL, "Failed to initialize pool");
+        STASIS_ASSERT((task = mp_pool_task(p, "task", NULL, (char *) test->input_cmd)) != NULL, "Failed to queue task");
+        STASIS_ASSERT(mp_pool_join(p, get_cpu_count(), test->input_join_flags) == test->expected_result, "Unexpected result");
+        STASIS_ASSERT(task->status == test->expected_status, "Unexpected status");
+        STASIS_ASSERT(task->signaled_by == test->expected_signal, "Unexpected signal");
+        STASIS_ASSERT(task->pid == MP_POOL_PID_UNUSED, "Unexpected PID. Should be marked UNUSED.");
+        mp_pool_show_summary(p);
+        mp_pool_free(&p);
+    }
+}
+
+int main(int argc, char *argv[]) {
+    STASIS_TEST_BEGIN_MAIN();
+    STASIS_TEST_FUNC *tests[] = {
+        test_mp_pool_init,
+        test_mp_task,
+        test_mp_pool_join,
+        test_mp_pool_free,
+        test_mp_pool_workflow,
+    };
+    STASIS_TEST_RUN(tests);
+    STASIS_TEST_END_MAIN();
+}
\ No newline at end of file
diff --git a/tests/test_str.c b/tests/test_str.c
index 85c3b78..f8aa1e9 100644
--- a/tests/test_str.c
+++ b/tests/test_str.c
@@ -79,6 +79,7 @@ void test_strdup_array_and_strcmp_array() {
     for (size_t outer = 0; outer < sizeof(tc) / sizeof(*tc); outer++) {
         char **result = strdup_array((char **) tc[outer].data);
         STASIS_ASSERT(strcmp_array((const char **) result, tc[outer].expected) == 0, "array members were different");
+        GENERIC_ARRAY_FREE(result);
     }
 
     const struct testcase tc_bad[] = {
@@ -94,6 +95,7 @@ void test_strdup_array_and_strcmp_array() {
     for (size_t outer = 0; outer < sizeof(tc_bad) / sizeof(*tc_bad); outer++) {
         char **result = strdup_array((char **) tc_bad[outer].data);
         STASIS_ASSERT(strcmp_array((const char **) result, tc_bad[outer].expected) != 0, "array members were identical");
+        GENERIC_ARRAY_FREE(result);
     }
 }