Statistics
| Branch: | Revision:

streamers / streamer.c @ 4ca3a6bd

History | View | Annotate | Download (8.76 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 <grapes_msg_types.h>
16
#include <net_helper.h>
17

    
18
#ifdef WIN32
19
#include <winsock2.h>
20
#endif
21

    
22
#include "net_helpers.h"
23
#include "loop.h"
24
#include "output.h"
25
#include "channel.h"
26
#include "topology.h"
27
#include "measures.h"
28
#include "streamer.h"
29

    
30
static struct nodeID *my_sock;
31

    
32
const char *peername = NULL;
33

    
34
static const char *my_iface = NULL;
35
static int port = 6666;
36
static int srv_port;
37
static const char *srv_ip = "";
38
static int period = 40;
39
static int chunks_per_second = 25;
40
#ifdef HTTPIO
41
//input-http.c needs this in order to accomplish the -m multiple send_chunk()
42
int multiply = 3;
43
#else
44
static int multiply = 3;
45
#endif
46
static int buff_size = 50;
47
static int outbuff_size = 50;
48
static const char *fname = "/dev/stdin";
49
static const char *output_config;
50
static bool loop_input = false;
51
static const char *net_helper_config = "";
52
static const char *topo_config = "";
53
unsigned char msgTypes[] = {MSG_TYPE_CHUNK,MSG_TYPE_SIGNALLING};
54
bool chunk_log = false;
55
static int randomize_start = 0;
56

    
57
extern int NEIGHBORHOOD_TARGET_SIZE;
58
extern uint64_t CB_SIZE_TIME;
59
extern double desired_rtt;
60
extern double alpha_target;
61

    
62
#ifndef MONL
63
extern struct timeval print_tdiff;
64
extern struct timeval tstartdiff;
65
#endif
66

    
67
static void print_usage(int argc, char *argv[])
68
{
69
  fprintf (stderr,
70
    "Usage:%s [options]\n"
71
    "\n"
72
    "Peer options\n"
73
    "\t[-p port]: port of the remote peer to connect at during bootstrap.\n"
74
    "\t           Usually it is the source peer port.\n"
75
    "\t[-i IP]: IP address of the remote peer to connect at during bootstrap.\n"
76
    "\t         Usually it is the source peer IP\n"
77
    "\t[-C name]: set the channel name to use on the repository.\n"
78
    "\t           All peers should use the same channel name.\n"
79
    "\n"
80
    "\t[-b size]: set the peer Chunk Buffer size.\n"
81
    "\t           This is also the chunk trading window size.\n"
82
    "\t[-o size]: set the Output Buffer size.\n"
83
    "\t[-c chunks]: set the number of chunks a peer can send per seconds.\n"
84
    "\t             it controls the upload capacity of peer as well.\n"
85
    "\t[-M peers]: neighbourhood target size.\n"
86
    "\t[-t config]: topology config.\n"
87
    "\t[-P port]: local UDP port to be used by the peer.\n"
88
    "\t[-I iface]: local netwok interface to be used by the peer.\n"
89
    "\t         Useful if the host has several interfaces/addresses.\n"
90
    "\t[-N name]: set the name of the peer.\n"
91
    "\t         This name will be used when publishing in the repository.\n"
92
    "\t[-n options]: pass configuration options to the net-helper\n"
93
    "\t[--chunk_log]: print a chunk level log on stderr\n"
94
    "\t[-F config]: configure the output module\n"
95
    "\t[-a alpha]: set the topology alpha value (from 0 to 100)\n"
96
    "\t[-t rtt]: set the RTT threshold (in ms) for desired neighbours\n"
97
    "\n"
98
    "Special Source Peer options\n"
99
    "\t[-m chunks]: set the number of copies the source injects in the overlay.\n"
100
    "\t[-f filename]: name of the video stream file to transmit.\n"
101
    "\t[-l]: loop the video stream.\n"
102
    "\n"
103
    "Special options\n"
104
    "\t[--randomize_start ms]: random wait before starting in to ms millisecs.\n"
105
    "\n"
106
    "NOTE: by deafult the peer will dump the received video on STDOUT in raw format\n"
107
    "      it can be played by your favourite player simply using a pipe\n"
108
    "      e.g., | cvlc /dev/stdin\n"
109
    "\n"
110
    "Examples:\n"
111
    "\n"
112
    "Start a source peer on port 6600:\n"
113
    "\n"
114
    "%s -l -f foreman.mpg -P 6600\n"
115
    "\n"
116
    "Start a peer connecting to the previous source, and using videolan as player:\n"
117
    "\n"
118
    "%s -i <sourceIP> -p <sourcePort> | cvlc /dev/stdin\n"
119
    "=======================================================\n", argv[0], argv[0], argv[0]
120
    );
121
  }
122

    
123

    
124

    
125
static void cmdline_parse(int argc, char *argv[])
126
{
127
  int o;
128

    
129
  int option_index = 0;
130
  static struct option long_options[] = {
131
        {"chunk_log", no_argument, 0, 0},
132
        {"measure_start", required_argument, 0, 0},
133
        {"measure_every", required_argument, 0, 0},
134
        {"playout_limit", required_argument, 0, 0},
135
        {"randomize_start", required_argument, 0, 0},
136
        {0, 0, 0, 0}
137
  };
138

    
139
    while ((o = getopt_long (argc, argv, "r:a:b:o:c:p:i:P:I:f:F:m:lC:N:n:M:t:",long_options, &option_index)) != -1) { //use this function to manage long options
140
    switch(o) {
141
      case 0: //for long options
142
        if( strcmp( "chunk_log", long_options[option_index].name ) == 0 ) { chunk_log = true; }
143
#ifndef MONL
144
        if( strcmp( "measure_start", long_options[option_index].name ) == 0 ) { tstartdiff.tv_sec = atoi(optarg); }
145
        if( strcmp( "measure_every", long_options[option_index].name ) == 0 ) { print_tdiff.tv_sec = atoi(optarg); }
146
#endif
147
        if( strcmp( "playout_limit", long_options[option_index].name ) == 0 ) { CB_SIZE_TIME = atoi(optarg); }
148
        if( strcmp( "randomize_start", long_options[option_index].name ) == 0 ) { randomize_start = atoi(optarg); }
149
        break;
150
      case 'a':
151
        alpha_target = (double)atoi(optarg) / 100.0;
152
        break;
153
      case 'r':
154
        desired_rtt = (double)atoi(optarg) / 1000.0;
155
        break;
156
      case 'b':
157
        buff_size = atoi(optarg);
158
        break;
159
      case 'o':
160
        outbuff_size = atoi(optarg);
161
        break;
162
        case 'c':
163
        chunks_per_second = atoi(optarg);
164
        break;
165
      case 'm':
166
        multiply = atoi(optarg);
167
        break;
168
      case 'p':
169
        srv_port = atoi(optarg);
170
        break;
171
      case 'i':
172
        srv_ip = strdup(optarg);
173
        break;
174
      case 'P':
175
        port =  atoi(optarg);
176
        break;
177
      case 'I':
178
        my_iface = strdup(optarg);
179
        break;
180
      case 'f':
181
        fname = strdup(optarg);
182
        break;
183
      case 'F':
184
        output_config = strdup(optarg);
185
        break;
186
      case 'l':
187
        loop_input = true;
188
        break;
189
      case 'C':
190
        channel_set_name(optarg);
191
        break;
192
      case 'n':
193
        net_helper_config = strdup(optarg);
194
        break;
195
      case 'N':
196
        peername = strdup(optarg);
197
        break;
198
      case 'M':
199
        NEIGHBORHOOD_TARGET_SIZE = atoi(optarg);
200
        break;
201
      case 't':
202
        topo_config = strdup(optarg);
203
        break;
204
      default:
205
        fprintf(stderr, "Error: unknown option %c\n", o);
206
        print_usage(argc, argv);
207

    
208
        exit(-1);
209
    }
210
  }
211

    
212
  if (!channel_get_name()) {
213
    channel_set_name("generic");
214
  }
215

    
216
  if (argc <= 1) {
217
    print_usage(argc, argv);
218
    fprintf(stderr, "Trying to start a source with default parameters, reading from %s\n", fname);
219
  }
220
}
221

    
222
const struct nodeID *get_my_addr(void)
223
{
224
  return my_sock;
225
}
226

    
227
static void init_rand(const char * s)
228
{
229
  long i, l, x;
230
  struct timeval t;
231

    
232
  gettimeofday(&t, NULL);
233

    
234
  x = 1;
235
  l = strlen(s);
236
  for (i = 0; i < l; i++) {
237
    x *= s[i];
238
  }
239

    
240
  srand(t.tv_usec * t.tv_sec * x);
241
}
242

    
243
static struct nodeID *init(void)
244
{
245
  int i;
246
  struct nodeID *myID;
247
  char *my_addr;
248

    
249
  if (my_iface) {
250
    my_addr = iface_addr(my_iface);
251
  } else {
252
    my_addr = default_ip_addr();
253
  }
254

    
255
  if (my_addr == NULL) {
256
    fprintf(stderr, "Cannot find network interface %s\n", my_iface);
257

    
258
    return NULL;
259
  }
260
  for (i=0;i<2;i++)
261
          bind_msg_type(msgTypes[i]);
262
  myID = net_helper_init(my_addr, port, net_helper_config);
263
  if (myID == NULL) {
264
    fprintf(stderr, "Error creating my socket (%s:%d)!\n", my_addr, port);
265
    free(my_addr);
266

    
267
    return NULL;
268
  }
269
  free(my_addr);
270
  fprintf(stderr, "My network ID is: %s\n", node_addr(myID));
271

    
272
  init_rand(node_addr(myID));
273

    
274
  topologyInit(myID, topo_config);
275

    
276
  return myID;
277
}
278

    
279
void leave(int sig) {
280
  end_measures();
281
  exit(sig);
282
}
283

    
284
static void random_wait(int max) {
285
    struct timespec t;
286
    uint64_t ms;
287

    
288
    ms = (rand()/(RAND_MAX + 1.0)) * max;
289
    t.tv_sec = ms / 1000000;
290
    t.tv_nsec = (ms % 1000000) * 1000;
291
    nanosleep(&t, NULL);
292
}
293

    
294
int main(int argc, char *argv[])
295
{
296

    
297
  (void) signal(SIGTERM,leave);
298

    
299
  cmdline_parse(argc, argv);
300

    
301
  my_sock = init();
302
  if (my_sock == NULL) {
303
    fprintf(stderr, "Cannot initialize streamer, exiting!\n");
304
    return -1;
305
  }
306
  if (srv_port != 0) {
307
    struct nodeID *srv;
308

    
309
    //random wait a bit before starting
310
    if (randomize_start) random_wait(randomize_start);
311

    
312
    output_init(outbuff_size, output_config);
313

    
314
    srv = create_node(srv_ip, srv_port);
315
    if (srv == NULL) {
316
      fprintf(stderr, "Cannot resolve remote address %s:%d\n", srv_ip, srv_port);
317

    
318
      return -1;
319
    }
320
    topoAddNeighbour(srv, NULL, 0);
321

    
322
    loop(my_sock, 1000000 / chunks_per_second, buff_size);
323
  } else {
324
    source_loop(fname, my_sock, period * 1000, multiply, loop_input);
325
  }
326
  return 0;
327
}
328

    
329
int am_i_source()
330
{
331
  return (srv_port == 0);
332
}
333

    
334
int get_cb_size()
335
{
336
  return buff_size;
337
}