Statistics
| Branch: | Revision:

streamers / streamer.c @ ef7fb5ca

History | View | Annotate | Download (12.1 KB)

1 89e893e2 Luca
/*
2 8fed7779 CsabaKiraly
 *  Copyright (c) 2010 Luca Abeni
3
 *  Copyright (c) 2010 Csaba Kiraly
4 89e893e2 Luca
 *
5 8fed7779 CsabaKiraly
 *  This is free software; see gpl-3.0.txt
6 89e893e2 Luca
 */
7
#include <stdlib.h>
8
#include <stdint.h>
9
#include <stdio.h>
10 46c24e94 Csaba Kiraly
#include <stdbool.h>
11 89e893e2 Luca
#include <string.h>
12
#include <getopt.h>
13 5dcfffbc CsabaKiraly
#include <signal.h>
14 d7a4d49c Csaba Kiraly
#include <time.h>
15 839ae781 Csaba Kiraly
#include <errno.h>
16
#include <math.h>
17
#ifndef NAN        //NAN is missing in some old math.h versions
18
#define NAN            (0.0/0.0)
19
#endif
20
21 e28a1487 CsabaKiraly
#include <grapes_msg_types.h>
22 89e893e2 Luca
#include <net_helper.h>
23
24 95e6fe96 CsabaKiraly
#ifdef WIN32
25
#include <winsock2.h>
26
#endif
27
28 89e893e2 Luca
#include "net_helpers.h"
29
#include "loop.h"
30 f0bb64e2 Csaba Kiraly
#include "output.h"
31 581339c4 Csaba Kiraly
#include "channel.h"
32 f088e50e MarcoBiazzini
#include "topology.h"
33 a8a31ce0 Csaba Kiraly
#include "measures.h"
34 1d21cb23 Csaba Kiraly
#include "streamer.h"
35
36 648aa768 Csaba Kiraly
#ifndef EXTRAVERSION
37
#define EXTRAVERSION "Unknown"
38
#endif
39
40 1d21cb23 Csaba Kiraly
static struct nodeID *my_sock;
41 89e893e2 Luca
42 7c21b5b1 CsabaKiraly
const char *peername = NULL;
43
44 613263e9 Csaba Kiraly
static const char *my_iface = NULL;
45 89e893e2 Luca
static int port = 6666;
46 5f0575fb Stefano Traverso
static int srv_port = 0;
47 0f23522a Luca
static const char *srv_ip = "";
48 ef7fb5ca Stefano Traverso
static int period = 15;
49 66f50156 Csaba Kiraly
static int chunks_per_second = 25;
50 839ae781 Csaba Kiraly
static double capacity_override = NAN;
51 66f84460 CsabaKiraly
static int multiply = 3;
52 789d6502 Csaba Kiraly
static int buff_size = 50;
53 66f84460 CsabaKiraly
static int outbuff_size = 50;
54 adb2a9b2 CsabaKiraly
static const char *fname = "/dev/stdin";
55 03de31e0 Csaba Kiraly
static const char *output_config ="";
56 5f0575fb Stefano Traverso
static bool loop_input = false;
57 193622ac CsabaKiraly
static const char *net_helper_config = "";
58 5f72f6b3 Csaba Kiraly
static const char *topo_config = "";
59 33d23b91 MarcoBiazzini
unsigned char msgTypes[] = {MSG_TYPE_CHUNK,MSG_TYPE_SIGNALLING};
60 1cd06c26 CsabaKiraly
bool chunk_log = false;
61 9082bbcd Stefano Traverso
static bool hrc = true;
62
static int number_offer_thread = -1;
63 4536b69e Csaba Kiraly
static int randomize_start = 0;
64 a2fa286f Luca Abeni
int start_id = -1;
65
int end_id = -1;
66
int initial_id = -1;
67 89e893e2 Luca
68 20663df3 Csaba Kiraly
extern int NEIGHBORHOOD_TARGET_SIZE;
69 3730fda0 Csaba Kiraly
extern uint64_t CB_SIZE_TIME;
70 9818ca86 Csaba Kiraly
extern double desired_bw;
71 4ca3a6bd Luca Abeni
extern double desired_rtt;
72
extern double alpha_target;
73 f672dcde Csaba Kiraly
extern double topo_mem;
74 79ef0588 Csaba Kiraly
extern bool topo_out;
75
extern bool topo_in;
76 f6e9f400 Csaba Kiraly
extern bool topo_keep_best;
77
extern bool topo_add_best;
78 20663df3 Csaba Kiraly
79 d8288eb0 Csaba Kiraly
#ifndef MONL
80 75b2b975 Csaba Kiraly
extern struct timeval print_tdiff;
81
extern struct timeval tstartdiff;
82 d8288eb0 Csaba Kiraly
#endif
83 75b2b975 Csaba Kiraly
84 43c2d7d6 CsabaKiraly
static void print_usage(int argc, char *argv[])
85 a02fc4cf MarcoMellia
{
86
  fprintf (stderr,
87 60784aa1 CsabaKiraly
    "Usage:%s [options]\n"
88 a02fc4cf MarcoMellia
    "\n"
89
    "Peer options\n"
90
    "\t[-p port]: port of the remote peer to connect at during bootstrap.\n"
91
    "\t           Usually it is the source peer port.\n"
92
    "\t[-i IP]: IP address of the remote peer to connect at during bootstrap.\n"
93
    "\t         Usually it is the source peer IP\n"
94
    "\t[-C name]: set the channel name to use on the repository.\n"
95
    "\t           All peers should use the same channel name.\n"
96
    "\n"
97
    "\t[-b size]: set the peer Chunk Buffer size.\n"
98
    "\t           This is also the chunk trading window size.\n"
99
    "\t[-o size]: set the Output Buffer size.\n"
100 5f0575fb Stefano Traverso
    "\t[-c thread]: set the number of signalling threads a peer can send per seconds.\n"
101 9082bbcd Stefano Traverso
    "\t             Default: set it automatically via HRC.\n"
102 20663df3 Csaba Kiraly
    "\t[-M peers]: neighbourhood target size.\n"
103 5f72f6b3 Csaba Kiraly
    "\t[-t config]: topology config.\n"
104 a02fc4cf MarcoMellia
    "\t[-P port]: local UDP port to be used by the peer.\n"
105 60784aa1 CsabaKiraly
    "\t[-I iface]: local netwok interface to be used by the peer.\n"
106 a02fc4cf MarcoMellia
    "\t         Useful if the host has several interfaces/addresses.\n"
107 7c21b5b1 CsabaKiraly
    "\t[-N name]: set the name of the peer.\n"
108
    "\t         This name will be used when publishing in the repository.\n"
109 193622ac CsabaKiraly
    "\t[-n options]: pass configuration options to the net-helper\n"
110 1cd06c26 CsabaKiraly
    "\t[--chunk_log]: print a chunk level log on stderr\n"
111 60784aa1 CsabaKiraly
    "\t[-F config]: configure the output module\n"
112 4ca3a6bd Luca Abeni
    "\t[-a alpha]: set the topology alpha value (from 0 to 100)\n"
113 29aaab6c Csaba Kiraly
    "\t[-r rtt]: set the RTT threshold (in ms) for desired neighbours\n"
114 9818ca86 Csaba Kiraly
    "\t[--desired_bw bw]: set the BW threshold (in bits/s) for desired neighbours. Use of K(ilo), M(ega) allowed, e.g 0.8M\n"
115 f672dcde Csaba Kiraly
    "\t[--topo_mem p]: keep p (0..1) portion of peers between topology operations\n"
116 79ef0588 Csaba Kiraly
    "\t[--topo_out]: peers only choose out-neighbours\n"
117
    "\t[--topo_in]: peers only choose in-neighbours\n"
118
    "\t[--topo_bidir]: peers choose both in- and out-neighbours (bidir)\n"
119 f6e9f400 Csaba Kiraly
    "\t[--topo_keep_best]: keep best peers, not random subset\n"
120
    "\t[--topo_add_best]: add best peers among desired ones, not random subset\n"
121 a02fc4cf MarcoMellia
    "\n"
122
    "Special Source Peer options\n"
123
    "\t[-m chunks]: set the number of copies the source injects in the overlay.\n"
124
    "\t[-f filename]: name of the video stream file to transmit.\n"
125 5f0575fb Stefano Traverso
    "\t[-l]: loop the video stream.\n"
126 a2fa286f Luca Abeni
    "\t[-S]: set initial chunk_id (source only).\n"
127
    "\t[-s]: set start_id from which to start output.\n"
128
    "\t[-e]: set end_id at which to end output.\n"
129 a02fc4cf MarcoMellia
    "\n"
130 4536b69e Csaba Kiraly
    "Special options\n"
131 c4044fb9 Csaba Kiraly
    "\t[--randomize_start us]: random wait before starting [0..us] microseconds.\n"
132 648aa768 Csaba Kiraly
    "\t[-v]: print version.\n"
133 4536b69e Csaba Kiraly
    "\n"
134 60784aa1 CsabaKiraly
    "NOTE: by deafult the peer will dump the received video on STDOUT in raw format\n"
135 a02fc4cf MarcoMellia
    "      it can be played by your favourite player simply using a pipe\n"
136 60784aa1 CsabaKiraly
    "      e.g., | cvlc /dev/stdin\n"
137 a02fc4cf MarcoMellia
    "\n"
138
    "Examples:\n"
139
    "\n"
140
    "Start a source peer on port 6600:\n"
141
    "\n"
142 60784aa1 CsabaKiraly
    "%s -l -f foreman.mpg -P 6600\n"
143 a02fc4cf MarcoMellia
    "\n"
144
    "Start a peer connecting to the previous source, and using videolan as player:\n"
145
    "\n"
146 60784aa1 CsabaKiraly
    "%s -i <sourceIP> -p <sourcePort> | cvlc /dev/stdin\n"
147
    "=======================================================\n", argv[0], argv[0], argv[0]
148 a02fc4cf MarcoMellia
    );
149
  }
150
151
152 c7beb713 Csaba Kiraly
static double atod_kmg(const char *s) {
153
  double d;
154
  char *e;
155
156
  errno = 0;
157
  d = strtod(s, &e);
158
  if (errno) {
159
    fprintf(stderr, "Error parsing option: %s\n", s);
160
    exit(-1);
161
  }
162
  switch (*e) {
163
    case 'g':
164
    case 'G':
165
      d *= 1024;
166
    case 'm':
167
    case 'M':
168
      d *= 1024;
169
    case 'k':
170
    case 'K':
171
      d *= 1024;
172
    case 0:
173
      break;
174
    default:
175
      fprintf(stderr, "Error parsing option: %s\n", s);
176
      exit(-1);
177
  }
178
179
  return d;
180
}
181 a02fc4cf MarcoMellia
182 89e893e2 Luca
static void cmdline_parse(int argc, char *argv[])
183
{
184
  int o;
185
186 4aaa8891 MatteoSammarco
  int option_index = 0;
187
  static struct option long_options[] = {
188 e63b7606 MatteoSammarco
        {"chunk_log", no_argument, 0, 0},
189 75b2b975 Csaba Kiraly
        {"measure_start", required_argument, 0, 0},
190
        {"measure_every", required_argument, 0, 0},
191 3730fda0 Csaba Kiraly
        {"playout_limit", required_argument, 0, 0},
192 4536b69e Csaba Kiraly
        {"randomize_start", required_argument, 0, 0},
193 839ae781 Csaba Kiraly
        {"capacity_override", required_argument, 0, 0},
194 9818ca86 Csaba Kiraly
        {"desired_bw", required_argument, 0, 0},
195 f672dcde Csaba Kiraly
        {"topo_mem", required_argument, 0, 0},
196 79ef0588 Csaba Kiraly
        {"topo_in", no_argument, 0, 0},
197
        {"topo_out", no_argument, 0, 0},
198
        {"topo_bidir", no_argument, 0, 0},
199 f6e9f400 Csaba Kiraly
        {"topo_keep_best", no_argument, 0, 0},
200
        {"topo_add_best", no_argument, 0, 0},
201 e63b7606 MatteoSammarco
        {0, 0, 0, 0}
202 4aaa8891 MatteoSammarco
  };
203
204 648aa768 Csaba Kiraly
    while ((o = getopt_long (argc, argv, "r:a:b:o:c:p:i:P:I:f:F:m:lC:N:n:M:t:s:e:S:v",long_options, &option_index)) != -1) { //use this function to manage long options
205 89e893e2 Luca
    switch(o) {
206 4aaa8891 MatteoSammarco
      case 0: //for long options
207 1cd06c26 CsabaKiraly
        if( strcmp( "chunk_log", long_options[option_index].name ) == 0 ) { chunk_log = true; }
208 d8288eb0 Csaba Kiraly
#ifndef MONL
209 75b2b975 Csaba Kiraly
        if( strcmp( "measure_start", long_options[option_index].name ) == 0 ) { tstartdiff.tv_sec = atoi(optarg); }
210
        if( strcmp( "measure_every", long_options[option_index].name ) == 0 ) { print_tdiff.tv_sec = atoi(optarg); }
211 d8288eb0 Csaba Kiraly
#endif
212 3730fda0 Csaba Kiraly
        if( strcmp( "playout_limit", long_options[option_index].name ) == 0 ) { CB_SIZE_TIME = atoi(optarg); }
213 4536b69e Csaba Kiraly
        if( strcmp( "randomize_start", long_options[option_index].name ) == 0 ) { randomize_start = atoi(optarg); }
214 839ae781 Csaba Kiraly
        if( strcmp( "capacity_override", long_options[option_index].name ) == 0 ) { capacity_override = atod_kmg(optarg); }
215 9818ca86 Csaba Kiraly
        if( strcmp( "desired_bw", long_options[option_index].name ) == 0 ) { desired_bw = atod_kmg(optarg); }
216 f672dcde Csaba Kiraly
        if( strcmp( "topo_mem", long_options[option_index].name ) == 0 ) { topo_mem = atof(optarg); }
217 79ef0588 Csaba Kiraly
        else if( strcmp( "topo_in", long_options[option_index].name ) == 0 ) { topo_in = true; topo_out = false; }
218
        else if( strcmp( "topo_out", long_options[option_index].name ) == 0 ) { topo_in = false; topo_out = true; }
219
        else if( strcmp( "topo_bidir", long_options[option_index].name ) == 0 ) { topo_in = true; topo_out = true; }
220 f6e9f400 Csaba Kiraly
        else if( strcmp( "topo_keep_best", long_options[option_index].name ) == 0 ) { topo_keep_best = true; }
221
        else if( strcmp( "topo_add_best", long_options[option_index].name ) == 0 ) { topo_add_best = true; }
222 1cd06c26 CsabaKiraly
        break;
223 4ca3a6bd Luca Abeni
      case 'a':
224
        alpha_target = (double)atoi(optarg) / 100.0;
225
        break;
226
      case 'r':
227
        desired_rtt = (double)atoi(optarg) / 1000.0;
228
        break;
229 90f3a290 CsabaKiraly
      case 'b':
230 0a40460a Luca
        buff_size = atoi(optarg);
231
        break;
232 90f3a290 CsabaKiraly
      case 'o':
233 f0bb64e2 Csaba Kiraly
        outbuff_size = atoi(optarg);
234
        break;
235 5f0575fb Stefano Traverso
      case 'c':
236
        number_offer_thread = atoi(optarg);
237 9082bbcd Stefano Traverso
        if (number_offer_thread >= 0)
238
                hrc = false;
239
        else
240
                hrc = true;
241 0a40460a Luca
        break;
242 90f3a290 CsabaKiraly
      case 'm':
243 59b7cfc6 Csaba Kiraly
        multiply = atoi(optarg);
244
        break;
245 90f3a290 CsabaKiraly
      case 'p':
246 89e893e2 Luca
        srv_port = atoi(optarg);
247
        break;
248 90f3a290 CsabaKiraly
      case 'i':
249 89e893e2 Luca
        srv_ip = strdup(optarg);
250
        break;
251 90f3a290 CsabaKiraly
      case 'P':
252 89e893e2 Luca
        port =  atoi(optarg);
253
        break;
254 90f3a290 CsabaKiraly
      case 'I':
255 568bd1bd Luca Abeni
        my_iface = strdup(optarg);
256 89e893e2 Luca
        break;
257 90f3a290 CsabaKiraly
      case 'f':
258 e37511b1 Luca Abeni
        fname = strdup(optarg);
259
        break;
260 463000a4 CsabaKiraly
      case 'F':
261
        output_config = strdup(optarg);
262
        break;
263 5f0575fb Stefano Traverso
      case 'l':
264
        loop_input = true;
265
        break;
266 90f3a290 CsabaKiraly
      case 'C':
267 581339c4 Csaba Kiraly
        channel_set_name(optarg);
268 8253c8ea CsabaKiraly
        break;
269 193622ac CsabaKiraly
      case 'n':
270
        net_helper_config = strdup(optarg);
271 581339c4 Csaba Kiraly
        break;
272 7c21b5b1 CsabaKiraly
      case 'N':
273
        peername = strdup(optarg);
274
        break;
275 20663df3 Csaba Kiraly
      case 'M':
276
        NEIGHBORHOOD_TARGET_SIZE = atoi(optarg);
277
        break;
278 5f72f6b3 Csaba Kiraly
      case 't':
279
        topo_config = strdup(optarg);
280
        break;
281 a2fa286f Luca Abeni
      case 'S':
282
        initial_id = atoi(optarg);
283
        break;
284
      case 's':
285
        start_id = atoi(optarg);
286
        break;
287
      case 'e':
288
        end_id = atoi(optarg);
289
        break;
290 648aa768 Csaba Kiraly
      case 'v':
291
        fprintf(stderr, "Version: %s\n", EXTRAVERSION);
292
            exit(0);
293 89e893e2 Luca
      default:
294
        fprintf(stderr, "Error: unknown option %c\n", o);
295 43c2d7d6 CsabaKiraly
        print_usage(argc, argv);
296 89e893e2 Luca
297
        exit(-1);
298
    }
299
  }
300 581339c4 Csaba Kiraly
301
  if (!channel_get_name()) {
302
    channel_set_name("generic");
303
  }
304 37b749c8 CsabaKiraly
305
  if (argc <= 1) {
306
    print_usage(argc, argv);
307
    fprintf(stderr, "Trying to start a source with default parameters, reading from %s\n", fname);
308
  }
309 89e893e2 Luca
}
310
311 1d21cb23 Csaba Kiraly
const struct nodeID *get_my_addr(void)
312
{
313
  return my_sock;
314
}
315
316 7a673aa5 Csaba Kiraly
static void init_rand(const char * s)
317
{
318
  long i, l, x;
319
  struct timeval t;
320
321
  gettimeofday(&t, NULL);
322
323
  x = 1;
324
  l = strlen(s);
325
  for (i = 0; i < l; i++) {
326
    x *= s[i];
327
  }
328
329
  srand(t.tv_usec * t.tv_sec * x);
330
}
331
332 89e893e2 Luca
static struct nodeID *init(void)
333
{
334 bdd19faf Csaba Kiraly
  int i;
335 89e893e2 Luca
  struct nodeID *myID;
336 613263e9 Csaba Kiraly
  char *my_addr;
337
338
  if (my_iface) {
339
    my_addr = iface_addr(my_iface);
340
  } else {
341 371a4639 Csaba Kiraly
    my_addr = default_ip_addr();
342 613263e9 Csaba Kiraly
  }
343 89e893e2 Luca
344 f858e682 Luca Abeni
  if (my_addr == NULL) {
345
    fprintf(stderr, "Cannot find network interface %s\n", my_iface);
346
347
    return NULL;
348
  }
349 33d23b91 MarcoBiazzini
  for (i=0;i<2;i++)
350 bdd19faf Csaba Kiraly
          bind_msg_type(msgTypes[i]);
351 193622ac CsabaKiraly
  myID = net_helper_init(my_addr, port, net_helper_config);
352 4136911a Luca Abeni
  if (myID == NULL) {
353 89e893e2 Luca
    fprintf(stderr, "Error creating my socket (%s:%d)!\n", my_addr, port);
354 568bd1bd Luca Abeni
    free(my_addr);
355 4136911a Luca Abeni
356 e63ea799 Luca
    return NULL;
357 89e893e2 Luca
  }
358 568bd1bd Luca Abeni
  free(my_addr);
359 5331023a CsabaKiraly
  fprintf(stderr, "My network ID is: %s\n", node_addr(myID));
360 7a673aa5 Csaba Kiraly
361
  init_rand(node_addr(myID));
362
363 5f72f6b3 Csaba Kiraly
  topologyInit(myID, topo_config);
364 89e893e2 Luca
365
  return myID;
366
}
367
368 5dcfffbc CsabaKiraly
void leave(int sig) {
369 a4a7d0f1 Csaba Kiraly
  fprintf(stderr, "Received signal %d, exiting!\n", sig);
370 a8a31ce0 Csaba Kiraly
  end_measures();
371 5dcfffbc CsabaKiraly
  exit(sig);
372
}
373 89e893e2 Luca
374 c4044fb9 Csaba Kiraly
// wait [0..max] microsec
375 4536b69e Csaba Kiraly
static void random_wait(int max) {
376
    struct timespec t;
377
    uint64_t ms;
378
379
    ms = (rand()/(RAND_MAX + 1.0)) * max;
380
    t.tv_sec = ms / 1000000;
381
    t.tv_nsec = (ms % 1000000) * 1000;
382
    nanosleep(&t, NULL);
383
}
384
385 89e893e2 Luca
int main(int argc, char *argv[])
386
{
387
388 5dcfffbc CsabaKiraly
  (void) signal(SIGTERM,leave);
389 fbf7067c Csaba Kiraly
  (void) signal(SIGINT,leave);
390 5dcfffbc CsabaKiraly
391 89e893e2 Luca
  cmdline_parse(argc, argv);
392
393
  my_sock = init();
394 e63ea799 Luca
  if (my_sock == NULL) {
395 1931ff6c Csaba Kiraly
    fprintf(stderr, "Cannot initialize streamer, exiting!\n");
396 e63ea799 Luca
    return -1;
397
  }
398 89e893e2 Luca
  if (srv_port != 0) {
399 74a5d4ae CsabaKiraly
    struct nodeID *srv;
400 89e893e2 Luca
401 d7a4d49c Csaba Kiraly
    //random wait a bit before starting
402 4536b69e Csaba Kiraly
    if (randomize_start) random_wait(randomize_start);
403 d7a4d49c Csaba Kiraly
404 463000a4 CsabaKiraly
    output_init(outbuff_size, output_config);
405
406 4136911a Luca Abeni
    srv = create_node(srv_ip, srv_port);
407 0f23522a Luca
    if (srv == NULL) {
408
      fprintf(stderr, "Cannot resolve remote address %s:%d\n", srv_ip, srv_port);
409
410
      return -1;
411
    }
412 5f0575fb Stefano Traverso
    
413 f088e50e MarcoBiazzini
    topoAddNeighbour(srv, NULL, 0);
414 9082bbcd Stefano Traverso
    loop(my_sock, period * 1000, buff_size, hrc, number_offer_thread);
415 98c28f0f CsabaKiraly
  } else {
416 03de31e0 Csaba Kiraly
    source_loop(fname, my_sock, period * 1000, multiply, buff_size);
417 89e893e2 Luca
  }
418
  return 0;
419
}
420 b88c9f4a Csaba Kiraly
421 70921454 Csaba Kiraly
int am_i_source()
422
{
423
  return (srv_port == 0);
424
}
425
426 b88c9f4a Csaba Kiraly
int get_cb_size()
427
{
428
  return buff_size;
429
}
430 815ebf31 Csaba Kiraly
431 839ae781 Csaba Kiraly
//capacity in bits/s, or NAN if unknown
432
double get_capacity()
433
{
434
  return capacity_override;
435
}