diff options
69 files changed, 2657 insertions, 925 deletions
diff --git a/ChangeLog-96b b/ChangeLog-96b index d9f57161a00..dfaa77649f9 100644 --- a/ChangeLog-96b +++ b/ChangeLog-96b @@ -1,3 +1,59 @@ +Sun Dec 15 10:29:20 1996 Douglas C. Schmidt <schmidt@flamenco.cs.wustl.edu> + + * netsvcs/servers/svc.conf: Removed the "lib" prefix for the + netsvcs DLL. This is now added automatically by the + ACE::ldfind() operation. + + * ace/SString.cpp (ACE_CString): Removed the #pragmas for Win32. + They aren't necessary since we should replace the ACE_USHORT16 + cast with a char cast. Thanks to Amos Shapira <amos@dsi.co.il> + for reporting this. + +Sat Dec 14 14:25:38 1996 Douglas C. Schmidt <schmidt@tango.cs.wustl.edu> + + * build/SunOS5.5/tests/UPIPE_SAP_Test.cpp (main): Fixed several + minor bugs with UPIPE_SAP_Test.cpp. + + * ace/OS.i (thr_join): Added implementations for Solaris threads + and most versions of POSIX pthreads where ACE_hthread_t and + ACE_thread_t are the same type! + + * ace/OS: Began adding hooks so that we can eventually move away + from the current split between ACE_thread_t and ACE_hthread_t + and unify them via ACE_Thread_ID. + + * ace/{OS,Thread}.h: Changed the interface of thr_getprio() so + that it takes an int & rather than an int *. + + * ace/OS.i (thr_getprio): Fixed a minor bug for Win32 where we + weren't depositing the thread priority into the return value! + + * Makefile: Changed the order in which things are built so that + netsvcs are built right after libACE, followed by the tests. + +Sat Dec 14 11:54:22 1996 Douglas C. Schmidt <schmidt@flamenco.cs.wustl.edu> + + * apps/Gateway/Gateway/Consumer_Map: Change the Consumer_Map class + so that it was no longer templatized. There isn't any point in + doing this since we aren't going to be changing these types for + this application. + + * apps/Gateway/Gateway: Factored out the code for selecting the + concurrency strategy into a separate *.h file called + Concurrency_Strategy.h. + + * apps/Gateway/Gateway: Began revising the Gateway application to + use the new ACE Event Channel. + + * ace/Svc_Handler: Now that we've got put() and svc() with no-op + defaults in class ACE_Task_Base, we don't need them in + ACE_Svc_Handler anymore, so I removed them! + + * ace/Task: Finally got sick of having to provide no-op + open()/put()/close() routines in all ACE_Task subclasses, so I + changed these methods from pure virtual to virtual with default + no-op behavior. Updated all the tests, as well. + Sat Dec 14 11:39:15 1996 David L. Levine <levine@cs.wustl.edu> * ace/{Module,Stream,Svc_Handler,Synch_T,Task_T}.cpp and Synch_T.i: @@ -21,8 +77,6 @@ Fri Dec 13 22:07:11 1996 David L. Levine <levine@cs.wustl.edu> SOLINK step in build of shared objects for SunOS5 with SunC++ with symlink from .so to .o file. -Fri Dec 13 13:44:12 1996 David L. Levine <levine@cs.wustl.edu> - * ace/config-vxworks*.h: added ACE_NEEDS_SYSTIME_H to VxWorks configs because it's needed with inlining @@ -58,6 +112,16 @@ Thu Dec 12 18:51:04 1996 Douglas C. Schmidt <schmidt@tango.cs.wustl.edu> * ace/Thread: Added getprio() and setprio() methods to ACE_Thread. +Fri Dec 13 13:44:12 1996 David L. Levine <levine@cs.wustl.edu> + + * ace/config-vxworks*.h: added ACE_NEEDS_SYSTIME_H to VxWorks + configs because it's needed with inlining + + * include/makeinclude/platform_vxworks*.GNU: cleaned up VxWorks + config files + +Thu Dec 12 18:51:04 1996 Douglas C. Schmidt <schmidt@tango.cs.wustl.edu> + * ace: Added a new macro called ACE_UNUSED_ARG() to keep the compiler from outputting warnings about unused arguments. So far, this is mostly done for Win32, but it @@ -12,11 +12,11 @@ INFO = README \ VERSION DIRS = ace \ + netsvcs \ + tests \ apps \ examples \ - netsvcs \ - performance-tests \ - tests + performance-tests CLONE = Makefile \ ace \ @@ -12,32 +12,32 @@ An Object-Oriented Network Programming Toolkit Overview of ACE The ADAPTIVE Communication Environment (ACE) is an object-oriented -toolkit that implements strategic and tactical design patterns to -simplify the development of concurrent, event-driven communication -software. ACE provides a rich set of reusable C++ wrappers, class -categories, and frameworks that perform common communication software -tasks across a range of operating system platforms. The communication -software tasks provided by ACE include event demultiplexing and event -handler dispatching, connection establishment, interprocess -communication, shared memory management, message routing, dynamic -(re)configuration of distributed services, multi-threading, and -concurrency control. - -ACE is targeted for developers of high-performance concurrent network -applications and services. The primary goal of ACE is to simplify the -development of concurrent OO communication software that utilizes -interprocess communication, event demultiplexing, explicit dynamic -linking, and concurrency. In addition, ACE automates communication -software configuration and reconfiguration by dynamically linking -services into applications at run-time and executing these services on -one or more processes or threads. +(OO) toolkit that implements fundamental design patterns for +communication software. ACE provides a rich set of reusable C++ +wrappers, class categories, and frameworks that perform common +communication software tasks across a range of operating system +platforms. The communication software tasks provided by ACE include +event demultiplexing and event handler dispatching, service +initialization, interprocess communication, shared memory management, +message routing, dynamic (re)configuration of distributed services, +multi-threading, and concurrency control. + +ACE is targeted for developers of high-performance communication +services and applications on UNIX, POSIX, and Win32 platforms. ACE +simplifies the development of OO network applications and services +that utilize interprocess communication, event demultiplexing, +explicit dynamic linking, and concurrency. ACE automates system +configuration and reconfiguration by dynamically linking services into +applications at run-time and executing these services in one or more +processes or threads. ACE has been ported to a wide range of uni-processor and multi-process OS platforms including Win32 (i.e., WinNT and Win95), most versions of UNIX (e.g., SunOS 4.x and 5.x, SGI IRIX, HP-UX, OSF/1, AIX, Linux, and SCO), VxWorks, and MVS OpenEdition. It is currently used in commercial products by dozens of companies including Ericsson, -Bellcore, Siemens, Motorola, and Kodak. +Bellcore, Siemens, Motorola, and Kodak. There are both C++ and Java +versions of ACE available. The remainder of this document outlines the structure and participants of the layers in this diagram. @@ -130,12 +130,12 @@ medical engineering. OBTAINING ACE -The current ACE release is provided as a tar file that is slightly -larger than 1.5 Meg compressed using GNU gzip. ACE may be obtained -electronically from http://www.cs.wustl.edu/~schmidt/ACE-obtain.html. -This release contains contains the source code, test drivers, and -example applications for C++ wrapper libraries and the higher-level -ACE network programming framework developed as part of the ADAPTIVE +The current ACE release is provided as a tar file that is around 1.8 +Meg compressed using GNU gzip. ACE may be obtained electronically +from http://www.cs.wustl.edu/~schmidt/ACE-obtain.html. This release +contains contains the source code, test drivers, and example +applications for C++ wrapper libraries and the higher-level ACE +network programming framework developed as part of the ADAPTIVE project at the University of California, Irvine and at Washington University. diff --git a/ace/OS.cpp b/ace/OS.cpp index 6559f6209aa..e0b395e66a1 100644 --- a/ace/OS.cpp +++ b/ace/OS.cpp @@ -1455,3 +1455,47 @@ ace_mutex_lock_cleanup_adapter (void *args) { ACE_OS::mutex_lock_cleanup (args); } + +ACE_Thread_ID::ACE_Thread_ID (ACE_thread_t thread_id, + ACE_hthread_t thread_handle) + : thread_id_ (thread_id), + thread_handle_ (thread_handle) +{ +} + +ACE_thread_t +ACE_Thread_ID::id (void) +{ + return this->thread_id_; +} + +void +ACE_Thread_ID::id (ACE_thread_t thread_id) +{ + this->thread_id_ = thread_id; +} + +ACE_hthread_t +ACE_Thread_ID::handle (void) +{ + return this->thread_handle_; +} + +void +ACE_Thread_ID::handle (ACE_hthread_t thread_handle) +{ + this->thread_handle_ = thread_handle; +} + +int +ACE_Thread_ID::operator == (const ACE_Thread_ID &rhs) +{ + return this->thread_handle_ == rhs.thread_handle_ + && this->thread_id_ == rhs.thread_id_; +} + +int +ACE_Thread_ID::operator != (const ACE_Thread_ID &rhs) +{ + return !((*this) == rhs); +} @@ -1852,6 +1852,33 @@ typedef int ucontext_t; #undef t_errno #endif /* ACE_HAS_BROKEN_T_ERRNO */ +class ACE_Export ACE_Thread_ID + // = TITLE + // Defines a platform-independent thread ID. +{ +public: + ACE_Thread_ID (ACE_thread_t, ACE_hthread_t); + + // = Set/Get the Thread ID. + ACE_thread_t id (void); + void id (ACE_thread_t); + + // = Set/Get the Thread handle. + ACE_hthread_t handle (void); + void handle (ACE_hthread_t); + + // != Comparison operator. + int operator == (const ACE_Thread_ID &); + int operator != (const ACE_Thread_ID &); + +private: + ACE_thread_t thread_id_; + // Identify the thread. + + ACE_hthread_t thread_handle_; + // Handle to the thread (typically used to "wait" on Win32). +}; + // Type of the extended signal handler. typedef void (*ACE_Sig_Handler_Ex) (int, siginfo_t *siginfo, ucontext_t *ucontext); @@ -2320,7 +2347,26 @@ public: static int t_sync (ACE_HANDLE fildes); static int t_unbind (ACE_HANDLE fildes); - // = A set of wrappers for threads. +#if 0 + // = A set of wrappers for threads (these are portable since they use the ACE_Thread_ID). + static int thr_continue (const ACE_Thread_ID &thread); + static int thr_create (ACE_THR_FUNC, + void *args, + long flags, + ACE_Thread_ID *, + u_int priority = 0, + void *stack = 0, + size_t stacksize = 0); + static int thr_getprio (ACE_Thread_ID thr_id, int &prio, int *policy = 0); + static int thr_join (ACE_Thread_ID waiter_id, void **status); + static int thr_kill (ACE_Thread_ID thr_id, int signum); + static ACE_Thread_ID thr_self (void); + static int thr_setprio (ACE_Thread_ID thr_id, int prio); + static int thr_suspend (ACE_Thread_ID target_thread); + static int thr_cancel (ACE_Thread_ID t_id); +#endif /* 0 */ + + // = A set of wrappers for threads (these are non-portable since they use ACE_thread_t and ACE_hthread_t and will go away in a future release). static int thr_continue (ACE_hthread_t target_thread); static int thr_create (ACE_THR_FUNC, void *args, @@ -2330,30 +2376,31 @@ public: u_int priority = 0, void *stack = 0, size_t stacksize = 0); + static int thr_getprio (ACE_hthread_t thr_id, int &prio); + static int thr_join (ACE_hthread_t waiter_id, void **status); + static int thr_join (ACE_thread_t waiter_id, ACE_thread_t *thr_id, void **status); + static int thr_kill (ACE_thread_t thr_id, int signum); + static ACE_thread_t thr_self (void); + static void thr_self (ACE_hthread_t &); + static int thr_setprio (ACE_hthread_t thr_id, int prio); + static int thr_suspend (ACE_hthread_t target_thread); + static int thr_cancel (ACE_thread_t t_id); + static int thr_cmp (ACE_hthread_t t1, ACE_hthread_t t2); static int thr_equal (ACE_thread_t t1, ACE_thread_t t2); static void thr_exit (void *status = 0); static int thr_getconcurrency (void); - static int thr_getprio (ACE_hthread_t thr_id, int *prio); static int thr_getspecific (ACE_thread_key_t key, void **data); - static int thr_join (ACE_hthread_t waiter_id, void **status); - static int thr_join (ACE_thread_t waiter_id, ACE_thread_t *thr_id, void **status); static int thr_keyfree (ACE_thread_key_t key); static int thr_key_detach (void *inst); static int thr_keycreate (ACE_thread_key_t *key, ACE_THR_DEST, void *inst = 0); static int thr_key_used (ACE_thread_key_t key); - static int thr_kill (ACE_thread_t thr_id, int signum); static size_t thr_min_stack (void); - static ACE_thread_t thr_self (void); - static void thr_self (ACE_hthread_t &); static int thr_setconcurrency (int hint); - static int thr_setprio (ACE_hthread_t thr_id, int prio); static int thr_setspecific (ACE_thread_key_t key, void *data); static int thr_sigsetmask (int how, const sigset_t *nsm, sigset_t *osm); - static int thr_suspend (ACE_hthread_t target_thread); static int thr_setcancelstate (int new_state, int *old_state); static int thr_setcanceltype (int new_type, int *old_type); - static int thr_cancel (ACE_thread_t t_id); static int sigwait (sigset_t *set, int *sig = 0); static void thr_testcancel (void); static void thr_yield (void); @@ -3069,20 +3069,26 @@ ACE_OS::thr_getconcurrency (void) } ACE_INLINE int -ACE_OS::thr_getprio (ACE_hthread_t thr_id, int *prio) +ACE_OS::thr_getprio (ACE_hthread_t thr_id, int &prio) { // ACE_TRACE ("ACE_OS::thr_getprio"); #if defined (ACE_HAS_THREADS) #if defined (ACE_HAS_STHREADS) - ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::thr_getprio (thr_id, prio), ace_result_), int, -1); -#elif defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS) - ACE_NOTSUP_RETURN (-1); + ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::thr_getprio (thr_id, &prio), ace_result_), int, -1); +#elif (defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS)) && !defined (ACE_LACKS_SETSCHED) + struct sched_param param; + int result; + int policy = 0; + + ACE_OSCALL (ACE_ADAPT_RETVAL (::pthread_getschedparam (thr_id, &policy, ¶m), result), int, -1, result); + prio = param.sched_priority; + return result; #elif defined (ACE_HAS_WTHREADS) ACE_UNUSED_ARG(prio); // why is the thread prio not dropped into prio ? - int result = ::GetThreadPriority (thr_id); - return result == THREAD_PRIORITY_ERROR_RETURN ? -1 : result; + prio = ::GetThreadPriority (thr_id); + return prio == THREAD_PRIORITY_ERROR_RETURN ? -1 : 0; #elif defined (VXWORKS) ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::taskPriorityGet (thr_id, prio), ace_result_), int, -1); #endif /* ACE_HAS_STHREADS */ @@ -3159,9 +3165,11 @@ ACE_OS::thr_join (ACE_hthread_t thr_handle, void **status) status = status; #if defined (ACE_HAS_THREADS) #if defined (ACE_HAS_STHREADS) - ACE_NOTSUP_RETURN (-1); -#elif defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS) - ACE_NOTSUP_RETURN (-1); + ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::thr_join (thr_handle, 0, status), ace_result_), + int, -1); +#elif (defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS)) && !defined (ACE_HAS_THREAD_SELF) + ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::pthread_join (thr_handle, status), ace_result_), + int, -1); #elif defined (ACE_HAS_WTHREADS) void *local_status = 0; @@ -3259,12 +3267,12 @@ ACE_OS::thr_setcanceltype (int new_type, int *old_type) } ACE_INLINE int -ACE_OS::thr_cancel (ACE_thread_t t_id) +ACE_OS::thr_cancel (ACE_thread_t thr_id) { // ACE_TRACE ("ACE_OS::thr_cancel"); #if defined (ACE_HAS_THREADS) #if defined (ACE_HAS_DCETHREADS) || (defined (ACE_HAS_PTHREADS) && defined (ACE_HAS_STHREADS)) - ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::pthread_cancel(t_id), + ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::pthread_cancel (thr_id), ace_result_), int, -1); #elif defined (ACE_HAS_PTHREADS) @@ -3275,7 +3283,7 @@ ACE_OS::thr_cancel (ACE_thread_t t_id) #elif defined (ACE_HAS_STHREADS) ACE_NOTSUP_RETURN (-1); #elif defined (ACE_HAS_WTHREADS) - ACE_UNUSED_ARG(t_id); + ACE_UNUSED_ARG(thr_id); ACE_NOTSUP_RETURN (-1); #elif defined (VXWORKS) @@ -3554,7 +3562,13 @@ ACE_OS::thr_setprio (ACE_hthread_t thr_id, int prio) ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::thr_setprio (thr_id, prio), ace_result_), int, -1); -#elif defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS) +#elif (defined (ACE_HAS_DCETHREADS) || defined (ACE_HAS_PTHREADS)) && !defined (ACE_LACKS_SETSCHED) + struct sched_param param; + int result; + + ACE_OSCALL (ACE_ADAPT_RETVAL (::pthread_setschedparam (thr_id, policy, ¶m), result), int, -1, result); + prio = param.sched_priority; + return result; ACE_NOTSUP_RETURN (-1); #elif defined (ACE_HAS_WTHREADS) ACE_OSCALL_RETURN (ACE_ADAPT_RETVAL (::SetThreadPriority (thr_id, prio), @@ -5842,3 +5856,91 @@ ACE_OS::mkdir (const wchar_t *path, mode_t mode) #endif /* ACE_WIN32 */ #endif /* ACE_HAS_UNICODE */ + +#if 0 +ACE_INLINE int +ACE_OS::thr_continue (const ACE_Thread_ID &thr_id) +{ + // ACE_TRACE ("ACE_OS::thr_continue"); + return ACE_OS::thr_continue (thr_id.id ()); +} + +ACE_INLINE int +ACE_OS::thr_create (ACE_THR_FUNC func, + void *args, + long flags, + ACE_Thread_ID *thr_id = 0, + u_int priority = 0, + void *stack = 0, + size_t stacksize = 0); +{ + // ACE_TRACE ("ACE_OS::thr_create"); + ACE_thread_t thread_id; + ACE_hthread_t thread_handle; + + int result = ACE_OS::thr_create (func, args, flags, + &thread_id, &thread_handle, + priority, stack, stacksize); + if (result == -1) + return -1; + else if (thr_id != 0) + { + thr_id->id (thread_id); + thr_id->handle (thread_handle); + return result; + } +} + +ACE_INLINE int +ACE_OS::thr_getprio (const ACE_Thread_ID &thr_id, int &prio) +{ + // ACE_TRACE ("ACE_OS::thr_getprio"); + return ACE_OS::thr_getprio (thr_id.handle (), prio); +} + +ACE_INLINE int +ACE_OS::thr_join (const ACE_Thread_ID &thr_id, void **status) +{ +#if defined (ACE_WIN32) + return ACE_OS::join (thr_id.id (), status); +#else + return ACE_OS::join (thr_id.handle (), status); +#endif /* ACE_WIN32 */ +} + +ACE_INLINE int +ACE_OS::thr_cancel (const ACE_Thread_ID &thr_id) +{ + return ACE_OS::thr_cancel (thr_id.id ()); +} + +ACE_INLINE int +ACE_OS::thr_kill (const ACE_Thread_ID &thr_id, int signum) +{ + return ACE_OS::thr_kill (thr_id.id (), signum); +} + +ACE_INLINE ACE_Thread_ID +ACE_OS::thr_self (void) +{ + ACE_hthread_t thr_handle; + ACE_OS::thr_self (thr_handle); + ACE_thread_t thr_id = ACE_OS::thr_self (); + + return ACE_Thread_ID (thr_id, thr_handle); +} + +ACE_INLINE int +ACE_OS::thr_setprio (const ACE_Thread_ID &thr_id, int prio) +{ + // ACE_TRACE ("ACE_OS::thr_getprio"); + return ACE_OS::thr_setprio (thr_id.handle (), prio); +} + +ACE_INLINE int +ACE_OS::thr_suspend (const ACE_Thread_ID &thr_id) +{ + return ACE_OS::thr_suspend (thr_id.handle ()); +} + +#endif /* 0 */ diff --git a/ace/SString.cpp b/ace/SString.cpp index a8269f6362e..47c52efcf68 100644 --- a/ace/SString.cpp +++ b/ace/SString.cpp @@ -108,10 +108,7 @@ ACE_CString::ACE_CString (const ACE_USHORT16 *s, ACE_Allocator *alloc) // Copy the ACE_USHORT16 * string byte-by-byte into the char * // string. for (size_t i = 0; i < this->len_; i++) -#pragma warning(disable: 4244) /* Possible loss of data */ - this->rep_[i] = (ACE_USHORT16) s[i]; -#pragma warning(default: 4244) /* Possible loss of data */ - + this->rep_[i] = char (s[i]); this->rep_[this->len_] = '\0'; } } diff --git a/ace/Svc_Handler.cpp b/ace/Svc_Handler.cpp index 4285901216f..922b6b89eaa 100644 --- a/ace/Svc_Handler.cpp +++ b/ace/Svc_Handler.cpp @@ -217,21 +217,6 @@ ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::close (unsigned long) } template <PR_ST_1, ACE_SYNCH_1> int -ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::svc (void) -{ - ACE_TRACE ("ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::svc"); - return -1; -} - -template <PR_ST_1, ACE_SYNCH_1> int -ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::put (ACE_Message_Block *, - ACE_Time_Value *) -{ - ACE_TRACE ("ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::put"); - return -1; -} - -template <PR_ST_1, ACE_SYNCH_1> int ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::init (int, char *[]) { ACE_TRACE ("ACE_Svc_Handler<PR_ST_2, ACE_SYNCH_2>::init"); diff --git a/ace/Svc_Handler.h b/ace/Svc_Handler.h index b66aa783ebd..1228ad0964a 100644 --- a/ace/Svc_Handler.h +++ b/ace/Svc_Handler.h @@ -97,9 +97,6 @@ public: // Returns the underlying PEER_STREAM (used by // ACE_Acceptor::accept() and ACE_Connector::connect() factories). - virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0); - // Provide a default implementation to simplify ancestors... - virtual void destroy (void); // Call this instead of <delete> to free up dynamically allocated // <Svc_Handler>. This method knows whether or not the object was @@ -119,9 +116,6 @@ public: // This really should be private so that users are forced to call // destroy(). - virtual int svc (void); - // Provide a default implementation to simplify ancestors... - private: void shutdown (void); // Close down the descriptor diff --git a/ace/Task.h b/ace/Task.h index 0b11f6ead24..8e6f4752f27 100644 --- a/ace/Task.h +++ b/ace/Task.h @@ -57,11 +57,11 @@ public: ACE_Task_Base (ACE_Thread_Manager *); // = Initialization and termination hooks (note that these *must* be defined by subclasses). - virtual int open (void *args = 0) = 0; + virtual int open (void *args = 0); // Hook called to open a Task. <args> can be used to pass arbitrary // information into <open>. - virtual int close (u_long flags = 0) = 0; + virtual int close (u_long flags = 0); // Hook called from ACE_Task_Exit when during thread exit and from // the default implemenation of module_closed(). @@ -75,7 +75,7 @@ public: // ACE_Task instance exits. // = Immediate and deferred processing methods, respectively. - virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0) = 0; + virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0); // Transfer msg into the queue to handle immediate processing. virtual int svc (void); diff --git a/ace/Task.i b/ace/Task.i index bf6d310d2fc..7e682513761 100644 --- a/ace/Task.i +++ b/ace/Task.i @@ -60,3 +60,30 @@ ACE_Task_Base::svc (void) ACE_TRACE ("ACE_Task_Base::svc"); return 0; } + +// Default ACE_Task open routine + +ACE_INLINE int +ACE_Task_Base::open (void *) +{ + ACE_TRACE ("ACE_Task_Base::open"); + return 0; +} + +// Default ACE_Task close routine + +ACE_INLINE int +ACE_Task_Base::close (u_long) +{ + ACE_TRACE ("ACE_Task_Base::close"); + return 0; +} + +// Default ACE_Task put routine. + +ACE_INLINE int +ACE_Task_Base::put (ACE_Message_Block *, ACE_Time_Value *) +{ + ACE_TRACE ("ACE_Task_Base::put"); + return 0; +} diff --git a/ace/Thread.i b/ace/Thread.i index 6696a08e3f4..1a5e2746f65 100644 --- a/ace/Thread.i +++ b/ace/Thread.i @@ -244,7 +244,7 @@ ACE_Thread::self (ACE_hthread_t &t_id) } ACE_INLINE int -ACE_Thread::getprio (ACE_hthread_t t_id, int *prio) +ACE_Thread::getprio (ACE_hthread_t t_id, int &prio) { ACE_TRACE ("ACE_Thread::getprio"); return ACE_OS::thr_getprio (t_id, prio); diff --git a/ace/config-aix-3.2.5.h b/ace/config-aix-3.2.5.h index cb5bf928a89..eb60aa501ea 100644 --- a/ace/config-aix-3.2.5.h +++ b/ace/config-aix-3.2.5.h @@ -11,23 +11,14 @@ #define ACE_HAS_ONLY_TWO_PARAMS_FOR_ASCTIME_R_AND_CTIME_R #define ACE_HAS_RECURSIVE_THR_EXIT_SEMANTICS #define ACE_HAS_CONSISTENT_SIGNAL_PROTOTYPES - #define ACE_HAS_TEMPLATE_TYPEDEFS - #define ACE_HAS_STRERROR - #define ACE_HAS_SIG_ATOMIC_T - #define ACE_HAS_SSIZE_T - #define ACE_HAS_CPLUSPLUS_HEADERS - #define ACE_HAS_POLL - #define ACE_HAS_POSIX_NONBLOCK - #define ACE_HAS_AIX_GETTIMEOFDAY - #define ACE_HAS_NO_SYSCALL_H // Compiler/platform has the getrusage() system call. diff --git a/apps/Gateway/Gateway/Concurrency_Strategies.h b/apps/Gateway/Gateway/Concurrency_Strategies.h new file mode 100644 index 00000000000..e2fbc934c93 --- /dev/null +++ b/apps/Gateway/Gateway/Concurrency_Strategies.h @@ -0,0 +1,74 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Concurrency_strategies.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_CONCURRENCY_STRATEGIES) +#define _CONCURRENCY_STRATEGIES + +#include "ace/Synch.h" + +// The following typedefs are used in order to parameterize the +// synchronization policies without changing the source code! + +// If we don't have threads then use the single-threaded synchronization. +#if !defined (ACE_HAS_THREADS) +#define SYNCH_STRATEGY ACE_NULL_SYNCH +typedef ACE_Null_Mutex MAP_MUTEX; +#else /* ACE_HAS_THREADS */ + +// Note that we only need to make the ACE_Task thread-safe if we are +// using the multi-threaded Thr_Consumer_Handler... +#if defined (USE_OUTPUT_MT) +#define SYNCH_STRATEGY ACE_MT_SYNCH +#else +#define SYNCH_STRATEGY ACE_NULL_SYNCH +#endif /* USE_OUTPUT_MT || USE_INPUT_MT */ + +// Note that we only need to make the ACE_Map_Manager thread-safe if +// we are using the multi-threaded Thr_Supplier_Handler. In this +// case, we use an RW_Mutex since we'll lookup Consumers far more +// often than we'll update them. +#if defined (USE_INPUT_MT) +typedef ACE_RW_Mutex MAP_MUTEX; +#else +typedef ACE_Null_Mutex MAP_MUTEX; +#endif /* USE_INPUT_MT */ +#endif /* ACE_HAS_THREADS */ + +// = Forward decls +class Thr_Consumer_Handler; +class Thr_Supplier_Handler; +class Consumer_Handler; +class Supplier_Handler; + +#if defined (ACE_HAS_THREADS) && (defined (USE_OUTPUT_MT) || defined (USE_INPUT_MT)) +#if defined (USE_OUTPUT_MT) +typedef Thr_Consumer_Handler CONSUMER_HANDLER; +#else +typedef Consumer_Handler CONSUMER_HANDLER; +#endif /* USE_OUTPUT_MT */ + +#if defined (USE_INPUT_MT) +typedef Thr_Supplier_Handler SUPPLIER_HANDLER; +#else +typedef Supplier_Handler SUPPLIER_HANDLER; +#endif /* USE_INPUT_MT */ +#else +// Instantiate a non-multi-threaded Gateway. +typedef Supplier_Handler SUPPLIER_HANDLER; +typedef Consumer_Handler CONSUMER_HANDLER; +#endif /* ACE_HAS_THREADS */ + +#endif /* _CONCURRENCY_STRATEGIES */ diff --git a/apps/Gateway/Gateway/Config_Files.cpp b/apps/Gateway/Gateway/Config_Files.cpp index f2466544e5b..4c2648addf0 100644 --- a/apps/Gateway/Gateway/Config_Files.cpp +++ b/apps/Gateway/Gateway/Config_Files.cpp @@ -8,25 +8,23 @@ typedef FP::Return_Type FP_RETURN_TYPE; FP_RETURN_TYPE -RT_Config_File_Parser::read_entry (RT_Config_File_Entry &entry, - int &line_number) +Consumer_Config_File_Parser::read_entry (Consumer_Config_File_Entry &entry, + int &line_number) { FP_RETURN_TYPE read_result; - // increment the line count + + // Increment the line count. line_number++; - // Ignore comments, check for EOF and EOLINE - // if this succeeds, we have our connection id + // Ignore comments, check for EOF and EOLINE if this succeeds, we + // have our connection id. while ((read_result = this->getint (entry.conn_id_)) != FP::SUCCESS) { if (read_result == FP::EOFILE) return FP::EOFILE; else if (read_result == FP::EOLINE || read_result == FP::COMMENT) - { - // increment the line count - line_number++; - } + line_number++; } // Get the logic id. @@ -51,8 +49,8 @@ RT_Config_File_Parser::read_entry (RT_Config_File_Entry &entry, } FP_RETURN_TYPE -CC_Config_File_Parser::read_entry (CC_Config_File_Entry &entry, - int &line_number) +Connection_Config_File_Parser::read_entry (Connection_Config_File_Entry &entry, + int &line_number) { char buf[BUFSIZ]; FP_RETURN_TYPE read_result; @@ -67,10 +65,7 @@ CC_Config_File_Parser::read_entry (CC_Config_File_Entry &entry, return FP::EOFILE; else if (read_result == FP::EOLINE || read_result == FP::COMMENT) - { - // increment the line count - line_number++; - } + line_number++; } // get the hostname @@ -83,7 +78,7 @@ CC_Config_File_Parser::read_entry (CC_Config_File_Entry &entry, if ((read_result = this->getint (port)) != FP::SUCCESS) return read_result; else - entry.remote_port_ = (u_short) port; + entry.remote_poconsumer_ = (u_short) port; // Get the direction. if ((read_result = this->getword (buf)) != FP::SUCCESS) @@ -99,7 +94,7 @@ CC_Config_File_Parser::read_entry (CC_Config_File_Entry &entry, if ((read_result = this->getint (port)) != FP::SUCCESS) return read_result; else - entry.local_port_ = (u_short) port; + entry.local_poconsumer_ = (u_short) port; return FP::SUCCESS; } @@ -113,8 +108,8 @@ int main (int argc, char *argv[]) exit (1); } FP_RETURN_TYPE result; - CC_Config_File_Entry CCentry; - CC_Config_File_Parser CCfile; + Connection_Config_File_Entry CCentry; + Connection_Config_File_Parser CCfile; CCfile.open (argv[1]); @@ -130,13 +125,13 @@ int main (int argc, char *argv[]) cerr << "Error at line " << line_number << endl; else printf ("%d\t%s\t%d\t%c\t%d\t%c\t%d\n", - CCentry.conn_id_, CCentry.host_, CCentry.remote_port_, CCentry.direction_, - CCentry.max_retry_delay_, CCentry.transform_, CCentry.local_port_); + CCentry.conn_id_, CCentry.host_, CCentry.remote_poconsumer_, CCentry.direction_, + CCentry.max_retry_delay_, CCentry.transform_, CCentry.local_poconsumer_); } CCfile.close(); - RT_Config_File_Entry RTentry; - RT_Config_File_Parser RTfile; + Consumer_Config_File_Entry RTentry; + Consumer_Config_File_Parser RTfile; RTfile.open (argv[2]); @@ -165,6 +160,6 @@ int main (int argc, char *argv[]) #endif /* DEBUGGING */ #if defined (ACE_TEMPLATES_REQUIRE_SPECIALIZATION) -template class File_Parser<CC_Config_File_Entry>; -template class File_Parser<RT_Config_File_Entry>; +template class File_Parser<Connection_Config_File_Entry>; +template class File_Parser<Consumer_Config_File_Entry>; #endif /* ACE_TEMPLATES_REQUIRE_SPECIALIZATION */ diff --git a/apps/Gateway/Gateway/Config_Files.h b/apps/Gateway/Gateway/Config_Files.h index 9418815ff09..145c3233bae 100644 --- a/apps/Gateway/Gateway/Config_Files.h +++ b/apps/Gateway/Gateway/Config_Files.h @@ -1,7 +1,6 @@ /* -*- C++ -*- */ // $Id$ - // ============================================================================ // // = LIBRARY @@ -21,42 +20,42 @@ #include "ace/OS.h" #include "File_Parser.h" -class CC_Config_File_Entry +class Connection_Config_File_Entry // = TITLE - // Stores the information in a Channel Connection entry. + // Stores the IO_Handler entry for connection configuration. { public: int conn_id_; - // Connection id for this Channel. + // Connection id for this IO_Handler. char host_[BUFSIZ]; // Host to connect with. - u_short remote_port_; + u_short remote_poconsumer_; // Port to connect with. char direction_; - // 'I' (input) or 'O' (output) + // 'S' (supplier) or 'C' (consumer). int max_retry_delay_; // Maximum amount of time to wait for reconnecting. - u_short local_port_; + u_short local_poconsumer_; // Our local port number. }; -class CC_Config_File_Parser : public File_Parser<CC_Config_File_Entry> +class Connection_Config_File_Parser : public File_Parser<Connection_Config_File_Entry> // = TITLE - // Parser for the Channel Connection file. + // Parser for the IO_Handler Connection file. { public: virtual FP::Return_Type - read_entry (CC_Config_File_Entry &entry, int &line_number); + read_entry (Connection_Config_File_Entry &entry, int &line_number); }; -class RT_Config_File_Entry +class Consumer_Config_File_Entry // = TITLE - // Stores the information in a Routing Table entry. + // Stores the information in a Consumer Map entry. { public: enum { @@ -79,13 +78,13 @@ public: // Total number of these destinations. }; -class RT_Config_File_Parser : public File_Parser<RT_Config_File_Entry> +class Consumer_Config_File_Parser : public File_Parser<Consumer_Config_File_Entry> // = TITLE - // Parser for the Routing Table file. + // Parser for the Consumer Map file. { public: virtual FP::Return_Type - read_entry (RT_Config_File_Entry &entry, int &line_number); + read_entry (Consumer_Config_File_Entry &entry, int &line_number); }; #endif /* _CONFIG_FILES */ diff --git a/apps/Gateway/Gateway/Consumer_Entry.cpp b/apps/Gateway/Gateway/Consumer_Entry.cpp new file mode 100644 index 00000000000..c3dcd96ebbf --- /dev/null +++ b/apps/Gateway/Gateway/Consumer_Entry.cpp @@ -0,0 +1,31 @@ +// Defines an entry in the Consumer Map. +// $Id$ + +#include "Consumer_Entry.h" + +Consumer_Entry::Consumer_Entry (void) +{ + ACE_NEW (this->destinations_, Consumer_Entry::ENTRY_SET); +} + +Consumer_Entry::~Consumer_Entry (void) +{ + delete this->destinations_; +} + +// Get the associated set of destinations. + +Consumer_Entry::ENTRY_SET * +Consumer_Entry::destinations (void) +{ + return this->destinations_; +} + +// Set the associated set of destinations. + +void +Consumer_Entry::destinations (Consumer_Entry::ENTRY_SET *s) +{ + this->destinations_ = s; +} + diff --git a/apps/Gateway/Gateway/Consumer_Entry.h b/apps/Gateway/Gateway/Consumer_Entry.h new file mode 100644 index 00000000000..fe502991514 --- /dev/null +++ b/apps/Gateway/Gateway/Consumer_Entry.h @@ -0,0 +1,45 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Consumer_Entry.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_ROUTING_ENTRY) +#define _ROUTING_ENTRY + +#include "ace/Set.h" + +// Forward reference. +class IO_Handler; + +class Consumer_Entry +{ + // = TITLE + // Defines an entry in the Consumer_Map. +public: + Consumer_Entry (void); + ~Consumer_Entry (void); + + typedef ACE_Unbounded_Set<IO_Handler *> ENTRY_SET; + typedef ACE_Unbounded_Set_Iterator<IO_Handler *> ENTRY_ITERATOR; + + // = Set/get the associated set of destinations. + ENTRY_SET *destinations (void); + void destinations (ENTRY_SET *); + +protected: + ENTRY_SET *destinations_; + // The set of destinations; +}; + +#endif /* _ROUTING_ENTRY */ diff --git a/apps/Gateway/Gateway/Consumer_Map.cpp b/apps/Gateway/Gateway/Consumer_Map.cpp new file mode 100644 index 00000000000..6d16601f949 --- /dev/null +++ b/apps/Gateway/Gateway/Consumer_Map.cpp @@ -0,0 +1,61 @@ +/* -*- C++ -*- */ +// $Id$ + +#if !defined (_CONSUMER_MAP_C) +#define _CONSUMER_MAP_C + +#include "Consumer_Map.h" + +// Bind the Event_Addr to the INT_ID. + +int +Consumer_Map::bind (Event_Addr event_addr, + Consumer_Entry *Consumer_Entry) +{ + return this->map_.bind (event_addr, Consumer_Entry); +} + +// Find the Consumer_Entry corresponding to the Event_Addr. + +int +Consumer_Map::find (Event_Addr event_addr, + Consumer_Entry *&Consumer_Entry) +{ + return this->map_.find (event_addr, Consumer_Entry); +} + +// Unbind (remove) the Event_Addr from the map. + +int +Consumer_Map::unbind (Event_Addr event_addr) +{ + return this->map_.unbind (event_addr); +} + +Consumer_Map_Iterator::Consumer_Map_Iterator (Consumer_Map &rt) + : map_iter_ (rt.map_) +{ +} + +int +Consumer_Map_Iterator::next (Consumer_Entry *&ss) +{ + // Loop in order to skip over inactive entries if necessary. + + for (ACE_Map_Entry<Event_Addr, Consumer_Entry *> *temp = 0; + this->map_iter_.next (temp) != 0; + this->advance ()) + { + // Otherwise, return the next item. + ss = temp->int_id_; + return 1; + } + return 0; +} + +int +Consumer_Map_Iterator::advance (void) +{ + return this->map_iter_.advance (); +} +#endif /* _CONSUMER_MAP_C */ diff --git a/apps/Gateway/Gateway/Consumer_Map.h b/apps/Gateway/Gateway/Consumer_Map.h new file mode 100644 index 00000000000..fd392afaf6e --- /dev/null +++ b/apps/Gateway/Gateway/Consumer_Map.h @@ -0,0 +1,62 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Consumer_Map.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_CONSUMER_MAP_H) +#define _CONSUMER_MAP_H + +#include "ace/Map_Manager.h" +#include "Concurrency_Strategies.h" +#include "Event.h" +#include "Consumer_Entry.h" + +class Consumer_Map +{ + // = TITLE + // Define a generic consumer map based on the ACE Map_Manager. + // + // = DESCRIPTION + // This class makes it easier to use the Map_Manager. +public: + int bind (Event_Addr event, Consumer_Entry *Consumer_Entry); + // Associate Event with the Consumer_Entry. + + int find (Event_Addr event, Consumer_Entry *&Consumer_Entry); + // Break any association of EXID. + + int unbind (Event_Addr event); + // Locate EXID and pass out parameter via INID. If found, + // return 0, else -1. + +public: + ACE_Map_Manager<Event_Addr, Consumer_Entry *, MAP_MUTEX> map_; + // Map that associates Event Addrs (external ids) with Consumer_Entry *'s + // <internal IDs>. +}; + +class Consumer_Map_Iterator +{ + // = TITLE + // Define an iterator for the Consumer Map. +public: + Consumer_Map_Iterator (Consumer_Map &mm); + int next (Consumer_Entry *&); + int advance (void); + +private: + ACE_Map_Iterator<Event_Addr, Consumer_Entry *, MAP_MUTEX> map_iter_; + // Map we are iterating over. +}; +#endif /* _CONSUMER_MAP_H */ diff --git a/apps/Gateway/Gateway/Event.h b/apps/Gateway/Gateway/Event.h new file mode 100644 index 00000000000..a8a9374be3c --- /dev/null +++ b/apps/Gateway/Gateway/Event.h @@ -0,0 +1,86 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Event.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (EVENT) +#define EVENT + +// This is the unique connection identifier that denotes a particular +// IO_Handler in the Gateway. +typedef short CONN_ID; + +class Event_Addr + // = TITLE + // Address used to identify the source/destination of an event. +{ +public: + Event_Addr (CONN_ID cid = -1, unsigned char lid = 0, unsigned char pay = 0) + : conn_id_ (cid), logical_id_ (lid), payload_ (pay) {} + + int operator== (const Event_Addr &pa) const + { + return this->conn_id_ == pa.conn_id_ + && this->logical_id_ == pa.logical_id_ + && this->payload_ == pa.payload_; + } + + CONN_ID conn_id_; + // Unique connection identifier that denotes a particular IO_Handler. + + unsigned char logical_id_; + // Logical ID. + + unsigned char payload_; + // Payload type. +}; + + +class Event_Header + // = TITLE + // Fixed sized header. +{ +public: + typedef unsigned short SUPPLIER_ID; + // Type used to route messages from gatewayd. + + enum + { + INVALID_ID = -1 // No peer can validly use this number. + }; + + SUPPLIER_ID routing_id_; + // Source ID. + + size_t len_; + // Length of the message in bytes. +}; + +class Event + // = TITLE + // Variable-sized message (buf_ may be variable-sized between + // 0 and MAX_PAYLOAD_SIZE). +{ +public: + enum { MAX_PAYLOAD_SIZE = 1024 }; + // The maximum size of an Event. + + Event_Header header_; + // Message header. + + char buf_[MAX_PAYLOAD_SIZE]; + // Message payload. +}; + +#endif /* EVENT */ diff --git a/apps/Gateway/Gateway/Event_Channel.h b/apps/Gateway/Gateway/Event_Channel.h new file mode 100644 index 00000000000..47dd8572012 --- /dev/null +++ b/apps/Gateway/Gateway/Event_Channel.h @@ -0,0 +1,99 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Event_Channel.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (ACE_EVENT_CHANNEL) +#define ACE_EVENT_CHANNEL + +#include "IO_Handler_Connector.h" + +template <class SUPPLIER_HANDLER, class CONSUMER_HANDLER> +class ACE_Svc_Export ACE_Event_Channel : public ACE_Event_Handler + // = TITLE + // Define a generic Event_Channel. +{ +public: + // = Initialization and termination methods. + ACE_Event_Channel (void); + + int open (int argc, char *argv[]); + // Initialize the Channel. + + int close (void); + // Close down the Channel. + +private: + int parse_args (int argc, char *argv[]); + // Parse the command-line arguments. + + int parse_connection_config_file (void); + // Parse the connection configuration file. + + int parse_consumer_config_file (void); + // Parse the consumer map configuration file. + + int initiate_connections (void); + // Initiate connections to the peers. + + virtual int handle_timeout (const ACE_Time_Value &, const void *arg); + // Perform timer-based performance profiling. + + const char *connection_config_file_; + // Name of the connection configuration file. + + const char *consumer_config_file_; + // Name of the consumer map configuration file. + + int active_connector_role_; + // Enabled if we are playing the role of the active Connector. + + int performance_window_; + // Number of seconds after connection establishment to report + // throughput. + + int blocking_semantics_; + // 0 == blocking connects, ACE_NONBLOCK == non-blocking connects. + + int debug_; + // Are we debugging? + + IO_Handler_Connector *connector_; + // This is used to establish the connections actively. + + int socket_queue_size_; + // Size of the socket queue (0 means "use default"). + + // = Make life easier by defining typedefs. + typedef ACE_Map_Manager<CONN_ID, IO_Handler *, MAP_MUTEX> CONNECTION_MAP; + typedef ACE_Map_Iterator<CONN_ID, IO_Handler *, MAP_MUTEX> CONNECTION_MAP_ITERATOR; + typedef ACE_Map_Entry<CONN_ID, IO_Handler *> CONNECTION_MAP_ENTRY; + + CONNECTION_MAP connection_map_; + // Table that maps Connection IDs to IO_Handler *'s. + + Consumer_Map consumer_map_; + // Map that associates event addresses to a set of Consumer_Handler + // *'s. +}; + +#if defined (ACE_TEMPLATES_REQUIRE_SOURCE) +#include "ace/Event_Channel.cpp" +#endif /* ACE_TEMPLATES_REQUIRE_SOURCE */ + +#if defined (ACE_TEMPLATES_REQUIRE_PRAGMA) +#pragma implementation ("Event_Channel.cpp") +#endif /* ACE_TEMPLATES_REQUIRE_PRAGMA */ + +#endif /* ACE_EVENT_CHANNEL */ diff --git a/apps/Gateway/Gateway/File_Parser.h b/apps/Gateway/Gateway/File_Parser.h index e36ae71ca94..776d1b2f338 100644 --- a/apps/Gateway/Gateway/File_Parser.h +++ b/apps/Gateway/Gateway/File_Parser.h @@ -1,7 +1,6 @@ /* -*- C++ -*- */ // $Id$ - // ============================================================================ // // = LIBRARY @@ -38,8 +37,8 @@ public: template <class ENTRY> class File_Parser // = TITLE - // Class used to parse the configuration file for the routing - // table. + // Class used to parse the configuration file for the Consumer + // Map. { public: // = Open and Close the file specified diff --git a/apps/Gateway/Gateway/Gateway.cpp b/apps/Gateway/Gateway/Gateway.cpp index f249eb2f37d..82666406070 100644 --- a/apps/Gateway/Gateway/Gateway.cpp +++ b/apps/Gateway/Gateway/Gateway.cpp @@ -1,315 +1,87 @@ /* -*- C++ -*- */ // $Id$ - - -#include "ace/Get_Opt.h" #include "ace/Service_Config.h" -#include "Config_Files.h" +#include "Event_Channel.h" #include "Gateway.h" -#include "Channel_Connector.h" -template <class INPUT_CHANNEL, class OUTPUT_CHANNEL> class Gateway : public ACE_Service_Object { public: - Gateway (ACE_Thread_Manager * = 0); + // = Initialization method. + Gateway (void); + // = Service configurator hooks. virtual int init (int argc, char *argv[]); // Perform initialization. virtual int fini (void); // Perform termination. + virtual int info (char **, size_t) const; + // Return info about this service. + protected: int handle_input (ACE_HANDLE); + // Shut down the Gateway when input comes in from the controlling + // console. int handle_signal (int signum, siginfo_t * = 0, ucontext_t * = 0); - - typedef ACE_Map_Manager<CONN_ID, Channel *, MUTEX> CONFIG_TABLE; - typedef ACE_Map_Iterator<CONN_ID, Channel *, MUTEX> CONFIG_ITERATOR; - - CONFIG_TABLE config_table_; - // Table that maps Connection IDs to Channel *'s. - - ROUTING_TABLE routing_table_; - // Table that maps Peer addresses to a set of Channel *'s for output. - - virtual int info (char **, size_t) const; - // Return info about this service. + // Shut down the Gateway when a signal arrives. int parse_args (int argc, char *argv[]); - // Parse gateway configuration arguments obtained from svc.conf file. - - int parse_cc_config_file (void); - // Parse the channel connection configuration file. - - int parse_rt_config_file (void); - // Parse the routing table configuration file. - - int initiate_connections (void); - // Initiate connections to the peers. - - virtual int handle_timeout (const ACE_Time_Value &, const void *arg); - // Perform timer-based performance profiling. - - const char *cc_config_file_; - // Name of the channel connection configuration file. + // Parse gateway configuration arguments obtained from svc.conf + // file. - const char *rt_config_file_; - // Name of the routing table configuration file. - - int performance_window_; - // Number of seconds after connection establishment to report throughput. - - int blocking_semantics_; - // 0 == blocking connects, ACE_NONBLOCK == non-blocking connects. - - int debug_; - // Are we debugging? - - Channel_Connector *connector_; - // This is used to establish the connections actively. - - int socket_queue_size_; - // Size of the socket queue (0 means "use default"). - - // = Manage output and input channel threads (if used.) - // if both input and output mt is used, they will share thr_mgr_, - // thr_mgr_ will always reference the thread manager being used - // regardless of whether input, output, or both channels are using mt. - ACE_Thread_Manager *thr_mgr_; - ACE_Thread_Manager *input_thr_mgr_; - ACE_Thread_Manager *output_thr_mgr_; + ACE_Event_Channel<SUPPLIER_HANDLER, CONSUMER_HANDLER> event_channel_; }; // Convenient shorthands. +// #define IC SUPPLIER_HANDLER +// #define OC CONSUMER_HANDLER -#define IC INPUT_CHANNEL -#define OC OUTPUT_CHANNEL - -template <class IC, class OC> int -Gateway<IC, OC>::handle_signal (int signum, siginfo_t *, ucontext_t *) +int +Gateway::handle_signal (int signum, siginfo_t *, ucontext_t *) { if (signum > 0) ACE_DEBUG ((LM_DEBUG, "(%t) %S\n", signum)); - if (this->thr_mgr_ != 0) - { -#if defined (ACE_HAS_THREADS) - ACE_DEBUG ((LM_DEBUG, "(%t) suspending all threads\n")); - if (this->thr_mgr_->suspend_all () == -1) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "suspend_all"), -1); -#endif /* ACE_HAS_THREADS */ - } + this->event_channel_.close (); // Shut down the main event loop. ACE_Service_Config::end_reactor_event_loop (); return 0; } -template <class IC, class OC> int -Gateway<IC, OC>::handle_input (ACE_HANDLE h) +int +Gateway::handle_input (ACE_HANDLE h) { - if (ACE_Service_Config::reactor ()->remove_handler (0, - ACE_Event_Handler::READ_MASK - | ACE_Event_Handler::DONT_CALL) == -1) + if (ACE_Service_Config::reactor ()->remove_handler + (0, ACE_Event_Handler::READ_MASK | ACE_Event_Handler::DONT_CALL) == -1) ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "remove_handler"), -1); + char buf[BUFSIZ]; // Consume the input... ACE_OS::read (h, buf, sizeof (buf)); - return this->handle_signal (h); -} -template <class IC, class OC> int -Gateway<IC, OC>::handle_timeout (const ACE_Time_Value &, const void *) -{ - ACE_DEBUG ((LM_DEBUG, "(%t) doing the performance timeout here...\n")); - CONFIG_ITERATOR cti (this->config_table_); - - // If we've got a ACE_Thread Manager then use it to suspend all - // the threads. This will enable us to get an accurate count. - - if (this->thr_mgr_ != 0) - { -#if defined (ACE_HAS_THREADS) - if (this->thr_mgr_->suspend_all () == -1) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "suspend_all"), -1); - ACE_DEBUG ((LM_DEBUG, "(%t) suspending all threads...")); -#endif /* ACE_HAS_THREADS */ - } - - size_t total_bytes_in = 0; - size_t total_bytes_out = 0; - - // Iterate through the routing table connecting all the channels. - - for (ACE_Map_Entry <CONN_ID, Channel *> *me = 0; - cti.next (me) != 0; - cti.advance ()) - { - Channel *channel = me->int_id_; - if (channel->direction () == 'O') - total_bytes_out += channel->total_bytes (); - else - total_bytes_in += channel->total_bytes (); - } - -#if defined (ACE_NLOGGING) - ACE_OS::fprintf (stderr, "After %d seconds, \ntotal_bytes_in = %d\ntotal_bytes_out = %d\n", - performance_window_, - total_bytes_in, - total_bytes_out); - - ACE_OS::fprintf (stderr, "%f Mbits/sec received.\n", - (float) (total_bytes_in * 8 / (float) (1024*1024*this->performance_window_))); - - ACE_OS::fprintf (stderr, "%f Mbits/sec sent.\n", - (float) (total_bytes_out * 8 / (float) (1024*1024*this->performance_window_))); -#else - ACE_DEBUG ((LM_DEBUG, "(%t) after %d seconds, \ntotal_bytes_in = %d\ntotal_bytes_out = %d\n", - this->performance_window_, - total_bytes_in, - total_bytes_out)); - ACE_DEBUG ((LM_DEBUG, "(%t) %f Mbits/sec received.\n", - (float) (total_bytes_in * 8 / (float) (1024*1024*this->performance_window_)))); - ACE_DEBUG ((LM_DEBUG, "(%t) %f Mbits/sec sent.\n", - (float) (total_bytes_out * 8 / (float) (1024*1024*this->performance_window_)))); -#endif /* ACE_NLOGGING */ - // Resume all the threads again. - if (this->thr_mgr_ != 0) - { -#if defined (ACE_HAS_THREADS) - this->thr_mgr_->resume_all (); - ACE_DEBUG ((LM_DEBUG, "(%t) resuming all threads...")); -#endif /* ACE_HAS_THREADS */ - } - return 0; + // Shut us down. + return this->handle_signal (h); } // Give default values to data members. -template <class IC, class OC> -Gateway<IC, OC>::Gateway (ACE_Thread_Manager *thr_mgr) - : cc_config_file_ ("cc_config"), - rt_config_file_ ("rt_config"), - performance_window_ (0), - blocking_semantics_ (ACE_NONBLOCK), - debug_ (0), - connector_ (0), - socket_queue_size_ (0), - thr_mgr_ (thr_mgr), - input_thr_mgr_ (thr_mgr), - output_thr_mgr_ (thr_mgr) -{ -} -// Parse the "command-line" arguments and set the corresponding flags. - -template <class IC, class OC> int -Gateway<IC, OC>::parse_args (int argc, char *argv[]) -{ - ACE_Get_Opt get_opt (argc, argv, "bc:dr:q:w:", 0); - - for (int c; (c = get_opt ()) != -1; ) - { - switch (c) - { - case 'b': // Use blocking connection establishment. - this->blocking_semantics_ = 0; - break; - case 'c': - this->cc_config_file_ = get_opt.optarg; - break; - case 'd': - this->debug_ = 1; - break; - case 'q': - this->socket_queue_size_ = ACE_OS::atoi (get_opt.optarg); - break; - case 'r': - this->rt_config_file_ = get_opt.optarg; - break; - case 'w': // Time performance for a designated amount of time. - this->performance_window_ = ACE_OS::atoi (get_opt.optarg); - // Use blocking connection semantics so that we get accurate - // timings (since all connections start at once). - this->blocking_semantics_ = 0; - break; - default: - break; - } - } - return 0; -} - -// Initiate connections with the peers. - -template <class IC, class OC> int -Gateway<IC, OC>::initiate_connections (void) -{ - CONFIG_ITERATOR cti (this->config_table_); - - // Iterate through the routing table connecting all the channels. - - for (ACE_Map_Entry <CONN_ID, Channel *> *me = 0; - cti.next (me) != 0; - cti.advance ()) - { - Channel *channel = me->int_id_; - if (this->connector_->initiate_connection - (channel, this->blocking_semantics_ == ACE_NONBLOCK - ? ACE_Synch_Options::asynch : ACE_Synch_Options::synch) == -1) - continue; - } - - return 0; -} - -// This method is automatically called when the gateway -// is shutdown. It gracefully shuts down all the Channels -// in the Channel connection Config_Table. - -template <class IC, class OC> int -Gateway<IC, OC>::fini (void) +Gateway::Gateway (void) { - // Question: do we need to do anything special about the Routing_Table? - - CONFIG_ITERATOR cti (this->config_table_); - - for (ACE_Map_Entry <CONN_ID, Channel *> *me; - cti.next (me) != 0; - cti.advance ()) - { - Channel *channel = me->int_id_; - ACE_DEBUG ((LM_DEBUG, "(%t) closing down route %d\n", - channel->id ())); - if (channel->state () != Channel::IDLE) - // Mark channel as DISCONNECTING so we don't try to reconnect... - channel->state (Channel::DISCONNECTING); - - // Deallocate Channel resources. - channel->destroy (); // Will trigger a delete. - } - - // Free up the resources allocated dynamically by the ACE_Connector. - delete this->connector_; - delete this->thr_mgr_; - - return 0; } -template <class IC, class OC> int -Gateway<IC, OC>::init (int argc, char *argv[]) +int +Gateway::init (int argc, char *argv[]) { - this->parse_args (argc, argv); - - ACE_NEW_RETURN (this->connector_, Channel_Connector (), -1); - - if (this->connector_ == 0) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "out of memory"), -1); + if (this->event_channel_.open (argc, argv) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "open"), -1); - // Ignore SIPPIPE so each Output_Channel can handle it. + // Ignore SIPPIPE so each Consumer_Handler can handle it. ACE_Sig_Action sig (ACE_SignalHandler (SIG_IGN), SIGPIPE); ACE_Sig_Set sig_set; @@ -327,46 +99,23 @@ Gateway<IC, OC>::init (int argc, char *argv[]) this, ACE_Event_Handler::READ_MASK) == -1) ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "register_handler"), -1); + return 0; +} - if (this->thr_mgr_ == 0) - // Create a thread manager if using some combination of multi-threaded channels. -#if defined (USE_OUTPUT_MT) && defined (USE_INPUT_MT) - this->thr_mgr_ = this->output_thr_mgr_ = - this->input_thr_mgr_ = ACE_Service_Config::thr_mgr (); -#elif defined (USE_OUTPUT_MT) - this->thr_mgr_ = this->output_thr_mgr_ = ACE_Service_Config::thr_mgr (); -#elif defined (USE_INPUT_MT) - this->thr_mgr_ = this->input_thr_mgr_ = ACE_Service_Config::thr_mgr (); -#endif - - // Parse the connection configuration file. - this->parse_cc_config_file (); - - // Parse the routing table config file and build the routing table. - this->parse_rt_config_file (); - - // Initiate connections with the peers. - this->initiate_connections (); - - // If this->performance_window_ > 0 start a timer. - - if (this->performance_window_ > 0) - { - if (ACE_Service_Config::reactor ()->schedule_timer (this, 0, - this->performance_window_) == -1) - ACE_ERROR ((LM_ERROR, "(%t) %p\n", "schedule_timer")); - else - ACE_DEBUG ((LM_DEBUG, "starting timer for %d seconds...\n", - this->performance_window_)); - } +// This method is automatically called when the gateway is shutdown. +// It closes down the Event Channel. +int +Gateway::fini (void) +{ + this->event_channel_.close (); return 0; } // Returns information on the currently active service. -template <class IC, class OC> int -Gateway<IC, OC>::info (char **strp, size_t length) const +int +Gateway::info (char **strp, size_t length) const { char buf[BUFSIZ]; @@ -380,182 +129,7 @@ Gateway<IC, OC>::info (char **strp, size_t length) const return ACE_OS::strlen (buf); } -// Parse and build the connection table. - -template <class IC, class OC> int -Gateway<IC, OC>::parse_cc_config_file (void) -{ - // File that contains the routing table configuration information. - CC_Config_File_Parser cc_file; - CC_Config_File_Entry entry; - int file_empty = 1; - int line_number = 0; - - if (cc_file.open (this->cc_config_file_) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", this->cc_config_file_), -1); - - // Read config file line at a time. - while (cc_file.read_entry (entry, line_number) != FP::EOFILE) - { - file_empty = 0; - - if (this->debug_) - ACE_DEBUG ((LM_DEBUG, "(%t) conn id = %d, host = %s, remote port = %d, " - "direction = %c, max retry timeout = %d, local port = %d\n", - entry.conn_id_, entry.host_, entry.remote_port_, entry.direction_, - entry.max_retry_delay_, entry.local_port_)); - - Channel *channel = 0; - - // The next few lines of code are dependent on whether we are making - // an Input_Channel or an Output_Channel. - - if (entry.direction_ == 'O') // Configure an output channel. - ACE_NEW_RETURN (channel, - OUTPUT_CHANNEL (&this->routing_table_, - this->connector_, - this->output_thr_mgr_, - this->socket_queue_size_), - -1); - else /* direction == 'I' */ // Configure an input channel. - ACE_NEW_RETURN (channel, - INPUT_CHANNEL (&this->routing_table_, - this->connector_, - this->input_thr_mgr_, - this->socket_queue_size_), - -1); - if (channel == 0) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) out of memory\n"), -1); - - // The following code is common to both Input_ and Output_Channels. - - // Initialize the routing entry's peer addressing info. - channel->bind (ACE_INET_Addr (entry.remote_port_, entry.host_), - ACE_INET_Addr (entry.local_port_), entry.conn_id_); - - // Initialize max timeout. - channel->max_timeout (entry.max_retry_delay_); - - // Try to bind the new Channel to the connection ID. - switch (this->config_table_.bind (entry.conn_id_, channel)) - { - case -1: - ACE_ERROR_RETURN ((LM_ERROR, "(%t) bind failed for connection %d\n", - entry.conn_id_), -1); - /* NOTREACHED */ - case 1: // Oops, found a duplicate! - ACE_DEBUG ((LM_DEBUG, "(%t) duplicate connection %d, already bound\n", - entry.conn_id_)); - break; - case 0: - // Success. - break; - } - } - - if (file_empty) - ACE_ERROR ((LM_WARNING, - "warning: connection channel configuration file was empty\n")); - - return 0; -} - -template <class IC, class OC> int -Gateway<IC, OC>::parse_rt_config_file (void) -{ - // File that contains the routing table configuration information. - RT_Config_File_Parser rt_file; - RT_Config_File_Entry entry; - int file_empty = 1; - int line_number = 0; - - if (rt_file.open (this->rt_config_file_) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", this->rt_config_file_), -1); - - // Read config file line at a time. - while (rt_file.read_entry (entry, line_number) != FP::EOFILE) - { - file_empty = 0; - - if (this->debug_) - { - ACE_DEBUG ((LM_DEBUG, "(%t) conn id = %d, logical id = %d, payload = %d, " - "number of destinations = %d\n", - entry.conn_id_, entry.logical_id_, entry.payload_type_, - entry.total_destinations_)); - for (int i = 0; i < entry.total_destinations_; i++) - ACE_DEBUG ((LM_DEBUG, "(%t) destination[%d] = %d\n", - i, entry.destinations_[i])); - } - - Routing_Entry *re; - ACE_NEW_RETURN (re, Routing_Entry, -1); - Routing_Entry::ENTRY_SET *channel_set = new Routing_Entry::ENTRY_SET; - Peer_Addr peer_addr (entry.conn_id_, entry.logical_id_, - entry.payload_type_); - - // Add the destinations to the Routing Entry. - for (int i = 0; i < entry.total_destinations_; i++) - { - Channel *channel = 0; - - // Lookup destination and add to Routing_Entry set if found. - if (this->config_table_.find (entry.destinations_[i], - channel) != -1) - channel_set->insert (channel); - else - ACE_ERROR ((LM_ERROR, "(%t) not found: destination[%d] = %d\n", - i, entry.destinations_[i])); - } - - // Attach set of destination channels to routing entry. - re->destinations (channel_set); - - // Bind with routing table, keyed by peer address. - switch (this->routing_table_.bind (peer_addr, re)) - { - case -1: - ACE_ERROR_RETURN ((LM_ERROR, "(%t) bind failed for connection %d\n", - entry.conn_id_), -1); - /* NOTREACHED */ - case 1: // Oops, found a duplicate! - ACE_DEBUG ((LM_DEBUG, "(%t) duplicate routing table entry %d, " - "already bound\n", entry.conn_id_)); - break; - case 0: - // Success. - break; - } - } - - if (file_empty) - ACE_ERROR ((LM_WARNING, - "warning: routing table configuration file was empty\n")); - - return 0; -} - -#if defined (ACE_HAS_THREADS) && (defined (USE_OUTPUT_MT) || defined (USE_INPUT_MT)) -#if defined (USE_OUTPUT_MT) -typedef Thr_Output_Channel OUTPUT_CHANNEL; -#else -typedef Output_Channel OUTPUT_CHANNEL; -#endif /* USE_OUTPUT_MT */ - -#if defined (USE_INPUT_MT) -typedef Thr_Input_Channel INPUT_CHANNEL; -#else -typedef Input_Channel INPUT_CHANNEL; -#endif /* USE_INPUT_MT */ -#else -// Instantiate a non-multi-threaded Gateway. -typedef Input_Channel INPUT_CHANNEL; -typedef Output_Channel OUTPUT_CHANNEL; -#endif /* ACE_HAS_THREADS */ - -typedef Gateway<INPUT_CHANNEL, OUTPUT_CHANNEL> ACE_Gateway; - // The following is a "Factory" used by the ACE_Service_Config and // svc.conf file to dynamically initialize the state of the Gateway. -ACE_SVC_FACTORY_DEFINE (ACE_Gateway) +ACE_SVC_FACTORY_DEFINE (Gateway) diff --git a/apps/Gateway/Gateway/Gateway.h b/apps/Gateway/Gateway/Gateway.h index b00d87628de..057ce981701 100644 --- a/apps/Gateway/Gateway/Gateway.h +++ b/apps/Gateway/Gateway/Gateway.h @@ -1,7 +1,6 @@ /* -*- C++ -*- */ // $Id$ - // ============================================================================ // // = LIBRARY @@ -20,11 +19,7 @@ #include "ace/OS.h" -ACE_SVC_FACTORY_DECLARE (ACE_Gateway) +ACE_SVC_FACTORY_DECLARE (Gateway) #endif /* ACE_GATEWAY */ - - - - diff --git a/apps/Gateway/Gateway/IO_Handler.cpp b/apps/Gateway/Gateway/IO_Handler.cpp new file mode 100644 index 00000000000..94997955979 --- /dev/null +++ b/apps/Gateway/Gateway/IO_Handler.cpp @@ -0,0 +1,710 @@ +// $Id$ + +#include "Consumer_Entry.h" +#include "IO_Handler_Connector.h" + +// Convenient short-hands. +#define CO CONDITION +#define MU MAP_MUTEX + +// The total number of bytes sent/received on this channel. + +size_t +IO_Handler::total_bytes (void) +{ + return this->total_bytes_; +} + +void +IO_Handler::total_bytes (size_t bytes) +{ + this->total_bytes_ += bytes; +} + +IO_Handler::IO_Handler (Consumer_Map *consumer_map, + IO_Handler_Connector *ioc, + ACE_Thread_Manager *thr_mgr, + int socket_queue_size) + : ACE_Svc_Handler<ACE_SOCK_STREAM, SYNCH_STRATEGY> (thr_mgr), + consumer_map_ (consumer_map), + id_ (-1), + total_bytes_ (0), + state_ (IO_Handler::IDLE), + connector_ (ioc), + timeout_ (1), + max_timeout_ (IO_Handler::MAX_RETRY_TIMEOUT), + socket_queue_size_ (socket_queue_size) +{ +} + +// Set the associated channel. + +void +IO_Handler::active (int a) +{ + this->state (a == 0 ? IO_Handler::IDLE : IO_Handler::ESTABLISHED); +} + +// Get the associated channel. + +int +IO_Handler::active (void) +{ + return this->state () == IO_Handler::ESTABLISHED; +} + +// Set the direction. + +void +IO_Handler::direction (char d) +{ + this->direction_ = d; +} + +// Get the direction. + +char +IO_Handler::direction (void) +{ + return this->direction_; +} + +// Sets the timeout delay. + +void +IO_Handler::timeout (int to) +{ + if (to > this->max_timeout_) + to = this->max_timeout_; + + this->timeout_ = to; +} + +// Recalculate the current retry timeout delay using exponential +// backoff. Returns the original timeout (i.e., before the +// recalculation). + +int +IO_Handler::timeout (void) +{ + int old_timeout = this->timeout_; + this->timeout_ *= 2; + + if (this->timeout_ > this->max_timeout_) + this->timeout_ = this->max_timeout_; + + return old_timeout; +} + +// Sets the max timeout delay. + +void +IO_Handler::max_timeout (int mto) +{ + this->max_timeout_ = mto; +} + +// Gets the max timeout delay. + +int +IO_Handler::max_timeout (void) +{ + return this->max_timeout_; +} + +// Restart connection asynchronously when timeout occurs. + +int +IO_Handler::handle_timeout (const ACE_Time_Value &, const void *) +{ + ACE_DEBUG ((LM_DEBUG, + "(%t) attempting to reconnect IO_Handler %d with timeout = %d\n", + this->id (), this->timeout_)); + return this->connector_->initiate_connection (this, ACE_Synch_Options::asynch); +} + +// Restart connection (blocking_semantics dicates whether we +// restart synchronously or asynchronously). + +int +IO_Handler::reinitiate_connection (void) +{ + // Skip over deactivated descriptors. + if (this->get_handle () != -1) + { + // Make sure to close down peer to reclaim descriptor. + this->peer ().close (); + +#if 0 +// if (this->state () == FAILED) +// { + // Reinitiate timeout to improve reconnection time. +// this->timeout (1); +#endif + + ACE_DEBUG ((LM_DEBUG, + "(%t) scheduling reinitiation of IO_Handler %d\n", + this->id ())); + + // Reschedule ourselves to try and connect again. + if (ACE_Service_Config::reactor ()->schedule_timer + (this, 0, this->timeout ()) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "schedule_timer"), -1); + } + return 0; +} + +// Handle shutdown of the IO_Handler object. + +int +IO_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask) +{ + ACE_DEBUG ((LM_DEBUG, + "(%t) shutting down IO_Handler %d on handle %d\n", + this->id (), this->get_handle ())); + + return this->reinitiate_connection (); +} + +// Set the state of the channel. + +void +IO_Handler::state (IO_Handler::State s) +{ + this->state_ = s; +} + +// Perform the first-time initiation of a connection to the peer. + +int +IO_Handler::initialize_connection (void) +{ + this->state_ = IO_Handler::ESTABLISHED; + + // Restart the timeout to 1. + this->timeout (1); + +#if defined (ASSIGN_SUPPLIER_ID) + // Action that sends the route id to the peerd. + + CONN_ID id = htons (this->id ()); + + ssize_t n = this->peer ().send ((const void *) &id, sizeof id); + + if (n != sizeof id) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + n == 0 ? "gatewayd has closed down unexpectedly" : "send"), + -1); +#endif /* ASSIGN_SUPPLIER_ID */ + return 0; +} + +// Set the size of the socket queue. + +void +IO_Handler::socket_queue_size (void) +{ + if (this->socket_queue_size_ > 0) + { + int option = this->direction_ == 'S' ? SO_RCVBUF : SO_SNDBUF; + + if (this->peer ().set_option (SOL_SOCKET, option, + &this->socket_queue_size_, sizeof (int)) == -1) + ACE_ERROR ((LM_ERROR, "(%t) %p\n", "set_option")); + } +} + +// Upcall from the ACE_Acceptor::handle_input() that +// delegates control to our application-specific IO_Handler. + +int +IO_Handler::open (void *a) +{ + ACE_DEBUG ((LM_DEBUG, "(%t) IO_Handler's fd = %d\n", + this->peer ().get_handle ())); + + // Set the size of the socket queue. + this->socket_queue_size (); + + // Turn on non-blocking I/O. + if (this->peer ().enable (ACE_NONBLOCK) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "enable"), -1); + + // Call down to the base class to activate and register this handler. + if (this->ACE_Svc_Handler<ACE_SOCK_STREAM, SYNCH_STRATEGY>::open (a) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "activate"), -1); + + return this->initialize_connection (); +} + +// Return the current state of the channel. + +IO_Handler::State +IO_Handler::state (void) +{ + return this->state_; +} + +void +IO_Handler::id (CONN_ID id) +{ + this->id_ = id; +} + +CONN_ID +IO_Handler::id (void) +{ + return this->id_; +} + +// Set the peer's address information. +int +IO_Handler::bind (const ACE_INET_Addr &remote_addr, + const ACE_INET_Addr &local_addr, + CONN_ID id) +{ + this->remote_addr_ = remote_addr; + this->local_addr_ = local_addr; + this->id_ = id; + return 0; +} + +ACE_INET_Addr & +IO_Handler::remote_addr (void) +{ + return this->remote_addr_; +} + +ACE_INET_Addr & +IO_Handler::local_addr (void) +{ + return this->local_addr_; +} + +// Constructor sets the consumer map pointer. + +Consumer_Handler::Consumer_Handler (Consumer_Map *consumer_map, + IO_Handler_Connector *ioc, + ACE_Thread_Manager *thr_mgr, + int socket_queue_size) + : IO_Handler (consumer_map, ioc, thr_mgr, socket_queue_size) +{ + this->direction_ = 'C'; + this->msg_queue ()->high_water_mark (Consumer_Handler::QUEUE_SIZE); +} + +// This method should be called only when the peer shuts down +// unexpectedly. This method simply marks the IO_Handler as +// having failed so that handle_close () can reconnect. + +int +Consumer_Handler::handle_input (ACE_HANDLE) +{ + char buf[1]; + + this->state (IO_Handler::FAILED); + + switch (this->peer ().recv (buf, sizeof buf)) + { + case -1: + ACE_ERROR_RETURN ((LM_ERROR, + "(%t) Peer has failed unexpectedly for Output IO_Handler %d\n", + this->id ()), -1); + /* NOTREACHED */ + case 0: + ACE_ERROR_RETURN ((LM_ERROR, + "(%t) Peer has shutdown unexpectedly for Output IO_Handler %d\n", + this->id ()), -1); + /* NOTREACHED */ + default: + ACE_ERROR_RETURN ((LM_ERROR, + "(%t) Peer is sending input on Output IO_Handler %d\n", + this->id ()), -1); + /* NOTREACHED */ + } +} + +// Perform a non-blocking put() of event MB. If we are unable to +// send the entire event the remainder is re-queued at the *front* of +// the Event_List. + +int +Consumer_Handler::nonblk_put (ACE_Message_Block *mb) +{ + // Try to send the event. If we don't send it all (e.g., due to + // flow control), then re-queue the remainder at the head of the + // Event_List and ask the ACE_Reactor to inform us (via + // handle_output()) when it is possible to try again. + + ssize_t n; + + if ((n = this->send (mb)) == -1) + { + // Things have gone wrong, let's try to close down and set up a new reconnection. + this->state (IO_Handler::FAILED); + this->handle_close (); + return -1; + } + else if (errno == EWOULDBLOCK) // Didn't manage to send everything. + { + ACE_DEBUG ((LM_DEBUG, "(%t) queueing activated on handle %d to routing id %d\n", + this->get_handle (), this->id ())); + + // ACE_Queue in *front* of the list to preserve order. + if (this->msg_queue ()->enqueue_head + (mb, (ACE_Time_Value *) &ACE_Time_Value::zero) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "enqueue_head"), -1); + + // Tell ACE_Reactor to call us back when we can send again. + else if (ACE_Service_Config::reactor ()-> + schedule_wakeup (this, ACE_Event_Handler::WRITE_MASK) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "schedule_wakeup"), -1); + return 0; + } + else + return n; +} + +int +Consumer_Handler::send (ACE_Message_Block *mb) +{ + ssize_t n; + size_t len = mb->length (); + + if ((n = this->peer ().send (mb->rd_ptr (), len)) <= 0) + return errno == EWOULDBLOCK ? 0 : n; + else if (n < len) + // Re-adjust pointer to skip over the part we did send. + mb->rd_ptr (n); + else /* if (n == length) */ + { + // The whole event is sent, we can now safely deallocate the + // buffer. Note that this should decrement a reference count... + delete mb; + errno = 0; + } + this->total_bytes (n); + return n; +} + +// Finish sending an event when flow control conditions abate. +// This method is automatically called by the ACE_Reactor. + +int +Consumer_Handler::handle_output (ACE_HANDLE) +{ + ACE_Message_Block *mb = 0; + int status = 0; + + ACE_DEBUG ((LM_DEBUG, + "(%t) in handle_output on handle %d\n", + this->get_handle ())); + // The list had better not be empty, otherwise there's a bug! + + if (this->msg_queue ()->dequeue_head + (mb, (ACE_Time_Value *) &ACE_Time_Value::zero) != -1) + { + switch (this->nonblk_put (mb)) + { + case 0: // Partial send. + ACE_ASSERT (errno == EWOULDBLOCK); + // Didn't write everything this time, come back later... + break; + + case -1: + // Caller is responsible for freeing a ACE_Message_Block if failures occur. + delete mb; + ACE_ERROR ((LM_ERROR, "(%t) %p\n", "transmission failure")); + + /* FALLTHROUGH */ + default: // Sent the whole thing. + + // If we succeed in writing the entire event (or we did not + // fail due to EWOULDBLOCK) then check if there are more + // events on the Event_List. If there aren't, tell the + // ACE_Reactor not to notify us anymore (at least until + // there are new events queued up). + + if (this->msg_queue ()->is_empty ()) + { + ACE_DEBUG ((LM_DEBUG, + "(%t) queueing deactivated on handle %d to routing id %d\n", + this->get_handle (), this->id ())); + + + if (ACE_Service_Config::reactor ()-> + cancel_wakeup (this, ACE_Event_Handler::WRITE_MASK) == -1) + ACE_ERROR ((LM_ERROR, "(%t) %p\n", "cancel_wakeup")); + } + } + } + else + ACE_ERROR ((LM_ERROR, "(%t) %p\n", "dequeue_head")); + return 0; +} + +// Send an event to a peer (may queue if necessary). + +int +Consumer_Handler::put (ACE_Message_Block *mb, ACE_Time_Value *) +{ + if (this->msg_queue ()->is_empty ()) + // Try to send the event *without* blocking! + return this->nonblk_put (mb); + else + // If we have queued up events due to flow control then just + // enqueue and return. + return this->msg_queue ()->enqueue_tail + (mb, (ACE_Time_Value *) &ACE_Time_Value::zero); +} + +// Constructor sets the consumer map pointer and the connector +// pointer. + +Supplier_Handler::Supplier_Handler (Consumer_Map *consumer_map, + IO_Handler_Connector *ioc, + ACE_Thread_Manager *thr_mgr, + int socket_queue_size) + : msg_frag_ (0), + IO_Handler (consumer_map, ioc, thr_mgr, socket_queue_size) +{ + this->direction_ = 'S'; + this->msg_queue ()->high_water_mark (0); +} + +// Receive a Peer event from peerd. Handles fragmentation. +// +// The routing event returned from recv consists of two parts: +// 1. The Address part, contains the virtual routing id. +// 2. The Data part, which contains the actual data to be routed. +// +// The reason for having two parts is to shield the higher layers +// of software from knowledge of the event structure. + +int +Supplier_Handler::recv (ACE_Message_Block *&forward_addr) +{ + Event *event; + size_t len; + ssize_t n = 0; + ssize_t m = 0; + size_t offset = 0; + + if (this->msg_frag_ == 0) + // No existing fragment... + ACE_NEW_RETURN (this->msg_frag_, + ACE_Message_Block (sizeof (Event)), + -1); + + event = (Event *) this->msg_frag_->rd_ptr (); + + const ssize_t HEADER_SIZE = sizeof (Event_Header); + ssize_t header_bytes_left_to_read = HEADER_SIZE - this->msg_frag_->length (); + + if (header_bytes_left_to_read > 0) + { + n = this->peer ().recv (this->msg_frag_->wr_ptr (), + header_bytes_left_to_read); + + if (n == -1 /* error */ + || n == 0 /* EOF */) + { + ACE_ERROR ((LM_ERROR, "%p\n", + "Recv error during header read ")); + ACE_DEBUG ((LM_DEBUG, + "attempted to read %d\n", + header_bytes_left_to_read)); + delete this->msg_frag_; + this->msg_frag_ = 0; + return n; + } + + // Bump the write pointer by the amount read. + this->msg_frag_->wr_ptr (n); + + // At this point we may or may not have the ENTIRE header. + if (this->msg_frag_->length () < HEADER_SIZE) + { + ACE_DEBUG ((LM_DEBUG, + "Partial header received: only %d bytes\n", + this->msg_frag_->length ())); + // Notify the caller that we didn't get an entire event. + errno = EWOULDBLOCK; + return -1; + } + } + + // At this point there is a complete, valid header in msg_frag_ + len = sizeof event->buf_ + HEADER_SIZE - this->msg_frag_->length (); + + // Try to receive the remainder of the event + + switch (m = this->peer ().recv (event->buf_ + offset, len)) + { + case -1: + if (errno == EWOULDBLOCK) + { + // This shouldn't happen since the ACE_Reactor + // just triggered us to handle pending I/O! + ACE_DEBUG ((LM_DEBUG, "(%t) unexpected recv failure\n")); + errno = EWOULDBLOCK; + return -1; + } + else + /* FALLTHROUGH */; + + case 0: // Premature EOF. + delete this->msg_frag_; + this->msg_frag_ = 0; + return 0; + + default: + if (m != len) + // Re-adjust pointer to skip over the part we've read. + { + this->msg_frag_->wr_ptr (m); + errno = EWOULDBLOCK; + return -1; // Inform caller that we didn't get the whole event. + } + else + { + // Set the write pointer at 1 past the end of the event. + this->msg_frag_->wr_ptr (m); + + // Set the read pointer to the beginning of the event. + this->msg_frag_->rd_ptr (this->msg_frag_->base ()); + + // Allocate an event forwarding header and chain the data + // portion onto its continuation field. + ACE_NEW_RETURN (forward_addr, + ACE_Message_Block (sizeof (Event_Addr), + ACE_Message_Block::MB_PROTO, + this->msg_frag_), + -1); + + Event_Addr event_addr (this->id (), event->header_.routing_id_, 0); + // Copy the forwarding address from the Event_Addr into + // forward_addr. + forward_addr->copy ((char *) &event_addr, sizeof (Event)); + + // Reset the pointer to indicate we've got an entire event. + this->msg_frag_ = 0; + } + this->total_bytes (m + n); +#if defined (VERBOSE) + ACE_DEBUG ((LM_DEBUG, "(%t) channel id = %d, route id = %d, len = %d, payload = %*s", + event_addr.conn_id_, event->header_.routing_id_, event->header_.len_, + event->header_.len_, event->buf_)); +#else + ACE_DEBUG ((LM_DEBUG, "(%t) route id = %d, cur len = %d, total bytes read = %d\n", + event->header_.routing_id_, event->header_.len_, this->total_bytes ())); +#endif + return m + n; + } +} + +// Receive various types of input (e.g., Peer event from the +// gatewayd, as well as stdio). + +int +Supplier_Handler::handle_input (ACE_HANDLE) +{ + ACE_Message_Block *forward_addr = 0; + + switch (this->recv (forward_addr)) + { + case 0: + // Note that a peer should never initiate a shutdown. + this->state (IO_Handler::FAILED); + ACE_ERROR_RETURN ((LM_ERROR, + "(%t) Peer has closed down unexpectedly for Input IO_Handler %d\n", + this->id ()), -1); + /* NOTREACHED */ + case -1: + if (errno == EWOULDBLOCK) + // A short-read, we'll come back and finish it up later on! + return 0; + else // A weird problem occurred, shut down and start again. + { + this->state (IO_Handler::FAILED); + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p for Input IO_Handler %d\n", + "Peer has failed unexpectedly", + this->id ()), -1); + } + /* NOTREACHED */ + default: + return this->forward (forward_addr); + } +} + +// Route an event to its appropriate destination. + +int +Supplier_Handler::forward (ACE_Message_Block *forward_addr) +{ + // We got a valid event, so determine its virtual routing id, + // which is stored in the first of the two event blocks chained + // together. + + Event_Addr *forwarding_key = (Event_Addr *) forward_addr->rd_ptr (); + + // Skip over the address portion. + const ACE_Message_Block *const data = forward_addr->cont (); + + // RE points to the routing entry located for this routing id. + Consumer_Entry *re = 0; + + if (this->consumer_map_->find (*forwarding_key, re) != -1) + { + // Check to see if there are any destinations. + if (re->destinations ()->size () == 0) + ACE_DEBUG ((LM_WARNING, + "there are no active destinations for this event currently\n")); + + else // There are destinations, so forward the event. + { + Consumer_Entry::ENTRY_SET *esp = re->destinations (); + Consumer_Entry::ENTRY_ITERATOR si (*esp); + + for (IO_Handler **channel = 0; si.next (channel) != 0; si.advance ()) + { + // Only process active channels. + if ((*channel)->active ()) + { + // Clone the event portion (should be doing reference counting here...) + ACE_Message_Block *newmsg = data->clone (); + + ACE_DEBUG ((LM_DEBUG, "(%t) sending to peer %d\n", (*channel)->id ())); + + if ((*channel)->put (newmsg) == -1) + { + if (errno == EWOULDBLOCK) // The queue has filled up! + ACE_ERROR ((LM_ERROR, "(%t) %p\n", + "gateway is flow controlled, so we're dropping events")); + else + ACE_ERROR ((LM_ERROR, "(%t) %p transmission error to route %d\n", + "put", (*channel)->id ())); + + // Caller is responsible for freeing a ACE_Message_Block if failures occur. + delete newmsg; + } + } + } + // Will become superfluous once we have reference counting... + delete forward_addr; + return 0; + } + } + delete forward_addr; + // Failure return. + ACE_ERROR ((LM_DEBUG, "(%t) find failed on conn id = %d, logical id = %d, payload = %d\n", + forwarding_key->conn_id_, forwarding_key->logical_id_, forwarding_key->payload_)); + return 0; +} + +#if defined (ACE_TEMPLATES_REQUIRE_SPECIALIZATION) +template class ACE_Map_Manager<Event_Addr, Consumer_Entry *, MAP_MUTEX>; +template class ACE_Map_Iterator<Event_Addr, Consumer_Entry *, MAP_MUTEX>; +template class ACE_Map_Entry<Event_Addr, Consumer_Entry *>; +#endif /* ACE_TEMPLATES_REQUIRE_SPECIALIZATION */ diff --git a/apps/Gateway/Gateway/IO_Handler.h b/apps/Gateway/Gateway/IO_Handler.h new file mode 100644 index 00000000000..c22f1a3df26 --- /dev/null +++ b/apps/Gateway/Gateway/IO_Handler.h @@ -0,0 +1,224 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// IO_Handler.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_IO_HANDLER) +#define _IO_HANDLER + +#include "ace/Service_Config.h" +#include "ace/SOCK_Connector.h" +#include "ace/Svc_Handler.h" +#include "Consumer_Map.h" +#include "Consumer_Entry.h" +#include "Event.h" + +// Forward declaration. +class IO_Handler_Connector; + +class IO_Handler : public ACE_Svc_Handler<ACE_SOCK_STREAM, SYNCH_STRATEGY> + // = TITLE + // IO_Handler contains info about connection state and addressing. + // + // = DESCRIPTION + // The IO_Handler classes process events sent from the peers to the + // gateway. These classes works as follows: + // + // 1. IO_Handler_Connector creates a number of connections with the set of + // peers specified in a configuration file. + // + // 2. For each peer that connects successfully, IO_Handler_Connector + // creates an IO_Handler object. Each object assigns a unique routing + // id to its associated peer. The Handlers are used by gatewayd + // that to receive, route, and forward events from source peer(s) + // to destination peer(s). +{ +public: + IO_Handler (Consumer_Map *, + IO_Handler_Connector *, + ACE_Thread_Manager * = 0, + int socket_queue_size = 0); + + virtual int open (void * = 0); + // Initialize and activate a single-threaded IO_Handler (called by + // ACE_Connector::handle_output()). + + int bind (const ACE_INET_Addr &remote_addr, + const ACE_INET_Addr &local_addr, + CONN_ID); + // Set the peer's addressing and routing information. + + ACE_INET_Addr &remote_addr (void); + // Returns the peer's routing address. + + ACE_INET_Addr &local_addr (void); + // Returns our local address. + + // = Set/get routing id. + CONN_ID id (void); + void id (CONN_ID); + + // = Set/get the current state of the IO_Handler. + enum State + { + IDLE = 1, // Prior to initialization. + CONNECTING, // During connection establishment. + ESTABLISHED, // IO_Handler is established and active. + DISCONNECTING, // IO_Handler is in the process of connecting. + FAILED // IO_Handler has failed. + }; + + // = Set/get the current state. + State state (void); + void state (State); + + // = Set/get the current retry timeout delay. + int timeout (void); + void timeout (int); + + // = Set/get the maximum retry timeout delay. + int max_timeout (void); + void max_timeout (int); + + // = Set/get IO_Handler activity status. + int active (void); + void active (int); + + // = Set/get direction (necessary for error checking). + char direction (void); + void direction (char); + + // = The total number of bytes sent/received on this channel. + size_t total_bytes (void); + void total_bytes (size_t bytes); + // Increment count by <bytes>. + + virtual int handle_timeout (const ACE_Time_Value &, const void *arg); + // Perform timer-based IO_Handler reconnection. + +protected: + enum + { + MAX_RETRY_TIMEOUT = 300 // 5 minutes is the maximum timeout. + }; + + int initialize_connection (void); + // Perform the first-time initiation of a connection to the peer. + + int reinitiate_connection (void); + // Reinitiate a connection asynchronously when peers fail. + + void socket_queue_size (void); + // Set the socket queue size. + + virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE, + ACE_Reactor_Mask = ACE_Event_Handler::RWE_MASK); + // Perform IO_Handler termination. + + Consumer_Map *consumer_map_; + // Pointer to table that maps an event + // to a Set of IO_Handler *'s for output. + + ACE_INET_Addr remote_addr_; + // Address of peer. + + ACE_INET_Addr local_addr_; + // Address of us. + + CONN_ID id_; + // The assigned routing ID of this entry. + + size_t total_bytes_; + // The total number of bytes sent/received on this channel. + + State state_; + // The current state of the channel. + + IO_Handler_Connector *connector_; + // Back pointer to IO_Handler_Connector to reestablish broken + // connections. + + int timeout_; + // Amount of time to wait between reconnection attempts. + + int max_timeout_; + // Maximum amount of time to wait between reconnection attempts. + + char direction_; + // Indicates which direction data flows through the channel ('O' == + // output and 'I' == input). + + int socket_queue_size_; + // Size of the socket queue (0 means "use default"). +}; + +class Supplier_Handler : public IO_Handler + // = TITLE + // Handle reception of Peer events arriving as events. +{ +public: + Supplier_Handler (Consumer_Map *, + IO_Handler_Connector *, + ACE_Thread_Manager * = 0, + int socket_queue_size = 0); + // Constructor sets the consumer map pointer. + + virtual int handle_input (ACE_HANDLE = ACE_INVALID_HANDLE); + // Receive and process peer events. + +protected: + virtual int recv (ACE_Message_Block *&); + // Receive an event from a Supplier. + + int forward (ACE_Message_Block *event); + // Forward the Event to a Consumer. + + ACE_Message_Block *msg_frag_; + // Keep track of event fragment to handle non-blocking recv's from + // Suppliers. +}; + +class Consumer_Handler : public IO_Handler + // = TITLE + // Handle transmission of events to other Peers using a + // single-threaded approach. +{ +public: + Consumer_Handler (Consumer_Map *, + IO_Handler_Connector *, + ACE_Thread_Manager * = 0, + int socket_queue_size = 0); + + virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0); + // Send an event to a Consumer (may be queued if necessary). + +protected: + // = We'll allow up to 16 megabytes to be queued per-output + // channel. + enum {QUEUE_SIZE = 1024 * 1024 * 16}; + + virtual int handle_input (ACE_HANDLE); + // Receive and process shutdowns from a Consumer. + + virtual int handle_output (ACE_HANDLE); + // Finish sending event when flow control conditions abate. + + int nonblk_put (ACE_Message_Block *mb); + // Perform a non-blocking put(). + + virtual int send (ACE_Message_Block *); + // Send an event to a Consumer. +}; + +#endif /* _IO_HANDLER */ diff --git a/apps/Gateway/Gateway/IO_Handler_Connector.cpp b/apps/Gateway/Gateway/IO_Handler_Connector.cpp new file mode 100644 index 00000000000..712b348951d --- /dev/null +++ b/apps/Gateway/Gateway/IO_Handler_Connector.cpp @@ -0,0 +1,92 @@ +#include "IO_Handler_Connector.h" +// $Id$ + + +IO_Handler_Connector::IO_Handler_Connector (void) +{ +} + +// Override the connection-failure method to add timer support. +// Note that these timers perform "expoential backoff" to +// avoid rapidly trying to reestablish connections when a link +// goes down. + +int +IO_Handler_Connector::handle_close (ACE_HANDLE sd, ACE_Reactor_Mask) +{ + ACE_Connector<IO_Handler, ACE_SOCK_CONNECTOR>::AST *stp = 0; + + // Locate the ACE_Svc_Handler corresponding to the socket descriptor. + if (this->handler_map_.find (sd, stp) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) can't locate channel %d in map, %p\n", + sd, "find"), -1); + + IO_Handler *channel = stp->svc_handler (); + + // Schedule a reconnection request at some point in the future + // (note that channel uses an exponential backoff scheme). + if (ACE_Service_Config::reactor ()->schedule_timer (channel, 0, + channel->timeout ()) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "schedule_timer"), -1); + return 0; +} + +// Initiate (or reinitiate) a connection to the IO_Handler. + +int +IO_Handler_Connector::initiate_connection (IO_Handler *channel, + ACE_Synch_Options &synch_options) +{ + char buf[MAXHOSTNAMELEN]; + + // Mark ourselves as idle so that the various iterators + // will ignore us until we are reconnected. + channel->state (IO_Handler::IDLE); + + if (channel->remote_addr ().addr_to_string (buf, sizeof buf) == -1 + || channel->local_addr ().addr_to_string (buf, sizeof buf) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "can't obtain peer's address"), -1); + + // Try to connect to the Peer. + + if (this->connect (channel, channel->remote_addr (), + synch_options, channel->local_addr ()) == -1) + { + if (errno != EWOULDBLOCK) + { + channel->state (IO_Handler::FAILED); + ACE_DEBUG ((LM_DEBUG, "(%t) %p on address %s\n", + "connect", buf)); + + // Reschedule ourselves to try and connect again. + if (synch_options[ACE_Synch_Options::USE_REACTOR]) + { + if (ACE_Service_Config::reactor ()->schedule_timer + (channel, 0, channel->timeout ()) == 0) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "schedule_timer"), -1); + } + else + // Failures on synchronous connects are reported as errors + // so that the caller can decide how to proceed. + return -1; + } + else + { + channel->state (IO_Handler::CONNECTING); + ACE_DEBUG ((LM_DEBUG, + "(%t) in the process of connecting %s to %s\n", + synch_options[ACE_Synch_Options::USE_REACTOR] + ? "asynchronously" : "synchronously", buf)); + } + } + else + { + channel->state (IO_Handler::ESTABLISHED); + ACE_DEBUG ((LM_DEBUG, "(%t) connected to %s on %d\n", + buf, channel->get_handle ())); + } + return 0; +} diff --git a/apps/Gateway/Gateway/IO_Handler_Connector.h b/apps/Gateway/Gateway/IO_Handler_Connector.h new file mode 100644 index 00000000000..585428c88ee --- /dev/null +++ b/apps/Gateway/Gateway/IO_Handler_Connector.h @@ -0,0 +1,40 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// IO_Handler_Connector.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_IO_HANDLER_CONNECTOR) +#define _IO_HANDLER_CONNECTOR + +#include "ace/Connector.h" +#include "Thr_IO_Handler.h" + +class IO_Handler_Connector : public ACE_Connector<IO_Handler, ACE_SOCK_CONNECTOR> + // = TITLE + // A concrete factory class that setups connections to peerds + // and produces a new IO_Handler object to do the dirty work... +{ +public: + IO_Handler_Connector (void); + + // Initiate (or reinitiate) a connection on the IO_Handler. + int initiate_connection (IO_Handler *, + ACE_Synch_Options & = ACE_Synch_Options::synch); + +protected: + // Override the connection-failure method to add timer support. + virtual int handle_close (ACE_HANDLE sd, ACE_Reactor_Mask); +}; + +#endif /* _IO_HANDLER_CONNECTOR */ diff --git a/apps/Gateway/Gateway/Makefile b/apps/Gateway/Gateway/Makefile index b2115abbd59..3b54556c4f6 100644 --- a/apps/Gateway/Gateway/Makefile +++ b/apps/Gateway/Gateway/Makefile @@ -9,17 +9,18 @@ #---------------------------------------------------------------------------- BIN = gatewayd -LIB = libGateway.a +#LIB = libGateway.a SHLIB = libGateway.so -FILES = Channel \ - Channel_Connector \ +FILES = Event_Channel \ + IO_Handler \ + IO_Handler_Connector \ Config_Files \ File_Parser \ Gateway \ - Routing_Entry \ - Routing_Table \ - Thr_Channel + Consumer_Entry \ + Consumer_Map \ + Thr_IO_Handler LSRC = $(addsuffix .cpp,$(FILES)) LOBJ = $(addsuffix .o,$(FILES)) @@ -51,7 +52,7 @@ include $(WRAPPER_ROOT)/include/makeinclude/rules.local.GNU # Default behavior is to use single-threading. See the README # file for information on how to configure this with multiple # strategies for threading the input and output channels. -DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT +DEFFLAGS += -DASSIGN_SUPPLIER_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT #---------------------------------------------------------------------------- # Dependencies @@ -60,8 +61,8 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT # DO NOT DELETE THIS LINE -- g++dep uses it. # DO NOT PUT ANYTHING AFTER THIS LINE, IT WILL GO AWAY. -.obj/Channel.o .shobj/Channel.so: Channel.cpp Routing_Entry.h \ - $(WRAPPER_ROOT)/ace/Set.h \ +.obj/Event_Channel.o .shobj/Event_Channel.so: Event_Channel.cpp \ + $(WRAPPER_ROOT)/ace/Get_Opt.h \ $(WRAPPER_ROOT)/ace/ACE.h \ $(WRAPPER_ROOT)/ace/OS.h \ $(WRAPPER_ROOT)/ace/Time_Value.h \ @@ -73,7 +74,7 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Log_Priority.h \ $(WRAPPER_ROOT)/ace/Log_Record.i \ $(WRAPPER_ROOT)/ace/ACE.i \ - Channel_Connector.h \ + Config_Files.h File_Parser.h IO_Handler_Connector.h \ $(WRAPPER_ROOT)/ace/Connector.h \ $(WRAPPER_ROOT)/ace/Service_Config.h \ $(WRAPPER_ROOT)/ace/Service_Object.h \ @@ -87,19 +88,80 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.i \ $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.i \ $(WRAPPER_ROOT)/ace/Synch_T.h \ + $(WRAPPER_ROOT)/ace/Signal.h \ + $(WRAPPER_ROOT)/ace/Set.h \ + $(WRAPPER_ROOT)/ace/Reactor.h \ + $(WRAPPER_ROOT)/ace/Handle_Set.h \ + $(WRAPPER_ROOT)/ace/Timer_Queue.h \ + $(WRAPPER_ROOT)/ace/Token.h \ + $(WRAPPER_ROOT)/ace/Pipe.h \ + $(WRAPPER_ROOT)/ace/Pipe.i \ + $(WRAPPER_ROOT)/ace/SOCK_Stream.h \ + $(WRAPPER_ROOT)/ace/SOCK_IO.h \ + $(WRAPPER_ROOT)/ace/SOCK.h \ + $(WRAPPER_ROOT)/ace/Addr.h \ + $(WRAPPER_ROOT)/ace/IPC_SAP.h \ + $(WRAPPER_ROOT)/ace/IPC_SAP.i \ + $(WRAPPER_ROOT)/ace/SOCK.i \ + $(WRAPPER_ROOT)/ace/SOCK_IO.i \ + $(WRAPPER_ROOT)/ace/INET_Addr.h \ + $(WRAPPER_ROOT)/ace/SOCK_Stream.i \ + $(WRAPPER_ROOT)/ace/Reactor.i \ $(WRAPPER_ROOT)/ace/Proactor.h \ $(WRAPPER_ROOT)/ace/Message_Block.h \ $(WRAPPER_ROOT)/ace/Malloc.h \ $(WRAPPER_ROOT)/ace/Malloc_T.h \ $(WRAPPER_ROOT)/ace/Memory_Pool.h \ - $(WRAPPER_ROOT)/ace/Signal.h \ $(WRAPPER_ROOT)/ace/Mem_Map.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.i \ $(WRAPPER_ROOT)/ace/ReactorEx.h \ - $(WRAPPER_ROOT)/ace/Token.h \ + $(WRAPPER_ROOT)/ace/Message_Queue.h \ + $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ + $(WRAPPER_ROOT)/ace/Strategies.h \ + $(WRAPPER_ROOT)/ace/Strategies_T.h \ + $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ + $(WRAPPER_ROOT)/ace/Map_Manager.h \ + $(WRAPPER_ROOT)/ace/Svc_Handler.h \ + $(WRAPPER_ROOT)/ace/Synch_Options.h \ + $(WRAPPER_ROOT)/ace/Task.h \ + $(WRAPPER_ROOT)/ace/Task_T.h \ + $(WRAPPER_ROOT)/ace/Connector.i \ + Thr_IO_Handler.h IO_Handler.h \ + $(WRAPPER_ROOT)/ace/SOCK_Connector.h \ + $(WRAPPER_ROOT)/ace/SOCK_Connector.i \ + Consumer_Map.h Concurrency_Strategies.h Event.h Consumer_Entry.h \ + Event_Channel.h +.obj/IO_Handler.o .shobj/IO_Handler.so: IO_Handler.cpp Consumer_Entry.h \ + $(WRAPPER_ROOT)/ace/Set.h \ + $(WRAPPER_ROOT)/ace/ACE.h \ + $(WRAPPER_ROOT)/ace/OS.h \ + $(WRAPPER_ROOT)/ace/Time_Value.h \ + $(WRAPPER_ROOT)/ace/config.h \ + $(WRAPPER_ROOT)/ace/stdcpp.h \ + $(WRAPPER_ROOT)/ace/Trace.h \ + $(WRAPPER_ROOT)/ace/Log_Msg.h \ + $(WRAPPER_ROOT)/ace/Log_Record.h \ + $(WRAPPER_ROOT)/ace/Log_Priority.h \ + $(WRAPPER_ROOT)/ace/Log_Record.i \ + $(WRAPPER_ROOT)/ace/ACE.i \ + IO_Handler_Connector.h \ + $(WRAPPER_ROOT)/ace/Connector.h \ + $(WRAPPER_ROOT)/ace/Service_Config.h \ + $(WRAPPER_ROOT)/ace/Service_Object.h \ + $(WRAPPER_ROOT)/ace/Shared_Object.h \ + $(WRAPPER_ROOT)/ace/Event_Handler.h \ + $(WRAPPER_ROOT)/ace/Thread_Manager.h \ + $(WRAPPER_ROOT)/ace/Thread.h \ + $(WRAPPER_ROOT)/ace/Synch.h \ + $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.h \ + $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.h \ + $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.i \ + $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.i \ + $(WRAPPER_ROOT)/ace/Synch_T.h \ + $(WRAPPER_ROOT)/ace/Signal.h \ $(WRAPPER_ROOT)/ace/Reactor.h \ $(WRAPPER_ROOT)/ace/Handle_Set.h \ + $(WRAPPER_ROOT)/ace/Timer_Queue.h \ + $(WRAPPER_ROOT)/ace/Token.h \ $(WRAPPER_ROOT)/ace/Pipe.h \ $(WRAPPER_ROOT)/ace/Pipe.i \ $(WRAPPER_ROOT)/ace/SOCK_Stream.h \ @@ -113,21 +175,30 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/INET_Addr.h \ $(WRAPPER_ROOT)/ace/SOCK_Stream.i \ $(WRAPPER_ROOT)/ace/Reactor.i \ + $(WRAPPER_ROOT)/ace/Proactor.h \ + $(WRAPPER_ROOT)/ace/Message_Block.h \ + $(WRAPPER_ROOT)/ace/Malloc.h \ + $(WRAPPER_ROOT)/ace/Malloc_T.h \ + $(WRAPPER_ROOT)/ace/Memory_Pool.h \ + $(WRAPPER_ROOT)/ace/Mem_Map.h \ + $(WRAPPER_ROOT)/ace/ReactorEx.h \ + $(WRAPPER_ROOT)/ace/Message_Queue.h \ + $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ + $(WRAPPER_ROOT)/ace/Strategies.h \ + $(WRAPPER_ROOT)/ace/Strategies_T.h \ $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ $(WRAPPER_ROOT)/ace/Map_Manager.h \ $(WRAPPER_ROOT)/ace/Svc_Handler.h \ $(WRAPPER_ROOT)/ace/Synch_Options.h \ $(WRAPPER_ROOT)/ace/Task.h \ $(WRAPPER_ROOT)/ace/Task_T.h \ - $(WRAPPER_ROOT)/ace/Message_Queue.h \ - $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ - $(WRAPPER_ROOT)/ace/Strategies.h \ $(WRAPPER_ROOT)/ace/Connector.i \ - Thr_Channel.h Channel.h \ + Thr_IO_Handler.h IO_Handler.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.i \ - Routing_Table.h Peer_Message.h -.obj/Channel_Connector.o .shobj/Channel_Connector.so: Channel_Connector.cpp Channel_Connector.h \ + Consumer_Map.h Concurrency_Strategies.h Event.h +.obj/IO_Handler_Connector.o .shobj/IO_Handler_Connector.so: IO_Handler_Connector.cpp \ + IO_Handler_Connector.h \ $(WRAPPER_ROOT)/ace/Connector.h \ $(WRAPPER_ROOT)/ace/Service_Config.h \ $(WRAPPER_ROOT)/ace/Service_Object.h \ @@ -152,20 +223,12 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.i \ $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.i \ $(WRAPPER_ROOT)/ace/Synch_T.h \ - $(WRAPPER_ROOT)/ace/Set.h \ - $(WRAPPER_ROOT)/ace/Proactor.h \ - $(WRAPPER_ROOT)/ace/Message_Block.h \ - $(WRAPPER_ROOT)/ace/Malloc.h \ - $(WRAPPER_ROOT)/ace/Malloc_T.h \ - $(WRAPPER_ROOT)/ace/Memory_Pool.h \ $(WRAPPER_ROOT)/ace/Signal.h \ - $(WRAPPER_ROOT)/ace/Mem_Map.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.i \ - $(WRAPPER_ROOT)/ace/ReactorEx.h \ - $(WRAPPER_ROOT)/ace/Token.h \ + $(WRAPPER_ROOT)/ace/Set.h \ $(WRAPPER_ROOT)/ace/Reactor.h \ $(WRAPPER_ROOT)/ace/Handle_Set.h \ + $(WRAPPER_ROOT)/ace/Timer_Queue.h \ + $(WRAPPER_ROOT)/ace/Token.h \ $(WRAPPER_ROOT)/ace/Pipe.h \ $(WRAPPER_ROOT)/ace/Pipe.i \ $(WRAPPER_ROOT)/ace/SOCK_Stream.h \ @@ -179,20 +242,28 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/INET_Addr.h \ $(WRAPPER_ROOT)/ace/SOCK_Stream.i \ $(WRAPPER_ROOT)/ace/Reactor.i \ + $(WRAPPER_ROOT)/ace/Proactor.h \ + $(WRAPPER_ROOT)/ace/Message_Block.h \ + $(WRAPPER_ROOT)/ace/Malloc.h \ + $(WRAPPER_ROOT)/ace/Malloc_T.h \ + $(WRAPPER_ROOT)/ace/Memory_Pool.h \ + $(WRAPPER_ROOT)/ace/Mem_Map.h \ + $(WRAPPER_ROOT)/ace/ReactorEx.h \ + $(WRAPPER_ROOT)/ace/Message_Queue.h \ + $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ + $(WRAPPER_ROOT)/ace/Strategies.h \ + $(WRAPPER_ROOT)/ace/Strategies_T.h \ $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ $(WRAPPER_ROOT)/ace/Map_Manager.h \ $(WRAPPER_ROOT)/ace/Svc_Handler.h \ $(WRAPPER_ROOT)/ace/Synch_Options.h \ $(WRAPPER_ROOT)/ace/Task.h \ $(WRAPPER_ROOT)/ace/Task_T.h \ - $(WRAPPER_ROOT)/ace/Message_Queue.h \ - $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ - $(WRAPPER_ROOT)/ace/Strategies.h \ $(WRAPPER_ROOT)/ace/Connector.i \ - Thr_Channel.h Channel.h \ + Thr_IO_Handler.h IO_Handler.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.i \ - Routing_Table.h Routing_Entry.h Peer_Message.h + Consumer_Map.h Concurrency_Strategies.h Event.h Consumer_Entry.h .obj/Config_Files.o .shobj/Config_Files.so: Config_Files.cpp \ $(WRAPPER_ROOT)/ace/OS.h \ $(WRAPPER_ROOT)/ace/Time_Value.h \ @@ -220,7 +291,9 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Log_Record.i \ File_Parser.h .obj/Gateway.o .shobj/Gateway.so: Gateway.cpp \ - $(WRAPPER_ROOT)/ace/Get_Opt.h \ + $(WRAPPER_ROOT)/ace/Service_Config.h \ + $(WRAPPER_ROOT)/ace/Service_Object.h \ + $(WRAPPER_ROOT)/ace/Shared_Object.h \ $(WRAPPER_ROOT)/ace/ACE.h \ $(WRAPPER_ROOT)/ace/OS.h \ $(WRAPPER_ROOT)/ace/Time_Value.h \ @@ -232,9 +305,6 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Log_Priority.h \ $(WRAPPER_ROOT)/ace/Log_Record.i \ $(WRAPPER_ROOT)/ace/ACE.i \ - $(WRAPPER_ROOT)/ace/Service_Config.h \ - $(WRAPPER_ROOT)/ace/Service_Object.h \ - $(WRAPPER_ROOT)/ace/Shared_Object.h \ $(WRAPPER_ROOT)/ace/Event_Handler.h \ $(WRAPPER_ROOT)/ace/Thread_Manager.h \ $(WRAPPER_ROOT)/ace/Thread.h \ @@ -244,20 +314,12 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.i \ $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.i \ $(WRAPPER_ROOT)/ace/Synch_T.h \ - $(WRAPPER_ROOT)/ace/Set.h \ - $(WRAPPER_ROOT)/ace/Proactor.h \ - $(WRAPPER_ROOT)/ace/Message_Block.h \ - $(WRAPPER_ROOT)/ace/Malloc.h \ - $(WRAPPER_ROOT)/ace/Malloc_T.h \ - $(WRAPPER_ROOT)/ace/Memory_Pool.h \ $(WRAPPER_ROOT)/ace/Signal.h \ - $(WRAPPER_ROOT)/ace/Mem_Map.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.i \ - $(WRAPPER_ROOT)/ace/ReactorEx.h \ - $(WRAPPER_ROOT)/ace/Token.h \ + $(WRAPPER_ROOT)/ace/Set.h \ $(WRAPPER_ROOT)/ace/Reactor.h \ $(WRAPPER_ROOT)/ace/Handle_Set.h \ + $(WRAPPER_ROOT)/ace/Timer_Queue.h \ + $(WRAPPER_ROOT)/ace/Token.h \ $(WRAPPER_ROOT)/ace/Pipe.h \ $(WRAPPER_ROOT)/ace/Pipe.i \ $(WRAPPER_ROOT)/ace/SOCK_Stream.h \ @@ -271,23 +333,20 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/INET_Addr.h \ $(WRAPPER_ROOT)/ace/SOCK_Stream.i \ $(WRAPPER_ROOT)/ace/Reactor.i \ - $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ - Config_Files.h File_Parser.h Gateway.h Channel_Connector.h \ - $(WRAPPER_ROOT)/ace/Connector.h \ - $(WRAPPER_ROOT)/ace/Map_Manager.h \ - $(WRAPPER_ROOT)/ace/Svc_Handler.h \ - $(WRAPPER_ROOT)/ace/Synch_Options.h \ - $(WRAPPER_ROOT)/ace/Task.h \ - $(WRAPPER_ROOT)/ace/Task_T.h \ + $(WRAPPER_ROOT)/ace/Proactor.h \ + $(WRAPPER_ROOT)/ace/Message_Block.h \ + $(WRAPPER_ROOT)/ace/Malloc.h \ + $(WRAPPER_ROOT)/ace/Malloc_T.h \ + $(WRAPPER_ROOT)/ace/Memory_Pool.h \ + $(WRAPPER_ROOT)/ace/Mem_Map.h \ + $(WRAPPER_ROOT)/ace/ReactorEx.h \ $(WRAPPER_ROOT)/ace/Message_Queue.h \ $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ $(WRAPPER_ROOT)/ace/Strategies.h \ - $(WRAPPER_ROOT)/ace/Connector.i \ - Thr_Channel.h Channel.h \ - $(WRAPPER_ROOT)/ace/SOCK_Connector.h \ - $(WRAPPER_ROOT)/ace/SOCK_Connector.i \ - Routing_Table.h Routing_Entry.h Peer_Message.h -.obj/Routing_Entry.o .shobj/Routing_Entry.so: Routing_Entry.cpp Routing_Entry.h \ + $(WRAPPER_ROOT)/ace/Strategies_T.h \ + $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ + Gateway.h +.obj/Consumer_Entry.o .shobj/Consumer_Entry.so: Consumer_Entry.cpp Consumer_Entry.h \ $(WRAPPER_ROOT)/ace/Set.h \ $(WRAPPER_ROOT)/ace/ACE.h \ $(WRAPPER_ROOT)/ace/OS.h \ @@ -300,7 +359,7 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Log_Priority.h \ $(WRAPPER_ROOT)/ace/Log_Record.i \ $(WRAPPER_ROOT)/ace/ACE.i -.obj/Routing_Table.o .shobj/Routing_Table.so: Routing_Table.cpp Routing_Table.h \ +.obj/Consumer_Map.o .shobj/Consumer_Map.so: Consumer_Map.cpp Consumer_Map.h \ $(WRAPPER_ROOT)/ace/Map_Manager.h \ $(WRAPPER_ROOT)/ace/ACE.h \ $(WRAPPER_ROOT)/ace/OS.h \ @@ -312,8 +371,10 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Log_Record.h \ $(WRAPPER_ROOT)/ace/Log_Priority.h \ $(WRAPPER_ROOT)/ace/Log_Record.i \ - $(WRAPPER_ROOT)/ace/ACE.i -.obj/Thr_Channel.o .shobj/Thr_Channel.so: Thr_Channel.cpp Thr_Channel.h Channel.h \ + $(WRAPPER_ROOT)/ace/ACE.i \ + Concurrency_Strategies.h Event.h Consumer_Entry.h \ + $(WRAPPER_ROOT)/ace/Set.h +.obj/Thr_IO_Handler.o .shobj/Thr_IO_Handler.so: Thr_IO_Handler.cpp Thr_IO_Handler.h IO_Handler.h \ $(WRAPPER_ROOT)/ace/Service_Config.h \ $(WRAPPER_ROOT)/ace/Service_Object.h \ $(WRAPPER_ROOT)/ace/Shared_Object.h \ @@ -337,20 +398,12 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/SV_Semaphore_Simple.i \ $(WRAPPER_ROOT)/ace/SV_Semaphore_Complex.i \ $(WRAPPER_ROOT)/ace/Synch_T.h \ - $(WRAPPER_ROOT)/ace/Set.h \ - $(WRAPPER_ROOT)/ace/Proactor.h \ - $(WRAPPER_ROOT)/ace/Message_Block.h \ - $(WRAPPER_ROOT)/ace/Malloc.h \ - $(WRAPPER_ROOT)/ace/Malloc_T.h \ - $(WRAPPER_ROOT)/ace/Memory_Pool.h \ $(WRAPPER_ROOT)/ace/Signal.h \ - $(WRAPPER_ROOT)/ace/Mem_Map.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.h \ - $(WRAPPER_ROOT)/ace/Timer_Queue.i \ - $(WRAPPER_ROOT)/ace/ReactorEx.h \ - $(WRAPPER_ROOT)/ace/Token.h \ + $(WRAPPER_ROOT)/ace/Set.h \ $(WRAPPER_ROOT)/ace/Reactor.h \ $(WRAPPER_ROOT)/ace/Handle_Set.h \ + $(WRAPPER_ROOT)/ace/Timer_Queue.h \ + $(WRAPPER_ROOT)/ace/Token.h \ $(WRAPPER_ROOT)/ace/Pipe.h \ $(WRAPPER_ROOT)/ace/Pipe.i \ $(WRAPPER_ROOT)/ace/SOCK_Stream.h \ @@ -364,6 +417,17 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/INET_Addr.h \ $(WRAPPER_ROOT)/ace/SOCK_Stream.i \ $(WRAPPER_ROOT)/ace/Reactor.i \ + $(WRAPPER_ROOT)/ace/Proactor.h \ + $(WRAPPER_ROOT)/ace/Message_Block.h \ + $(WRAPPER_ROOT)/ace/Malloc.h \ + $(WRAPPER_ROOT)/ace/Malloc_T.h \ + $(WRAPPER_ROOT)/ace/Memory_Pool.h \ + $(WRAPPER_ROOT)/ace/Mem_Map.h \ + $(WRAPPER_ROOT)/ace/ReactorEx.h \ + $(WRAPPER_ROOT)/ace/Message_Queue.h \ + $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ + $(WRAPPER_ROOT)/ace/Strategies.h \ + $(WRAPPER_ROOT)/ace/Strategies_T.h \ $(WRAPPER_ROOT)/ace/Svc_Conf_Tokens.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.h \ $(WRAPPER_ROOT)/ace/SOCK_Connector.i \ @@ -371,13 +435,11 @@ DEFFLAGS += -DASSIGN_ROUTING_ID # -DUSE_OUTPUT_MT -DUSE_INPUT_MT $(WRAPPER_ROOT)/ace/Synch_Options.h \ $(WRAPPER_ROOT)/ace/Task.h \ $(WRAPPER_ROOT)/ace/Task_T.h \ - $(WRAPPER_ROOT)/ace/Message_Queue.h \ - $(WRAPPER_ROOT)/ace/IO_Cntl_Msg.h \ - Routing_Table.h \ + Consumer_Map.h \ $(WRAPPER_ROOT)/ace/Map_Manager.h \ - Routing_Entry.h Peer_Message.h Channel_Connector.h \ + Concurrency_Strategies.h Event.h Consumer_Entry.h \ + IO_Handler_Connector.h \ $(WRAPPER_ROOT)/ace/Connector.h \ - $(WRAPPER_ROOT)/ace/Strategies.h \ $(WRAPPER_ROOT)/ace/Connector.i # IF YOU PUT ANYTHING HERE IT WILL GO AWAY diff --git a/apps/Gateway/Gateway/README b/apps/Gateway/Gateway/README index ceb17528d0d..4e986354aaa 100644 --- a/apps/Gateway/Gateway/README +++ b/apps/Gateway/Gateway/README @@ -1,22 +1,23 @@ -This application illustrates an application-level Gateway which -routes messages between a set of Peers in a distributed environment. +This application illustrates an application-level Gateway which routes +messages between Consumer and Suppliers in a distributed environment. -The default configuration is single-threaded, i.e., all Input_Channels -and Output_Channels are multiplexed via the Reactor on a single thread -of control. To obtain a version that multi-threads both input and -output simply set the following flag in the Makefile: +The default configuration is single-threaded, i.e., all +Supplier_Handlers and Consumer_Handlers are multiplexed via the ACE +Reactor within a single thread of control. To obtain a version that +multi-threads both Consumer_Handlers and Supplier_Handlers simply set +the following flag in the Makefile: DEFFLAGS += -DUSE_OUTPUT_MT -DUSE_INPUT_MT -To get a version that uses single-threading for all Input_Channels, -but a separate thread per-Output_Channel set the following flag in the -Makefile: +To get a version that uses single-threading for all Supplier_Handlers, +but a separate thread per-Consumer_Handler set the following flag in +the Makefile: DEFFLAGS += -DUSE_OUTPUT_MT If you examine the source code, you'll see that very few changes are required in the source code to switch between single-threading and multi-threading. The ACE Task class is primarily responsible for -enabling the flexible modification of concurrency strategies with -little modification to the source code, design, and system +enabling the flexible modification of concurrency strategies with only +minor changes required to the source code, design, and system architecture. diff --git a/apps/Gateway/Gateway/Thr_IO_Handler.cpp b/apps/Gateway/Gateway/Thr_IO_Handler.cpp new file mode 100644 index 00000000000..109cfad9c3f --- /dev/null +++ b/apps/Gateway/Gateway/Thr_IO_Handler.cpp @@ -0,0 +1,204 @@ +#include "Thr_IO_Handler.h" +// $Id$ + +#include "IO_Handler_Connector.h" + +#if defined (ACE_HAS_THREADS) +Thr_Consumer_Handler::Thr_Consumer_Handler (Consumer_Map *consumer_map, + IO_Handler_Connector *ioc, + ACE_Thread_Manager *thr_mgr, + int socket_queue_size) + : Consumer_Handler (consumer_map, ioc, thr_mgr, socket_queue_size) +{ +} + +// This method should be called only when the peer shuts down +// unexpectedly. This method marks the IO_Handler as having failed and +// deactivates the ACE_Message_Queue (to wake up the thread blocked on +// <dequeue_head> in svc()). Thr_Output_Handler::handle_close () will +// eventually try to reconnect... + +int +Thr_Consumer_Handler::handle_input (ACE_HANDLE h) +{ + this->Consumer_Handler::handle_input (h); + ACE_Service_Config::reactor ()->remove_handler (h, + ACE_Event_Handler::RWE_MASK + | ACE_Event_Handler::DONT_CALL); + // Deactivate the queue while we try to get reconnected. + this->msg_queue ()->deactivate (); + return 0; +} + +// Initialize the threaded Consumer_Handler object and spawn a new +// thread. + +int +Thr_Consumer_Handler::open (void *) +{ + // Set the size of the socket queue. + this->socket_queue_size (); + + // Turn off non-blocking I/O. + if (this->peer ().disable (ACE_NONBLOCK) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "enable"), -1); + + // Register ourselves to receive input events (which indicate that + // the Peer has shut down unexpectedly). + if (ACE_Service_Config::reactor ()->register_handler (this, + ACE_Event_Handler::READ_MASK) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "register_handler"), -1); + + if (this->initialize_connection ()) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "initialize_connection"), -1); + + // Reactivate message queue. If it was active then this is the + // first time in and we need to spawn a thread, otherwise the queue + // was inactive due to some problem and we've already got a thread. + if (this->msg_queue ()->activate () == ACE_Message_Queue<SYNCH_STRATEGY>::WAS_ACTIVE) + { + ACE_DEBUG ((LM_DEBUG, "(%t) spawning new thread\n")); + // Become an active object by spawning a new thread to transmit + // messages to peers. + return this->activate (THR_NEW_LWP | THR_DETACHED); + } + else + { + ACE_DEBUG ((LM_DEBUG, "(%t) reusing existing thread\n")); + return 0; + } +} + +// ACE_Queue up a message for transmission (must not block since all +// Supplier_Handlers are single-threaded). + +int +Thr_Consumer_Handler::put (ACE_Message_Block *mb, ACE_Time_Value *) +{ + // Perform non-blocking enqueue. + return this->msg_queue ()->enqueue_tail (mb, (ACE_Time_Value *) &ACE_Time_Value::zero); +} + +// Transmit messages to the peer (note simplification resulting from +// threads...) + +int +Thr_Consumer_Handler::svc (void) +{ + for (;;) + { + ACE_DEBUG ((LM_DEBUG, "(%t) connected! Thr_Consumer_Handler's fd = %d\n", + this->peer ().get_handle ())); + + // Since this method runs in its own thread it is OK to block on + // output. + + for (ACE_Message_Block *mb = 0; + this->msg_queue ()->dequeue_head (mb) != -1; ) + if (this->send (mb) == -1) + ACE_ERROR ((LM_ERROR, "(%t) %p\n", "send failed")); + + ACE_ASSERT (errno == ESHUTDOWN); + + ACE_DEBUG ((LM_DEBUG, "(%t) shutting down threaded Consumer_Handler %d on handle %d\n", + this->id (), this->get_handle ())); + + this->peer ().close (); + + for (this->timeout (1); + // Default is to reconnect synchronously. + this->connector_->initiate_connection (this) == -1; ) + { + ACE_Time_Value tv (this->timeout ()); + ACE_ERROR ((LM_ERROR, + "(%t) reattempting connection, sec = %d\n", + tv.sec ())); + ACE_OS::sleep (tv); + } + } + + return 0; +} + +Thr_Supplier_Handler::Thr_Supplier_Handler (Consumer_Map *consumer_map, + IO_Handler_Connector *ioc, + ACE_Thread_Manager *thr_mgr, + int socket_queue_size) + : Supplier_Handler (consumer_map, ioc, thr_mgr, socket_queue_size) +{ +} + +int +Thr_Supplier_Handler::open (void *) +{ + // Set the size of the socket queue. + this->socket_queue_size (); + + // Turn off non-blocking I/O. + if (this->peer ().disable (ACE_NONBLOCK) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "enable"), -1); + + if (this->initialize_connection ()) + ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", + "initialize_connection"), -1); + + // Reactivate message queue. If it was active then this is the + // first time in and we need to spawn a thread, otherwise the queue + // was inactive due to some problem and we've already got a thread. + if (this->msg_queue ()->activate () == ACE_Message_Queue<SYNCH_STRATEGY>::WAS_ACTIVE) + { + ACE_DEBUG ((LM_DEBUG, "(%t) spawning new thread\n")); + // Become an active object by spawning a new thread to transmit + // messages to peers. + return this->activate (THR_NEW_LWP | THR_DETACHED); + } + else + { + ACE_DEBUG ((LM_DEBUG, "(%t) reusing existing thread\n")); + return 0; + } +} + +// Receive messages from a Peer in a separate thread (note reuse of +// existing code!). + +int +Thr_Supplier_Handler::svc (void) +{ + for (;;) + { + ACE_DEBUG ((LM_DEBUG, "(%t) connected! Thr_Supplier_Handler's fd = %d\n", + this->peer ().get_handle ())); + + // Since this method runs in its own thread and processes + // messages for one connection it is OK to block on input and + // output. + + while (this->handle_input () != -1) + continue; + + ACE_DEBUG ((LM_DEBUG, + "(%t) shutting down threaded Supplier_Handler %d on handle %d\n", + this->id (), + this->get_handle ())); + + this->peer ().close (); + + // Deactivate the queue while we try to get reconnected. + this->msg_queue ()->deactivate (); + + for (this->timeout (1); + // Default is to reconnect synchronously. + this->connector_->initiate_connection (this) == -1; ) + { + ACE_Time_Value tv (this->timeout ()); + ACE_ERROR ((LM_ERROR, + "(%t) reattempting connection, sec = %d\n", tv.sec ())); + ACE_OS::sleep (tv); + } + } + return 0; +} + +#endif /* ACE_HAS_THREADS */ diff --git a/apps/Gateway/Gateway/Thr_IO_Handler.h b/apps/Gateway/Gateway/Thr_IO_Handler.h new file mode 100644 index 00000000000..ee056b35361 --- /dev/null +++ b/apps/Gateway/Gateway/Thr_IO_Handler.h @@ -0,0 +1,64 @@ +/* -*- C++ -*- */ +// $Id$ + +// ============================================================================ +// +// = LIBRARY +// apps +// +// = FILENAME +// Thr_IO_Handler.h +// +// = AUTHOR +// Doug Schmidt +// +// ============================================================================ + +#if !defined (_THR_IO_HANDLER) +#define _THR_IO_HANDLER + +#include "IO_Handler.h" + +#if defined (ACE_HAS_THREADS) +class Thr_Consumer_Handler : public Consumer_Handler + // = TITLE + // Runs each Output IO_Handler in a separate thread. +{ +public: + Thr_Consumer_Handler (Consumer_Map *, + IO_Handler_Connector *, + ACE_Thread_Manager *, + int socket_queue_size); + + virtual int open (void *); + // Initialize the threaded Consumer_Handler object and spawn a new + // thread. + + virtual int handle_input (ACE_HANDLE); + // Called when Peer shutdown unexpectedly. + + virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0); + // Send a message to a peer. + + virtual int svc (void); + // Transmit peer messages. +}; + +class Thr_Supplier_Handler : public Supplier_Handler + // = TITLE + // Runs each Input IO_Handler in a separate thread. +{ +public: + Thr_Supplier_Handler (Consumer_Map *, + IO_Handler_Connector *, + ACE_Thread_Manager *, + int socket_queue_size); + + virtual int open (void *); + // Initialize the object and spawn a new thread. + + virtual int svc (void); + // Transmit peer messages. +}; +#endif /* ACE_HAS_THREADS */ +#endif /* _THR_IO_HANDLER */ diff --git a/apps/Gateway/Gateway/consumer_config b/apps/Gateway/Gateway/consumer_config new file mode 100644 index 00000000000..d33469ee157 --- /dev/null +++ b/apps/Gateway/Gateway/consumer_config @@ -0,0 +1,8 @@ +# Consumer map configuration file +# Conn ID Logical ID Payload Destinations +# ------- ---------- ------- ------------ +# 1 1 0 3,4,5 + 1 1 0 3 + 3 1 0 3 +# 4 1 0 4 +# 5 1 0 5 diff --git a/apps/Gateway/Gateway/svc.conf b/apps/Gateway/Gateway/svc.conf index 7ae8d4b0080..6c41415a9ce 100644 --- a/apps/Gateway/Gateway/svc.conf +++ b/apps/Gateway/Gateway/svc.conf @@ -1,3 +1,3 @@ #static Svc_Manager "-d -p 2913" -dynamic Gateway Service_Object * ./libGateway:_make_ACE_Gateway() active "-d -c cc_config -f rt_config" +dynamic Gateway Service_Object * ./libGateway:_make_ACE_Gateway() active "-d -c connection_config -f consumer_config" diff --git a/apps/Gateway/Peer/Gateway_Handler.cpp b/apps/Gateway/Peer/Gateway_Handler.cpp index 15ca0a58807..cfc9a7dad6f 100644 --- a/apps/Gateway/Peer/Gateway_Handler.cpp +++ b/apps/Gateway/Peer/Gateway_Handler.cpp @@ -84,10 +84,10 @@ Gateway_Handler::xmit_stdin (void) ACE_Message_Block *mb; ACE_NEW_RETURN (mb, - ACE_Message_Block (sizeof (Peer_Message)), + ACE_Message_Block (sizeof (Event)), -1); - Peer_Message *peer_msg = (Peer_Message *) mb->rd_ptr (); + Event *peer_msg = (Event *) mb->rd_ptr (); peer_msg->header_.routing_id_ = this->routing_id_; n = ACE_OS::read (ACE_STDIN, peer_msg->buf_, sizeof peer_msg->buf_); @@ -270,7 +270,7 @@ Gateway_Handler::send_peer (ACE_Message_Block *mb) int Gateway_Handler::recv_peer (ACE_Message_Block *&mb) { - Peer_Message *peer_msg; + Event *peer_msg; size_t len; ssize_t n; size_t offset = 0; @@ -278,14 +278,14 @@ Gateway_Handler::recv_peer (ACE_Message_Block *&mb) if (this->msg_frag_ == 0) { ACE_NEW_RETURN (this->msg_frag_, - ACE_Message_Block (sizeof (Peer_Message)), + ACE_Message_Block (sizeof (Event)), -1); // No existing fragment... if (this->msg_frag_ == 0) ACE_ERROR_RETURN ((LM_ERROR, "out of memory\n"), -1); - peer_msg = (Peer_Message *) this->msg_frag_->rd_ptr (); + peer_msg = (Event *) this->msg_frag_->rd_ptr (); switch (n = this->peer ().recv (peer_msg, sizeof (Peer_Header))) { @@ -441,7 +441,7 @@ Gateway_Handler::await_messages (void) // We got a valid message, so let's process it now! At the // moment, we just print out the message contents... - Peer_Message *peer_msg = (Peer_Message *) mb->rd_ptr (); + Event *peer_msg = (Event *) mb->rd_ptr (); this->total_bytes_ += mb->length (); #if defined (VERBOSE) diff --git a/apps/Gateway/Peer/Gateway_Handler.h b/apps/Gateway/Peer/Gateway_Handler.h index 6dc4539e6b7..82477264c4f 100644 --- a/apps/Gateway/Peer/Gateway_Handler.h +++ b/apps/Gateway/Peer/Gateway_Handler.h @@ -32,7 +32,7 @@ #include "ace/SOCK_Acceptor.h" #include "ace/INET_Addr.h" #include "ace/Map_Manager.h" -#include "Peer_Message.h" +#include "Event.h" // Forward declaration. class Gateway_Handler; diff --git a/apps/Gateway/Peer/Makefile b/apps/Gateway/Peer/Makefile index f88aaf14926..9909eb2ef2a 100644 --- a/apps/Gateway/Peer/Makefile +++ b/apps/Gateway/Peer/Makefile @@ -14,9 +14,9 @@ FILES = Gateway_Handler LSRC = $(addsuffix .cpp,$(FILES)) LOBJ = $(addsuffix .o,$(FILES)) -VSHOBJS = $(LSRC:%.cpp=$(VSHDIR)%.so) +SHOBJ = $(addsuffix .so,$(FILES)) -LDLIBS = $(VSHOBJS) +LDLIBS = $(addprefix .shobj/,$(SHOBJ)) VLDLIBS = $(LDLIBS:%=%$(VAR)) @@ -111,6 +111,6 @@ include $(WRAPPER_ROOT)/include/makeinclude/rules.local.GNU $(WRAPPER_ROOT)/ace/Acceptor.i \ $(WRAPPER_ROOT)/ace/SOCK_Acceptor.h \ $(WRAPPER_ROOT)/ace/Map_Manager.h \ - Peer_Message.h + Event.h # IF YOU PUT ANYTHING HERE IT WILL GO AWAY diff --git a/apps/Gateway/README b/apps/Gateway/README index 892fe43ba6d..ffd7e52bdf4 100644 --- a/apps/Gateway/README +++ b/apps/Gateway/README @@ -2,8 +2,12 @@ OVERVIEW This directory contains source code for a prototype application-level gateway implemented with ACE. This prototype was developed in my -cs422 grad OS class at Washington University. You can get a paper -that explains the patterns used in this implementation at URL +cs422 OS class at Washington University. It illustrates the use of +Event Channels to forward events from Suppliers to Consumers in a +distributed system. + +You can get a paper that explains the patterns used in this +implementation at the following WWW URL: http://www.cs.wustl.edu/~schmidt/TAPOS-95.ps.gz @@ -15,28 +19,33 @@ Gateway -- The application Gateway, which must be started *after* all the Peers described below). This process reads the - cc_config and rt_config files. The cc_config file tells - the Gateway what connections to establish with which hosts - on which ports, etc. The rt_config file tells the Gateway - how to route data coming from "sources" to the appropriate - "destinations." - + connection_config and consumer_config files: + + 1. The connection_config file is used to establish the "physical + configuration." It tells the Gateway what connections + to establish with particular hosts using particular + ports. + + 2. The consumer_config file is used to establish the "logical + configuration." It tells the Gateway how to forward + data coming from "sources" to the appropriate + "destinations." Peer -- The test driver programs that must be started *before* the - Gateway. To do anything interesting you'll need at - least two Peers: one for supplying events and one for consuming - them. In the configuration files, these two types of Peers - are designated as follows: + Gateway. To do anything interesting you'll need at least + two Peers: one to supply events and one to consume events. + In the configuration files, these two types of Peers are + designated as follows: - (1) Input Peers (designated by an "I" in the Gateway's - cc_config configuration file). These Peers are "sources" - of messages to the Gateway. + 1. Supplier Peers (designated by an 'S' in the Gateway's + connection_config configuration file). These Peers are + "suppliers" of events to the Gateway. - (2) Output Peers (designated by an "O" in the Gateway's - cc_config file). These Peers are "destinations" of - messages routed by the Gateway (routing is based on - the settings in the rt_config configuration file). + 2. Consumer Peers (designated by an 'C' in the Gateway's + connection_config file). These Peers are "consumers" of + events forwarded by the Gateway (forwarding is based on + the settings in the consumer_config configuration file). RUNNING THE TESTS @@ -45,39 +54,39 @@ To run the tests do the following: 1. Compile everything (i.e., first compile the ACE libraries, then compile the the Gateway directories). -2. Edit the rt_config and cc_config files as discussed above. +2. Edit the consumer_config and connection_config files as discussed + above to indicate the desired physical and logical mappings. 3. Start up the Peers (peerd). You can start up as many as you - like, as per the cc_config file, but you'll need at least - two (one for supplying input and one for consuming output). I - typically start up each peer in a different window on a different - machine. The peers should print out some diagnostic info and then - block awaiting connections from the Gateway. - -4. Start up the Gateway (gatewayd). This will print out - a bunch of messages as it reads the config files and connects - to all the Peers. Assuming everything works, then all the - Peers will be connected. If some of the Peers aren't set up - correctly then the Gateway will use an exponential backoff - algorithm to attempt to reestablish those connections. + like, as per the connection_config file, but you'll need at least + two (one to supply and one to consume). I typically start up each + Peer in a different window on a different machine. The Peers + should print out some diagnostic info and then block awaiting + connections from the Gateway. + +4. Start up the Gateway (gatewayd). This will print out a bunch of + events as it reads the config files and connects to all the Peers. + Assuming everything works, then all the Peers will be connected. + If some of the Peers aren't set up correctly then the Gateway will + use an exponential backoff algorithm to attempt to reestablish + those connections. 5. Once the Gateway has connected with all the Peers you can send - messages from Input Peers by typing commands in the Peer window. - This input will be sent to the Gateway, which will forward - the message to all Output Peers that have "subscribed" to receive - these messages. + events from Supplier Peers by typing commands in the Peer window. + This Supplier will be sent to the Gateway, which will forward the + event to all Consumer Peers that have "subscribed" to receive these + events. Note that if you type ^C in a Peer window the Peer will shutdown - its handlers and exit. The Gateway will detect this and will - start trying to reestablish the connection using the same - exponential backoff algorithm it used for the initial connection - establishment. + its handlers and exit. The Gateway will detect this and will start + trying to reestablish the connection using the same exponential + backoff algorithm it used for the initial connection establishment. -7. When you want to terminate a Gateway, just type ^C - and the process will shut down gracefully. +7. When you want to terminate a Gateway, just type ^C and the process + will shut down gracefully. Please let me know if there are any questions. Doug -schmidt@cs.wustl.edu + schmidt@cs.wustl.edu diff --git a/apps/Orbix-Examples/Event_Comm/Supplier/Notifier_Handler.h b/apps/Orbix-Examples/Event_Comm/Supplier/Notifier_Handler.h index 293d597a99f..43730693c93 100644 --- a/apps/Orbix-Examples/Event_Comm/Supplier/Notifier_Handler.h +++ b/apps/Orbix-Examples/Event_Comm/Supplier/Notifier_Handler.h @@ -1,7 +1,6 @@ /* -*- C++ -*- */ // $Id$ - // ============================================================================ // // = LIBRARY diff --git a/examples/ASX/CCM_App/CCM_App.cpp b/examples/ASX/CCM_App/CCM_App.cpp index 93335efcc9c..e0330cf53de 100644 --- a/examples/ASX/CCM_App/CCM_App.cpp +++ b/examples/ASX/CCM_App/CCM_App.cpp @@ -14,9 +14,6 @@ class ACE_Svc_Export Test_Task : public MT_Task public: virtual int open (void *); virtual int close (u_long); - virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0); - virtual int svc (void); - virtual int info (char **, size_t) const; virtual int init (int, char *[]); virtual int fini (void); virtual int suspend (void); @@ -52,24 +49,6 @@ Test_Task::resume (void) } int -Test_Task::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - -int -Test_Task::svc (void) -{ - return 0; -} - -int -Test_Task::info (char **, size_t) const -{ - return 0; -} - -int Test_Task::init (int, char *[]) { ACE_DEBUG ((LM_DEBUG, "initializing %s\n", this->name () ? this->name () : "task")); diff --git a/examples/ASX/Event_Server/Event_Server/Event_Analyzer.h b/examples/ASX/Event_Server/Event_Server/Event_Analyzer.h index 7451d38b2b7..498e91d476e 100644 --- a/examples/ASX/Event_Server/Event_Server/Event_Analyzer.h +++ b/examples/ASX/Event_Server/Event_Server/Event_Analyzer.h @@ -19,7 +19,6 @@ public: virtual int open (void *a = 0); virtual int close (u_long flags = 0); virtual int put (ACE_Message_Block *msg, ACE_Time_Value * = 0); - virtual int svc (void) { return 0; } /* Dynamic linking hooks */ virtual int init (int argc, char *argv[]); diff --git a/examples/ASX/Message_Queue/buffer_stream.cpp b/examples/ASX/Message_Queue/buffer_stream.cpp index fbfed5d13ce..cc2c475bef1 100644 --- a/examples/ASX/Message_Queue/buffer_stream.cpp +++ b/examples/ASX/Message_Queue/buffer_stream.cpp @@ -30,12 +30,6 @@ public: // ACE_Task hooks virtual int open (void * = 0); virtual int close (u_long = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0) { return 0; } - - // ACE_Service_Object hooks - virtual int init (int, char **) { return 0; } - virtual int fini (void) { return 0; } - virtual int info (char **, size_t) const { return 0; } }; // Define the Producer interface. diff --git a/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h b/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h index 3dd706eaa6a..a815e7d642a 100644 --- a/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h +++ b/examples/ASX/UPIPE_Event_Server/Event_Analyzer.h @@ -19,7 +19,6 @@ public: virtual int open (void *a = 0); virtual int close (u_long flags = 0); virtual int put (ACE_Message_Block *msg, ACE_Time_Value * = 0); - virtual int svc (void) { return 0; } // Dynamic linking hooks. virtual int init (int argc, char *argv[]); diff --git a/examples/Connection/non_blocking/CPP-connector.h b/examples/Connection/non_blocking/CPP-connector.h index 3cf2a6eb2fb..81be8b6ba71 100644 --- a/examples/Connection/non_blocking/CPP-connector.h +++ b/examples/Connection/non_blocking/CPP-connector.h @@ -34,9 +34,6 @@ protected: // Keeps track of which state we are in. private: - // = Disallow these methods... - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } - virtual int svc (void) { return 0; } }; template <class SVC_HANDLER, ACE_PEER_CONNECTOR_1> diff --git a/examples/Reactor/Misc/test_demuxing.cpp b/examples/Reactor/Misc/test_demuxing.cpp index cacac69f5e6..f8f7992ce37 100644 --- a/examples/Reactor/Misc/test_demuxing.cpp +++ b/examples/Reactor/Misc/test_demuxing.cpp @@ -212,11 +212,6 @@ public: // Run the "event-loop" periodically putting messages to our // internal <Message_Queue> that we inherit from <ACE_Task>. - // = Not used... - virtual int open (void *) { return 0; } - virtual int close (u_long) { return 0; } - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } - private: ACE_Reactor_Notification_Strategy notification_strategy_; // This strategy will notify the <ACE_Reactor> Singleton when a new diff --git a/examples/Reactor/Misc/test_reactors.cpp b/examples/Reactor/Misc/test_reactors.cpp index dbb60941f24..5426af32d2a 100644 --- a/examples/Reactor/Misc/test_reactors.cpp +++ b/examples/Reactor/Misc/test_reactors.cpp @@ -1,6 +1,6 @@ -// Perform a torture test of multiple ACE_Reactors and ACE_Tasks in // $Id$ +// Perform a torture test of multiple ACE_Reactors and ACE_Tasks in // the same process... Thanks to Detlef Becker for contributing this. #include "ace/Reactor.h" @@ -21,7 +21,6 @@ public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0); virtual int svc (void); virtual int handle_input (ACE_HANDLE handle); @@ -29,7 +28,6 @@ public: ACE_Reactor_Mask close_mask); private: - ACE_Reactor *r_; int handled_; static int task_count_; @@ -64,7 +62,7 @@ Test_Task::~Test_Task (void) int Test_Task::open (void *args) { - r_ = (ACE_Reactor *) args; + this->reactor ((ACE_Reactor *) args); return this->activate (THR_NEW_LWP); } @@ -81,13 +79,6 @@ Test_Task::close (u_long) } int -Test_Task::put (ACE_Message_Block *, - ACE_Time_Value *) -{ - return 0; -} - -int Test_Task::svc (void) { for (int i = 0; i < NUM_INVOCATIONS; i++) @@ -96,7 +87,7 @@ Test_Task::svc (void) // ACE_DEBUG ((LM_DEBUG, "(%t) calling notify %d\n", i)); - if (r_->notify (this, ACE_Event_Handler::READ_MASK) == -1) + if (this->reactor ()->notify (this, ACE_Event_Handler::READ_MASK) == -1) ACE_ERROR_RETURN ((LM_ERROR, "(%t) %p\n", "notify"), -1); // ACE_DEBUG ((LM_DEBUG, "(%t) leaving notify %d\n", i)); diff --git a/examples/Reactor/ReactorEx/test_reactorEx.cpp b/examples/Reactor/ReactorEx/test_reactorEx.cpp index 334cfce80aa..488fa1e683a 100644 --- a/examples/Reactor/ReactorEx/test_reactorEx.cpp +++ b/examples/Reactor/ReactorEx/test_reactorEx.cpp @@ -76,9 +76,6 @@ public: // Called when output events should start private: - int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } - // Make Task happy. - ACE_SOCK_Stream stream_; // Socket that we have connected to the server. diff --git a/examples/Reactor/WFMO_Reactor/test_reactorEx.cpp b/examples/Reactor/WFMO_Reactor/test_reactorEx.cpp index 334cfce80aa..488fa1e683a 100644 --- a/examples/Reactor/WFMO_Reactor/test_reactorEx.cpp +++ b/examples/Reactor/WFMO_Reactor/test_reactorEx.cpp @@ -76,9 +76,6 @@ public: // Called when output events should start private: - int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } - // Make Task happy. - ACE_SOCK_Stream stream_; // Socket that we have connected to the server. diff --git a/examples/System_V_IPC/SV_Semaphores/Semaphores_1.cpp b/examples/System_V_IPC/SV_Semaphores/Semaphores_1.cpp new file mode 100644 index 00000000000..8af66b90f34 --- /dev/null +++ b/examples/System_V_IPC/SV_Semaphores/Semaphores_1.cpp @@ -0,0 +1,89 @@ +// $Id$ + +#include "ace/SV_Shared_Memory.h" +#include "ace/SV_Semaphore_Simple.h" +#include "ace/Malloc.h" + +#if defined (ACE_HAS_SYSV_IPC) + +// Shared memory allocator (note that this chews up the +// ACE_DEFAULT_SEM_KEY). +static ACE_Malloc<ACE_SHARED_MEMORY_POOL, ACE_SV_Semaphore_Simple> allocator; + +const int SEM_KEY = ACE_DEFAULT_SEM_KEY + 1; +const int SHMSZ = 27; + +static int +parent (char *shm) +{ + char *s = shm; + + ACE_SV_Semaphore_Complex sem (SEM_KEY, ACE_SV_Semaphore_Complex::ACE_CREATE, 0, 2); + + for (char c = 'a'; c <= 'z'; c++) + *s++ = c; + + *s = '\0'; + + if (sem.release (0) == -1) + ACE_ERROR ((LM_ERROR, "%p", "parent sem.release(0)")); + else if (sem.acquire (1) == -1) + ACE_ERROR ((LM_ERROR, "%p", "parent sem.acquire(1)")); + + if (allocator.remove () == -1) + ACE_ERROR ((LM_ERROR, "%p\n", "allocator.remove")); + if (sem.remove () == -1) + ACE_ERROR ((LM_ERROR, "%p\n", "sem.remove")); + return 0; +} + +static int +child (char *shm) +{ + ACE_SV_Semaphore_Complex sem (SEM_KEY, ACE_SV_Semaphore_Complex::ACE_CREATE, 0, 2); + + while (sem.tryacquire (0) == -1) + if (errno == EAGAIN) + ACE_DEBUG ((LM_DEBUG, "spinning in client!\n")); + else + ACE_ERROR_RETURN ((LM_ERROR, "client mutex.tryacquire(0)"), 1); + + for (char *s = (char *) shm; *s != '\0'; s++) + ACE_DEBUG ((LM_DEBUG, "%c", *s)); + + ACE_DEBUG ((LM_DEBUG, "\n")); + + if (sem.release (1) < 0) + ACE_ERROR ((LM_ERROR, "client sem.release(1)")); + return 0; +} + +int +main (void) +{ + char *shm = (char *) allocator.malloc (27); + + switch (ACE_OS::fork ()) + { + case -1: + ACE_ERROR_RETURN ((LM_ERROR, "fork failed\n"), -1); + /* NOTREACHED */ + case 0: + // Child. + return child (shm); + default: + return parent (shm); + } +} +#else +int main (void) +{ + ACE_ERROR ((LM_ERROR, + "SYSV IPC is not supported on this platform\n")); + return 0; +} +#endif /* ACE_HAS_SYSV_IPC */ + +#if defined (ACE_TEMPLATES_REQUIRE_SPECIALIZATION) +template class ACE_Malloc<ACE_SHARED_MEMORY_POOL, ACE_SV_Semaphore_Simple>; +#endif /* ACE_TEMPLATES_REQUIRE_SPECIALIZATION */ diff --git a/examples/System_V_IPC/SV_Semaphores/Semaphores_2.cpp b/examples/System_V_IPC/SV_Semaphores/Semaphores_2.cpp new file mode 100644 index 00000000000..0405440364c --- /dev/null +++ b/examples/System_V_IPC/SV_Semaphores/Semaphores_2.cpp @@ -0,0 +1,100 @@ +// $Id$ + +// Illustrates the use of the ACE_SV_Semaphore_Complex class and the +// ACE_Malloc class using the ACE_Shared_Memory_Pool (which uses +// System V shared memory). Note that it doesn't matter whether the +// parent or the child creates the semaphore since Semaphore_Complex +// will correctly serialize the intialization of the mutex and synch +// objects. + +#include "ace/Malloc.h" +#include "ace/SV_Semaphore_Complex.h" + +#if defined (ACE_HAS_SYSV_IPC) + +// Shared memory allocator (note that this chews up the +// ACE_DEFAULT_SEM_KEY). +static ACE_Malloc<ACE_SHARED_MEMORY_POOL, ACE_SV_Semaphore_Simple> allocator; + +const int SEM_KEY_1 = ACE_DEFAULT_SEM_KEY + 1; +const int SEM_KEY_2 = ACE_DEFAULT_SEM_KEY + 2; +const int SHMSZ = 27; + +static int +parent (char *shm) +{ + char *s = shm; + + ACE_SV_Semaphore_Complex mutex (SEM_KEY_1, ACE_SV_Semaphore_Complex::ACE_CREATE, 0); + ACE_SV_Semaphore_Complex synch (SEM_KEY_2, ACE_SV_Semaphore_Complex::ACE_CREATE, 0); + + for (char c = 'a'; c <= 'z'; c++) + *s++ = c; + + *s = '\0'; + + if (mutex.release () == -1) + ACE_ERROR ((LM_ERROR, "%p", "parent mutex.release")); + else if (synch.acquire () == -1) + ACE_ERROR ((LM_ERROR, "%p", "parent synch.acquire")); + + if (allocator.remove () == -1) + ACE_ERROR ((LM_ERROR, "%p\n", "allocator.remove")); + if (mutex.remove () == -1) + ACE_ERROR ((LM_ERROR, "%p\n", "mutex.remove")); + if (synch.remove () == -1) + ACE_ERROR ((LM_ERROR, "%p\n", "synch.remove")); + return 0; +} + +static int +child (char *shm) +{ + ACE_SV_Semaphore_Complex mutex (SEM_KEY_1, ACE_SV_Semaphore_Complex::ACE_CREATE, 0); + ACE_SV_Semaphore_Complex synch (SEM_KEY_2, ACE_SV_Semaphore_Complex::ACE_CREATE, 0); + + while (mutex.tryacquire () == -1) + if (errno == EAGAIN) + ACE_DEBUG ((LM_DEBUG, "spinning in child!\n")); + else + ACE_ERROR_RETURN ((LM_ERROR, "child mutex.tryacquire"), 1); + + for (char *s = (char *) shm; *s != '\0'; s++) + ACE_DEBUG ((LM_DEBUG, "%c", *s)); + + ACE_DEBUG ((LM_DEBUG, "\n")); + + if (synch.release () == -1) + ACE_ERROR_RETURN ((LM_ERROR, "child synch.release"), 1); + return 0; +} + +int +main (void) +{ + char *shm = (char *) allocator.malloc (27); + + switch (ACE_OS::fork ()) + { + case -1: + ACE_ERROR_RETURN ((LM_ERROR, "fork failed\n"), -1); + /* NOTREACHED */ + case 0: + // Child. + return child (shm); + default: + return parent (shm); + } +} +#else +int main (void) +{ + ACE_ERROR ((LM_ERROR, + "SYSV IPC is not supported on this platform\n")); + return 0; +} +#endif /* ACE_HAS_SYSV_IPC */ + +#if defined (ACE_TEMPLATES_REQUIRE_SPECIALIZATION) +template class ACE_Malloc<ACE_SHARED_MEMORY_POOL, ACE_SV_Semaphore_Simple>; +#endif /* ACE_TEMPLATES_REQUIRE_SPECIALIZATION */ diff --git a/examples/Threads/future1.cpp b/examples/Threads/future1.cpp index ea295e487e1..00814fee881 100644 --- a/examples/Threads/future1.cpp +++ b/examples/Threads/future1.cpp @@ -57,7 +57,6 @@ public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0); virtual int svc (void); ACE_Future<double> work (double param, int count); @@ -196,13 +195,6 @@ Scheduler::close (u_long) return 0; } -// put... ?? -int -Scheduler::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - // service.. int Scheduler::svc (void) diff --git a/examples/Threads/future2.cpp b/examples/Threads/future2.cpp index 55ce8c05a40..f034d765c1e 100644 --- a/examples/Threads/future2.cpp +++ b/examples/Threads/future2.cpp @@ -73,10 +73,6 @@ private: virtual int close (u_long flags = 0); // Should not be accessible from outside... (use end () instead). - virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0) - { return 0; }; - // Doesn't have any use for this example. - virtual int svc (void); // Here the actual servicing of all requests is happening.. diff --git a/examples/Threads/task_four.cpp b/examples/Threads/task_four.cpp index 64209cb3430..87a2f3abfa8 100644 --- a/examples/Threads/task_four.cpp +++ b/examples/Threads/task_four.cpp @@ -55,11 +55,6 @@ private: // Number of threads per task. int n_iterations_; // Number of iterations per thread. - - // = Not needed for this test. - virtual int open (void *) { return 0; } - virtual int close (u_long) { return 0; } - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } }; class Worker_Task : public ACE_Task<ACE_MT_SYNCH> diff --git a/examples/Threads/task_one.cpp b/examples/Threads/task_one.cpp index d0a8a12e6c4..7acf8de3789 100644 --- a/examples/Threads/task_one.cpp +++ b/examples/Threads/task_one.cpp @@ -31,11 +31,6 @@ private: int n_iterations_; // Number of iterations to run. - - // = Not needed for this test. - virtual int open (void *) { return 0; } - virtual int close (u_long) { return 0; } - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } }; Barrier_Task::Barrier_Task (ACE_Thread_Manager *thr_mgr, diff --git a/examples/Threads/task_three.cpp b/examples/Threads/task_three.cpp index 0214ac10ddf..4b365aba8a5 100644 --- a/examples/Threads/task_three.cpp +++ b/examples/Threads/task_three.cpp @@ -33,7 +33,6 @@ public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0); virtual int svc (void); virtual int handle_input (ACE_HANDLE fd); @@ -86,12 +85,6 @@ Test_Task::close (u_long) return 0; } -int -Test_Task::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - Test_Task::svc (void) { // Every thread must register the same stream to write to file. diff --git a/examples/Threads/task_two.cpp b/examples/Threads/task_two.cpp index 1c6366c4b12..c6e88d8eaa5 100644 --- a/examples/Threads/task_two.cpp +++ b/examples/Threads/task_two.cpp @@ -31,7 +31,6 @@ class Task_Test : public ACE_Task<ACE_MT_SYNCH> public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0); virtual int svc (void); private: @@ -66,13 +65,6 @@ Task_Test::close (u_long) } int -Task_Test::put (ACE_Message_Block *, - ACE_Time_Value *) -{ - return 0; -} - -int Task_Test::svc (void) { wait_count++; diff --git a/examples/Threads/thread_pool.cpp b/examples/Threads/thread_pool.cpp index 9478ed0883d..ddcad02e4dd 100644 --- a/examples/Threads/thread_pool.cpp +++ b/examples/Threads/thread_pool.cpp @@ -34,9 +34,6 @@ public: private: virtual int close (u_long); - - // = Not needed for this test. - virtual int open (void *) { return 0; } }; int diff --git a/examples/Threads/token.cpp b/examples/Threads/token.cpp index 5a51496d011..e8c72be3d1d 100644 --- a/examples/Threads/token.cpp +++ b/examples/Threads/token.cpp @@ -10,9 +10,6 @@ class My_Task : public ACE_Task<ACE_MT_SYNCH> { public: My_Task (int n); - virtual int open (void *) { return 0; } - virtual int close (u_long) { return 0; } - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } virtual int svc (void); static void sleep_hook (void *); diff --git a/examples/Threads/tss1.cpp b/examples/Threads/tss1.cpp index 7efdc9dc3ef..dd21023fa19 100644 --- a/examples/Threads/tss1.cpp +++ b/examples/Threads/tss1.cpp @@ -96,9 +96,6 @@ public: virtual int open (void *theArgs = 0); virtual int close (u_long theArg = 0); - virtual int put (ACE_Message_Block *theMsgBlock, - ACE_Time_Value *theTimeVal = 0); - virtual int svc (void); }; template <ACE_SYNCH_1> int @@ -119,18 +116,6 @@ int Tester<ACE_SYNCH_2>::close (u_long) return 0; } -template <ACE_SYNCH_1> int -Tester<ACE_SYNCH_2>::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - -template <ACE_SYNCH_1> int -Tester<ACE_SYNCH_2>::svc (void) -{ - return 0; -} - int main (int, char *[]) { diff --git a/netsvcs/servers/svc.conf b/netsvcs/servers/svc.conf index f8634065b13..b136e97bb33 100644 --- a/netsvcs/servers/svc.conf +++ b/netsvcs/servers/svc.conf @@ -1,5 +1,5 @@ # These are the services that can be linked into ACE. -# Note that you can replace the hardcoded "../lib/libnet_svcs" with +# Note that you can replace the hardcoded "../lib/net_svcs" with # a relative path if you set your LD search path correctly -- ACE will # locate this for you automatically by reading your LD search path. # Moreover, ACE will automatically insert the correct suffix (e.g., @@ -7,11 +7,11 @@ # "-p 20xxx" with "-p $PORTxxx" if you set your environment variables # correctly. -dynamic Logger Service_Object * ../lib/libnet_svcs:_make_ACE_Logger() "-s foobar -f STDERR|OSTREAM" -dynamic Time_Service Service_Object * ../lib/libnet_svcs:_make_ACE_TS_Server_Acceptor() "-p 10222" -dynamic Name_Server Service_Object * ../lib/libnet_svcs:_make_ACE_Name_Acceptor() "-p 10012" -dynamic Token_Service Service_Object * ../lib/libnet_svcs:_make_ACE_Token_Acceptor() "-p 10202" -dynamic Server_Logging_Service Service_Object * ../lib/libnet_svcs:_make_ACE_Server_Logging_Acceptor() active "-p 10009" -dynamic Thr_Server_Logging_Service Service_Object * ../lib/libnet_svcs:_make_ACE_Thr_Server_Logging_Acceptor() active "-p 10020" -dynamic Client_Logging_Service Service_Object * ../lib/libnet_svcs:_make_ACE_Client_Logging_Connector() active "-p 10009" +dynamic Logger Service_Object * ../lib/net_svcs:_make_ACE_Logger() "-s foobar -f STDERR|OSTREAM" +dynamic Time_Service Service_Object * ../lib/net_svcs:_make_ACE_TS_Server_Acceptor() "-p 10222" +dynamic Name_Server Service_Object * ../lib/net_svcs:_make_ACE_Name_Acceptor() "-p 10012" +dynamic Token_Service Service_Object * ../lib/net_svcs:_make_ACE_Token_Acceptor() "-p 10202" +dynamic Server_Logging_Service Service_Object * ../lib/net_svcs:_make_ACE_Server_Logging_Acceptor() active "-p 10009" +dynamic Thr_Server_Logging_Service Service_Object * ../lib/net_svcs:_make_ACE_Thr_Server_Logging_Acceptor() active "-p 10020" +dynamic Client_Logging_Service Service_Object * ../lib/net_svcs:_make_ACE_Client_Logging_Connector() active "-p 10009" diff --git a/tests/Buffer_Stream_Test.cpp b/tests/Buffer_Stream_Test.cpp index a228d94e783..aa5ec27414c 100644 --- a/tests/Buffer_Stream_Test.cpp +++ b/tests/Buffer_Stream_Test.cpp @@ -46,12 +46,6 @@ public: // ACE_Task hooks virtual int open (void * = 0); virtual int close (u_long = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0) { return 0; } - - // ACE_Service_Object hooks - virtual int init (int, char **) { return 0; } - virtual int fini (void) { return 0; } - virtual int info (char **, size_t) const { return 0; } }; // Define the Producer interface. diff --git a/tests/Future_Test.cpp b/tests/Future_Test.cpp index 00ed154ebbf..8207ede913c 100644 --- a/tests/Future_Test.cpp +++ b/tests/Future_Test.cpp @@ -58,7 +58,6 @@ public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0); virtual int svc (void); ACE_Future<double> work (double param, int count); @@ -196,13 +195,6 @@ Scheduler::close (u_long) return 0; } -// put... ?? -int -Scheduler::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - // service.. int Scheduler::svc (void) diff --git a/tests/Priority_Task_Test.cpp b/tests/Priority_Task_Test.cpp index 5444e5b987c..bcc52856f43 100644 --- a/tests/Priority_Task_Test.cpp +++ b/tests/Priority_Task_Test.cpp @@ -29,7 +29,6 @@ class Priority_Task : public ACE_Task<ACE_MT_SYNCH> public: Priority_Task (void); - int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } int close (u_long = 0); int open (void *); int svc (void); @@ -65,7 +64,7 @@ Priority_Task::svc (void) ACE_hthread_t thr_handle; ACE_Thread::self (thr_handle); int prio; - ACE_Thread::getprio (thr_handle, &prio); + ACE_Thread::getprio (thr_handle, prio); ACE_ASSERT (this->priority_ == prio); return 0; } diff --git a/tests/Reactors_Test.cpp b/tests/Reactors_Test.cpp index 8bad943910d..2c459cdbf15 100644 --- a/tests/Reactors_Test.cpp +++ b/tests/Reactors_Test.cpp @@ -36,7 +36,6 @@ public: virtual int open (void *args = 0); virtual int close (u_long flags = 0); - virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0); virtual int svc (void); virtual int handle_input (ACE_HANDLE handle); @@ -97,12 +96,6 @@ Test_Task::close (u_long) } int -Test_Task::put (ACE_Message_Block *, ACE_Time_Value *) -{ - return 0; -} - -int Test_Task::svc (void) { ACE_NEW_THREAD; diff --git a/tests/Task_Test.cpp b/tests/Task_Test.cpp index be1256c3023..27c4d1bd98e 100644 --- a/tests/Task_Test.cpp +++ b/tests/Task_Test.cpp @@ -43,11 +43,6 @@ private: int n_iterations_; // Number of iterations to run. - - // = Not needed for this test. - virtual int open (void *) { return 0; } - virtual int close (u_long) { return 0; } - virtual int put (ACE_Message_Block *, ACE_Time_Value *) { return 0; } }; Barrier_Task::Barrier_Task (ACE_Thread_Manager *thr_mgr, diff --git a/tests/Thread_Pool_Test.cpp b/tests/Thread_Pool_Test.cpp index cb7d9069b9a..2363850e7f3 100644 --- a/tests/Thread_Pool_Test.cpp +++ b/tests/Thread_Pool_Test.cpp @@ -41,7 +41,7 @@ public: // Iterate <n_iterations> time printing off a message and "waiting" // for all other threads to complete this iteration. - virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv=0); + virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0); // This allows the producer to pass messages to the <Thread_Pool>. private: diff --git a/tests/UPIPE_SAP_Test.cpp b/tests/UPIPE_SAP_Test.cpp index f1036c563cf..64a851c28bf 100644 --- a/tests/UPIPE_SAP_Test.cpp +++ b/tests/UPIPE_SAP_Test.cpp @@ -100,17 +100,6 @@ acceptor (void *args) ACE_UPIPE_Acceptor *acceptor = (ACE_UPIPE_Acceptor *) args; ACE_UPIPE_Stream s_stream; - ACE_hthread_t thr_handle; - - // Spawn a connector thread. - if (ACE_Thread::spawn (ACE_THR_FUNC (connector), - (void *) 0, - THR_NEW_LWP | THR_DETACHED, - 0, - &thr_handle) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 0); - - ACE_DEBUG ((LM_DEBUG, "(%t) acceptor starting accept\n")); if (acceptor->accept (s_stream) == -1) ACE_DEBUG ((LM_DEBUG, @@ -163,7 +152,7 @@ main (int, char *[]) // Spawn a acceptor thread. if (ACE_Thread::spawn (ACE_THR_FUNC (acceptor), (void *) &acc, - THR_NEW_LWP | THR_DETACHED, + THR_NEW_LWP, 0, &thr_handle_acceptor) == -1) ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1); @@ -171,15 +160,22 @@ main (int, char *[]) // Spawn a connector thread. if (ACE_Thread::spawn (ACE_THR_FUNC (connector), (void *) 0, - THR_NEW_LWP | THR_DETACHED, + THR_NEW_LWP, 0, &thr_handle_connector) == -1) ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "spawn"), 1); // Wait for both the acceptor and connector threads to exit. - ACE_Thread::join (thr_handle_connector); - ACE_Thread::join (thr_handle_acceptor); + if (ACE_Thread::join (thr_handle_connector) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "join"), -1); + else + ACE_DEBUG ((LM_DEBUG, "(%t) joined with connector thread\n")); + if (ACE_Thread::join (thr_handle_acceptor) == -1) + ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "join"), -1); + else + ACE_DEBUG ((LM_DEBUG, "(%t) joined with acceptor thread\n")); + #else ACE_ERROR ((LM_ERROR, "threads and/or UPIPE not supported on this platform\n")); #endif /* ACE_HAS_THREADS */ diff --git a/tests/run_tests.sh b/tests/run_tests.sh index 92ac675fa09..830260bbb6c 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -66,7 +66,7 @@ run Reader_Writer_Test # uses Thread_Manager, Mutex # ifdef ACE_HAS_STREAM_PIPES run SPIPE_Test # uses SPIPE_Acceptor/Connector, Thread_Manager -# run UPIPE_SAP_Test # uses UPIPE, Thread, Thread_Manager +run UPIPE_SAP_Test # uses UPIPE, Thread, Thread_Manager run Barrier_Test # uses Service_Config, Barrier run Buffer_Stream_Test # uses Service_Config, Module (Stream,Task, Message_Queue) |