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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
|
#ifndef DSRT_CV_DISPATCHER_IMPL_T_CPP
#define DSRT_CV_DISPATCHER_IMPL_T_CPP
#include "DSRT_CV_Dispatcher_Impl_T.h"
#if !defined (__ACE_INLINE__)
//#include "DSRT_CV_Dispatcher_Impl_T.i"
#endif /* __ACE_INLINE__ */
namespace Kokyu
{
/*
//@@VS: This is somehow not being recognized by MSVC, which results
//in a link error. For now, the definition has been moved to the .h
//file. Needs further investigation.
template <class DSRT_Scheduler_Traits>
int Comparator_Adapter_Generator<DSRT_Scheduler_Traits>::MoreEligible::
operator ()(const DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits>& item1,
const DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits>& item2)
{
int rc = qos_comparator_ (item1->qos (), item2->qos ());
//more eligible
if (rc == 1)
return 1;
//if equally eligible, then resolve tie with the creation time of
//the item
if (rc == 0 && item1->insertion_time () < item2->insertion_time ())
return 1;
return 0;
}
*/
template <class DSRT_Scheduler_Traits>
DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
DSRT_CV_Dispatcher_Impl (ACE_Sched_Params::Policy sched_policy,
int sched_scope)
:DSRT_Dispatcher_Impl<DSRT_Scheduler_Traits> (sched_policy, sched_scope),
run_cond_ (run_cond_lock_)
{
}
template <class DSRT_Scheduler_Traits> int
DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
init_i (const DSRT_ConfigInfo&)
{
return 0;
}
template <class DSRT_Scheduler_Traits>
int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
schedule_i (Guid_t id, const DSRT_QoSDescriptor& qos)
{
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):schedule_i enter\n"));
#endif
DSRT_Dispatch_Item<DSRT_Scheduler_Traits>* item;
ACE_hthread_t thr_handle;
ACE_Thread::self (thr_handle);
if (ACE_OS::thr_setprio (thr_handle,
this->blocked_prio_,
this->sched_policy_) == -1)
{
ACE_ERROR_RETURN ((LM_ERROR,
ACE_TEXT ("%p\n"),
ACE_TEXT ("thr_setprio failed")), -1);
}
ACE_NEW_RETURN (item,
DSRT_Dispatch_Item<DSRT_Scheduler_Traits> (id, qos),
-1);
item->thread_handle (thr_handle);
ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
if (this->ready_queue_.insert (item) == -1)
return -1;
#ifdef KOKYU_DSRT_LOGGING
this->ready_queue_.dump ();
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):schedule_i after ready_q.insert\n"));
#endif
DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> item_var;
ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, cond_guard, run_cond_lock_, -1);
this->ready_queue_.most_eligible (item_var);
guard.release ();
ACE_hthread_t most_eligible_thr_handle = item_var->thread_handle ();
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):curr thr handle = %d\n",
thr_handle));
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):curr scheduled thr handle = %d\n",
this->curr_scheduled_thr_handle_));
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):most eligible thr handle = %d\n",
most_eligible_thr_handle));
#endif
if (this->curr_scheduled_thr_handle_ == thr_handle &&
most_eligible_thr_handle != thr_handle)
{
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):curr sched thr handle = thr_handle & "
"most eligible thr handle != curr thr handle. "
"about to do a broadcast on CV to wake up most eligible\n"));
#endif
this->curr_scheduled_thr_handle_ = most_eligible_thr_handle;
//wake up the most eligible thread
this->run_cond_.broadcast ();
}
//if the current thread is not the most eligible, then wait.
//if the current thread is most eligible, but some thread is
//scheduled currently, then wait.
while (most_eligible_thr_handle != thr_handle ||
(most_eligible_thr_handle == thr_handle &&
this->curr_scheduled_thr_handle_ != thr_handle &&
this->curr_scheduled_thr_handle_ != 0))
{
ACE_Time_Value tv (60,0);
tv += ACE_OS::gettimeofday ();
//wait a maximum of 1 min. This is an escape latch against lockups.
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T): About to block on cv\n"));
#endif
if (this->run_cond_.wait (&tv) == -1)
{
ACE_ERROR ((LM_ERROR,
"(%t|%T): run_cond.wait timed out -- Possible Lockup\n"));
}
this->ready_queue_.most_eligible (item_var);
most_eligible_thr_handle = item_var->thread_handle ();
}
this->curr_scheduled_guid_ = item_var->guid ();
this->curr_scheduled_thr_handle_ = most_eligible_thr_handle;
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T): %d is currently running\n",
thr_handle));
#endif
if (ACE_OS::thr_setprio (thr_handle,
this->active_prio_,
this->sched_policy_) == -1)
{
ACE_ERROR ((LM_ERROR,
ACE_TEXT ("%p\n"),
ACE_TEXT ("thr_setprio failed")));
}
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG,
"(%t|%T):schedule_i exit\n"));
#endif
return 0;
}
template <class DSRT_Scheduler_Traits>
int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
update_schedule_i (Guid_t guid, const DSRT_QoSDescriptor& qos)
{
return this->schedule_i (guid, qos);
}
template <class DSRT_Scheduler_Traits>
int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
update_schedule_i (Guid_t guid, Block_Flag_t flag)
{
ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block entered\n"));
#endif
DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> dispatch_item;
ACE_hthread_t thr_handle;
ACE_Thread::self (thr_handle);
int found = this->ready_queue_.find (guid, dispatch_item);
#ifdef KOKYU_DSRT_LOGGING
if (found == 0)
ACE_DEBUG ((LM_DEBUG, "(%t|%T): %d found in ready queue\n", thr_handle));
else
ACE_DEBUG ((LM_DEBUG, "(%t|%T): %d not found in ready queue\n",
thr_handle));
#endif
if (found == 0 && flag == BLOCK)
{
thr_handle = dispatch_item->thread_handle ();
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG, "(%t|%T): update schedule: %d found\n", thr_handle));
#endif
if (ACE_OS::thr_setprio (thr_handle,
this->blocked_prio_,
this->sched_policy_) == -1)
{
ACE_ERROR ((LM_ERROR,
ACE_TEXT ("%p\n"),
ACE_TEXT ("thr_setprio failed")));
}
int rc = this->cancel_schedule (guid);
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
#endif
return rc;
}
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
#endif
return -1;
}
template <class DSRT_Scheduler_Traits> int
DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
cancel_schedule_i (Guid_t guid)
{
ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
ACE_hthread_t thr_handle;
ACE_Thread::self (thr_handle);
#ifdef KOKYU_DSRT_LOGGING
ACE_DEBUG ((LM_DEBUG, "(%t|%T): about to remove guid\n"));
#endif
this->ready_queue_.remove (guid);
#ifdef KOKYU_DSRT_LOGGING
this->ready_queue_.dump ();
#endif
if (this->curr_scheduled_thr_handle_ == thr_handle)
{
this->curr_scheduled_guid_ = 0;
this->curr_scheduled_thr_handle_ = 0;
}
ACE_GUARD_RETURN (cond_lock_t,
mon, this->run_cond_lock_, 0);
this->run_cond_.broadcast ();
return 0;
}
template <class DSRT_Scheduler_Traits> int
DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
shutdown_i ()
{
this->shutdown_flagged_ = 1;
return 0;
}
}
#endif /* DSRT_CV_DISPATCHER_IMPL_T_CPP */
|
