memoria condivisa in foreach parallelo R

Question

Descrizione del problema:

Ho una grande matrice c, caricato in memoria RAM. Il mio obiettivo è attraverso l'elaborazione parallela di avere accesso in sola lettura ad esso. Tuttavia, quando creo le connessioni, utilizzo doSNOW, doMPI, big.matrix, ecc. La quantità di ram utilizzata aumenta in modo significativo.

C'è un modo per creare correttamente una memoria condivisa, da cui tutti i processi possono leggere, senza creare una copia locale di tutti i dati?

Esempio:

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: nell'immagine sopra, si potrebbe scoprire che la memoria aumenta molto fino foreach estremità e viene liberato.

Answer 1

Penso che la soluzione al problema possa essere vista dal post di Steve Weston, l'autore del pacchetto foreach, here. Qui:

Il pacchetto doParallel esegue l'esportazione automatica di variabili per gli operatori a cui viene fatto riferimento nel ciclo foreach.

Quindi penso che il problema è che nel codice vostro grande matrice c fa riferimento nell'assegnazione c<-m[1,1]. Prova semplicemente lo xyz <- m[1,1] e guarda cosa succede.

Ecco un esempio con un file-backed big.matrix:

#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

In alternativa, se siete su Linux/Mac e si desidera una mucca condiviso memoria, usa le forchette. Caricare dapprima tutti i dati nel thread principale e quindi avviare i thread di lavoro (fork) con la funzione generale mcparallel dal pacchetto parallel.

È possibile raccogliere i loro risultati con mccollect o con l'uso della memoria realmente condivisa utilizzando la libreria Rdsm, in questo modo:

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 

è possibile confermare, che il valore realmente viene aggiornato in backgruound, se si ritarda la scrittura:

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) 

da controllare per concurency ed evitare condizioni di gara utilizzano serrature:

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:

ho semplificato le dipendenze un po scambiando Rdsm::mgrmakevar in bigmemory::big.matrix.mgrmakevar chiama internamente big.matrix in ogni caso e non abbiamo bisogno di altro.