# Distributed: Passing views of an array for read access to workers (using pmap)

**URL:** <https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379>\
**Category:** General Usage\
**Tags:** question, performance, parallel, distributed, views\
**Created:** [January 29, 2024, 7:55am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379 "2024-01-29T07:55:20Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![james](https://avatars.discourse-cdn.com/v4/letter/j/bc8723/32.png) [@james](https://discourse.julialang.org/u/james)\
**Post date:** [January 29, 2024, 7:55am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/1 "2024-01-29T07:55:21Z")

</div>

I have a large array and want multiple workers to have read access to that array in parallel. Each worker works on a disjoint interval of the array.

My plan was to create a view of each of the sections of the array, and then pmap the workers’ function on a list of these views.  
However, I found that the creation of these workers is slow; with the example below only the _creation_ takes about 1.4 seconds; with more data to be shared it takes much more time. In addition, there is much more data allocated in the parallel case (874.982 MiB) compared to the sequential case (2.307 MiB).  
Is it possible that my approach to share parallel reader access to disjoint junks of the array is inefficient? Is the data possibly copied? Is there a more efficient way to provide parallel reader access to disjoint intervals?

Here is a minimal example, showing how long it takes to create the workers:

```julia
using Distributed

arr = rand(10^8)

# we partition arr in n_workers (almost) many equally sized junks
n_workers = length(workers())
size_per_worker = ceil(Int64,length(arr)/n_workers)
taskseq = []
for i in 1:n_workers
    start = (i-1)*size_per_worker + 1
    stop = min(i*size_per_worker,length(arr))
    push!(taskseq,@view(arr[start:stop]))
end

# this is the function that each worker should executed in parallel
# input: A view of the array, and the start time of pmap
@everywhere function worker_function((myarr,pmap_start_time))
    worker_start_time = time()
    println("Time it took to start function (worker): ", worker_start_time - pmap_start_time)
    flush(stdout)
    retval = 0
    for i in 1:length(myarr)
        retval += myarr[i]
    end
    println("Time to process in function (worker):", time() - worker_start_time)
    flush(stdout)
    return retval
end

seq_dict = Dict()
seq_dict[0] = 0 
retval = 0

println("starting sequential")
start_time = time()
@time worker_function((arr,start_time))
end_time = time()
println("time for sequential: ", end_time - start_time)

println("starting pmap"); flush(stdout)
start_time = time()
start_time_vec = ones(length(taskseq)).*start_time

# we execute worker_function in parallel
@time pmap(worker_function,zip(taskseq,start_time_vec))

end_time = time()
println("time of parallel: ", end_time - start_time)

```

The output would be e.g. when calling this script with 4 workers (`julia -p 4 filename.jl`):

> starting sequential  
> Time it took to start function (worker): 0.027009010314941406  
> Time to process in function (worker):0.561241865158081  
> 0.569265 seconds (32.95 k allocations: 2.307 MiB, 3.94% compilation time)  
> time for sequential: 0.5942938327789307  
> starting pmap  
> From worker 2: Time it took to start function (worker): 1.4049251079559326  
> From worker 3: Time it took to start function (worker): 1.405060052871704  
> From worker 5: Time it took to start function (worker): 1.409548044204712  
> From worker 4: Time it took to start function (worker): 1.4143550395965576  
> From worker 2: Time to process in function (worker):0.21155595779418945  
> From worker 3: Time to process in function (worker):0.22400593757629395  
> From worker 5: Time to process in function (worker):0.23674607276916504  
> From worker 4: Time to process in function (worker):0.24160385131835938  
> 1.692557 seconds (1.72 M allocations: 874.982 MiB, 5.34% gc time, 28.35% compilation time)  
> time of parallel: 1.7246119976043701

You can see that the time to just start the functions on the workers is quite significant (and increases when arr is larger). One possible reason could be that the memory allocation is much larger in the parallel case (874.982 MiB) compared to the sequential one ( 2.307 MiB).

Note that while each worker does a summation task, which could of course be done more efficient by mapping the (+) function, I explictiy want to execute a for loop on each worker, since my (more complex) usecase requires multiple function calls _per item_ of the array.

---

<div class="post-metadata">

**Author:** ![fatteneder](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/fatteneder/32/33991_2.png) [@fatteneder](https://discourse.julialang.org/u/fatteneder)\
**Post date:** [January 29, 2024, 9:09am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/2 "2024-01-29T09:09:03Z")

</div>

Take a look at [Introduction · DistributedArrays.jl](https://juliaparallel.org/DistributedArrays.jl/stable/#Distributed-Arrays)

---

<div class="post-metadata">

**Author:** ![james](https://avatars.discourse-cdn.com/v4/letter/j/bc8723/32.png) [@james](https://discourse.julialang.org/u/james)\
**Post date:** [January 29, 2024, 9:39am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/3 "2024-01-29T09:39:59Z")

</div>

Many thanks, this seems to be the right way. In particular, when constructing a distributed array ([[Introduction · DistributedArrays.jl](https://juliaparallel.org/DistributedArrays.jl/stable/#Constructing-Distributed-Arrays)]) it is possible to pass the `procs` argument, so that each of the processes gets a part of the array. That seems to be the solution.

However, I am not sure from the link how to most efficiently _transform_ my data into a distributed array with each worker having a separate junk. My use case is that the data (`arr`) has been generated in the main process and now needs to be shared among the workers. This is described [in this section](https://juliaparallel.org/DistributedArrays.jl/stable/#Construction-from-arrays-generated-on-separate-processes)

But calling simply

` darr = DArray(arr)`  
yields

> ERROR: LoadError: type Float64 has no field where

and calling  
`darr = DArray(arr;procs=workers())`  
yields

> (M1Max) (base) tass@arm64-apple-darwin20 Aljaz % julia -p 4 test\_parallel\_darray.jl  
> ERROR: LoadError: MethodError: no method matching DArray(::Vector{Float64}; procs::Vector{Int64})

The example code does some reshaping, but I do not understand why and how these dimensions are determined. Maybe I need to reshape as well?

Do you know how to distribute the data to all workers?

---

<div class="post-metadata">

**Author:** ![fatteneder](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/fatteneder/32/33991_2.png) [@fatteneder](https://discourse.julialang.org/u/fatteneder)\
**Post date:** [January 29, 2024, 2:25pm UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/4 "2024-01-29T14:25:51Z")

</div>

Maybe something like this

```julia
julia> using Distributed

julia> addprocs(4)
4-element Vector{Int64}:
 2
 3
 4
 5

julia> @everywhere using DistributedArrays

julia> A = randn((100,100))
100×100 Matrix{Float64}:
...

julia> DA = distribute(A)

julia> pmap(procs(DA)) do i
           DA[:l] # this gets you the local part of DA that is available on the ith process
       end
2×2 Matrix{Matrix{Float64}}:
 [-1.32192 -1.03872 … -1.45507 -1.03656; 0.9152 0.935673 … 0.552808 -0.524983; … ; 1.29931 0.584092 … -0.0659655 0.558668; 0.779895 1.13761 … 0.105637 -1.30604] … [1.47834 -2.07625 … 0.286428 0.723728; 0.870625 1.94568 … -0.68 1.21642; … ; -1.33853 0.28787 … -1.09238 -0.573003; 0.89278 -0.146452 … -0.214458 -1.67541]
 [0.863741 0.474698 … 0.155879 -0.0570954; 0.559998 -1.62241 … -0.16028 1.39969; … ; -0.447555 0.427656 … 0.293934 -2.45839; 0.150076 0.542755 … -0.324436 0.664975] [1.05879 0.769584 … -0.427145 -0.44544; 0.0646921 0.11107 … -0.639559 -0.348907; … ; -0.802571 1.13975 … 0.954727 -1.00489; -2.02937 1.72174 … 0.157684 1.59962]

```

---

<div class="post-metadata">

**Author:** ![james](https://avatars.discourse-cdn.com/v4/letter/j/bc8723/32.png) [@james](https://discourse.julialang.org/u/james)\
**Post date:** [January 29, 2024, 2:31pm UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/5 "2024-01-29T14:31:21Z")

</div>

Thank you, that does help. Using the DistributedArrays package (and the `[:L]` as you suggested), I could now make it run. There still seems to be a performance issue, but the bottle neck seems to be the number of allocations; I will mark this thread as solved (the distributed arrays did the trick) and open a new one with using (a) distributed arrays and (b) highlighting the memory allocation problem.

---

<div class="post-metadata">

**Author:** ![abraemer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/abraemer/32/51403_2.png) [@abraemer](https://discourse.julialang.org/u/abraemer)\
**Post date:** [January 30, 2024, 3:01am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/6 "2024-01-30T03:01:22Z")

</div>

Is there a reason why you went for processes (via Distributed.jl) instead of threads? Threads are generally much more light-weight and easier to work with. You really only need processes when you want to distribute work across multiple physical machines.

Just replace this line

```julia
@time pmap(worker_function,zip(taskseq,start_time_vec))

```

With something like

```julia
@time Threads.@threads for task in taskseq
    worker_function((task, time()))
end

```

and start Julia with some threads by replacing `-p` with `-t`: `julia -t 4`

---

<div class="post-metadata">

**Author:** ![carstenbauer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/carstenbauer/32/4981_2.png) [@carstenbauer](https://discourse.julialang.org/u/carstenbauer)\
**Post date:** [January 30, 2024, 5:53am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/7 "2024-01-30T05:53:28Z")

</div>

> [@abraemer](#):
>
> Just replace this line
> 
> ```julia
> @time pmap(worker_function,zip(taskseq,start_time_vec))
> 
> ```
> 
> With something like
> 
> ```julia
> @time Threads.@threads for task in taskseq
> worker_function((task, time()))
> end
> 
> ```
> 
> and start Julia with some threads by replacing `-p` with `-t`: `julia -t 4`

Since you want to record the results and also, potentially, want load-balancing you probably want to translate this as

```julia
tasks = map(zip(taskseq,start_time_vec)) do x
    Threads.@spawn worker_function(x)
end
fetch.(tasks)

```

---

<div class="post-metadata">

**Author:** ![james](https://avatars.discourse-cdn.com/v4/letter/j/bc8723/32.png) [@james](https://discourse.julialang.org/u/james)\
**Post date:** [January 31, 2024, 8:40am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/8 "2024-01-31T08:40:37Z")

</div>

Many thanks for your responses. That does the trick; my computation is much faster now!

However, I want to understand why. My reason for using multiple processes rather than multiple threads is because I wanted to have “real parallelism” i.e. the jobs should run on different cores (in my case, it’s a single machine with 10-40 cores) at the same time. My understanding of threads was that the computer never runs multiple threads at the same time, but rather starts them, one after the other, and the by context switching executes some lines of one thread, then some lines of another thread. I.e. if there was a line of code waiting for an outside action to be triggered (e.g. waiting for a network package), it makes sense to put that waiting part into a separate thread, since that allows other parts of the program to be run while waiting.  
Here, however, there does not seem to be a line waiting for an external event to happen. Hence I wanted to use processes rather than threads.

Is my understanding wrong? An answer [here](https://discourse.julialang.org/t/multi-threading-or-multi-processing-how-to-know-which-to-use-and-when/72278/3) says

> Well, both multithreading as well as multiprocessing give you parallelism.

So do both give parallelism in the sense of “true parallelism” as described above?

---

<div class="post-metadata">

**Author:** ![abraemer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/abraemer/32/51403_2.png) [@abraemer](https://discourse.julialang.org/u/abraemer)\
**Post date:** [January 31, 2024, 9:42am UTC](https://discourse.julialang.org/t/distributed-passing-views-of-an-array-for-read-access-to-workers-using-pmap/109379/9 "2024-01-31T09:42:42Z")

</div>

> [@james](#):
>
> So do both give parallelism in the sense of “true parallelism” as described above?

Yes - in Julia they do. In Python they don’t. Python (at least as long as the Global Interpreter Lock is still present) is fundamentally single threaded and so Python “threads” work as you describe: You start them and the Python runtime switches between them but at any given point in time only a single thread is actively working.

Julia has a similar concept to that: [`Task`s](https://docs.julialang.org/en/v1/manual/asynchronous-programming/). Basically a `Task` is a chunk of work. You can create multiple `Task`s and Julia will switch between them to minimize total execution (e.g. switch to another `Task` if the current `Task` needs to wait for I/O).

Threading build on top of Julia’s `Task`s. Basically a thread is a something that can work on completing `Task` and they can work “truly parallel”, i.e. each thread can use a different CPU core \[1\]. So if you start Julia with more threads, you can have more `Task`s running in parallel.

Another way of achieving “true parallelism” is using multiple processes. This is basically a different layer of abstraction. See when you tell your OS to start a program, it puts everything belonging to the program in a process.

So what is the difference between processes and threads? For starter each Julia process have their own separate threads. But what does that mean in practice? Having multiple processes is much more expensive than having multiple threads. That is because the OS allocates resources for process separately. So each Julia process needs to load the external libraries again\[2\] and needs to compile the code again. Sharing memory is also harder, since the OS usually prohibits a process from accessing a different process’ memory. So accessing common data has a higher overhead.

The main advantage of using different processes is the possibility of running them on physically different hardware, i.e. on multiple _computers_. In that sense processes are another layer on top of threads.

Summary:

1. `Task`s allow the queuing of multiple independent pieces of work.
2. Threads work on `Task`s. So multiple threads allow parallel execution on the same machine (by using multiple cores).
3. Processes allow distribution of work to multiple physical machines.

* * *

1. Note that by default the threads are not pinned to a specific CPU core, you’d need to use ThreadPinning.jl for that. It can have performance advantages. If you want more on this technicalities please ask 🙂 

2. This is the reason that sometimes you need to use multiple processes though. For example BLAS (linear algebra library) uses it’s own threads and that’s a [very complicated story](https://carstenbauer.github.io/ThreadPinning.jl/stable/explanations/blas/)