1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
use crate::broadcast::broadcast_trait::{GenericReturn, TCB};
use crate::configuration::middleware_configuration::Configuration;
use crate::graph::communication::{acceptor, connector};
use crate::graph::middleware::dot::Dot;
use crate::graph::middleware::message_types::ClientMessage;
use crate::graph::middleware::middleware_thread;
use crate::graph::structs::message_type::ClientPeerMiddleware;
use crossbeam::crossbeam_channel::unbounded;
use crossbeam::{Receiver, RecvError, RecvTimeoutError, SendError, Sender, TryRecvError};
use std::sync::{Arc, Barrier};
use std::time::Duration;
use std::{thread, usize};

/**
 * Client side of the graph based middleware service.
 * Maintains the API and necessary state to send and deliver messages.
 */
pub struct GRAPH {
    ///Receiver end of the channel between the client and the middleware thread
    receive_channel: Receiver<ClientMessage>,
    ///Sender end of the channel between the client and the middleware thread
    middleware_channel: Sender<ClientPeerMiddleware>,
    ///Dot of the next sent message
    dot: Dot,
    ///Context of the next sent message
    context: Vec<Dot>,
}

impl GRAPH {
    /**
     * Updates the next sent message's context upon a delivery.
     *
     * # Arguments
     *
     * `message` - Delivered or stable message.
     */
    fn handle_delivery(&mut self, message: ClientMessage) -> GenericReturn {
        match message {
            ClientMessage::Delivery {
                ref payload,
                dot,
                ref context,
            } => {
                Self::update_context(&dot, context, &mut self.context);

                GenericReturn::Delivery(payload.to_vec(), dot.id, dot.counter)
            }
            ClientMessage::Stable { dot } => GenericReturn::Stable(dot.id, dot.counter),
            _ => {
                panic!("ERROR: Received an EMPTY when it shouldn't!");
            }
        }
    }

    /**
     * Updates the Peer's context upon delivery. A dot's context dots are removed
     * from the Peer's context when adding it.
     *
     * # Arguments
     *
     * `dot` - Delivered message dot.
     *
     * `message_context` - Delivered message context.
     *
     * `local_context` - Next sent message message context.
     */
    fn update_context(dot: &Dot, message_context: &Vec<Dot>, local_context: &mut Vec<Dot>) {
        local_context.retain(|&client_dot| !message_context.contains(&client_dot));
        local_context.push(dot.clone());
    }

    /**
     * Starting method of the Middleware service. It creates and initializes
     * the necessary variables, communication channels and threads.
     *
     * # Arguments
     *
     * `local_id` - Local peer's globally unique id.
     *
     * `local_port` - Port where the middleware will be listening for connections.
     *
     * `peer_addresses` - Addresses the middleware will connect to.
     *
     * `configuration` - Middleware's configuration file.
     */
    fn start_service(
        local_id: usize,
        local_port: usize,
        peer_addresses: Vec<String>,
        configuration: Arc<Configuration>,
    ) -> (Sender<ClientPeerMiddleware>, Receiver<ClientMessage>) {
        let setup_end_barrier = Arc::new(Barrier::new(peer_addresses.len() + 1));

        //Creating the clone of the middleware configuration arc
        let configuration_clone = Arc::clone(&configuration);

        //Creating the channel where the middleware writes to
        //and the client reads from
        let (middleware_send_channel, peer_receive_channel) = unbounded::<ClientMessage>();

        //Creating the channel where the main middleware thread reads from
        //and the peer threads and client write to
        let (peer_reader_send_channel, middleware_receive_channel) =
            unbounded::<ClientPeerMiddleware>();

        let peer_reader_send_channel_clone = peer_reader_send_channel.clone();

        //Cloning the port array for the acceptor thread
        let acceptor_thread_peer_addresses = peer_addresses.clone();

        //Formatting the peer's acceptor thread name
        let thread_name = format!("acceptor_thread_{}", local_id);
        let builder = thread::Builder::new()
            .name(thread_name)
            .stack_size(configuration.thread_stack_size);

        let setup_end_barrier_clone = Arc::clone(&setup_end_barrier);

        //Spawning the acceptor thread
        builder
            .spawn(move || {
                acceptor::start(
                    local_id,
                    local_port,
                    acceptor_thread_peer_addresses,
                    peer_reader_send_channel_clone,
                    configuration,
                    setup_end_barrier_clone,
                );
            })
            .unwrap();

        //Connecting to the peers' ports and getting the channels sender ends
        //between the middleware and the sender thread
        let channels_to_socket_threads: Vec<Sender<(Arc<Barrier>, Arc<Vec<u8>>)>> =
            connector::start(local_id, &peer_addresses, &configuration_clone);

        //Formatting the peer's middlware thread name
        let thread_name = format!("middleware_thread_{}", local_id);
        let builder = thread::Builder::new()
            .name(thread_name)
            .stack_size(configuration_clone.middleware_thread_stack_size);

        //Spawning the main middleware thread
        builder
            .spawn(move || {
                middleware_thread::start(
                    local_id,
                    peer_addresses,
                    middleware_receive_channel,
                    middleware_send_channel,
                    channels_to_socket_threads,
                    configuration_clone,
                )
            })
            .unwrap();

        setup_end_barrier.wait();
        //Return the channels the peer writes and reads from to the middleware
        (peer_reader_send_channel, peer_receive_channel)
    }
}

impl TCB for GRAPH {
    /**
     * Type of the return from a send call, which is the sent message context or an error.
     */
    type SendCallReturn = Result<Vec<Dot>, SendError<ClientPeerMiddleware>>;

    /**
     * Creates a new middleware instance. This function only returns after the middleware
     * has a connection to every other peer in both directions.
     *
     * # Arguments
     *
     * `local_id` - Peer's globally unique id in the group.
     *
     * `local_port` - Port where the middleware will be listening for connections.
     *
     * `peer_addresses` - Addresses the middleware will connect to.
     *
     * `configuration` - Middleware's configuration file.
     */
    fn new(
        local_id: usize,
        local_port: usize,
        peer_addresses: Vec<String>,
        configuration: Configuration,
    ) -> Self {
        let configuration = Arc::new(configuration);

        let (middleware_channel, receive_channel) =
            Self::start_service(local_id, local_port, peer_addresses, configuration);

        //Initializing the context and dot variables
        let context: Vec<Dot> = Vec::new();
        let dot = Dot::new(local_id, 0);

        GRAPH {
            receive_channel,
            middleware_channel,
            dot,
            context,
        }
    }

    /**
     * Broadcasts a message to every peer in the group.
     * Returns the sent message context if successfull.
     *
     * # Arguments
     *
     * `msg` - Serialized message to be broadcast
     */
    fn send(&mut self, msg: Vec<u8>) -> Self::SendCallReturn {
        //Incrementing the dot's counter entry
        self.dot.counter += 1;

        //Building the enum of the new message
        let client_message = ClientPeerMiddleware::Client {
            dot: self.dot.clone(),
            msg,
            context: self.context.clone(),
        };

        //Sending the enum to the middleware thread
        self.middleware_channel.send(client_message)?;
        //.expect("ERROR: Client could not send message to main middleware");

        //Clearing the context for the next sent message
        let context: Vec<Dot> = self.context.drain(..).collect();

        //Adding the last sent message's dot to the new context
        self.context.push(self.dot.clone());

        //Returning the previous message's context
        Ok(context)
    }

    /**
     * Signals and waits for the middleware to terminate.
     */
    fn end(&self) {
        let end_message = ClientPeerMiddleware::End;
        self.middleware_channel.send(end_message).unwrap();

        loop {
            match self.receive_channel.recv() {
                Ok(msg) => match msg {
                    ClientMessage::Empty => {
                        break;
                    }
                    _ => {}
                },
                Err(_) => {}
            }
        }
    }

    /**
     * Delivers a message from the middleware. Blocks the calling thread
     * until a message is delivered or the channel to the middleware is
     * empty or disconnected.
     */
    fn recv(&mut self) -> Result<GenericReturn, RecvError> {
        match self.receive_channel.recv() {
            Ok(message) => Ok(self.handle_delivery(message)),
            Err(e) => Err(e),
        }
    }

    /**
     * Attempts to deliver a message from the middleware without blocking
     * the caller thread. Either a message is immeadiately delivered
     * from the channel or an error is returned if the channel is empty.
     */
    fn try_recv(&mut self) -> Result<GenericReturn, TryRecvError> {
        match self.receive_channel.try_recv() {
            Ok(message) => Ok(self.handle_delivery(message)),
            Err(e) => Err(e),
        }
    }

    /**
     * Waits for a message to be delivered from the middleware for a
     * limited time. If the channel is empty and not disconnected, the
     * caller thread is blocked until a message is received in the channel
     * or the timeout ends. If there are no messages until the timeout ends or
     * the channel becomes disconnected, an error is returned.
     *
     * # Arguments
     *
     * `duration` - Timeout duration
     */
    fn recv_timeout(&mut self, duration: Duration) -> Result<GenericReturn, RecvTimeoutError> {
        match self.receive_channel.recv_timeout(duration) {
            Ok(message) => Ok(self.handle_delivery(message)),
            Err(e) => Err(e),
        }
    }

    /**
     * ACKS a stable message. This is needed for the GRAPH approach so the node with
     * the message's information can be deleted from the graph and its position in the
     * array be available and reused for another message. Otherwise the array that maps
     * the causal dependency graph will grow exponentially. However, if stability was
     * disabled from the configuration file, then the message's are directly removed
     * from the graph upon delivery, rendering the call to this method unnecessary.
     *
     * The VV implementation doesn't require the call of this method.
     *
     * # Arguments
     *
     * `id` - Stable dot id field
     *
     * `counter` - Stable dot counter field
     */
    fn tcbstable(&mut self, id: usize, counter: usize) {
        let dot = Dot::new(id, counter);
        let stable_dot = ClientPeerMiddleware::Stable { dot };

        self.middleware_channel
            .send(stable_dot)
            .expect("ERROR: When the Client sends a STABLE message");
    }
}