Master/slave multicore processing in R

Question

I would like to parallelize the while loop in the following code:

work <- function(n) {
  # Do some intensive work (e.g explore a graph starting at n).
  # After this, we don't need to execute work() on nodes in excluding.
  # (e.g exclude could be the nodes explored/reached from n)
  # n is just an example. exclude can be a potentially large set.
  Sys.sleep(2)
  exclude <- c(n, sample(nodes, rbinom(1, length(nodes), 0.5)))
  return(exclude)
}

nodes <- 1:1e3

#Order of execution doesn't matter
nodes <- sample(nodes)

#parallelize this loop
while(length(nodes) > 0) {
  n <- nodes[1]
  exclude <- work(n)
  nodes <- setdiff(nodes, exclude)
}

It doesn't matter if work() is executed on an excluded node, but we would like to minimize such instances. The objective of the while loop above is to run work() as few times as possible

This is not an embarrassingly parallel computation, so I don't know how to use parLapply directly. A master-slave framework could be used, but I don't know of any for multicore programming (on Windows).

As a concrete example, you can think of work(n) as graph_exploration(n) (the function the finds all nodes connected to n) and exclude as the nodes in the connected component of n. The final objective is to find one node from each connected component. You want to run graph_exploration(n) as few times as necessary because it is an expensive operation.

user_2733597 · Answer

Miheer,

Here is a proposed solution.

Pre-amble:

The core problem here (as I understand it) is to unblock the while loop whilst work() is number crunching. Essentially, you want the loop to be non-blocked so long as resources remain to initiate more work() calls and processing. Ok, how? Well, my recommendation is you use the future package.

The example below essentially creates a new process invocation of work() for each call. The invocation, however, will not block the while loop unless all the assigned worker processes are busy. You can see this as each work() invocation has a different process id as shown in the runtime output.

Thus, each work() runs independently, to finish we resolve all the futures and return the finals results.

Results:

Sequential runtime: 20.61 sec elapsed
Parallel runtime: 8.22 sec elapsed

I hope this points you in the right direction.

Caveat: You have to run through all nodes, but it does improve runtime.

Machine setup:

R version 3.4.1 (2017-06-30)
Platform: x86_64-w64-mingw32/x64 (64-bit)
Running under: Windows >= 8 x64 (build 9200)
[Windows 10, 8 Core Xeon, 64Gb RAM]

Parallel Code Example:

# Check for, and install and load required packages.
requiredPackages <-
  c("tictoc", "listenv", "future")


ipak <- function(pkg) {
  new.pkg <- pkg[!(pkg %in% installed.packages()[, "Package"])]
  if (length(new.pkg))
    install.packages(new.pkg, dependencies = TRUE)
  sapply(pkg, require, character.only = TRUE)
}

ipak(requiredPackages)

work <- function(n) {
  # Do some intensive work (e.g explore a graph starting at n).
  # After this, we don't need to execute work() on nodes in exclude.
  # (e.g exclude could be the nodes explored/reached from n)
  # n is just an example. exclude can be a potentially large set.
  Sys.sleep(2) # sample(.5:5))
  exclude <- n
  return(exclude)
}

plan(multiprocess, workers = 4L)
#plan(sequential)

nodesGraph  <- 1:10
nodesGraph  <- sample(nodesGraph)
nodesCount  <- length(nodesGraph)
resultsList <- listenv()

tic()
while ( nodesCount > 0 ) {
  n <- nodesGraph[[nodesCount]]
  ## This is evaluated in parallel and will only block
  ## if all workers are busy.
  resultsList[[nodesCount]] %<-% {
      list( exclude = work(n), 
            iteration = length(nodesGraph), 
            pid = Sys.getpid())
  }

  nodesGraph <- setdiff(nodesGraph, nodesGraph[[nodesCount]] )
  cat("nodesGraph",nodesGraph,"
")
  cat("nodesCount",nodesCount,"
")
  nodesCount = nodesCount - 1
}
toc()

## Resolve all futures (blocks if not already finished)
resultsList <- as.list(resultsList)
str(resultsList)

Parallel Runtime Output:

> source('/dev/stackoverflow/47230384/47230384v5.R')
nodesGraph 2 5 8 4 6 10 7 1 9 
nodesCount 10 
nodesGraph 2 5 8 4 6 10 7 1 
nodesCount 9 
nodesGraph 2 5 8 4 6 10 7 
nodesCount 8 
nodesGraph 2 5 8 4 6 10 
nodesCount 7 
nodesGraph 2 5 8 4 6 
nodesCount 6 
nodesGraph 2 5 8 4 
nodesCount 5 
nodesGraph 2 5 8 
nodesCount 4 
nodesGraph 2 5 
nodesCount 3 
nodesGraph 2 
nodesCount 2 
nodesGraph  
nodesCount 1 
8.22 sec elapsed
List of 10
 $ :List of 3
  ..$ exclude  : int 2
  ..$ iteration: int 1
  ..$ pid      : int 10692
 $ :List of 3
  ..$ exclude  : int 5
  ..$ iteration: int 2
  ..$ pid      : int 2032
 $ :List of 3
  ..$ exclude  : int 8
  ..$ iteration: int 3
  ..$ pid      : int 16356
 $ :List of 3
  ..$ exclude  : int 4
  ..$ iteration: int 4
  ..$ pid      : int 7756
 $ :List of 3
  ..$ exclude  : int 6
  ..$ iteration: int 5
  ..$ pid      : int 10692
 $ :List of 3
  ..$ exclude  : int 10
  ..$ iteration: int 6
  ..$ pid      : int 2032
 $ :List of 3
  ..$ exclude  : int 7
  ..$ iteration: int 7
  ..$ pid      : int 16356
 $ :List of 3
  ..$ exclude  : int 1
  ..$ iteration: int 8
  ..$ pid      : int 7756
 $ :List of 3
  ..$ exclude  : int 9
  ..$ iteration: int 9
  ..$ pid      : int 10692
 $ :List of 3
  ..$ exclude  : int 3
  ..$ iteration: int 10
  ..$ pid      : int 2032

Sequential Runtime output

> source('/dev/stackoverflow/47230384/47230384v5.R')
nodesGraph 6 2 1 9 4 8 10 7 3 
nodesCount 10 
nodesGraph 6 2 1 9 4 8 10 7 
nodesCount 9 
nodesGraph 6 2 1 9 4 8 10 
nodesCount 8 
nodesGraph 6 2 1 9 4 8 
nodesCount 7 
nodesGraph 6 2 1 9 4 
nodesCount 6 
nodesGraph 6 2 1 9 
nodesCount 5 
nodesGraph 6 2 1 
nodesCount 4 
nodesGraph 6 2 
nodesCount 3 
nodesGraph 6 
nodesCount 2 
nodesGraph  
nodesCount 1 
20.61 sec elapsed
List of 10
 $ :List of 3
  ..$ exclude  : int 6
  ..$ iteration: int 1
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 2
  ..$ iteration: int 2
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 1
  ..$ iteration: int 3
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 9
  ..$ iteration: int 4
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 4
  ..$ iteration: int 5
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 8
  ..$ iteration: int 6
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 10
  ..$ iteration: int 7
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 7
  ..$ iteration: int 8
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 3
  ..$ iteration: int 9
  ..$ pid      : int 12484
 $ :List of 3
  ..$ exclude  : int 5
  ..$ iteration: int 10
  ..$ pid      : int 12484