Creazione di un contatore che rimane sincronizzato tra i processi MPI

Question

Ho un bel po 'di esperienza nell'usare i metodi base di comunicazione e di gruppo MPI2, e faccio un po' di lavoro di simulazione parallelo imbarazzante usando MPI. Fino ad ora, ho strutturato il mio codice per avere un nodo di invio e un mucchio di nodi di lavoro. Il nodo di invio ha un elenco di file di parametri che verranno eseguiti con il simulatore. Seziona ciascun nodo di lavoro con un file di parametri. I nodi worker eseguono la simulazione, quindi richiedono un altro file di parametri, fornito dal nodo di invio. Una volta che tutti i file dei parametri sono stati eseguiti, il nodo di invio arresta ogni nodo di lavoro, prima di spegnersi.

I file dei parametri sono in genere denominati "Par_N.txt" dove N è il numero intero identificativo (ad es. N = 1-1000). Quindi stavo pensando, se potessi creare un contatore, e potessi avere questo contatore sincronizzato su tutti i miei nodi, potrei eliminare la necessità di avere un nodo di invio e rendere il sistema un po 'più semplice. Per quanto semplice possa sembrare, in pratica sospetto che sia un po 'più difficile, dato che dovrei assicurarmi che il contatore sia bloccato mentre viene cambiato, ecc. E pensavo che ci potesse essere un modo integrato per MPI di gestire questo. qualche idea? Sto pensando troppo a questo?

Answer 1

L'implementazione di un contatore condiviso non è banale, ma una volta eseguita e conservata in una libreria da qualche parte, è possibile eseguire un lotto con esso.

Nel libro Using MPI-2, che si dovrebbe avere a disposizione se si intende implementare questa roba, uno degli esempi (il codice è available online) è un contatore condiviso. Lo "non scalabile" dovrebbe funzionare bene con diverse dozzine di processi: il contatore è un array di 0..size-1 di interi, uno per grado, e quindi l'operazione di `get next work item # 'consiste di bloccando la finestra, leggendo il contributo di tutti gli altri al contatore (in questo caso, quanti elementi hanno preso), aggiornando il proprio (++), chiudendo la finestra e calcolando il totale. Questo è tutto fatto con operazioni passive unilaterali. (La migliore scala utilizza solo un albero piuttosto che un array 1-d).

Quindi l'uso sarebbe dire che il grado 0 ospita il contatore e tutti continuano a fare unità di lavoro e ad aggiornare il contatore per ottenere quello successivo fino a quando non c'è più lavoro; quindi aspetti ad una barriera o qualcosa e finalizzi.

Una volta che hai qualcosa di simile - utilizzando un valore condiviso per mettere a disposizione la prossima unità di lavoro - funzionante, puoi generalizzare a un approccio più sofisticato. Quindi, come suggerito da suzterpatt, tutti quelli che prendono "la loro parte" di unità di lavoro all'inizio funzionano alla grande, ma cosa fare se alcuni finiscono più velocemente di altri? La solita risposta ora è il furto del lavoro; ognuno mantiene la propria lista di unità di lavoro in una coda, e poi quando si esaurisce il lavoro, ruba unità di lavoro dall'altra parte di qualcuno, finché non rimane più lavoro.Questa è davvero la versione completamente distribuita di master-worker, dove non c'è più un singolo lavoro di partizionamento master. Una volta che hai funzionato un singolo contatore condiviso, puoi creare mutex da quelli e da questo puoi implementare la rimozione. Ma se il semplice contatore condiviso funziona abbastanza bene, potresti non aver bisogno di andare lì.

Aggiornamento: Ok, quindi ecco una hacky-tentativo di fare il contatore condiviso - la mia versione della semplice nel libro MPI-2: sembra funzionare, ma non avrebbe detto niente molto più forte di quello (Non ho giocato con questa roba per molto tempo). C'è una semplice implementazione del contatore (corrispondente alla versione non scalabile nel libro MPI-2) con due semplici test, uno corrispondente all'incirca al tuo caso di lavoro; ogni articolo aggiorna il contatore per ottenere un oggetto di lavoro, quindi esegue il "lavoro" (dorme per una quantità casuale di tempo). Alla fine di ogni test, viene stampata la struttura dei dati del contatore, che è il numero di incrementi che ogni grado ha eseguito.

#include <mpi.h> 
#include <stdlib.h> 
#include <stdio.h> 
#include <unistd.h> 

struct mpi_counter_t { 
    MPI_Win win; 
    int hostrank ; 
    int myval; 
    int *data; 
    int rank, size; 
}; 

struct mpi_counter_t *create_counter(int hostrank) { 
    struct mpi_counter_t *count; 

    count = (struct mpi_counter_t *)malloc(sizeof(struct mpi_counter_t)); 
    count->hostrank = hostrank; 
    MPI_Comm_rank(MPI_COMM_WORLD, &(count->rank)); 
    MPI_Comm_size(MPI_COMM_WORLD, &(count->size)); 

    if (count->rank == hostrank) { 
     MPI_Alloc_mem(count->size * sizeof(int), MPI_INFO_NULL, &(count->data)); 
     for (int i=0; i<count->size; i++) count->data[i] = 0; 
     MPI_Win_create(count->data, count->size * sizeof(int), sizeof(int), 
         MPI_INFO_NULL, MPI_COMM_WORLD, &(count->win)); 
    } else { 
     count->data = NULL; 
     MPI_Win_create(count->data, 0, 1, 
         MPI_INFO_NULL, MPI_COMM_WORLD, &(count->win)); 
    } 
    count -> myval = 0; 

    return count; 
} 

int increment_counter(struct mpi_counter_t *count, int increment) { 
    int *vals = (int *)malloc(count->size * sizeof(int)); 
    int val; 

    MPI_Win_lock(MPI_LOCK_EXCLUSIVE, count->hostrank, 0, count->win); 

    for (int i=0; i<count->size; i++) { 

     if (i == count->rank) { 
      MPI_Accumulate(&increment, 1, MPI_INT, 0, i, 1, MPI_INT, MPI_SUM, 
          count->win); 
     } else { 
      MPI_Get(&vals[i], 1, MPI_INT, 0, i, 1, MPI_INT, count->win); 
     } 
    } 

    MPI_Win_unlock(0, count->win); 
    count->myval += increment; 

    vals[count->rank] = count->myval; 
    val = 0; 
    for (int i=0; i<count->size; i++) 
     val += vals[i]; 

    free(vals); 
    return val; 
} 

void delete_counter(struct mpi_counter_t **count) { 
    if ((*count)->rank == (*count)->hostrank) { 
     MPI_Free_mem((*count)->data); 
    } 
    MPI_Win_free(&((*count)->win)); 
    free((*count)); 
    *count = NULL; 

    return; 
} 

void print_counter(struct mpi_counter_t *count) { 
    if (count->rank == count->hostrank) { 
     for (int i=0; i<count->size; i++) { 
      printf("%2d ", count->data[i]); 
     } 
     puts(""); 
    } 
} 

int test1() { 
    struct mpi_counter_t *c; 
    int rank; 
    int result; 

    c = create_counter(0); 

    MPI_Comm_rank(MPI_COMM_WORLD, &rank); 
    result = increment_counter(c, 1); 
    printf("%d got counter %d\n", rank, result); 

    MPI_Barrier(MPI_COMM_WORLD); 
    print_counter(c); 
    delete_counter(&c); 
} 


int test2() { 
    const int WORKITEMS=50; 

    struct mpi_counter_t *c; 
    int rank; 
    int result = 0; 

    c = create_counter(0); 

    MPI_Comm_rank(MPI_COMM_WORLD, &rank); 
    srandom(rank); 

    while (result < WORKITEMS) { 
     result = increment_counter(c, 1); 
     if (result <= WORKITEMS) { 
      printf("%d working on item %d...\n", rank, result); 
      sleep(random() % 10); 
     } else { 
      printf("%d done\n", rank); 
     } 
    } 

    MPI_Barrier(MPI_COMM_WORLD); 
    print_counter(c); 
    delete_counter(&c); 
} 

int main(int argc, char **argv) { 

    MPI_Init(&argc, &argv); 

    test1(); 
    test2(); 

    MPI_Finalize(); 
} 

Answer 2

Non riesco a pensare a nessun meccanismo integrato per risolvere questo problema, dovresti implementarlo manualmente. A giudicare dai commenti si desidera decentrare il programma, nel qual caso ogni processo (o almeno gruppi di processi) dovrebbe mantenere i propri valori del contatore e mantenerlo sincronizzato. Ciò potrebbe probabilmente essere fatto con un uso intelligente di send/receive non bloccanti, ma la semantica di quelli non è banale.

Invece, risolverei il problema di saturazione semplicemente immettendo più file contemporaneamente in processi di lavoro. Ciò ridurrebbe il traffico di rete e consentirà di mantenere la configurazione semplice di un singolo dispatcher.

Answer 3

Sembra che si stia utilizzando il nodo di invio per eseguire il bilanciamento dinamico del carico (assegnazione del lavoro ai processori quando diventano disponibili). Un contatore condiviso che non richiede l'arresto di tutti i processori non lo farà. Vorrei raccomandare di stare con quello che hai ora o fare ciò che suszterpatt suggerisce, inviare batch di file alla volta.

Answer 4

Non è chiaro se sia necessario passare attraverso i file in modo rigoroso o meno. In caso contrario, perché non fare in modo che ogni nodo i gestisca tutti i file in cui è N % total_workers == i - cioè la distribuzione ciclica del lavoro?