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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
#include "pyi_timing_thread.h"
#include <Python.h>
#include <time.h>
#include <float.h>
#include "pyi_floatclock.h"
static volatile double current_time = 0.0;
static PyThread_type_lock subscriber_lock = NULL;
static PyThread_type_lock update_lock = NULL;
static int thread_should_exit = 0;
static int thread_alive = 0;
// Structure to hold subscriptions
typedef struct Subscription {
double interval;
int id;
} Subscription;
#define MAX_SUBSCRIBERS 1000
static Subscription subscribers[MAX_SUBSCRIBERS];
static int subscriber_count = 0;
static double get_interval(double max_interval) {
double min_interval = max_interval;
for (int i = 0; i < subscriber_count; i++) {
if (subscribers[i].interval < min_interval) {
min_interval = subscribers[i].interval;
}
}
return min_interval;
}
static void timing_thread(void* args) {
while (!thread_should_exit) {
double interval = get_interval(1.0);
// sleep for the interval, or until we're woken up by a change
PyLockStatus status = PyThread_acquire_lock_timed(
update_lock,
(PY_TIMEOUT_T)(interval * 1e6),
0
);
if (status == PY_LOCK_ACQUIRED) {
// rather than finishing the wait, another thread signaled a
// change by releasing the lock. The lock was just for the sake of
// the wakeup, so let's release it again.
PyThread_release_lock(update_lock);
}
current_time = pyi_floatclock(PYI_FLOATCLOCK_DEFAULT);
}
}
int pyi_timing_thread_subscribe(double desiredInterval) {
if (subscriber_lock == NULL) {
subscriber_lock = PyThread_allocate_lock();
}
if (update_lock == NULL) {
update_lock = PyThread_allocate_lock();
}
PyThread_acquire_lock(subscriber_lock, WAIT_LOCK);
if (!thread_alive) {
PyThread_acquire_lock(update_lock, WAIT_LOCK); // Initially hold the lock
thread_should_exit = 0;
PyThread_start_new_thread(timing_thread, NULL);
thread_alive = 1;
// initialise the current_time in case it's read immediately
current_time = pyi_floatclock(PYI_FLOATCLOCK_DEFAULT);
}
int new_id = 0;
// find an unused ID
for (; new_id < MAX_SUBSCRIBERS; new_id++) {
int already_exists = 0;
for (int i = 0; i < subscriber_count; i++) {
if (subscribers[i].id == new_id) {
already_exists = 1;
break;
}
}
if (!already_exists) {
break;
}
}
if (new_id == MAX_SUBSCRIBERS) {
// Too many subscribers
PyThread_release_lock(subscriber_lock);
return PYI_TIMING_THREAD_TOO_MANY_SUBSCRIBERS;
}
int index = subscriber_count;
subscribers[index].id = new_id;
subscribers[index].interval = desiredInterval;
subscriber_count++;
// signal a possible change in the interval
PyThread_release_lock(update_lock);
PyThread_acquire_lock(update_lock, WAIT_LOCK);
PyThread_release_lock(subscriber_lock);
return new_id;
}
int pyi_timing_thread_unsubscribe(int id) {
PyThread_acquire_lock(subscriber_lock, WAIT_LOCK);
int removals = 0;
for (int i = 0; i < subscriber_count; i++) {
if (subscribers[i].id == id) {
// Removal: overwrite this one with with the last element and decrement count.
subscribers[i] = subscribers[subscriber_count-1];
subscriber_count--;
removals++;
break;
}
}
// if the last subscriber was removed, stop the thread
if (subscriber_count == 0) {
thread_should_exit = 1;
PyThread_release_lock(update_lock);
thread_alive = 0;
}
PyThread_release_lock(subscriber_lock);
if (removals == 0) {
return PYI_TIMING_THREAD_NOT_SUBSCRIBED;
} else {
return 0;
}
}
double pyi_timing_thread_get_time(void) {
return current_time;
}
double pyi_timing_thread_get_interval(void) {
if (thread_alive) {
return get_interval(DBL_MAX);
} else {
return -1.0;
}
}
|
