Merge pull request #36 from kjellwinblad/kjell/timeout_nif

Add support for timeouts in the NIF backend

Author: kjellwinblad

README.md

@@ -50,19 +50,21 @@ application is loaded (the jq Erlang application is loaded automatically when
 * `jq_filter_program_lru_cache_max_size` (default value = 500) - Sets the size of
   the LRU caches that are holding compiled JQ programs to prevent frequent
   expensive recompilation of the same program.
-* `jq_implementation_module` (default value = `jq_port`) - Sets the implementation
+* `jq_implementation_module` (default value = `jq_nif`) - Sets the implementation
   that will be used. The options are:
   * `jq_port` - This implementation uses a port program to interact with jq.
     This is the most safe option as a bug in jq cannot
     cause the Erlang VM to crash or leak memory.
   * `jq_nif` - This implementation uses a NIF library to interact with jq. This
     option is faster than the `jq_port` option but it is also less safe even
     though we are currently not not aware of any problems with this option.
+  One can use the function `jq:set_implementation_module/2` to change the
+  implementation while the application is running.
 * `jq_port_nr_of_jq_port_servers` (default value =
   `erlang:system_info(schedulers)`) (only relevant for the `jq_port` option) -
   Use this option to set how many port programs that will handle jq requests.
   Higher values can lead to better performance (due to parallelism) at the
-  expense of increased memory usage and cache locality. 
+  expense of increased memory usage and worse cache locality. 
 * `jq_port_restart_period` (default value = 1000000) (only relevant for the
   `jq_port` option) - Use this option to set how many `jq:process_json/2` calls
   a port program can process before it is restarted. This is a safety option

c_src/Makefile

@@ -13,9 +13,11 @@ NIF_MODULE := jq_nif
 JQURL := https://github.com/emqx/jqc.git
 JQSRC_DIR := libs/jqc
 JQSRC := $(JQSRC_DIR)/src/jv.c
+JQSRC_VERSION := 73c51aa79b67b99c575ef8923aa0c03cd87a7ac1
 RDSURL := https://github.com/emqx/c_reusable_data_structures.git
 RDSSRC_DIR := libs/c_reusable_data_structures
 RDSSRC := $(RDSSRC_DIR)/Makefile
+RDSSRC_VERSION := 77c7a96416065a839417e5078fce9ba9a14982dc
 LIBJQ_DIR ?= $(JQSRC_DIR)/.libs
 LIBJQ_PREFIX := /usr/local
 EXT_LIBS := ext_libs
@@ -104,7 +106,9 @@ PORT_OBJECTS = $(addsuffix .o, $(basename $(PORT_SOURCES)))
 COMPILE_C = $(c_verbose) $(CC) $(CFLAGS) $(CPPFLAGS) -c
 COMPILE_CPP = $(cpp_verbose) $(CXX) $(CXXFLAGS) $(CPPFLAGS) -c
 
-.PHONY: clean
+.PHONY: clean check_jqc_version all check_rds_version
+
+all: check_jqc_version check_rds_version $(C_SRC_OUTPUT)
 
 ifeq ($(JQ_MEMSAN_DEBUG), 1)
 $(C_SRC_OUTPUT): $(OBJECTS) port_nif_common.h
@@ -131,16 +135,25 @@ $(ERL_PORT_PROGRAM): $(PORT_OBJECTS)
 %.o: %.c $(LIBJQ_NAME) $(JQERLMODSRC) $(RDSSRC) $(PORT_SOURCES)
 	$(COMPILE_C) $(OUTPUT_OPTION) $<
 
+check_jqc_version:
+	( cd libs/jqc && \
+	  [ `git rev-parse --verify HEAD` = '$(JQSRC_VERSION)' ] || (cd .. && rm -rf jqc))
+
+
 $(JQSRC):
 	git clone -b jq-1.6-emqx --single-branch "$(JQURL)" "$(JQSRC_DIR)"
-	((cd "$(JQSRC_DIR)" && git checkout 51df83ccf4808867eee79a011c71a93a121be1b6) || \
+	((cd "$(JQSRC_DIR)" && git checkout $(JQSRC_VERSION)) || \
 	     (echo "Failed to check out jq commit" && \
 	      rm -r "$(JQSRC_DIR)" && \
 	      false))
 
+check_rds_version:
+	( cd libs/c_reusable_data_structures && \
+	  [ `git rev-parse --verify HEAD` = '$(RDSSRC_VERSION)' ] || (cd .. && rm -rf c_reusable_data_structures))
+
 $(RDSSRC):
 	git clone -b master --single-branch "$(RDSURL)" "$(RDSSRC_DIR)"
-	((cd "$(RDSSRC_DIR)" && git checkout 77c7a96416065a839417e5078fce9ba9a14982dc) || \
+	((cd "$(RDSSRC_DIR)" && git checkout $(RDSSRC_VERSION)) || \
 	     (echo "Failed to check out c_reusable_data_structures commit" && \
 	      rm -r "$(RDSSRC_DIR)" && \
 	      false))

c_src/erlang_jq_nif.c

@@ -1,7 +1,9 @@
 #include <erl_nif.h>
+#include "jq.h"
 #include "jv.h"
 #include "port_nif_common.h"
 
+#include <stdatomic.h>
 #include <stdbool.h>
 #include <setjmp.h>
 
@@ -96,6 +98,9 @@ typedef struct {
     JQStateCacheEntry_lru_ptr_dynarr caches;
     // Lock protecting the above field from concurrent modifications
     ErlNifMutex * lock;
+    // NIF resource that holds a jq state that is used when timing out
+    // NIF calls
+    ErlNifResourceType * jq_state_holder_resource_type; 
 } module_private_data;
 
 
@@ -162,6 +167,9 @@ jq_state* get_jq_state(
             JQStateCacheEntry_lru_get(cache, new_entry);
         if (cache_entry != NULL) {
             // Return jq_state from cache
+            // It is important that we reset the cancel state so
+            // that the jq execution don't get canceled immediately
+            jq_reset_cancel_state(cache_entry->state);
             return cache_entry->state;
         }
     } else {
@@ -178,6 +186,7 @@ jq_state* get_jq_state(
         erljq_free(error_message);
         return NULL;
     }
+    jq_reset_cancel_state(jq);
     if (cache != NULL) {
         // Add new entry to cache
         char * filter_program_string = erljq_alloc(erl_jq_filter.size);
@@ -192,7 +201,7 @@ jq_state* get_jq_state(
 }
 
 
-// Process the JSON obejct value using the compiled filter program in the
+// Process the JSON object value using the compiled filter program in the
 // given jq_state
 static int process_json(
         jq_state *jq,
@@ -233,6 +242,14 @@ static int process_json(
             (char*)enif_make_new_binary(env, binsz, error_msg_bin_ptr);
         memcpy(bin_data, error_message, binsz);
         erljq_free(error_message);
+        // We might have results in the result strings (e.g., when there is a
+        // timeout) so we need to free them as well
+        size_t nr_of_result_objects = String_dynarr_size(&result_strings);
+        for (size_t i = 0; i < nr_of_result_objects; i++) {
+            String result = String_dynarr_item_at(&result_strings, i);
+            erljq_free(result.string);
+        }
+        String_dynarr_destroy(&result_strings);
     }
     return res;
 }
@@ -253,9 +270,66 @@ static ERL_NIF_TERM make_ok_return(ErlNifEnv* env, ERL_NIF_TERM result) {
     return enif_make_tuple2(env, enif_make_atom(env, "ok"), result);
 }
 
+typedef struct {
+    ErlNifMutex* lock;
+    volatile atomic_bool consumed_by_process_json_nif;
+    jq_state* state;
+} JQCancelableStateHolderResource;
+
+static ERL_NIF_TERM
+create_jq_resource_nif(
+        ErlNifEnv* env,
+        int argc,
+        const ERL_NIF_TERM argv[]) {
+    module_private_data* module_data = enif_priv_data(env);
+    JQCancelableStateHolderResource* cancelable_state_holder =
+        enif_alloc_resource(module_data->jq_state_holder_resource_type,
+                            sizeof(JQCancelableStateHolderResource));
+    cancelable_state_holder->state = NULL;
+    static char* lock_name = "jq_nif_resource_lock";
+    cancelable_state_holder->lock = enif_mutex_create(lock_name);
+    atomic_init(&cancelable_state_holder->consumed_by_process_json_nif, false);
+    ERL_NIF_TERM term = enif_make_resource(env, cancelable_state_holder);
+    enif_release_resource(cancelable_state_holder);
+    return term;
+}
+
+static ERL_NIF_TERM
+cancel_jq_resource_nif(
+        ErlNifEnv* env,
+        int argc,
+        const ERL_NIF_TERM argv[]) {
+    module_private_data* module_data = enif_priv_data(env);
+    JQCancelableStateHolderResource* cancelable_state_holder;
+    if (enif_get_resource(
+                env,
+                argv[0],
+                module_data->jq_state_holder_resource_type,
+                (void**)&cancelable_state_holder)) {
+        if (!atomic_load(&cancelable_state_holder->consumed_by_process_json_nif)) {
+            // The resource has not been passed to process_json_nif yet. We
+            // schedule ourselves out and will try again later
+            enif_consume_timeslice(env, 100);
+            return enif_make_atom(env, "retry");
+        }
+        enif_mutex_lock(cancelable_state_holder->lock);
+        if (cancelable_state_holder->state != NULL) {
+            // jq_cancel is a function to cancel the jq execution from
+            // another thread. This function only exists in the our
+            // patched version of the jq library.
+            jq_cancel(cancelable_state_holder->state);
+        }
+        enif_mutex_unlock(cancelable_state_holder->lock);
+    } else {
+        return enif_make_badarg(env);
+    }
+    return enif_make_atom(env, "ok");
+}
+
 // NIF function taking a binaries for a filter program and a JSON text
 // and returning the result of processing the JSON text with the
-// filter program 
+// filter program. The third argument can be a NIF resource that other
+// threads can use to cancel the execution
 static ERL_NIF_TERM process_json_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
     ERL_NIF_TERM error_msg_bin; 
     // ----------------------------- init --------------------------------------
@@ -264,7 +338,25 @@ static ERL_NIF_TERM process_json_nif(ErlNifEnv* env, int argc, const ERL_NIF_TER
     int ret = JQ_ERROR_UNKNOWN;
     ERL_NIF_TERM ret_term;
     int dumpopts = 512; // JV_PRINT_SPACE1
-
+    JQCancelableStateHolderResource* cancelable_state_holder = NULL; 
+    bool cancelable_state_holder_opened = false;
+    if (argc > 2) {
+        // Third argument is a NIF resource for canceling
+        // the jq execution from another thread
+        module_private_data* module_data = enif_priv_data(env);
+        if (!enif_get_resource(
+                    env,
+                    argv[2],
+                    module_data->jq_state_holder_resource_type,
+                    (void**)&cancelable_state_holder)) {
+            ret = JQ_ERROR_BADARG;
+            const char* error_message =
+                "Expected a resource as third argument";
+            error_msg_bin =
+                make_error_msg_bin(env, error_message, strlen(error_message));
+            goto out;
+        }
+    }
     // --------------------------- read args -----------------------------------
     ErlNifBinary erl_jq_filter;
     ErlNifBinary erl_json_text;
@@ -280,13 +372,22 @@ static ERL_NIF_TERM process_json_nif(ErlNifEnv* env, int argc, const ERL_NIF_TER
     if (setjmp(nomem_handling_jmp_buf)) {
         // Give badarg exception as this seems more reasonable than allocating and
         // returning an error tuple when we have run out of memory
+        if (cancelable_state_holder != NULL && !cancelable_state_holder_opened) {
+            atomic_store(&cancelable_state_holder->consumed_by_process_json_nif, true);
+        }
         return enif_make_badarg(env);
     }
     // --------- get jq state and compile filter program if not cached ---------
     jq = get_jq_state(env, &error_msg_bin, &ret, erl_jq_filter, &remove_jq_object);
     if (jq == NULL) {
         goto out;
+    } else if (cancelable_state_holder != NULL) {
+        cancelable_state_holder->state = jq;
+        // jq state can be timed out after the following unlock call
+        atomic_store(&cancelable_state_holder->consumed_by_process_json_nif, true);
+        cancelable_state_holder_opened = true;
     }
+
     ERL_NIF_TERM ret_list = enif_make_list(env, 0);
     ret = process_json(
             jq,
@@ -296,8 +397,17 @@ static ERL_NIF_TERM process_json_nif(ErlNifEnv* env, int argc, const ERL_NIF_TER
             0,
             dumpopts,
             &error_msg_bin);
+    if (cancelable_state_holder != NULL) {
+        // We don't want any timeouts after this point 
+        enif_mutex_lock(cancelable_state_holder->lock);
+        cancelable_state_holder->state = NULL;
+        enif_mutex_unlock(cancelable_state_holder->lock);
+    }
 
 out:// ----------------------------- release -----------------------------------
+    if (cancelable_state_holder != NULL && !cancelable_state_holder_opened) {
+        atomic_store(&cancelable_state_holder->consumed_by_process_json_nif, true);
+    }
     switch (ret) {
         default:
             assert(0 && "invalid ret");
@@ -306,6 +416,7 @@ static ERL_NIF_TERM process_json_nif(ErlNifEnv* env, int argc, const ERL_NIF_TER
         case JQ_ERROR_BADARG:
         case JQ_ERROR_COMPILE:
         case JQ_ERROR_PARSE:
+        case JQ_ERROR_TIMEOUT:
         case JQ_ERROR_PROCESS: {
             ret_term = make_error_return(env, ret, error_msg_bin);
             break;
@@ -386,6 +497,12 @@ static int get_int_config(
     return 0;
 }
 
+
+static void jq_state_holder_resource_dtor(ErlNifEnv* caller_env, void* obj) {
+    JQCancelableStateHolderResource* cancelable_state_holder = obj;
+    enif_mutex_destroy(cancelable_state_holder->lock);
+}
+
 static int load_helper(
         ErlNifEnv* caller_env,
         void** priv_data,
@@ -436,6 +553,14 @@ static int load_helper(
         erljq_free(data);
         return 1;
     }
+    static const char * jq_state_holder_resource_type_name = "jq_state_holder_resource_type";
+    data->jq_state_holder_resource_type = enif_open_resource_type(
+            caller_env,
+            NULL,
+            jq_state_holder_resource_type_name,
+            jq_state_holder_resource_dtor,
+            ERL_NIF_RT_CREATE | ERL_NIF_RT_TAKEOVER,
+            NULL); 
     JQStateCacheEntry_lru_ptr_dynarr_init(&data->caches);
     *priv_data = data;
     return 0;
@@ -478,6 +603,9 @@ static ErlNifFunc nif_funcs[] = {
        for a much longer time than is should).
     */
     {"process_json", 2, process_json_nif, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+    {"create_jq_resource", 0, create_jq_resource_nif, 0},
+    {"cancel_jq_resource", 1, cancel_jq_resource_nif, 0},
+    {"process_json_with_jq_resource", 3, process_json_nif, ERL_NIF_DIRTY_JOB_CPU_BOUND},
     {"set_filter_program_lru_cache_max_size", 1, set_filter_program_lru_cache_max_size_nif, 0},
     {"get_filter_program_lru_cache_max_size", 0, get_filter_program_lru_cache_max_size_nif, 0}
 };

c_src/erlang_jq_port.c

@@ -271,6 +271,14 @@ static bool handle_process_json() {
         return false;
     } else {
         const char* error_str = "error";
+        // We might have results in the result strings. This can happen
+        // when we get a result out from jq before an error happens
+        size_t nr_of_result_objects = String_dynarr_size(&result_strings);
+        for (size_t i = 0; i < nr_of_result_objects; i++) {
+            String result = String_dynarr_item_at(&result_strings, i);
+            erljq_free(result.string);
+        }
+        String_dynarr_destroy(&result_strings);
         if (write_packet((byte*)error_str, strlen(error_str)) <= 0) {
             goto error_on_write_out_1;
         }

c_src/port_nif_common.c

@@ -13,7 +13,8 @@ char* err_tags[] = {
     "jq_err_badarg",    // 3
     "jq_err_compile",   // 4
     "jq_err_parse",     // 5
-    "jq_err_process"    // 6
+    "jq_err_process",   // 6
+    "timeout"           // 7 
 };
 static const char* ERR_MSG_COULD_NOT_INIT = "jq_init: Could not initialize jq";
 static const char* ERR_MSG_COMPILATION_FAILED = "Compilation of jq filter failed";
@@ -143,23 +144,38 @@ int process_json_common(
       String_dynarr_push(result_strings, port_res_str);
       jv_free(res_jv_str);
   }
-
-  if (jv_invalid_has_msg(jv_copy(result))) {
+  if (jq_canceled(jq)) {
+        // Use define for format because the compiler complains if its
+        // not a literal
+      #define ERL_JQ_ERR_MSG \
+          "jq program canceled as it took too long time to execute"  
+      size_t binsz =
+          snprintf(NULL, 0, ERL_JQ_ERR_MSG) + 1;
+      char* bin_data = erljq_alloc(binsz);
+      snprintf(bin_data, binsz, ERL_JQ_ERR_MSG);
+      #undef ERL_JQ_ERR_MSG
+      *error_msg_wb = bin_data;
+      ret = JQ_ERROR_TIMEOUT;
+  } else if (jv_invalid_has_msg(jv_copy(result))) {
     // Uncaught jq exception
     jv msg = jv_invalid_get_msg(jv_copy(result));
     if (jv_get_kind(msg) == JV_KIND_STRING) {
+        #define ERL_JQ_ERR_MSG "jq error: %s\n"  
         size_t binsz =
-            snprintf(NULL, 0, "jq error: %s\n", jv_string_value(msg)) + 1;
+            snprintf(NULL, 0, ERL_JQ_ERR_MSG, jv_string_value(msg)) + 1;
         char* bin_data = erljq_alloc(binsz);
-        snprintf(bin_data, binsz, "jq error: %s\n", jv_string_value(msg));
+        snprintf(bin_data, binsz, ERL_JQ_ERR_MSG, jv_string_value(msg));
+        #undef ERL_JQ_ERR_MSG
         *error_msg_wb = bin_data;
     } else {
         msg = jv_dump_string(msg, 0);
-        size_t binsz = snprintf(NULL, 0, "jq error (not a string): %s\n",
+        #define ERL_JQ_ERR_MSG "jq error (not a string): %s\n"  
+        size_t binsz = snprintf(NULL, 0, ERL_JQ_ERR_MSG,
                 jv_string_value(msg)) + 1;
         char* bin_data = erljq_alloc(binsz);
-        snprintf(bin_data, binsz, "jq error (not a string): %s\n",
+        snprintf(bin_data, binsz, ERL_JQ_ERR_MSG,
                 jv_string_value(msg));
+        #undef ERL_JQ_ERR_MSG
         *error_msg_wb = bin_data;
     }
     ret = JQ_ERROR_PROCESS;

c_src/port_nif_common.h

@@ -19,7 +19,8 @@ enum {
     JQ_ERROR_BADARG    =  3,
     JQ_ERROR_COMPILE   =  4,
     JQ_ERROR_PARSE     =  5,
-    JQ_ERROR_PROCESS   =  6
+    JQ_ERROR_PROCESS   =  6,
+    JQ_ERROR_TIMEOUT   =  7
 };
 
 extern char* err_tags[];