Statistics
| Branch: | Revision:

streamers / measures-monl.c @ 461946de

History | View | Annotate | Download (12.1 KB)

1
/*
2
 *  Copyright (c) 2010 Csaba Kiraly
3
 *
4
 *  This is free software; see gpl-3.0.txt
5
 */
6
#include <math.h>
7
#ifndef NAN        //NAN is missing in some old math.h versions
8
#define NAN            (0.0/0.0)
9
#endif
10
#ifndef INFINITY
11
#define INFINITY       (1.0/0.0)
12
#endif
13

    
14
#include <mon.h>
15
#include <ml.h>
16
#include <net_helper.h>
17
#include <grapes_msg_types.h>
18

    
19
#include "channel.h"
20
#include "dbg.h"
21
#include "measures.h"
22

    
23
#define PEER_PUBLISH_INTERVAL 10 //in seconds
24
#define P2P_PUBLISH_INTERVAL 60 //in seconds
25

    
26
extern const char *peername;
27

    
28
typedef struct nodeID {
29
        socketID_handle addr;
30
        int connID;        // connection associated to this node, -1 if myself
31
        int refcnt;
32
        //a quick and dirty static vector for measures TODO: make it dinamic
33
        MonHandler mhs[20];
34
        int n_mhs;
35
} nodeID;
36

    
37
static MonHandler chunk_dup, chunk_playout, neigh_size, chunk_receive, chunk_send, offer_accept, chunk_hops, chunk_delay, playout_delay;
38
//static MonHandler rx_bytes_chunk_per_sec, tx_bytes_chunk_per_sec, rx_bytes_sig_per_sec, tx_bytes_sig_per_sec;
39
//static MonHandler rx_chunks, tx_chunks;
40

    
41
/*
42
 * Initialize one measure
43
*/
44
void add_measure(MonHandler *mh, MeasurementId id, MeasurementCapabilities mc, MonParameterValue rate, const char *pubname, enum stat_types st[], int length, SocketId dst, MsgType mt)
45
{
46
        *mh = monCreateMeasure(id, mc);
47
        if (rate) monSetParameter (*mh, P_PUBLISHING_RATE, rate);
48
        if (length) monPublishStatisticalType(*mh, pubname, channel_get_name(), st , length, NULL);
49
        monActivateMeasure(*mh, dst, mt);
50
}
51

    
52
/*
53
 * Register duplicate arrival
54
*/
55
void reg_chunk_duplicate()
56
{
57
        if (!chunk_dup) {
58
                enum stat_types st[] = {SUM, RATE};
59
                add_measure(&chunk_dup, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunkDuplicates", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
60
                monNewSample(chunk_dup, 0);        //force publish even if there are no events
61
        }
62
        monNewSample(chunk_dup, 1);
63
}
64

    
65
/*
66
 * Register playout/loss of a chunk before playout
67
*/
68
void reg_chunk_playout(int id, bool b, uint64_t timestamp)
69
{
70
        static MonHandler chunk_loss_burst_size;
71
        static int last_arrived_chunk = -1;
72

    
73
        struct timeval tnow;
74

    
75
        if (!chunk_playout && b) {        //don't count losses before the first arrived chunk
76
                enum stat_types st[] = {WIN_AVG, AVG, SUM, RATE};
77
                add_measure(&chunk_playout, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunksPlayed", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
78
        }
79
        monNewSample(chunk_playout, b);
80

    
81
        if (!playout_delay) {
82
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
83
                add_measure(&playout_delay, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ReorderDelay", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
84
        }
85
        if (b) {        //count delay only if chunk has arrived
86
                gettimeofday(&tnow, NULL);
87
                monNewSample(playout_delay, ((int64_t)(tnow.tv_usec + tnow.tv_sec * 1000000ULL) - (int64_t)timestamp) / 1000000.0);
88
        }
89

    
90
        //if (!chunk_loss_burst_size) {
91
        //        enum stat_types st[] = {WIN_AVG, WIN_VAR};
92
        //        add_measure(&chunk_loss_burst_size, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunkLossBurstSize", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
93
        //}
94
        if (b) {
95
                if (last_arrived_chunk >= 0) {
96
                        int burst_size = id - 1 - last_arrived_chunk;
97
                        if (burst_size) monNewSample(chunk_loss_burst_size, burst_size);
98
                }
99
                last_arrived_chunk = id;
100
        }
101
}
102

    
103
/*
104
 * Register actual neghbourhood size
105
*/
106
void reg_neigh_size(int s)
107
{
108
        if (!neigh_size) {
109
                enum stat_types st[] = {LAST};
110
                add_measure(&neigh_size, GENERIC, 0, PEER_PUBLISH_INTERVAL, "NeighSize", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
111
        }
112
        monNewSample(neigh_size, s);
113
}
114

    
115
/*
116
 * Register chunk receive event
117
*/
118
void reg_chunk_receive(int id, uint64_t timestamp, int hopcount, bool old, bool dup)
119
{
120
        struct timeval tnow;
121

    
122
        if (!chunk_receive) {
123
                enum stat_types st[] = {RATE};
124
                add_measure(&chunk_receive, GENERIC, 0, PEER_PUBLISH_INTERVAL, "RxChunkAll", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
125
                monNewSample(chunk_receive, 0);        //force publish even if there are no events
126
        }
127
        monNewSample(chunk_receive, 1);
128

    
129
        if (!chunk_hops) {
130
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
131
                add_measure(&chunk_hops, GENERIC, 0, PEER_PUBLISH_INTERVAL, "Hops", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
132
        }
133
        monNewSample(chunk_hops, hopcount);
134

    
135
        if (!chunk_delay) {
136
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
137
                add_measure(&chunk_delay, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ReceiveDelay", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
138
        }
139
        gettimeofday(&tnow, NULL);
140
        monNewSample(chunk_delay, ((int64_t)(tnow.tv_usec + tnow.tv_sec * 1000000ULL) - (int64_t)timestamp) / 1000000.0);
141
}
142

    
143
/*
144
 * Register chunk send event
145
*/
146
void reg_chunk_send(int id)
147
{
148
        if (!chunk_send) {
149
                enum stat_types st[] = {RATE};
150
                add_measure(&chunk_send, GENERIC, 0, PEER_PUBLISH_INTERVAL, "TxChunkAll", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
151
                monNewSample(chunk_send, 0);        //force publish even if there are no events
152
        }
153
        monNewSample(chunk_send, 1);
154
}
155

    
156
/*
157
 * Register chunk accept evemt
158
*/
159
void reg_offer_accept(bool b)
160
{
161
        if (!offer_accept) {
162
                enum stat_types st[] = {AVG};
163
                add_measure(&offer_accept, GENERIC, 0, PEER_PUBLISH_INTERVAL, "OfferAccept", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
164
        }
165
        monNewSample(offer_accept, b);
166
}
167

    
168
/*
169
 * Initialize peer level measurements
170
*/
171
void init_measures()
172
{
173
        //enum stat_types stavg[] = {WIN_AVG};
174
        //enum stat_types stsum[] = {SUM};
175

    
176
        if (peername) monSetPeerName(peername);
177

    
178
        // Traffic
179
       //add_measure(&rx_bytes_chunk_per_sec, BULK_TRANSFER, RXONLY | PACKET | TIMER_BASED, PEER_PUBLISH_INTERVAL, "RxBytesChunkPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), NULL, MSG_TYPE_CHUNK);        //[bytes/s]
180
       //add_measure(&tx_bytes_chunk_per_sec, BULK_TRANSFER, TXONLY | PACKET | TIMER_BASED, PEER_PUBLISH_INTERVAL, "TxBytesChunkPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), NULL, MSG_TYPE_CHUNK);        //[bytes/s]
181
       //add_measure(&rx_bytes_sig_per_sec, BULK_TRANSFER, RXONLY | PACKET | TIMER_BASED, PEER_PUBLISH_INTERVAL, "RxBytesSigPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), NULL, MSG_TYPE_SIGNALLING);        //[bytes/s]
182
       //add_measure(&tx_bytes_sig_per_sec, BULK_TRANSFER, TXONLY | PACKET | TIMER_BASED, PEER_PUBLISH_INTERVAL, "TxBytesSigPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), NULL, MSG_TYPE_SIGNALLING);        //[bytes/s]
183

    
184
        // Chunks
185
       // replaced by reg_chun_receive add_measure(&rx_chunks, COUNTER, RXONLY | DATA | IN_BAND, PEER_PUBLISH_INTERVAL, "RxChunksAll", stsum, sizeof(stsum)/sizeof(enum stat_types), NULL, MSG_TYPE_CHUNK);        //[chunks]
186
       // replaced by reg_chun_send add_measure(&tx_chunks, COUNTER, TXONLY | DATA | IN_BAND, PEER_PUBLISH_INTERVAL, "TxChunksAll", stsum, sizeof(stsum)/sizeof(enum stat_types), NULL, MSG_TYPE_CHUNK);        //[chunks]
187
}
188

    
189
/*
190
 * End peer level measurements
191
*/
192
void end_measures()
193
{
194
}
195

    
196
/*
197
 * Initialize p2p measurements towards a peer
198
*/
199
void add_measures(struct nodeID *id)
200
{
201
        // Add measures
202
        int j = 0;
203
        enum stat_types stavg[] = {AVG};
204
        enum stat_types stwinavg[] = {WIN_AVG};
205
        enum stat_types stsum[] = {SUM};
206
        enum stat_types stsumrate[] = {SUM, RATE};
207

    
208
        dprintf("adding measures to %s\n",node_addr(id));
209

    
210
        /* HopCount */
211
       add_measure(&id->mhs[j++], HOPCOUNT, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "HopCount", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[IP hops]
212

    
213
        /* Round Trip Time */
214
       add_measure(&id->mhs[j++], RTT, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "RoundTripDelay", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[seconds]
215

    
216
        /* Loss */
217
       add_measure(&id->mhs[j++], SEQWIN, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
218
       add_measure(&id->mhs[j++], LOSS, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "LossRate", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //LossRate_avg [probability 0..1] LossRate_rate [lost_pkts/sec]
219

    
220
        // Cumulative Traffic
221
       //add_measure(&id->mhs[j++], BYTE, RXONLY | PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "RxBytes", stsum, sizeof(stsum)/sizeof(enum stat_types), id->addr, MSG_TYPE_ANY);
222
       //add_measure(&id->mhs[j++], BYTE, TXONLY | PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "TxBytes", stsum, sizeof(stsum)/sizeof(enum stat_types), id->addr, MSG_TYPE_ANY);
223
       add_measure(&id->mhs[j++], RX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "RxBytesChunk", stsum, sizeof(stsum)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
224
       add_measure(&id->mhs[j++], TX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "TxBytesChunk", stsum, sizeof(stsum)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
225

    
226
        // Traffic
227
       add_measure(&id->mhs[j++], RX_BULK_TRANSFER, PACKET | TIMER_BASED, P2P_PUBLISH_INTERVAL, "RxBytesChunkPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes/s]
228
       add_measure(&id->mhs[j++], TX_BULK_TRANSFER, PACKET | TIMER_BASED, P2P_PUBLISH_INTERVAL, "TxBytesChunkPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes/s]
229
       add_measure(&id->mhs[j++], RX_BULK_TRANSFER, PACKET | TIMER_BASED, P2P_PUBLISH_INTERVAL, "RxBytesSigPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes/s]
230
       add_measure(&id->mhs[j++], TX_BULK_TRANSFER, PACKET | TIMER_BASED, P2P_PUBLISH_INTERVAL, "TxBytesSigPSec", stavg, sizeof(stavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes/s]
231

    
232
        // Chunks
233
       add_measure(&id->mhs[j++], RX_PACKET, DATA | IN_BAND, P2P_PUBLISH_INTERVAL, "RxChunks", stsumrate, sizeof(stsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //RxChunks_sum [chunks] RxChunks_rate [chunks/sec]
234
       add_measure(&id->mhs[j++], TX_PACKET, DATA | IN_BAND, P2P_PUBLISH_INTERVAL, "TxChunks", stsumrate, sizeof(stsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //TxChunks_sum [chunks] TxChunks_rate [chunks/sec]
235
//        // Capacity
236
//        add_measure(&id->mhs[j++], CLOCKDRIFT, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
237
//        monSetParameter (id->mhs[j], P_CLOCKDRIFT_ALGORITHM, 1);
238
//        add_measure(&id->mhs[j++], CORRECTED_DELAY, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
239
//        add_measure(&id->mhs[j++], CAPACITY_CAPPROBE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "Capacity", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes/s]
240
//        monSetParameter (id->mhs[j], P_CAPPROBE_DELAY_TH, -1);
241
//        monSetParameter (mh, P_CAPPROBE_PKT_TH, 100);
242
//        monSetParameter (mh, P_CAPPROBE_IPD_TH, 60);
243
//        monPublishStatisticalType(mh, NULL, st , sizeof(st)/sizeof(enum stat_types), repoclient);
244

    
245
        // for static must not be more then 10 or whatever size is in net_helper-ml.c
246
        id->n_mhs = j;
247
}
248

    
249
/*
250
 * Delete p2p measurements towards a peer
251
*/
252
void delete_measures(struct nodeID *id)
253
{
254
        dprintf("deleting measures from %s\n",node_addr(id));
255
        while(id->n_mhs) {
256
                monDestroyMeasure(id->mhs[--(id->n_mhs)]);
257
        }
258
}
259

    
260
/*
261
 * Helper to retrieve a measure
262
*/
263
double get_measure(struct nodeID *id, int j, enum stat_types st)
264
{
265
        return (id->n_mhs > j) ? monRetrieveResult(id->mhs[j], st) : NAN;
266
}
267

    
268
/*
269
 * Hopcount to a given peer
270
*/
271
int get_hopcount(struct nodeID *id){
272
        double r = get_measure(id, 0, LAST);
273
        return isnan(r) ? -1 : (int) r;
274
}
275

    
276
/*
277
 * RTT to a given peer in seconds
278
*/
279
double get_rtt(struct nodeID *id){
280
        return get_measure(id, 1, WIN_AVG);
281
}
282

    
283
/*
284
 * average RTT to a set of peers in seconds
285
*/
286
double get_average_rtt(struct nodeID **ids, int len){
287
        int i;
288
        int n = 0;
289
        double sum = 0;
290

    
291
        for (i = 0; i < len; i++) {
292
                double l = get_rtt(ids[i]);
293
                if (!isnan(l)) {
294
                        sum += l;
295
                        n++;
296
                }
297
        }
298
        return (n > 0) ? sum / n : NAN;
299
}
300

    
301
/*
302
 * loss ratio from a given peer as 0..1
303
*/
304
double get_lossrate(struct nodeID *id){
305
        return get_measure(id, 3, WIN_AVG);
306
}
307

    
308
/*
309
 * average loss ratio from a set of peers as 0..1
310
*/
311
double get_average_lossrate(struct nodeID **ids, int len){
312
        int i;
313
        int n = 0;
314
        double sum = 0;
315

    
316
        for (i = 0; i < len; i++) {
317
                double l = get_lossrate(ids[i]);
318
                if (!isnan(l)) {
319
                        sum += l;
320
                        n++;
321
                }
322
        }
323
        return (n > 0) ? sum / n : NAN;
324
}
325

    
326
double get_receive_delay(void) {
327
        return chunk_delay ? monRetrieveResult(chunk_delay, WIN_AVG) : NAN;
328
}