diff options
Diffstat (limited to 'ACE/examples/APG/Threads')
| -rw-r--r-- | ACE/examples/APG/Threads/.cvsignore | 12 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Activate.cpp | 33 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Condition_Variables.cpp | 118 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Guards.cpp | 95 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Makefile.am | 104 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Message_Blocks.cpp | 50 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Message_Queue.cpp | 179 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Message_Receiver.h | 90 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/Mutexes.cpp | 75 | ||||
| -rw-r--r-- | ACE/examples/APG/Threads/threads.mpc | 44 |
10 files changed, 800 insertions, 0 deletions
diff --git a/ACE/examples/APG/Threads/.cvsignore b/ACE/examples/APG/Threads/.cvsignore new file mode 100644 index 00000000000..feb9ec3ab97 --- /dev/null +++ b/ACE/examples/APG/Threads/.cvsignore @@ -0,0 +1,12 @@ +Activate +Activate +Condition_Variables +Condition_Variables +Guards +Guards +Message_Blocks +Message_Blocks +Message_Queue +Message_Queue +Mutexes +Mutexes diff --git a/ACE/examples/APG/Threads/Activate.cpp b/ACE/examples/APG/Threads/Activate.cpp new file mode 100644 index 00000000000..2afa6316f91 --- /dev/null +++ b/ACE/examples/APG/Threads/Activate.cpp @@ -0,0 +1,33 @@ +// $Id$ + +// Listing 1 code/ch12 +#include "ace/Task.h" +#include "ace/OS_NS_unistd.h" + +class HA_CommandHandler : public ACE_Task_Base +{ +public: + virtual int svc (void) + { + ACE_DEBUG + ((LM_DEBUG, ACE_TEXT ("(%t) Handler Thread running\n"))); + ACE_OS::sleep (4); + return 0; + } +}; + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + ACE_DEBUG + ((LM_DEBUG, ACE_TEXT ("(%t) Main Thread running\n"))); + + HA_CommandHandler handler; + int result = handler.activate (); + ACE_ASSERT (result == 0); + + ACE_UNUSED_ARG (result); + + handler.wait (); + return 0; +} +// Listing 1 diff --git a/ACE/examples/APG/Threads/Condition_Variables.cpp b/ACE/examples/APG/Threads/Condition_Variables.cpp new file mode 100644 index 00000000000..ccb29895032 --- /dev/null +++ b/ACE/examples/APG/Threads/Condition_Variables.cpp @@ -0,0 +1,118 @@ +// $Id$ + +#include "ace/config-lite.h" +#if defined (ACE_HAS_THREADS) + +#include "ace/Task.h" +#include "ace/Synch.h" + +// Listing 1 code/ch12 +class HA_Device_Repository +{ +public: + HA_Device_Repository() : owner_(0) + { } + + int is_free (void) + { return (this->owner_ == 0); } + + int is_owner (ACE_Task_Base* tb) + { return (this->owner_ == tb); } + + ACE_Task_Base *get_owner (void) + { return this->owner_; } + + void set_owner (ACE_Task_Base *owner) + { this->owner_ = owner; } + + int update_device (int device_id); + +private: + ACE_Task_Base * owner_; +}; +// Listing 1 + +class HA_CommandHandler : public ACE_Task_Base +{ +public: + enum {NUM_USES = 10}; + + HA_CommandHandler (HA_Device_Repository& rep, + ACE_Condition<ACE_Thread_Mutex> &wait, + ACE_Thread_Mutex& mutex) + : rep_(rep), waitCond_(wait), mutex_(mutex) + { } + + virtual int svc (void); + +private: + HA_Device_Repository &rep_; + ACE_Condition<ACE_Thread_Mutex> &waitCond_; + ACE_Thread_Mutex &mutex_; +}; +// Listing 2 code/ch12 +int +HA_CommandHandler::svc (void) +{ + ACE_DEBUG ((LM_DEBUG, + ACE_TEXT ("(%t) Handler Thread running\n"))); + + for (int i = 0; i < NUM_USES; i++) + { + this->mutex_.acquire (); + while (!this->rep_.is_free ()) + this->waitCond_.wait (); + this->rep_.set_owner (this); + this->mutex_.release (); + + this->rep_.update_device (i); + + ACE_ASSERT (this->rep_.is_owner (this)); + this->rep_.set_owner (0); + + this->waitCond_.signal (); + } + + return 0; +} +// Listing 2 +int +HA_Device_Repository::update_device (int device_id) +{ + ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Updating device %d\n"), + device_id)); + + ACE_OS::sleep (1); + return 0; +} +// Listing 3 code/ch12 +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + HA_Device_Repository rep; + ACE_Thread_Mutex rep_mutex; + ACE_Condition<ACE_Thread_Mutex> wait (rep_mutex); + + HA_CommandHandler handler1 (rep, wait, rep_mutex); + HA_CommandHandler handler2 (rep, wait, rep_mutex); + + handler1.activate (); + handler2.activate (); + + handler1.wait (); + handler2.wait (); + + return 0; +} +// Listing 3 + +#else +#include "ace/OS_main.h" +#include "ace/OS_NS_stdio.h" + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + ACE_OS::puts (ACE_TEXT ("This example requires threads.")); + return 0; +} + +#endif /* ACE_HAS_THREADS */ diff --git a/ACE/examples/APG/Threads/Guards.cpp b/ACE/examples/APG/Threads/Guards.cpp new file mode 100644 index 00000000000..7ef23e1e4da --- /dev/null +++ b/ACE/examples/APG/Threads/Guards.cpp @@ -0,0 +1,95 @@ +// $Id$ +#include "ace/config-lite.h" +#if defined (ACE_HAS_THREADS) + +#include "ace/OS_main.h" +#include "ace/OS_Memory.h" +#include "ace/Guard_T.h" +#include "ace/Log_Msg.h" +#include "ace/Thread_Mutex.h" + +// This file exists primarily to get code into the book to show different +// ways to do the same thing. For complete context and explanation, please +// see APG chapter 12. + +class HA_Device_Repository { +public: + int update_device (int device_id); + +private: + ACE_Thread_Mutex mutex_; +}; + +class Object { +}; +static Object *object; + +#if 0 +// This is less-desired way to do this... + +// Listing 1 code/ch12 +int +HA_Device_Repository::update_device (int device_id) +{ + this->mutex_.acquire (); + ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Updating device %d\n"), + device_id)); + + // Allocate a new object. + ACE_NEW_RETURN (object, Object, -1); + // ... + // Use the object + + this->mutex_.release (); +} +// Listing 1 +// Listing 2 code/ch12 +int +HA_Device_Repository::update_device (int device_id) +{ + // Construct a guard specifying the type of the mutex as + // a template parameter and passing in the mutex to hold + // as a parameter. + ACE_Guard<ACE_Thread_Mutex> guard (this->mutex_); + + // This can throw an exception that is not caught here. + ACE_NEW_RETURN (object, Object, -1); + // .. + // Use the object. + // .. + // Guard is destroyed, automatically releasing the lock. +} +// Listing 2 +#endif /* 0 */ + +// Listing 3 code/ch12 +int +HA_Device_Repository::update_device (int /* device_id */) +{ + ACE_GUARD_RETURN (ACE_Thread_Mutex, mon, mutex_, -1); + + ACE_NEW_RETURN (object, Object, -1); + // Use the object. + // ... + return 0; +} +// Listing 3 + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + HA_Device_Repository rep; + rep.update_device (42); + return 0; +} + +#else +#include "ace/OS_main.h" +#include "ace/OS_NS_stdio.h" + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + ACE_OS::puts (ACE_TEXT ("This example requires threads.")); + return 0; +} + +#endif /* ACE_HAS_THREADS */ diff --git a/ACE/examples/APG/Threads/Makefile.am b/ACE/examples/APG/Threads/Makefile.am new file mode 100644 index 00000000000..a5a1b77d76a --- /dev/null +++ b/ACE/examples/APG/Threads/Makefile.am @@ -0,0 +1,104 @@ +## Process this file with automake to create Makefile.in +## +## $Id$ +## +## This file was generated by MPC. Any changes made directly to +## this file will be lost the next time it is generated. +## +## MPC Command: +## /acebuilds/ACE_wrappers-repository/bin/mwc.pl -include /acebuilds/MPC/config -include /acebuilds/MPC/templates -feature_file /acebuilds/ACE_wrappers-repository/local.features -noreldefs -type automake -exclude build,Kokyu + +ACE_BUILDDIR = $(top_builddir) +ACE_ROOT = $(top_srcdir) + +## Makefile.Activate.am +noinst_PROGRAMS = Activate + +Activate_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Activate_SOURCES = \ + Activate.cpp \ + Message_Receiver.h + +Activate_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Makefile.Condition_Variables.am +noinst_PROGRAMS += Condition_Variables + +Condition_Variables_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Condition_Variables_SOURCES = \ + Condition_Variables.cpp \ + Message_Receiver.h + +Condition_Variables_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Makefile.Guards.am +noinst_PROGRAMS += Guards + +Guards_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Guards_SOURCES = \ + Guards.cpp \ + Message_Receiver.h + +Guards_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Makefile.Message_Blocks.am +noinst_PROGRAMS += Message_Blocks + +Message_Blocks_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Message_Blocks_SOURCES = \ + Message_Blocks.cpp \ + Message_Receiver.h + +Message_Blocks_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Makefile.Message_Queue.am +noinst_PROGRAMS += Message_Queue + +Message_Queue_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Message_Queue_SOURCES = \ + Message_Queue.cpp \ + Message_Receiver.h + +Message_Queue_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Makefile.Mutexes.am +noinst_PROGRAMS += Mutexes + +Mutexes_CPPFLAGS = \ + -I$(ACE_ROOT) \ + -I$(ACE_BUILDDIR) + +Mutexes_SOURCES = \ + Mutexes.cpp \ + Message_Receiver.h + +Mutexes_LDADD = \ + $(ACE_BUILDDIR)/ace/libACE.la + +## Clean up template repositories, etc. +clean-local: + -rm -f *~ *.bak *.rpo *.sym lib*.*_pure_* core core.* + -rm -f gcctemp.c gcctemp so_locations *.ics + -rm -rf cxx_repository ptrepository ti_files + -rm -rf templateregistry ir.out + -rm -rf ptrepository SunWS_cache Templates.DB diff --git a/ACE/examples/APG/Threads/Message_Blocks.cpp b/ACE/examples/APG/Threads/Message_Blocks.cpp new file mode 100644 index 00000000000..96cbf3e8cfe --- /dev/null +++ b/ACE/examples/APG/Threads/Message_Blocks.cpp @@ -0,0 +1,50 @@ +// $Id$ + +#include "ace/OS_main.h" +#include "ace/OS_Memory.h" +#include "ace/OS_NS_stdio.h" +#include "ace/OS_NS_string.h" +#include "ace/Log_Msg.h" +#include "ace/Message_Block.h" + +int ACE_TMAIN (int, ACE_TCHAR **) +{ +#if 0 +// Just for the book... + +// Listing 1 code/ch12 + ACE_Message_Block *mb; + ACE_NEW_RETURN (mb, ACE_Message_Block (128), -1); + + const char *deviceAddr= "Dev#12"; + mb->copy (deviceAddr, ACE_OS::strlen (deviceAddr)+1); +// Listing 1 +#endif /* 0 */ +// Listing 2 code/ch12 + ACE_Message_Block *mb; + ACE_NEW_RETURN (mb, ACE_Message_Block (128), -1); + + const char *commandSeq= "CommandSeq#14"; + ACE_OS::sprintf (mb->wr_ptr (), commandSeq); + // Move the wr_ptr() forward in the buffer by the + // amount of data we just put in. + mb->wr_ptr (ACE_OS::strlen (commandSeq) +1); +// Listing 2 +// Listing 3 code/ch12 + ACE_DEBUG((LM_DEBUG, + ACE_TEXT ("Command Sequence --> %C\n"), + mb->rd_ptr ())); + mb->rd_ptr (ACE_OS::strlen (mb->rd_ptr ())+1); + mb->release (); +// Listing 3 +// Listing 4 code/ch12 + // Send a hangup notification to the receiver. + ACE_NEW_RETURN + (mb, ACE_Message_Block (128, ACE_Message_Block::MB_HANGUP), -1); + // Send an error notification to the receiver. + mb->msg_type (ACE_Message_Block::MB_ERROR); +// Listing 4 + mb->release (); + + return 0; +} diff --git a/ACE/examples/APG/Threads/Message_Queue.cpp b/ACE/examples/APG/Threads/Message_Queue.cpp new file mode 100644 index 00000000000..3544d6bcefa --- /dev/null +++ b/ACE/examples/APG/Threads/Message_Queue.cpp @@ -0,0 +1,179 @@ +// $Id$ + +#include "ace/config-lite.h" +#if defined (ACE_HAS_THREADS) + +#include "ace/SOCK_Acceptor.h" +#include "ace/Acceptor.h" +#include "Message_Receiver.h" + +// Listing 5 code/ch12 +int +HA_CommandHandler::svc (void) +{ + while(1) + { + ACE_Message_Block *mb; + if (this->getq (mb) == -1) + break; + if (mb->msg_type () == ACE_Message_Block::MB_HANGUP) + { + mb->release (); + break; + } + else + { + // Get header pointer, then move past header to payload. + DeviceCommandHeader *dch + = (DeviceCommandHeader*)mb->rd_ptr (); + mb->rd_ptr (sizeof (DeviceCommandHeader)); + ACE_DEBUG ((LM_DEBUG, + ACE_TEXT ("Message for device #%d with ") + ACE_TEXT ("command payload of:\n%s"), + dch->deviceId_, mb->rd_ptr ())); + this->rep_.update_device (dch->deviceId_, + mb->rd_ptr ()); + mb->release (); + } + } + + ACE_Reactor::instance ()->end_reactor_event_loop (); + + return 0; +} +// Listing 5 + +// Listing 4 code/ch12 +ACE_Message_Block * +Message_Receiver::shut_down_message (void) +{ + ACE_Message_Block *mb; + ACE_NEW_RETURN + (mb, ACE_Message_Block (0, ACE_Message_Block::MB_HANGUP), 0); + return mb; +} +// Listing 4 + +int +Message_Receiver::read_header (DeviceCommandHeader *dch) +{ + ssize_t result = + this->peer ().recv_n (dch, sizeof (DeviceCommandHeader)); + if (result <= 0) + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("Recieve Failure")), + -1); + return 0; +} +// Listing 3 code/ch12 +int +Message_Receiver::copy_payload (ACE_Message_Block *mb, + int payload_length) +{ + ssize_t result = + this->peer ().recv_n (mb->wr_ptr (), payload_length); + + if (result <= 0) + { + mb->release (); + return -1; + } + + mb->wr_ptr (payload_length); + return 0; +} +// Listing 3 +// Listing 2 code/ch12 +int +Message_Receiver::handle_input (ACE_HANDLE) +{ + DeviceCommandHeader dch; + if (this->read_header (&dch) < 0) + return -1; + + if (dch.deviceId_ < 0) + { + // Handle shutdown. + this->handler_->putq (shut_down_message ()); + return -1; + } + + ACE_Message_Block *mb; + ACE_NEW_RETURN + (mb, ACE_Message_Block (dch.length_ + sizeof dch), -1); + // Copy the header. + mb->copy ((const char*)&dch, sizeof dch); + // Copy the payload. + if (this->copy_payload (mb, dch.length_) < 0) + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("Recieve Failure")), -1); + // Pass it off to the handler thread. + this->handler_->putq (mb); + return 0; +} +// Listing 2 + +static void report_usage (int argc, ACE_TCHAR *argv[]) +{ + if (argc < 2) + { + ACE_DEBUG ((LM_ERROR, ACE_TEXT ("%s port\n"), argv[1])); + ACE_OS::exit (-1); + } +} + + +class Acceptor : public ACE_Acceptor<Message_Receiver, ACE_SOCK_ACCEPTOR> +{ +public: + Acceptor(HA_CommandHandler *handler) : handler_(handler) + { } + +protected: + virtual int make_svc_handler (Message_Receiver *&mr) + { + ACE_NEW_RETURN (mr, Message_Receiver (handler_), -1); + return 0; + } + +private: + HA_CommandHandler *handler_; +}; + +int ACE_TMAIN (int argc, ACE_TCHAR *argv[]) +{ + report_usage (argc, argv); + + u_short port = ACE_OS::atoi (argv[1]); + + HA_Device_Repository rep; + HA_CommandHandler handler (rep); + ACE_ASSERT(handler.activate()==0); + //start up the handler. + + Acceptor acceptor (&handler); + ACE_INET_Addr addr (port); + if (acceptor.open (addr) == -1) + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("Failed to open connection")), -1); + + ACE_Reactor::instance()->run_reactor_event_loop (); + //run the reactive event loop + + handler.wait (); + //reap the handler before exiting. + + return 0; +} + +#else +#include "ace/OS_main.h" +#include "ace/OS_NS_stdio.h" + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + ACE_OS::puts (ACE_TEXT ("This example requires threads.")); + return 0; +} + +#endif /* ACE_HAS_THREADS */ diff --git a/ACE/examples/APG/Threads/Message_Receiver.h b/ACE/examples/APG/Threads/Message_Receiver.h new file mode 100644 index 00000000000..dee0731f007 --- /dev/null +++ b/ACE/examples/APG/Threads/Message_Receiver.h @@ -0,0 +1,90 @@ +/** + * $Id$ + * + * Sample code from The ACE Programmer's Guide, + * copyright 2003 Addison-Wesley. All Rights Reserved. + */ + +#ifndef __MESSAGE_RECEIVER_H_ +#define __MESSAGE_RECEIVER_H_ + +#include "ace/Log_Msg.h" +#include "ace/Message_Block.h" +#include "ace/SOCK_Stream.h" +#include "ace/Svc_Handler.h" +#include "ace/Synch.h" +#include "ace/Task.h" + +// Listing 1 code/ch12 +struct DeviceCommandHeader +{ + int length_; + int deviceId_; +}; +// Listing 1 + +class HA_Device_Repository +{ +public: + HA_Device_Repository (); + + int update_device (int device_id, char *commands); + +private: + ACE_Task_Base *owner_; +}; + +HA_Device_Repository::HA_Device_Repository () +{ } + +int +HA_Device_Repository::update_device (int, char *) +{ + return 0; +} + +class HA_CommandHandler : public ACE_Task<ACE_MT_SYNCH> +{ +public: + HA_CommandHandler (HA_Device_Repository &rep) : rep_(rep) + { } + + virtual int svc(); + +private: + HA_Device_Repository &rep_; +}; + +class Message_Receiver : + public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH> +{ +public: + Message_Receiver () : handler_(0) + { + ACE_ASSERT(0); + } + + Message_Receiver (HA_CommandHandler *ch) : handler_(ch) + { } + + ACE_Message_Block *shut_down_message (void); + + virtual int handle_input (ACE_HANDLE fd); + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::ALL_EVENTS_MASK) + { + this->peer ().close (); + delete this; + return 0; + } + +private: + int read_header (DeviceCommandHeader *dch); + int copy_payload (ACE_Message_Block *mb, int payload_length); + +private: + HA_CommandHandler *handler_; +}; + +#endif /* __MESSAGE_RECEIVER_H */ diff --git a/ACE/examples/APG/Threads/Mutexes.cpp b/ACE/examples/APG/Threads/Mutexes.cpp new file mode 100644 index 00000000000..ca5ebcdcbe4 --- /dev/null +++ b/ACE/examples/APG/Threads/Mutexes.cpp @@ -0,0 +1,75 @@ +// $Id$ + +#include "ace/config-lite.h" +#if defined (ACE_HAS_THREADS) + +#include "ace/Synch.h" +#include "ace/Task.h" + +// Listing 1 code/ch12 +class HA_Device_Repository +{ +public: + HA_Device_Repository () + { } + + void update_device (int device_id) + { + mutex_.acquire (); + ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Updating device %d\n"), + device_id)); + ACE_OS::sleep (1); + mutex_.release (); + } + +private: + ACE_Thread_Mutex mutex_; +}; +// Listing 1 +// Listing 2 code/ch12 +class HA_CommandHandler : public ACE_Task_Base +{ +public: + enum {NUM_USES = 10}; + + HA_CommandHandler (HA_Device_Repository& rep) : rep_(rep) + { } + + virtual int svc (void) + { + ACE_DEBUG + ((LM_DEBUG, ACE_TEXT ("(%t) Handler Thread running\n"))); + for (int i=0; i < NUM_USES; i++) + this->rep_.update_device (i); + return 0; + } + +private: + HA_Device_Repository & rep_; +}; + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + HA_Device_Repository rep; + HA_CommandHandler handler1 (rep); + HA_CommandHandler handler2 (rep); + handler1.activate (); + handler2.activate (); + + handler1.wait (); + handler2.wait (); + return 0; +} +// Listing 2 + +#else +#include "ace/OS_main.h" +#include "ace/OS_NS_stdio.h" + +int ACE_TMAIN (int, ACE_TCHAR *[]) +{ + ACE_OS::puts (ACE_TEXT ("This example requires threads.")); + return 0; +} + +#endif /* ACE_HAS_THREADS */ diff --git a/ACE/examples/APG/Threads/threads.mpc b/ACE/examples/APG/Threads/threads.mpc new file mode 100644 index 00000000000..a980a867d26 --- /dev/null +++ b/ACE/examples/APG/Threads/threads.mpc @@ -0,0 +1,44 @@ +// -*- MPC -*- +// $Id$ + +project(Activate) : aceexe { + exename = Activate + Source_Files { + Activate.cpp + } +} + +project(Condition Variables) : aceexe { + exename = Condition_Variables + Source_Files { + Condition_Variables.cpp + } +} + +project(Guards) : aceexe { + exename = Guards + Source_Files { + Guards.cpp + } +} + +project(Message Blocks) : aceexe { + exename = Message_Blocks + Source_Files { + Message_Blocks.cpp + } +} + +project(Message Queue) : aceexe { + exename = Message_Queue + Source_Files { + Message_Queue.cpp + } +} + +project(Mutexes) : aceexe { + exename = Mutexes + Source_Files { + Mutexes.cpp + } +} |