Parallel Computing

For newcomers to multi-threading and parallel computing it can be useful to first appreciate the different levels of parallelism offered by Julia. We can divide them in three main categories :

  1. Julia Coroutines (Green Threading)
  2. Multi-Threading
  3. Multi-Core or Distributed Processing

We will first consider Julia Tasks (aka Coroutines) and other modules that rely on the Julia runtime library, that allow us to suspend and resume computations with full control of inter-Tasks communication without having to manually interface with the operating system’s scheduler. Julia also supports communication between Tasks through operations like wait and fetch. Communication and data synchronization is managed through Channels, which are the conduits that provide inter-Tasks communication.

Julia also supports experimental multi-threading, where execution is forked and an anonymous function is run across all threads. Known as the fork-join approach, parallel threads execute independently, and must ultimately be joined in Julia’s main thread to allow serial execution to continue. Multi-threading is supported using the Base.Threads module that is still considered experimental, as Julia is not yet fully thread-safe. In particular segfaults seem to occur during I/O operations and task switching. As an up-to-date reference, keep an eye on the issue tracker. Multi-Threading should only be used if you take into consideration global variables, locks and atomics, all of which are explained later.

In the end we will present Julia’s approach to distributed and parallel computing. With scientific computing in mind, Julia natively implements interfaces to distribute a process across multiple cores or machines. Also we will mention useful external packages for distributed programming like MPI.jl and DistributedArrays.jl.

Coroutines

Julia’s parallel programming platform uses Tasks (aka Coroutines) to switch among multiple computations. To express an order of execution between lightweight threads communication primitives are necessary. Julia offers Channel(func::Function, ctype=Any, csize=0, taskref=nothing) that creates a new task from func, binds it to a new channel of type ctype and size csize and schedule the task. Channels can serve as a way to communicate between tasks, as Channel{T}(sz::Int) creates a buffered channel of type T and size sz. Whenever code performs a communication operation like fetch or wait, the current task is suspended and a scheduler picks another task to run. A task is restarted when the event it is waiting for completes.

For many problems, it is not necessary to think about tasks directly. However, they can be used to wait for multiple events at the same time, which provides for dynamic scheduling. In dynamic scheduling, a program decides what to compute or where to compute it based on when other jobs finish. This is needed for unpredictable or unbalanced workloads, where we want to assign more work to processes only when they finish their current tasks.

Channels

The section on Tasks in Control Flow discussed the execution of multiple functions in a co-operative manner. Channels can be quite useful to pass data between running tasks, particularly those involving I/O operations.

Examples of operations involving I/O include reading/writing to files, accessing web services, executing external programs, etc. In all these cases, overall execution time can be improved if other tasks can be run while a file is being read, or while waiting for an external service/program to complete.

A channel can be visualized as a pipe, i.e., it has a write end and a read end :

  • Multiple writers in different tasks can write to the same channel concurrently via put! calls.
  • Multiple readers in different tasks can read data concurrently via take! calls.
  • As an example:

``` 1. # Given Channels c1 and c2, 2. c1 = Channel(32) 3. c2 = Channel(32)

  1. and a function foo which reads items from c1, processes the item read

  2. and writes a result to c2,

  3. function foo()
  4. while true
  5. data = take!(c1)
  6. [...] # process data
  7. put!(c2, result) # write out result
  8. end

  9. end

  10. we can schedule n instances of foo to be active concurrently.

  11. for _ in 1:n
  12. @async foo()
  13. end ```
  14. Channels are created via the Channel{T}(sz) constructor. The channel will only hold objects of type T. If the type is not specified, the channel can hold objects of any type. sz refers to the maximum number of elements that can be held in the channel at any time. For example, Channel(32) creates a channel that can hold a maximum of 32 objects of any type. A Channel{MyType}(64) can hold up to 64 objects of MyType at any time.
  15. If a Channel is empty, readers (on a take! call) will block until data is available.
  16. If a Channel is full, writers (on a put! call) will block until space becomes available.
  17. isready tests for the presence of any object in the channel, while wait waits for an object to become available.
  18. A Channel is in an open state initially. This means that it can be read from and written to freely via take! and put! calls. close closes a Channel. On a closed Channel, put! will fail. For example:

``` 1. julia> c = Channel(2);

  1. julia> put!(c, 1) # put! on an open channel succeeds

  2. 1

  3. julia> close(c);

  4. julia> put!(c, 2) # put! on a closed channel throws an exception.

  5. ERROR: InvalidStateException("Channel is closed.",:closed)
  6. Stacktrace:
  7. [...] ```
  8. take! and fetch (which retrieves but does not remove the value) on a closed channel successfully return any existing values until it is emptied. Continuing the above example:

`` 1. julia> fetch(c) # Any number offetch` calls succeed. 2. 1

  1. julia> fetch(c)

  2. 1

  3. julia> take!(c) # The first take! removes the value.

  4. 1

  5. julia> take!(c) # No more data available on a closed channel.

  6. ERROR: InvalidStateException("Channel is closed.",:closed)
  7. Stacktrace:
  8. [...] ```

A Channel can be used as an iterable object in a for loop, in which case the loop runs as long as the Channel has data or is open. The loop variable takes on all values added to the Channel. The for loop is terminated once the Channel is closed and emptied.

For example, the following would cause the for loop to wait for more data:

1. julia> c = Channel{Int}(10);

3. julia> foreach(i->put!(c, i), 1:3) # add a few entries

5. julia> data = [i for i in c]

while this will return after reading all data:

1. julia> c = Channel{Int}(10);

3. julia> foreach(i->put!(c, i), 1:3); # add a few entries

5. julia> close(c);                    # `for` loops can exit

7. julia> data = [i for i in c]
8. 3-element Array{Int64,1}:
9. 1
10. 2
11. 3

Consider a simple example using channels for inter-task communication. We start 4 tasks to process data from a single jobs channel. Jobs, identified by an id (job_id), are written to the channel. Each task in this simulation reads a job_id, waits for a random amount of time and writes back a tuple of job_id and the simulated time to the results channel. Finally all the results are printed out.

1. julia> const jobs = Channel{Int}(32);

3. julia> const results = Channel{Tuple}(32);

5. julia> function do_work()
6. for job_id in jobs
7. exec_time = rand()
8. sleep(exec_time)                # simulates elapsed time doing actual work
9. # typically performed externally.
10. put!(results, (job_id, exec_time))
11. end
12. end;

14. julia> function make_jobs(n)
15. for i in 1:n
16. put!(jobs, i)
17. end
18. end;

20. julia> n = 12;

22. julia> @async make_jobs(n); # feed the jobs channel with "n" jobs

24. julia> for i in 1:4 # start 4 tasks to process requests in parallel
25. @async do_work()
26. end

28. julia> @elapsed while n > 0 # print out results
29. job_id, exec_time = take!(results)
30. println("$job_id finished in $(round(exec_time; digits=2)) seconds")
31. global n = n - 1
32. end
33. 4 finished in 0.22 seconds
34. 3 finished in 0.45 seconds
35. 1 finished in 0.5 seconds
36. 7 finished in 0.14 seconds
37. 2 finished in 0.78 seconds
38. 5 finished in 0.9 seconds
39. 9 finished in 0.36 seconds
40. 6 finished in 0.87 seconds
41. 8 finished in 0.79 seconds
42. 10 finished in 0.64 seconds
43. 12 finished in 0.5 seconds
44. 11 finished in 0.97 seconds
45. 0.029772311

The current version of Julia multiplexes all tasks onto a single OS thread. Thus, while tasks involving I/O operations benefit from parallel execution, compute bound tasks are effectively executed sequentially on a single OS thread. Future versions of Julia may support scheduling of tasks on multiple threads, in which case compute bound tasks will see benefits of parallel execution too.

Multi-Threading (Experimental)

In addition to tasks Julia natively supports multi-threading. Note that this section is experimental and the interfaces may change in the future.

Setup

By default, Julia starts up with a single thread of execution. This can be verified by using the command Threads.nthreads():

1. julia> Threads.nthreads()
2. 1

The number of threads Julia starts up with is controlled by an environment variable called JULIA_NUM_THREADS. Now, let’s start up Julia with 4 threads:

Bash on Linux/OSX:

1. export JULIA_NUM_THREADS=4

C shell on Linux/OSX, CMD on Windows:

1. set JULIA_NUM_THREADS=4

Powershell on Windows:

1. $env:JULIA_NUM_THREADS=4

Let’s verify there are 4 threads at our disposal.

1. julia> Threads.nthreads()
2. 4

But we are currently on the master thread. To check, we use the function Threads.threadid

1. julia> Threads.threadid()
2. 1

The @threads Macro

Let’s work a simple example using our native threads. Let us create an array of zeros:

1. julia> a = zeros(10)
2. 10-element Array{Float64,1}:
3. 0.0
4. 0.0
5. 0.0
6. 0.0
7. 0.0
8. 0.0
9. 0.0
10. 0.0
11. 0.0
12. 0.0

Let us operate on this array simultaneously using 4 threads. We’ll have each thread write its thread ID into each location.

Julia supports parallel loops using the Threads.@threads macro. This macro is affixed in front of a for loop to indicate to Julia that the loop is a multi-threaded region:

1. julia> Threads.@threads for i = 1:10
2. a[i] = Threads.threadid()
3. end

The iteration space is split amongst the threads, after which each thread writes its thread ID to its assigned locations:

1. julia> a
2. 10-element Array{Float64,1}:
3. 1.0
4. 1.0
5. 1.0
6. 2.0
7. 2.0
8. 2.0
9. 3.0
10. 3.0
11. 4.0
12. 4.0

Note that Threads.@threads does not have an optional reduction parameter like @distributed.

Atomic Operations

Julia supports accessing and modifying values atomically, that is, in a thread-safe way to avoid race conditions. A value (which must be of a primitive type) can be wrapped as Threads.Atomic to indicate it must be accessed in this way. Here we can see an example:

1. julia> i = Threads.Atomic{Int}(0);

3. julia> ids = zeros(4);

5. julia> old_is = zeros(4);

7. julia> Threads.@threads for id in 1:4
8. old_is[id] = Threads.atomic_add!(i, id)
9. ids[id] = id
10. end

12. julia> old_is
13. 4-element Array{Float64,1}:
14. 0.0
15. 1.0
16. 7.0
17. 3.0

19. julia> ids
20. 4-element Array{Float64,1}:
21. 1.0
22. 2.0
23. 3.0
24. 4.0

Had we tried to do the addition without the atomic tag, we might have gotten the wrong answer due to a race condition. An example of what would happen if we didn’t avoid the race:

1. julia> using Base.Threads

3. julia> nthreads()
4. 4

6. julia> acc = Ref(0)
7. Base.RefValue{Int64}(0)

9. julia> @threads for i in 1:1000
10. acc[] += 1
11. end

13. julia> acc[]
14. 926

16. julia> acc = Atomic{Int64}(0)
17. Atomic{Int64}(0)

19. julia> @threads for i in 1:1000
20. atomic_add!(acc, 1)
21. end

23. julia> acc[]
24. 1000

Not all primitive types can be wrapped in an Atomic tag. Supported types are Int8, Int16, Int32, Int64, Int128, UInt8, UInt16, UInt32, UInt64, UInt128, Float16, Float32, and Float64. Additionally, Int128 and UInt128 are not supported on AAarch32 and ppc64le.

Side effects and mutable function arguments

When using multi-threading we have to be careful when using functions that are not pure as we might get a wrong answer. For instance functions that have their name ending with ! by convention modify their arguments and thus are not pure. However, there are functions that have side effects and their name does not end with !. For instance findfirst(regex, str) mutates its regex argument or rand() changes Base.GLOBAL_RNG :

1. julia> using Base.Threads

3. julia> nthreads()
4. 4

6. julia> function f()
7. s = repeat(["123", "213", "231"], outer=1000)
8. x = similar(s, Int)
9. rx = r"1"
10. @threads for i in 1:3000
11. x[i] = findfirst(rx, s[i]).start
12. end
13. count(v -> v == 1, x)
14. end
15. f (generic function with 1 method)

17. julia> f() # the correct result is 1000
18. 1017

20. julia> function g()
21. a = zeros(1000)
22. @threads for i in 1:1000
23. a[i] = rand()
24. end
25. length(unique(a))
26. end
27. g (generic function with 1 method)

29. julia> Random.seed!(1); g() # the result for a single thread is 1000
30. 781

In such cases one should redesign the code to avoid the possibility of a race condition or use synchronization primitives.

For example in order to fix findfirst example above one needs to have a separate copy of rx variable for each thread:

1. julia> function f_fix()
2. s = repeat(["123", "213", "231"], outer=1000)
3. x = similar(s, Int)
4. rx = [Regex("1") for i in 1:nthreads()]
5. @threads for i in 1:3000
6. x[i] = findfirst(rx[threadid()], s[i]).start
7. end
8. count(v -> v == 1, x)
9. end
10. f_fix (generic function with 1 method)

12. julia> f_fix()
13. 1000

We now use Regex("1") instead of r"1" to make sure that Julia creates separate instances of Regex object for each entry of rx vector.

The case of rand is a bit more complex as we have to ensure that each thread uses non-overlapping pseudorandom number sequences. This can be simply ensured by using Future.randjump function:

1. julia> using Random; import Future

3. julia> function g_fix(r)
4. a = zeros(1000)
5. @threads for i in 1:1000
6. a[i] = rand(r[threadid()])
7. end
8. length(unique(a))
9. end
10. g_fix (generic function with 1 method)

12. julia>  r = let m = MersenneTwister(1)
13. [m; accumulate(Future.randjump, fill(big(10)^20, nthreads()-1), init=m)]
14. end;

16. julia> g_fix(r)
17. 1000

We pass the r vector to g_fix as generating several RGNs is an expensive operation so we do not want to repeat it every time we run the function.

@threadcall (Experimental)

All I/O tasks, timers, REPL commands, etc are multiplexed onto a single OS thread via an event loop. A patched version of libuv (<http://docs.libuv.org/en/v1.x/>) provides this functionality. Yield points provide for co-operatively scheduling multiple tasks onto the same OS thread. I/O tasks and timers yield implicitly while waiting for the event to occur. Calling yield explicitly allows for other tasks to be scheduled.

Thus, a task executing a ccall effectively prevents the Julia scheduler from executing any other tasks till the call returns. This is true for all calls into external libraries. Exceptions are calls into custom C code that call back into Julia (which may then yield) or C code that calls jl_yield() (C equivalent of yield).

Note that while Julia code runs on a single thread (by default), libraries used by Julia may launch their own internal threads. For example, the BLAS library may start as many threads as there are cores on a machine.

The @threadcall macro addresses scenarios where we do not want a ccall to block the main Julia event loop. It schedules a C function for execution in a separate thread. A threadpool with a default size of 4 is used for this. The size of the threadpool is controlled via environment variable UV_THREADPOOL_SIZE. While waiting for a free thread, and during function execution once a thread is available, the requesting task (on the main Julia event loop) yields to other tasks. Note that @threadcall does not return till the execution is complete. From a user point of view, it is therefore a blocking call like other Julia APIs.

It is very important that the called function does not call back into Julia, as it will segfault.

@threadcall may be removed/changed in future versions of Julia.

Multi-Core or Distributed Processing

An implementation of distributed memory parallel computing is provided by module Distributed as part of the standard library shipped with Julia.

Most modern computers possess more than one CPU, and several computers can be combined together in a cluster. Harnessing the power of these multiple CPUs allows many computations to be completed more quickly. There are two major factors that influence performance: the speed of the CPUs themselves, and the speed of their access to memory. In a cluster, it’s fairly obvious that a given CPU will have fastest access to the RAM within the same computer (node). Perhaps more surprisingly, similar issues are relevant on a typical multicore laptop, due to differences in the speed of main memory and the cache. Consequently, a good multiprocessing environment should allow control over the “ownership” of a chunk of memory by a particular CPU. Julia provides a multiprocessing environment based on message passing to allow programs to run on multiple processes in separate memory domains at once.

Julia’s implementation of message passing is different from other environments such as MPI [1]. Communication in Julia is generally “one-sided”, meaning that the programmer needs to explicitly manage only one process in a two-process operation. Furthermore, these operations typically do not look like “message send” and “message receive” but rather resemble higher-level operations like calls to user functions.

Distributed programming in Julia is built on two primitives: remote references and remote calls. A remote reference is an object that can be used from any process to refer to an object stored on a particular process. A remote call is a request by one process to call a certain function on certain arguments on another (possibly the same) process.

Remote references come in two flavors: Future and RemoteChannel.

A remote call returns a Future to its result. Remote calls return immediately; the process that made the call proceeds to its next operation while the remote call happens somewhere else. You can wait for a remote call to finish by calling wait on the returned Future, and you can obtain the full value of the result using fetch.

On the other hand, RemoteChannel s are rewritable. For example, multiple processes can co-ordinate their processing by referencing the same remote Channel.

Each process has an associated identifier. The process providing the interactive Julia prompt always has an id equal to 1. The processes used by default for parallel operations are referred to as “workers”. When there is only one process, process 1 is considered a worker. Otherwise, workers are considered to be all processes other than process 1.

Let’s try this out. Starting with julia -p n provides n worker processes on the local machine. Generally it makes sense for n to equal the number of CPU threads (logical cores) on the machine. Note that the -p argument implicitly loads module Distributed.

1. $ ./julia -p 2

3. julia> r = remotecall(rand, 2, 2, 2)
4. Future(2, 1, 4, nothing)

6. julia> s = @spawnat 2 1 .+ fetch(r)
7. Future(2, 1, 5, nothing)

9. julia> fetch(s)
10. 2×2 Array{Float64,2}:
11. 1.18526  1.50912
12. 1.16296  1.60607

The first argument to remotecall is the function to call. Most parallel programming in Julia does not reference specific processes or the number of processes available, but remotecall is considered a low-level interface providing finer control. The second argument to remotecall is the id of the process that will do the work, and the remaining arguments will be passed to the function being called.

As you can see, in the first line we asked process 2 to construct a 2-by-2 random matrix, and in the second line we asked it to add 1 to it. The result of both calculations is available in the two futures, r and s. The @spawnat macro evaluates the expression in the second argument on the process specified by the first argument.

Occasionally you might want a remotely-computed value immediately. This typically happens when you read from a remote object to obtain data needed by the next local operation. The function remotecall_fetch exists for this purpose. It is equivalent to fetch(remotecall(...)) but is more efficient.

1. julia> remotecall_fetch(getindex, 2, r, 1, 1)
2. 0.18526337335308085

Remember that getindex(r,1,1) is equivalent to r[1,1], so this call fetches the first element of the future r.

To make things easier, the symbol :any can be passed to [@spawnat], which picks where to do the operation for you:

1. julia> r = @spawnat :any rand(2,2)
2. Future(2, 1, 4, nothing)

4. julia> s = @spawnat :any 1 .+ fetch(r)
5. Future(3, 1, 5, nothing)

7. julia> fetch(s)
8. 2×2 Array{Float64,2}:
9. 1.38854  1.9098
10. 1.20939  1.57158

Note that we used 1 .+ fetch(r) instead of 1 .+ r. This is because we do not know where the code will run, so in general a fetch might be required to move r to the process doing the addition. In this case, @spawnat is smart enough to perform the computation on the process that owns r, so the fetch will be a no-op (no work is done).

(It is worth noting that @spawnat is not built-in but defined in Julia as a macro. It is possible to define your own such constructs.)

An important thing to remember is that, once fetched, a Future will cache its value locally. Further fetch calls do not entail a network hop. Once all referencing Futures have fetched, the remote stored value is deleted.

@async is similar to @spawnat, but only runs tasks on the local process. We use it to create a “feeder” task for each process. Each task picks the next index that needs to be computed, then waits for its process to finish, then repeats until we run out of indices. Note that the feeder tasks do not begin to execute until the main task reaches the end of the @sync block, at which point it surrenders control and waits for all the local tasks to complete before returning from the function. As for v0.7 and beyond, the feeder tasks are able to share state via nextidx because they all run on the same process. Even if Tasks are scheduled cooperatively, locking may still be required in some contexts, as in asynchronous I/O. This means context switches only occur at well-defined points: in this case, when remotecall_fetch is called. This is the current state of implementation and it may change for future Julia versions, as it is intended to make it possible to run up to N Tasks on M Process, aka M:N Threading. Then a lock acquiring\releasing model for nextidx will be needed, as it is not safe to let multiple processes read-write a resource at the same time.

Code Availability and Loading Packages

Your code must be available on any process that runs it. For example, type the following into the Julia prompt:

1. julia> function rand2(dims...)
2. return 2*rand(dims...)
3. end

5. julia> rand2(2,2)
6. 2×2 Array{Float64,2}:
7. 0.153756  0.368514
8. 1.15119   0.918912

10. julia> fetch(@spawnat :any rand2(2,2))
11. ERROR: RemoteException(2, CapturedException(UndefVarError(Symbol("#rand2"))
12. Stacktrace:
13. [...]

Process 1 knew about the function rand2, but process 2 did not.

Most commonly you’ll be loading code from files or packages, and you have a considerable amount of flexibility in controlling which processes load code. Consider a file, DummyModule.jl, containing the following code:

1. module DummyModule

3. export MyType, f

5. mutable struct MyType
6. a::Int
7. end

9. f(x) = x^2+1

11. println("loaded")

13. end

In order to refer to MyType across all processes, DummyModule.jl needs to be loaded on every process. Calling include("DummyModule.jl") loads it only on a single process. To load it on every process, use the @everywhere macro (starting Julia with julia -p 2):

1. julia> @everywhere include("DummyModule.jl")
2. loaded
3. From worker 3:    loaded
4. From worker 2:    loaded

As usual, this does not bring DummyModule into scope on any of the process, which requires using or import. Moreover, when DummyModule is brought into scope on one process, it is not on any other:

1. julia> using .DummyModule

3. julia> MyType(7)
4. MyType(7)

6. julia> fetch(@spawnat 2 MyType(7))
7. ERROR: On worker 2:
8. UndefVarError: MyType not defined
9. ⋮

11. julia> fetch(@spawnat 2 DummyModule.MyType(7))
12. MyType(7)

However, it’s still possible, for instance, to send a MyType to a process which has loaded DummyModule even if it’s not in scope:

1. julia> put!(RemoteChannel(2), MyType(7))
2. RemoteChannel{Channel{Any}}(2, 1, 13)

A file can also be preloaded on multiple processes at startup with the -L flag, and a driver script can be used to drive the computation:

1. julia -p <n> -L file1.jl -L file2.jl driver.jl

The Julia process running the driver script in the example above has an id equal to 1, just like a process providing an interactive prompt.

Finally, if DummyModule.jl is not a standalone file but a package, then using DummyModule will load DummyModule.jl on all processes, but only bring it into scope on the process where using was called.

Starting and managing worker processes

The base Julia installation has in-built support for two types of clusters:

  • A local cluster specified with the -p option as shown above.
  • A cluster spanning machines using the --machine-file option. This uses a passwordless ssh login to start Julia worker processes (from the same path as the current host) on the specified machines.

Functions addprocs, rmprocs, workers, and others are available as a programmatic means of adding, removing and querying the processes in a cluster.

1. julia> using Distributed

3. julia> addprocs(2)
4. 2-element Array{Int64,1}:
5. 2
6. 3

Module Distributed must be explicitly loaded on the master process before invoking addprocs. It is automatically made available on the worker processes.

Note that workers do not run a ~/.julia/config/startup.jl startup script, nor do they synchronize their global state (such as global variables, new method definitions, and loaded modules) with any of the other running processes. You may use addprocs(exeflags="--project") to initialize a worker with a particular environment, and then @everywhere using <modulename> or @everywhere include("file.jl").

Other types of clusters can be supported by writing your own custom ClusterManager, as described below in the ClusterManagers section.

Data Movement

Sending messages and moving data constitute most of the overhead in a distributed program. Reducing the number of messages and the amount of data sent is critical to achieving performance and scalability. To this end, it is important to understand the data movement performed by Julia’s various distributed programming constructs.

fetch can be considered an explicit data movement operation, since it directly asks that an object be moved to the local machine. @spawnat (and a few related constructs) also moves data, but this is not as obvious, hence it can be called an implicit data movement operation. Consider these two approaches to constructing and squaring a random matrix:

Method 1:

1. julia> A = rand(1000,1000);

3. julia> Bref = @spawnat :any A^2;

5. [...]

7. julia> fetch(Bref);

Method 2:

1. julia> Bref = @spawnat :any rand(1000,1000)^2;

3. [...]

5. julia> fetch(Bref);

The difference seems trivial, but in fact is quite significant due to the behavior of @spawnat. In the first method, a random matrix is constructed locally, then sent to another process where it is squared. In the second method, a random matrix is both constructed and squared on another process. Therefore the second method sends much less data than the first.

In this toy example, the two methods are easy to distinguish and choose from. However, in a real program designing data movement might require more thought and likely some measurement. For example, if the first process needs matrix A then the first method might be better. Or, if computing A is expensive and only the current process has it, then moving it to another process might be unavoidable. Or, if the current process has very little to do between the @spawnat and fetch(Bref), it might be better to eliminate the parallelism altogether. Or imagine rand(1000,1000) is replaced with a more expensive operation. Then it might make sense to add another @spawnat statement just for this step.

Global variables

Expressions executed remotely via @spawnat, or closures specified for remote execution using remotecall may refer to global variables. Global bindings under module Main are treated a little differently compared to global bindings in other modules. Consider the following code snippet:

1. A = rand(10,10)
2. remotecall_fetch(()->sum(A), 2)

In this case sum MUST be defined in the remote process. Note that A is a global variable defined in the local workspace. Worker 2 does not have a variable called A under Main. The act of shipping the closure ()->sum(A) to worker 2 results in Main.A being defined on 2. Main.A continues to exist on worker 2 even after the call remotecall_fetch returns. Remote calls with embedded global references (under Main module only) manage globals as follows:

  • New global bindings are created on destination workers if they are referenced as part of a remote call.
  • Global constants are declared as constants on remote nodes too.
  • Globals are re-sent to a destination worker only in the context of a remote call, and then only if its value has changed. Also, the cluster does not synchronize global bindings across nodes. For example:

1. A = rand(10,10) 2. remotecall_fetch(()->sum(A), 2) # worker 2 3. A = rand(10,10) 4. remotecall_fetch(()->sum(A), 3) # worker 3 5. A = nothing

Executing the above snippet results in Main.A on worker 2 having a different value from Main.A on worker 3, while the value of Main.A on node 1 is set to nothing.

As you may have realized, while memory associated with globals may be collected when they are reassigned on the master, no such action is taken on the workers as the bindings continue to be valid. clear! can be used to manually reassign specific globals on remote nodes to nothing once they are no longer required. This will release any memory associated with them as part of a regular garbage collection cycle.

Thus programs should be careful referencing globals in remote calls. In fact, it is preferable to avoid them altogether if possible. If you must reference globals, consider using let blocks to localize global variables.

For example:

1. julia> A = rand(10,10);

3. julia> remotecall_fetch(()->A, 2);

5. julia> B = rand(10,10);

7. julia> let B = B
8. remotecall_fetch(()->B, 2)
9. end;

11. julia> @fetchfrom 2 InteractiveUtils.varinfo()
12. name           size summary
13. ––––––––– ––––––––– ––––––––––––––––––––––
14. A         800 bytes 10×10 Array{Float64,2}
15. Base                Module
16. Core                Module
17. Main                Module

As can be seen, global variable A is defined on worker 2, but B is captured as a local variable and hence a binding for B does not exist on worker 2.

Parallel Map and Loops

Fortunately, many useful parallel computations do not require data movement. A common example is a Monte Carlo simulation, where multiple processes can handle independent simulation trials simultaneously. We can use @spawnat to flip coins on two processes. First, write the following function in count_heads.jl:

1. function count_heads(n)
2. c::Int = 0
3. for i = 1:n
4. c += rand(Bool)
5. end
6. c
7. end

The function count_heads simply adds together n random bits. Here is how we can perform some trials on two machines, and add together the results:

1. julia> @everywhere include_string(Main, $(read("count_heads.jl", String)), "count_heads.jl")

3. julia> a = @spawnat :any count_heads(100000000)
4. Future(2, 1, 6, nothing)

6. julia> b = @spawnat :any count_heads(100000000)
7. Future(3, 1, 7, nothing)

9. julia> fetch(a)+fetch(b)
10. 100001564

This example demonstrates a powerful and often-used parallel programming pattern. Many iterations run independently over several processes, and then their results are combined using some function. The combination process is called a reduction, since it is generally tensor-rank-reducing: a vector of numbers is reduced to a single number, or a matrix is reduced to a single row or column, etc. In code, this typically looks like the pattern x = f(x,v[i]), where x is the accumulator, f is the reduction function, and the v[i] are the elements being reduced. It is desirable for f to be associative, so that it does not matter what order the operations are performed in.

Notice that our use of this pattern with count_heads can be generalized. We used two explicit @spawnat statements, which limits the parallelism to two processes. To run on any number of processes, we can use a parallel for loop, running in distributed memory, which can be written in Julia using @distributed like this:

1. nheads = @distributed (+) for i = 1:200000000
2. Int(rand(Bool))
3. end

This construct implements the pattern of assigning iterations to multiple processes, and combining them with a specified reduction (in this case (+)). The result of each iteration is taken as the value of the last expression inside the loop. The whole parallel loop expression itself evaluates to the final answer.

Note that although parallel for loops look like serial for loops, their behavior is dramatically different. In particular, the iterations do not happen in a specified order, and writes to variables or arrays will not be globally visible since iterations run on different processes. Any variables used inside the parallel loop will be copied and broadcast to each process.

For example, the following code will not work as intended:

1. a = zeros(100000)
2. @distributed for i = 1:100000
3. a[i] = i
4. end

This code will not initialize all of a, since each process will have a separate copy of it. Parallel for loops like these must be avoided. Fortunately, Shared Arrays can be used to get around this limitation:

1. using SharedArrays

3. a = SharedArray{Float64}(10)
4. @distributed for i = 1:10
5. a[i] = i
6. end

Using “outside” variables in parallel loops is perfectly reasonable if the variables are read-only:

1. a = randn(1000)
2. @distributed (+) for i = 1:100000
3. f(a[rand(1:end)])
4. end

Here each iteration applies f to a randomly-chosen sample from a vector a shared by all processes.

As you could see, the reduction operator can be omitted if it is not needed. In that case, the loop executes asynchronously, i.e. it spawns independent tasks on all available workers and returns an array of Future immediately without waiting for completion. The caller can wait for the Future completions at a later point by calling fetch on them, or wait for completion at the end of the loop by prefixing it with @sync, like @sync @distributed for.

In some cases no reduction operator is needed, and we merely wish to apply a function to all integers in some range (or, more generally, to all elements in some collection). This is another useful operation called parallel map, implemented in Julia as the pmap function. For example, we could compute the singular values of several large random matrices in parallel as follows:

1. julia> M = Matrix{Float64}[rand(1000,1000) for i = 1:10];

3. julia> pmap(svdvals, M);

Julia’s pmap is designed for the case where each function call does a large amount of work. In contrast, @distributed for can handle situations where each iteration is tiny, perhaps merely summing two numbers. Only worker processes are used by both pmap and @distributed for for the parallel computation. In case of @distributed for, the final reduction is done on the calling process.

Remote References and AbstractChannels

Remote references always refer to an implementation of an AbstractChannel.

A concrete implementation of an AbstractChannel (like Channel), is required to implement put!, take!, fetch, isready and wait. The remote object referred to by a Future is stored in a Channel{Any}(1), i.e., a Channel of size 1 capable of holding objects of Any type.

RemoteChannel, which is rewritable, can point to any type and size of channels, or any other implementation of an AbstractChannel.

The constructor RemoteChannel(f::Function, pid)() allows us to construct references to channels holding more than one value of a specific type. f is a function executed on pid and it must return an AbstractChannel.

For example, RemoteChannel(()->Channel{Int}(10), pid), will return a reference to a channel of type Int and size 10. The channel exists on worker pid.

Methods put!, [take!](/read/julialang-1.

（本文超出站点单文档体积上限，已截断）