1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
// $Id$
#ifndef ACE_FUTURE_SET_CPP
#define ACE_FUTURE_SET_CPP
#include "ace/Future_Set.h"
#if !defined (ACE_LACKS_PRAGMA_ONCE)
#pragma once
#endif /* ACE_LACKS_PRAGMA_ONCE */
#if defined (ACE_HAS_THREADS)
ACE_BEGIN_VERSIONED_NAMESPACE_DECL
template <class T>
ACE_Future_Set<T>::ACE_Future_Set (ACE_Message_Queue<ACE_SYNCH> *new_queue)
: delete_queue_ (0)
{
if (new_queue)
this->future_notification_queue_ = new_queue;
else
{
ACE_NEW (this->future_notification_queue_,
ACE_Message_Queue<ACE_SYNCH>);
this->delete_queue_ = 1;
}
}
template <class T>
ACE_Future_Set<T>::~ACE_Future_Set (void)
{
// Detach ourselves from all remaining futures, if any, in our map.
ACE_TYPENAME FUTURE_HASH_MAP::iterator iterator =
this->future_map_.begin ();
ACE_TYPENAME FUTURE_HASH_MAP::iterator end =
this->future_map_.end ();
for (;
iterator != end;
++iterator)
{
FUTURE_HOLDER *future_holder = (*iterator).int_id_;
future_holder->item_.detach (this);
delete future_holder;
}
if (this->delete_queue_ != 0)
delete this->future_notification_queue_;
}
template <class T> int
ACE_Future_Set<T>::is_empty () const
{
return (((ACE_Future_Set<T>*)this)->future_map_.current_size () == 0 );
}
template <class T> int
ACE_Future_Set<T>::insert (ACE_Future<T> &future)
{
FUTURE_HOLDER *future_holder;
ACE_NEW_RETURN (future_holder,
FUTURE_HOLDER (future),
-1);
FUTURE_REP *future_rep = future.get_rep ();
int result = this->future_map_.bind (future_rep,
future_holder);
// If a new map entry was created, then attach to the future,
// otherwise we were already attached to the future or some error
// occurred so just delete the future holder.
if ( result == 0 )
// Attach ourself to the ACE_Futures list of observer
future.attach (this);
else
delete future_holder;
return result;
}
template <class T> void
ACE_Future_Set<T>::update (const ACE_Future<T> &future)
{
ACE_Message_Block *mb;
FUTURE &local_future = const_cast<ACE_Future<T> &> (future);
ACE_NEW (mb,
ACE_Message_Block ((char *) local_future.get_rep (), 0));
// Enqueue in priority order.
this->future_notification_queue_->enqueue (mb, 0);
}
template <class T> int
ACE_Future_Set<T>::next_readable (ACE_Future<T> &future,
ACE_Time_Value *tv)
{
if (this->is_empty ())
return 0;
ACE_Message_Block *mb = 0;
FUTURE_REP *future_rep = 0;
// Wait for a "readable future" signal from the message queue.
if (this->future_notification_queue_->dequeue_head (mb,
tv) != -1)
{
// Extract future rep from the message block.
future_rep = reinterpret_cast<FUTURE_REP *> (mb->base ());
// Delete the message block.
mb->release ();
}
else
return 0;
// Remove the hash map entry with the specified future rep from our map.
FUTURE_HOLDER *future_holder;
if (this->future_map_.find (future_rep,
future_holder) != -1)
{
future = future_holder->item_;
this->future_map_.unbind (future_rep);
delete future_holder;
return 1;
}
return 0;
}
ACE_END_VERSIONED_NAMESPACE_DECL
#endif /* ACE_HAS_THREADS */
#endif /* ACE_FUTURE_SET_CPP */
|
