Log using a new task but using explicit context switching

This is effectively synchronous logging, but the new task allows us to have a clean context in which to call into julia.
JuliaLang · Jan 16, 2018 · dde9e4e · dde9e4e
1 parent b283281
commit dde9e4e
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 44 deletions.
diff --git a/base/logging.jl b/base/logging.jl
@@ -359,13 +359,17 @@ end
     nothing
 end
 
-# Log a message. Called from the julia C code; kwargs is in the format
-# Any[key1,val1, ...] for simplicity in construction on the C side.
-function logmsg_shim(level, message, _module, group, id, file, line, kwargs)
+# Log a message. Called from the julia C code; `kwargs` is in the format
+# `Any[key1,val1, ...]` for simplicity in construction on the C side.
+# `srctask` is `nothing` or the task from which the message originated and
+# which we should yield to when done. (The source task may be in a state
+# where it can not call julia functions directly.)
+function logmsg_shim(srctask, level, message, _module, group, id, file, line, kwargs)
     real_kws = Any[(kwargs[i],kwargs[i+1]) for i in 1:2:length(kwargs)]
     @logmsg(convert(LogLevel, level), message,
             _module=_module, _id=id, _group=group,
             _file=String(file), _line=line, real_kws...)
+    srctask === nothing || yieldto(srctask)
 end
 
 # Global log limiting mechanism for super fast but inflexible global log

diff --git a/src/julia_internal.h b/src/julia_internal.h
@@ -705,7 +705,6 @@ size_t rec_backtrace_ctx_dwarf(uintptr_t *data, size_t maxsize, bt_context_t *ct
 #endif
 JL_DLLEXPORT void jl_get_backtrace(jl_array_t **bt, jl_array_t **bt2);
 JL_DLLEXPORT jl_value_t *jl_apply_with_saved_exception_state(jl_value_t **args, uint32_t nargs, int drop_exceptions);
-jl_task_t *jl_apply_in_new_task(jl_value_t **args, uint32_t nargs);
 void jl_critical_error(int sig, bt_context_t *context, uintptr_t *bt_data, size_t *bt_size);
 JL_DLLEXPORT void jl_raise_debugger(void);
 int jl_getFunctionInfo(jl_frame_t **frames, uintptr_t pointer, int skipC, int noInline);

diff --git a/src/rtutils.c b/src/rtutils.c
@@ -1035,21 +1035,14 @@ void jl_log(int level, jl_value_t *module, jl_value_t *group, jl_value_t *id,
             jl_value_t *file, jl_value_t *line, jl_value_t *kwargs, int async,
             jl_value_t *msg)
 {
-    static jl_value_t *logmsg_func = NULL;
-    jl_array_t *Workqueue = NULL;
-    if (!logmsg_func && jl_base_module) {
+    static jl_value_t *logmsg_shim = NULL;
+    if (!logmsg_shim && jl_base_module) {
         jl_value_t *corelogging = jl_get_global(jl_base_module, jl_symbol("CoreLogging"));
         if (corelogging && jl_is_module(corelogging)) {
-            logmsg_func = jl_get_global((jl_module_t*)corelogging, jl_symbol("logmsg_shim"));
+            logmsg_shim = jl_get_global((jl_module_t*)corelogging, jl_symbol("logmsg_shim"));
         }
     }
-    if (!Workqueue) {
-        jl_value_t *wq = jl_get_global(jl_base_module, jl_symbol("Workqueue"));
-        if (wq && jl_is_array(wq) && jl_array_eltype(wq) == jl_task_type) {
-            Workqueue = (jl_array_t*)wq;
-        }
-    }
-    if (!logmsg_func || (async && !Workqueue)) {
+    if (!logmsg_shim) {
         ios_t str_;
         ios_mem(&str_, 300);
         uv_stream_t* str = (uv_stream_t*)&str_;
@@ -1077,25 +1070,32 @@ void jl_log(int level, jl_value_t *module, jl_value_t *group, jl_value_t *id,
         return;
     }
     jl_value_t **args;
-    const int nargs = 9;
+    const int nargs = 10;
     JL_GC_PUSHARGS(args, nargs);
-    args[0] = logmsg_func;
-    args[1] = jl_box_long(level);
-    args[2] = msg;
+    args[0] = logmsg_shim;
+    args[1] = async ? (jl_value_t*)jl_current_task : jl_nothing;
+    args[2] = jl_box_long(level);
+    args[3] = msg;
     // Some of the jl_nothing here would perhaps be better as `missing` instead.
-    args[3] = module ? module  : jl_nothing;
-    args[4] = group  ? group   : jl_nothing;
-    args[5] = id     ? id      : jl_nothing;
-    args[6] = file   ? file    : jl_nothing;
-    args[7] = line   ? line    : jl_nothing;
-    args[8] = kwargs ? kwargs  : (jl_value_t*)jl_alloc_vec_any(0);
+    args[4] = module ? module  : jl_nothing;
+    args[5] = group  ? group   : jl_nothing;
+    args[6] = id     ? id      : jl_nothing;
+    args[7] = file   ? file    : jl_nothing;
+    args[8] = line   ? line    : jl_nothing;
+    args[9] = kwargs ? kwargs  : (jl_value_t*)jl_alloc_vec_any(0);
     if (async) {
-        jl_task_t *task = jl_apply_in_new_task(args, nargs);
-        // The following duplicates Base.enq_work() without calling into julia.
-        uv_stop(jl_global_event_loop());
-        jl_array_grow_end(Workqueue, 1);
-        jl_array_ptr_set(Workqueue, jl_array_len(Workqueue)-1, (jl_value_t*)task);
-        task->state = jl_symbol("queued");
+        // In certain circumstances the current task may not be able to call
+        // into julia.  In that case do so in a new task instead.
+        jl_array_t *argvec = jl_alloc_vec_any(nargs);
+        jl_value_t *closure = NULL;
+        jl_task_t *logtask = NULL;
+        JL_GC_PUSH3(&argvec, &closure, &logtask);
+        for (uint32_t i = 0; i < nargs; ++i)
+            jl_array_ptr_set(argvec, i, args[i]);
+        closure = jl_new_struct(jl_deferredcall_type, argvec);
+        logtask = jl_new_task(closure, 0);
+        jl_switchto(&logtask);
+        JL_GC_POP();
     }
     else {
         jl_apply(args, nargs);
@@ -1142,19 +1142,6 @@ JL_DLLEXPORT void jl_depwarn_partial_indexing(size_t n)
     JL_GC_POP();
 }
 
-JL_DLLEXPORT jl_task_t *jl_apply_in_new_task(jl_value_t **args0, uint32_t nargs)
-{
-    jl_array_t *args = jl_alloc_vec_any(nargs);
-    jl_value_t *closure = NULL;
-    JL_GC_PUSH2(&args, &closure);
-    for (uint32_t i = 0; i < nargs; ++i)
-        jl_array_ptr_set(args, i, args0[i]);
-    closure = jl_new_struct(jl_deferredcall_type, args);
-    jl_task_t* task = jl_new_task(closure, 0);
-    JL_GC_POP();
-    return task;
-}
-
 #ifdef __cplusplus
 }
 #endif