blob: 70cec656c1c50f6f56b28d7da8900ae6cca8abe4 (
plain)
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
|
#pragma once
#include <atomic>
#include <Exceptions.h>
#include <Local.h>
#include <Threads.h>
namespace Balau {
class AsyncManager;
class AsyncFinishWorker;
typedef void (*IdleReadyCallback_t)(void *);
class AsyncOperation {
protected:
AsyncOperation() { }
virtual void run() { }
virtual void finish() { }
virtual void done() { }
virtual bool needsMainQueue() { return true; }
virtual bool needsFinishWorker() { return false; }
virtual bool needsSynchronousCallback() { return true; }
protected:
virtual ~AsyncOperation() { }
private:
CQueue<AsyncOperation> * m_idleQueue = NULL;
IdleReadyCallback_t m_idleReadyCallback = NULL;
void * m_idleReadyParam = NULL;
void finalize();
AsyncOperation(const AsyncOperation &) = delete;
AsyncOperation & operator=(const AsyncOperation &) = delete;
friend class AsyncManager;
friend class AsyncFinishWorker;
};
class AsyncFinishWorker : public Thread {
public:
AsyncFinishWorker(AsyncManager * async, Queue<AsyncOperation> * queue) : m_async(async), m_queue(queue), m_stopped(false) { }
bool stopped() { return m_stopped; }
private:
AsyncFinishWorker(const AsyncFinishWorker &) = delete;
AsyncOperation & operator=(const AsyncFinishWorker &) = delete;
virtual void * proc();
AsyncManager * m_async;
Queue<AsyncOperation> * m_queue;
bool m_stopping = false;
std::atomic<bool> m_stopped;
};
class AsyncManager : public Thread {
public:
AsyncManager() : m_numTLSes(0), m_ready(false), m_stopperPushed(false) { }
void setFinishers(int minIdle, int maxIdle) {
AAssert(minIdle < maxIdle, "Minimum number of threads needs to be less than maximum number of threads.");
m_minIdle = minIdle;
m_maxIdle = maxIdle;
}
void setIdleReadyCallback(IdleReadyCallback_t idleReadyCallback, void * param);
void queueOp(AsyncOperation * op);
void idle();
bool isReady() { return m_ready; }
protected:
virtual void threadExit();
private:
AsyncManager(const AsyncManager &) = delete;
AsyncManager & operator=(const AsyncManager &) = delete;
void checkIdle();
void killOneFinisher();
void startOneFinisher();
void joinStoppedFinishers();
void stopAllWorkers();
virtual void * proc();
struct TLS {
CQueue<AsyncOperation> idleQueue;
IdleReadyCallback_t idleReadyCallback;
void * idleReadyParam;
};
TLS * getTLS() {
TLS * tls = (TLS *) m_tlsManager.getTLS();
if (!tls) {
tls = new TLS();
m_tlsManager.setTLS(tls);
m_TLSes.push(tls);
++m_numTLSes;
}
return tls;
}
Queue<AsyncOperation> m_queue;
Queue<AsyncOperation> m_finished;
Queue<TLS> m_TLSes;
std::atomic<int> m_numTLSes;
PThreadsTLSManager m_tlsManager;
std::list<AsyncFinishWorker *> m_workers;
int m_numFinishers = 0;
int m_numFinishersIdle = 0;
int m_minIdle = 1;
int m_maxIdle = 4;
bool m_stopping = false;
std::atomic<bool> m_ready;
std::atomic<bool> m_stopperPushed;
void incIdle() { m_numFinishersIdle++; }
void decIdle() { m_numFinishersIdle--; }
friend class AsyncFinishWorker;
};
};
|
