Réordonner des données calculées en parallèle

Version imprimable

24/02/2016, 17h59
orochimaru

Réordonner des données calculées en parallèle

Bonjour,

J'ai multithreadé une application de décodage vidéo et chaque frame est décodé de manière parallèle par n threads, le résultat étant envoyé dans une fifo utilisée par un thread consommateur.
Seulement le temps de décodage étant variable, les frames n'arrivent pas forcément dans le bon ordre, il me faut donc les réordonner pour que le résultat soit utilisable.

J'ai pensé utiliser une std::list dans laquelle je peut insérer au bon endroit la frame décodée seulement cela ne me semble pas être optimal, j'aimerai donc savoir si il existe une meilleure méthode pour faire cela.

Merci par avance.
24/02/2016, 19h00
Bousk

Salut,

pourquoi pas un simple array, en indiquant au décodeur à quelle place dans l'array il doit mettre son résultat ?
Le thread consommateur a juste a lire cet array, éventuellement il peut réinitialiser le tout pour laisser la place à la suite, les décodeurs étant alors en attente que la place se libère de leur côté s'ils ont fini de processer la frame suivante assignée mais que la place dans l'array est prise (c'est claire comme phrase ? je me perds moi-même :o ).
25/02/2016, 10h15
orochimaru

J'ai oublié de préciser que l’application doit tourner en temps réel et sur des très longues durées.
Du coup je ne voit pas comment utiliser un array qui grandirait linéairement avec l'arrivée des images décodées et exploserait très vite la mémoire (20Mo par frames ce qui donne 1.2Go/s à 60 fps quand même).
De plus je ne peux pas attendre que l'array soit rempli, le vider et le re-remplir avec les nouvelles frames car cela créerait une discontinuité dans le flux et le consommateur attend impérativement une frame toutes les 16.6ms.
25/02/2016, 10h21
ternel

Regarde du coté des "circular buffer", par exemple dans Boost.circular_buffer

Citation:

Envoyé par orochimaru

Du coup je ne voit pas comment utiliser un array qui grandirait linéairement avec l'arrivée des images décodées et exploserait très vite la mémoire (20Mo par frames ce qui donne 1.2Go/s à 60 fps quand même).
De plus je ne peux pas attendre que l'array soit rempli, le vider et le re-remplir avec les nouvelles frames car cela créerait une discontinuité dans le flux et le consommateur attend impérativement une frame toutes les 16.6ms.

J'ai parlé d'array, pas de vector.
Tu as 5 threads de décodage, tu utilises un array de 5 emplacements. Ton array sert de liste fifo.
Grossièrement ça part de
Code:

1 2 3 4 5 const unsigned int buffersize = N; array<buffersize, image*> buffer; void threaddecode(unsigned int index) { process lent de decodage; while(buffer[index] != nullptr) :: sleep(1); /*on attend que la place soit libre*/ buffer[index] = image décodée;} void mainthread() { for(unsigned int i = 0; !finished; i = (++i)%buffersize;) {utiliser buffer[i]; buffer[i] = nullptr;} }

25/02/2016, 14h17
stendhal666

Je pense que std::list est un bon choix.

Le désavantage des listes par rapport aux vecteurs c'est que l'accès à un membre n se fait en O(n). En revanche elles sont plus compétitives pour l'insertion et la suppression d'éléments.
Dans ton cas l'accès et la suppression se font toujours au premier élément, sinon il s'agit d'insertions pour lesquelles le désavantage de l'accès en O(n) est compensé par le fait que l'opération d'insertion elle-même est en O(1), comparée à un tableau où il faut déplacer (N-n) éléments, voire réallouer la mémoire (cas du vecteur ou du circular-buffer). De toute façon, avec un nombre de frames quoiqu'il arrive assez limité, l'accès en O(n) ne devrait pas être un facteur de lenteur vraiment gênant.
L'autre avantage que je vois par rapport à un array, dont la taille est fixe, c'est que tu pourras utiliser des stratégies d'utilisation de la mémoire (buffering, par exemple).
25/02/2016, 14h20
orochimaru

Bousk, ok je vois
Mais cela me dérange que les threads décodeurs soient en attente que la place soit libre, le durée du décodage à une forte amplitude et certaines frames sont lentes à décoder, du coup je préférerai une structure sous forme de queue qui me serve de buffer pour amortir cette variabilité.

leternel, j'ai regardé le circular_buffer de boost et effectivement cela semble répondre à la problématique. Seulement je n'ai pas accès à boost car je travaille sur de l'embarqué et de toute façon la structure n'est pas thread safe. Ou pourrais-je trouver une structure similaire mais thread safe et légère à intégrer ?

edit:
stendhal666, effectivement, std::list semble un bon candidat, il ne reste plus qu'a la rendre thread safe.

j'ai fini par utiliser une priority_queue que j'ai encapsulée pour qu'elle soit thread safe,
voici le résultat :

Code:

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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
template <typename T, class Compare>
class ThreadSafePriorityQueue
{
private:
    typedef std::priority_queue<T, std::deque<T>, Compare> priority_queue;
    priority_queue m_queue;
    size_t         m_max_size;
    size_t         m_min_pop_size;
    std::mutex     m_mutex;
    std::condition_variable m_cvPush;
    std::condition_variable m_cvPop;
 
public:
    ThreadSafePriorityQueue(const Compare &cmp, const size_t max_size, const size_t min_pop_size=1):
       m_queue(cmp), m_max_size(max_size), m_min_pop_size(min_pop_size)
    {
    }
 
    ~ThreadSafePriorityQueue()
    {
 
    }
 
    void push(const T &v) noexcept
    {
        std::unique_lock<std::mutex> lock( m_mutex );
 
        // Wait until there is space in the queue.
        while( m_queue.size() == m_max_size )
        {
            m_cvPush.wait( lock );
        }
 
        // Push to queue.
        m_queue.push(v);
 
        // Wake up one popping thread.
        if( m_queue.size() >= m_min_pop_size )
        {
            m_cvPop.notify_one();
        }
    }
 
    void push(T &&v) noexcept
    {
        std::unique_lock<std::mutex> lock( m_mutex );
 
        // Wait until there is space in the queue.
        while( m_queue.size() == m_max_size )
        {
            m_cvPush.wait( lock );
        }
 
        // Push to queue.
        m_queue.push(v);
 
        // Wake up one popping thread.
        if( m_queue.size() >= m_min_pop_size )
        {
            m_cvPop.notify_one();
        }
    }
 
    T pop() noexcept
    {
        std::unique_lock<std::mutex> lock( m_mutex );
 
        // If there is no item then we wait until there is one.
        while( m_queue.empty() )
        {
            m_cvPop.wait( lock );
        }
 
        // If we reach here then there is an item, get it.
        T ret = m_queue.top();
        m_queue.pop();
 
        // Wake up one pushing thread.
        m_cvPush.notify_one();
 
        return ret;
    }
 
    size_t size() noexcept
    {
        std::lock_guard<std::mutex> lock( m_mutex );
        return m_queue.size();
    }
 
    bool empty() noexcept
    {
        std::lock_guard<std::mutex> lock( m_mutex );
        return m_queue.empty();
    }
 
    void set_max_size(const size_t max_size) noexcept
    {
        std::lock_guard<std::mutex> lock( m_mutex );
        m_max_size = max_size;
    }
 
    void set_min_pop_size(const size_t min_pop_size) noexcept
    {
        std::lock_guard<std::mutex> lock( m_mutex );
        m_min_pop_size = min_pop_size;
    }
 
    void flush(const std::function<void(T&)> &func = [](){}) noexcept
    {
        std::lock_guard<std::mutex> lock( m_mutex );
        while( !m_queue.empty() )
        {
            T ret = m_queue.top();
            func( ret );
            m_queue.pop();
        }
 
        // Wake up one pushing thread.
        m_cvPush.notify_one();
    }
 
};

Cela fonctionne bien mais je ne suis pas sur de l'avoir fait dans les règles de l'art, des commentaires dessus seraient appréciés :-)

07/03/2016, 11h18
stendhal666

Comme ça ça me paraît bien!
Juste deux observations:
- tu n'implémentes pas la fonction membre emplace. Selon la façon dont est fait ton programme, tu peux peut-être gratter un peu de performance en utilisant cette sémantique de construction "sur place".
- pour la fonction flush, quelques réserves: il y a une différence de signature entre la fonction par défaut et le type de la std::function; tu copies l'objet retourné par top() dans ret -il vaudrait mieux au moins utiliser std::move; pour gratter un tout petit peu de performance en plus tu pourrais te passer de std::function (à voir selon l'utilisation mais il peut y avoir des copies / déréférencement) et prendre le callable en paramètre template, avec une spécialisation pour void (qui pourrait être l'argument template par défaut) qui ne ferait que pop() tant que !empty().