1 // $Id: Select_Reactor_Base.cpp 81153 2008-03-29 08:17:58Z johnnyw $
3 #include "ace/Select_Reactor_Base.h"
4 #include "ace/Reactor.h"
5 #include "ace/Thread.h"
6 #include "ace/SOCK_Acceptor.h"
7 #include "ace/SOCK_Connector.h"
8 #include "ace/Timer_Queue.h"
9 #include "ace/Log_Msg.h"
10 #include "ace/Signal.h"
11 #include "ace/OS_NS_fcntl.h"
13 #if !defined (__ACE_INLINE__)
14 #include "ace/Select_Reactor_Base.inl"
15 #endif /* __ACE_INLINE__ */
19 #endif /* !ACE_WIN32 */
23 "$Id: Select_Reactor_Base.cpp 81153 2008-03-29 08:17:58Z johnnyw $")
26 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
28 template<typename iterator
>
29 inline ACE_Event_Handler
*
30 ACE_SELECT_REACTOR_EVENT_HANDLER (iterator i
)
36 #endif /* ACE_WIN32 */
39 // Performs sanity checking on the ACE_HANDLE.
42 ACE_Select_Reactor_Handler_Repository::invalid_handle (ACE_HANDLE handle
)
44 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::invalid_handle");
45 #if defined (ACE_WIN32)
46 // It's too expensive to perform more exhaustive validity checks on
47 // Win32 due to the way that they implement SOCKET HANDLEs.
48 if (handle
== ACE_INVALID_HANDLE
)
49 #else /* !ACE_WIN32 */
51 || static_cast<size_type
> (handle
) >= this->event_handlers_
.size ())
52 #endif /* ACE_WIN32 */
61 // Performs sanity checking on the ACE_HANDLE.
64 ACE_Select_Reactor_Handler_Repository::handle_in_range (ACE_HANDLE handle
)
66 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::handle_in_range");
67 #if defined (ACE_WIN32)
68 // It's too expensive to perform more exhaustive validity checks on
69 // Win32 due to the way that they implement SOCKET HANDLEs.
70 if (handle
!= ACE_INVALID_HANDLE
)
71 #else /* !ACE_WIN32 */
72 if (handle
>= 0 && handle
< this->max_handlep1_
)
73 #endif /* ACE_WIN32 */
78 // Don't bother setting errno. It isn't used in the select()-based
79 // reactors and incurs a TSS access.
86 ACE_Select_Reactor_Handler_Repository::open (size_t size
)
88 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::open");
90 #if defined (ACE_WIN32)
91 if (this->event_handlers_
.open (size
) == -1)
94 if (this->event_handlers_
.size (size
) == -1)
97 // Initialize the ACE_Event_Handler pointers to 0.
98 std::fill (this->event_handlers_
.begin (),
99 this->event_handlers_
.end (),
100 static_cast<ACE_Event_Handler
*> (0));
102 this->max_handlep1_
= 0;
103 #endif /* ACE_WIN32 */
105 // Try to increase the number of handles if <size> is greater than
106 // the current limit.
107 return ACE::set_handle_limit (static_cast<int> (size
), 1);
110 // Initialize a repository of the appropriate <size>.
112 ACE_Select_Reactor_Handler_Repository::ACE_Select_Reactor_Handler_Repository (ACE_Select_Reactor_Impl
&select_reactor
)
113 : select_reactor_ (select_reactor
),
116 #endif /* !ACE_WIN32 */
119 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::ACE_Select_Reactor_Handler_Repository");
123 ACE_Select_Reactor_Handler_Repository::unbind_all (void)
125 // Unbind all of the <handle, ACE_Event_Handler>s.
127 map_type::iterator
const end
= this->event_handlers_
.end ();
128 for (map_type::iterator pos
= this->event_handlers_
.begin ();
132 // Post-increment (*not* pre-increment) before unbind()ing since
133 // the current iterator will be invalidated during the unbind()
135 map_type::iterator
const the_pos (pos
++);
137 ACE_HANDLE
const handle
= (*the_pos
).key ();
138 (void) this->unbind (handle
,
140 ACE_Event_Handler::ALL_EVENTS_MASK
);
143 // We could use the "end()" iterator but leveraging max_handlep1_
144 // allows us to optimize away unnecessary accesses of nil event
146 map_type::iterator pos
=
147 this->event_handlers_
.begin (); // iterator == ACE_Event_Handler*
149 for (ACE_HANDLE handle
= 0;
150 handle
< this->max_handlep1_
;
153 (void) this->unbind (handle
,
155 ACE_Event_Handler::ALL_EVENTS_MASK
);
158 #endif /* ACE_WIN32 */
164 ACE_Select_Reactor_Handler_Repository::close (void)
166 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::close");
168 return this->unbind_all ();
171 ACE_Select_Reactor_Handler_Repository::map_type::iterator
172 ACE_Select_Reactor_Handler_Repository::find_eh (ACE_HANDLE handle
)
174 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::find_eh");
176 map_type::iterator
pos (this->event_handlers_
.end ());
178 // this code assumes the handle is in range.
179 #if defined (ACE_WIN32)
180 this->event_handlers_
.find (handle
, pos
);
182 map_type::iterator
const tmp
= &this->event_handlers_
[handle
];
186 #endif /* ACE_WIN32 */
191 // Bind the <ACE_Event_Handler *> to the <ACE_HANDLE>.
193 ACE_Select_Reactor_Handler_Repository::bind (ACE_HANDLE handle
,
194 ACE_Event_Handler
*event_handler
,
195 ACE_Reactor_Mask mask
)
197 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::bind");
199 if (event_handler
== 0)
202 if (handle
== ACE_INVALID_HANDLE
)
203 handle
= event_handler
->get_handle ();
205 if (this->invalid_handle (handle
))
208 // Is this handle already in the Reactor?
209 bool existing_handle
= false;
211 #if defined (ACE_WIN32)
213 map_type::ENTRY
* entry
= 0;
216 this->event_handlers_
.bind (handle
, event_handler
, entry
);
222 else if (result
== 1) // Entry already exists.
224 // Cannot use a different handler for an existing handle.
225 if (event_handler
!= entry
->item ())
231 // Remember that this handle is already registered in the
233 existing_handle
= true;
239 // Check if this handle is already registered.
240 ACE_Event_Handler
* const current_handler
=
241 this->event_handlers_
[handle
];
245 // Cannot use a different handler for an existing handle.
246 if (current_handler
!= event_handler
)
249 // Remember that this handle is already registered in the
251 existing_handle
= true;
254 this->event_handlers_
[handle
] = event_handler
;
256 if (this->max_handlep1_
< handle
+ 1)
257 this->max_handlep1_
= handle
+ 1;
259 #endif /* ACE_WIN32 */
261 if (this->select_reactor_
.is_suspended_i (handle
))
263 this->select_reactor_
.bit_ops (handle
,
265 this->select_reactor_
.suspend_set_
,
266 ACE_Reactor::ADD_MASK
);
270 this->select_reactor_
.bit_ops (handle
,
272 this->select_reactor_
.wait_set_
,
273 ACE_Reactor::ADD_MASK
);
275 // Note the fact that we've changed the state of the <wait_set_>,
276 // which is used by the dispatching loop to determine whether it can
277 // keep going or if it needs to reconsult select().
278 // this->select_reactor_.state_changed_ = 1;
281 // If new entry, call add_reference() if needed.
282 if (!existing_handle
)
283 event_handler
->add_reference ();
288 // Remove the binding of <ACE_HANDLE>.
291 ACE_Select_Reactor_Handler_Repository::unbind (
293 map_type::iterator pos
,
294 ACE_Reactor_Mask mask
)
296 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::unbind");
298 // Retrieve event handler before unbinding it from the map. The
299 // iterator pointing to it will no longer be valid once the handler
301 ACE_Event_Handler
* const event_handler
=
302 (pos
== this->event_handlers_
.end ()
304 : ACE_SELECT_REACTOR_EVENT_HANDLER (pos
));
306 // Clear out the <mask> bits in the Select_Reactor's wait_set.
307 this->select_reactor_
.bit_ops (handle
,
309 this->select_reactor_
.wait_set_
,
310 ACE_Reactor::CLR_MASK
);
313 this->select_reactor_
.bit_ops (handle
,
315 this->select_reactor_
.suspend_set_
,
316 ACE_Reactor::CLR_MASK
);
318 // Note the fact that we've changed the state of the <wait_set_>,
319 // which is used by the dispatching loop to determine whether it can
320 // keep going or if it needs to reconsult select().
321 // this->select_reactor_.state_changed_ = 1;
323 // If there are no longer any outstanding events on this <handle>
324 // then we can totally shut down the Event_Handler.
326 bool const has_any_wait_mask
=
327 (this->select_reactor_
.wait_set_
.rd_mask_
.is_set (handle
)
328 || this->select_reactor_
.wait_set_
.wr_mask_
.is_set (handle
)
329 || this->select_reactor_
.wait_set_
.ex_mask_
.is_set (handle
));
330 bool const has_any_suspend_mask
=
331 (this->select_reactor_
.suspend_set_
.rd_mask_
.is_set (handle
)
332 || this->select_reactor_
.suspend_set_
.wr_mask_
.is_set (handle
)
333 || this->select_reactor_
.suspend_set_
.ex_mask_
.is_set (handle
));
335 bool complete_removal
= false;
337 if (!has_any_wait_mask
&& !has_any_suspend_mask
)
339 #if defined (ACE_WIN32)
340 if (event_handler
!= 0 && this->event_handlers_
.unbind (pos
) == -1)
341 return -1; // Should not happen!
343 this->event_handlers_
[handle
] = 0;
345 if (this->max_handlep1_
== handle
+ 1)
347 // We've deleted the last entry, so we need to figure out
348 // the last valid place in the array that is worth looking
350 ACE_HANDLE
const wait_rd_max
=
351 this->select_reactor_
.wait_set_
.rd_mask_
.max_set ();
352 ACE_HANDLE
const wait_wr_max
=
353 this->select_reactor_
.wait_set_
.wr_mask_
.max_set ();
354 ACE_HANDLE
const wait_ex_max
=
355 this->select_reactor_
.wait_set_
.ex_mask_
.max_set ();
357 ACE_HANDLE
const suspend_rd_max
=
358 this->select_reactor_
.suspend_set_
.rd_mask_
.max_set ();
359 ACE_HANDLE
const suspend_wr_max
=
360 this->select_reactor_
.suspend_set_
.wr_mask_
.max_set ();
361 ACE_HANDLE
const suspend_ex_max
=
362 this->select_reactor_
.suspend_set_
.ex_mask_
.max_set ();
364 // Compute the maximum of six values.
365 this->max_handlep1_
= wait_rd_max
;
366 if (this->max_handlep1_
< wait_wr_max
)
367 this->max_handlep1_
= wait_wr_max
;
368 if (this->max_handlep1_
< wait_ex_max
)
369 this->max_handlep1_
= wait_ex_max
;
371 if (this->max_handlep1_
< suspend_rd_max
)
372 this->max_handlep1_
= suspend_rd_max
;
373 if (this->max_handlep1_
< suspend_wr_max
)
374 this->max_handlep1_
= suspend_wr_max
;
375 if (this->max_handlep1_
< suspend_ex_max
)
376 this->max_handlep1_
= suspend_ex_max
;
378 ++this->max_handlep1_
;
381 #endif /* ACE_WIN32 */
383 // The handle has been completely removed.
384 complete_removal
= true;
387 if (event_handler
== 0)
390 bool const requires_reference_counting
=
391 event_handler
->reference_counting_policy ().value () ==
392 ACE_Event_Handler::Reference_Counting_Policy::ENABLED
;
394 // Close down the <Event_Handler> unless we've been instructed not
396 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::DONT_CALL
) == 0)
397 (void) event_handler
->handle_close (handle
, mask
);
399 // Call remove_reference() if the removal is complete and reference
400 // counting is needed.
401 if (complete_removal
&& requires_reference_counting
)
403 (void) event_handler
->remove_reference ();
409 ACE_Select_Reactor_Handler_Repository_Iterator::ACE_Select_Reactor_Handler_Repository_Iterator
410 (ACE_Select_Reactor_Handler_Repository
const * s
)
412 current_ (s
->event_handlers_
.begin ())
415 // Don't use ACE_Array_Base::end() since it may be larger than
416 // event_handlers[max_handlep1_].
417 const_base_iterator
const end
=
418 &this->rep_
->event_handlers_
[this->rep_
->max_handlep1 ()];
420 // Advance to the next element containing a non-zero event handler.
421 // There's no need to do this for the Windows case since the hash
422 // map will only contain non-zero event handlers.
423 while (this->current_
!= end
&& (*(this->current_
) == 0))
428 // Pass back the <next_item> that hasn't been seen in the Set.
429 // Returns 0 when all items have been seen, else 1.
432 ACE_Select_Reactor_Handler_Repository_Iterator::next (
433 ACE_Event_Handler
*&next_item
)
440 next_item
= ACE_SELECT_REACTOR_EVENT_HANDLER (this->current_
);
445 // Move forward by one element in the set.
448 ACE_Select_Reactor_Handler_Repository_Iterator::advance (void)
451 // No need to explicitly limit search to "current" to
452 // max_handlep1_ range.
453 const_base_iterator
const end
= this->rep_
->event_handlers_
.end ();
455 // Don't use ACE_Array_Base::end() since it may be larger than
456 // event_handlers[max_handlep1_].
457 const_base_iterator
const end
=
458 &this->rep_
->event_handlers_
[this->rep_
->max_handlep1 ()];
459 #endif /* ACE_WIN32 */
461 if (this->current_
!= end
)
465 // Advance to the next element containing a non-zero event handler.
466 // There's no need to do this for the Windows case since the hash
467 // map will only contain non-zero event handlers.
468 while (this->current_
!= end
&& (*(this->current_
) == 0))
470 #endif /* !ACE_WIN32 */
472 return this->current_
!= end
;
475 // Dump the state of an object.
478 ACE_Select_Reactor_Handler_Repository_Iterator::dump (void) const
480 #if defined (ACE_HAS_DUMP)
481 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository_Iterator::dump");
483 ACE_DEBUG ((LM_DEBUG
, ACE_BEGIN_DUMP
, this));
484 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("rep_ = %u"), this->rep_
));
486 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("current_ = ")));
487 this->current_
.dump ();
489 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("current_ = %@"), this->current_
));
490 # endif /* ACE_WIN32 */
491 ACE_DEBUG ((LM_DEBUG
, ACE_END_DUMP
));
492 #endif /* ACE_HAS_DUMP */
496 ACE_Select_Reactor_Handler_Repository::dump (void) const
498 #if defined (ACE_HAS_DUMP)
499 ACE_TRACE ("ACE_Select_Reactor_Handler_Repository::dump");
502 # define ACE_HANDLE_FORMAT_SPECIFIER ACE_TEXT("%@")
503 # define ACE_MAX_HANDLEP1_FORMAT_SPECIFIER ACE_TEXT("%u")
505 # define ACE_HANDLE_FORMAT_SPECIFIER ACE_TEXT("%d")
506 # define ACE_MAX_HANDLEP1_FORMAT_SPECIFIER ACE_TEXT("%d")
507 # endif /* ACE_WIN32 */
509 ACE_DEBUG ((LM_DEBUG
, ACE_BEGIN_DUMP
, this));
510 ACE_DEBUG ((LM_DEBUG
,
511 ACE_TEXT ("max_handlep1_ = ")
512 ACE_MAX_HANDLEP1_FORMAT_SPECIFIER
514 this->max_handlep1 ()));
515 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("[")));
517 ACE_Event_Handler
*event_handler
= 0;
519 for (ACE_Select_Reactor_Handler_Repository_Iterator
iter (this);
520 iter
.next (event_handler
) != 0;
522 ACE_DEBUG ((LM_DEBUG
,
523 ACE_TEXT (" (event_handler = %@,")
524 ACE_TEXT (" event_handler->handle_ = ")
525 ACE_HANDLE_FORMAT_SPECIFIER
528 event_handler
->get_handle ()));
530 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT (" ]\n")));
531 ACE_DEBUG ((LM_DEBUG
, ACE_END_DUMP
));
532 #endif /* ACE_HAS_DUMP */
535 ACE_ALLOC_HOOK_DEFINE(ACE_Select_Reactor_Handler_Repository_Iterator
)
537 ACE_Select_Reactor_Notify::ACE_Select_Reactor_Notify (void)
538 : max_notify_iterations_ (-1)
542 ACE_Select_Reactor_Notify::~ACE_Select_Reactor_Notify (void)
547 ACE_Select_Reactor_Notify::max_notify_iterations (int iterations
)
549 // Must always be > 0 or < 0 to optimize the loop exit condition.
553 this->max_notify_iterations_
= iterations
;
557 ACE_Select_Reactor_Notify::max_notify_iterations (void)
559 return this->max_notify_iterations_
;
562 // purge_pending_notifications
563 // Removes all entries from the notify_queue_ and each one that
564 // matches <eh> is put on the free_queue_. The rest are saved on a
565 // local queue and copied back to the notify_queue_ at the end.
566 // Returns the number of entries removed. Returns -1 on error.
567 // ACE_NOTSUP_RETURN if ACE_HAS_REACTOR_NOTIFICATION_QUEUE is not defined.
569 ACE_Select_Reactor_Notify::purge_pending_notifications (ACE_Event_Handler
*eh
,
570 ACE_Reactor_Mask mask
)
572 ACE_TRACE ("ACE_Select_Reactor_Notify::purge_pending_notifications");
574 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
576 return notification_queue_
.purge_pending_notifications(eh
, mask
);
578 #else /* defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE) */
580 ACE_UNUSED_ARG (mask
);
581 ACE_NOTSUP_RETURN (-1);
582 #endif /* defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE) */
586 ACE_Select_Reactor_Notify::dump (void) const
588 #if defined (ACE_HAS_DUMP)
589 ACE_TRACE ("ACE_Select_Reactor_Notify::dump");
591 ACE_DEBUG ((LM_DEBUG
, ACE_BEGIN_DUMP
, this));
592 ACE_DEBUG ((LM_DEBUG
, ACE_TEXT ("select_reactor_ = %x"), this->select_reactor_
));
593 this->notification_pipe_
.dump ();
594 ACE_DEBUG ((LM_DEBUG
, ACE_END_DUMP
));
595 #endif /* ACE_HAS_DUMP */
599 ACE_Select_Reactor_Notify::open (ACE_Reactor_Impl
*r
,
601 int disable_notify_pipe
)
603 ACE_TRACE ("ACE_Select_Reactor_Notify::open");
605 if (disable_notify_pipe
== 0)
607 this->select_reactor_
=
608 dynamic_cast<ACE_Select_Reactor_Impl
*> (r
);
610 if (select_reactor_
== 0)
616 if (this->notification_pipe_
.open () == -1)
618 #if defined (F_SETFD)
619 ACE_OS::fcntl (this->notification_pipe_
.read_handle (), F_SETFD
, 1);
620 ACE_OS::fcntl (this->notification_pipe_
.write_handle (), F_SETFD
, 1);
623 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
624 if (notification_queue_
.open() == -1)
628 #endif /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
630 // There seems to be a Win32 bug with this... Set this into
631 // non-blocking mode.
632 if (ACE::set_flags (this->notification_pipe_
.read_handle (),
636 return this->select_reactor_
->register_handler
637 (this->notification_pipe_
.read_handle (),
639 ACE_Event_Handler::READ_MASK
);
643 this->select_reactor_
= 0;
649 ACE_Select_Reactor_Notify::close (void)
651 ACE_TRACE ("ACE_Select_Reactor_Notify::close");
653 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
654 notification_queue_
.reset();
656 if (this->notification_pipe_
.read_handle() != ACE_INVALID_HANDLE
)
658 // Please see Bug 2820, if we just close the pipe then we break
659 // the reference counting rules. Basically, all the event
660 // handlers "stored" in the pipe had their reference counts
661 // increased. We need to decrease them before closing the
663 ACE_Notification_Buffer b
;
664 for (int r
= read_notify_pipe(notification_pipe_
.read_handle(), b
);
666 r
= read_notify_pipe(notification_pipe_
.read_handle(), b
))
670 b
.eh_
->remove_reference();
674 #endif /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
676 return this->notification_pipe_
.close ();
680 ACE_Select_Reactor_Notify::notify (ACE_Event_Handler
*event_handler
,
681 ACE_Reactor_Mask mask
,
682 ACE_Time_Value
*timeout
)
684 ACE_TRACE ("ACE_Select_Reactor_Notify::notify");
686 // Just consider this method a "no-op" if there's no
687 // <ACE_Select_Reactor> configured.
688 if (this->select_reactor_
== 0)
691 ACE_Event_Handler_var
safe_handler (event_handler
);
694 event_handler
->add_reference ();
696 ACE_Notification_Buffer
buffer (event_handler
, mask
);
698 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
699 int const notification_required
=
700 notification_queue_
.push_new_notification(buffer
);
702 if (notification_required
== -1)
707 if (notification_required
== 0)
709 // No failures, the handler is now owned by the notification queue
710 safe_handler
.release ();
714 #endif /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
716 ssize_t
const n
= ACE::send (this->notification_pipe_
.write_handle (),
724 safe_handler
.release ();
729 // Handles pending threads (if any) that are waiting to unblock the
733 ACE_Select_Reactor_Notify::dispatch_notifications (int &number_of_active_handles
,
734 ACE_Handle_Set
&rd_mask
)
736 ACE_TRACE ("ACE_Select_Reactor_Notify::dispatch_notifications");
738 ACE_HANDLE
const read_handle
=
739 this->notification_pipe_
.read_handle ();
741 if (read_handle
!= ACE_INVALID_HANDLE
742 && rd_mask
.is_set (read_handle
))
744 --number_of_active_handles
;
745 rd_mask
.clr_bit (read_handle
);
746 return this->handle_input (read_handle
);
754 ACE_Select_Reactor_Notify::notify_handle (void)
756 ACE_TRACE ("ACE_Select_Reactor_Notify::notify_handle");
758 return this->notification_pipe_
.read_handle ();
763 ACE_Select_Reactor_Notify::is_dispatchable (ACE_Notification_Buffer
&buffer
)
765 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
766 ACE_UNUSED_ARG(buffer
);
769 // If eh == 0 then another thread is unblocking the
770 // <ACE_Select_Reactor> to update the <ACE_Select_Reactor>'s
771 // internal structures. Otherwise, we need to dispatch the
772 // appropriate handle_* method on the <ACE_Event_Handler>
773 // pointer we've been passed.
780 // has no dispatchable buffer
783 #endif /*ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
787 ACE_Select_Reactor_Notify::dispatch_notify (ACE_Notification_Buffer
&buffer
)
791 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
792 // Dispatch one message from the notify queue, and put another in
793 // the pipe if one is available. Remember, the idea is to keep
794 // exactly one message in the pipe at a time.
796 bool more_messages_queued
= false;
797 ACE_Notification_Buffer next
;
799 result
= notification_queue_
.pop_next_notification(buffer
,
800 more_messages_queued
,
813 if(more_messages_queued
)
815 (void) ACE::send(this->notification_pipe_
.write_handle(),
816 (char *)&next
, sizeof(ACE_Notification_Buffer
));
818 #endif /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
820 // If eh == 0 then another thread is unblocking the
821 // <ACE_Select_Reactor> to update the <ACE_Select_Reactor>'s
822 // internal structures. Otherwise, we need to dispatch the
823 // appropriate handle_* method on the <ACE_Event_Handler> pointer
824 // we've been passed.
827 ACE_Event_Handler
*event_handler
= buffer
.eh_
;
829 bool const requires_reference_counting
=
830 event_handler
->reference_counting_policy ().value () ==
831 ACE_Event_Handler::Reference_Counting_Policy::ENABLED
;
833 switch (buffer
.mask_
)
835 case ACE_Event_Handler::READ_MASK
:
836 case ACE_Event_Handler::ACCEPT_MASK
:
837 result
= event_handler
->handle_input (ACE_INVALID_HANDLE
);
839 case ACE_Event_Handler::WRITE_MASK
:
840 result
= event_handler
->handle_output (ACE_INVALID_HANDLE
);
842 case ACE_Event_Handler::EXCEPT_MASK
:
843 result
= event_handler
->handle_exception (ACE_INVALID_HANDLE
);
845 case ACE_Event_Handler::QOS_MASK
:
846 result
= event_handler
->handle_qos (ACE_INVALID_HANDLE
);
848 case ACE_Event_Handler::GROUP_QOS_MASK
:
849 result
= event_handler
->handle_group_qos (ACE_INVALID_HANDLE
);
852 // Should we bail out if we get an invalid mask?
853 ACE_ERROR ((LM_ERROR
,
854 ACE_TEXT ("invalid mask = %d\n"),
859 event_handler
->handle_close (ACE_INVALID_HANDLE
,
860 ACE_Event_Handler::EXCEPT_MASK
);
862 if (requires_reference_counting
)
864 event_handler
->remove_reference ();
872 ACE_Select_Reactor_Notify::read_notify_pipe (ACE_HANDLE handle
,
873 ACE_Notification_Buffer
&buffer
)
875 ACE_TRACE ("ACE_Select_Reactor_Notify::read_notify_pipe");
877 ssize_t
const n
= ACE::recv (handle
, (char *) &buffer
, sizeof buffer
);
881 // Check to see if we've got a short read.
882 if (n
!= sizeof buffer
)
884 ssize_t
const remainder
= sizeof buffer
- n
;
886 // If so, try to recover by reading the remainder. If this
887 // doesn't work we're in big trouble since the input stream
888 // won't be aligned correctly. I'm not sure quite what to
889 // do at this point. It's probably best just to return -1.
890 if (ACE::recv (handle
,
891 ((char *) &buffer
) + n
,
892 remainder
) != remainder
)
900 // Return -1 if things have gone seriously wrong.
901 if (n
<= 0 && (errno
!= EWOULDBLOCK
&& errno
!= EAGAIN
))
909 ACE_Select_Reactor_Notify::handle_input (ACE_HANDLE handle
)
911 ACE_TRACE ("ACE_Select_Reactor_Notify::handle_input");
912 // Precondition: this->select_reactor_.token_.current_owner () ==
913 // ACE_Thread::self ();
915 int number_dispatched
= 0;
917 ACE_Notification_Buffer buffer
;
919 while ((result
= this->read_notify_pipe (handle
, buffer
)) > 0)
921 // Dispatch the buffer
922 // NOTE: We count only if we made any dispatches ie. upcalls.
923 if (this->dispatch_notify (buffer
) > 0)
926 // Bail out if we've reached the <notify_threshold_>. Note that
927 // by default <notify_threshold_> is -1, so we'll loop until all
928 // the notifications in the pipe have been dispatched.
929 if (number_dispatched
== this->max_notify_iterations_
)
933 // Reassign number_dispatched to -1 if things have gone seriously
936 number_dispatched
= -1;
938 // Enqueue ourselves into the list of waiting threads. When we
939 // reacquire the token we'll be off and running again with ownership
940 // of the token. The postcondition of this call is that
941 // <select_reactor_.token_.current_owner> == <ACE_Thread::self>.
942 this->select_reactor_
->renew ();
943 return number_dispatched
;
946 // -------------------------------------------
949 ACE_Select_Reactor_Impl::purge_pending_notifications (ACE_Event_Handler
*eh
,
950 ACE_Reactor_Mask mask
)
952 if (this->notify_handler_
== 0)
955 return this->notify_handler_
->purge_pending_notifications (eh
, mask
);
959 // Perform GET, CLR, SET, and ADD operations on the Handle_Sets.
961 // GET = 1, Retrieve current value
962 // SET = 2, Set value of bits to new mask (changes the entire mask)
963 // ADD = 3, Bitwise "or" the value into the mask (only changes
965 // CLR = 4 Bitwise "and" the negation of the value out of the mask
966 // (only changes enabled bits)
968 // Returns the original mask. Must be called with locks held.
970 ACE_Select_Reactor_Impl::bit_ops (ACE_HANDLE handle
,
971 ACE_Reactor_Mask mask
,
972 ACE_Select_Reactor_Handle_Set
&handle_set
,
975 ACE_TRACE ("ACE_Select_Reactor_Impl::bit_ops");
976 if (this->handler_rep_
.handle_in_range (handle
) == 0)
979 #if !defined (ACE_WIN32)
981 this->mask_signals_
); // Block out all signals until method returns.
982 #endif /* ACE_WIN32 */
984 ACE_FDS_PTMF ptmf
= &ACE_Handle_Set::set_bit
;
985 u_long omask
= ACE_Event_Handler::NULL_MASK
;
987 // Find the old reactor masks. This automatically does the work of
988 // the GET_MASK operation.
989 if (handle_set
.rd_mask_
.is_set (handle
))
990 ACE_SET_BITS (omask
, ACE_Event_Handler::READ_MASK
);
991 if (handle_set
.wr_mask_
.is_set (handle
))
992 ACE_SET_BITS (omask
, ACE_Event_Handler::WRITE_MASK
);
993 if (handle_set
.ex_mask_
.is_set (handle
))
994 ACE_SET_BITS (omask
, ACE_Event_Handler::EXCEPT_MASK
);
998 case ACE_Reactor::GET_MASK
:
999 // The work for this operation is done in all cases at the
1000 // begining of the function.
1002 case ACE_Reactor::CLR_MASK
:
1003 ptmf
= &ACE_Handle_Set::clr_bit
;
1004 // State was changed. we need to reflect that change in the
1005 // dispatch_mask I assume that only ACE_Reactor::CLR_MASK should
1006 // be treated here which means we need to clear the handle|mask
1007 // from the current dispatch handler
1008 this->clear_dispatch_mask (handle
, mask
);
1010 case ACE_Reactor::SET_MASK
:
1012 case ACE_Reactor::ADD_MASK
:
1014 // The following code is rather subtle... Note that if we are
1015 // doing a ACE_Reactor::SET_MASK then if the bit is not enabled
1016 // in the mask we need to clear the bit from the ACE_Handle_Set.
1017 // On the other hand, if we are doing a ACE_Reactor::CLR_MASK or
1018 // a ACE_Reactor::ADD_MASK we just carry out the operations
1019 // specified by the mask.
1021 // READ, ACCEPT, and CONNECT flag will place the handle in the
1023 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::READ_MASK
)
1024 || ACE_BIT_ENABLED (mask
, ACE_Event_Handler::ACCEPT_MASK
)
1025 || ACE_BIT_ENABLED (mask
, ACE_Event_Handler::CONNECT_MASK
))
1027 (handle_set
.rd_mask_
.*ptmf
) (handle
);
1029 else if (ops
== ACE_Reactor::SET_MASK
)
1030 handle_set
.rd_mask_
.clr_bit (handle
);
1032 // WRITE and CONNECT flag will place the handle in the write set
1033 if (ACE_BIT_ENABLED (mask
,
1034 ACE_Event_Handler::WRITE_MASK
)
1035 || ACE_BIT_ENABLED (mask
,
1036 ACE_Event_Handler::CONNECT_MASK
))
1038 (handle_set
.wr_mask_
.*ptmf
) (handle
);
1040 else if (ops
== ACE_Reactor::SET_MASK
)
1041 handle_set
.wr_mask_
.clr_bit (handle
);
1043 // EXCEPT (and CONNECT on Win32) flag will place the handle in
1045 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::EXCEPT_MASK
)
1046 #if defined (ACE_WIN32)
1047 || ACE_BIT_ENABLED (mask
, ACE_Event_Handler::CONNECT_MASK
)
1048 #endif /* ACE_WIN32 */
1051 (handle_set
.ex_mask_
.*ptmf
) (handle
);
1053 else if (ops
== ACE_Reactor::SET_MASK
)
1054 handle_set
.ex_mask_
.clr_bit (handle
);
1063 ACE_Select_Reactor_Impl::clear_dispatch_mask (ACE_HANDLE handle
,
1064 ACE_Reactor_Mask mask
)
1066 ACE_TRACE ("ACE_Select_Reactor_Impl::clear_dispatch_mask");
1068 // Use handle and mask in order to modify the sets
1069 // (wait/suspend/ready/dispatch), that way, the dispatch_io_set loop
1070 // will not be interrupt, and there will no reason to rescan the
1071 // wait_set and re-calling select function, which is *very*
1072 // expensive. It seems that wait/suspend/ready sets are getting
1073 // updated in register/remove bind/unbind etc functions. The only
1074 // thing need to be updated is the dispatch_set (also can be found
1075 // in that file code as dispatch_mask). Because of that, we need
1076 // that dispatch_set to be member of the ACE_Select_Reactor_impl in
1077 // Select_Reactor_Base.h file That way we will have access to that
1078 // member in that function.
1080 // We kind of invalidate the iterator in dispatch_io_set because its
1081 // an array and index built from the original dispatch-set. Take a
1082 // look at dispatch_io_set for more details.
1084 // We only need to clr_bit, because we are interested in clearing the
1085 // handles that was removed, so no dispatching to these handles will
1087 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::READ_MASK
) ||
1088 ACE_BIT_ENABLED (mask
, ACE_Event_Handler::ACCEPT_MASK
))
1090 this->dispatch_set_
.rd_mask_
.clr_bit (handle
);
1092 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::WRITE_MASK
))
1094 this->dispatch_set_
.wr_mask_
.clr_bit (handle
);
1096 if (ACE_BIT_ENABLED (mask
, ACE_Event_Handler::EXCEPT_MASK
))
1098 this->dispatch_set_
.ex_mask_
.clr_bit (handle
);
1101 // That will make the dispatch_io_set iterator re-start and rescan
1102 // the dispatch set.
1103 this->state_changed_
= true;
1108 ACE_Select_Reactor_Impl::resumable_handler (void)
1110 // The select reactor has no handlers that can be resumed by the
1111 // application. So return 0;
1116 ACE_END_VERSIONED_NAMESPACE_DECL