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
|
// $Id$
#include "ace/POSIX_CB_Proactor.h"
#if defined (ACE_HAS_AIO_CALLS) && !defined (ACE_HAS_BROKEN_SIGEVENT_STRUCT)
#include "ace/Task_T.h"
#include "ace/Log_Msg.h"
#include "ace/Object_Manager.h"
#include "ace/OS_NS_sys_time.h"
ACE_RCSID (ace,
POSIX_CB_Proactor,
"$Id$")
ACE_BEGIN_VERSIONED_NAMESPACE_DECL
ACE_POSIX_CB_Proactor::ACE_POSIX_CB_Proactor (size_t max_aio_operations)
: ACE_POSIX_AIOCB_Proactor (max_aio_operations,
ACE_POSIX_Proactor::PROACTOR_CB),
sema_ ((unsigned int) 0)
{
// we should start pseudo-asynchronous accept task
// one per all future acceptors
this->get_asynch_pseudo_task ().start ();
}
// Destructor.
ACE_POSIX_CB_Proactor::~ACE_POSIX_CB_Proactor (void)
{
this->close ();
}
ACE_POSIX_Proactor::Proactor_Type
ACE_POSIX_CB_Proactor::get_impl_type (void)
{
return PROACTOR_CB;
}
void ACE_POSIX_CB_Proactor::aio_completion_func (sigval cb_data)
{
ACE_POSIX_CB_Proactor * impl = static_cast<ACE_POSIX_CB_Proactor *> (cb_data.sival_ptr);
if ( impl != 0 )
impl->notify_completion (0);
}
#if defined (ACE_HAS_SIG_C_FUNC)
extern "C" void
ACE_POSIX_CB_Proactor_aio_completion (sigval cb_data)
{
ACE_POSIX_CB_Proactor::aio_completion_func (cb_data);
}
#endif /* ACE_HAS_SIG_C_FUNC */
int
ACE_POSIX_CB_Proactor::handle_events (ACE_Time_Value &wait_time)
{
// Decrement <wait_time> with the amount of time spent in the method
ACE_Countdown_Time countdown (&wait_time);
return this->handle_events_i (wait_time.msec ());
}
int
ACE_POSIX_CB_Proactor::handle_events (void)
{
return this->handle_events_i (ACE_INFINITE);
}
int
ACE_POSIX_CB_Proactor::notify_completion (int sig_num)
{
ACE_UNUSED_ARG (sig_num);
return this->sema_.release();
}
ssize_t
ACE_POSIX_CB_Proactor::allocate_aio_slot (ACE_POSIX_Asynch_Result *result)
{
ssize_t slot = ACE_POSIX_AIOCB_Proactor::allocate_aio_slot (result);
if (slot == -1)
return -1;
// setup OS notification methods for this aio
// @@ TODO: This gets the completion method back to this proactor to
// find the completed aiocb. It would be so much better to not only get
// the proactor, but the aiocb as well.
#if defined(__sgi)
result->aio_sigevent.sigev_notify = SIGEV_CALLBACK;
result->aio_sigevent.sigev_func = aio_completion_func ;
#else
result->aio_sigevent.sigev_notify = SIGEV_THREAD;
# if defined (ACE_HAS_SIG_C_FUNC)
result->aio_sigevent.sigev_notify_function =
ACE_POSIX_CB_Proactor_aio_completion;
# else
result->aio_sigevent.sigev_notify_function = aio_completion_func;
# endif /* ACE_HAS_SIG_C_FUNC */
result->aio_sigevent.sigev_notify_attributes = 0;
#endif /* __sgi */
result->aio_sigevent.sigev_value.sival_ptr = this ;
return slot;
}
int
ACE_POSIX_CB_Proactor::handle_events_i (u_long milli_seconds)
{
int result_wait=0;
// Wait for the signals.
if (milli_seconds == ACE_INFINITE)
{
result_wait = this->sema_.acquire();
}
else
{
// Wait for <milli_seconds> amount of time.
ACE_Time_Value abs_time = ACE_OS::gettimeofday ()
+ ACE_Time_Value (0, milli_seconds * 1000);
result_wait = this->sema_.acquire(abs_time);
}
// Check for errors
// but let continue work in case of errors
// we should check "post_completed" queue
if (result_wait == -1)
{
int const lerror = errno;
if (lerror != ETIME && // timeout
lerror != EINTR ) // interrupted system call
ACE_ERROR ((LM_ERROR,
"%N:%l:(%P | %t)::%p\n",
"ACE_POSIX_CB_Proactor::handle_events:"
"semaphore acquire failed"
));
}
size_t index = 0; // start index to scan aiocb list
size_t count = this->aiocb_list_max_size_; // max number to iterate
int error_status = 0;
size_t return_status = 0;
int ret_aio = 0;
int ret_que = 0;
for (;; ret_aio++)
{
ACE_POSIX_Asynch_Result * asynch_result =
this->find_completed_aio (error_status,
return_status,
index,
count);
if (asynch_result == 0)
break;
// Call the application code.
this->application_specific_code (asynch_result,
return_status, // Bytes transferred.
0, // No completion key.
error_status); // Error
}
// process post_completed results
ret_que = this->process_result_queue ();
// Uncomment this if you want to test
// and research the behavior of you system
// ACE_DEBUG ((LM_DEBUG,
// "(%t) NumAIO=%d NumQueue=%d\n",
// ret_aio, ret_que));
return ret_aio + ret_que > 0 ? 1 : 0;
}
ACE_END_VERSIONED_NAMESPACE_DECL
#endif /* ACE_HAS_AIO_CALLS && !ACE_HAS_BROKEN_SIGEVENT_STRUCT */
|
