[New Feature] Timeout-capable channels

.github/workflows/cmake.yml

@@ -46,6 +46,16 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+
+      - name: Install GCC 12
+        if: runner.os == 'Linux'
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y gcc-12 g++-12 cmake
+          sudo update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-12 100
+          sudo update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-12 100
+          gcc --version
+          g++ --version
 
       - name: Create Build Environment
         # Some projects don't allow in-source building, so create a separate build directory

README.md

@@ -12,6 +12,7 @@
 * Blocking (forever waiting to fetch).
 * Range-based for loop supported.
 * Close to prevent pushing and stop waiting to fetch.
+* Optional timeout for read/write operations using `std::chrono`.
 * Integrates well with STL algorithms in some cases. Eg: std::move(ch.begin(), ch.end(), ...).
 * Tested with GCC, Clang, and MSVC.
 
@@ -29,7 +30,6 @@ see [CMakeLists.txt](./examples/cmake-project/CMakeLists.txt) from the [CMake pr
 
 ```c++
 #include <cassert>
-
 #include <msd/channel.hpp>
 
 int main() {
@@ -83,7 +83,6 @@ int main() {
 
 ```c++
 #include <iostream>
-
 #include <msd/channel.hpp>
 
 int main() {
@@ -100,6 +99,31 @@ int main() {
 }
 ```
 
+````c++
+#include <iostream>
+#include <msd/channel.hpp>
+
+int main() {
+    msd::channel<int> ch{2};
+    ch.setTimeout(std::chrono::milliseconds(100));
+
+    std::clog << "Testing write timeout on full buffer:\n";
+    try {
+        ch << 1;
+        ch << 2;
+        std::clog << "Attempting to write to full channel...\n";
+        ch << 3;
+    }
+    catch (const msd::channel_timeout& e) {
+        std::clog << "Expected timeout occurred: " << e.what() << "\n";
+    }
+
+    return 0;
+}
+
+
+```
+
 See [examples](examples).
 
 <br>

examples/CMakeLists.txt

@@ -36,3 +36,6 @@ add_example(example_multithreading multithreading.cpp)
 
 add_example(example_streaming streaming.cpp)
 run_example(example_streaming)
+
+add_example(example_timeout timeout.cpp)
+run_example(example_timeout)

examples/timeout.cpp

@@ -0,0 +1,125 @@
+#include <chrono>
+#include <iostream>
+#include <thread>
+
+#include "msd/channel.hpp"
+
+// using namespace std::chrono_literals; for post-C++11 code, use this to save some headaches
+
+void demonstrateTimeouts()
+{
+    // small capacity, short timeout
+    msd::channel<int> ch{2};
+    ch.setTimeout(std::chrono::milliseconds(100));
+
+    std::cout << "Testing write timeout on full buffer:\n";
+    try {
+        ch << 1;
+        ch << 2;
+        std::cout << "Attempting to write to full channel...\n";
+        ch << 3;
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Expected timeout occurred: " << e.what() << "\n";
+    }
+
+    std::cout << "\nTesting read timeout on empty channel:\n";
+
+    msd::channel<int> ch2{5};
+    ch2.setTimeout(std::chrono::milliseconds(200));
+
+    try {
+        int value;
+        std::cout << "Attempting to read from empty channel...\n";
+        ch2 >> value;
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Expected timeout occurred: " << e.what() << "\n";
+    }
+
+    std::cout << "\nDemonstrating timeout with range-based for loop:\n";
+
+    msd::channel<int> ch3{5};
+    ch3.setTimeout(std::chrono::milliseconds(200));  // lower this to see the timeout
+
+    // Producer
+    std::thread writer([&ch3]() {
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        ch3 << 1;
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        ch3 << 2;
+        ch3.close();
+    });
+
+    // Consumer
+    try {
+        for (const auto& value : ch3) {
+            std::cout << "Received value: " << value << "\n";
+        }
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Timeout in for loop: " << e.what() << "\n";
+    }
+
+    writer.join();
+}
+
+int main()
+{
+    // small capacity, short timeout
+    msd::channel<int> ch{2};
+    ch.setTimeout(std::chrono::milliseconds(100));
+
+    std::cout << "Testing write timeout on full buffer:\n";
+    try {
+        ch << 1;
+        ch << 2;
+        std::cout << "Attempting to write to full channel...\n";
+        ch << 3;
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Expected timeout occurred: " << e.what() << "\n";
+    }
+
+    std::cout << "\nTesting read timeout on empty channel:\n";
+
+    msd::channel<int> ch2{5};
+    ch2.setTimeout(std::chrono::milliseconds(200));
+
+    try {
+        int value;
+        std::cout << "Attempting to read from empty channel...\n";
+        ch2 >> value;
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Expected timeout occurred: " << e.what() << "\n";
+    }
+
+    std::cout << "\nDemonstrating timeout with range-based for loop:\n";
+
+    msd::channel<int> ch3{5};
+    ch3.setTimeout(std::chrono::milliseconds(200));  // lower this to see the timeout
+
+    // Producer
+    std::thread writer([&ch3]() {
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        ch3 << 1;
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        ch3 << 2;
+        ch3.close();
+    });
+
+    // Consumer
+    try {
+        for (const auto& value : ch3) {
+            std::cout << "Received value: " << value << "\n";
+        }
+    }
+    catch (const msd::channel_timeout& e) {
+        std::cout << "Timeout in for loop: " << e.what() << "\n";
+    }
+
+    writer.join();
+
+    return 0;
+}

include/msd/channel.hpp

@@ -1,9 +1,9 @@
 // Copyright (C) 2023 Andrei Avram
-
 #ifndef MSD_CHANNEL_HPP_
 #define MSD_CHANNEL_HPP_
 
 #include <atomic>
+#include <chrono>
 #include <condition_variable>
 #include <cstdlib>
 #include <mutex>
@@ -30,6 +30,14 @@ class closed_channel : public std::runtime_error {
     explicit closed_channel(const char* msg) : std::runtime_error{msg} {}
 };
 
+/**
+ * @brief Exception thrown when channel operation times out.
+ */
+class channel_timeout : public std::runtime_error {
+   public:
+    explicit channel_timeout(const char* msg) : std::runtime_error{msg} {}
+};
+
 /**
  * @brief Thread-safe container for sharing data between threads.
  *
@@ -56,10 +64,24 @@ class channel {
      */
     explicit constexpr channel(size_type capacity);
 
+    /**
+     * Sets a timeout for channel operations.
+     *
+     * @param timeout Duration after which operations will time out.
+     */
+    template <typename Rep, typename Period>
+    void setTimeout(const std::chrono::duration<Rep, Period>& timeout);
+
+    /**
+     * Clears any previously set timeout.
+     */
+    void clearTimeout() noexcept;
+
     /**
      * Pushes an element into the channel.
      *
      * @throws closed_channel if channel is closed.
+     * @throws channel_timeout if operation times out.
      */
     template <typename Type>
     friend channel<typename std::decay<Type>::type>& operator<<(channel<typename std::decay<Type>::type>&, Type&&);
@@ -68,6 +90,7 @@ class channel {
      * Pops an element from the channel.
      *
      * @tparam Type The type of the elements
+     * @throws channel_timeout if operation times out.
      */
     template <typename Type>
     friend channel<Type>& operator>>(channel<Type>&, Type&);
@@ -114,14 +137,17 @@ class channel {
     std::mutex mtx_;
     std::condition_variable cnd_;
     std::atomic<bool> is_closed_{false};
+    std::atomic<std::chrono::nanoseconds> timeout_{std::chrono::nanoseconds::zero()};
+
+    template <typename Predicate>
+    bool waitWithTimeout(std::unique_lock<std::mutex>&, Predicate);
+    bool waitBeforeRead(std::unique_lock<std::mutex>&);
+    bool waitBeforeWrite(std::unique_lock<std::mutex>&);
 
-    inline void waitBeforeRead(std::unique_lock<std::mutex>&);
-    inline void waitBeforeWrite(std::unique_lock<std::mutex>&);
     friend class blocking_iterator<channel>;
 };
 
 }  // namespace msd
 
 #include "channel.inl"
-
 #endif  // MSD_CHANNEL_HPP_