Statistics
| Branch: | Revision:

grapes / src / TopologyManager / cloudcast_proto.c @ 36dc16e0

History | View | Annotate | Download (4.26 KB)

1
/*
2
 *  Copyright (c) 2010 Andrea Zito
3
 *
4
 *  This is free software; see lgpl-2.1.txt
5
 */
6

    
7
#include <stdint.h>
8
#include <stdlib.h>
9
#include <stdio.h>
10

    
11
#include "net_helper.h"
12
#include "topocache.h"
13
#include "proto.h"
14
#include "topo_proto.h"
15
#include "cloudcast_proto.h"
16
#include "grapes_msg_types.h"
17

    
18
// TODO: This should not be duplicated here. should inherit from topo_proto.c
19
#define MAX_MSG_SIZE 1500
20

    
21
#define CLOUD_VIEW_KEY "view"
22

    
23
struct topo_header cloud_header = {MSG_TYPE_TOPOLOGY, CLOUDCAST_CLOUD};
24

    
25
struct cloudcast_proto_context {
26
  struct peer_cache* myEntry;
27
  struct topo_context *topo_context;
28
  struct cloud_helper_context *cloud_context;
29
};
30

    
31
struct cloudcast_proto_context* cloudcast_proto_init(struct nodeID *s, void *meta, int meta_size)
32
{
33
  struct cloudcast_proto_context *con;
34
  con = malloc(sizeof(struct cloudcast_proto_context));
35

    
36
  if (!con) return NULL;
37

    
38
  con->topo_context = topo_proto_init(s, meta, meta_size);
39
  if (!con->topo_context){
40
    free(con);
41
    return NULL;
42
  }
43

    
44
  con->cloud_context = get_cloud_helper_for(s);
45
  if (!con->cloud_context) {
46
    free(con);
47
    return NULL;
48
  }
49

    
50
  con->myEntry = cache_init(1, meta_size, 0);
51
  cache_add(con->myEntry, s, meta, meta_size);
52

    
53
  return con;
54
}
55

    
56
struct nodeID* cloudcast_get_cloud_node(struct cloudcast_proto_context *con)
57
{
58
  return get_cloud_node(con->cloud_context, 0);
59
}
60

    
61
struct nodeID** cloudcast_get_cloud_nodes(struct cloudcast_proto_context *con, uint8_t number)
62
{
63
  int i;
64
  struct nodeID* *cloud_nodes;
65

    
66
  if (number == 0) return NULL;
67
  cloud_nodes = calloc(number, sizeof(struct nodeID*));
68

    
69
  for (i=0; i< number; i++)
70
    cloud_nodes[i] = get_cloud_node(con->cloud_context, i);
71
  return cloud_nodes;
72
}
73

    
74
int cloudcast_is_cloud_node(struct cloudcast_proto_context *con, struct nodeID* node)
75
{
76
  return is_cloud_node(con->cloud_context, node);
77
}
78

    
79
int cloudcast_reply_peer(struct cloudcast_proto_context *context, const struct peer_cache *c, struct peer_cache *local_cache)
80
{
81
  return topo_reply(context->topo_context, c, local_cache, MSG_TYPE_TOPOLOGY, CLOUDCAST_REPLY, 0, 0);
82
}
83

    
84
int cloudcast_query_peer(struct cloudcast_proto_context *context, struct peer_cache *sent_cache, struct nodeID *dst)
85
{
86
  return topo_query_peer(context->topo_context, sent_cache, dst, MSG_TYPE_TOPOLOGY, CLOUDCAST_QUERY, 0);
87
}
88

    
89

    
90
int cloudcast_payload_fill(struct cloudcast_proto_context *context, uint8_t *payload, int size, struct peer_cache *c, int max_peers, int include_me)
91
{
92
  int i;
93
  uint8_t *p = payload;
94

    
95
  if (!max_peers) max_peers = MAX_MSG_SIZE; // just to be sure to dump the whole cache...
96
  p += cache_header_dump(p, c, include_me);
97
  if (include_me) {
98
    p += entry_dump(p, context->myEntry, 0, size - (p - payload));
99
    max_peers--;
100
  }
101
  for (i = 0; nodeid(c, i) && max_peers; i++) {
102
    if (!is_cloud_node(context->cloud_context, nodeid(c, i))) {
103
      int res;
104
      res = entry_dump(p, c, i, size - (p - payload));
105
      if (res < 0) {
106
        fprintf(stderr, "too many entries!\n");
107
        return -1;
108
      }
109
      p += res;
110
      --max_peers;
111
    }
112
  }
113

    
114
  return p - payload;
115
}
116

    
117

    
118
int cloudcast_reply_cloud(struct cloudcast_proto_context *context, struct peer_cache *cloud_cache)
119
{
120
  uint8_t headerless_pkt[MAX_MSG_SIZE - sizeof(cloud_header)];
121
  int len, res;
122

    
123
  len = cloudcast_payload_fill(context, headerless_pkt, MAX_MSG_SIZE - sizeof(cloud_header), cloud_cache, 0, 1);
124

    
125
  if (len > 0)
126
    res = put_on_cloud(context->cloud_context, CLOUD_VIEW_KEY, headerless_pkt, len);
127
  else
128
    res = 0;
129
  return res;
130
}
131

    
132
int cloudcast_query_cloud(struct cloudcast_proto_context *context)
133
{
134
  return get_from_cloud(context->cloud_context, CLOUD_VIEW_KEY, (uint8_t *)&cloud_header, sizeof(cloud_header));
135
}
136

    
137
struct peer_cache * cloudcast_cloud_default_reply(struct peer_cache *template, struct nodeID *cloud_entry)
138
{
139
  int size;
140
  struct peer_cache *cloud_reply;
141
  get_metadata(template, &size);
142
  cloud_reply = cache_init(1, size, 0);
143
  cache_add(cloud_reply, cloud_entry, NULL, 0);
144
  return cloud_reply;
145
}
146

    
147
time_t cloudcast_timestamp_cloud(struct cloudcast_proto_context *context)
148
{
149
  return timestamp_cloud(context->cloud_context);
150
}
151

    
152
int cloudcast_proto_change_metadata(struct cloudcast_proto_context *context, void *metadata, int metadata_size)
153
{
154
  if (topo_proto_metadata_update(context->topo_context, metadata, metadata_size) <= 0) {
155
    return -1;
156
  }
157

    
158
  return 1;
159
}