remove threadedregion and move jl_threading_run to julia

Author: JeffBezanson

base/task.jl

@@ -409,6 +409,14 @@ function __preinit_threads__()
     nothing
 end
 
+function _run_on(t::Task, tid)
+    @assert !istaskstarted(t)
+    t.sticky = true
+    ccall(:jl_set_task_tid, Cvoid, (Any, Cint), t, tid-1)
+    schedule(t)
+    return t
+end
+
 function enq_work(t::Task)
     (t.state == :runnable && t.queue === nothing) || error("schedule: Task not runnable")
     tid = Threads.threadid(t)

base/threadingconstructs.jl

@@ -71,14 +71,22 @@ function _threadsfor(iter,lbody)
             Base.invokelatest(threadsfor_fun, true)
         else
             in_threaded_loop[] = true
-            # the ccall is not expected to throw
-            ccall(:jl_threading_run, Cvoid, (Any,), threadsfor_fun)
+            threading_run(threadsfor_fun)
             in_threaded_loop[] = false
         end
         nothing
     end
 end
 
+function threading_run(func)
+    tasks = Vector{Task}(undef, nthreads())
+    for tid = 1:nthreads()
+        tasks[tid] = Base._run_on(Task(func), tid)
+    end
+    foreach(wait, tasks)
+    return nothing
+end
+
 """
     Threads.@threads
 

src/jl_uv.c

@@ -201,12 +201,10 @@ JL_DLLEXPORT void jl_uv_req_set_data(uv_req_t *req, void *data) { req->data = da
 JL_DLLEXPORT void *jl_uv_handle_data(uv_handle_t *handle) { return handle->data; }
 JL_DLLEXPORT void *jl_uv_write_handle(uv_write_t *req) { return req->handle; }
 
-extern volatile unsigned _threadedregion;
-
 JL_DLLEXPORT int jl_run_once(uv_loop_t *loop)
 {
     jl_ptls_t ptls = jl_get_ptls_states();
-    if (loop && (_threadedregion || ptls->tid == 0)) {
+    if (loop) {
         jl_gc_safepoint_(ptls);
         JL_UV_LOCK();
         loop->stop_flag = 0;
@@ -220,7 +218,7 @@ JL_DLLEXPORT int jl_run_once(uv_loop_t *loop)
 JL_DLLEXPORT int jl_process_events(uv_loop_t *loop)
 {
     jl_ptls_t ptls = jl_get_ptls_states();
-    if (loop && (_threadedregion || ptls->tid == 0)) {
+    if (loop) {
         jl_gc_safepoint_(ptls);
         if (jl_mutex_trylock(&jl_uv_mutex)) {
             loop->stop_flag = 0;

src/partr.c

@@ -411,8 +411,6 @@ static int may_sleep(jl_ptls_t ptls)
     return jl_atomic_load(&sleep_check_state) == sleeping && jl_atomic_load(&ptls->sleep_check_state) == sleeping;
 }
 
-extern volatile unsigned _threadedregion;
-
 JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *getsticky)
 {
     jl_ptls_t ptls = jl_get_ptls_states();
@@ -434,7 +432,7 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *getsticky)
 #endif
 
         jl_cpu_pause();
-        if (sleep_check_after_threshold(&start_cycles) || (!_threadedregion && ptls->tid == 0)) {
+        if (sleep_check_after_threshold(&start_cycles)) {
             if (!sleep_check_now(ptls->tid))
                 continue;
             jl_atomic_store(&ptls->sleep_check_state, sleeping); // acquire sleep-check lock
@@ -446,16 +444,8 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *getsticky)
             // although none are allowed to create new ones
             // outside of threaded regions, all IO is permitted,
             // but only on thread 1
-            int uvlock = 0;
-            if (_threadedregion) {
-                uvlock = jl_mutex_trylock(&jl_uv_mutex);
-            }
-            else if (ptls->tid == 0) {
-                uvlock = 1;
-                JL_UV_LOCK();
-            }
+            int uvlock = jl_mutex_trylock(&jl_uv_mutex);
             if (uvlock) {
-                int active = 1;
                 if (jl_atomic_load(&jl_uv_n_waiters) != 0) {
                     // but if we won the race against someone who actually needs
                     // the lock to do real work, we need to let them have it instead
@@ -467,7 +457,7 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *getsticky)
                     jl_gc_safepoint();
                     if (may_sleep(ptls)) {
                         loop->stop_flag = 0;
-                        active = uv_run(loop, UV_RUN_ONCE);
+                        uv_run(loop, UV_RUN_ONCE);
                     }
                     JL_UV_UNLOCK();
                     // optimization: check again first if we may have work to do
@@ -483,12 +473,6 @@ JL_DLLEXPORT jl_task_t *jl_task_get_next(jl_value_t *getsticky)
                     //       to the last thread to do an explicit operation,
                     //       which may starve other threads of critical work
                 }
-                if (!_threadedregion && active && ptls->tid == 0) {
-                    // thread 0 is the only thread permitted to run the event loop
-                    // so it needs to stay alive
-                    start_cycles = 0;
-                    continue;
-                }
             }
 
             // the other threads will just wait for on signal to resume

src/threading.c

@@ -473,78 +473,6 @@ void jl_start_threads(void)
 
 #endif
 
-unsigned volatile _threadedregion; // HACK: keep track of whether it is safe to do IO
-
-// simple fork/join mode code
-JL_DLLEXPORT void jl_threading_run(jl_value_t *func)
-{
-    jl_ptls_t ptls = jl_get_ptls_states();
-    int8_t gc_state = jl_gc_unsafe_enter(ptls);
-    size_t world = jl_world_counter;
-    jl_method_instance_t *mfunc = jl_lookup_generic(&func, 1, jl_int32hash_fast(jl_return_address()), world);
-    // Ignore constant return value for now.
-    jl_code_instance_t *fptr = jl_compile_method_internal(mfunc, world);
-    if (fptr->invoke == jl_fptr_const_return)
-        return;
-
-    size_t nthreads = jl_n_threads;
-    jl_svec_t *ts = jl_alloc_svec(nthreads);
-    JL_GC_PUSH1(&ts);
-    jl_value_t *wait_func = jl_get_global(jl_base_module, jl_symbol("wait"));
-    jl_value_t *schd_func = jl_get_global(jl_base_module, jl_symbol("schedule"));
-    // create and schedule all tasks
-    _threadedregion += 1;
-    for (int i = 0; i < nthreads; i++) {
-        jl_value_t *args2[2];
-        args2[0] = (jl_value_t*)jl_task_type;
-        args2[1] = func;
-        jl_task_t *t = (jl_task_t*)jl_apply(args2, 2);
-        jl_svecset(ts, i, t);
-        t->sticky = 1;
-        t->tid = i;
-        args2[0] = schd_func;
-        args2[1] = (jl_value_t*)t;
-        jl_apply(args2, 2);
-#ifdef JULIA_ENABLE_THREADING
-        if (i == 1) {
-            // let threads know work is coming (optimistic)
-            jl_wakeup_thread(-1);
-        }
-#endif
-    }
-#ifdef JULIA_ENABLE_THREADING
-    if (nthreads > 2) {
-        // let threads know work is ready (guaranteed)
-        jl_wakeup_thread(-1);
-    }
-#endif
-    // join with all tasks
-    JL_TRY {
-        for (int i = 0; i < nthreads; i++) {
-            jl_value_t *t = jl_svecref(ts, i);
-            jl_value_t *args[2] = { wait_func, t };
-            jl_apply(args, 2);
-        }
-    }
-    JL_CATCH {
-        _threadedregion -= 1;
-        jl_wake_libuv();
-        JL_UV_LOCK();
-        JL_UV_UNLOCK();
-        jl_rethrow();
-    }
-    // make sure no threads are sitting in the event loop
-    _threadedregion -= 1;
-    jl_wake_libuv();
-    // make sure no more callbacks will run while user code continues
-    // outside thread region and might touch an I/O object.
-    JL_UV_LOCK();
-    JL_UV_UNLOCK();
-    JL_GC_POP();
-    jl_gc_unsafe_leave(ptls, gc_state);
-}
-
-
 #ifndef JULIA_ENABLE_THREADING
 
 void jl_init_threading(void)