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
|
class WordCounter: public Block
{
unsigned long long m_word_cnt;
int m_msg_recv;
unsigned int m_reset;
uint32_t m_msg_type;
int m_ingate_id;
std::map<const std::string,unsigned long long> words[MAX_SIZE];
struct timeval lasttime;
long receivedMessages;
long receivedTuple;
long receivedMessagesOld;
timeval firsttime;
int reported;
public:
/**
* @brief Constructor
* @param name The name of the packet counter block
* @param invocation Invocation type of the block (Indirect, Direct, Async)
*/
WordCounter(const std::string &name, invocation_type invocation)
: Block(name, invocation),
m_ingate_id(register_input_gate("in_word")),
receivedMessages(0),
receivedTuple(0),
receivedMessagesOld(0),
reported(0)
{
m_reset = 0;
m_word_cnt = 0;
m_msg_type = type_to_id<WordRecord>::id();
register_variable("wordcnt",make_rd_var(no_mutex_t(), m_word_cnt));
register_variable("reset",make_wr_var(no_mutex_t(), m_reset));
//Monitor mn(this);
}
void HandleMonitor(int monitored) {
if(receivedMessagesOld != receivedMessages) {
gettimeofday(&lasttime, NULL);
}
if( (monitored % 10) == 0 && (reported < receivedMessages) ) {
long deltat = (lasttime.tv_sec - firsttime.tv_sec) * 1000000 + (lasttime.tv_usec - firsttime.tv_usec);
double deltatt = (double) deltat / 1000000.0;
std::cout << get_name() << ": rate = " << receivedMessages/deltatt << " " << receivedTuple/deltatt << " (received " << receivedMessages << " in " << deltatt << "sec" << std::endl;
reported = receivedMessages;
}
receivedMessagesOld = receivedMessages;
if( (monitored % 10) == 0) {
for( std::map<const std::string, unsigned long long>::iterator ii = words[0].begin();
ii != words[0].end();
++ii ) {
std::cout << (*ii).first << ": " << (*ii).second << std::endl;
}
}
}
/**
* @brief Destructor
*/
~WordCounter() {}
/**
* @brief Configures the block.
* @param n The configuration parameters
*/
void _configure(const pugi::xml_node& n )
{}
/**
* @brief Initialize the block
*/
void _initialize()
{}
/**
* If the message received is not of type RawPacket throw an exception,
* otherwise count it
* @param m The message
* @param index The index of the gate the message came on
*/
void _receive_msg(std::shared_ptr<const Msg>&& m, int /* index */)
{
receivedTuple++;
if(m->type() != m_msg_type ){
throw std::runtime_error("WordCounter::wrong message type");
}
// stuff for monitor
/* if(receivedMessages == 0) {
gettimeofday(&firsttime, NULL);
}*/
const WordRecord* msg_ptr = static_cast<const WordRecord *>(m.get());
//int idx = msg_ptr->key() % MAX_SIZE;
int idx = 0;
assert(idx>=0 && idx < MAX_SIZE);
for(int i = 0; i < msg_ptr->get_word_number(); i++) {
receivedMessages++;
std::map<const std::string,unsigned long long>::iterator it;
if( (it = words[idx].find( msg_ptr->get_word(i))) != words[idx].end() ){
it->second++;
//std::cout << "PRESENTE " << it->first << " e countno. " << it->second << std::endl;
}
else{
words[idx].insert( std::pair<const std::string,unsigned long long>(msg_ptr->get_word(i),1) );
std::cout << "INSERISCO " << msg_ptr->get_word(i) << std::endl;
}
}
if(!(m_msg_recv % 50000)) {
std::cout << "Received " << m_msg_recv << std::endl;
}
m_msg_recv++;
}
};
#ifndef _BLOCKMON_DOXYGEN_SKIP_
REGISTER_BLOCK(WordCounter,"WordCounter");
#endif /* _BLOCKMON_DOXYGEN_SKIP_ */
}//blockmon
#endif // _BASE_PACKETCOUNTER_HPP_
|
