RTO Mechanism for Retransmission

- Acknowledgment reception and handling
- Error handling
- Communication component states
- Bus Off recovery mechanism

Author: ShoshiBnayahu

communication/src/communication.cpp

@@ -139,7 +139,13 @@ void GlobalClock::runClock()
         pthread_mutex_lock(
             &shared_clock
                  ->tick_mutex); // Lock the mutex to safely increment the tick
-        shared_clock->current_tick++; // Increment the tick
+        // Increment the tick and check for overflow
+        if (shared_clock->current_tick.load() >= MAX_TICK) {
+            shared_clock->current_tick.store(0); // Reset to 0 if max is reached
+        }
+        else {
+            shared_clock->current_tick++; // Increment the tick
+        }
         pthread_cond_broadcast(
             &shared_clock
                  ->tick_cond); // Notify all waiting processes that the tick has changed

communication/src/communication.h

@@ -10,8 +10,12 @@
 #include <unistd.h>
 #include <csignal>
 #include <pthread.h>
+#include <limits> 
+
 #define SHM_NAME "/clock__shm"
-#define TICK_DURATION std::chrono::milliseconds(500) // Define the tick duration
+#define TICK_DURATION std::chrono::milliseconds(50) // Define the tick duration
+#define MAX_TICK std::numeric_limits<int>::max() // Define the maximum value for the tick
+
 /**
  * @struct SharedClock
  * @brief A structure representing the shared state of the global clock.
@@ -25,11 +29,9 @@
  */
 struct SharedClock {
     std::atomic<int> current_tick; ///< The current tick value.
-    std::atomic<bool> is_running;  ///< Whether the clock is currently running.
-    pthread_mutex_t
-        tick_mutex; ///< Mutex to synchronize access to the tick count.
-    pthread_cond_t
-        tick_cond; ///< Condition variable to notify waiting threads on tick change.
+    std::atomic<bool> is_running;   ///< Whether the clock is currently running.
+    pthread_mutex_t tick_mutex;     ///< Mutex to synchronize access to the tick count.
+    pthread_cond_t tick_cond;       ///< Condition variable to notify waiting threads on tick change.
 };
 /**
  * @class GlobalClock

communication/src/global_clock.cpp

@@ -21,7 +21,7 @@ uint32_t Message::generateMessageID(MessageType messageType, uint16_t srcID) {
 
     auto now = std::chrono::system_clock::now().time_since_epoch();
     uint32_t timestamp = static_cast<uint32_t>(
-        std::chrono::duration_cast<std::chrono::milliseconds>(now).count() & 0xFF
+        std::chrono::duration_cast<std::chrono::nanoseconds>(now).count() & 0xFF
     );
 
     return (messageTypeID << 24) | 
@@ -37,9 +37,11 @@ Message::Message(uint32_t tps)
 
 // Add a packet to the received message
 bool Message::addPacket(const Packet &p)
-{    
-    if (p.getPSN() != packets.size()) {
-        return false;
+{
+    if (!packets.empty()) {
+        if (p.getPSN() <= packets.back().getPSN()) {
+            return false;
+        }
     }
     packets.push_back(p);
     return true;
@@ -48,7 +50,8 @@ bool Message::addPacket(const Packet &p)
 // Check if the message is complete
 bool Message::isComplete() const
 {
-    return packets.size() == tps;
+    if(packets.size() == tps)
+       return packets.size() == tps;
 }
 
 // Get the complete data of the message
@@ -67,3 +70,10 @@ std::vector<Packet> &Message::getPackets()
 {
     return packets;
 }
+
+// Extract the MessageType from the message ID
+MessageType Message::getMessageTypeFromID(uint32_t messageID) {
+    uint32_t messageTypeID = (messageID >> 24) & 0xFF;  // Extract the message type
+    return static_cast<MessageType>(messageTypeID);      // Cast to MessageType
+}
+

communication/src/global_clock.h

@@ -47,4 +47,7 @@ class Message {
 
     // Get the packets of the message
     std::vector<Packet>& getPackets();
+
+    // Extract the MessageType from the message ID
+    static MessageType getMessageTypeFromID(uint32_t messageID);
 };

communication/src/message.cpp

@@ -0,0 +1,85 @@
+#include "scheduler.h"
+
+Scheduler::Scheduler()
+{
+}
+
+Scheduler::~Scheduler()
+{
+    stopAllTimers();
+}
+
+void Scheduler::stopAllTimers()
+{
+    for (auto &future : futures)
+    {
+        if (future.valid())
+        {
+            future.wait(); // Wait for all threads to finish
+        }
+    }
+}
+
+void Scheduler::startRetransmissionTimer(int packetID, Callback callback, std::shared_ptr<std::promise<bool>> ackPromise)
+{
+    // Promise to manage the lifecycle of the thread itself
+    std::promise<void> threadCompletionPromise;
+    
+    // Future object to track the thread's completion
+    std::future<void> future = threadCompletionPromise.get_future();
+    
+    // Store the future to ensure we can wait for the thread to finish later
+    futures.push_back(std::move(future));
+
+    // Start a new thread for handling retransmission and ACK wait
+    std::thread([this, packetID, callback, threadCompletionPromise = std::move(threadCompletionPromise), ackPromise]() mutable
+                {
+                    int retryCount = 0;
+                    
+                    {
+                        // Lock the mutex to synchronize access to shared data (ackReceived)
+                        std::unique_lock<std::mutex> lock(mutex);
+
+                        // Wait for an ACK or timeout
+                        if (cv.wait_for(lock, MAX_ACK_TIMEOUT, [this, packetID]()
+                                        { return ackReceived[packetID]; }))
+                        {
+                            // ACK received within the timeout period
+                            std::cout << "ACK received for packet ID: " << packetID << std::endl;
+                            clearPacketData(packetID); // Clear packet data
+                            
+                            // Set both promises to indicate success and thread completion
+                            threadCompletionPromise.set_value(); 
+                            ackPromise->set_value(true); // ACK was received, set to true
+                            return; // Exit the thread
+                        }
+                        else
+                        {
+                            // Timeout occurred, retransmit the packet
+                            retryCounts[packetID]++;
+                            retryCount = retryCounts[packetID];
+                            std::cout << "Timeout! Retransmitting packet ID: " << packetID << std::endl;
+                        }
+                    }
+                    
+                    // Call the callback function with the updated retry count
+                    callback(retryCount);
+                    
+                    // Set the promise to indicate the thread has finished
+                    threadCompletionPromise.set_value(); 
+                })
+        .detach(); // Detach the thread to allow it to run independently
+}
+
+void Scheduler::receiveACK(int packetID)
+{
+    std::unique_lock<std::mutex> lock(mutex);
+    ackReceived[packetID] = true;
+    cv.notify_all(); // Notify all waiting threads
+}
+
+void Scheduler::clearPacketData(int packetID)
+{
+    ackReceived.erase(packetID);
+    retryCounts.erase(packetID);
+}

communication/src/message.h

@@ -0,0 +1,81 @@
+#ifndef __SCHEDULER_H__
+#define __SCHEDULER_H__
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+#include "global_clock.h"
+
+// must be set
+#define MAX_ACK_TIMEOUT \
+    (20 * TICK_DURATION) ///< Maximum time to wait for ACK
+
+/**
+ * @class Scheduler
+ * @brief Manages retransmission timers and tracks acknowledgments for packets.
+ *
+ * This class provides functionality to manage retransmission timers for
+ * packets, handle acknowledgments, and clean up packet data once
+ * retransmissions are complete.
+ */
+class Scheduler
+{
+public:
+    using Callback = std::function<void(int)>;
+
+    Scheduler();
+    ~Scheduler();
+
+    /**
+     * @brief Stops all active timers and waits for their completion.
+     *
+     * This method ensures that all active timers (threads) complete their
+     * execution before the program exits.
+     */
+    void stopAllTimers();
+
+    /**
+     * @brief Starts a retransmission timer for a given packet ID.
+     * 
+     * The function initiates a timer to wait for an ACK (Acknowledgment) for the specified packet.
+     * If the ACK is received within the MAX_ACK_TIMEOUT, it clears the packet data and sets the result 
+     * of the ackPromise to `true`, indicating success. If the timeout occurs and no ACK is received,
+     * it triggers the provided callback function to retransmit the packet and increments the retry count.
+     * 
+     * @param packetID The unique ID of the packet being transmitted.
+     * @param callback The callback function to call when retransmitting the packet after a timeout.
+     * @param ackPromise A shared promise that communicates whether the packet transmission was successful 
+     *                   (i.e., ACK was received).
+     */
+    void startRetransmissionTimer(int packetID, Callback callback, std::shared_ptr<std::promise<bool>> ackPromise);
+
+    /**
+     * @brief Receives an acknowledgment for a packet.
+     *
+     * @param packetID The ID of the packet that has been acknowledged.
+     */
+    void receiveACK(int packetID);
+
+    /**
+     * @brief Clears the data associated with a packet.
+     *
+     * @param packetID The ID of the packet whose data is to be cleared.
+     */
+    void clearPacketData(int packetID);
+
+private:
+    std::unordered_map<int, bool>
+        ackReceived; ///< Map to track received acknowledgments
+    std::unordered_map<int, int>
+        retryCounts;            ///< Map to track retry counts for packets
+    std::mutex mutex;           ///< Mutex for thread safety
+    std::condition_variable cv; ///< Condition variable for synchronization
+    std::vector<std::future<void>>
+        futures; ///< Vector to store futures of active threads
+};
+
+#endif // __SCHEDULER_H__