Statistics
| Branch: | Revision:

streamers / measures-monl.c @ 45395cd8

History | View | Annotate | Download (10.7 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
                // number of chunks which have been received more then once
60
                add_measure(&chunk_dup, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunkDuplicates", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
61
                monNewSample(chunk_dup, 0);        //force publish even if there are no events
62
        }
63
        monNewSample(chunk_dup, 1);
64
}
65

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

    
74
        struct timeval tnow;
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
                //number of chunks played
78
                add_measure(&chunk_playout, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ChunksPlayed", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks]
79
        }
80
        monNewSample(chunk_playout, b);
81

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

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

    
106
/*
107
 * Register actual neghbourhood size
108
*/
109
void reg_neigh_size(int s)
110
{
111
        if (!neigh_size) {
112
                enum stat_types st[] = {LAST};
113
                // number of peers in the neighboorhood
114
                add_measure(&neigh_size, GENERIC, 0, PEER_PUBLISH_INTERVAL, "NeighSize", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
115
        }
116
        monNewSample(neigh_size, s);
117
}
118

    
119
/*
120
 * Register chunk receive event
121
*/
122
void reg_chunk_receive(int id, uint64_t timestamp, int hopcount, bool old, bool dup)
123
{
124
        struct timeval tnow;
125

    
126
        if (!chunk_receive) {
127
                enum stat_types st[] = {RATE};
128
                // total number of received chunks per second
129
                add_measure(&chunk_receive, GENERIC, 0, PEER_PUBLISH_INTERVAL, "TotalRxChunk", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks/s]
130
                monNewSample(chunk_receive, 0);        //force publish even if there are no events
131
        }
132
        monNewSample(chunk_receive, 1);
133

    
134
        if (!chunk_hops) {
135
                enum stat_types st[] = {WIN_AVG};
136
                // number of hops from source on the p2p network
137
                add_measure(&chunk_hops, GENERIC, 0, PEER_PUBLISH_INTERVAL, "OverlayHops", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[peers]
138
        }
139
        monNewSample(chunk_hops, hopcount);
140

    
141
        if (!chunk_delay) {
142
                enum stat_types st[] = {WIN_AVG, WIN_VAR};
143
                // time elapsed since the source emitted the chunk
144
                add_measure(&chunk_delay, GENERIC, 0, PEER_PUBLISH_INTERVAL, "ReceiveDelay", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[seconds]
145
        }
146
        gettimeofday(&tnow, NULL);
147
        monNewSample(chunk_delay, ((int64_t)(tnow.tv_usec + tnow.tv_sec * 1000000ULL) - (int64_t)timestamp) / 1000000.0);
148
}
149

    
150
/*
151
 * Register chunk send event
152
*/
153
void reg_chunk_send(int id)
154
{
155
        if (!chunk_send) {
156
                enum stat_types st[] = {RATE};
157
                add_measure(&chunk_send, GENERIC, 0, PEER_PUBLISH_INTERVAL, "TotalTxChunk", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[chunks/s]
158
                monNewSample(chunk_send, 0);        //force publish even if there are no events
159
        }
160
        monNewSample(chunk_send, 1);
161
}
162

    
163
/*
164
 * Register chunk accept evemt
165
*/
166
void reg_offer_accept(bool b)
167
{
168
        if (!offer_accept) {
169
                enum stat_types st[] = {WIN_AVG};
170
                // ratio between number of offers and number of accepts
171
                add_measure(&offer_accept, GENERIC, 0, PEER_PUBLISH_INTERVAL, "OfferAccept", st, sizeof(st)/sizeof(enum stat_types), NULL, MSG_TYPE_ANY);        //[no unit -> ratio]
172
        }
173
        monNewSample(offer_accept, b);
174
}
175

    
176
/*
177
 * Initialize peer level measurements
178
*/
179
void init_measures()
180
{
181
        if (peername) monSetPeerName(peername);
182
}
183

    
184
/*
185
 * End peer level measurements
186
*/
187
void end_measures()
188
{
189
}
190

    
191
/*
192
 * Initialize p2p measurements towards a peer
193
*/
194
void add_measures(struct nodeID *id)
195
{
196
        // Add measures
197
        int j = 0;
198
        enum stat_types stwinavgwinvar[] = {WIN_AVG, WIN_VAR};
199
        enum stat_types stwinavg[] = {WIN_AVG};
200
        enum stat_types stwinavgrate[] = {WIN_AVG, RATE};
201
        enum stat_types stsum[] = {SUM};
202
        enum stat_types stsumwinsumrate[] = {SUM, WIN_SUM, RATE};
203

    
204
        dprintf("adding measures to %s\n",node_addr(id));
205

    
206
        /* HopCount */
207
        // number of hops at IP level
208
       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]
209

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

    
213
        /* Loss */
214
       add_measure(&id->mhs[j++], SEQWIN, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
215
       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]
216

    
217
       /* RX,TX volume in bytes (only chunks) */
218
       add_measure(&id->mhs[j++], RX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "RxBytesChunk", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
219
       add_measure(&id->mhs[j++], TX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "TxBytesChunk", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //[bytes]
220

    
221
       /* RX,TX volume in bytes (only signaling) */
222
       add_measure(&id->mhs[j++], RX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "RxBytesSig", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes]
223
       add_measure(&id->mhs[j++], TX_BYTE, PACKET | IN_BAND, P2P_PUBLISH_INTERVAL, "TxBytesSig", stsumwinsumrate, sizeof(stsumwinsumrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_SIGNALLING);        //[bytes]
224

    
225
        // Chunks
226
       add_measure(&id->mhs[j++], RX_PACKET, DATA | IN_BAND, P2P_PUBLISH_INTERVAL, "RxChunks", stwinavgrate, sizeof(stwinavgrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //RxChunks_sum [chunks] RxChunks_rate [chunks/sec]
227
       add_measure(&id->mhs[j++], TX_PACKET, DATA | IN_BAND, P2P_PUBLISH_INTERVAL, "TxChunks", stwinavgrate, sizeof(stwinavgrate)/sizeof(enum stat_types), id->addr, MSG_TYPE_CHUNK);        //TxChunks_sum [chunks] TxChunks_rate [chunks/sec]
228
//        // Capacity
229
//        add_measure(&id->mhs[j++], CLOCKDRIFT, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
230
//        monSetParameter (id->mhs[j], P_CLOCKDRIFT_ALGORITHM, 1);
231
//        add_measure(&id->mhs[j++], CORRECTED_DELAY, PACKET | IN_BAND, 0, NULL, NULL, 0, id->addr, MSG_TYPE_CHUNK);
232
//        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]
233
//        monSetParameter (id->mhs[j], P_CAPPROBE_DELAY_TH, -1);
234
//        monSetParameter (mh, P_CAPPROBE_PKT_TH, 100);
235
//        monSetParameter (mh, P_CAPPROBE_IPD_TH, 60);
236
//        monPublishStatisticalType(mh, NULL, st , sizeof(st)/sizeof(enum stat_types), repoclient);
237

    
238
        // for static must not be more then 10 or whatever size is in net_helper-ml.c
239
        id->n_mhs = j;
240
}
241

    
242
/*
243
 * Delete p2p measurements towards a peer
244
*/
245
void delete_measures(struct nodeID *id)
246
{
247
        dprintf("deleting measures from %s\n",node_addr(id));
248
        while(id->n_mhs) {
249
                monDestroyMeasure(id->mhs[--(id->n_mhs)]);
250
        }
251
}
252

    
253
/*
254
 * Helper to retrieve a measure
255
*/
256
double get_measure(struct nodeID *id, int j, enum stat_types st)
257
{
258
        return (id->n_mhs > j) ? monRetrieveResult(id->mhs[j], st) : NAN;
259
}
260

    
261
/*
262
 * Hopcount to a given peer
263
*/
264
int get_hopcount(struct nodeID *id){
265
        double r = get_measure(id, 0, LAST);
266
        return isnan(r) ? -1 : (int) r;
267
}
268

    
269
/*
270
 * RTT to a given peer in seconds
271
*/
272
double get_rtt(struct nodeID *id){
273
        return get_measure(id, 1, WIN_AVG);
274
}
275

    
276
/*
277
 * average RTT to a set of peers in seconds
278
*/
279
double get_average_rtt(struct nodeID **ids, int len){
280
        int i;
281
        int n = 0;
282
        double sum = 0;
283

    
284
        for (i = 0; i < len; i++) {
285
                double l = get_rtt(ids[i]);
286
                if (!isnan(l)) {
287
                        sum += l;
288
                        n++;
289
                }
290
        }
291
        return (n > 0) ? sum / n : NAN;
292
}
293

    
294
/*
295
 * loss ratio from a given peer as 0..1
296
*/
297
double get_lossrate(struct nodeID *id){
298
        return get_measure(id, 3, WIN_AVG);
299
}
300

    
301
/*
302
 * average loss ratio from a set of peers as 0..1
303
*/
304
double get_average_lossrate(struct nodeID **ids, int len){
305
        int i;
306
        int n = 0;
307
        double sum = 0;
308

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

    
319
double get_receive_delay(void) {
320
        return chunk_delay ? monRetrieveResult(chunk_delay, WIN_AVG) : NAN;
321
}