Recherche de conseils pour écriture du résultat de plusieurs workitems dans un fichier

**_skip** · 12/10/2009, 12h49

Bonjour,
Voilà, je dispose d'un vecteur d'éléments (des items avec des listes d'attributs).

Un score de 1-100 sous forme de short est calculé pour chaque couple (x,y) de ce vecteur. de façon à obtenir
(element_1, element_2) = 24;
(element_1, element_3) = 50;
(element_1, element_4) = 10;
...

Puisque f( x ,y) est égal à f( y, x), je compare le premier élément à tous, le second à tous sauf le premier, le 3e à tous sauf les deux premiers... Bref chaque element d'index x du vecteur est comparé à tous les éléments dont l'index est compris entre x+1 et n. Afin d'éviter le recalcul (en gros si le résultat était sous forme de matrice, on ne calculerait que la diagonale haut/droite).

Pour accélérer les calculs, j'ai créé un pool consommant des workitems. Un workItem contient le vecteur d'élément à comparer, et simplement l'index de départ plus une ou deux ressources partagées.

Mon code ressemble à ceci :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
void computeAllAttributeSim( const QVector<itemAttributes*> itemsWithAttributes, const QVector<float> weights )
{
    QThreadPool pool;
    pool.setMaxThreadCount(2);
 
    for( int i = 0; i <  itemsWithAttributes.size(); i++ )
    {
        //chaque workitem part de i, et calcule (i, i+1), (i,i+2),... (i, i+n) 
        workItem *item = new workItem( itemsWithAttributes,  weights, i  );
        pool.start(item);
    }
 
    pool.waitForDone();
}

Pour le moment cela marche très bien, mais vu la quantité d'élément que peut contenir le vecteur source (50'000 items). J'aurai besoin d'écrire les résultats des couples dans un fichier au fur et à mesure afin de ne pas avoir à tout stocker en mémoire car le nombre de valeurs est joliment de l'ordre de grandeur (n^2) / 2.

Donc lorsqu'un workitem est fini, il me faudrait pouvoir écrire les valeurs qu'il a calculées dans un fichier, le problème c'est que l'ordre dans lequel j'écris doit être identique à celui du vecteur (c'est une contrainte du programme qui va lire mon fichier généré).
Donc si 4 workitems tournent en même temps, je dois pouvoir les écrire dans l'ordre de création quand ils ont fini, même si le 3e finit avant le premier.

En fait je sèche sur la façon dont je pourrai gérer cette écriture de résultat efficacement sans massacrer les performances à grand coups de mutex dans tous les sens...

Si quelqu'un a une piste, c'est bienvenu...

**Mac LAK** · 12/10/2009, 14h18

Essaie d'aller voir cette discussion, je parle d'un problème similaire (post #5 notamment).

**_skip** · 12/10/2009, 15h07

J'ai déjà une file d'exécution, juste que j'ai besoin de stocker les résultats dans une 2e structure managée par un autre thread (celui qui écrira les données).

Mon problème c'est que :
- Je ne peux pas garantir que l'ordre de complétion soit l'odre de création des tâches (comme tu l'as précisé).
- Et surtout je dois empêcher les résultats d'arriver plus vite qu'ils ne peuvent être écrit sur disque (ce qui reviendrait au même que tout conserver en mémoire).

En ce qui concerne ce 2e point, je dois trouver un moyen de bloquer un thread au moment ou il ajoute un résultat à la structure si celle-ci est déjà trop grande, mais je dois me gaffer à éviter que le thread qui produit le résultat sur lequel j'attend reste coincé dehors.

**Mac LAK** · 12/10/2009, 15h32

T'as pas bien lu cette partie du post #5 :

Donc, si les paquets "source" doivent être traités de façon à ce que les paquets "résultat" aient le même ordre, il faut parfois utiliser un étage de plus à la sortie des threads pour rétablir l'ordre.
Pour ma part, j'utilise le plus souvent un numéro séquentiel unique généré lors de la récupération d'un paquet "source", plus une FIFO unitaire de sortie pour chaque thread. Ensuite, un sérialiseur connecte toutes les FIFO des threads, attends "le" bon numéro, et sérialise dans le bon ordre vers la FIFO finale.

En l'occurrence, il te faut un étage de plus dans ta sérialisation de sortie, afin de réordonnancer les paquets.

Comme tu as des contraintes de place, ce que je te conseille, c'est de swapper les données en attente d'ordonnancement sur le disque, dans des fichiers temporaires, et d'attendre le "bon" paquet afin de débloquer un thread qui ira recoller les données aux bons endroits depuis ces fichiers temporaires.

**_skip** · 12/10/2009, 16h26

J'ai lu ce passage...

J'essaie actuellement d'implémenter la solution suivante :

- la sortie est limitée à 10 résultats
- le thread est mis en attente si les 10 slots de résultats disponibles sont occupés, sauf si ce premier correspond au no qui est attendu. Lors de l'ajout le thread disque est notifié.
- un thread disque consomme les résultats et notifie le thread éventuellement bloqué en ajout.

Bref, c'est le bordel et je suis pas sur que dans les situations ou il y a plusieurs producteurs je n'ai jamais de deadlocks. J'y réfléchis encore

**Mac LAK** · 12/10/2009, 16h38

Il faut maintenir en RAM une FIFO qui possède, pour chaque élément, le numéro unique de sérialisation / réordonnancement, ainsi que le nom du fichier de swap (forcément unique, bien sûr, cf. GetTempFileName).
Ensuite, les deadlocks ne peuvent plus se produire ainsi : d'une part, ta RAM est libérée via le swap, et d'autre part, tu garantis un accès forcément exclusif à chaque fichier de swap :

Quand il est créé, il n'existe pas encore dans la FIFO de sérialisation.
Une fois dans cette FIFO, le thread créateur ne gère plus cette donnée (qui est "terminée"), et n'écrasera donc pas non plus le fichier ni les données en FIFO.

Le seul et unique risque, c'est la saturation si jamais une donnée attendue n'arrive jamais (thread planté), mais c'est normalement une situation que tu devrais savoir gérer au sein de ton application.

**_skip** · 12/10/2009, 18h54

Créer 50'000 fichiers temporaires et ensuite les rassembler c'est jouable ça?
Merci

**Mac LAK** · 13/10/2009, 09h45

Tu n'en auras pas 50.000, mais (2 x N) (N étant le nombre de threads de calcul) simultanément au maximum, je pense... Pour le reste, si les données sont envoyées dans l'ordre, tu as de fortes chances (mais pas de garantie hélas) que les threads se termineront à peu près dans l'ordre aussi. Au pire, décales l'envoi des données brutes aux threads (disons de l'ordre de 20% du temps de calcul) de façon à mieux séquencer tout ça, avec ou sans augmentation de priorité du thread ayant "le" prochain bloc à traiter.

EDIT : C'est pas clair, je me rends compte... Si le bloc terminé est le bon, inutile de l'envoyer sur un fichier temporaire. Il part alors directement dans le fichier final.

**_skip** · 13/10/2009, 11h41

En m'inspirant de ce que tu as dit, j'ai écrit ça pour ma file de sortie :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
void SyncResultWriter::addResult( int pos, QList<QPair<int, short> > results )
{
 
    QIODevice* tempFile;
 
    if( results.size() > 10000)
    {
         tempFile = new QTemporaryFile();
    }
    else
    {
         tempFile = new QBuffer();
    }
 
 
 
    if( !tempFile->open( QIODevice::ReadWrite ) )
        return;
 
    //qDebug() << tempFile->fileName();
 
    QTextStream out(tempFile);
    out.setCodec( "UTF-8" );
 
    for(int i = 0; i < results.size(); i++)
    {
        out << "(" << results[i].first << "," << results[i].second << ")\t";
    }
 
    out.flush();
    tempFile->close();
 
    mapMutex.lock();
    map.insert(pos, tempFile);
    mapMutex.unlock();
 
    elementAvailable.wakeAll();
}

C'est encore un peu brouillon, mais l'idée est la suivante :

En gros cette méthode est appelée par un thread qui a fini sa rangée de calcul. Si le nombre de résultat est plus grand que 10000 (sachant que je ne retiens que les pairs significatives), il sera écrit provisoirement dans un fichier, sinon ce sera dans un buffer en mémoire.

Après avoir écrit son résultat, le thread ajoute une entrée dans une map ayant pour clef le no de sérialisation (fourni en paramètre par "pos"). Il notifie au besoin qu'un nouvel élément est disponible.

Le consommateur qui écrit le fichier final, il ressemble à ceci :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
void SyncResultWriter::run()
{
    QFile file( finalFileName );
    file.remove();
 
    if(!file.open( QIODevice::ReadWrite))
        return;
 
    QTextStream out(&file);
    out.setCodec( "UTF-8" );
 
    //write file header
    out << nbItems << '\n';
 
    while( current < nbItems )
    {
        //get the next chunk if available
        QIODevice* nextFile = tryFetch();
 
        if( nextFile == NULL)
        {
            //no chunk available, wait for a while
            if( !elementAvailable.wait( &elementAvailableMutex, 1000 ))
                qDebug() << "timeout";
        }
        else
        {
            //adds chunk to the given result file
            mergeFile( nextFile, out);
        }
    }
 
    file.close();
}

Il tente de fetcher un nouvel élément depuis la map, s'il trouve l'élément qu il cherche il le sort de la map, le merge, puis le delete. Sinon il se met en attente un moment.

Qu'en penses-tu sur le fond?

**Mac LAK** · 13/10/2009, 12h11

Sur le fond, ça me parait bien, même si je ne vois pas bien où tu gères l'ordonnancement sur la deuxième partie...

**_skip** · 13/10/2009, 12h20

dans cette méthode qui essaie de me sortir l'élément suivant dont le no est enregistré dans la variable "current":

Les résultats en attente d'être traités sont stockés dans une map membre :
map<int, QIODevice*>

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
QIODevice* SyncResultWriter::tryFetch()
{
 
    QIODevice* nextFile = NULL;
 
    mapMutex.lock();
    QMap<int, QIODevice*>::iterator iter = map.find(current);
 
    if ( iter != map.end())
    {
        nextFile = iter.value();
        map.erase(iter);
        ++current;
    }
 
    mapMutex.unlock();
 
    return nextFile;
}

**Mac LAK** · 13/10/2009, 12h28

Dans ce cas, c'est le moment du "choix" entre buffer interne et fichier temporaire que je ne vois pas... A moins que tu ne fasses tout passer par le polymorphisme sur "tempfile" ?

(Pour info, je ne connais pas l'API QT, donc j'ignore les propriétés de tous les objets QT que tu as utilisé...)

**_skip** · 13/10/2009, 13h11

En effet, QIODevice est une classe de base dont les descendants sont QFile, QTemporaryFile et QBuffer notamment.

QFile est un fichier disque physique permanent.
QTemporaryFile est un fichier disque temporaire.
QBuffer est une structure en mémoire qui encapsule un tableau de bytes

Tout ceci peut être manipuler par la même interface commune définie dans QIODevice.

En fait par polymorphisme, je choisis d'instancier un QTemporaryFile ou alors un QBuffer en sortie de thread selon la quantité de valeurs significatives dans le résultat.