diff --git a/Project.toml b/Project.toml index d662a16..e69062e 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "ParallelProcessingTools" uuid = "8e8a01fc-6193-5ca1-a2f1-20776dae4199" -version = "0.4.10" +version = "0.4.11" [deps] ArgCheck = "dce04be8-c92d-5529-be00-80e4d2c0e197" diff --git a/ext/ParallelProcessingToolsThreadPinningExt.jl b/ext/ParallelProcessingToolsThreadPinningExt.jl index 04f249a..632da77 100644 --- a/ext/ParallelProcessingToolsThreadPinningExt.jl +++ b/ext/ParallelProcessingToolsThreadPinningExt.jl @@ -200,4 +200,4 @@ ThreadPinning.pinthreads(::ParallelProcessingTools.AutoThreadPinning) = nothing end # if _threadpinning_supported -end # module ChangesOfVariablesInverseFunctionsExt +end # module ParallelProcessingToolsThreadPinningExt diff --git a/src/display.jl b/src/display.jl index e79b0ff..b1e4409 100644 --- a/src/display.jl +++ b/src/display.jl @@ -24,7 +24,7 @@ const _g_unicode_occupancy = ( """ - ParallelProcessingTools.in_vscode_notebook():Bool + ParallelProcessingTools.in_vscode_notebook()::Bool Test if running within a Visual Studio Code notebook. """ @@ -32,10 +32,10 @@ in_vscode_notebook() = haskey(ENV, "VSCODE_CWD") """ - ParallelProcessingTools.printover(f_show::Function, io::IOBuffer) + ParallelProcessingTools.printover(f_show::Function, io::IO) -Runs `f_show(tmpio)` with an IO buffer, then clears the required number of -lines on `io` (typically `stdout`) and prints the output over them. +Runs `f_show(tmpio)` with a temporary IO buffer, then clears the required +number of lines on `io` (typically `stdout`) and prints the output over them. """ function printover(f_show, io) vscode_nb_mode = in_vscode_notebook() diff --git a/src/exceptions.jl b/src/exceptions.jl index d424480..c960605 100644 --- a/src/exceptions.jl +++ b/src/exceptions.jl @@ -34,9 +34,8 @@ original_exception(err::RemoteException) = original_exception(err.captured.ex) """ ParallelProcessingTools.onlyfirst_exception(err) -Replaces `CompositeException`s with their first exception. - -Also employs `inner_exception` if `simplify` is `true`. +Replaces `CompositeException`s with their first exception. Leaves other +exceptions unchanged. """ function onlyfirst_exception end export onlyfirst_exception @@ -54,7 +53,7 @@ If multiple exceptions originate from parallel code in `expr`, only one is rethrown, and `TaskFailedException`s and `RemoteException`s are replaced by the original exceptions that caused them. -See [`inner_exception`] and [`onlyfirst_exception`](@ref). +See [`inner_exception`](@ref) and [`onlyfirst_exception`](@ref). """ macro userfriendly_exceptions(expr) quote diff --git a/src/fileio.jl b/src/fileio.jl index 10ceacd..faadba8 100644 --- a/src/fileio.jl +++ b/src/fileio.jl @@ -87,7 +87,7 @@ end ParallelProcessingTools.default_cache_dir()::String Returns the default cache directory, e.g. for [`write_files`](@ref) and -`read_files`(@ref). +[`read_files`](@ref). See also [`default_cache_dir!`](@ref). """ @@ -114,7 +114,7 @@ end Sets the default cache directory to `dir` and returns it. -See also [`default_cache_dir!`](@ref). +See also [`default_cache_dir`](@ref). """ function default_cache_dir!(dir::AbstractString) lock(_g_default_cachedir_lock) do @@ -180,8 +180,8 @@ _should_overwrite_if_necessary(::CreateOrIgnore) = false """ CreateNew() isa WriteMode -Indicates that new files should be created and to throw and eror if the files -already exist. +Indicates that new files should be created and that an error should be thrown +if the files already exist. See [`WriteMode`](@ref) and [`write_files`](@ref). """ @@ -190,7 +190,7 @@ export CreateNew function _already_done(::CreateNew, target_fnames::AbstractVector{<:String}, any_pre_existing::Bool, all_pre_existing::Bool, loglevel::LogLevel) if any_pre_existing - throw(ErrorException("Some, but not all of $target_fnames exist, but not allowed to replace files")) + throw(ErrorException("Some or all of $target_fnames exist already, but not allowed to replace files")) else return false end @@ -220,7 +220,7 @@ _should_overwrite_if_necessary(::CreateOrReplace) = true """ - CreateOrIgnore() isa WriteMode + CreateOrModify() isa WriteMode Indicates that either new files should be created, or that existing files should be modified. @@ -279,7 +279,7 @@ write to. With `ftw::FilesToWrite`, use `collect(ftw)` or `iterate(ftw)` to access the filenames to write to. Use `close(ftw)` or `close(ftw, true)` to close things in good order, indicating success, and use `close(ftw, false)` or -`close(ftw, err:Exception)` to abort, indicating failure. +`close(ftw, err::Exception)` to abort, indicating failure. See [`write_files`](@ref) for example code. @@ -436,9 +436,9 @@ on the OS and file-system used). [`ModifyExisting()`](@ref). If a writing function `f_write` is given, calls -`f_create(temporary_filenames...)`. If `f_create` doesn't throw an exception, +`f_write(temporary_filenames...)`. If `f_write` doesn't throw an exception, the files `temporary_filenames` are renamed to `filenames`, otherwise -the temporary files are are either deleted (if `delete_tmp_onerror` is `true) +the temporary files are either deleted (if `delete_tmp_onerror` is `true`) or left in place (e.g. for debugging purposes). Set `ENV["JULIA_DEBUG"] = "ParallelProcessingTools"` to see a log of all @@ -457,7 +457,7 @@ end files were (re-)written or `nothing` if there was nothing to do (depending on `mode`). -If no writing funcion `f_write` is given then, `write_files` returns an object +If no writing function `f_write` is given then, `write_files` returns an object of type [`FilesToWrite`](@ref) that holds the temporary filenames. Closing it will, like above, either rename temporary files to `filenames` or remove them. So @@ -524,7 +524,7 @@ end function write_files( @nospecialize(filenames::AbstractString...); mode::WriteMode = CreateOrIgnore(), - use_cache::Bool = false, @nospecialize(cache_dirname::AbstractString = default_cache_dir()), + use_cache::Bool = false, @nospecialize(cache_dir::AbstractString = default_cache_dir()), create_dirs::Bool = true, delete_tmp_onerror::Bool=true, verbose::Bool = false ) @@ -532,7 +532,7 @@ function write_files( loglevel = verbose ? Info : Debug - cache_dir = String(cache_dirname) # Fix type + cache_dir = String(cache_dir) # Fix type target_fnames = String[filenames...] # Fix type staging_fnames = String[] cache_fnames = String[] @@ -643,9 +643,9 @@ function Base.close(ftr::FilesToRead, @nospecialize(reason::Union{Bool,Exception ftr._isopen[] = false if reason == true - # @debug "Reading from to $(ftr._source_fnames) was indicated to have succeeded." + # @debug "Reading from $(ftr._source_fnames) was indicated to have succeeded." elseif reason == false - @debug "Reading from to $(ftr._source_fnames) was indicated to have failed." + @debug "Reading from $(ftr._source_fnames) was indicated to have failed." else @debug "Aborted reading from $(ftr._source_fnames) due to exception:" reason end @@ -704,7 +704,7 @@ result = read_files("foo.txt", "bar.txt", use_cache = true) do foo, bar end ``` -If no reading funcion `f_read` is given, then `read_files` returns an object +If no reading function `f_read` is given, then `read_files` returns an object of type [`FilesToRead`](@ref) that holds the temporary filenames. Closing it will clean up temporary files, like described above. So diff --git a/src/htcondor.jl b/src/htcondor.jl index 1423754..bf0caec 100644 --- a/src/htcondor.jl +++ b/src/htcondor.jl @@ -83,7 +83,7 @@ function _generate_condor_worker_script(filename, runmode::OnHTCondor, manager:: additional_julia_flags = `$jl_threads_flag $jl_heap_size_hint_flag $julia_flags` worker_cmd = worker_local_startcmd( manager; - julia_flags = `$julia_flags $additional_julia_flags`, + julia_flags = additional_julia_flags, redirect_output = runmode.redirect_output, env = runmode.env ) depot_path = join(runmode.julia_depot, ":") diff --git a/src/localhost.jl b/src/localhost.jl index f8ddb07..2c92be2 100644 --- a/src/localhost.jl +++ b/src/localhost.jl @@ -16,12 +16,12 @@ Example: runmode = OnLocalhost(n = 4) task, n = runworkers(runmode) -Threads.@async begin +@async begin wait(task) - @info "SLURM workers have terminated." + @info "Local workers have terminated." end -@wait_while nprocs()-1 < n) +@wait_while nprocs()-1 < n ``` Workers can also be started manually, use @@ -56,7 +56,7 @@ end function runworkers(runmode::OnLocalhost, manager::ElasticManager) start_cmd, m, n = worker_start_command(runmode, manager) - task = Threads.@async begin + task = @async begin processes = Base.Process[] for _ in 1:m push!(processes, open(start_cmd)) diff --git a/src/memory.jl b/src/memory.jl index ea49b61..397b647 100644 --- a/src/memory.jl +++ b/src/memory.jl @@ -21,7 +21,7 @@ Values of `-1` mean unlimited. !!! note Currently only works on Linux, simply returns `(Int64(-1), Int64(-1))` on - other operationg systems. + other operating systems. """ function memory_limit end export memory_limit @@ -75,5 +75,5 @@ export memory_limit! return rlim[].cur, rlim[].max end else - memory_limit!(soft_limit::Integer, hard_limit::Integer) = Int64(-1), Int64(-1) + memory_limit!(soft_limit::Integer, hard_limit::Integer = Int64(-1)) = Int64(-1), Int64(-1) end diff --git a/src/onthreads.jl b/src/onthreads.jl index 83ec6c5..76d6d50 100644 --- a/src/onthreads.jl +++ b/src/onthreads.jl @@ -19,9 +19,8 @@ end # Adapted from Julia PR 32477: function _threading_run(func, threadsel::AbstractVector{<:Integer}) - tasks = Vector{Task}(undef, length(eachindex(threadsel))) - for tid in threadsel - i = firstindex(tasks) + (tid - first(threadsel)) + tasks = Vector{Task}(undef, length(threadsel)) + for (i, tid) in enumerate(threadsel) tasks[i] = _run_on(Task(func), tid) end foreach(wait, tasks) @@ -52,9 +51,12 @@ end """ allthreads() -Convencience function, returns an equivalent of `1:Base.Threads.nthreads()`. +Returns the thread IDs of all threads in the default threadpool. + +Returns an equivalent of `1:Base.Threads.nthreads()` if no interactive +threads are present. """ -allthreads() = Base.OneTo(Base.Threads.nthreads()) +allthreads() = Threads.threadpooltids(:default) export allthreads @@ -65,11 +67,11 @@ Execute code in `expr` in parallel on the threads in `threadsel`. `threadsel` should be a single thread-ID or a range (or array) of thread-ids. If `threadsel == Base.Threads.threadid()`, `expr` is run on the current -tread with only minimal overhead. +thread with only minimal overhead. Example 1: -```juliaexpr +```julia tlsum = ThreadLocal(0.0) data = rand(100) @onthreads allthreads() begin @@ -81,13 +83,13 @@ sum(getallvalues(tlsum)) ≈ sum(data) Example 2: ```julia -# Assuming 4 threads: +# Assuming 4 threads in the default threadpool: tl = ThreadLocal(42) -threadsel = 2:3 +threadsel = allthreads()[2:3] @onthreads threadsel begin tl[] = Base.Threads.threadid() end -getallvalues(tl)[threadsel] == [2, 3] +getallvalues(tl)[2:3] == threadsel getallvalues(tl)[[1,4]] == [42, 42] ``` """ @@ -99,7 +101,6 @@ export @onthreads function ThreadLocal{T}(f::Base.Callable) where {T} result = ThreadLocal{T}(undef) - result.value @onthreads allthreads() result.value[threadid()] = f() result end @@ -115,12 +116,13 @@ multi-threaded tasks. Example: -``` +```julia @mt_out_of_order begin a = foo() bar() c = baz() end +``` will run `a = foo()`, `bar()` and `c = baz()` in parallel and in arbitrary order, results of assignments will appear in the outside scope. @@ -145,7 +147,6 @@ macro mt_out_of_order(ex) exprs[i] = esc(exprs[i]) end elseif exprs[i] isa Expr - ftvar = gensym() exprs[i] = :(push!($tasks, Base.Threads.@spawn($(esc(exprs[i]))))) push!(handle_results, :(wait(popfirst!($tasks)))) else diff --git a/src/onworkers.jl b/src/onworkers.jl index 6484c49..1331e19 100644 --- a/src/onworkers.jl +++ b/src/onworkers.jl @@ -4,7 +4,7 @@ """ TimelimitExceeded <: Exception -Exception thrown something timed out. +Exception thrown when something times out. """ struct TimelimitExceeded <: Exception max_time::Float64 @@ -48,166 +48,132 @@ terminated. If a problem occurs (maxtime or worker failure) while running the activity, reschedules the task if the maximum number of tries has not yet been reached, -otherwise throws an exception. +otherwise throws an exception. Worker failures do not count against `tries`, +but only up to `3 * tries` worker failures are tolerated. """ -function onworker end -export onworker - function onworker( - f::Function; + f::Function, args...; @nospecialize(pool::AbstractWorkerPool = ppt_worker_pool()), @nospecialize(maxtime::Real = 0), @nospecialize(tries::Integer = 1), @nospecialize(label::AbstractString = "") ) - R = _return_type(f, ()) - untyped_result = _on_worker_impl(f, (), pool, Float64(maxtime), Int(tries), String(label)) + R = Base.promote_op(f, map(typeof, args)...) + untyped_result = _on_worker_impl(f, args, pool, Float64(maxtime), Int(tries), String(label)) return convert(R, untyped_result)::R end +export onworker -function onworker( - f::Function, arg1, args...; - @nospecialize(pool::AbstractWorkerPool = ppt_worker_pool()), - @nospecialize(maxtime::Real = 0), @nospecialize(tries::Integer = 1), @nospecialize(label::AbstractString = "") -) - all_args = (arg1, args...) - R = _return_type(f, all_args) - untyped_result = _on_worker_impl(f, all_args, pool, Float64(maxtime), Int(tries), String(label)) - @assert !(untyped_result isa Exception) - return convert(R, untyped_result)::R +# Outcome of a single attempt to run an activity on a worker: +struct _AttemptSucceeded; value::Any; end +struct _AttemptFailed; err::Exception; retriable::Bool; end +struct _AttemptTimedOut; elapsed::Float64; end +struct _WorkerUnusable; err::Exception; end + +const _AttemptOutcome = Union{_AttemptSucceeded,_AttemptFailed,_AttemptTimedOut,_WorkerUnusable} + +function _attempt_onworker(@nospecialize(f::Function), @nospecialize(args::Tuple), worker::Int, maxtime::Float64) + t_start = time() + try + future_result = remotecall(f, worker, args...) + if maxtime > 0 + wait_for_any(future_result, maxtime = maxtime) + isready(future_result) || return _AttemptTimedOut(time() - t_start) + end + # With a `remotecall` to the current process, fetch will return exceptions + # originating in the called function, while if run on a remote process they + # will be thrown to the caller of fetch. `@return_exceptions` unifies this: + result = @return_exceptions fetch(future_result) + return result isa Exception ? _classify_failure(result) : _AttemptSucceeded(result) + catch err + if err isa Union{ProcessExitedException,RemoteException} + return _classify_failure(err) + else + rethrow() + end + end end -_return_type(f, args::Tuple) = Core.Compiler.return_type(f, typeof(args)) +function _classify_failure(err::Exception) + orig_err = inner_exception(err) + if orig_err isa ProcessExitedException + return _WorkerUnusable(err) + elseif orig_err isa MethodError && _worker_seems_corrupted(orig_err) + return _WorkerUnusable(err) + else + return _AttemptFailed(err, _should_retry(err)) + end +end +# A method that exists locally but is missing on the worker indicates a +# corrupted (serializer-)state on the worker: +function _worker_seems_corrupted(err::MethodError) + func_module = nameof(parentmodule(parentmodule(typeof(err.f)))) + func_module == :Serialization && hasmethod(err.f, map(typeof, err.args)) +end @noinline function _on_worker_impl( @nospecialize(f::Function), @nospecialize(args::Tuple), @nospecialize(pool::AbstractWorkerPool), maxtime::Float64, tries::Int, label::String ) + activity = _Activity(f, label, tries) n_tries::Int = 0 - while n_tries < tries - n_tries += 1 - activity = _Activity(f, label, tries) + n_workers_lost::Int = 0 + max_workers_lost = 3 * tries + while true + n_tries += 1 @debug "Preparing to run $activity, taking a worker from $(getlabel(pool))" worker = take!(pool) - start_time = time() - elapsed_time = zero(start_time) - - try + outcome::_AttemptOutcome = try @debug "Running $activity on worker $worker" + _attempt_onworker(f, args, worker, maxtime) + finally + put!(pool, worker) + end - future_result = remotecall(f, worker, args...) - - result_isready = try - if maxtime > 0 - # May throw an exception: - wait_for_any(future_result, maxtime = maxtime) - else - # May throw an exception: - wait(future_result) - end - elapsed_time = time() - start_time - - isready(future_result) - catch err - # Testing if future is ready may throw exceptions from f already: - if _should_retry(err) - if !(n_tries < tries) - inner_err = inner_exception(err) - throw(MaxTriesExceeded(tries, n_tries, inner_err)) - else - @debug "Will retry $activity ($n_tries tries so far) due to" err - end - else - throw(err) - end - true + if outcome isa _AttemptSucceeded + @debug "Worker $worker ran $activity successfully" + return outcome.value + elseif outcome isa _WorkerUnusable + orig_err = inner_exception(outcome.err) + @warn "Worker $worker became unusable during $activity, removing it." orig_err + rmprocs(worker) + # Worker loss doesn't count as a try, but don't tolerate it indefinitely: + n_tries -= 1 + n_workers_lost += 1 + if n_workers_lost > max_workers_lost + throw(MaxTriesExceeded(tries, n_tries, orig_err)) end - - if result_isready - # With a `remotecall` to the current process, fetch will return exceptions - # originating in the called function, while if run on a remote process they - # will be thrown to the caller of fetch. We need to unify this behavior: - - fetched_result = @return_exceptions fetch(future_result) - - if fetched_result isa Exception - err = fetched_result - if _should_retry(err) - if !(n_tries < tries) - inner_err = inner_exception(err) - throw(MaxTriesExceeded(tries, n_tries, inner_err)) - else - @debug "Will retry $activity ($n_tries tries so far) due to" err - end - else - throw(err) - end - else - @debug "Worker $worker ran $activity successfully in $elapsed_time s" - return fetched_result - end + elseif outcome isa _AttemptTimedOut + @warn "Running $activity on worker $worker timed out after $(outcome.elapsed) s (max runtime $maxtime s)" + if worker == myid() + # ToDo: Cancel the task running the timed-out activity, once Julia + # supports robust task cancellation + # (see https://github.com/JuliaLang/julia/pull/60281): + @warn "Will not terminate main process $worker, making it available again, but it may still be running timed-out $activity" else - # Sanity check: if we got here, we must have timed out: - @assert maxtime > 0 && elapsed_time > maxtime - - @warn "Running $activity on worker $worker timed out after $elapsed_time s (max runtime $(maxtime) s)" - - if worker == myid() - @warn "Will not terminate main process $worker, making it available again, but it may still running timed-out $activity" - else - @warn "Terminating worker $worker due to activity maxtime" - rmprocs(worker) - end - - if !(n_tries < tries) - err = TimelimitExceeded(maxtime, elapsed_time) - @debug "Giving up on $activity after $n_tries tries due to" err - throw(MaxTriesExceeded(tries, n_tries, err)) - end - end - catch err - if err isa ProcessExitedException - @warn "Worker $worker seems to have terminated during $activity" - # This try doesn't count: - n_tries -= 1 - # Make certain that worker is really gone: + @warn "Terminating worker $worker due to activity maxtime" rmprocs(worker) - elseif err isa RemoteException - orig_err = inner_exception(err) - if orig_err isa MethodError - func = orig_err.f - func_args = orig_err.args - func_name = string(typeof(func)) - func_module = nameof(parentmodule(parentmodule(typeof(func)))) - func_hasmethod_local = hasmethod(func, map(typeof, func_args)) - if func_module == :Serialization && func_hasmethod_local - @warn "Function $func_name may be corrupted on worker $worker (missing method), terminating worker." - rmprocs(worker) - # This try doesn't count: - n_tries -= 1 - else - rethrow() - end - else - @debug "Encountered exception while trying to run $activity on worker $worker:" orig_err - rethrow() - end - elseif err isa MaxTriesExceeded - retry_reason = err.retry_reason - @debug "Giving up on $activity after $err.n_tries tries due to" retry_reason - rethrow() + end + if !(n_tries < tries) + err = TimelimitExceeded(maxtime, outcome.elapsed) + @debug "Giving up on $activity after $n_tries tries due to" err + throw(MaxTriesExceeded(tries, n_tries, err)) + end + elseif outcome isa _AttemptFailed + err = outcome.err + if !outcome.retriable + @debug "Encountered exception while trying to run $activity on worker $worker:" inner_exception(err) + throw(err) + elseif !(n_tries < tries) + @debug "Giving up on $activity after $n_tries tries due to" err + throw(MaxTriesExceeded(tries, n_tries, inner_exception(err))) else - @debug "Encountered unexpected exception while trying to run $activity on worker $worker:" err - rethrow() + @debug "Will retry $activity ($n_tries tries so far) due to" err end - finally - put!(pool, worker) end end - # Should never reach this point: - @assert false end diff --git a/src/procinit.jl b/src/procinit.jl index ff079dd..1bbb845 100644 --- a/src/procinit.jl +++ b/src/procinit.jl @@ -198,15 +198,14 @@ See also [`ParallelProcessingTools.get_procinit_code`](@ref) and _global_procinit_level[] = next_init_level _current_procinit_level[] = next_init_level - - return nothing finally unlock(allprocs_management_lock()) end if run_everywhere - ensure_procinit_or_kill(pids) + ensure_procinit_or_kill(workers()) end + return nothing end @@ -288,11 +287,8 @@ When using a [`FlexWorkerPool`](@ref), worker initialization can safely be run in the background though, as the pool will only offer workers (via `take!(pool)`) after it has fully initialized them. -See also [`ParallelProcessingTools.get_procinit_code`](@ref) -and [`ParallelProcessingTools.add_procinit_code`](@ref). - See also [`ParallelProcessingTools.get_procinit_code`](@ref), -[`ParallelProcessingTools.ensure_procinit`](@ref), +[`ParallelProcessingTools.add_procinit_code`](@ref), [`ParallelProcessingTools.global_procinit_level`](@ref) and [`ParallelProcessingTools.current_procinit_level`](@ref). """ @@ -362,7 +358,7 @@ function ensure_procinit_or_kill(pid::Int) ensure_procinit(pid) catch err orig_err = inner_exception(err) - @warn "Error while initializig process $pid, removing it." orig_err + @warn "Error while initializing process $pid, removing it." orig_err rmprocs(pid) end return nothing diff --git a/src/runworkers.jl b/src/runworkers.jl index da232b6..7e346a1 100644 --- a/src/runworkers.jl +++ b/src/runworkers.jl @@ -9,7 +9,7 @@ ParallelProcessingTools default thread pinning mode. Constructor: ```julia -AutoThreadPinning(; random::Bool = false, pin_blas::Bool = false) +AutoThreadPinning(; random::Bool = false, blas::Bool = false) ``` Arguments: @@ -155,7 +155,7 @@ function _get_elasticmgr_add_to_pool_callback(get_workerpool::Function = ppt_wor function mgr_add_too_pool(::ElasticManager, pid::Integer, op::Symbol) pool = get_workerpool()::AbstractWorkerPool if op == :register - Threads.@async begin + @async begin @debug "Adding process $pid to worker pool $(getlabel(pool))." push!(pool, pid) @debug "Added process $pid to worker pool $(getlabel(pool))." @@ -263,9 +263,10 @@ function _elastic_worker_startjl( """import ParallelProcessingTools; ParallelProcessingTools.elastic_worker("$cookie", "$address", $port, forward_stdout=$redirect_output, env=$env_vec)""" end -const _default_addprocs_params = Distributed.default_addprocs_params() - -_default_julia_cmd() = `$(_default_addprocs_params[:exename]) $(_default_addprocs_params[:exeflags])` +function _default_julia_cmd() + params = Distributed.default_addprocs_params() + `$(params[:exename]) $(params[:exeflags])` +end _default_julia_flags() = `` _default_julia_project() = Pkg.project().path diff --git a/src/slurm.jl b/src/slurm.jl index 1e1e794..d30b20e 100644 --- a/src/slurm.jl +++ b/src/slurm.jl @@ -21,9 +21,9 @@ Example: ```julia runmode = OnSlurm(slurm_flags = `--ntasks=4 --cpus-per-task=8 --mem-per-cpu=8G`) -task = runworkers(runmode) +task, n = runworkers(runmode) -Threads.@async begin +@async begin wait(task) @info "SLURM workers have terminated." end @@ -70,7 +70,7 @@ function worker_start_command(runmode::OnSlurm, manager::ElasticManager) worker_cmd = worker_local_startcmd( manager; - julia_flags = `$julia_flags $additional_julia_flags`, + julia_flags = additional_julia_flags, redirect_output = runmode.redirect_output, env = runmode.env ) @@ -103,7 +103,7 @@ end function runworkers(runmode::OnSlurm, manager::ElasticManager) srun_cmd, m, n = worker_start_command(runmode, manager) @info "Starting SLURM job: $srun_cmd" - task = Threads.@async begin + task = @async begin process = open(srun_cmd) wait(process) @info "SLURM job terminated: $srun_cmd" @@ -120,7 +120,7 @@ function _default_slurm_flags() end -const _slurm_memunits = IdDict{Char,Int}('K' => 1024^1, 'M' => 1024^2, 'G' => 1024^3, 'T' => 1024^4) +const _slurm_memunits = IdDict{Char,Int64}('K' => Int64(1024)^1, 'M' => Int64(1024)^2, 'G' => Int64(1024)^3, 'T' => Int64(1024)^4) const _slurm_memsize_regex = r"^([0-9]+)(([KMGT])B?)?$" function _slurm_parse_memoptval(memsize::AbstractString) @@ -129,7 +129,7 @@ function _slurm_parse_memoptval(memsize::AbstractString) if isnothing(m) throw(ArgumentError("Invalid SLURM memory size specification \"$s\"")) else - value = parse(Int, m.captures[1]) + value = parse(Int64, m.captures[1]) unitchar = only(something(m.captures[3], 'M')) unitmult = _slurm_memunits[unitchar] return value * unitmult @@ -150,11 +150,7 @@ function _slurm_parse_shortopt(opt::Char, args::Vector{String}, i::Int, default) throw(ArgumentError("Missing value for option \"-$opt\"")) end elseif startswith(arg, "-$opt") - if length(arg) > 2 - return arg[begin+2:end], i+1 - else - throw(ArgumentError("Missing value for option \"-$opt\"")) - end + return arg[begin+2:end], i+1 else return default, i end diff --git a/src/states.jl b/src/states.jl index 7a56594..e5deebd 100644 --- a/src/states.jl +++ b/src/states.jl @@ -1,10 +1,10 @@ # This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). """ - ParallelProcessingTools.NonZeroExitCode(cmd::Cmd, exitcode::Integer) isa Exception + ParallelProcessingTools.NonZeroExitCode(exitcode::Integer) isa Exception -Exception to indicate that a an external process running `cmd` failed with the -given exit code (not equal zero). +Exception to indicate that an external process failed with the given exit +code (not equal zero). """ struct NonZeroExitCode <: Exception exitcode::Int diff --git a/src/threadlocal.jl b/src/threadlocal.jl index a6252a3..0a3c967 100644 --- a/src/threadlocal.jl +++ b/src/threadlocal.jl @@ -7,12 +7,12 @@ Abstract type for thread-local values of type `T`. The value for the current thread is accessed via -`getindex(::AbstractThreadLocal)` and `setindex(::AbstractThreadLocal, x). +`getindex(::AbstractThreadLocal)` and `setindex!(::AbstractThreadLocal, x)`. To access both regular and thread-local values in a unified manner, use the function [`getlocalvalue`](@ref). -To get the all values across all threads, use the function +To get all values across all threads, use the function [`getallvalues`](@ref). Default implementation is [`ThreadLocal`](@ref). @@ -34,8 +34,9 @@ export getlocalvalue """ getallvalues(v::AbstractThreadLocal{T})::AbstractVector{T} -Access the all values (one for each thread) of a thread-local value as a -vector. Can only be called in single-threaded code sections. +Access all values (one for each thread in the default threadpool, in +thread-ID order) of a thread-local value as a vector. Can only be called +in single-threaded code sections. """ function getallvalues end export getallvalues @@ -62,6 +63,10 @@ ThreadLocal(value::T) where {T} ThreadLocal{T}(f::Base.Callable) where {T} ``` +`ThreadLocal{T}()` initializes the value on each thread to `T()`, +`ThreadLocal{T}(f)` to `f()`, and `ThreadLocal(value)` to a `deepcopy` +of `value`. + Examples: ```julia @@ -79,10 +84,10 @@ struct ThreadLocal{T} <: AbstractThreadLocal{T} value::Vector{T} ThreadLocal{T}(::UndefInitializer) where {T} = - new{T}(_protect_from_resize(Vector{T}(undef, nthreads()))) + new{T}(_protect_from_resize(Vector{T}(undef, Threads.maxthreadid()))) ThreadLocal{T}(value::T) where {T} = - new{T}(_protect_from_resize([deepcopy(value) for i in 1:nthreads()])) + new{T}(_protect_from_resize([deepcopy(value) for i in 1:Threads.maxthreadid()])) end export ThreadLocal @@ -123,6 +128,4 @@ Base.get!(x::ThreadLocal, default) = get!(() -> default, x) Base.isassigned(x::ThreadLocal) = isassigned(x.value, threadid()) -function getallvalues(x::ThreadLocal) - x.value -end +getallvalues(x::ThreadLocal) = view(x.value, allthreads()) diff --git a/src/threadsafe.jl b/src/threadsafe.jl index 9e66308..1b9148e 100644 --- a/src/threadsafe.jl +++ b/src/threadsafe.jl @@ -53,7 +53,7 @@ LockableIO(x::T) where {T<:IO} = LockableIO{T}(x) end @inline Base.read!(lio::LockableIO, args...; kwargs...) = map(lio) do io - read(io, args...; kwargs...) + read!(io, args...; kwargs...) end @inline Base.write(lio::LockableIO, args...; kwargs...) = map(lio) do io @@ -80,9 +80,9 @@ Example: @onthreads allthreads() begin @critical @info Base.Threads.threadid() end +``` Without `@critical`, the above will typically crash Julia. -``` """ macro critical(expr) quote diff --git a/src/waiting.jl b/src/waiting.jl index c9a3c3e..0906aeb 100644 --- a/src/waiting.jl +++ b/src/waiting.jl @@ -10,7 +10,7 @@ const _g_sleep_yield_threshold = 3 * _g_sleep_0_time_ns const _g_sleep_sleep_0_threshold = 3 * _g_sleep_t_time_ns """ - sleep_ns(t_in_ns::Real) + sleep_ns(t_in_ns::Integer) Sleep for `t_in_ns` nanoseconds, using a mixture of `yield()`, `sleep(0)` and `sleep(t)` to be able sleep for short times as well as long times with diff --git a/src/workerpool.jl b/src/workerpool.jl index 4c62da9..656ac9f 100644 --- a/src/workerpool.jl +++ b/src/workerpool.jl @@ -8,12 +8,12 @@ FlexWorkerPool(; caching = false, withmyid::Bool = true, kwargs...) -An flexible worker pool, intended to work with cluster managers that may +A flexible worker pool, intended to work with cluster managers that may add and remove Julia processes dynamically. If the current process (`Distributed.myid()`) is part of the pool, resp. if `withmyid` is `true`, it will be used as a fallback when no other workers are -in are members of the pool (e.g. because no other processes have been added +members of the pool (e.g. because no other processes have been added yet or because all other processes in the pool have terminated and been removed from it). The current process will *not* be used as a fallback when all other workers are currently in use. @@ -26,7 +26,7 @@ If `maxoccupancy`is greater than one, individual workers can be used ID `pid` multiple times without a `put!(pool, pid)` in between. Such a (ideally moderate) oversubscription can be useful to reduce latency-related idle times on workers: e.g. if communication latency to the worker -is not short compared the the runtime of the function called on them. Or if +is not short compared to the runtime of the function called on them. Or if the remote functions are often blocked waiting for I/O. Note: Workers still must be put back the same number of times they were taken from the pool, in total. @@ -236,10 +236,7 @@ function Base.take!(fwp::FlexWorkerPool) return pid catch err orig_err = inner_exception(err) - @warn "Error while initializig process $pid, removing it." orig_err - lock(fwp._worker_mgmt) do - fwp._worker_occupancy[pid] -= 1 - end + @warn "Error while initializing process $pid, removing it." orig_err rmprocs(pid) put!(fwp, pid) end @@ -323,7 +320,7 @@ end """ clear_worker_caches!(pool::AbstractWorkerPool) -Clear the worker caches (cached function closures, etc.) on the workers In +Clear the worker caches (cached function closures, etc.) on the workers in `pool`. Does nothing if the pool doesn't perform any on-worker caching. diff --git a/src/workpartition.jl b/src/workpartition.jl index fbf90ab..ff247f3 100644 --- a/src/workpartition.jl +++ b/src/workpartition.jl @@ -83,9 +83,11 @@ A = rand(100) # ... sub_A = workpart(A, workers(), myid()) # ... -idxs = workpart(eachindex(sub_A), allthreads(), threadid()) -for i in idxs - # ... +@onthreads allthreads() begin + idxs = workpart(eachindex(sub_A), allthreads(), threadid()) + for i in idxs + # ... + end end ``` """ @@ -111,10 +113,10 @@ end @deprecate( workpartition(A::AbstractArray, n::Integer, i::Integer), - workpart(A, 1::n, i) + workpart(A, 1:n, i) ) @deprecate( threadpartition(A::AbstractArray, n_threads::Integer = length(allthreads()), i::Integer = threadid()), - workpart(A, 1::n_threads, i) + workpart(A, 1:n_threads, i) ) diff --git a/test/runtests.jl b/test/runtests.jl index 84807a7..4c52b39 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -13,7 +13,9 @@ Test.@testset "Package ParallelProcessingTools" begin include("test_aqua.jl") include("test_memory.jl") + include("test_display.jl") include("test_waiting.jl") + include("test_exceptions.jl") include("test_states.jl") include("test_fileio.jl") include("test_threadsafe.jl") @@ -24,6 +26,8 @@ Test.@testset "Package ParallelProcessingTools" begin include("test_procinit.jl") include("test_workerpool.jl") include("test_onworkers.jl") + include("test_slurm.jl") + include("test_htcondor.jl") include("test_ext_threadpinning.jl") include("test_deprecated.jl") include("test_docs.jl") diff --git a/test/test_aqua.jl b/test/test_aqua.jl index 9a4e2b1..35286fa 100644 --- a/test/test_aqua.jl +++ b/test/test_aqua.jl @@ -5,7 +5,7 @@ import Aqua import ParallelProcessingTools Test.@testset "Package ambiguities" begin - Test.@test isempty(Test.detect_ambiguities(ParallelProcessingTools)) + Test.@test isempty(Test.detect_ambiguities(ParallelProcessingTools, recursive = true)) end # testset Test.@testset "Aqua tests" begin diff --git a/test/test_deprecated.jl b/test/test_deprecated.jl index 9e719f5..a52cf15 100644 --- a/test/test_deprecated.jl +++ b/test/test_deprecated.jl @@ -24,7 +24,7 @@ include("testtools.jl") end @testset "macro mt_async" begin - @test_deprecated begin + @test_deprecated r"deprecated"i begin n = 128 A = zeros(n) @sync for i in eachindex(A) @@ -39,7 +39,7 @@ include("testtools.jl") pids = classic_addprocs(2) @testset "macro mp_async" begin - @test_deprecated begin + @test_deprecated r"deprecated"i begin n = 128 A = Vector{Future}(undef, n) @sync for i in 1:n @@ -53,5 +53,11 @@ include("testtools.jl") end rmprocs(pids) + @testset "workpartition and threadpartition" begin + A = collect(1:10) + @test (@test_deprecated ParallelProcessingTools.workpartition(A, 3, 1)) == workpart(A, 1:3, 1) + @test (@test_deprecated ParallelProcessingTools.threadpartition(A, 2, 2)) == workpart(A, 1:2, 2) + end + @test_deprecated pinthreads_auto() isa Nothing end diff --git a/test/test_display.jl b/test/test_display.jl new file mode 100644 index 0000000..6a71cc7 --- /dev/null +++ b/test/test_display.jl @@ -0,0 +1,33 @@ +# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). + +using Test +using ParallelProcessingTools + +using ParallelProcessingTools: in_vscode_notebook, printover + +@testset "display" begin + withenv("VSCODE_CWD" => nothing) do + @test in_vscode_notebook() == false + + io = IOBuffer() + printover(io) do tmpio + println(tmpio, "foo") + println(tmpio, "bar") + end + output = String(take!(io)) + @test occursin("foo", output) + @test occursin("bar", output) + end + + withenv("VSCODE_CWD" => pwd()) do + @test in_vscode_notebook() == true + + io = IOBuffer() + printover(io) do tmpio + println(tmpio, "foo") + println(tmpio, "bar") + end + output = String(take!(io)) + @test occursin("foo | bar", output) + end +end diff --git a/test/test_exceptions.jl b/test/test_exceptions.jl new file mode 100644 index 0000000..960b1a7 --- /dev/null +++ b/test/test_exceptions.jl @@ -0,0 +1,69 @@ +# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). + +using Test +using ParallelProcessingTools + +using Distributed +using ParallelProcessingTools: inner_exception, original_exception, onlyfirst_exception + +@testset "exceptions" begin + err = ErrorException("Some error") + + throw_in_task(e) = try + wait(Threads.@spawn throw(e)) + catch tfe + tfe + end + + tfe = throw_in_task(err) + nested_tfe = try + wait(Threads.@spawn wait(Threads.@spawn throw(err))) + catch e + e + end + re = RemoteException(1, CapturedException(err, [])) + + @testset "inner_exception" begin + @test inner_exception(err) === err + @test tfe isa TaskFailedException + @test inner_exception(tfe) === err + @test inner_exception(re) === err + ce = inner_exception(CompositeException([tfe, re])) + @test ce isa CompositeException + @test all(e -> e === err, ce.exceptions) + @test inner_exception(nested_tfe) isa TaskFailedException + end + + @testset "original_exception" begin + @test original_exception(err) === err + @test original_exception(tfe) === err + @test original_exception(re) === err + @test original_exception(nested_tfe) === err + ce = original_exception(CompositeException([nested_tfe])) + @test only(ce.exceptions) === err + end + + @testset "onlyfirst_exception" begin + @test onlyfirst_exception(err) === err + @test onlyfirst_exception(CompositeException([tfe, re])) === tfe + end + + @testset "macro userfriendly_exceptions" begin + @test (@userfriendly_exceptions 42) == 42 + + caught = try + @userfriendly_exceptions @sync begin + Threads.@spawn throw(err) + Threads.@spawn throw(ErrorException("Some other error")) + end + catch e + e + end + @test caught isa ErrorException + end + + @testset "macro return_exceptions" begin + @test (@return_exceptions 42) == 42 + @test (@return_exceptions throw(err)) === err + end +end diff --git a/test/test_fileio.jl b/test/test_fileio.jl index 5f9ab6c..10c49bb 100644 --- a/test/test_fileio.jl +++ b/test/test_fileio.jl @@ -57,8 +57,11 @@ ENV["JULIA_DEBUG"] = old_julia_debug * ",ParallelProcessingTools" fn1 = joinpath(dir, "targetdir", "hello.txt") fn2 = joinpath(dir, "targetdir", "world.txt") + foo_fn = joinpath(dir, "foo.txt") + bar_fn = joinpath(dir, "bar.txt") + @test write_files() isa Nothing - ftw = write_files("foo.txt", "bar.txt", mode = CreateOrIgnore(), use_cache = use_cache) + ftw = write_files(foo_fn, bar_fn, mode = CreateOrIgnore(), use_cache = use_cache) @test ftw isa ParallelProcessingTools.FilesToWrite{CreateOrIgnore} tmp_foo, tmp_bar = ftw try @@ -71,10 +74,10 @@ ENV["JULIA_DEBUG"] = old_julia_debug * ",ParallelProcessingTools" close(ftw, err) end if !(throw_dummy_error || test_abort_write) - @test all(isfile, ["foo.txt", "bar.txt"]) - @test read.(["foo.txt", "bar.txt"], String) == ["Hello", "World"] - @test write_files("foo.txt", "bar.txt") isa Nothing - rm.(["foo.txt", "bar.txt"]) + @test all(isfile, [foo_fn, bar_fn]) + @test read.([foo_fn, bar_fn], String) == ["Hello", "World"] + @test write_files(foo_fn, bar_fn) isa Nothing + rm.([foo_fn, bar_fn]) end @test !in(tmp_foo, ParallelProcessingTools._g_files_to_clean_up) @test !in(tmp_bar, ParallelProcessingTools._g_files_to_clean_up) @@ -260,6 +263,74 @@ ENV["JULIA_DEBUG"] = old_julia_debug * ",ParallelProcessingTools" end end end + + @testset "write_files edge cases" begin + mktempdir() do dir + fn1 = joinpath(dir, "hello.txt") + fn2 = joinpath(dir, "world.txt") + + @test write_files() isa Nothing + @test write_files(() -> error("must not be called")) isa Nothing + + # Only some of the target files existing is an inconsistent state: + write(fn1, "Hello") + @test_throws ErrorException write_files(fn1, fn2) + @test_throws ErrorException write_files(fn1, fn2, mode = CreateOrModify()) + @test_throws ErrorException write_files(fn1, fn2, mode = ModifyExisting()) + rm(fn1) + + # Cache directory is created on demand: + cache_dir = joinpath(dir, "write_cache") + @test write_files(fn1, use_cache = true, cache_dir = cache_dir) do fn + write(fn, "Hello") + end == (fn1,) + @test isdir(cache_dir) + rm(fn1) + + # Closing twice is a no-op: + ftw = write_files(fn1) + write(only(collect(ftw)), "Hello") + close(ftw) + @test close(ftw) isa Nothing + @test read(fn1, String) == "Hello" + rm(fn1) + + # Target files that appear while writing are not replaced with CreateOrIgnore: + ftw = write_files(fn1) + write(only(collect(ftw)), "Hello") + write(fn1, "already there") + close(ftw) + @test read(fn1, String) == "already there" + @test !any(isfile, ftw._staging_fnames) + rm(fn1) + + # Only some target files appearing while writing is an error: + ftw = write_files(fn1, fn2) + foreach(fn -> write(fn, "Hello"), ftw) + write(fn1, "already there") + @test_throws ErrorException close(ftw) + rm(fn1) + end + end + + @testset "read_files edge cases" begin + mktempdir() do dir + fn1 = joinpath(dir, "hello.txt") + write(fn1, "Hello") + + @test_throws ErrorException read_files(fn -> error("Some error"), fn1) + + # Cache directory is created on demand: + cache_dir = joinpath(dir, "read_cache") + @test read_files(fn -> read(fn, String), fn1, use_cache = true, cache_dir = cache_dir) == "Hello" + @test isdir(cache_dir) + + # A closed FilesToRead can't be iterated anymore: + ftr = read_files(fn1, use_cache = false) + close(ftr) + @test_throws InvalidStateException collect(ftr) + end + end end ENV["JULIA_DEBUG"] = old_julia_debug; nothing diff --git a/test/test_htcondor.jl b/test/test_htcondor.jl new file mode 100644 index 0000000..b6e699e --- /dev/null +++ b/test/test_htcondor.jl @@ -0,0 +1,35 @@ +# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). + +using Test +using ParallelProcessingTools + +@testset "htcondor" begin + @testset "worker_start_command" begin + manager = ParallelProcessingTools.ppt_cluster_manager() + mktempdir(prefix = "ppt-htcondor-test") do dir + runmode = OnHTCondor( + n = 3, jobfile_dir = dir, julia_flags = `--depwarn=yes`, + condor_settings = Dict("request_memory" => "4GB") + ) + cmd, m, n = worker_start_command(runmode, manager) + @test m == 1 + @test n == 3 + @test first(cmd.exec) == "condor_submit" + + submit_file = last(cmd.exec) + @test isfile(submit_file) + submit_content = read(submit_file, String) + @test occursin("queue 3", submit_content) + @test occursin("request_memory=4GB", submit_content) + @test occursin("executable = /bin/bash", submit_content) + + worker_script = replace(submit_file, r"\.sub$" => ".sh") + @test isfile(worker_script) + script_content = read(worker_script, String) + @test occursin("JULIA_DEPOT_PATH", script_content) + @test count("--depwarn=yes", script_content) == 1 + @test occursin("--threads=1", script_content) + @test occursin("--heap-size-hint=2048M", script_content) + end + end +end diff --git a/test/test_memory.jl b/test/test_memory.jl index bce6468..cda8eb7 100644 --- a/test/test_memory.jl +++ b/test/test_memory.jl @@ -10,11 +10,13 @@ using ParallelProcessingTools new_limits = (limit, -1) if Sys.islinux() @test @inferred(memory_limit!(new_limits...)) == new_limits + @test @inferred(memory_limit!(limit)) == new_limits @test @inferred(memory_limit()) == new_limits stricter_limit = round(Int, 0.9*limit) @test_throws ArgumentError @inferred(memory_limit!(limit, stricter_limit)) else @test @inferred(memory_limit!(new_limits...)) == (-1, -1) + @test @inferred(memory_limit!(limit)) == (-1, -1) @test @inferred(memory_limit()) == (-1, -1) end end diff --git a/test/test_onprocs.jl b/test/test_onprocs.jl index 97ba661..8af0a9d 100644 --- a/test/test_onprocs.jl +++ b/test/test_onprocs.jl @@ -22,7 +22,7 @@ include("testtools.jl") @test (@onprocs workers() myid()) == workers() - threadinfo = [collect(1:n) for n in [fetch(@spawnat w nthreads()) for w in workers()]] + threadinfo = [fetch(@spawnat w allthreads()) for w in workers()] ref_result = ((w,t) -> (proc = w, threads = t)).(workers(), threadinfo) @test (@onprocs workers() begin diff --git a/test/test_onthreads.jl b/test/test_onthreads.jl index b195f27..bd336f9 100644 --- a/test/test_onthreads.jl +++ b/test/test_onthreads.jl @@ -32,11 +32,54 @@ using Base.Threads tl = ThreadLocal(0) @onthreads allthreads() tl[] = threadid() getallvalues(tl) - end) == 1:nthreads() + end) == allthreads() + end + + @testset "interactive threadpool" begin + @test allthreads() == Threads.threadpooltids(:default) + + prog = """ + using ParallelProcessingTools, Base.Threads + @assert nthreads(:interactive) == 1 && nthreads(:default) == 2 + @assert allthreads() == Threads.threadpooltids(:default) + tl = ThreadLocal(0) + @onthreads allthreads() tl[] = threadid() + @assert getallvalues(tl) == allthreads() + tl2 = ThreadLocal{Vector{Int}}() + @onthreads allthreads() push!(tl2[], threadid()) + @assert sort!(reduce(vcat, getallvalues(tl2))) == allthreads() + println("OK") + """ + cmd = `$(Base.julia_cmd()) --startup-file=no --project=$(Base.active_project()) -t 2,1 -e $prog` + @test strip(read(cmd, String)) == "OK" + end + + @testset "current-thread fast path" begin + tl = ThreadLocal(0) + @onthreads threadid() tl[] = threadid() + 100 + @test tl[] == threadid() + 100 + end + + @testset "non-contiguous threadsel" begin + @test (begin + tl = ThreadLocal(0) + @onthreads reverse(collect(allthreads())) tl[] = threadid() + getallvalues(tl) + end) == allthreads() + + if nthreads() >= 3 + tl = ThreadLocal(0) + threadsel = allthreads()[[1, end]] + @onthreads threadsel tl[] = threadid() + @test getallvalues(tl)[[1, end]] == threadsel + @test all(iszero, getallvalues(tl)[2:end-1]) + end end @testset "macro mt_out_of_order" begin + @test_throws ErrorException @macroexpand @mt_out_of_order 42 + @test begin b = 0 foo() = b = 9 @@ -65,14 +108,14 @@ using Base.Threads if nthreads() >= 4 @testset "Example 2" begin - # Assuming 4 threads: + # Assuming 4 threads in the default threadpool: tl = ThreadLocal(42) - threadsel = 2:3 + threadsel = allthreads()[2:3] @onthreads threadsel begin tl[] = Base.Threads.threadid() end - @test getallvalues(tl)[threadsel] == [2, 3] - @test getallvalues(tl)[[1,4]] == [42, 42] + @test getallvalues(tl)[2:3] == threadsel + @test getallvalues(tl)[[1,4]] == fill(42, 2) end end end diff --git a/test/test_onworkers.jl b/test/test_onworkers.jl index 260c3c4..44fe9c8 100644 --- a/test/test_onworkers.jl +++ b/test/test_onworkers.jl @@ -54,6 +54,19 @@ end end end + if !Sys.iswindows() + @testset "write_worker_start_script $(nameof(typeof(runmode)))" begin + mktempdir(prefix = "ppt-startscript-test") do dir + startscript = write_worker_start_script(joinpath(dir, "startjlworkers.sh"), runmode) + @test isfile(startscript) + script_content = read(startscript, String) + @test occursin("julia", script_content) + @test occursin("xargs", script_content) + @test_throws ArgumentError write_worker_start_script(joinpath(dir, "startjlworkers.txt"), runmode) + end + end + end + #= # Run manually for now, fails when run during CI tests for some reason: @@ -85,7 +98,28 @@ end @test_throws ParallelProcessingTools.MaxTriesExceeded onworker(gen_mayfail(1), "bar"; tries = 2, label = "mayfail") @test_throws ParallelProcessingTools.MaxTriesExceeded onworker(mytask, 2, "foo", maxtime = 0.5, tries = 2) - + + @test original_exception( + @return_exceptions onworker(() -> throw(MyExceptionNoRetry("no retry")), label = "noretry") + ) isa MyExceptionNoRetry + + @testset "worker loss" begin + # Worker losses don't count against tries, but are capped at 3 * tries: + pids = classic_addprocs(4) + die_pool = FlexWorkerPool{WorkerPool}(pids, init_workers = false) + @test_throws ParallelProcessingTools.MaxTriesExceeded onworker( + () -> exit(), pool = die_pool, label = "worker_killer" + ) + @test !any(in(procs()), pids) + + # Recovers if a worker is lost and another can take over: + pids2 = classic_addprocs(2) + mixed_pool = FlexWorkerPool{WorkerPool}([myid(), pids2[1]], init_workers = false) + @test onworker(() -> (myid() == 1 ? 42 : exit()), pool = mixed_pool, label = "lossy") == 42 + @test !(pids2[1] in procs()) + stopworkers() + end + runworkers(OnLocalhost(n = 2)) timer = Timer(30) @@ -106,6 +140,15 @@ end @test_throws ParallelProcessingTools.MaxTriesExceeded onworker(gen_mayfail(1), "bar"; tries = 2, label = "mayfail") + @test original_exception( + @return_exceptions onworker(() -> throw(MyExceptionNoRetry("no retry")), label = "noretry") + ) isa MyExceptionNoRetry + + @testset "elastic manager pool callback" begin + callback = ParallelProcessingTools._get_elasticmgr_add_to_pool_callback() + manager = ParallelProcessingTools.ppt_cluster_manager() + @test_logs (:error, r"Unknown ElasticManager manage op") callback(manager, 9999, :bogus) + end #= # Run these manually for now. Not sure how to make Test enviroment ignore the diff --git a/test/test_procinit.jl b/test/test_procinit.jl index a6e45cb..85706e2 100644 --- a/test/test_procinit.jl +++ b/test/test_procinit.jl @@ -57,6 +57,8 @@ ENV["JULIA_DEBUG"] = old_julia_debug * ",ParallelProcessingTools" @test _execute_procinit_code(get_procinit_code(), global_procinit_level()) isa Nothing @test current_procinit_level() == global_procinit_level() @test Main._g_inittest3 == 103 + # Re-running at the same init level is a no-op: + @test _execute_procinit_code(get_procinit_code(), global_procinit_level()) isa Nothing @info "The following \"Failed to raise process 1 init level\" error message is expected" @test_throws ErrorException _execute_procinit_code(get_procinit_code(), global_procinit_level() + 1) @@ -79,10 +81,14 @@ ENV["JULIA_DEBUG"] = old_julia_debug * ",ParallelProcessingTools" classic_addprocs(2) ensure_procinit(workers()[end]) - @test remotecall_fetch(last(workers())) do + @test remotecall_fetch(last(workers())) do _g_inittest1 + _g_inittest2 + _g_inittest3 + _g_inittest4 + _g_somevar1 + _g_somevar2 end == 813 + # @always_everywhere must init all current workers without explicit ensure_procinit: + @always_everywhere _g_somevar3 = 203 + @test all(pid -> remotecall_fetch(() -> Main._g_somevar3, pid) == 203, workers()) + rmprocs(workers()) end diff --git a/test/test_readme_examples.jl b/test/test_readme_examples.jl deleted file mode 100644 index 3b36b78..0000000 --- a/test/test_readme_examples.jl +++ /dev/null @@ -1,40 +0,0 @@ -# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). - -using ParallelProcessingTools -using Test - -using Distributed - -include("testtools.jl") - -if length(workers()) < 2 - classic_addprocs(2) -end - -@testset "workpartition" begin - @testset "parallel histogramming" begin - using Distributed, ParallelProcessingTools - classic_addprocs(2) - @everywhere using ParallelProcessingTools, Base.Threads, - DistributedArrays, Statistics, StatsBase - - data = drandn(10^8) - procsel = procs(data) - @onprocs procsel size(localpart(data)) - - @onprocs procsel nthreads() - - proc_hists = @onprocs procsel begin - local_data = localpart(data) - tl_hist = ThreadLocal(Histogram((-6:0.1:6,), :left)) - @onthreads allthreads() begin - data_for_this_thread = workpart(local_data, allthreads(), threadid()) - append!(tl_hist[], data_for_this_thread) - end - merged_hist = merge(getallvalues(tl_hist)...) - end - final_hist = merge(proc_hists...) - - @test sum(final_hist.weights) ≈ length(data) - end -end diff --git a/test/test_slurm.jl b/test/test_slurm.jl new file mode 100644 index 0000000..d849054 --- /dev/null +++ b/test/test_slurm.jl @@ -0,0 +1,103 @@ +# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). + +using Test +using ParallelProcessingTools + +using ParallelProcessingTools: _slurm_parse_memoptval, _slurm_parse_intoptval, + _get_slurm_taskconf, _slurm_nworkers, _slurm_mem_per_task + +@testset "slurm" begin + @testset "SLURM option values" begin + @test _slurm_parse_memoptval(nothing) === nothing + @test _slurm_parse_memoptval("100") == 100 * Int64(1024)^2 + @test _slurm_parse_memoptval("16K") == 16 * Int64(1024) + @test _slurm_parse_memoptval("2M") == 2 * Int64(1024)^2 + @test _slurm_parse_memoptval("32G") == 32 * Int64(1024)^3 + @test _slurm_parse_memoptval("32GB") == 32 * Int64(1024)^3 + @test _slurm_parse_memoptval("1T") == Int64(1024)^4 + @test_throws ArgumentError _slurm_parse_memoptval("32Q") + @test_throws ArgumentError _slurm_parse_memoptval("foo") + + @test _slurm_parse_intoptval(nothing) === nothing + @test _slurm_parse_intoptval("42") == 42 + end + + @testset "SLURM task configuration" begin + no_env = Dict{String,String}() + + tc = _get_slurm_taskconf(``, no_env) + @test tc == ( + n_tasks = nothing, cpus_per_task = nothing, mem_per_cpu = nothing, + n_nodes = nothing, ntasks_per_node = nothing, mem_per_node = nothing + ) + + tc = _get_slurm_taskconf(`--ntasks=4 --cpus-per-task=8 --mem-per-cpu=2G`, no_env) + @test tc == ( + n_tasks = 4, cpus_per_task = 8, mem_per_cpu = 2 * Int64(1024)^3, + n_nodes = nothing, ntasks_per_node = nothing, mem_per_node = nothing + ) + + tc = _get_slurm_taskconf(`-n 4 -c8 -N 2 --mem=16G --ntasks-per-node=2`, no_env) + @test tc == ( + n_tasks = 4, cpus_per_task = 8, mem_per_cpu = nothing, + n_nodes = 2, ntasks_per_node = 2, mem_per_node = 16 * Int64(1024)^3 + ) + + slurm_env = Dict( + "SLURM_NTASKS" => "6", "SLURM_CPUS_PER_TASK" => "2", + "SLURM_MEM_PER_CPU" => "1G", "SLURM_JOB_NUM_NODES" => "3" + ) + tc = _get_slurm_taskconf(``, slurm_env) + @test tc == ( + n_tasks = 6, cpus_per_task = 2, mem_per_cpu = Int64(1024)^3, + n_nodes = 3, ntasks_per_node = nothing, mem_per_node = nothing + ) + + tc = _get_slurm_taskconf(`--ntasks=12`, slurm_env) + @test tc.n_tasks == 12 + @test tc.cpus_per_task == 2 + + # Unknown options are skipped, separate-value long options work: + tc = _get_slurm_taskconf(`--partition=main --ntasks 4`, no_env) + @test tc.n_tasks == 4 + + @test_throws ArgumentError _get_slurm_taskconf(`-n`, no_env) + @test_throws ArgumentError _get_slurm_taskconf(`-n -c4`, no_env) + @test_throws ArgumentError _get_slurm_taskconf(`--ntasks`, no_env) + @test_throws ArgumentError _get_slurm_taskconf(`--ntasks=`, no_env) + @test_throws ArgumentError _get_slurm_taskconf(`--ntasks --nodes=2`, no_env) + end + + @testset "workers and memory per task" begin + template = ( + n_tasks = nothing, cpus_per_task = nothing, mem_per_cpu = nothing, + n_nodes = nothing, ntasks_per_node = nothing, mem_per_node = nothing + ) + + @test _slurm_nworkers(merge(template, (n_tasks = 4,))) == 4 + @test _slurm_nworkers(merge(template, (n_nodes = 2, ntasks_per_node = 3))) == 6 + @test_throws ArgumentError _slurm_nworkers(template) + + GiB = Int64(1024)^3 + @test _slurm_mem_per_task(merge(template, (cpus_per_task = 4, mem_per_cpu = GiB))) == 4 * GiB + @test _slurm_mem_per_task(merge(template, (n_nodes = 2, ntasks_per_node = 4, mem_per_node = 8 * GiB))) == 2 * GiB + @test _slurm_mem_per_task(merge(template, (n_nodes = 2, n_tasks = 4, mem_per_node = 8 * GiB))) == 4 * GiB + @test _slurm_mem_per_task(template) === nothing + end + + @testset "worker_start_command" begin + manager = ParallelProcessingTools.ppt_cluster_manager() + runmode = OnSlurm( + slurm_flags = `--ntasks=2 --cpus-per-task=4 --mem-per-cpu=1G`, + julia_flags = `--depwarn=yes` + ) + cmd, m, n = worker_start_command(runmode, manager) + @test m == 1 + @test n == 2 + cmd_string = string(cmd) + @test first(cmd.exec) == "srun" + @test count("--depwarn=yes", cmd_string) == 1 + @test occursin("--threads=4", cmd_string) + @test occursin("--heap-size-hint=2048M", cmd_string) + end +end diff --git a/test/test_states.jl b/test/test_states.jl index 4da7e4d..ef5635f 100644 --- a/test/test_states.jl +++ b/test/test_states.jl @@ -122,4 +122,14 @@ using Distributed: myid, remotecall @test_throws ArgumentError whyfailed(empty_open_channel) @test whyfailed(bad_closed_channel) isa ErrorException end + + @testset "non-exception task failure" begin + stuck_task = Task(() -> wait()) + schedule(stuck_task) + yield() + schedule(stuck_task, :stopped, error = true) + @wait_while maxtime=10 timeout_error=true !istaskdone(stuck_task) + @test hasfailed(stuck_task) + @test whyfailed(stuck_task) isa ErrorException + end end diff --git a/test/test_threadlocal.jl b/test/test_threadlocal.jl index 9f6bdf2..a8d615d 100644 --- a/test/test_threadlocal.jl +++ b/test/test_threadlocal.jl @@ -1,4 +1,4 @@ -# This file is a part of BAT.jl, licensed under the MIT License (MIT). +# This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). using ParallelProcessingTools using Test @@ -8,9 +8,9 @@ using Base.Threads @testset "ThreadLocal" begin tl = @inferred ThreadLocal{Float32}(undef) @test typeof(tl) <: ThreadLocal{Float32} - @test length(tl.value) == nthreads() + @test length(getallvalues(tl)) == nthreads() - @test (@inferred getallvalues(ThreadLocal{Int}(threadid))) == 1:nthreads() + @test (@inferred getallvalues(ThreadLocal{Int}(threadid))) == allthreads() tmp = 2.5 tl = @inferred ThreadLocal(tmp) diff --git a/test/test_threadsafe.jl b/test/test_threadsafe.jl index 67b0287..fdf8c67 100644 --- a/test/test_threadsafe.jl +++ b/test/test_threadsafe.jl @@ -35,7 +35,7 @@ using Base.Threads map(seekstart, lv) @test read(lv, Int) == 10 map(seekstart, lv) - @test read!(lv, Int) == 10 + @test read!(lv, zeros(Int, 1)) == [10] map(seekstart, lv) write(lv, 11) diff --git a/test/test_workpartition.jl b/test/test_workpartition.jl index 0f60565..5985d4f 100644 --- a/test/test_workpartition.jl +++ b/test/test_workpartition.jl @@ -66,6 +66,15 @@ using Test res = vcat(res, @inferred workpart(cmp_res, 1:part, i)) end @test res == cmp_res + + A = collect(1:10) + @test workpart(A, [2, 5, 7], 5) == workpart(A, 1:3, 2) + @test_throws ArgumentError workpart(A, [5, 2, 7], 5) + @test_throws ArgumentError workpart(A, [2, 2, 7], 2) + @test_throws ArgumentError workpart(A, [2, 5, 7], 3) + @test workpart(A, 4, 4) === A + @test_throws ArgumentError workpart(A, 4, 5) + @test isempty(ParallelProcessingTools._workpart_scheme(Base.OneTo(10), 3, 0)) end @testset "Examples" begin @@ -75,9 +84,11 @@ using Test # ... sub_A = workpart(A, procs(), myid()) # ... - idxs = workpart(eachindex(sub_A), allthreads(), threadid()) - for i in idxs - # ... + @onthreads allthreads() begin + idxs = workpart(eachindex(sub_A), allthreads(), threadid()) + for i in idxs + # ... + end end true end diff --git a/test/testtools.jl b/test/testtools.jl index c60f089..88407ca 100644 --- a/test/testtools.jl +++ b/test/testtools.jl @@ -1,17 +1,8 @@ # This file is a part of ParallelProcessingTools.jl, licensed under the MIT License (MIT). -import Pkg - if !isdefined(@__MODULE__, :classic_addprocs) -function classic_addprocs(n::Integer) - if VERSION >= v"1.10" - addprocs(2) - else - # addprocs doesn't set project automatically on some older Julia versions - addprocs(2; exeflags=`--project=$(Pkg.project().path)`) - end -end +classic_addprocs(n::Integer) = addprocs(n) function test_runprocs(f_runprocs, additional_n)