remove Scheduler

base/boot.jl

@@ -107,6 +107,7 @@
 #     last::Task
 #     storage::Any
 #     consumers
+#     started::Bool
 #     done::Bool
 #     runnable::Bool
 # end

base/client.jl

@@ -325,21 +325,15 @@ function init_load_path()
     push!(LOAD_PATH,abspath(JULIA_HOME,"..","share","julia","site",vers))
 end
 
-function init_sched()
-    global const Workqueue = Any[]
-end
+global const Workqueue = Any[]
 
 function init_head_sched()
     # start in "head node" mode
-    global const Scheduler = Task(()->event_loop(true), 1024*1024)
     global PGRP
     global LPROC
     LPROC.id = 1
     assert(length(PGRP.workers) == 0)
     register_worker(LPROC)
-    # make scheduler aware of current (root) task
-    unshift!(Workqueue, roottask)
-    yield()
 end
 
 function init_profiler()
@@ -376,10 +370,10 @@ function _start()
     LinAlg.init()
     GMP.gmp_init()
     init_profiler()
+    start_gc_msgs_task()
 
     #atexit(()->flush(STDOUT))
     try
-        init_sched()
         any(a->(a=="--worker"), ARGS) || init_head_sched()
         init_load_path()
         (quiet,repl,startup,color_set,history) = process_options(copy(ARGS))

base/inference.jl

@@ -1093,26 +1093,6 @@ f_argnames(ast) =
 
 is_rest_arg(arg::ANY) = (ccall(:jl_is_rest_arg,Int32,(Any,), arg) != 0)
 
-# function typeinf_task(caller)
-#     result = ()
-#     while true
-#         (caller, args) = yieldto(caller, result)
-#         result = typeinf_ext_(args...)
-#     end
-# end
-
-#Inference_Task = Task(typeinf_task, 2097152)
-#yieldto(Inference_Task, current_task())
-
-#function typeinf_ext(linfo, atypes, sparams, cop)
-    #C = current_task()
-    #args = (linfo, atypes, sparams, cop)
-    #if is(C, Inference_Task)
-    #    return typeinf_ext_(args...)
-    #end
-    #return yieldto(Inference_Task, C, args)
-#end
-
 function typeinf_ext(linfo, atypes::ANY, sparams::ANY, def)
     global inference_stack
     last = inference_stack

base/multi.jl

@@ -394,24 +394,6 @@ type RemoteRef
     RemoteRef(w::Worker) = RemoteRef(w.id)
     RemoteRef() = RemoteRef(myid())
 
-    global WeakRemoteRef
-    function WeakRemoteRef(w, wh, id)
-        return new(w, wh, id)
-    end
-
-    function WeakRemoteRef(pid::Integer)
-        rr = WeakRemoteRef(pid, myid(), REQ_ID)
-        REQ_ID += 1
-        if mod(REQ_ID,200) == 0
-            gc()
-        end
-        rr
-    end
-
-    WeakRemoteRef(w::LocalProcess) = WeakRemoteRef(myid())
-    WeakRemoteRef(w::Worker) = WeakRemoteRef(w.id)
-    WeakRemoteRef() = WeakRemoteRef(myid())
-
     global next_id
     next_id() = (id=(myid(),REQ_ID); REQ_ID+=1; id)
 end
@@ -467,7 +449,14 @@ function del_clients(pairs::(Any,Any)...)
     end
 end
 
-any_gc_flag = false
+any_gc_flag = Condition()
+function start_gc_msgs_task()
+    enq_work(Task(()->
+        while true
+            wait(any_gc_flag)
+            flush_gc_msgs()
+        end))
+end
 
 function send_del_client(rr::RemoteRef)
     if rr.where == myid()
@@ -480,7 +469,7 @@ function send_del_client(rr::RemoteRef)
         w = worker_from_id(rr.where)
         push!(w.del_msgs, (rr2id(rr), myid()))
         w.gcflag = true
-        global any_gc_flag = true
+        notify(any_gc_flag)
     end
 end
 
@@ -508,7 +497,7 @@ function send_add_client(rr::RemoteRef, i)
         #println("$(myid()) adding $((rr2id(rr), i)) for $(rr.where)")
         push!(w.add_msgs, (rr2id(rr), i))
         w.gcflag = true
-        global any_gc_flag = true
+        notify(any_gc_flag)
     end
 end
 
@@ -735,35 +724,6 @@ end
 take_ref(rid) = take(lookup_ref(rid))
 take(rr::RemoteRef) = call_on_owner(take_ref, rr)
 
-## work queue ##
-
-function enq_work(t::Task)
-    ccall(:uv_stop,Void,(Ptr{Void},),eventloop())
-    unshift!(Workqueue, t)
-end
-
-function perform_work()
-    perform_work(pop!(Workqueue))
-end
-
-function perform_work(t::Task)
-    if !istaskstarted(t)
-        # starting new task
-        yieldto(t)
-    else
-        # continuing interrupted work item
-        arg = t.result
-        t.result = nothing
-        t.runnable = true
-        yieldto(t, arg)
-    end
-    t = current_task().last
-    if t.runnable
-        # still runnable; return to queue
-        enq_work(t)
-    end
-end
-
 function deliver_result(sock::IO, msg, oid, value)
     #print("$(myid()) sending result $oid\n")
     if is(msg,:call_fetch)
@@ -956,11 +916,9 @@ function start_worker(out::IO)
 
     ccall(:jl_install_sigint_handler, Void, ())
 
-    global const Scheduler = current_task()
-
     try
         check_master_connect(60.0)
-        event_loop(false)
+        wait()
     catch err
         print(STDERR, "unhandled exception on $(myid()): $(err)\nexiting.\n")
     end
@@ -1216,37 +1174,6 @@ end
 
 ## higher-level functions: spawn, pmap, pfor, etc. ##
 
-sync_begin() = task_local_storage(:SPAWNS, ({}, get(task_local_storage(), :SPAWNS, ())))
-
-function sync_end()
-    spawns = get(task_local_storage(), :SPAWNS, ())
-    if is(spawns,())
-        error("sync_end() without sync_begin()")
-    end
-    refs = spawns[1]
-    task_local_storage(:SPAWNS, spawns[2])
-    for r in refs
-        wait(r)
-    end
-end
-
-macro sync(block)
-    quote
-        sync_begin()
-        v = $(esc(block))
-        sync_end()
-        v
-    end
-end
-
-function sync_add(r)
-    spawns = get(task_local_storage(), :SPAWNS, ())
-    if !is(spawns,())
-        push!(spawns[1], r)
-    end
-    r
-end
-
 let nextidx = 1
     global chooseproc
     function chooseproc(thunk::Function)
@@ -1297,23 +1224,6 @@ macro fetchfrom(p, expr)
     :(remotecall_fetch($(esc(p)), $(esc(expr))))
 end
 
-function spawnlocal(thunk)
-    t = Task(thunk)
-    sync_add(t)
-    enq_work(t)
-    t
-end
-
-macro async(expr)
-    expr = localize_vars(:(()->($expr)), false)
-    :(spawnlocal($(esc(expr))))
-end
-
-macro spawnlocal(expr)
-    warn_once("@spawnlocal is deprecated, use @async instead.")
-    :(@async $(esc(expr)))
-end
-
 function at_each(f, args...)
     for w in PGRP.workers
         sync_add(remotecall(w.id, f, args...))
@@ -1531,75 +1441,6 @@ end
 #     2/(nc/niter)
 # end
 
-## event processing, I/O and work scheduling ##
-
-function yield(args...)
-    ct = current_task()
-    # preserve Task.last across calls to the scheduler
-    prev = ct.last
-    v = yieldto(Scheduler, args...)
-    ct.last = prev
-    return v
-end
-
-function pause()
-    @unix_only    ccall(:pause, Void, ())
-    @windows_only ccall(:Sleep,stdcall, Void, (Uint32,), 0xffffffff)
-end
-
-function event_loop(isclient)
-    iserr, lasterr, bt = false, nothing, {}
-    while true
-        try
-            if iserr
-                display_error(lasterr, bt)
-                println(STDERR)
-                iserr, lasterr, bt = false, nothing, {}
-            else
-                while true
-                    if isempty(Workqueue)
-                        if any_gc_flag
-                            flush_gc_msgs()
-                        end
-                        c = process_events(true)
-                        if c==0 && eventloop()!=C_NULL && isempty(Workqueue) && !any_gc_flag
-                            # if there are no active handles and no runnable tasks, just
-                            # wait for signals.
-                            pause()
-                        end
-                    else
-                        perform_work()
-                        process_events(false)
-                    end
-                end
-            end
-        catch err
-            if iserr
-                ccall(:jl_, Void, (Any,), (
-                    "\n!!!An ERROR occurred while printing the last error!!!\n",
-                    lasterr,
-                    bt
-                ))
-            end
-            iserr, lasterr = true, err
-            bt = catch_backtrace()
-            if isclient && isa(err,InterruptException)
-                # root task is waiting for something on client. allow C-C
-                # to interrupt.
-                interrupt_waiting_task(roottask,err)
-                iserr, lasterr = false, nothing
-            end
-        end
-    end
-end
-
-# force a task to stop waiting with an exception
-function interrupt_waiting_task(t::Task, err)
-    if !t.runnable
-        t.exception = err
-        enq_work(t)
-    end
-end
 
 function check_master_connect(timeout)
     # If we do not have at least process 1 connect to us within timeout

base/precompile.jl

@@ -7,7 +7,6 @@ precompile(start, (Dict{Any,Any},))
 precompile(perform_work, ())
 precompile(isempty, (Array{Any,1},))
 precompile(getindex, (Dict{Any,Any}, Int32))
-precompile(event_loop, (Bool,))
 precompile(_start, ())
 precompile(process_options, (Array{Any,1},))
 precompile(run_repl, ())

base/stream.jl

@@ -723,50 +723,33 @@ end
 write(s::TTY, p::Ptr, nb::Integer) = @uv_write nb ccall(:jl_write_no_copy, Int32, (Ptr{Void}, Ptr{Void}, Uint, Ptr{Void}, Ptr{Void}), handle(s), p, nb, uvw, uv_jl_writecb::Ptr{Void})
 
 function write(s::AsyncStream, b::Uint8)
-    if isdefined(Main.Base,:Scheduler) && current_task() != Main.Base.Scheduler
-        @uv_write 1 ccall(:jl_putc_copy, Int32, (Uint8, Ptr{Void}, Ptr{Void}, Ptr{Void}), b, handle(s), uvw, uv_jl_writecb_task::Ptr{Void})
-        uv_req_set_data(uvw,current_task())
-        wait()
-    else
-        @uv_write 1 ccall(:jl_putc_copy, Int32, (Uint8, Ptr{Void}, Ptr{Void}, Ptr{Void}), b, handle(s), uvw, uv_jl_writecb::Ptr{Void})
-    end
+    @uv_write 1 ccall(:jl_putc_copy, Int32, (Uint8, Ptr{Void}, Ptr{Void}, Ptr{Void}), b, handle(s), uvw, uv_jl_writecb_task::Ptr{Void})
+    uv_req_set_data(uvw,current_task())
+    wait()
     return 1
 end
 function write(s::AsyncStream, c::Char)
-    if isdefined(Main.Base,:Scheduler) && current_task() != Main.Base.Scheduler
-        @uv_write utf8sizeof(c) ccall(:jl_pututf8_copy, Int32, (Ptr{Void},Uint32, Ptr{Void}, Ptr{Void}), handle(s), c, uvw, uv_jl_writecb_task::Ptr{Void})
-        uv_req_set_data(uvw,current_task())
-        wait()
-    else
-        @uv_write utf8sizeof(c) ccall(:jl_pututf8_copy, Int32, (Ptr{Void},Uint32, Ptr{Void}, Ptr{Void}), handle(s), c, uvw, uv_jl_writecb::Ptr{Void})
-    end
+    @uv_write utf8sizeof(c) ccall(:jl_pututf8_copy, Int32, (Ptr{Void},Uint32, Ptr{Void}, Ptr{Void}), handle(s), c, uvw, uv_jl_writecb_task::Ptr{Void})
+    uv_req_set_data(uvw,current_task())
+    wait()
     return utf8sizeof(c)
 end
 function write{T}(s::AsyncStream, a::Array{T})
     if isbits(T)
-        if isdefined(Main.Base,:Scheduler) && current_task() != Main.Base.Scheduler
-            n = uint(length(a)*sizeof(T))
-            @uv_write n ccall(:jl_write_no_copy, Int32, (Ptr{Void}, Ptr{Void}, Uint, Ptr{Void}, Ptr{Void}), handle(s), a, n, uvw, uv_jl_writecb_task::Ptr{Void})
-            uv_req_set_data(uvw,current_task())
-            wait()
-        else
-            write!(s,copy(a))
-        end
+        n = uint(length(a)*sizeof(T))
+        @uv_write n ccall(:jl_write_no_copy, Int32, (Ptr{Void}, Ptr{Void}, Uint, Ptr{Void}, Ptr{Void}), handle(s), a, n, uvw, uv_jl_writecb_task::Ptr{Void})
+        uv_req_set_data(uvw,current_task())
+        wait()
         return int(length(a)*sizeof(T))
     else
         check_open(s)
         invoke(write,(IO,Array),s,a)
     end
 end
 function write(s::AsyncStream, p::Ptr, nb::Integer)
-    if isdefined(Main.Base,:Scheduler) && current_task() != Main.Base.Scheduler
-        @uv_write nb ccall(:jl_write_no_copy, Int32, (Ptr{Void}, Ptr{Void}, Uint, Ptr{Void}, Ptr{Void}), handle(s), p, nb, uvw, uv_jl_writecb_task::Ptr{Void})
-        uv_req_set_data(uvw,current_task())
-        wait()
-    else
-        check_open(s)
-        ccall(:jl_write, Uint, (Ptr{Void},Ptr{Void},Uint), handle(s), p, uint(nb))
-    end
+    @uv_write nb ccall(:jl_write_no_copy, Int32, (Ptr{Void}, Ptr{Void}, Uint, Ptr{Void}, Ptr{Void}), handle(s), p, nb, uvw, uv_jl_writecb_task::Ptr{Void})
+    uv_req_set_data(uvw,current_task())
+    wait()
     return int(nb)
 end
 

base/sysimg.jl

@@ -103,7 +103,7 @@ include("process.jl")
 include("multimedia.jl")
 importall .Multimedia
 reinit_stdio()
-ccall(:jl_get_uv_hooks, Void, ())
+ccall(:jl_get_uv_hooks, Void, (Cint,), 0)
 include("grisu.jl")
 import .Grisu.print_shortest
 include("printf.jl")