drl/drl_state.h

   1 /* See the DRL-LICENSE file for this file's software license. */
   2
   3 #ifndef _DRL_STATE_
   4 #define _DRL_STATE_
   5
   6 #define _XOPEN_SOURCE 600
   7
   8 /* FILE */
   9 #include <stdio.h>
  10
  11 /* uint32_t */
  12 #include <sys/types.h>
  13 #include <inttypes.h>
  14
  15 /* in_addr_t, in_port_t */
  16 #include <arpa/inet.h>
  17
  18 /* pthread functions. */
  19 #include <pthread.h>
  20
  21 /* fd_set */
  22 #include <sys/select.h>
  23
  24 /* Hash map types/functions. */
  25 #ifdef STANDALONE
  26 #include "../util.h"
  27 #else
  28 #include "util.h"
  29 #endif
  30
  31 #define MAX_IDENTS (1024)
  32 #define MAX_LIMITERS (128)
  33 #define TRUE (1)
  34 #define FALSE (0)
  35
  36 enum transports { UDP = 0, TCP = 1 };
  37 enum view_confidences { IN = 0, NOTIN = 1, UNSURE = 2 };
  38 enum reachabilities { REACHABLE = 0, SUSPECT = 1, UNREACHABLE = 2 };
  39
  40 typedef struct gossipval {
  41     /* Fields that don't change. */
  42     int gossip_branch;
  43     enum memberships membership;
  44     enum failure_behaviors failure_behavior;
  45
  46     /* Fields that change only on restart. */
  47     int32_t view;
  48
  49     /* Fields that change frequently. */
  50     double value;
  51     double weight;
  52 } gossip_t;
  53
  54 typedef struct out_neighbor {
  55     uint32_t next_seqno;
  56     uint32_t first_seqno;
  57     double saved_value;
  58     double saved_weight;
  59 } out_neighbor_t;
  60
  61 typedef struct in_neighbor {
  62     uint32_t seen_seqno;
  63     double saved_value;
  64     double saved_weight;
  65 } in_neighbor_t;
  66
  67 typedef struct remote_node {
  68     in_addr_t addr;
  69     in_port_t port;
  70 } remote_node_t;
  71
  72 //TODO: Clean this up
  73 typedef struct remote_limiter {
  74     /** The last known value at the remote limiter. */
  75     double rate;
  76
  77     in_neighbor_t incoming;
  78     out_neighbor_t outgoing;
  79
  80     /* Socket to contact this remote limiter, if using TCP. */
  81     //int socket;
  82
  83     /** IP address of the remote limiter, in network byte order. */
  84     in_addr_t addr;
  85     in_port_t port;
  86
  87     /** Keeps track of the number of messages we have sent to this peer without
  88      * having heard from them. */
  89     int awol;
  90
  91     /** Whether or not we think this peer is reachable. */
  92     enum reachabilities reachability;
  93
  94     /**Count of the rounds since doubt has risen and count of friends which
  95      * suspect this node to be awol or alive*/
  96     int count_rounds;
  97     int count_awol;
  98     int count_alive;
  99
 100     uint32_t incarnation;
 101
 102     int32_t view;
 103     enum view_confidences view_confidence;
 104 } remote_limiter_t;
 105
 106 //TODO: Reduce the size of this?
 107 typedef struct message {
 108     uint32_t magic;
 109     uint32_t ident_id;
 110     double value;
 111     double weight;
 112     uint32_t seqno;
 113     uint32_t min_seqno;
 114     uint16_t type;
 115
 116     /** tell ping target the address of node which requested ping */
 117     in_addr_t ping_source;
 118     in_port_t ping_port;
 119     /** friend needs to be told the address of node suspected to be down */
 120     in_addr_t check_target;
 121     in_port_t check_port;
 122     /** friend responds with ALIVE / AWOL */
 123     uint32_t checkack_value;
 124     /*Whether the message has an update piggy backed onto it*/
 125     uint32_t update_present; // TRUE or FALSE
 126     /*Node is reachable or not*/
 127     uint32_t reachability;
 128     /*Incarnation number of the node whose update
 129      * is being sent piggy backed on the message*/
 130     uint32_t incarnation;
 131     /*Address of the node whose update is being sent*/
 132     remote_node_t node;
 133
 134     uint32_t view;
 135 } message_t;
 136
 137 typedef struct comm {
 138     /** Communication policy. (COMM_MESH, COMM_GOSSIP) */
 139     enum commfabrics comm_fabric;
 140
 141     /** Transport protocol. */
 142     enum transports transport_proto;
 143
 144     /** Current local value. */
 145     double local_rate;
 146
 147     /** Previous local value. */
 148     double last_local_rate;
 149
 150     /** The number of remote nodes in the identity */
 151     uint32_t remote_node_count;
 152
 153     remote_node_t *remote_nodes;
 154
 155     /** Array containing all known remote limiters in this identity.
 156      * Contains the same information as the remote_node_map. */
 157     remote_limiter_t *remote_limiters;
 158
 159     /** Hash map containing all the remote limiters in this identity.
 160      * Indexed by the remote_node_t used to create the remote_limiter_t.
 161      * Maps to the appropriate element of the remote_limiters array. */
 162     map_handle remote_node_map;
 163
 164     /** A mutex to protect the comm structure. */
 165     pthread_mutex_t lock;
 166
 167     /** Gossip values for our local identity */
 168     gossip_t gossip;
 169
 170     /** Function pointer to send function. */
 171     int (*send_function)(struct comm *comm, uint32_t id, int sock);
 172
 173     /** Function pointer to recv function for group membership. When a message
 174      * is received, it is proccessed normally and then handed to this function
 175      * in case additional processing is necessary for group membership. */
 176     int (*recv_function)(struct comm *comm, uint32_t id, int sock, remote_limiter_t *remote, message_t *msg);
 177
 178     /** Function to restart the communication protocol. */
 179     void (*restart_function)(struct comm *comm, int32_t view_number);
 180
 181     /** Flag indicating whether or not we are "connected" to the group
 182      * membership service.  This can only be false for membership schemes that
 183      * require a persistent connection (Zookeeper). */
 184     int connected;
 185
 186     /** Array of integers specifiying which nodes have been selected for
 187      * message transmissions during the current round. */
 188     int *selected;
 189
 190     /** Array of indicies into remote_limiters.  Used to keep a shuffled
 191      * ordering for future gossip targets. */
 192     int *indices;
 193
 194     /** The next index to use for target peer selection.  The indicies are
 195      * re-shuffled when this reaches remote_node_count. */
 196     uint32_t shuffle_index;
 197
 198     void *membership_state;
 199
 200 #if 0
 201     /** Thread for handling incoming TCP data. */
 202     pthread_t tcp_recv_thread;
 203
 204     /** Descriptor set for reading TCP messages */
 205     fd_set fds;
 206 #endif
 207 } comm_t;
 208
 209 /**
 210  * Fills in the communication structure of an identity.
 211  *
 212  * @param comm The communication structure to be created/populated.
 213  *
 214  * @param config The configuration options for the identity.
 215  *
 216  * @param nodes An array of remote nodes belonging to this identity.
 217  *
 218  * @returns 0 on success, ENOMEM if memory cannot be allocated.
 219  */
 220 int new_comm(comm_t *comm, ident_config *config, remote_node_t *nodes);
 221
 222 /**
 223  * Frees the memory associated with an identity's communication structure.
 224  *
 225  * @param comm The communication structure to free.
 226  */
 227 void free_comm(comm_t *comm);
 228
 229 /**
 230  * Calculates and reads the current aggregate value for an identity.
 231  * This value includes the locally observed value.
 232  *
 233  * @returns 0 on success, EINVAL on error.
 234  */
 235 int read_comm(double *aggregate, uint32_t *effective_global, comm_t *comm, uint32_t global_limit);
 236
 237 /**
 238  * Updates the locally observed value of an identity.
 239  *
 240  * @param comm The comm structure of the identity to update.
 241  *
 242  * @param value The new locally observed value.
 243  *
 244  * @returns 0 on success, EINVAL on error.
 245  */
 246 int write_local_value(comm_t *comm, const double value);
 247
 248 /**
 249  * Sends the local state information to one or more peer limiters in the same
 250  * identity.  If the identity is configured as a mesh, it will send to all
 251  * peers.  If the identity is configured using gossip, it will send to the
 252  * number of peers specified by the gossip_branch field of the comm_config_t
 253  * that was used to configure the identity.
 254  *
 255  * @param comm The communication structure of the identity whose value should
 256  * be propagated.
 257  *
 258  * @param id The unique id of the identity that is sending.
 259  *
 260  * @returns 0 on success, ENOMEM if there was not enough memory, or possibly
 261  * other E values as a result of socket-related results.
 262  */
 263 int send_update(comm_t *comm, uint32_t id);
 264
 265 /**
 266  * Thread that is responsible for receiving data from other limiters.
 267  *
 268  * @param limiter The limiter_t that is to be doing the receiving.
 269  */
 270 void *limiter_receive_thread(void *unused);
 271
 272 #if 0
 273 typedef struct hello_message {
 274     uint32_t magic;
 275     uint32_t ident_id;
 276     uint16_t port;
 277 } hello_t;
 278
 279 struct recv_thread_args {
 280     comm_ident_t *ident;
 281     pthread_rwlock_t *lock;
 282     uint16_t port;
 283 };
 284
 285 /**
 286  * Initializes the global limiter.
 287  *
 288  * @param ipaddr The IP address on which the limiter should listen.
 289  * INADDR_ANY will suffice.  Should be specified in network byte order.
 290  *
 291  * @param port The port on which the limiter should listen. Should be specified
 292  * in network byte order.
 293  */
 294 void init_limiter(const in_addr_t ipaddr, const in_port_t port);
 295
 296 /**
 297  * Deallocates the entire global limiter.
 298  */
 299 void destroy_limiter();
 300 #endif
 301
 302 #endif  /* _DRL_STATE_ */