useRparallel

Parallel computing in R

using packages foreach and doSNOW

a classical linearly executed loop is performed in the base package of R by:

  	for(i in 1:iterations) { expr(i)  }

R uses only one processor to execute this task, because we don't tell it how to split it over several cores. If we have a task, where the single iterations are basically independent of the outcome of the other iterations, we can apply parallel code. Parallel computing is not trivial but requires a lot of tracking (What iterations are started where?) and communication (Master: “you have to calculate that now!” Worker: “Ok, I am done, here is the result! What now?”). There are packages in R that are providing all this. The simplest solution is the foreach package, because it mimics the classic for-loop. It splits the job into several independent jobs and distributes it to a parallel-backend, a structure of multiple processors that is known to R (How many cores? What type of workers?).

This is how it is called:

	foreach(i = 1:iterations) %dopar% { expr(i)  } 

The handling is a bit more complicated than the classical for-loop, but not much. Get information on that by calling the help file ?foreach.

Before calling foreach, we need to register a parallel backend. This is easiest set up with doSNOW, a wrapper, a simplifying toolkit for the package snow. We need to

1. tell it the number and the character of the nodes that we want to use.
2. register it for the use with foreach

library(foreach) #required once at the beginning of the programm
library(doSNOW) #required once at the beginning of the programm

workerlist <- rep("localhost", times = 5)

cl <- makeSOCKcluster(workerlist)

registerDoSNOW(cl)

foreach(i = 1:iterations) %dopar% {  expr(i)  }
 
stopCluster(cl)

the short description localhost is sufficient to tell the function makeCluster() that the workers should be created on the same mashine that is presently running the R process. The times = argument tells snow how many workers should be invoked. You must not call more workers than are available on the computer minus one. Our workstation has 24 virtual cores which means you can call 23 workers (leaving one core unoccupied for all the other stuff the computer has to do). Most desktop computers today have rather 2, 4 or 6 cores.

Finally, The call of registerDoSNOW() makes the cluster available for the use with foreach() which you apply to call a parallel loop to execute your task (here, expr(i)).

It is very important to terminate the Cluster after the job is done by calling stopCluster().

some tips:

• You can check if the program is running and if it is using the number of cores that you assigned by running the little program htop (just type that into your command line). This gives you a nice view of the processes. you can exit the program by typing q.

• If you started a job, that is not producing the expected output, you should use top to view the list of processes and than kill the running R processes. Just hit k which will ask you for a PID number. Enter the number of the first R process on your username on the list. You will be asked for confirmation, which you give by typing y. This should end all jobs of R in the list. If not, type the next PID and proceed until R is silenced.

• After starting the cluster, other functions are available with the package snow, e.g. a parallel apply function or a parallel random number generator. Check the manual for the snow package.

However, not all iterative tasks are quicker with parallel implementation. The use of these functions uses quite a bit of calculation power itself and therefore it just doesn't make sense to use it for operations that are only iterated a few times.

working remotely

Since the computer on your desk is not equipped with many cores, in most cases you want to run jobs on a remote computer such as the workstation.

To connect remotely to the workstation we are going to use SSH. This is a program in Linux that allows you to log into the other computer. By default this gives access to anybody who has a proper account with a password on that computer after asking for a password. To connect to our workstation, you would open the terminal and type

	ssh <user>@kefi118

or

	ssh <user>@168.38.184.118

and you will be asked

	<user>@kefi118's password:

After you provide it (you won't see any signs or stars to fill in as you type) you will be recognized and logged in, and the terminal will behave as the normal bash terminal in Linux. Now, you could execute R by typing R. Everything you run here, will be calculated on the workstation, with all the power it provides. You could execute code for a parallel calculation with up to 23 cores.

This is quite convenient for testing parallel tasks from home with low resolution and short timeseries. However, if you run a long simulation, this would occupy your terminal at home while the calculation is running on the remote computer. This is simply not necessary. The easiest alternative is to implement the code as a standalone R-Script, that runs autonomously on the workstation. This means:

• include all functions in the head of the R-script, before calling them

• configure the cluster with "localhost" workers.

• supress all terminating promts by wrapping into try(). e.g. a function call which returns an error would stop the whole simulation.

• implement saving of the output in one or multiple files by using save(), write.table() or write.csv()

• use the return() function and the .combine parameter of foreach() to collect the output into a single R object.

• If you load data from saved objects (a csv table or a previously saved R-object) make sure that the path works. It is best to use absolute paths to the file. You can use the tilde ~ as a shortcut to your home directory. If you want to load an R object called file12 stored in a folder results right in your home directory you can call

  load("~/results/file12")
  

  to restore the R-object with its original name. Keep in mind that Linux uses a different path structure than Windows. On a Windows cluster this would need to spell:

  load("~\\results\\file12")
  

  Thus, you might need to modify your path definitions for loading and saving if you developed the code on a different OS than you run it in parallel.

• preferentially, do not set a working directory. For reasons of simplicity, you should copy your R-script in a separate, empty folder on the remote computer (using scp or a graphical ssh client such as FileZilla, or a USB stick) and navigate to that folder before starting it. All output is then saved in that folder.

• for debugging, specify an output file for error messages of the cluster:

  cl <- makeSOCKcluster(workerlist, outfile='out_messages.txt')

Now, we could easily run the remote computers installation of R in the terminal and copy our code into that window. The problem is, when we close the R-terminal, all R processes will be shut down and the simulation will be terminated. This is not exactly what we want. Alternatively, we could call the programm Rscript in the Linux terminal, which runs an r.-file from the beginning to the end and returns the output requested. But once we close the SSH connection, the Rscript process – which is owned by the user – will be terminated as well. Instead, we want to be able to close the connection between our home computer and the server, while the simulation keeps running. This is enabled by the program screen in Linux. Among many other features, screen provides the ability to start a process as a detached process, or daemon (option -d). This means, it runs in background and is not connected to the user any more.

So, you can start a Terminal at your home computer, log in via SSH, start an R-script file, containing our parallelised simulation code as detached process and disconnect from the server by calling in the Terminal:

	ssh <user>@kefi118
	cd path/to/file/
	screen -d -m Rscript filename.r 
	exit

Before you exit, you can check if the program is running by visiting htop to see the occupation of processors.

R Studio - Server

R multi-node cluster

It might be an attractive feature, once in a while, to run parallel tasks on our own temporal cluster. A cluster is a network of multiple computers, each having multiple cores. This would join the calculation power of the workstation with your local computer and even somebody elses computer.

The R packages foreach and snow provide this feature.

Requirements

set-up

http://homepage.stat.uiowa.edu/~luke/R/cluster/cluster.html

http://blog.revolutionanalytics.com/2011/02/parallel-computing-r-openss-dosnow.html

http://helmingstay.blogspot.fr/2011/02/snow-and-ssh-secure-inter-machine.html

Particularities in Windows

If starting the cluster from a Windows mashine, several extra steps are necessary:

• enable ssh from command line in the standard shell: However you do ssh in windows, usually it is run in an own terminal application like cygwin, msys or the git powershell. The commands are not available the standard command line shell (called cmd). To ensure the availability of commands like ssh <username>@162.38.184.118 you will need to add an entry to the PATH variable of your Windows environment. PATH

• Set a HOME variable: ssh is used to search for a directory called .ssh within the home directory of the current user. The location of the home directory in Linux is usualy specified by a variable called $HOME which makes ssh in windows to expect a variable %HOME%. However, this variable by default is called %HOMEPATH% in Windows, so we have to create a new environmental variables. You can do that from the terminal by typing

  set HOME=C:\Users\yourusername

• identify the locations of the Rscript.exe and snow-package on the cluster node.

This is very similar to setting up a local cluster for parallel computing. While for a local cluster all processes are started on localhost, now the single nodes, a.k.a. the workers, need to be specified more precisely including their adress on the network, the username to log-in with, as well as the location of the Rscript executable and the snow package. This looks very different depending on the computer and operating system they are running on. For a remote Ubuntu-system with IP adress 162.38.184.54 and a user-account called that you can access passwordless via ssh using a public-private key, this would be:

ubuWorker <- list(host = "162.38.184.54", user = "<username>",
         rscript = "/<path>/<to>/R/bin/Rscript",
		 snowlib = "/<path>/<to>/R/library/")

In the particular case of our workstation, it reads:

workstation <- list(host = "162.38.184.118", user = "<username>",
         rscript = "/usr/lib/R/bin/Rscript",
		 snowlib = "/usr/lib/R/library/")

For Windows computers it is something like:

winWorker <- list(host = "162.38.184.63", user = "<username>"
				rscript = "C:/R/R-2.15.3/bin/x64/Rscript.exe", 
				snowlib = "C:/R/R-2.15.3/library")

After setting these, you can compile a workerlist using several nodes with the same specifications, and fire up the cluster. Note, that you now might need to specify the master adress, if you adress nodes on a different computer, so each worker knows were to report back. Before running foreach loops, you need to register the cluster for use with the foreach package. Don't forget to shut down the cluster after use!

workerlist <- rep(list(workstation), times = 20)
cl <- makeSOCKcluster(workerlist, master = "your.ip.adress")
registerDoSNOW(cl)

foreach(i = 1:iterations) %dopar% {  expr(i)  }
 
stopCluster(cl)

Of course you can combine different nodes to one big cluster.

workerlist <- c( 
				rep(list(workstation), times = 20), 
				rep(list(ubuWorker), times = 3),
				rep("localhost", times = 10)  
				) 
cl <- makeSOCKcluster(workerlist, master = "your.ip.adress")
registerDoSNOW(cl)

foreach(i = 1:iterations) %dopar% {  expr(i)  }
 
stopCluster(cl)

Once the cluster is created by makeSOCKcluster() you can check its structure with

clusterCall(cl,function() Sys.info()[c("nodename","machine")])

However, there are limitations for this kind of parallelisation. If you save the output of each iteration into a file, the destination needs to be specified to be on the clusters master computer. Otherwise, the output will be saved on the different computers and need to be collected afterwards. The tasks best suited for this kind of cluster are tasks that produce a combined output into a single R-object. By default, foreach() is collecting all the nodes' output in one list object. The output of a single iteration must be specified using return(). It is also good practice to clear the occupied memory after each iteration using R's garbage collection gc().

foreach(i = 1:100) %dopar% {  

	temp <- rnorm(10)
	
	return(temp)
	gc() 
} -> listoutput

This saves a list object with all the collected returns:

> listoutput
[[1]]
 [1] -1.2329679  0.2285091  1.1509093  0.6880026 -0.3177907  0.4330497
 [7]  1.8169844  2.9795603  1.7304972 -1.7805568

[[2]]
 [1] -0.4377486  0.6299124 -0.6652313 -0.8424409  0.4772709 -0.3884619
 [7] -0.1358137  0.6812075 -0.8121486 -0.4542451

[[3]]
 [1] -1.0560805  0.9194420 -0.1892013 -0.2738968  1.1095158 -0.4951034
 [7]  0.5720476 -0.9425755 -0.3075008 -0.9093797

[[4]]
 [1]  0.81097289  0.65003397 -1.11184687 -1.06623967 -1.52171816  0.58934076
 [7]  0.08573481 -0.99550821  0.13800923  1.27614739

[[5]]
 [1] -0.2244869 -0.5261722 -1.1098975 -1.6393781  0.8345173 -0.6166367
 [7]  0.7453575 -0.5893772 -0.1341751 -0.3650106

[[6]]
 [1] -0.9992371  0.8430372  1.3805995 -1.3803696  0.6018764 -0.4097727
 [7] -0.3208447 -1.3857410 -0.4699092 -0.1074893

 ... 

Depending on the type of content you might want to combine the output of the foreach loop. You can specify how that should be done by using the .combine parameter of foreach() to bind the returns as rows in a data.frame table object.

foreach(i = 1:43, .combine = "rbind") %dopar% {  

	temp <- rnorm(5)
	
	return(temp)
	gc() 
} -> dataframeoutput

> dataframeoutput
                 [,1]        [,2]        [,3]        [,4]          [,5]
result.1   0.30429707 -0.09549955 -0.06769123  0.44828502 -1.2775854271
result.2   0.92825846  0.92929921 -0.54575716  0.06493218  1.6735847726
result.3  -0.06634894 -0.61437407 -1.07826831  0.55867509  0.8711318092
result.4  -1.01309855 -1.59606418 -1.85581323 -0.59560250  0.5535813540
result.5  -1.19435367  0.22395252  1.81436254  0.70218067 -1.4648754506
result.6   1.53370503  0.60823151 -0.28388320 -1.00578356 -0.0003790473
result.7  -0.62307511 -0.80386098  1.05256562 -1.39991735  1.3667377560
result.8  -0.30588025 -0.37303218  0.54065203 -0.15943930  0.0869040794
result.9  -0.93557312  1.71011575  0.08336999  0.65264126 -0.2366355002
result.10 -1.25065149  1.51511372 -1.29809495  0.23927960 -1.1575295759
...