Statistics
| Branch: | Tag: | Revision:

sssimulator / td.c @ master

History | View | Annotate | Download (4.29 KB)

1
#include <stdlib.h>
2
#include <stdio.h>
3
#include <string.h>
4
#include <stdint.h>
5

    
6
#include "core.h"
7
#include "sched.h"
8
#include "stats.h"
9
#include "td.h"
10

    
11
FILE *resfile;
12
FILE *delayfile;
13
FILE *statusfile;
14
static int ts;
15
static struct peer *peers;
16
static struct chunk *chunks;
17
static int num_peers;
18
extern float convergence_precision;
19
extern uint32_t analysis_window;
20
extern float netload;
21

    
22
int comp(int chunk)
23
{
24
  return (chunks[chunk].received == num_peers);
25
}
26

    
27

    
28
///==========================================================
29
///     the network model :)
30
///==========================================================
31
//chunks propagate through receiver's recv_chunks array
32
static void chunk_send(int k, struct peer *p, int dl)
33
{
34
#if 0
35
  if (p->chunks[k]) {
36
    fprintf(stderr, "Error: sending chunk %d to %d which already has it!\n", k, p->id);
37
    exit(-1);
38
  }
39
#endif
40
  if (p->recv_chunks[win_idx(p, k)]) {
41
    fprintf(stderr, "Error: sending chunk %d (idx: %d) to %d which already receives it (%d)!\n", k, win_idx(p, k), p->id, p->recv_chunks[k]->chunk);
42
    exit(-1);
43
  }
44
  if (p->chunks[win_idx(p, k)]) {
45
    p->recv_chunks[win_idx(p, k)] = p->chunks[win_idx(p, k)];
46
    p->chunks[win_idx(p, k)] = NULL;
47
  } else {
48
    p->recv_chunks[win_idx(p, k)] = malloc(sizeof(struct chunk_instance));
49
  }
50
  p->recv_chunks[win_idx(p, k)]->dl = dl;
51
  p->recv_chunks[win_idx(p, k)]->chunk = k;
52

    
53
  //CACHE latest_not_needed
54
  p->latest_not_needed = max(k,p->latest_not_needed);
55
  chunks[k].received += 1;
56
}
57

    
58
static int receive(int t)
59
{
60
  int p, c;
61
  int cnt;
62

    
63
  cnt = 0;
64
  for (p = 0; p < num_peers; p++) {
65
    for (c = 0; c < peers[p].buf_size; c++) {
66
      int idx = win_idx(&peers[p], c /* + peers[p].buf_size*/);
67

    
68
      if (peers[p].recv_chunks[idx]) {
69
        cnt++;
70
        peers[p].chunks[idx] = peers[p].recv_chunks[idx];
71
        peers[p].chunks[idx]->completed = t;
72
        peers[p].recv_chunks[idx] = NULL;
73
        chunks[peers[p].chunks[idx]->chunk].avg_delay += t - peers[p].chunks[idx]->chunk;
74
        chunks[peers[p].chunks[idx]->chunk].max_delay = max(chunks[peers[p].chunks[idx]->chunk].max_delay, t - peers[p].chunks[idx]->chunk);
75
        if (delayfile) {
76
            fprintf(delayfile, "%d,%d,%d,%d\n", t, p, peers[p].chunks[idx]->chunk,  t - peers[p].chunks[idx]->chunk);
77
            fflush(delayfile);
78
        }
79
        if (chunks[peers[p].chunks[idx]->chunk].received == num_peers) {
80
          if (resfile) {
81
            fprintf(resfile, "Chunk %d received at %d in %d\n", peers[p].chunks[idx]->chunk, t, t - peers[p].chunks[idx]->chunk);
82
            fflush(resfile);
83
          }
84
        }
85
      }
86
    }
87
  }
88
#if 0
89
  for (c = 0; c < num_chunks; j++) {
90
    if (completed[c] == 0) {
91
      completed[c] = t;
92
      for (p = 0; p < num_peers; p++) {
93
        if (peers[p].chunks[c] == NULL) {
94
          completed[c] = 0;
95
        }
96
      }
97
    }
98
  }
99
#endif
100

    
101
  return cnt;
102
}
103

    
104
struct chunk *td_loop(struct peer *p, int np, int num_chunks, int server_period)
105
{
106
  int done, t, offers;
107
  struct stats * stat = NULL;
108

    
109
  offers = ((int)netload) > 0 ? ((int)netload) : 1;
110

    
111
  chunks = malloc(num_chunks * sizeof(struct chunk));
112
  if (chunks == NULL) {
113
    perror("MAlloc");
114

    
115
    return NULL;
116
  }
117
  memset(chunks, 0, num_chunks * sizeof(struct chunk));
118
  num_peers = np;
119
  peers = p;
120
  ts = server_period;
121

    
122
  done = 0; t = 0;  // t is the time in chunk time (discrete)
123
  while (!done) {
124
    int i, j;
125
    struct peer *target;
126

    
127
    // loop stuff
128
    done = ((t != 0) && (receive(t) == 0));
129
    status_print(statusfile, peers, num_peers, chunks, num_chunks, t);
130
    if (trace) fprintf(trace, "Time %d:\n", t);
131

    
132
    // source seeding
133
    source_send(t, &target);
134
    if (target != NULL) {
135
      chunk_send(t, target, t + ts);
136
    }
137

    
138
    // peer offering
139
    for (i = 0; i < num_peers; i++) {
140
      int chunk;
141

    
142
      for (j = 0; j < offers; j++)
143
      {
144
              peer_send(&peers[i], t, &chunk, &target);
145
              if (target != NULL) {
146
                peers[i].chunks[win_idx(&peers[i], chunk)]->dl += ts;
147
                chunk_send(chunk, target, peers[i].chunks[win_idx(&peers[i], chunk)]->dl);
148
              }
149
      }
150
    }
151
    t++;
152
    if (convergence_precision && chunk_stats_converge(&stat, chunks, num_chunks, peers, num_peers, t - 1, analysis_window))
153
            done = 1;
154
  }
155
  chunk_stats_converge(&stat, chunks, num_chunks, peers, num_peers, t - 1, analysis_window);
156
  chunk_stats_print(stat, CSV_STATS);
157
  if (stat) free(stat);
158

    
159
  return chunks;
160
}