1 /*------------------------------------------------------------------------
2   Junction: Concurrent data structures in C++
3   Copyright (c) 2016 Jeff Preshing
4   Distributed under the Simplified BSD License.
5   Original location: https://github.com/preshing/junction
6   This software is distributed WITHOUT ANY WARRANTY; without even the
7   implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
8   See the LICENSE file for more information.
9 ------------------------------------------------------------------------*/
10 
11 #ifndef SIMPLEJOBCOORDINATOR_H
12 #define SIMPLEJOBCOORDINATOR_H
13 
14 #include <QMutex>
15 #include <QWaitCondition>
16 #include <QMutexLocker>
17 
18 #include "kis_assert.h"
19 #include "atomic.h"
20 
21 #define SANITY_CHECK
22 
23 class SimpleJobCoordinator
24 {
25 public:
26     struct Job {
~JobJob27         virtual ~Job()
28         {
29         }
30 
31         virtual void run() = 0;
32     };
33 
34 private:
35     Atomic<quint64> m_job;
36     QMutex mutex;
37     QWaitCondition condVar;
38 
39 public:
SimpleJobCoordinator()40     SimpleJobCoordinator() : m_job(quint64(NULL))
41     {
42     }
43 
loadConsume()44     Job* loadConsume() const
45     {
46         return (Job*) m_job.load(Consume);
47     }
48 
storeRelease(Job * job)49     void storeRelease(Job* job)
50     {
51         {
52             QMutexLocker guard(&mutex);
53             m_job.store(quint64(job), Release);
54         }
55 
56         condVar.wakeAll();
57     }
58 
participate()59     void participate()
60     {
61         quint64 prevJob = quint64(NULL);
62 
63         for (;;) {
64             quint64 job = m_job.load(Consume);
65             if (job == prevJob) {
66                 QMutexLocker guard(&mutex);
67 
68                 for (;;) {
69                     job = m_job.loadNonatomic(); // No concurrent writes inside lock
70                     if (job != prevJob) {
71                         break;
72                     }
73 
74                     condVar.wait(&mutex);
75                 }
76             }
77 
78             if (job == 1) {
79                 return;
80             }
81 
82             reinterpret_cast<Job*>(job)->run();
83             prevJob = job;
84         }
85     }
86 
runOne(Job * job)87     void runOne(Job* job)
88     {
89 #ifdef SANITY_CHECK
90         KIS_ASSERT_RECOVER_NOOP(job != (Job*) m_job.load(Relaxed));
91 #endif // SANITY_CHECK
92         storeRelease(job);
93         job->run();
94     }
95 
end()96     void end()
97     {
98         {
99             QMutexLocker guard(&mutex);
100             m_job.store(1, Release);
101         }
102 
103         condVar.wakeAll();
104     }
105 };
106 
107 #endif // SIMPLEJOBCOORDINATOR_H
108