TAO/tao/Messaging/Messaging_Queueing_Strategies.cpp

   1 #include "tao/Messaging/Messaging_Queueing_Strategies.h"
   2 #include "tao/Messaging/Buffering_Constraint_Policy.h"
   3 #include "tao/Stub.h"
   4 #include "tao/debug.h"
   5
   6 #include "ace/Log_Msg.h"
   7 #include "ace/OS_NS_sys_time.h"
   8
   9 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  10
  11 namespace TAO
  12 {
  13 // ****************************************************************
  14
  15   #if (TAO_HAS_BUFFERING_CONSTRAINT_POLICY == 1)
  16
  17   bool
  18   Eager_Transport_Queueing_Strategy::must_queue (bool) const
  19   {
  20     return true;
  21   }
  22
  23   bool
  24   Eager_Transport_Queueing_Strategy::buffering_constraints_reached (
  25     TAO_Stub *stub,
  26     size_t msg_count,
  27     size_t total_bytes,
  28     bool &must_flush,
  29     const ACE_Time_Value &current_deadline,
  30     bool &set_timer,
  31     ACE_Time_Value &new_deadline) const
  32   {
  33     must_flush = false;
  34     set_timer = false;
  35
  36     TAO::BufferingConstraint buffering_constraint;
  37
  38     try
  39       {
  40         CORBA::Policy_var bcp_policy =
  41           stub->get_cached_policy (TAO_CACHED_POLICY_BUFFERING_CONSTRAINT);
  42
  43         TAO::BufferingConstraintPolicy_var bcpv =
  44           TAO::BufferingConstraintPolicy::_narrow (bcp_policy.in ());
  45
  46         TAO_Buffering_Constraint_Policy* bcp =
  47           dynamic_cast<TAO_Buffering_Constraint_Policy *> (bcpv.in ());
  48         if (bcp == 0)
  49           {
  50             return true;
  51           }
  52         bcp->get_buffering_constraint (buffering_constraint);
  53       }
  54     catch (const ::CORBA::Exception&)
  55       {
  56         return true;
  57       }
  58
  59
  60     if (buffering_constraint.mode == TAO::BUFFER_FLUSH)
  61       {
  62         must_flush = true;
  63         return true;
  64       }
  65
  66     bool constraints_reached = false;
  67
  68     if (ACE_BIT_ENABLED (buffering_constraint.mode,
  69                          TAO::BUFFER_MESSAGE_COUNT)
  70         && msg_count >= buffering_constraint.message_count)
  71       {
  72         constraints_reached = true;
  73       }
  74
  75     if (ACE_BIT_ENABLED (buffering_constraint.mode,
  76                          TAO::BUFFER_MESSAGE_BYTES)
  77         && total_bytes >= buffering_constraint.message_bytes)
  78       {
  79         constraints_reached = true;
  80       }
  81
  82     if (this->timer_check (buffering_constraint,
  83                            current_deadline,
  84                            set_timer,
  85                            new_deadline))
  86       {
  87         constraints_reached = true;
  88       }
  89
  90     return constraints_reached;
  91   }
  92
  93   bool
  94   Eager_Transport_Queueing_Strategy::timer_check (
  95     const TAO::BufferingConstraint &buffering_constraint,
  96     const ACE_Time_Value &current_deadline,
  97     bool &set_timer,
  98     ACE_Time_Value &new_deadline) const
  99   {
 100     set_timer = false;
 101
 102     if (!ACE_BIT_ENABLED (buffering_constraint.mode,
 103                           TAO::BUFFER_TIMEOUT))
 104       {
 105         return false;
 106       }
 107
 108     // Compute the next deadline...
 109     ACE_Time_Value const now = ACE_OS::gettimeofday ();
 110     ACE_Time_Value timeout =
 111       this->time_conversion (buffering_constraint.timeout);
 112     new_deadline = now + timeout;
 113
 114     // Check if the new deadline is more stringent, or if the deadline
 115     // has expired and thus must be reset anyway.
 116     if (current_deadline > new_deadline
 117         || current_deadline < now)
 118       {
 119         set_timer = true;
 120       }
 121
 122     // ... if there is no deadline we don't want to schedule output (the
 123     // deadline will be set because set_timer is set to 1 in that case).
 124     // If there is a deadline but but it has not been reached, we
 125     // don't want to schedule any output either...
 126     if (current_deadline == ACE_Time_Value::zero
 127         || current_deadline >= now)
 128       {
 129         return false;
 130       }
 131
 132     if (TAO_debug_level > 6)
 133       {
 134         TAOLIB_DEBUG ((LM_DEBUG,
 135                     "TAO (%P|%t) - TAO_Eager_Buffering_Sync_Strategy::timer_check, "
 136                     "Now = %u, Current = %u, New = %u\n",
 137                     now.msec (), current_deadline.msec (),
 138                     new_deadline.msec ()));
 139       }
 140
 141     return true;
 142   }
 143
 144   ACE_Time_Value
 145   Eager_Transport_Queueing_Strategy::time_conversion (
 146     const TimeBase::TimeT &time) const
 147   {
 148     TimeBase::TimeT seconds = time / 10000000u;
 149     TimeBase::TimeT microseconds = (time % 10000000u) / 10;
 150     return ACE_Time_Value (ACE_U64_TO_U32 (seconds),
 151                            ACE_U64_TO_U32 (microseconds));
 152   }
 153
 154 // ****************************************************************
 155
 156   bool
 157   Delayed_Transport_Queueing_Strategy::must_queue (bool queue_empty) const
 158   {
 159     // If the queue is empty we want to send immediately
 160     return !queue_empty;
 161   }
 162
 163   #endif /* TAO_HAS_BUFFERING_CONSTRAINT_POLICY == 1 */
 164
 165 }
 166
 167 TAO_END_VERSIONED_NAMESPACE_DECL