Multiprocessing and distributed computing

Your code should be available to any process in which it is executed. For example, type the following in the Julia command prompt:

Process 1 knows the function `rand2', but process 2 does not.

The code is usually loaded from files or packages, and you can flexibly control which processes the code is loaded into. Let’s take the DummyModule' file.jl containing the following code.

So that all processes can reference the MyType, the DummyModule' file.jl must be loaded into each process. When calling include("DummyModule.jl")`it is loaded into only one process. To upload a file to each process, use the macro @everywhere` (Julia should be launched with the command `julia -p 2').

As usual, the DummyModule module is not included in the process scope. This requires the use of a keyword. using or import. Moreover, when the DummyModule is injected into the scope in one process, it doesn’t happen in others.:

However, you can still pass MyType to a process in which the DummyModule module is loaded but not included in the scope.

The file can also be preloaded into several processes at startup using the -L flag, and an activation script can be used to activate its calculation.

In the example above, the Julia process executing the activation script has an id with the value 1, as does the process providing the interactive command prompt.

Finally, if DummyModule.jl is not a separate file, but a package. The using DummyModule operator loads DummyModule.jl` is included in all processes, but includes the scope only in the process where the operator was called. using.

Two types of clusters are supported in the basic Julia installation:

The following functions are available for programmatically adding, deleting, and querying processes in a cluster: addprocs, rmprocs, workers and others.

Before the challenge addprocs module Distributed must be explicitly loaded into the main process. It becomes available automatically in workflows.

To support other types of clusters, you can write a custom implementation of ClusterManager, as described below in the section ClusterManager objects.

In a distributed program, most of the work is done on sending messages and moving data. To ensure high performance and scalability, it is important to minimize the number of messages and the amount of data transferred. To do this, you need to understand how the various distributed programming constructs in Julia move data.

fetch can be regarded as an explicit data transfer operation, since moving an object to a local computer is requested directly. The macro @spawnat (and a number of related constructs) also moves data, but in a less obvious way, so this can be called an implicit data movement operation. Let’s consider two approaches to creating a random matrix and squaring it.

Method 1

Method 2

The difference seems trivial, but it actually makes a significant difference due to the behavior of the macro. @spawnat. In the first method, a random matrix is created locally, and then transferred to another process, where it is squared. In the second method, a random matrix is both created and squared in another process. Therefore, much less data is transmitted in the second case than in the first.

In this simplest example, the difference between the two methods is obvious, so it’s easy to choose the right one. However, in a real-world program, it may be necessary to carefully consider moving data and possibly take measurements. For example, if the matrix A is needed by the first process, it may be better to use the first method. If calculating the matrix A requires a lot of resources and only the current process has them, then it will inevitably have to be moved to another process. Finally, if the current process has little to do between calls '@spawnat' and fetch(Bref)', it might be better to abandon parallelism altogether. Let’s also imagine that instead of `rand(1000,1000)"we have a more complicated operation. In this case, it might make sense to add another operator. @spawnat` exclusively for her.

Expressions executed remotely by @spawnat, or closures assigned to remote execution using remotecall, can refer to global variables. Global bindings in the Main module are handled slightly differently than global bindings in other modules. Consider the following code snippet.

In this case, the function sum MUST be defined in the remote process. Note that A is a global variable defined in the local workspace. Workflow 2 does not have a variable named A in Main'. Sending the closure `+()→sum(A)+ to workflow 2 causes it to define Main.A. Main.A continues to exist in workflow 2 even after call control returns. remotecall_fetch. Remote calls with embedded global references (only in the `Main' module) manage global variables as follows.

As you may have already guessed, the memory associated with global variables can be cleared when they are reassigned in the main process, but this does not happen in workflows, as the bindings continue to operate. If certain global variables in remote nodes are no longer needed, they can be set to nothing using the method clear!. As a result, the associated memory is released during the next garbage collection cycle.

Therefore, global variables should be referenced with caution in remote calls. And it’s even better to avoid it altogether if possible. If it is still necessary to refer to global variables, it is recommended to use let blocks to localize them.

For example:

As you can see, the global variable A is defined in workflow 2, but B is captured as a local variable, and therefore there is no binding for B in workflow 2.

Fortunately, for many useful parallel computations, it is not necessary to move data. A common example is Monte Carlo simulation, for which several processes can simultaneously conduct independent model tests. We can use @spawnat to flip coins in two processes. First, let’s write the following function in the 'count_heads.jl` file.

The 'count_heads` function simply adds up n random bits. As follows, you can run tests on two computers, and then add up the results.

This example demonstrates a common and very effective parallel programming pattern. Iterations are performed independently in several processes, and then their results are combined using a function. The process of combining is called reduction, as it usually involves lowering the valence of the tensor: the vector of numbers is reduced to a single number, or the matrix is reduced to a single row or column, etc. In the code, this usually looks like the pattern x = f(x,v[i]), where x is the adder, f is the reduction function, and v[i] is the reduced elements. It is desirable that the function f be associative.: This way, the operations can be performed in any order.

Please note that our use case for this pattern in relation to count_heads can be generalized. We used two explicit operators @spawnat, so parallelism is limited to two processes. To run in any number of processes, you can use a parallel for loop running in distributed memory. In Julia, it can be written using @distributed as follows.

This construction implements a pattern of assigning iterations to several processes, followed by combining their results using the specified reduction (in this case, (+)). The result of each iteration is the value of the last expression inside the loop. As a result of calculating the entire parallel loop expression, the final answer is obtained.

Keep in mind that although parallel for loops are similar to sequential ones, their behavior is fundamentally different. In particular, iterations do not occur in a specific order, and values assigned to variables or arrays will not be available globally, since iterations are performed in different processes. The variables used inside the parallel loop are copied and transferred to each process.

For example, the following code will not work properly.

This code does not initialize all the elements of a, as there will be a separate copy in each process. Such parallel for loops should be avoided. Fortunately, this limitation can be circumvented using shared arrays.

Using external variables in parallel loops is perfectly justified if the variables are read-only.

Here, in each iteration, the function f is applied to a randomly selected element of the vector a, which is shared by all processes.

As you can see, the reduction operator can be omitted if it is not needed. In this case, the loop runs asynchronously, that is, it generates independent tasks in all available workflows and immediately returns an array of objects. Future, without waiting for the execution to complete. The caller can wait for the objects to be completed. Future anywhere further, calling for them fetch, or wait for execution to complete at the end of the loop, for which you should specify before it @sync, for example, `@sync @distributed for'.

In some cases, the reduction operator is not needed if we just want to apply the function to all integers in a certain range (or more generally to all elements of some collection). This is another useful operation called parallel matching. It is implemented in Julia through the function pmap. For example, you can calculate singular values for several large random matrices in parallel mode as follows.

Function 'pmap` in Julia is designed for cases where each function call performs a large amount of work. On the contrary, @distributed for is suitable for situations where each iteration is very simple, even to the point where two numbers are simply added together. For parallel computing using pmap and @distributed for use only workflows. In the case of @distributed for, the final reduction is performed in the calling process.

Deleted links are always associated with the implementation of the `AbstractChannel'.

In a specific implementation of AbstractChannel (for example, Channel), the methods must be implemented put!, take!, fetch, isready and wait. The deleted object that is referenced Future, stored in Channel{Any}(1), that is, in a channel of size 1 for objects of type `Any'.

An object RemoteChannel', which supports overwriting, can point to channels of any type and size, or any other implementation of `AbstractChannel.

The RemoteChannel(f) constructor::Function, pid)() allows you to create links to channels containing more than one value of a certain type. 'f` is a function executed in the 'pid` process. It should return the AbstractChannel object.

For example, the expression RemoteChannel(()->Channel{Int}(10), pid) returns a reference to an Int type channel of size 10. The channel exists in the 'pid` workflow.

Methods put!, take!, fetch, isready and wait for the object RemoteChannel are proxied to the auxiliary storage of the remote process.

Thus, RemoteChannel' can be used to reference `AbstractChannel objects implemented by the user. A simple example is the following DictChannel, which uses a dictionary as a remote storage.:

The above example of channels can be changed so that interprocess communication takes place, as shown below.

To process a single remote jobs channel, four workflows are started. Jobs identified by job_id are written to the channel. In this example, each remotely executed task reads the job_id, waits for a random amount of time, and writes a tuple consisting of the job_id, the timeout, and its own pid to the results channel. Finally, the main process outputs all the data from the results channel.

The data is inevitably copied to the remote call for execution. This happens both during remote calls and when data is saved in RemoteChannel' or `Future in another node. As expected, this leads to the formation of copies of the serialized objects in the remote node. However, when the destination node is a local node, that is, the ID of the calling process matches the ID of the remote node, the call is executed as local. It is usually (but not always) executed in another task, but there is no serialization or deserialization of the data. Therefore, the call refers to the same instances of objects that are being passed, but no copies are created. This behavior is demonstrated below.

As can be seen from the example, method calls put! for the local channel RemoteChannel when the object v is changed between calls, the same instance of the object is saved. When the node that owns rc is different, copies of v are created.

It should be noted that, as a rule, there is no problem in this. This should only be taken into account if the object is saved locally and at the same time changed after the call. In such cases, you may need to keep a `deepcopy' of the object.

This is also true for remote calls on the local node, as shown in the following example.

As you can see in this case, a remote call to a local node works the same way as a direct call. It modifies the local objects passed as arguments. When making a remote call, operations are performed with a copy of the arguments.

Once again, we repeat that, as a rule, there is no problem in this. If the local node is also used as a computing node and the arguments are used after the call, this feature should be taken into account and, if necessary, pass deep copies of the arguments to the call made in the local node. Calls to remote nodes always work with copies of arguments.

Shared arrays use the shared memory of the system to map the same array across multiple processes. SharedArray is a good choice when a large amount of data needs to be accessed simultaneously by two processes or more on the same computer. Support for shared arrays is provided by the SharedArrays module, which must be explicitly loaded into all participating processes.

An additional data structure is provided by an external package https://github.com/JuliaParallel/DistributedArrays.jl [DistributedArrays.jl] as a DArray'. Although in some ways the object is similar to `SharedArray, behavior https://github.com/JuliaParallel/DistributedArrays.jl [DArray] is different in many ways. In the case of SharedArray each process working with it has access to the entire array. On the contrary, in the case of an object https://github.com/JuliaParallel/DistributedArrays.jl [DArray] each process has local access to only a part of the data, and these parts are different for all processes.

SharedArray is indexed (for the purpose of assigning values and accessing them) just like a regular array, and this happens efficiently, since the corresponding memory area is available to the local process. Therefore, most algorithms naturally work with SharedArray, albeit in single-process mode. If the algorithm requires an input array Array, the base array can be extracted from SharedArray by calling the function sdata. For other types of AbstractArray' function 'sdata just returns the object itself, so sdata can be safely applied to any object of the Array type.

The constructor of the shared array looks like this.

It creates an N-dimensional shared array of bit type T and size dims, accessible to the processes specified in pids. Unlike distributed arrays, the shared array is accessible only from the participating processes specified in the named argument pids' (as well as from the process where it is created if this process is running on the same host). Note that only elements of types are supported in SharedArray. `isbits.

If the init function is specified with the signature initfn(S::SharedArray), it is called in all participating workflows. You can specify that each workflow should execute the init function for a separate part of the array in order to parallelize initialization.

Here is a small example.

Function SharedArrays.localindices provides separate one-dimensional ranges of indexes and is sometimes useful for dividing tasks between processes. Of course, you can divide the work as you like.

Since the underlying data is available to all processes, it is important to ensure that there are no conflicts. For example:

will result in undefined behavior. Since each process fills the entire array with its own pid, the result will be the pid of the process that was executed last (for any S element).

To consider a more complex example, let’s run the following kernel in parallel.

In this case, if we try to divide the work using a one-dimensional index, trouble is likely to arise: if q[i,j,t] is close to the end of the block assigned to one workflow, and q[i,j,t+1] is close to the beginning of the block. assigned to another process, there is a high probability that the part of q[i,j,t] will not be ready by the time it is needed to calculate q[i,j,t+1]. In such cases, it is better to divide the array into parts manually. Let’s divide the array by the second dimension. Define a function that returns the indexes (irange, jrange) assigned to this workflow.

Next, we define the core.

We will also define an auxiliary wrapper for the implementation of `SharedArray'.

Now let’s compare the three different versions. Running in the same process:

Using @distributed:

Dividing the work into parts:

If we create SharedArray arrays and measure the running time of these functions, we will get the following results (when using the julia -p 4 command).

Let’s execute the functions once to perform JIT compilation, and then apply the macro @time on the second execution.

The main advantage of `advection_shared! The reason is that the traffic between the workflows is minimized, so that each process can work with the assigned part for a long time.

Julia processes are started, managed, and combined into a logical cluster using cluster managers. The ClusterManager is responsible for the following actions:

The cluster in Julia has the following features.

Connections between workflows (using the built-in TCP/IP transport) are established as follows.

Although by default, the transport layer uses a regular socket. TCPSocket, the Julia cluster can provide its own transport.

Julia has two built-in cluster managers.:

The LocalManager is used to run additional workflows on the same host, allowing the use of multi-core and multiprocessor hardware.

addprocs(manager::FooManager) requires the implementation of the following methods in the FooManager.

For example, let’s see how the LocalManager manager is implemented, which is responsible for running workflows on the same host.

Method launch accepts the following arguments:

Method launch is called asynchronously in a separate task. Completion of this task signals that all requested workflows are running. Therefore, exiting the function launch SHOULD be performed as soon as all requested workflows are running.

The running workflows are connected to each other and to the main process according to the "all to all" scheme. If you specify the command line argument --worker[=<cookie>], the running processes will initialize themselves as workers and establish connections via TCP/IP sockets.

All the workflows in the cluster use the same cookie file, which is the main process. If the cookie is not specified (when using the '--worker` option), the workflow tries to read it from the standard input stream. The LocalManager and SSHManager pass the cookie to the new workflows through their standard input streams.

By default, the workflow listens on a free port at the address returned as a result of the call. getipaddr(). Using the optional --bind-to bind_addr[:port] argument, you can specify a specific listening address. This can be useful if the host is connected to multiple physical data lines.

An example of a non-TCP/IP transport in an implementation is MPI. In this case, the --worker parameter CANNOT be specified. Instead, new worker processes should call init_worker(cookie) before using any parallel constructs.

For each running workflow, the method launch must add a WorkerConfig object to launched (with initialization of the corresponding fields).

Most of the fields in the WorkerConfig are used by the built-in dispatchers. User cluster managers usually specify only io or host and `port'.

The manage(manager' method::FooManager, id::Integer, config::WorkerConfig, op::Symbol) is called at different points during the existence of the workflow with the corresponding values of op:

Named argument topology' passed to the function 'addprocs, defines how workflows should connect to each other:

The named argument lazy=true|false is valid only if the topology parameter has the value :all_to_all'. If the value is set to `true, immediately after the cluster is started, the main process is connected to all workers. Connections between specific workflows are established on the first remote call between them. This helps to reduce the amount of resources allocated initially for interaction within the cluster. Connections are established depending on the needs of the parallel program at runtime. The default value for the `lazy' argument is `true'.

Currently, sending a message between unconnected worker processes results in an error. This functionality and its associated interface should be considered experimental in nature and may change in future releases.

In addition to implementing parallelism in Julia, there are many external packages worth mentioning. For example, https://github.com/JuliaParallel/MPI.jl [MPI.jl] is the Julia wrapper for the MPI protocol. https://github.com/JuliaParallel/Dagger.jl [Dagger.jl] provides functionality similar to the library https://dask.org /[Dask] in Python, and https://github.com/JuliaParallel/Distributedarrays.jl [DistributedArrays.jl] supports distributed array operations in multiple workflows, as already described. above.

It is also necessary to mention the Julia GPU programming ecosystem, which includes the following components.

In the following example, we use DistributedArrays.jl and CUDA.jl to distribute the array across multiple processes. To do this, we will first bring it up using distribute() and `CuArray()'.

Don’t forget that DistributedArrays.jl must be imported into all processes using @everywhere.

In the following example, we use DistributedArrays.jl and CUDA.jl to distribute an array across multiple processes and call a universal function for it.

'power_method` systematically creates vectors and normalizes them. We did not specify the type signature in the function declaration. Let’s see if this works for the previously mentioned data types.

To conclude this brief overview of external packages, consider MPI.jl, the Julia wrapper for the MPI protocol. It would take too long to review all the internal functions, so it would be better to just evaluate the protocol implementation approach.

Let’s consider a simple script that calls each subprocess, creates an instance of its rank, and summarizes the ranks when the main process is reached.