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
123
124
125
126
127
128
129
130
131
132
|
#pragma once
#include <stdlib.h>
#include <coro.h>
#include <ev++.h>
#include <vector>
#include <Exceptions.h>
#include <Printer.h>
namespace Balau {
namespace Events { class BaseEvent; };
class EAgain : public GeneralException {
public:
EAgain(Events::BaseEvent * evt) : GeneralException("Try Again"), m_evt(evt) { }
Events::BaseEvent * getEvent() { return m_evt; }
private:
Events::BaseEvent * m_evt;
};
class TaskMan;
class Task;
namespace Events {
class Callback {
protected:
virtual void gotEvent(BaseEvent *) = 0;
friend class BaseEvent;
};
class BaseEvent {
public:
BaseEvent() : m_cb(NULL), m_signal(false), m_task(NULL) { Printer::elog(E_TASK, "Creating event at %p", this); }
virtual ~BaseEvent() { if (m_cb) delete m_cb; }
bool gotSignal() { return m_signal; }
void doSignal();
void reset() { Assert(m_task != NULL); m_signal = false; gotOwner(m_task); }
Task * taskWaiting() { Assert(m_task); return m_task; }
void registerOwner(Task * task) { if (m_task == task) return; Assert(m_task == NULL); m_task = task; gotOwner(task); }
protected:
virtual void gotOwner(Task * task) { }
private:
Callback * m_cb;
bool m_signal;
Task * m_task;
};
class Timeout : public BaseEvent {
public:
Timeout(ev_tstamp tstamp);
void evt_cb(ev::timer & w, int revents);
void set(ev_tstamp tstamp);
private:
virtual void gotOwner(Task * task);
ev::timer m_evt;
};
class TaskEvent : public BaseEvent {
public:
TaskEvent(Task * taskWaited);
Task * taskWaited() { return m_taskWaited; }
private:
Task * m_taskWaited;
};
class Custom : public BaseEvent {
public:
void doSignal() { BaseEvent::doSignal(); ev_break(m_loop, EVBREAK_ALL); }
protected:
virtual void gotOwner(Task * task);
private:
struct ev_loop * m_loop;
};
};
class Task {
public:
enum Status {
STARTING,
RUNNING,
IDLE,
STOPPED,
FAULTED,
};
Task();
virtual ~Task();
virtual const char * getName() = 0;
Status getStatus() { return m_status; }
static Task * getCurrentTask();
static void yield(Events::BaseEvent * evt, bool interruptible = false) throw (GeneralException) {
Task * t = getCurrentTask();
t->waitFor(evt, true);
do {
t->yield(true);
} while ((!interruptible || !t->m_okayToEAgain) && !evt->gotSignal());
if (interruptible && t->m_okayToEAgain && !evt->gotSignal())
throw EAgain(evt);
}
TaskMan * getTaskMan() { return m_taskMan; }
struct ev_loop * getLoop();
protected:
void yield(bool override = false);
virtual void Do() = 0;
void waitFor(Events::BaseEvent * event, bool override = false);
bool setPreemptible(bool enable);
bool setOkayToEAgain(bool enable) {
bool oldValue = m_okayToEAgain;
m_okayToEAgain = enable;
return oldValue;
}
private:
size_t stackSize() { return 128 * 1024; }
void switchTo();
static void coroutine(void *);
void * m_stack;
coro_context m_ctx;
TaskMan * m_taskMan;
Status m_status;
void * m_tls;
friend class TaskMan;
friend class Events::TaskEvent;
typedef std::vector<Events::TaskEvent *> waitedByList_t;
waitedByList_t m_waitedBy;
struct ev_loop * m_loop;
bool m_okayToEAgain;
};
};
|
