Statistics
| Branch: | Revision:

streamers / measures-monl.c @ 59e70a74

History | View | Annotate | Download (11.5 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
 * Start one measure
43
*/
44
void start_measure(MonHandler mh, MonParameterValue rate, const char *pubname, enum stat_types st[], int length, SocketId dst, MsgType mt)
45
{
46
        if (rate) monSetParameter (mh, P_PUBLISHING_RATE, rate);
47
        if (length) monPublishStatisticalType(mh, pubname, channel_get_name(), st , length, NULL);
48
        monActivateMeasure(mh, dst, mt);
49
}
50

    
51
/*
52
 * Initialize and start one measure
53
 */
54
void add_measure(MonHandler *mhp, MeasurementId id, MeasurementCapabilities mc, MonParameterValue rate, const char *pubname, enum stat_types st[], int length, SocketId dst, MsgType mt)
55
{
56
        *mhp = monCreateMeasure(id, mc);
57
        start_measure(*mhp, rate, pubname, st, length, dst, mt);
58
}
59

    
60
/*
61
 * Register duplicate arrival
62
*/
63
void reg_chunk_duplicate()
64
{
65
        if (!chunk_dup) {
66
                enum stat_types st[] = {SUM, RATE};
67
                // number of chunks which have been received more then once
68
                add_measure(&chunk_dup, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunkDuplicates", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
69
                monNewSample(chunk_dup, 0);        //force publish even if there are no events
70
        }
71
        monNewSample(chunk_dup, 1);
72
}
73

    
74
/*
75
 * Register playout/loss of a chunk before playout
76
*/
77
void reg_chunk_playout(int id, bool b, uint64_t timestamp)
78
{
79
        static MonHandler chunk_loss_burst_size;
80
        static int last_arrived_chunk = -1;
81

    
82
        struct timeval tnow;
83
        if (!chunk_playout && b) {        //don't count losses before the first arrived chunk
84
                enum stat_types st[] = {WIN_AVG, AVG, SUM, RATE};
85
                //number of chunks played
86
                add_measure(&chunk_playout, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunksPlayed", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
87
        }
88
        monNewSample(chunk_playout, b);
89

    
90
        if (!playout_delay) {
91
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
92
                //delay after reorder buffer, however http module does not use reorder buffer
93
                add_measure(&playout_delay, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ReorderDelay", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[seconds]
94
        }
95
        if (b) {        //count delay only if chunk has arrived
96
                gettimeofday(&tnow, NULL);
97
                monNewSample(playout_delay, ((int64_t)(tnow.tv_usec + tnow.tv_sec * 1000000ULL) - (int64_t)timestamp) / 1000000.0);
98
        }
99

    
100
        //if (!chunk_loss_burst_size) {
101
        //        enum stat_types st[] = {WIN_AVG, WIN_VAR};
102
        //        // number of consecutive lost chunks
103
        //        add_measure(&chunk_loss_burst_size, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunkLossBurstSize", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
104
        //}
105
        if (b) {
106
                if (last_arrived_chunk >= 0) {
107
                        int burst_size = id - 1 - last_arrived_chunk;
108
                        if (burst_size) monNewSample(chunk_loss_burst_size, burst_size);
109
                }
110
                last_arrived_chunk = id;
111
        }
112
}
113

    
114
/*
115
 * Register actual neghbourhood size
116
*/
117
void reg_neigh_size(int s)
118
{
119
        if (!neigh_size) {
120
                enum stat_types st[] = {LAST};
121
                // number of peers in the neighboorhood
122
                add_measure(&neigh_size, GENERIC, 0, PEER_PUBLISH_INTERVAL, "NeighSize", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
123
        }
124
        monNewSample(neigh_size, s);
125
}
126

    
127
/*
128
 * Register chunk receive event
129
*/
130
void reg_chunk_receive(int id, uint64_t timestamp, int hopcount, bool old, bool dup)
131
{
132
        struct timeval tnow;
133

    
134
        if (!chunk_receive) {
135
                enum stat_types st[] = {RATE};
136
                // total number of received chunks per second
137
                add_measure(&chunk_receive, GENERIC, 0, PEER_PUBLISH_INTERVAL, "TotalRxChunk", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks/s]
138
                monNewSample(chunk_receive, 0);        //force publish even if there are no events
139
        }
140
        monNewSample(chunk_receive, 1);
141

    
142
        if (!chunk_hops) {
143
                enum stat_types st[] = {WIN_AVG};
144
                // number of hops from source on the p2p network
145
                add_measure(&chunk_hops, GENERIC, 0, PEER_PUBLISH_INTERVAL, "OverlayHops", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
146
        }
147
        monNewSample(chunk_hops, hopcount);
148

    
149
        if (!chunk_delay) {
150
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
151
                // time elapsed since the source emitted the chunk
152
                add_measure(&chunk_delay, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ReceiveDelay", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[seconds]
153
        }
154
        gettimeofday(&tnow, NULL);
155
        monNewSample(chunk_delay, ((int64_t)(tnow.tv_usec + tnow.tv_sec * 1000000ULL) - (int64_t)timestamp) / 1000000.0);
156
}
157

    
158
/*
159
 * Register chunk send event
160
*/
161
void reg_chunk_send(int id)
162
{
163
        if (!chunk_send) {
164
                enum stat_types st[] = {RATE};
165
                add_measure(&chunk_send, GENERIC, 0, PEER_PUBLISH_INTERVAL, "TotalTxChunk", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks/s]
166
                monNewSample(chunk_send, 0);        //force publish even if there are no events
167
        }
168
        monNewSample(chunk_send, 1);
169
}
170

    
171
/*
172
 * Register chunk accept evemt
173
*/
174
void reg_offer_accept(bool b)
175
{
176
        if (!offer_accept) {
177
                enum stat_types st[] = {WIN_AVG};
178
                // ratio between number of offers and number of accepts
179
                add_measure(&offer_accept, GENERIC, 0, PEER_PUBLISH_INTERVAL, "OfferAccept", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[no unit -> ratio]
180
        }
181
        monNewSample(offer_accept, b);
182
}
183

    
184
/*
185
 * Initialize peer level measurements
186
*/
187
void init_measures()
188
{
189
        if (peername) monSetPeerName(peername);
190
}
191

    
192
/*
193
 * End peer level measurements
194
*/
195
void end_measures()
196
{
197
}
198

    
199
/*
200
 * Initialize p2p measurements towards a peer
201
*/
202
void add_measures(struct nodeID *id)
203
{
204
        // Add measures
205
        int j = 0;
206
        enum stat_types stwinavgwinvar[] = {WIN_AVG, WIN_VAR};
207
        enum stat_types stwinavg[] = {WIN_AVG};
208
        enum stat_types stwinavgrate[] = {WIN_AVG, RATE};
209
//        enum stat_types stsum[] = {SUM};
210
        enum stat_types stsumwinsumrate[] = {SUM, PERIOD_SUM, WIN_SUM, RATE};
211

    
212
        dprintf("adding measures to %s\n",node_addr(id));
213

    
214
        /* HopCount */
215
        // number of hops at IP level
216
       id->mhs[j] = monCreateMeasure(HOPCOUNT, PACKET | IN_BAND);
217
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "HopCount", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[IP hops]
218

    
219
        /* Round Trip Time */
220
       id->mhs[j] = monCreateMeasure(RTT, PACKET | IN_BAND);
221
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "RoundTripDelay", stwinavgwinvar, sizeof(stwinavgwinvar)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[seconds]
222

    
223
        /* Loss */
224
       id->mhs[j] = monCreateMeasure(SEQWIN, PACKET | IN_BAND);
225
       start_measure(id->mhs[j++], 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
226
       id->mhs[j] = monCreateMeasure(LOSS, PACKET | IN_BAND);
227
       start_measure(id->mhs[j++], 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]
228

    
229
       /* RX,TX volume in bytes (only chunks) */
230
       id->mhs[j] = monCreateMeasure(RX_BYTE, PACKET | IN_BAND);
231
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "RxBytesChunk", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
232
       id->mhs[j] = monCreateMeasure(TX_BYTE, PACKET | IN_BAND);
233
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "TxBytesChunk", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
234

    
235
       /* RX,TX volume in bytes (only signaling) */
236
       id->mhs[j] = monCreateMeasure(RX_BYTE, PACKET | IN_BAND);
237
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "RxBytesSig", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes]
238
       id->mhs[j] = monCreateMeasure(TX_BYTE, PACKET | IN_BAND);
239
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "TxBytesSig", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes]
240

    
241
        // Chunks
242
       id->mhs[j] = monCreateMeasure(RX_PACKET, DATA | IN_BAND);
243
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "RxChunks", stwinavgrate, sizeof(stwinavgrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //RxChunks_sum [chunks] RxChunks_rate [chunks/sec]
244
       id->mhs[j] = monCreateMeasure(TX_PACKET, DATA | IN_BAND);
245
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "TxChunks", stwinavgrate, sizeof(stwinavgrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //TxChunks_sum [chunks] TxChunks_rate [chunks/sec]
246
       id->mhs[j] = monCreateMeasure(BULK_TRANSFER, PACKET | DATA | IN_BAND);
247
       start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "BulkTransfer", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK); //Bulktransfer [bit/s]
248

    
249
//        // Capacity
250
        id->mhs[j] = monCreateMeasure(CLOCKDRIFT, PACKET | IN_BAND);
251
        monSetParameter (id->mhs[j], P_CLOCKDRIFT_ALGORITHM, 1);
252
        start_measure(id->mhs[j++], 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
253
        id->mhs[j] = monCreateMeasure(CORRECTED_DELAY, PACKET | IN_BAND);
254
        start_measure(id->mhs[j++], 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
255
        id->mhs[j] = monCreateMeasure(CAPACITY_CAPPROBE, PACKET | IN_BAND);
256
        monSetParameter (id->mhs[j], P_CAPPROBE_DELAY_TH, -1);
257
//        monSetParameter (id->mhs[j], P_CAPPROBE_PKT_TH, 5);
258
        start_measure(id->mhs[j++], P2P_PUBLISH_INTERVAL, "Capacity", stwinavg, sizeof(stwinavg)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes/s]
259

    
260
        // for static must not be more then 10 or whatever size is in net_helper-ml.c
261
        id->n_mhs = j;
262
}
263

    
264
/*
265
 * Delete p2p measurements towards a peer
266
*/
267
void delete_measures(struct nodeID *id)
268
{
269
        dprintf("deleting measures from %s\n",node_addr(id));
270
        while(id->n_mhs) {
271
                monDestroyMeasure(id->mhs[--(id->n_mhs)]);
272
        }
273
}
274

    
275
/*
276
 * Helper to retrieve a measure
277
*/
278
double get_measure(struct nodeID *id, int j, enum stat_types st)
279
{
280
        return (id->n_mhs > j) ? monRetrieveResult(id->mhs[j], st) : NAN;
281
}
282

    
283
/*
284
 * Hopcount to a given peer
285
*/
286
int get_hopcount(struct nodeID *id){
287
        double r = get_measure(id, 0, LAST);
288
        return isnan(r) ? -1 : (int) r;
289
}
290

    
291
/*
292
 * RTT to a given peer in seconds
293
*/
294
double get_rtt(struct nodeID *id){
295
        return get_measure(id, 1, WIN_AVG);
296
}
297

    
298
/*
299
 * average RTT to a set of peers in seconds
300
*/
301
double get_average_rtt(struct nodeID **ids, int len){
302
        int i;
303
        int n = 0;
304
        double sum = 0;
305

    
306
        for (i = 0; i < len; i++) {
307
                double l = get_rtt(ids[i]);
308
                if (!isnan(l)) {
309
                        sum += l;
310
                        n++;
311
                }
312
        }
313
        return (n > 0) ? sum / n : NAN;
314
}
315

    
316
/*
317
 * loss ratio from a given peer as 0..1
318
*/
319
double get_lossrate(struct nodeID *id){
320
        return get_measure(id, 3, WIN_AVG);
321
}
322

    
323
/*
324
 * average loss ratio from a set of peers as 0..1
325
*/
326
double get_average_lossrate(struct nodeID **ids, int len){
327
        int i;
328
        int n = 0;
329
        double sum = 0;
330

    
331
        for (i = 0; i < len; i++) {
332
                double l = get_lossrate(ids[i]);
333
                if (!isnan(l)) {
334
                        sum += l;
335
                        n++;
336
                }
337
        }
338
        return (n > 0) ? sum / n : NAN;
339
}
340

    
341
double get_receive_delay(void) {
342
        return chunk_delay ? monRetrieveResult(chunk_delay, WIN_AVG) : NAN;
343
}