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
|
// $Id$
#include "Batch_Buffering_Strategy.h"
#include "../Method_Request_Event.h"
#include "ace/Null_Condition.h"
#if ! defined (__ACE_INLINE__)
#include "Batch_Buffering_Strategy.inl"
#endif /* __ACE_INLINE__ */
ACE_RCSID (Notify, TAO_Notify_Batch_Buffering_Strategy, "$Id$")
TAO_Notify_Batch_Buffering_Strategy::TAO_Notify_Batch_Buffering_Strategy (TAO_Notify_Message_Queue& msg_queue, TAO_Notify_AdminProperties_var& admin_properties, CORBA::Long batch_size)
:TAO_Notify_Buffering_Strategy (msg_queue, admin_properties, batch_size)
{
}
TAO_Notify_Batch_Buffering_Strategy::~TAO_Notify_Batch_Buffering_Strategy ()
{
}
int
TAO_Notify_Batch_Buffering_Strategy::dequeue_batch (CosNotification::EventBatch& event_batch)
{
ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->global_queue_lock_, -1);
// if batch_size is infinite, simply dequeue everything available.
int pending = 0; // not used.
if (this->batch_size_ == 0)
{
return this->dequeue_available (event_batch, pending);
}
else
{
// block till batch size of events are available.
while (this->msg_queue_.message_count () < this->batch_size_)
{
if (this->shutdown_ == 1) // if we're shutdown, don't play this silly game.
return -1;
this->batch_size_reached_condition_.wait ();
}
return this->dequeue_i (this->batch_size_, event_batch);
}
}
int
TAO_Notify_Batch_Buffering_Strategy::dequeue_available (CosNotification::EventBatch& event_batch, int &pending)
{
ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->global_queue_lock_, -1);
int deq_count = this->msg_queue_.message_count ();
if (this->batch_size_ != 0 && deq_count > this->batch_size_) // Restrict upto batch size.
deq_count = this->batch_size_;
pending = this->msg_queue_.message_count () - deq_count;
return this->dequeue_i (deq_count, event_batch);
}
int
TAO_Notify_Batch_Buffering_Strategy::dequeue_i (int max_deq_count, CosNotification::EventBatch& event_batch)
{
ACE_Message_Block *mb;
int deq_count = 0;
event_batch.length (max_deq_count);
for (; deq_count < max_deq_count; ++deq_count)
{
if (this->msg_queue_.dequeue (mb) == -1)
break; // error, simply return what we could extract so far.
--this->global_queue_length_;
TAO_Notify_Method_Request_Event* mre = ACE_dynamic_cast (TAO_Notify_Method_Request_Event*, mb);
mre->event ()->convert (event_batch[deq_count]);
ACE_Message_Block::release (mb);
}
event_batch.length (deq_count);
this->global_queue_not_full_condition_.signal ();
this->local_queue_not_full_condition_.signal ();
return deq_count;
}
|
