summaryrefslogtreecommitdiff
path: root/ACE/examples/Reactor/Misc/notification.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'ACE/examples/Reactor/Misc/notification.cpp')
-rw-r--r--ACE/examples/Reactor/Misc/notification.cpp385
1 files changed, 385 insertions, 0 deletions
diff --git a/ACE/examples/Reactor/Misc/notification.cpp b/ACE/examples/Reactor/Misc/notification.cpp
new file mode 100644
index 00000000000..a04663b28ad
--- /dev/null
+++ b/ACE/examples/Reactor/Misc/notification.cpp
@@ -0,0 +1,385 @@
+// $Id$
+
+#include "ace/OS_NS_unistd.h"
+#include "ace/Service_Config.h"
+#include "ace/Reactor.h"
+#include "ace/Thread_Manager.h"
+#include "ace/Thread.h"
+#include "ace/Signal.h"
+
+ACE_RCSID(Misc, notification, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+#if defined (CHORUS)
+// Chorus does not have signal, so we'll stop after a number of rounds.
+#define MAX_ITERATIONS 3
+#else
+#define MAX_ITERATIONS 10000
+#endif /* CHORUS */
+
+class Thread_Handler : public ACE_Event_Handler
+{
+ // = TITLE
+ // Illustrate how the ACE_Reactor's thread-safe event notification
+ // mechanism works.
+ //
+ // = DESCRIPTION
+ // Handle timeouts in the main thread via the ACE_Reactor and I/O
+ // events in a separate thread. Just before the separate I/O
+ // thread exits it notifies the ACE_Reactor in the main thread
+ // using the ACE_Reactor's notification mechanism.
+public:
+ Thread_Handler (long delay,
+ long interval,
+ size_t n_threads,
+ size_t max_iterations);
+ // Constructor.
+
+ Thread_Handler (size_t id,
+ size_t max_iterations);
+
+ ~Thread_Handler (void);
+ // Destructor.
+
+ virtual int handle_signal (int signum,
+ siginfo_t * = 0,
+ ucontext_t * = 0);
+ // Handle signals.
+
+ virtual int handle_exception (ACE_HANDLE);
+ // Print data from main thread.
+
+ virtual int handle_output (ACE_HANDLE);
+ // Print data from main thread.
+
+ virtual int handle_timeout (const ACE_Time_Value &,
+ const void *);
+ // Handle timeout events in the main thread.
+
+ virtual int handle_input (ACE_HANDLE);
+ // General notification messages to the Reactor.
+
+ virtual int notify (ACE_Time_Value *tv = 0);
+ // Perform notifications.
+
+ virtual int svc (void);
+ // Handle I/O events in a separate threads.
+
+private:
+ static void *svc_run (void *);
+ // Glues C++ to C thread library functions.
+
+ size_t id_;
+ // ID passed in by Thread_Handler constructor.
+
+ size_t iterations_;
+
+ static sig_atomic_t shutdown_;
+ // Shutting down.
+
+ // = Timing variables.
+ // Delay factor for timer-driven I/O.
+ static ACE_Time_Value delay_;
+
+ // Interval factor for Event_Handler timer.
+ static ACE_Time_Value interval_;
+};
+
+// Shutdown flag.
+sig_atomic_t Thread_Handler::shutdown_ = 0;
+
+// Delay factor for timer-driven I/O.
+ACE_Time_Value Thread_Handler::delay_;
+
+// Interval factor for Event_Handler timer.
+ACE_Time_Value Thread_Handler::interval_;
+
+Thread_Handler::Thread_Handler (size_t id,
+ size_t max_iterations)
+ : id_ (id),
+ iterations_ (max_iterations)
+{
+}
+
+Thread_Handler::~Thread_Handler (void)
+{
+ // Cleanup resources so that we don't crash and burn when shutdown.
+ ACE_Event_Handler::remove_stdin_handler (ACE_Reactor::instance (),
+ ACE_Thread_Manager::instance ());
+ ACE_Reactor::instance ()->cancel_timer (this);
+}
+
+Thread_Handler::Thread_Handler (
+ long delay,
+ long interval,
+ size_t n_threads,
+ size_t max_iterations)
+ : iterations_ (max_iterations)
+{
+ ACE_Sig_Set sig_set;
+
+ sig_set.sig_add (SIGQUIT);
+ sig_set.sig_add (SIGINT);
+
+ delay_.set (delay);
+ interval_.set (interval);
+ this->id_ = 0;
+
+ if (ACE_Event_Handler::register_stdin_handler (this,
+ ACE_Reactor::instance (),
+ ACE_Thread_Manager::instance ()) == -1)
+ ACE_ERROR ((LM_ERROR,
+ "%p\n",
+ "register_stdin_handler"));
+ else if (ACE_Reactor::instance ()->register_handler (sig_set,
+ this) == -1)
+ ACE_ERROR ((LM_ERROR,
+ "(%t) %p\n",
+ "register_handler"));
+ else if (ACE_Reactor::instance ()->schedule_timer
+ (this,
+ 0,
+ Thread_Handler::delay_,
+ Thread_Handler::interval_) == -1)
+ ACE_ERROR ((LM_ERROR,
+ "(%t) %p\n",
+ "schedule_timer"));
+
+ // Set up this thread's signal mask to block all the signal in the
+ // <sig_set>, which is inherited by the threads it spawns.
+ ACE_Sig_Guard guard (&sig_set);
+
+ // Create N new threads of control Thread_Handlers.
+
+ for (size_t i = 0; i < n_threads; i++)
+ {
+ Thread_Handler *th;
+
+ ACE_NEW (th,
+ Thread_Handler (i + 1,
+ this->iterations_));
+
+ if (ACE_Thread::spawn (reinterpret_cast<ACE_THR_FUNC> (&Thread_Handler::svc_run),
+ reinterpret_cast<void *> (th),
+ THR_NEW_LWP | THR_DETACHED) != 0)
+ ACE_ERROR ((LM_ERROR,
+ "%p\n",
+ "ACE_Thread::spawn"));
+ }
+
+ // The destructor of <guard> unblocks the signal set so that only
+ // this thread receives them!
+}
+
+int
+Thread_Handler::notify (ACE_Time_Value *timeout)
+{
+ // Just do something to test the ACE_Reactor's multi-thread
+ // capabilities...
+
+ if (ACE_Reactor::instance ()->notify
+ (this,
+ ACE_Event_Handler::EXCEPT_MASK,
+ timeout) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR,
+ "(%t) %p\n",
+ "notification::notify:exception"),
+ -1);
+ else if (ACE_Reactor::instance ()->notify
+ (this,
+ ACE_Event_Handler::WRITE_MASK,
+ timeout) == -1)
+ ACE_ERROR_RETURN ((LM_ERROR,
+ "(%t) %p\n",
+ "notification::notify:write"),
+ -1);
+ return 0;
+}
+
+// Test stdin handling that uses <select> to demultiplex HANDLEs.
+// Input is only handled by the main thread.
+
+int
+Thread_Handler::handle_input (ACE_HANDLE handle)
+{
+ char buf[BUFSIZ];
+ ssize_t n = ACE_OS::read (handle, buf, sizeof buf);
+
+ if (n > 0)
+ {
+ ACE_DEBUG ((LM_DEBUG,
+ "input to (%t) %*s",
+ n,
+ buf));
+
+ ACE_DEBUG ((LM_DEBUG,
+ "%d more input to kill\n",
+ this->iterations_));
+
+ // Only wait up to 10 milliseconds to notify the Reactor.
+ ACE_Time_Value timeout (0,
+ 10 * 1000);
+
+ if (this->notify (&timeout) == -1)
+ ACE_ERROR ((LM_DEBUG,
+ "(%t), %p\n",
+ "notification::handle_input:notify"));
+ return 0;
+ }
+ else
+ return -1;
+}
+
+// Perform a task that will test the ACE_Reactor's multi-threading
+// capabilities in separate threads.
+
+int
+Thread_Handler::svc (void)
+{
+ ACE_Time_Value sleep_timeout (0,
+ // Transform this into microseconds and divide by 2.
+ (Thread_Handler::interval_.sec () * ACE_ONE_SECOND_IN_USECS) / 2);
+
+ for (int i = this->iterations_;
+ i > 0;
+ --i)
+ {
+ if (this->shutdown_ != 0)
+ break;
+
+ // Block for delay_.secs () / 2, then notify the Reactor.
+ ACE_OS::sleep (sleep_timeout);
+
+ // Wait up to 10 milliseconds to notify the Reactor.
+ ACE_Time_Value timeout (0,
+ 10 * 1000);
+ if (this->notify (&timeout) == -1)
+ ACE_ERROR ((LM_ERROR,
+ "(%t) %p\n",
+ "notify"));
+ }
+
+ ACE_Reactor::instance ()->remove_handler (this,
+ ALL_EVENTS_MASK);
+ ACE_DEBUG ((LM_DEBUG,
+ "(%t) exiting svc\n"));
+ return 0;
+}
+
+// Test signal handling.
+
+int
+Thread_Handler::handle_signal (int signum, siginfo_t *, ucontext_t *)
+{
+ // @@ Note that this code is not portable to all OS platforms since
+ // it uses print statements within signal handler context.
+ ACE_DEBUG ((LM_DEBUG,
+ "(%t) received signal %S\n",
+ signum));
+
+ switch (signum)
+ {
+ case SIGINT:
+ case SIGQUIT:
+ ACE_ERROR ((LM_ERROR,
+ "(%t) ******************** shutting down %n on signal %S\n",
+ signum));
+ this->shutdown_ = 1;
+ ACE_Reactor::end_event_loop();
+ }
+ return 0;
+}
+
+int
+Thread_Handler::handle_timeout (const ACE_Time_Value &time, const void *)
+{
+ ACE_DEBUG ((LM_DEBUG,
+ "(%t) received timeout at (%u, %u), iterations = %d\n",
+ time.sec (),
+ time.usec (),
+ this->iterations_));
+
+ if (--this->iterations_ <= 0
+ || Thread_Handler::interval_.sec () == 0)
+ ACE_Reactor::end_event_loop ();
+
+ return 0;
+}
+
+// Called by the ACE_Reactor when it receives a notification.
+
+int
+Thread_Handler::handle_exception (ACE_HANDLE)
+{
+ ACE_DEBUG ((LM_DEBUG,
+ "(%t) exception to id %d, iteration = %d\n",
+ this->id_,
+ this->iterations_));
+ return 0;
+}
+
+// Called by the ACE_Reactor when it receives a notification.
+
+int
+Thread_Handler::handle_output (ACE_HANDLE)
+{
+ ACE_DEBUG ((LM_DEBUG,
+ "(%t) output to id %d, iteration = %d\n",
+ this->id_,
+ // This decrement must come last since
+ // <handle_exception> is called before <handle_output>!
+ this->iterations_--));
+ return 0;
+}
+
+// "Shim" function that integrates C thread API with C++.
+
+void *
+Thread_Handler::svc_run (void *eh)
+{
+ Thread_Handler *this_handler =
+ reinterpret_cast<Thread_Handler *> (eh);
+
+ if (this_handler->svc () == 0)
+ return 0;
+ else
+ return reinterpret_cast<void *> (-1);
+}
+
+int
+ACE_TMAIN (int argc, ACE_TCHAR *argv[])
+{
+ ACE_LOG_MSG->open (argv[0]);
+
+ if (argc < 4)
+ {
+ ACE_ERROR ((LM_ERROR,
+ ACE_TEXT ("usage: %s delay interval n_threads [iterations]\n"),
+ argv[0]));
+ ACE_OS::exit (1);
+ }
+
+ int delay = ACE_OS::atoi (argv[1]);
+ int interval = ACE_OS::atoi (argv[2]);
+ size_t n_threads = ACE_OS::atoi (argv[3]);
+ size_t max_iterations = argc > 4 ? ACE_OS::atoi (argv[4]) : MAX_ITERATIONS;
+
+ Thread_Handler thr_handler (delay,
+ interval,
+ n_threads,
+ max_iterations);
+
+ ACE_Reactor::instance ()->run_reactor_event_loop ();
+
+ ACE_DEBUG ((LM_DEBUG,
+ ACE_TEXT ("exiting from main\n")));
+ return 0;
+}
+#else
+int
+main (int, char *[])
+{
+ ACE_ERROR_RETURN ((LM_ERROR,
+ "threads must be supported to run this application\n"), -1);
+}
+#endif /* ACE_HAS_THREADS */