Statistics
| Branch: | Revision:

streamers / streamer.c @ 03de31e0

History | View | Annotate | Download (11.8 KB)

1
/*
2
 *  Copyright (c) 2010 Luca Abeni
3
 *  Copyright (c) 2010 Csaba Kiraly
4
 *
5
 *  This is free software; see gpl-3.0.txt
6
 */
7
#include <stdlib.h>
8
#include <stdint.h>
9
#include <stdio.h>
10
#include <stdbool.h>
11
#include <string.h>
12
#include <getopt.h>
13
#include <signal.h>
14
#include <time.h>
15
#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
#include <grapes_msg_types.h>
22
#include <net_helper.h>
23

    
24
#ifdef WIN32
25
#include <winsock2.h>
26
#endif
27

    
28
#include "net_helpers.h"
29
#include "loop.h"
30
#include "output.h"
31
#include "channel.h"
32
#include "topology.h"
33
#include "measures.h"
34
#include "streamer.h"
35

    
36
static struct nodeID *my_sock;
37

    
38
const char *peername = NULL;
39

    
40
static const char *my_iface = NULL;
41
static int port = 6666;
42
static int srv_port;
43
static const char *srv_ip = "";
44
static int period = 40;
45
static int chunks_per_second = 25;
46
static double capacity_override = NAN;
47
#ifdef HTTPIO
48
//input-http.c needs this in order to accomplish the -m multiple send_chunk()
49
int multiply = 3;
50
#else
51
static int multiply = 3;
52
#endif
53
static int buff_size = 50;
54
static int outbuff_size = 50;
55
static const char *fname = "/dev/stdin";
56
static const char *output_config ="";
57
static const char *net_helper_config = "";
58
static const char *topo_config = "";
59
unsigned char msgTypes[] = {MSG_TYPE_CHUNK,MSG_TYPE_SIGNALLING};
60
bool chunk_log = false;
61
static int randomize_start = 0;
62
int start_id = -1;
63
int end_id = -1;
64
int initial_id = -1;
65

    
66
extern int NEIGHBORHOOD_TARGET_SIZE;
67
extern uint64_t CB_SIZE_TIME;
68
extern double desired_bw;
69
extern double desired_rtt;
70
extern double alpha_target;
71
extern double topo_mem;
72
extern bool topo_out;
73
extern bool topo_in;
74
extern bool topo_keep_best;
75
extern bool topo_add_best;
76

    
77
#ifndef MONL
78
extern struct timeval print_tdiff;
79
extern struct timeval tstartdiff;
80
#endif
81

    
82
static void print_usage(int argc, char *argv[])
83
{
84
  fprintf (stderr,
85
    "Usage:%s [options]\n"
86
    "\n"
87
    "Peer options\n"
88
    "\t[-p port]: port of the remote peer to connect at during bootstrap.\n"
89
    "\t           Usually it is the source peer port.\n"
90
    "\t[-i IP]: IP address of the remote peer to connect at during bootstrap.\n"
91
    "\t         Usually it is the source peer IP\n"
92
    "\t[-C name]: set the channel name to use on the repository.\n"
93
    "\t           All peers should use the same channel name.\n"
94
    "\n"
95
    "\t[-b size]: set the peer Chunk Buffer size.\n"
96
    "\t           This is also the chunk trading window size.\n"
97
    "\t[-o size]: set the Output Buffer size.\n"
98
    "\t[-c chunks]: set the number of chunks a peer can send per seconds.\n"
99
    "\t             it controls the upload capacity of peer as well.\n"
100
    "\t[-M peers]: neighbourhood target size.\n"
101
    "\t[-t config]: topology config.\n"
102
    "\t[-P port]: local UDP port to be used by the peer.\n"
103
    "\t[-I iface]: local netwok interface to be used by the peer.\n"
104
    "\t         Useful if the host has several interfaces/addresses.\n"
105
    "\t[-N name]: set the name of the peer.\n"
106
    "\t         This name will be used when publishing in the repository.\n"
107
    "\t[-n options]: pass configuration options to the net-helper\n"
108
    "\t[--chunk_log]: print a chunk level log on stderr\n"
109
    "\t[-F config]: configure the output module\n"
110
    "\t[-a alpha]: set the topology alpha value (from 0 to 100)\n"
111
    "\t[-r rtt]: set the RTT threshold (in ms) for desired neighbours\n"
112
    "\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"
113
    "\t[--topo_mem p]: keep p (0..1) portion of peers between topology operations\n"
114
    "\t[--topo_out]: peers only choose out-neighbours\n"
115
    "\t[--topo_in]: peers only choose in-neighbours\n"
116
    "\t[--topo_bidir]: peers choose both in- and out-neighbours (bidir)\n"
117
    "\t[--topo_keep_best]: keep best peers, not random subset\n"
118
    "\t[--topo_add_best]: add best peers among desired ones, not random subset\n"
119
    "\n"
120
    "Special Source Peer options\n"
121
    "\t[-m chunks]: set the number of copies the source injects in the overlay.\n"
122
    "\t[-f filename]: name of the video stream file to transmit.\n"
123
    "\t[-S]: set initial chunk_id (source only).\n"
124
    "\t[-s]: set start_id from which to start output.\n"
125
    "\t[-e]: set end_id at which to end output.\n"
126
    "\n"
127
    "Special options\n"
128
    "\t[--randomize_start ms]: random wait before starting in to ms millisecs.\n"
129
    "\n"
130
    "NOTE: by deafult the peer will dump the received video on STDOUT in raw format\n"
131
    "      it can be played by your favourite player simply using a pipe\n"
132
    "      e.g., | cvlc /dev/stdin\n"
133
    "\n"
134
    "Examples:\n"
135
    "\n"
136
    "Start a source peer on port 6600:\n"
137
    "\n"
138
    "%s -l -f foreman.mpg -P 6600\n"
139
    "\n"
140
    "Start a peer connecting to the previous source, and using videolan as player:\n"
141
    "\n"
142
    "%s -i <sourceIP> -p <sourcePort> | cvlc /dev/stdin\n"
143
    "=======================================================\n", argv[0], argv[0], argv[0]
144
    );
145
  }
146

    
147

    
148
static double atod_kmg(const char *s) {
149
  double d;
150
  char *e;
151

    
152
  errno = 0;
153
  d = strtod(s, &e);
154
  if (errno) {
155
    fprintf(stderr, "Error parsing option: %s\n", s);
156
    exit(-1);
157
  }
158
  switch (*e) {
159
    case 'g':
160
    case 'G':
161
      d *= 1024;
162
    case 'm':
163
    case 'M':
164
      d *= 1024;
165
    case 'k':
166
    case 'K':
167
      d *= 1024;
168
    case 0:
169
      break;
170
    default:
171
      fprintf(stderr, "Error parsing option: %s\n", s);
172
      exit(-1);
173
  }
174

    
175
  return d;
176
}
177

    
178
static void cmdline_parse(int argc, char *argv[])
179
{
180
  int o;
181

    
182
  int option_index = 0;
183
  static struct option long_options[] = {
184
        {"chunk_log", no_argument, 0, 0},
185
        {"measure_start", required_argument, 0, 0},
186
        {"measure_every", required_argument, 0, 0},
187
        {"playout_limit", required_argument, 0, 0},
188
        {"randomize_start", required_argument, 0, 0},
189
        {"capacity_override", required_argument, 0, 0},
190
        {"desired_bw", required_argument, 0, 0},
191
        {"topo_mem", required_argument, 0, 0},
192
        {"topo_in", no_argument, 0, 0},
193
        {"topo_out", no_argument, 0, 0},
194
        {"topo_bidir", no_argument, 0, 0},
195
        {"topo_keep_best", no_argument, 0, 0},
196
        {"topo_add_best", no_argument, 0, 0},
197
        {0, 0, 0, 0}
198
  };
199

    
200
    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:",long_options, &option_index)) != -1) { //use this function to manage long options
201
    switch(o) {
202
      case 0: //for long options
203
        if( strcmp( "chunk_log", long_options[option_index].name ) == 0 ) { chunk_log = true; }
204
#ifndef MONL
205
        if( strcmp( "measure_start", long_options[option_index].name ) == 0 ) { tstartdiff.tv_sec = atoi(optarg); }
206
        if( strcmp( "measure_every", long_options[option_index].name ) == 0 ) { print_tdiff.tv_sec = atoi(optarg); }
207
#endif
208
        if( strcmp( "playout_limit", long_options[option_index].name ) == 0 ) { CB_SIZE_TIME = atoi(optarg); }
209
        if( strcmp( "randomize_start", long_options[option_index].name ) == 0 ) { randomize_start = atoi(optarg); }
210
        if( strcmp( "capacity_override", long_options[option_index].name ) == 0 ) { capacity_override = atod_kmg(optarg); }
211
        if( strcmp( "desired_bw", long_options[option_index].name ) == 0 ) { desired_bw = atod_kmg(optarg); }
212
        if( strcmp( "topo_mem", long_options[option_index].name ) == 0 ) { topo_mem = atof(optarg); }
213
        else if( strcmp( "topo_in", long_options[option_index].name ) == 0 ) { topo_in = true; topo_out = false; }
214
        else if( strcmp( "topo_out", long_options[option_index].name ) == 0 ) { topo_in = false; topo_out = true; }
215
        else if( strcmp( "topo_bidir", long_options[option_index].name ) == 0 ) { topo_in = true; topo_out = true; }
216
        else if( strcmp( "topo_keep_best", long_options[option_index].name ) == 0 ) { topo_keep_best = true; }
217
        else if( strcmp( "topo_add_best", long_options[option_index].name ) == 0 ) { topo_add_best = true; }
218
        break;
219
      case 'a':
220
        alpha_target = (double)atoi(optarg) / 100.0;
221
        break;
222
      case 'r':
223
        desired_rtt = (double)atoi(optarg) / 1000.0;
224
        break;
225
      case 'b':
226
        buff_size = atoi(optarg);
227
        break;
228
      case 'o':
229
        outbuff_size = atoi(optarg);
230
        break;
231
        case 'c':
232
        chunks_per_second = atoi(optarg);
233
        break;
234
      case 'm':
235
        multiply = atoi(optarg);
236
        break;
237
      case 'p':
238
        srv_port = atoi(optarg);
239
        break;
240
      case 'i':
241
        srv_ip = strdup(optarg);
242
        break;
243
      case 'P':
244
        port =  atoi(optarg);
245
        break;
246
      case 'I':
247
        my_iface = strdup(optarg);
248
        break;
249
      case 'f':
250
        fname = strdup(optarg);
251
        break;
252
      case 'F':
253
        output_config = strdup(optarg);
254
        break;
255
      case 'C':
256
        channel_set_name(optarg);
257
        break;
258
      case 'n':
259
        net_helper_config = strdup(optarg);
260
        break;
261
      case 'N':
262
        peername = strdup(optarg);
263
        break;
264
      case 'M':
265
        NEIGHBORHOOD_TARGET_SIZE = atoi(optarg);
266
        break;
267
      case 't':
268
        topo_config = strdup(optarg);
269
        break;
270
      case 'S':
271
        initial_id = atoi(optarg);
272
        break;
273
      case 's':
274
        start_id = atoi(optarg);
275
        break;
276
      case 'e':
277
        end_id = atoi(optarg);
278
        break;
279
      default:
280
        fprintf(stderr, "Error: unknown option %c\n", o);
281
        print_usage(argc, argv);
282

    
283
        exit(-1);
284
    }
285
  }
286

    
287
  if (!channel_get_name()) {
288
    channel_set_name("generic");
289
  }
290

    
291
  if (argc <= 1) {
292
    print_usage(argc, argv);
293
    fprintf(stderr, "Trying to start a source with default parameters, reading from %s\n", fname);
294
  }
295
}
296

    
297
const struct nodeID *get_my_addr(void)
298
{
299
  return my_sock;
300
}
301

    
302
static void init_rand(const char * s)
303
{
304
  long i, l, x;
305
  struct timeval t;
306

    
307
  gettimeofday(&t, NULL);
308

    
309
  x = 1;
310
  l = strlen(s);
311
  for (i = 0; i < l; i++) {
312
    x *= s[i];
313
  }
314

    
315
  srand(t.tv_usec * t.tv_sec * x);
316
}
317

    
318
static struct nodeID *init(void)
319
{
320
  int i;
321
  struct nodeID *myID;
322
  char *my_addr;
323

    
324
  if (my_iface) {
325
    my_addr = iface_addr(my_iface);
326
  } else {
327
    my_addr = default_ip_addr();
328
  }
329

    
330
  if (my_addr == NULL) {
331
    fprintf(stderr, "Cannot find network interface %s\n", my_iface);
332

    
333
    return NULL;
334
  }
335
  for (i=0;i<2;i++)
336
          bind_msg_type(msgTypes[i]);
337
  myID = net_helper_init(my_addr, port, net_helper_config);
338
  if (myID == NULL) {
339
    fprintf(stderr, "Error creating my socket (%s:%d)!\n", my_addr, port);
340
    free(my_addr);
341

    
342
    return NULL;
343
  }
344
  free(my_addr);
345
  fprintf(stderr, "My network ID is: %s\n", node_addr(myID));
346

    
347
  init_rand(node_addr(myID));
348

    
349
  topologyInit(myID, topo_config);
350

    
351
  return myID;
352
}
353

    
354
void leave(int sig) {
355
  fprintf(stderr, "Received signal %d, exiting!\n", sig);
356
  end_measures();
357
  exit(sig);
358
}
359

    
360
static void random_wait(int max) {
361
    struct timespec t;
362
    uint64_t ms;
363

    
364
    ms = (rand()/(RAND_MAX + 1.0)) * max;
365
    t.tv_sec = ms / 1000000;
366
    t.tv_nsec = (ms % 1000000) * 1000;
367
    nanosleep(&t, NULL);
368
}
369

    
370
int main(int argc, char *argv[])
371
{
372

    
373
  (void) signal(SIGTERM,leave);
374
  (void) signal(SIGINT,leave);
375

    
376
  cmdline_parse(argc, argv);
377

    
378
  my_sock = init();
379
  if (my_sock == NULL) {
380
    fprintf(stderr, "Cannot initialize streamer, exiting!\n");
381
    return -1;
382
  }
383
  if (srv_port != 0) {
384
    struct nodeID *srv;
385

    
386
    //random wait a bit before starting
387
    if (randomize_start) random_wait(randomize_start);
388

    
389
    output_init(outbuff_size, output_config);
390

    
391
    srv = create_node(srv_ip, srv_port);
392
    if (srv == NULL) {
393
      fprintf(stderr, "Cannot resolve remote address %s:%d\n", srv_ip, srv_port);
394

    
395
      return -1;
396
    }
397
    topoAddNeighbour(srv, NULL, 0);
398

    
399
    loop(my_sock, 1000000 / chunks_per_second, buff_size);
400
  } else {
401
    source_loop(fname, my_sock, period * 1000, multiply, buff_size);
402
  }
403
  return 0;
404
}
405

    
406
int am_i_source()
407
{
408
  return (srv_port == 0);
409
}
410

    
411
int get_cb_size()
412
{
413
  return buff_size;
414
}
415

    
416
int get_chunks_per_sec()
417
{
418
  return chunks_per_second;
419
}
420

    
421
//capacity in bits/s, or NAN if unknown
422
double get_capacity()
423
{
424
  return capacity_override;
425
}