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 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305
|
//=============================================================================
/**
* @file post_completions.cpp
*
* $Id: post_completions.cpp 93639 2011-03-24 13:32:13Z johnnyw $
*
* This program demonstrates how to post fake completions to The
* Proactor. It also shows the how to specify the particular
* real-time signals to post completions. The Real-time signal
* based completion strategy is implemented with
* ACE_POSIX_SIG_PROACTOR.
* (So, it can be used only if both ACE_HAS_AIO_CALLS and
* ACE_HAS_POSIX_REALTIME_SIGNALS are defined.)
* Since it is faking results, you have to pay by knowing and
* using platform-specific implementation objects for Asynchronous
* Result classes.
* This example shows using an arbitrary result class for faking
* completions. You can also use the predefined Result classes for
* faking. The factory methods in the Proactor class create the
* Result objects.
* make
* ./post_completions
*
* @author Alexander Babu Arulanthu <alex@cs.wustl.edu>
*/
//=============================================================================
#include "ace/OS_NS_unistd.h"
#include "ace/OS_main.h"
#include "ace/Proactor.h"
#include "ace/Task.h"
#include "ace/WIN32_Proactor.h"
#include "ace/POSIX_Proactor.h"
#include "ace/Atomic_Op.h"
#include "ace/Thread_Mutex.h"
// Keep track of how many completions are still expected.
static ACE_Atomic_Op <ACE_SYNCH_MUTEX, size_t> Completions_To_Go;
#if defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)
// This only works on Win32 platforms and on Unix platforms supporting
// POSIX aio calls.
#if defined (ACE_HAS_AIO_CALLS)
#define RESULT_CLASS ACE_POSIX_Asynch_Result
#elif defined (ACE_HAS_WIN32_OVERLAPPED_IO)
#define RESULT_CLASS ACE_WIN32_Asynch_Result
#endif /* ACE_HAS_AIO_CALLS */
/**
* @class My_Result
*
* @brief Result Object that we will post to the Proactor.
*
*/
class My_Result : public RESULT_CLASS
{
public:
My_Result (ACE_Handler &handler,
const void *act,
int signal_number,
size_t sequence_number)
: RESULT_CLASS (handler.proxy (),
act,
ACE_INVALID_HANDLE,
0, // Offset
0, // OffsetHigh
0, // Priority
signal_number),
sequence_number_ (sequence_number)
{}
// Constructor.
virtual ~My_Result (void)
{}
// Destructor.
/**
* This is the method that will be called by the Proactor for
* dispatching the completion. This method generally calls one of
* the call back hood methods defined in the ACE_Handler
* class. But, we will just handle the completions here.
*/
void complete (size_t,
int success,
const void *completion_key,
u_long error)
{
this->success_ = success;
this->completion_key_ = completion_key;
this->error_ = error;
size_t to_go = --Completions_To_Go;
// Print the completion details.
ACE_DEBUG ((LM_DEBUG,
"(%t) Completion sequence number %d, success : %d, error : %d, signal_number : %d, %u more to go\n",
this->sequence_number_,
this->success_,
this->error_,
this->signal_number (),
to_go));
// Sleep for a while.
ACE_OS::sleep (4);
}
private:
/// Sequence number for the result object.
size_t sequence_number_;
};
/**
* @class My_Handler
*
* @brief Handler class for faked completions.
*
*/
class My_Handler : public ACE_Handler
{
public:
/// Constructor.
My_Handler (void) {}
/// Destructor.
virtual ~My_Handler (void) {}
};
/**
* @class My_Task:
*
* @brief Contains thread functions which execute event loops. Each
* thread waits for a different signal.
*/
class My_Task: public ACE_Task <ACE_NULL_SYNCH>
{
public:
/// Constructor.
My_Task (void) {}
/// Destructor.
virtual ~My_Task (void) {}
//FUZZ: disable check_for_lack_ACE_OS
int open (void *proactor)
{
//FUZZ: enable check_for_lack_ACE_OS
// Store the proactor.
this->proactor_ = (ACE_Proactor *) proactor;
// Activate the Task.
this->activate (THR_NEW_LWP, 5);
return 0;
}
int svc (void)
{
// Handle events for 13 seconds.
ACE_Time_Value run_time (13);
ACE_DEBUG ((LM_DEBUG, "(%t):Starting svc routine\n"));
if (this->proactor_->handle_events (run_time) == -1)
ACE_ERROR_RETURN ((LM_ERROR, "(%t):%p.\n", "Worker::svc"), -1);
ACE_DEBUG ((LM_DEBUG, "(%t) work complete\n"));
return 0;
}
private:
/// Proactor for this task.
ACE_Proactor *proactor_;
};
int
ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
ACE_UNUSED_ARG (argc);
ACE_UNUSED_ARG (argv);
ACE_DEBUG ((LM_DEBUG,
"(%P | %t):Test starts\n"));
// = Get two POSIX_SIG_Proactors, one with SIGRTMIN and one with
// SIGRTMAX.
ACE_Proactor proactor1;
// Proactor1. SIGRTMIN Proactor. (default).
// = Proactor2. SIGRTMAX Proactor.
#if defined (ACE_HAS_AIO_CALLS) && defined (ACE_HAS_POSIX_REALTIME_SIGNALS)
ACE_DEBUG ((LM_DEBUG, "Using ACE_POSIX_SIG_Proactor\n"));
sigset_t signal_set;
// Signal set that we want to mask.
// Clear the signal set.
if (ACE_OS::sigemptyset (&signal_set) == -1)
ACE_ERROR_RETURN ((LM_ERROR,
"Error:%p\n",
"sigemptyset failed"),
1);
// Add the SIGRTMAX to the signal set.
if (ACE_OS::sigaddset (&signal_set, ACE_SIGRTMAX) == -1)
ACE_ERROR_RETURN ((LM_ERROR,
"Error:%p\n",
"sigaddset failed"),
1);
// Make the POSIX Proactor.
ACE_POSIX_SIG_Proactor posix_proactor (signal_set);
// Get the Proactor interface out of it.
ACE_Proactor proactor2 (&posix_proactor);
#else /* ACE_HAS_AIO_CALLS && ACE_HAS_POSIX_REALTIME_SIGNALS */
ACE_Proactor proactor2;
#endif /* ACE_HAS_AIO_CALLS && ACE_HAS_POSIX_REALTIME_SIGNALS */
// = Create Tasks. One pool of threads to handle completions on
// SIGRTMIN and the other one to handle completions on SIGRTMAX.
My_Task task1, task2;
task1.open (&proactor1);
task2.open (&proactor2);
// Handler for completions.
My_Handler handler;
// = Create a few MyResult objects and post them to Proactor.
const size_t NrCompletions (10);
My_Result *result_objects [NrCompletions];
int signal_number = ACE_SIGRTMAX;
size_t ri = 0;
Completions_To_Go = NrCompletions;
// Creation.
for (ri = 0; ri < NrCompletions; ri++)
{
// Use RTMIN and RTMAX proactor alternatively, to post
// completions.
if (ri % 2)
signal_number = ACE_SIGRTMIN;
else
signal_number = ACE_SIGRTMAX;
// Create the result.
ACE_NEW_RETURN (result_objects [ri],
My_Result (handler,
0,
signal_number,
ri),
1);
}
ACE_OS::sleep(5);
// Post all the result objects.
ACE_Proactor *proactor;
for (ri = 0; ri < NrCompletions; ri++)
{
// Use RTMIN and RTMAX Proactor alternatively, to post
// completions.
if (ri % 2)
proactor = &proactor1;
else
proactor = &proactor2;
if (result_objects [ri]->post_completion (proactor->implementation ())
== -1)
ACE_ERROR_RETURN ((LM_ERROR,
"Test failed\n"),
1);
}
ACE_Thread_Manager::instance ()->wait ();
int status = 0;
size_t to_go = Completions_To_Go.value ();
if (size_t (0) != to_go)
{
ACE_ERROR ((LM_ERROR,
"Fail! Expected all completions to finish but %u to go\n",
to_go));
status = 1;
}
ACE_DEBUG ((LM_DEBUG,
"(%P | %t):Test ends\n"));
return status;
}
#else /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
int
ACE_TMAIN (int, ACE_TCHAR *[])
{
ACE_DEBUG ((LM_DEBUG,
"This example cannot work with AIOCB_Proactor.\n"));
return 1;
}
#endif /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
|