restart worker during tests depending on resident memory size

base/multi.jl

@@ -336,30 +336,35 @@ function rmprocs(args...; waitfor = 0.0)
         error("only process 1 can add and remove processes")
     end
 
-    rmprocset = []
-    for i in vcat(args...)
-        if i == 1
-            warn("rmprocs: process 1 not removed")
-        else
-            if haskey(map_pid_wrkr, i)
-                w = map_pid_wrkr[i]
-                set_worker_state(w, W_TERMINATING)
-                kill(w.manager, i, w.config)
-                push!(rmprocset, w)
+    lock(worker_lock)
+    try
+        rmprocset = []
+        for i in vcat(args...)
+            if i == 1
+                warn("rmprocs: process 1 not removed")
+            else
+                if haskey(map_pid_wrkr, i)
+                    w = map_pid_wrkr[i]
+                    set_worker_state(w, W_TERMINATING)
+                    kill(w.manager, i, w.config)
+                    push!(rmprocset, w)
+                end
             end
         end
-    end
 
-    start = time()
-    while (time() - start) < waitfor
-        if all(w -> w.state == W_TERMINATED, rmprocset)
-            break;
-        else
-            sleep(0.1)
+        start = time()
+        while (time() - start) < waitfor
+            if all(w -> w.state == W_TERMINATED, rmprocset)
+                break;
+            else
+                sleep(0.1)
+            end
         end
-    end
 
-    ((waitfor > 0) && any(w -> w.state != W_TERMINATED, rmprocset)) ? :timed_out : :ok
+        ((waitfor > 0) && any(w -> w.state != W_TERMINATED, rmprocset)) ? :timed_out : :ok
+    finally
+        unlock(worker_lock)
+    end
 end
 
 
@@ -1075,7 +1080,20 @@ end
 # `manager` is of type ClusterManager. The respective managers are responsible
 # for launching the workers. All keyword arguments (plus a few default values)
 # are available as a dictionary to the `launch` methods
+#
+# Only one addprocs can be in progress at any time
+#
+const worker_lock = ReentrantLock()
 function addprocs(manager::ClusterManager; kwargs...)
+    lock(worker_lock)
+    try
+        addprocs_locked(manager::ClusterManager; kwargs...)
+    finally
+        unlock(worker_lock)
+    end
+end
+
+function addprocs_locked(manager::ClusterManager; kwargs...)
 
     params = merge(default_addprocs_params(), AnyDict(kwargs))
     topology(symbol(params[:topology]))

base/sysinfo.jl

@@ -146,4 +146,6 @@ function set_process_title(title::AbstractString)
     uv_error("set_process_title", err)
 end
 
+maxrss() = ccall(:jl_maxrss, Csize_t, ())
+
 end # module Sys

src/sys.c

@@ -10,7 +10,9 @@
 #include <stdlib.h>
 #include <string.h>
 #include <assert.h>
-#ifndef _OS_WINDOWS_
+#ifdef _OS_WINDOWS_
+#include <psapi.h>
+#else
 #include <sys/sysctl.h>
 #include <sys/wait.h>
 #include <sys/ptrace.h>
@@ -742,6 +744,30 @@ DLLEXPORT jl_sym_t* jl_get_ARCH()
     return ARCH;
 }
 
+DLLEXPORT size_t jl_maxrss()
+{
+#if defined(_OS_WINDOWS_)
+	PROCESS_MEMORY_COUNTERS counter;
+	GetProcessMemoryInfo( GetCurrentProcess( ), &counter, sizeof(counter) );
+	return (size_t)counter.PeakWorkingSetSize;
+
+#elif defined(_OS_LINUX_) || defined(_OS_DARWIN_) || defined (_OS_FREEBSD_)
+	struct rusage rusage;
+	getrusage( RUSAGE_SELF, &rusage );
+
+#if defined(_OS_LINUX_)
+	return (size_t)(rusage.ru_maxrss * 1024);
+#else
+	return (size_t)rusage.ru_maxrss;
+#endif
+
+#else
+	return (size_t)0;
+#endif
+}
+
+
+
 #ifdef __cplusplus
 }
 #endif

test/choosetests.jl

@@ -32,7 +32,7 @@ function choosetests(choices = [])
         "nullable", "meta", "profile", "libgit2", "docs", "markdown",
         "base64", "serialize", "functors", "misc",
         "enums", "cmdlineargs", "i18n", "workspace", "libdl", "int",
-        "intset", "floatfuncs", "compile"
+        "intset", "floatfuncs", "compile", "parallel"
     ]
 
     if Base.USE_GPL_LIBS
@@ -43,9 +43,6 @@ function choosetests(choices = [])
         push!(testnames, "examples")
     end
 
-    # parallel tests depend on other workers - do them last
-    push!(testnames, "parallel")
-
     tests = []
     skip_tests = []
 

test/examples.jl

@@ -52,14 +52,13 @@ end
 dc_path = joinpath(dir, "dictchannel.jl")
 include(dc_path)
 
-w_set=filter!(x->x != myid(), workers())
-pid = length(w_set) > 0 ? w_set[1] : myid()
-
-remotecall_fetch(pid, dc_path) do f
+# Run the remote on pid 1, since runtests may terminate workers
+# at any time depending on memory usage
+remotecall_fetch(1, dc_path) do f
     include(f)
     nothing
 end
-dc=RemoteRef(()->DictChannel(), pid)
+dc=RemoteRef(()->DictChannel(), 1)
 @test typeof(dc) == RemoteRef{DictChannel}
 
 @test isready(dc) == false