add tests, initial protobuf and rabbitmq support, ...

Makefile

@@ -1,19 +1,44 @@
-.PHONY: all clean install uninstall
+.PHONY: all clean install uninstall test
 
 PROTOBUF=/usr/include/
 STARTER=starter
 GEN=gen
 SRCS=$(STARTER)/cgroup_prepare.c $(STARTER)/ns_exec.c $(STARTER)/util.c $(STARTER)/starter.c 
+RUNS=runs
+PROTO_OUT=$(GEN)/google/protobuf/empty.pb-c.c $(GEN)/google/protobuf/empty.pb-c.h $(GEN)/google/protobuf/timestamp.pb-c.c $(GEN)/google/protobuf/timestamp.pb-c.h $(GEN)/google/protobuf/duration.pb-c.c $(GEN)/google/protobuf/duration.pb-c.h gen/runner/v1/runner.pb-c.c gen/runner/v1/runner.pb-c.c
 
-all: proto $(STARTER)/alpine-make-rootfs $(STARTER)/minrootfs $(STARTER)/starter
+all: $(STARTER)/alpine-make-rootfs $(STARTER)/minrootfs $(STARTER)/starter transport/transport $(RUNS)
-proto: /usr/include/google/protobuf/empty.proto proto/runner/v1/runner.proto
+
+$(GEN):
 	mkdir -p $(GEN)
-	protoc --c_out=$(GEN) runner/v1/runner.proto -I proto
+
+$(GEN)/google/protobuf/empty.pb-c.c $(GEN)/google/protobuf/empty.pb-c.h: $(PROTOBUF)/google/protobuf/empty.proto $(GEN)
 	protoc --c_out=$(GEN) google/protobuf/empty.proto -I $(PROTOBUF)
+
+$(GEN)/google/protobuf/timestamp.pb-c.c $(GEN)/google/protobuf/timestamp.pb-c.h: $(PROTOBUF)/google/protobuf/timestamp.proto $(GEN)
+	protoc --c_out=$(GEN) google/protobuf/timestamp.proto -I $(PROTOBUF)
+
+$(GEN)/google/protobuf/duration.pb-c.c $(GEN)/google/protobuf/duration.pb-c.h: $(PROTOBUF)/google/protobuf/duration.proto $(GEN)
+	protoc --c_out=$(GEN) google/protobuf/duration.proto -I $(PROTOBUF)
+
+gen/runner/v1/runner.pb-c.c gen/runner/v1/runner.pb-c.h: proto/runner/v1/runner.proto $(GEN)
+	protoc --c_out=$(GEN) runner/v1/runner.proto -I proto
+
+
 $(STARTER)/alpine-make-rootfs:
 	git clone https://github.com/alpinelinux/alpine-make-rootfs starter/alpine-make-rootfs
+
 $(STARTER)/minrootfs: $(STARTER)/create_rootfs.sh
 	$(STARTER)/create_rootfs.sh
-$(STARTER)/starter: proto $(STARTER)/starter.c $(STARTER)/cgroup_prepare.c $(STARTER)/ns_exec.c Makefile
+
-	$(CC) -o $(STARTER)/starter $(SRCS) -g -fsanitize=address -fsanitize=leak -fno-omit-frame-pointer -lrabbitmq -lprotobuf-c -I gen
+$(STARTER)/starter: $(PROTO) $(STARTER)/starter.c $(STARTER)/cgroup_prepare.c $(STARTER)/ns_exec.c Makefile
+	#$(CC) -shared -o $(STARTER)/starter $(SRCS) -g -fsanitize=address -fsanitize=leak -fno-omit-frame-pointer -lrabbitmq -lprotobuf-c -I gen
+	$(CC) -fPIC -shared -o $(STARTER)/starter $(SRCS) -lrabbitmq -lprotobuf-c -I gen
 	#$(CC) $(STARTER)/starter.c -o $(STARTER)/starter
+
+transport/transport: $(PROTO) transport/utils.c transport/transport.c transport/transport.h
+	$(CC) -o $@ gen/runner/v1/runner.pb-c.c transport/utils.c transport/transport.c -I . -I gen -lrabbitmq -lprotobuf-c -g -fsanitize=address -fsanitize=leak -fno-omit-frame-pointer
+
+$(RUNS):
+	mkdir $(RUNS)
+include tests/Makefile

resource_handler/resource_handler.c

@@ -0,0 +1,43 @@
+#include "resource_handler/resource_handler.h"
+
+int extract_numbers(char* str, int* res) {
+    bool is_digit=false,last_digit=false;
+    int cur=0;
+    int begin=-1;
+    int* old_res=res;
+    bool go=true;
+    while(go) {
+        go=(*str)!=0;
+        is_digit = *str <= '9' && *str >= '0';
+        if(is_digit) {
+            cur *= 10;
+            cur += *str - '0';
+        }
+        else if(last_digit) {
+            if(*str == '-') {
+                begin = cur;
+            } else if(begin != -1) {
+                for(int core = begin; core <= cur; core++) {
+                    *res = core;
+                    res++;
+                }
+                begin = -1;
+            } else {
+                *res = cur;
+                res++;
+                begin = -1;
+            }
+            cur = 0;
+        }
+        str++;
+        last_digit = is_digit;
+    }
+    return res - old_res;
+}
+
+int get_isolated_cores(int** res) {
+    char buf[MAX_OPTION];
+    read(open("/sys/devices/system/cpu/isolated", O_RDONLY),buf,MAX_OPTION);
+    *res=malloc(sizeof(int)*MAX_CORES);
+    return extract_numbers(buf,*res);
+}

resource_handler/resource_handler.h

@@ -0,0 +1,10 @@
+#include <stdio.h>
+#include <stdbool.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#define MAX_CORES 128
+#define MAX_OPTION 4096
+
+int extract_numbers(char* str, int* res);
+int get_isolated_cores(int** res);

starter/starter.c

@@ -25,26 +25,6 @@ static int killafter(void *arg) {
 static char nmstack[STACK_SIZE];
 static char killstack[STACK_SIZE];
 
-static void parse_args(int argc, char **argv, struct params *params, struct limits *limits){
-    if (argc < 7) {
-        puts("usage:\n starter <max processes number> <core_id> <memory_amount> <time limit in ms> <shared folder> <command, arg1,arg2,...>");
-        exit(0);
-    }
-    argc--; argv++;
-    limits->processes = atoi(argv[0]);
-    argc--; argv++;
-    limits->core = atoi(argv[0]);
-    argc--; argv++;
-    limits->memory = atoi(argv[0]);
-    argc--; argv++;
-    limits->time = atoi(argv[0]);
-    argc--; argv++;
-    params->shared_folder = argv[0];
-    argc--; argv++;
-
-    params->argv = argv;
-}
-
 // setup user namespace
 static void prepare_userns(int pid) {
     char path[100];
@@ -68,26 +48,16 @@ static void prepare_userns(int pid) {
     write_file(path, line);
 }
 
-static void get_real_path(char* path, char* call_str) {
+int starter(char* working_path, struct limits limits, struct params params) {
-    if (realpath (call_str, path) == 0) die("unable to resolve real path: %m");// get absolute path to executable
-    for(int i = strlen(path); i > 0 && path[i] != '/';i--) path[i]=0;// cut filename to get directory name
-}
-
-int main(int argc,char** argv) {
     if(setuid(0)) die("must be run as root");
     if(setgid(0)) die("must be run as root");
     // get binary path
-    char real_path[PATH_MAX];
+    char* real_path=working_path;
-    get_real_path(real_path, argv[0]);
+    //get_real_path(real_path, argv[0]);
     if(chdir(real_path)) die("unable to chdir to binary path: %m");
     // set random seed
     srand(time(NULL));
     // setup parameters
-    struct params params;
-    memset(&params, 0, sizeof(struct params));
-    struct limits limits;
-    memset(&limits, 0, sizeof(struct limits));
-    parse_args(argc, argv, &params, &limits);
     prepare_cgroup(&limits);
     if (pipe(params.fd) < 0) die("can't open pipe: %m");// a pipe to report readiness
     int clone_flags = SIGCHLD | CLONE_NEWUTS | CLONE_NEWUSER | CLONE_NEWNS | CLONE_NEWPID | CLONE_NEWNET | CLONE_NEWIPC | CLONE_NEWCGROUP;

tests/Makefile

@@ -0,0 +1,11 @@
+.PHONY: test_all
+test: tests/resource_handler_test tests/transport_test
+
+tests/resource_handler_test: resource_handler/resource_handler.h resource_handler/resource_handler.c tests/resource_handler_test.c
+	$(CC) -o $@ resource_handler/resource_handler.c tests/resource_handler_test.c -I . -g -fsanitize=address -fsanitize=leak -fno-omit-frame-pointer
+
+tests/transport_test: transport/transport.c transport/transport.h transport/utils.c tests/transport_test.c
+	$(CC) -o $@ gen/runner/v1/runner.pb-c.c tests/transport_test.c -I . -I gen -g -fsanitize=address -fsanitize=leak -fno-omit-frame-pointer -lrabbitmq -lprotobuf-c
+
+test_all:
+	echo 1

tests/resource_handler_test.c

@@ -0,0 +1,38 @@
+#include <string.h>
+#include <stdio.h>
+#include "resource_handler/resource_handler.h"
+
+void check_extract_numbers(char *input, int correct_len, int *correct_buf) {
+    int test_buf[MAX_CORES];
+    int test_len = extract_numbers(input, test_buf);
+    //int correct_buf[] = {1,2,3,4,5,7,8};
+    //int correct_len = 7;
+    if(test_len==correct_len && memcmp(test_buf, correct_buf, test_len)==0) {
+        printf("test on string \"%s\" passed\n", input);
+    } else {
+        printf("test on string \"%s\" failed:\ncorrect len: %d,got len: %d\ngot result:", input, correct_len, test_len);
+        for(int i=0;i<test_len;i++) {
+            printf("%d ", *(test_buf+i));
+        }
+        printf("\n");
+    }
+}
+
+int main() {
+    int res1[]={1,2,3,4,5,7,8};
+    check_extract_numbers("1-5,7-8", 7, res1);
+    int res2[]={1};
+    check_extract_numbers("1", 1, res2);
+    int res3[]={};
+    check_extract_numbers("", 0, res3);
+    int* cores;
+    int num_cores;
+    num_cores=get_isolated_cores(&cores);
+    printf("your system has %d isolated cores: ", num_cores);
+    for(int i = 0; i < num_cores; i++) {
+        printf("%d,", *(cores+i));
+    }
+    printf("\n");
+    free(cores);
+    return 0;
+}

tests/transport_test.c

@@ -0,0 +1,114 @@
+#define _GNU_SOURCE
+#include "gen/runner/v1/runner.pb-c.h"
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <rabbitmq-c/amqp.h>
+#include <rabbitmq-c/tcp_socket.h>
+
+struct connection_data {
+    char const *hostname;
+    int port;
+    char const *exchange;
+    char const *routingkey;
+    amqp_socket_t *socket;
+    amqp_connection_state_t conn;
+};
+
+void prepare_connection(int argc, char const *const *argv, struct connection_data *condata) {
+  condata->socket=NULL;
+  int status;
+
+  if (argc < 5) {
+    fprintf(stderr, "Usage: transport_test host port exchange routingkey\n");
+    exit(1);
+  }
+
+  condata->hostname = argv[1];
+  condata->port = atoi(argv[2]);
+  condata->exchange = argv[3];
+  condata->routingkey = argv[4];
+
+  condata->conn = amqp_new_connection();
+
+  condata->socket = amqp_tcp_socket_new(condata->conn);
+
+  status = amqp_socket_open(condata->socket, condata->hostname, condata->port);
+
+  amqp_login(condata->conn, "/", 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, "rmuser", "rmpassword"), "Logging in";
+  amqp_channel_open(condata->conn, 1);
+  amqp_get_rpc_reply(condata->conn), "Opening channel";
+}
+
+void send_to_rbmq(struct connection_data condata, char const* msg)
+{
+    amqp_basic_properties_t props;
+    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
+    props.content_type = amqp_cstring_bytes("text/plain");
+    props.delivery_mode = 2;
+    amqp_basic_publish(condata.conn, 1, amqp_cstring_bytes(condata.exchange), amqp_cstring_bytes(condata.routingkey), 0, 0, &props, amqp_cstring_bytes(msg)), "Publishing";
+}
+
+void build_serialize(int32_t solution_id, char* binding_key, int32_t language, char* solution, void** buf) {
+    struct Runner__V1__Build build = RUNNER__V1__BUILD__INIT;
+    build.solution_id = solution_id;
+    build.binding_key = binding_key;
+    build.language    = language;
+    build.solution    = solution;
+
+    struct Runner__V1__Instruction inst = RUNNER__V1__INSTRUCTION__INIT;
+    inst.instruction_case = RUNNER__V1__INSTRUCTION__INSTRUCTION_BUILD;
+    inst.build = &build;
+
+    size_t len=runner__v1__instruction__get_packed_size(&inst);
+    *buf=malloc(len+1);
+    memset(*buf,0,len+1);
+    runner__v1__instruction__pack(&inst, *buf);
+}
+
+void run_serialize(int32_t solution_id, int32_t test_id, char* binding_key, void** buf) {
+    struct Runner__V1__Run run= RUNNER__V1__RUN__INIT;
+    run.solution_id = solution_id;
+    run.test_id     = test_id;
+    run.binding_key = binding_key;
+
+    struct Runner__V1__Instruction inst = RUNNER__V1__INSTRUCTION__INIT;
+    inst.instruction_case = RUNNER__V1__INSTRUCTION__INSTRUCTION_RUN;
+    inst.run = &run;
+
+    size_t len=runner__v1__instruction__get_packed_size(&inst);
+    *buf=malloc(len+1);
+    memset(*buf,0,len+1);
+    runner__v1__instruction__pack(&inst, *buf);
+}
+
+int main(int argc, char const *const *argv) {
+  struct connection_data condata;
+  prepare_connection(argc,argv,&condata);
+
+  void *messagebody;
+  char* binding_key;
+  asprintf(&binding_key,"bktest");
+  char* solution;
+  asprintf(&solution,"soltest");
+  build_serialize(0,binding_key,1,solution,&messagebody);
+  free(binding_key);
+  free(solution);
+  send_to_rbmq(condata,messagebody);
+  free(messagebody);
+
+  asprintf(&binding_key,"bktest");
+  run_serialize(0,1,binding_key,&messagebody);
+  free(binding_key);
+  send_to_rbmq(condata,messagebody);
+  free(messagebody);
+
+  /* closing connection */
+
+  amqp_channel_close(condata.conn, 1, AMQP_REPLY_SUCCESS), "Closing channel";
+  amqp_connection_close(condata.conn, AMQP_REPLY_SUCCESS), "Closing connection";
+  amqp_destroy_connection(condata.conn), "Ending connection";
+  return 0;
+}

transport/transport.c

@@ -0,0 +1,96 @@
+#include "transport/transport.h"
+#include "starter/starter.h"
+
+void prepare_amqp_connection(int argc, char const *const *argv,struct connection_data *condata) {
+  int status;
+  if(secure_getenv("AMQP_HOSTNAME")==NULL || secure_getenv("AMQP_HOSTNAME")==NULL || secure_getenv("AMQP_EXCHANGE")==NULL || secure_getenv("AMQP_BINDINGKEY")==NULL) {
+    fprintf(stderr,"no amqp connection parameters in environment");
+    exit(1);
+  }
+  condata->hostname = getenv("AMQP_HOSTNAME");
+  condata->port = atoi(getenv("AMQP_PORT"));
+  condata->exchange = getenv("AMQP_EXCHANGE");
+  condata->bindingkey = getenv("AMQP_BINDINGKEY");
+
+  condata->conn = amqp_new_connection();
+
+  condata->socket = amqp_tcp_socket_new(condata->conn);
+  if (!condata->socket) {
+    error_log("failed to create TCP socket");
+  }
+
+  status = amqp_socket_open(condata->socket, condata->hostname, condata->port);
+  if (status) {
+    error_log("failed to open TCP socket");
+  }
+
+  amqp_assert(amqp_login(condata->conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "rmuser", "rmpassword"));
+  amqp_channel_open(condata->conn, 1);
+  amqp_assert(amqp_get_rpc_reply(condata->conn));
+
+  {
+    amqp_queue_declare_ok_t *r = amqp_queue_declare( condata->conn, 1, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table);
+    amqp_assert(amqp_get_rpc_reply(condata->conn));
+    condata->queuename = amqp_bytes_malloc_dup(r->queue);
+    if (condata->queuename.bytes == NULL) {
+      fprintf(stderr, "Out of memory while copying queue name");
+      exit(1);
+    }
+  }
+
+  amqp_queue_bind(condata->conn, 1, condata->queuename, amqp_cstring_bytes(condata->exchange), amqp_cstring_bytes(condata->bindingkey), amqp_empty_table);
+  amqp_assert(amqp_get_rpc_reply(condata->conn));
+
+  amqp_basic_consume(condata->conn, 1, condata->queuename, amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
+  amqp_assert(amqp_get_rpc_reply(condata->conn));
+}
+
+int main(int argc, char const *const *argv) {
+  struct connection_data condata;
+  prepare_amqp_connection(argc,argv,&condata);
+
+  {
+    for (;;) {
+      amqp_rpc_reply_t res;
+      amqp_envelope_t envelope;
+
+      amqp_maybe_release_buffers(condata.conn);
+
+      res = amqp_consume_message(condata.conn, &envelope, NULL, 0);
+
+      if (AMQP_RESPONSE_NORMAL != res.reply_type) {
+        break;
+      }
+
+      Runner__V1__Instruction *inst = runner__v1__instruction__unpack(NULL,envelope.message.body.len,envelope.message.body.bytes);
+      if(inst==NULL) {
+        fprintf(stderr, "error reading buffer");
+        exit(1);
+      }
+      switch(inst->instruction_case) {
+        case RUNNER__V1__INSTRUCTION__INSTRUCTION_BUILD:
+          puts("build request recieved");
+          printf("solution_id: %d\nbinding_key:%s\nlanguage:%d\nsolution:%s\n\n",inst->build->solution_id,inst->build->binding_key,inst->build->language,inst->build->solution);
+          break;
+        case RUNNER__V1__INSTRUCTION__INSTRUCTION_RUN:
+          puts("run request recieved");
+          printf("solution_id: %d\ntest_id:%d\nbinding_key:%s\n\n",inst->run->solution_id,inst->run->test_id,inst->run->binding_key);
+          break;
+        case RUNNER__V1__INSTRUCTION__INSTRUCTION__NOT_SET:
+          puts("empty request recieved");
+          break;
+        default:
+          puts("unknown request recieved");
+      }
+      runner__v1__instruction__free_unpacked(inst, NULL);
+      amqp_destroy_envelope(&envelope);
+    }
+  }
+
+  amqp_bytes_free(condata.queuename);
+
+  amqp_assert(amqp_channel_close(condata.conn, 1, AMQP_REPLY_SUCCESS));
+  amqp_assert(amqp_connection_close(condata.conn, AMQP_REPLY_SUCCESS));
+  amqp_code_assert(amqp_destroy_connection(condata.conn));
+  return 0;
+}

transport/transport.h

@@ -0,0 +1,27 @@
+#define _GNU_SOURCE
+#include "gen/runner/v1/runner.pb-c.h"
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#include <rabbitmq-c/amqp.h>
+#include <rabbitmq-c/tcp_socket.h>
+
+struct connection_data {
+  char const *hostname;
+  int port;
+  char const *exchange;
+  char const *bindingkey;
+  amqp_socket_t *socket;
+  amqp_connection_state_t conn;
+  amqp_bytes_t queuename;
+};
+
+extern char* amqp_error(amqp_rpc_reply_t x);
+#define error_log(err) fprintf(stderr,"error in %s line %d:%s",__FILE__,__LINE__,err);
+#define amqp_assert(msg) {char* err=amqp_error(msg);if(err!=NULL) {fprintf(stderr,"amqp error in %s line %d:%s",__FILE__,__LINE__,err);free(err);}}
+#define amqp_code_assert(msg) {int32_t code=msg;if(code<0) {fprintf(stderr,"amqp error in %s line %d:%s",__FILE__,__LINE__,amqp_error_string2(code));}}
+
+

transport/utils.c

@@ -0,0 +1,48 @@
+#define _GNU_SOURCE
+#include <ctype.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "gen/runner/v1/runner.pb-c.h"
+#include <rabbitmq-c/amqp.h>
+#include <rabbitmq-c/framing.h>
+#include <rabbitmq-c/tcp_socket.h>
+#include <stdint.h>
+
+#include "transport/transport.h"
+
+char* amqp_error(amqp_rpc_reply_t x) {
+  char* res=NULL;
+  switch (x.reply_type) {
+    case AMQP_RESPONSE_NORMAL:
+      break;
+
+    case AMQP_RESPONSE_NONE:
+      asprintf(&res,"missing RPC reply type");
+      break;
+
+    case AMQP_RESPONSE_LIBRARY_EXCEPTION:
+      asprintf(&res,amqp_error_string2(x.library_error));
+      break;
+
+    case AMQP_RESPONSE_SERVER_EXCEPTION:
+      switch (x.reply.id) {
+        case AMQP_CONNECTION_CLOSE_METHOD: {
+          amqp_connection_close_t *m = (amqp_connection_close_t *)x.reply.decoded;
+          asprintf(&res,"server connection error %uh, message: %.*s\n", m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes);
+          break;
+        }
+        case AMQP_CHANNEL_CLOSE_METHOD: {
+          amqp_channel_close_t *m = (amqp_channel_close_t *)x.reply.decoded;
+          asprintf(&res,"server channel error %uh, message: %.*s\n", m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes);
+          break;
+        }
+        default:
+          asprintf(&res,"unknown server error, method id 0x%08X\n", x.reply.id);
+          break;
+      }
+  }
+  return res;
+}