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
  34 #define REMOTE_AWOL_THRESHOLD (5)
  35
  36 enum transports { UDP, TCP };
  37
  38 typedef struct gossipval {
  39     int gossip_branch;
  40     double last_nonzero;
  41     double value;
  42     double weight;
  43 } gossip_t;
  44
  45 typedef struct out_neighbor {
  46     uint32_t next_seqno;
  47     uint32_t first_seqno;
  48     double saved_value;
  49     double saved_weight;
  50 } out_neighbor_t;
  51
  52 typedef struct in_neighbor {
  53     uint32_t seen_seqno;
  54     double saved_value;
  55     double saved_weight;
  56 } in_neighbor_t;
  57
  58 typedef struct remote_node {
  59     in_addr_t addr;
  60     in_port_t port;
  61 } remote_node_t;
  62
  63 typedef struct remote_limiter {
  64     /** The last known value at the remote limiter. */
  65     double rate;
  66
  67     in_neighbor_t incoming;
  68     out_neighbor_t outgoing;
  69
  70     /* Socket to contact this remote limiter, if using TCP. */
  71     int socket;
  72
  73     in_addr_t addr;
  74     in_port_t port;
  75
  76     /** Flag to keep track of situations in which we read from this node's
  77      * value more than once before receiving an update from it.  We use this
  78      * value to know when it's safe to begin decaying the remote node's value
  79      * (because we assume that it has failed). */
  80     int awol;
  81
  82 } remote_limiter_t;
  83
  84 typedef struct comm {
  85     /** Communication policy. (COMM_MESH, COMM_GOSSIP) */
  86     enum commfabrics comm_fabric;
  87
  88     /** Transport protocol. */
  89     enum transports transport_proto;
  90
  91     /** Current local value. */
  92     double local_rate;
  93
  94     /** Previous local value. */
  95     double last_local_rate;
  96
  97     double rate_change;
  98
  99     /** The number of remote nodes in the identity */
 100     uint32_t remote_node_count;
 101
 102     remote_node_t *remote_nodes;
 103
 104     /** Array containing all known remote limiters in this identity.
 105      * Contains the same information as the remote_node_map. */
 106     remote_limiter_t *remote_limiters;
 107
 108     /** Hash map containing all the remote limiters in this identity.
 109      * Indexed by the remote_node_t used to create the remote_limiter_t.
 110      * Maps to the appropriate element of the remote_limiters array. */
 111     map_handle remote_node_map;
 112
 113     /** A mutex to protect the comm structure. */
 114     pthread_mutex_t lock;
 115
 116     /** Gossip values for our local identity */
 117     gossip_t gossip;
 118
 119     /** Function pointer to send function. */
 120     int (*send_function)(struct comm *comm, uint32_t id, int sock);
 121
 122 #if 0
 123     /** Thread for handling incoming TCP data. */
 124     pthread_t tcp_recv_thread;
 125
 126     /** Descriptor set for reading TCP messages */
 127     fd_set fds;
 128 #endif
 129
 130     /** Array of integers specifiying which nodes, if any, have outstanding
 131      * unacked data.  When nodes fall in this category, and it's time to send,
 132      * these nodes will be chosen first.  This only affects gossip.  A
 133      * negative number means there is no retransmit necessary.  Otherwise, the
 134      * value is the index into the remote_limiters array of the necessary
 135      * retransmit. */
 136     int *retrys;
 137
 138 } comm_t;
 139
 140 typedef struct message {
 141     uint32_t magic;
 142     uint32_t ident_id;
 143     uint32_t seqno;
 144     uint32_t min_seqno;
 145     double value;
 146     double weight;
 147     uint16_t type;
 148 } message_t;
 149
 150 typedef struct hello_message {
 151     uint32_t magic;
 152     uint32_t ident_id;
 153     uint16_t port;
 154 } hello_t;
 155
 156 #if 0
 157 struct recv_thread_args {
 158     comm_ident_t *ident;
 159     pthread_rwlock_t *lock;
 160     uint16_t port;
 161 };
 162 #endif
 163
 164 #if 0
 165 /**
 166  * Initializes the global limiter.
 167  *
 168  * @param ipaddr The IP address on which the limiter should listen.
 169  * INADDR_ANY will suffice.  Should be specified in network byte order.
 170  *
 171  * @param port The port on which the limiter should listen. Should be specified
 172  * in network byte order.
 173  */
 174 void init_limiter(const in_addr_t ipaddr, const in_port_t port);
 175
 176 /**
 177  * Deallocates the entire global limiter.
 178  */
 179 void destroy_limiter();
 180 #endif
 181
 182 /**
 183  * Fills in the communication structure of an identity.
 184  *
 185  * @param comm The communication structure to be created/populated.
 186  *
 187  * @param config The configuration options for the identity.
 188  *
 189  * @param nodes An array of remote nodes belonging to this identity.
 190  *
 191  * @returns 0 on success, ENOMEM if memory cannot be allocated.
 192  */
 193 int new_comm(comm_t *comm, ident_config *config, remote_node_t *nodes);
 194
 195 /**
 196  * Frees the memory associated with an identity's communication structure.
 197  *
 198  * @param comm The communication structure to free.
 199  */
 200 void free_comm(comm_t *comm);
 201
 202 /**
 203  * Calculates and reads the current aggregate value for an identity.
 204  * This value includes the locally observed value.
 205  *
 206  * @param comm The comm structure of the identity in question.
 207  *
 208  * @param aggregate The location at which the aggregate value will
 209  * be stored.
 210  *
 211  * @param decayto When using a mesh comm fabric, limiters whose value
 212  * has not been heard in several timesteps will decay to this value.
 213  * Generally globallimit/N.
 214  *
 215  * @returns 0 on success, EINVAL on error.
 216  */
 217 int read_comm(comm_t *comm, double *aggregate, double decayto);
 218
 219 /**
 220  * Updates the locally observed value of an identity.
 221  *
 222  * @param comm The comm structure of the identity to update.
 223  *
 224  * @param value The new locally observed value.
 225  *
 226  * @returns 0 on success, EINVAL on error.
 227  */
 228 int write_local_value(comm_t *comm, const double value);
 229
 230 /**
 231  * Sends the local state information to one or more peer limiters in the same
 232  * identity.  If the identity is configured as a mesh, it will send to all
 233  * peers.  If the identity is configured using gossip, it will send to the
 234  * number of peers specified by the gossip_branch field of the comm_config_t
 235  * that was used to configure the identity.
 236  *
 237  * @param comm The communication structure of the identity whose value should
 238  * be propagated.
 239  *
 240  * @param id The unique id of the identity that is sending.
 241  *
 242  * @returns 0 on success, ENOMEM if there was not enough memory, or possibly
 243  * other E values as a result of socket-related results.
 244  */
 245 int send_update(comm_t *comm, uint32_t id);
 246
 247 /**
 248  * Thread that is responsible for receiving data from other limiters.
 249  *
 250  * @param limiter The limiter_t that is to be doing the receiving.
 251  */
 252 void *limiter_receive_thread(void *unused);
 253
 254 #endif  /* _DRL_STATE_ */