Statistics
| Branch: | Revision:

streamers / chunk_signaling.c @ b947607c

History | View | Annotate | Download (5.18 KB)

1
/*
2
 *  Copyright (c) 2009 Alessandro Russo
3
 *  Copyright (c) 2009 Csaba Kiraly
4
 *
5
 *  This is free software; see gpl-3.0.txt
6
 *
7
 * Chunk Signaling API - Higher Abstraction
8
 *
9
 * The Chunk Signaling HA provides a set of primitives for chunks signaling negotiation with other peers, in order to collect information for the effective chunk exchange with other peers. <br>
10
 * This is a part of the Data Exchange Protocol which provides high level abstraction for chunks' negotiations, like requesting and proposing chunks.
11
 *
12
 */
13
#include <stdint.h>
14
#include <stdlib.h>
15
#include <stdio.h>
16
#include <sys/time.h>
17
#include <errno.h>
18
#include <assert.h>
19
#include <string.h>
20
#include "peer.h"
21
#include "peerset.h"
22
#include "chunkidset.h"
23
#include "trade_sig_la.h"
24
#include "chunk_signaling.h"
25
#include "net_helper.h"
26
#include <trade_sig_ha.h>
27

    
28
#include "streaming.h"
29
#include "topology.h"
30
#include "dbg.h"
31

    
32
static bool neigh_on_sign_recv = false;
33

    
34
void ack_received(const struct nodeID *fromid, struct chunkID_set *cset, int max_deliver, uint16_t trans_id) {
35
  struct peer *from = nodeid_to_peer(fromid,0);   //verify that we have really sent, 0 at least garantees that we've known the peer before
36
  dprintf("The peer %s acked our chunk %d chunks, max deliver %d, trans_id %d.\n", node_addr(fromid), chunkID_set_get_latest(cset), max_deliver, trans_id);
37

    
38
  if (from) {
39
    chunkID_set_clear(from->bmap,from->cb_size+5);        //TODO: some better solution might be needed to keep info about chunks we sent in flight.
40
    chunkID_set_union(from->bmap,cset);
41
    gettimeofday(&from->bmap_timestamp, NULL);
42
  }
43
}
44

    
45
void bmap_received(const struct nodeID *fromid, const struct nodeID *ownerid, struct chunkID_set *c_set, int cb_size, uint16_t trans_id) {
46
  struct peer *owner;
47
  if (nodeid_equal(fromid, ownerid)) {
48
    owner = nodeid_to_peer(ownerid, neigh_on_sign_recv);
49
  } else {
50
    dprintf("%s might be behind ",node_addr(ownerid));
51
    dprintf("NAT:%s\n",node_addr(fromid));
52
    owner = nodeid_to_peer(fromid, neigh_on_sign_recv);
53
  }
54
  
55
  if (owner) {        //now we have it almost sure
56
    chunkID_set_clear(owner->bmap,cb_size+5);        //TODO: some better solution might be needed to keep info about chunks we sent in flight.
57
    chunkID_set_union(owner->bmap,c_set);
58
    owner->cb_size = cb_size;
59
    gettimeofday(&owner->bmap_timestamp, NULL);
60
  }
61
}
62

    
63
void offer_received(const struct nodeID *fromid, struct chunkID_set *cset, int max_deliver, uint16_t trans_id) {
64
  struct chunkID_set *cset_acc;
65

    
66
  struct peer *from = nodeid_to_peer(fromid, neigh_on_sign_recv);
67
  dprintf("The peer %s offers %d chunks, max deliver %d.\n", node_addr(fromid), chunkID_set_size(cset), max_deliver);
68

    
69
  if (from) {
70
    //register these chunks in the buffermap. Warning: this should be changed when offers become selective.
71
    chunkID_set_clear(from->bmap,0);        //TODO: some better solution might be needed to keep info about chunks we sent in flight.
72
    chunkID_set_union(from->bmap,cset);
73
    gettimeofday(&from->bmap_timestamp, NULL);
74
  }
75

    
76
    //decide what to accept
77
    cset_acc = get_chunks_to_accept(fromid, cset, max_deliver, trans_id);
78

    
79
    //send accept message
80
    dprintf("\t accept %d chunks from peer %s, trans_id %d\n", chunkID_set_size(cset_acc), node_addr(fromid), trans_id);
81
    acceptChunks(fromid, cset_acc, trans_id);
82

    
83
    chunkID_set_free(cset_acc);
84
}
85

    
86
void accept_received(const struct nodeID *fromid, struct chunkID_set *cset, int max_deliver, uint16_t trans_id) {
87
  //TODO: verify if such a verification is needed
88
  //struct peer *from = nodeid_to_peer(fromid,0);   //verify that we have really offered, 0 at least garantees that we've known the peer before
89

    
90
  dprintf("The peer %s accepted our offer for %d chunks, max deliver %d.\n", node_addr(fromid), chunkID_set_size(cset), max_deliver);
91

    
92
  send_accepted_chunks(fromid, cset, max_deliver, trans_id);
93
}
94

    
95

    
96
 /**
97
 * Dispatcher for signaling messages.
98
 *
99
 * This method decodes the signaling messages, retrieving the set of chunk and the signaling
100
 * message, invoking the corresponding method.
101
 *
102
 * @param[in] buff buffer which contains the signaling message
103
 * @param[in] buff_len length of the buffer
104
 * @return 0 on success, <0 on error
105
 */
106

    
107
int sigParseData(const struct nodeID *fromid, uint8_t *buff, int buff_len) {
108
    struct chunkID_set *c_set;
109
    struct nodeID *ownerid;
110
    enum signaling_type sig_type;
111
    int max_deliver = 0;
112
    uint16_t trans_id = 0;
113
    int ret = 1;
114
    dprintf("Decoding signaling message...\n");
115

    
116
    ret = parseSignaling(buff + 1, buff_len-1, &ownerid, &c_set, &max_deliver, &trans_id, &sig_type);
117
    if (ret < 0) {
118
      fprintf(stdout, "ERROR parsing signaling message\n");
119
      return -1;
120
    }
121
    switch (sig_type) {
122
        case sig_send_buffermap:
123
          bmap_received(fromid, ownerid, c_set, max_deliver, trans_id); //FIXME: cb_size has gone from signaling
124
          break;
125
        case sig_offer:
126
          offer_received(fromid, c_set, max_deliver, trans_id);
127
          break;
128
        case sig_accept:
129
          accept_received(fromid, c_set, chunkID_set_size(c_set), trans_id);
130
          break;
131
            case sig_ack:
132
              ack_received(fromid, c_set, chunkID_set_size(c_set), trans_id);
133
          break;
134
        default:
135
          ret = -1;
136
    }
137
    chunkID_set_free(c_set);
138
    nodeid_free(ownerid);
139
    return ret;
140
}