Gemeinsamer Speicher parallel in foreach R

Question

Problembeschreibung:

Ich habe eine große Matrix c, im RAM-Speicher geladen. Mein Ziel ist es, durch parallele Verarbeitung nur Lesezugriff darauf zu haben. Wenn ich die Verbindungen jedoch entweder doSNOW, doMPI, big.matrix usw. erstelle, erhöht sich die verwendete RAM-Menge dramatisch.

Gibt es eine Möglichkeit, einen gemeinsamen Speicher ordnungsgemäß zu erstellen, in dem alle Prozesse lesen können, ohne eine lokale Kopie aller Daten zu erstellen?

Beispiel:

libs<-function(libraries){# Installs missing libraries and then load them 
    for (lib in libraries){ 
    if(!is.element(lib, .packages(all.available = TRUE))) { 
     install.packages(lib) 
    } 
    library(lib,character.only = TRUE) 
    } 
} 

libra<-list("foreach","parallel","doSNOW","bigmemory") 
libs(libra) 

#create a matrix of size 1GB aproximatelly 
c<-matrix(runif(10000^2),10000,10000) 
#convert it to bigmatrix 
x<-as.big.matrix(c) 
# get a description of the matrix 
mdesc <- describe(x) 
# Create the required connections  
cl <- makeCluster(detectCores()) 
registerDoSNOW(cl) 
out<-foreach(linID = 1:10, .combine=c) %dopar% { 
    #load bigmemory 
    require(bigmemory) 
    # attach the matrix via shared memory?? 
    m <- attach.big.matrix(mdesc) 
    #dummy expression to test data aquisition 
    c<-m[1,1] 
} 
closeAllConnections() 

RAM: im Bild oben, können Sie feststellen, dass der Speicher eine Menge bis foreach Ende erhöht und es wird befreit.

Answer 1

Ich denke, die Lösung für das Problem kann aus dem Beitrag von Steve Weston, der Autor des foreach Pakets, here gesehen werden. Dort gibt er an:

Das doParallel-Paket wird Variablen automatisch an die Worker exportieren, die in der foreach-Schleife referenziert werden.

Also ich denke, das Problem, dass in Ihrem Code ist Ihre große Matrix c in der Zuordnung c<-m[1,1] verwiesen wird. Versuchen Sie einfach xyz <- m[1,1] und sehen Sie, was passiert.

Hier ist ein Beispiel mit einer big.matrix Datei-backed:

#create a matrix of size 1GB aproximatelly 
n <- 10000 
m <- 10000 
c <- matrix(runif(n*m),n,m) 
#convert it to bigmatrix 
x <- as.big.matrix(x = c, type = "double", 
       separated = FALSE, 
       backingfile = "example.bin", 
       descriptorfile = "example.desc") 
# get a description of the matrix 
mdesc <- describe(x) 
# Create the required connections  
cl <- makeCluster(detectCores()) 
registerDoSNOW(cl) 
## 1) No referencing 
out <- foreach(linID = 1:4, .combine=c) %dopar% { 
    t <- attach.big.matrix("example.desc") 
    for (i in seq_len(30L)) { 
    for (j in seq_len(m)) { 
     y <- t[i,j] 
    } 
    } 
    return(0L) 
} 

## 2) Referencing 
out <- foreach(linID = 1:4, .combine=c) %dopar% { 
    invisible(c) ## c is referenced and thus exported to workers 
    t <- attach.big.matrix("example.desc") 
    for (i in seq_len(30L)) { 
    for (j in seq_len(m)) { 
     y <- t[i,j] 
    } 
    } 
    return(0L) 
} 
closeAllConnections() 

Answer 2

Alternativ, wenn Sie unter Linux/Mac sind und Sie möchten eine Kuh geteilt Speicher, benutze Gabeln. Laden Sie zunächst alle Ihre Daten in den Hauptthread und starten Sie dann die Arbeitsthreads (Forks) mit der allgemeinen Funktion mcparallel aus dem Paket parallel.

Sie können ihre Ergebnisse mit mccollect oder mit der Verwendung von wirklich gemeinsam genutzten Speicher mit der Rdsm Bibliothek, wie diese sammeln:

library(parallel) 
library(bigmemory) #for shared variables 
shared<-bigmemory::big.matrix(nrow = size, ncol = 1, type = 'double') 
shared[1]<-1 #Init shared memory with some number 

job<-mcparallel({shared[1]<-23}) #...change it in another forked thread 
shared[1,1] #...and confirm that it gets changed 
# [1] 23 

Sie können bestätigen, dass der Wert wirklich in backgruound aktualisiert wird, wenn Sie verzögern die Schreib:

fn<-function() 
{ 
    Sys.sleep(1) #One second delay 
    shared[1]<-11 
} 

job<-mcparallel(fn()) 
shared[1] #Execute immediately after last command 
# [1] 23 
aaa[1,1] #Execute after one second 
# [1] 11 
mccollect() #To destroy all forked processes (and possibly collect their output) 

Um concurency zu steuern und Rennbedingungen verwenden Sperren vermeiden:

library(synchronicity) #for locks 
m<-boost.mutex() #Lets create a mutex "m" 

bad.incr<-function() #This function doesn't protect the shared resource with locks: 
{ 
    a<-shared[1] 
    Sys.sleep(1) 
    shared[1]<-a+1 
} 

good.incr<-function() 
{ 
    lock(m) 
    a<-shared[1] 
    Sys.sleep(1) 
    shared[1]<-a+1 
    unlock(m) 
} 

shared[1]<-1 
for (i in 1:5) job<-mcparallel(bad.incr()) 
shared[1] #You can verify, that the value didn't get increased 5 times due to race conditions 

mccollect() #To clear all threads, not to get the values 
shared[1]<-1 
for (i in 1:5) job<-mcparallel(good.incr()) 
shared[1] #As expected, eventualy after 5 seconds of waiting you get the 6 
#[1] 6 

mccollect() 

---

Edit:

I Abhängigkeiten ein wenig vereinfacht, indem Rdsm::mgrmakevar in bigmemory::big.matrix auszutauschen.mgrmakevar intern ruft big.matrix auf jeden Fall, und wir brauchen nichts mehr.