blob: 15e0ac7e4b5a26ee0911f523ced0219a0abfae34 (
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
|
#pragma once
#include <Atomic.h>
#include <Exceptions.h>
#include <Local.h>
#include <Threads.h>
namespace Balau {
class AsyncManager;
class AsyncFinishWorker;
typedef void (*IdleReadyCallback_t)(void *);
class AsyncOperation {
protected:
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();
friend class AsyncManager;
friend class AsyncFinishWorker;
};
class AsyncFinishWorker : public Thread {
public:
AsyncFinishWorker(AsyncManager * async, Queue<AsyncOperation> * queue) : m_async(async), m_queue(queue) { }
virtual void * proc();
AsyncManager * m_async;
Queue<AsyncOperation> * m_queue;
bool m_stopping = false;
volatile bool m_stopped = false;
};
class AsyncManager : public Thread {
public:
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:
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);
Atomic::Increment(&m_numTLSes);
}
return tls;
}
Queue<AsyncOperation> m_queue;
Queue<AsyncOperation> m_finished;
Queue<TLS> m_TLSes;
volatile int m_numTLSes = 0;
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;
volatile bool m_ready = false;
volatile bool m_stopperPushed = false;
void incIdle() { Atomic::Increment(&m_numFinishersIdle); }
void decIdle() { Atomic::Decrement(&m_numFinishersIdle); }
friend class AsyncFinishWorker;
};
};
|