Adds throughput and chaining tests for C API and Seveless

.github/workflows/daily.yml

@@ -77,7 +77,7 @@ jobs:
           uses: actions/checkout@v3
         - name: Test Vine Serverless Mode
           run:  ./test-vine-serverless.sh
-  
+
   parrot-cvmfs-job:
     runs-on: ubuntu-20.04
     timeout-minutes: 10
@@ -96,6 +96,23 @@ jobs:
         - name: Test TopEFT + Work Queue
           run:  ./test-topeft.sh
 
+  vine-task-throughput:
+     runs-on: ubuntu-20.04
+     timeout-minutes: 20
+     steps:
+        - name: Checkout
+          uses: actions/checkout@v3
+        - name: Test Vine Throughput
+          run:  ./test-vine-task-throughput.sh
+
+  vine-throughput-capi-job:
+    runs-on: ubuntu-20.04
+    timeout-minutes: 20
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Test Vine Throughput Test C API
+        run:  ./test-vine-task-throughput-capi.sh
 # Removed plain coffea test 7/24/2023 to replace with coffea-dask-taskvine when ready.
 #  coffea-job:
 #    runs-on: ubuntu-20.04

test-vine-task-serverless.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+import ndcctools.taskvine as vine
+import time
+import sys
+
+def func():
+    return 
+
+def main():
+    print("Creating TaskVine manager...")
+    q = vine.Manager()
+    num_tasks = 1000
+    print("Creating Library from functions...")
+    function_lib = q.create_library_from_functions('test-library', func, add_env=False)
+    q.install_library(function_lib)
+
+    print("Starting worker factory for port {}...".format(q.port))
+    factory = vine.Factory("local",manager_host_port="localhost:{}".format(q.port))
+    factory.max_workers=1
+    factory.min_workers=1
+
+    with factory:
+        print("Submitting tasks...")
+        for i in range (num_tasks):
+            t = vine.FunctionCall('test-library', 'func')
+            task_id = q.submit(t)
+
+        print("Waiting for tasks to complete...")
+        start_timer = True
+        while not q.empty():
+            t = q.wait(5)
+            if start_timer:
+                start = time.time()
+                start_timer = False
+        end = time.time()
+        many = end - start
+
+        start = time.time()
+        for i in range(num_tasks):
+            while not q.empty():
+                result = q.wait(5)
+            t = vine.FunctionCall('test-library', 'func')
+            task_id = q.submit(t)
+        end = time.time()
+        one = end-start
+    throughput = num_tasks/many
+    chaining = num_tasks/one
+
+    print(f"\nThroughput was {throughput} tasks per second")
+    print(f"Chaining was {chaining} tasks per second")
+    print("all tasks complete!")
+    assert throughput >= 110
+    assert chaining >= 50
+
+if __name__ == '__main__':
+    main()

test-vine-task-throughput-capi.sh

@@ -0,0 +1,12 @@
+
+#!/bin/sh
+
+# Build the sources from github
+. ./install-github.sh
+
+# Compile C API test using environment 
+gcc test-vine-task-throughput.c -o test-vine-task-throughput -I $PREFIX/include/cctools/ -L $PREFIX/lib -ltaskvine -ldttools -lm -lcrypto -lssl -lz
+
+./test-vine-task-throughput & 
+
+vine_worker localhost 9123 --single-shot

test-vine-task-throughput.c

@@ -0,0 +1,78 @@
+#include "taskvine.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <time.h>
+#include <assert.h>
+int main(int argc, char *argv[])
+{
+	struct vine_manager *m;
+	int i;
+    int tasksC = 5000;
+
+	m = vine_create(VINE_DEFAULT_PORT);
+
+	if(!m) {
+		printf("couldn't create manager: %s\n", strerror(errno));
+		return 1;
+	}
+	printf("TaskVine listening on %d\n", vine_port(m));
+
+	for(i=0;i<tasksC;i++) {
+		struct vine_task *t = vine_task_create(":");
+		vine_task_set_cores(t, 1);
+
+		int task_id = vine_submit(m, t);
+
+	}
+	bool start_timer = true;
+	clock_t start = 0;
+	while(!vine_empty(m)) {
+        struct vine_task *t  = vine_wait(m, 5);
+		if (start_timer){
+			start = clock();
+			start_timer = false;
+		}
+        if(t) {
+            vine_task_delete(t);
+        }
+    }
+	clock_t end = clock();
+	double throughtput_time = ((double)(end - start)) / CLOCKS_PER_SEC;
+
+	double throughput = tasksC / throughtput_time;
+
+	start_timer = true;
+	start = 0;
+	for(i=0;i<tasksC;i++) {
+		struct vine_task *t = vine_task_create(":");
+		vine_task_set_cores(t, 1);
+		if (start_timer){
+			start = clock();
+			start_timer = false;
+		}
+		int task_id = vine_submit(m, t);
+        while(!vine_empty(m)) {
+            t  = vine_wait(m, 5);
+            if(t) {
+                vine_task_delete(t);
+            }
+        }
+	}
+	end = clock();
+	double chaining_time = ((double)(end - start)) / CLOCKS_PER_SEC;
+	double chaining = tasksC / chaining_time;
+
+	printf("Throughput was %f tasks per second\n", throughput);
+	assert(throughput > 1000);
+
+	printf("Chaining was %f tasks per second\n", chaining);
+	printf("all tasks complete!\n");
+	assert(chaining > 1000);
+	vine_delete(m);
+
+	return 0;
+}

test-vine-task-throughput.py

@@ -2,15 +2,15 @@
 
 import ndcctools.taskvine as vine
 import time
+import sys
 
-if __name__ == "__main__":
+def main():
 
     q = vine.Manager()
     print("listening on port", q.port)
     factory = vine.Factory("local",manager_host_port="localhost:{}".format(q.port))
     factory.max_workers=1
     factory.min_workers=1
-    factory.disk = 100
 
     num_tasks = 1000
 
@@ -29,11 +29,30 @@
                 start = time.time()
                 start_timer = False
 
-    end = time.time()
-    e = end-start
-    print(f"It took {e} seconds\n")
-    print(f"Throughput was {num_tasks/e} tasks per second")
-    print("all tasks complete!")
+        end = time.time()
+        many = end - start
+
+        start = time.time()
 
+        for i in range(num_tasks):
+            while not q.empty():
+                result = q.wait(5)
+            t = vine.Task(command=":")
+            task_id = q.submit(t)
+
+        print("waiting for tasks to complete...")
+        end = time.time()
+        one = end - start
+    throughput = num_tasks/many
+    chaining = num_tasks/one
+    print(f"\nThroughput was {throughput} tasks per second")
+    print(f"Chaining was {chaining} tasks per second")
+    print("all tasks complete!")
+    assert throughput >= 190
+    assert chaining >= 155
+
+if __name__ == '__main__':
+    main()
+
 # vim: set sts=4 sw=4 ts=4 expandtab ft=python:
 

test-vine-task-throughput.sh

@@ -6,5 +6,6 @@
 # Install conda development environment
 . ./install-github.sh
 
-# Run the serverless test case.
-python3 test-vine-task-throughput.py
+# Run the test case.
+python test-vine-task-serverless.py
+python test-vine-task-throughput.py