summaryrefslogtreecommitdiff
path: root/TAO/performance-tests/Throughput/Receiver.cpp
blob: da9f6df9b7162c9a20c752831b2f003cddd11beb (plain)
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
//
// $Id$
//

#include "Receiver.h"
#include "ace/High_Res_Timer.h"

ACE_RCSID(Throughput, Receiver, "$Id$")

Receiver::Receiver (void)
  : start_time_ (0)
  , message_count_ (0)
  , byte_count_ (0)
  , last_message_time_ (0)
  , last_message_id_ (0)
{
}

void
Receiver::receive_data (const Test::Message &the_message,
                        CORBA::Environment &)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  ACE_hrtime_t now = ACE_OS::gethrtime ();
  if (this->message_count_ == 0)
    {
      this->start_time_ = now;
    }
  else
    {
      if (this->last_message_id_ > the_message.message_id)
        {
          ACE_ERROR ((LM_ERROR,
                      "ERROR: (%P|%t) Receiver::receive_data, "
                      "message out of sequence %d / %d\n",
                      the_message.message_id,
                      this->last_message_id_));
          return;
        }
      this->last_message_id_ = the_message.message_id;
    }
  this->message_count_++;
  this->byte_count_ += the_message.the_payload.length ();
  this->last_message_time_ = now;
}

void
Receiver::done (CORBA::Environment &ACE_TRY_ENV)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  if (this->message_count_ == 0)
    {
      ACE_ERROR ((LM_ERROR,
                  "ERROR: (%P|%t) Receiver::done, "
                  "no messages received\n"));
    }
  else
    {
      ACE_UINT32 gsf =
        ACE_High_Res_Timer::global_scale_factor ();

      ACE_hrtime_t elapsed_time =
        this->last_message_time_ - this->start_time_;

      // convert to microseconds
#if !defined ACE_LACKS_LONGLONG_T
      ACE_UINT32 usecs = ACE_UINT32(elapsed_time / gsf);
#else  /* ! ACE_LACKS_LONGLONG_T */
      ACE_UINT32 usecs = elapsed_time / gsf;
#endif /* ! ACE_LACKS_LONGLONG_T */

      if (usecs != 0)
        {
          double bytes =
            (1000000.0 * this->byte_count_) / usecs;
          double kbytes = bytes / 1024;
          double mbytes = kbytes / 1024;
          double mbits  = bytes * 8 / 10000000;

          ACE_DEBUG ((LM_DEBUG,
                      "Receiver %f (bytes/sec), %f (Kb/sec)\n"
                      "Receiver %f (Mb/sec), %f Mbits\n",
                      bytes, kbytes,
                      mbytes, mbits));
        }
    }

  ACE_TRY
    {
      PortableServer::POA_var poa = this->_default_POA (ACE_TRY_ENV);
      ACE_TRY_CHECK;
      PortableServer::ObjectId_var oid =
        poa->servant_to_id (this, ACE_TRY_ENV);
      ACE_TRY_CHECK;
      poa->deactivate_object (oid.in (), ACE_TRY_ENV);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY {} ACE_ENDTRY;
}