Multithreading and concurrency

#include <iostream>
#include <thread>
#include <chrono>
#include <thread>
#include <vector>

struct Worker
{
    int& acc;

    Worker(int& acc): acc(acc) { }

    void operator()(int x)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        acc = acc + x * x;
    }
};

int main()
{
    int result = 0;

    std::vector<std::thread> thread_list{};
    for(int i = 1; i <= 10; i++)
    {
        thread_list.push_back( std::thread{Worker(result), i} );
    }

    for(auto& t: thread_list) { t.join(); }
    std::cout << " result   = " << result << std::endl;
    return 0;
}

$ g++ thread1.cpp -o thread1.bin -std=c++1z -Wall -Wextra -O0 -g -lpthread 

$ ./thread1.bin 
 result   = 385

$ ./thread1.bin 
 result   = 384

$ ./thread1.bin 
 result   = 385

$ ./thread1.bin 
 result   = 368

$ ./thread1.bin 
 result   = 385

./thread1.bin 
 result   = 376

#include <iostream>
#include <thread>
#include <chrono>
#include <thread>
#include <vector>

struct Worker
{
    int& acc;

    Worker(int& acc): acc(acc) { }

    void operator()(int x)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        acc = acc + x * x;
    }
};

int main()
{
    int result = 0;

    std::vector<std::thread> thread_list{};
    for(int i = 1; i <= 10; i++)
    {
        thread_list.push_back( std::thread{Worker(result), i} );
    }

    for(auto& t: thread_list) { t.join(); }
    std::cout << " result   = " << result << std::endl;
    return 0;
}

# Compiler swiches: 
#   =>> -O0                  Disable optimization 
#   =>> -g                   Enable debug building 
#   =>> -Wthread-safety      Enable Clang compiler warning for race conditions
#   =>> -fsanitize=thread    Enable thread sanitizer 
#   =>> -fsanitize=undefined Eanble undefined behavior sanitizer 
$ clang++ flawed-program.cpp -o out.bin -O0 -g -lpthread -Wall -Wextra -pedantic \
    -Wthread-safety \
    -fsanitize=thread \
    -fsanitize=undefined 

$ ./out.bin 

==================
WARNING: ThreadSanitizer: data race (pid=22979)
  Read of size 4 at 0x7ffdb2ae1b78 by thread T2:
    #0 Worker::operator()(int) /Users/dummy/projects/cmake-experiment/flawed-program.cpp:16:17 (out.bin+0x4b7516)
    #1 void std::__invoke_impl<void, Worker, int>(std::__invoke_other, Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:60:14 (out.bin+0x4b7421)
    #2 std::__invoke_result<Worker, int>::type std::__invoke<Worker, int>(Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:95:14 (out.bin+0x4b7026)
    #3 _ZNSt6thread8_InvokerISt5tupleIJ6WorkeriEEE9_M_invokeIJLm0ELm1EEEEDTclsr3stdE8__invokespcl10_S_declvalIXT_EEEEESt12_Index_tupleIJXspT_EEE /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:244:13 (out.bin+0x4b6eb6)
    #4 std::thread::_Invoker<std::tuple<Worker, int> >::operator()() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:253:11 (out.bin+0x4b6ca8)
    #5 std::thread::_State_impl<std::thread::_Invoker<std::tuple<Worker, int> > >::_M_run() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:196:13 (out.bin+0x4b5fe7)
    #6 <null> <null> (libstdc++.so.6+0xbf5c2)

  Previous write of size 4 at 0x7ffdb2ae1b78 by thread T1:
    #0 Worker::operator()(int) /Users/dummy/projects/cmake-experiment/flawed-program.cpp:16:15 (out.bin+0x4b75c6)
    #1 void std::__invoke_impl<void, Worker, int>(std::__invoke_other, Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:60:14 (out.bin+0x4b7421)
    #2 std::__invoke_result<Worker, int>::type std::__invoke<Worker, int>(Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:95:14 (out.bin+0x4b7026)
    #3 _ZNSt6thread8_InvokerISt5tupleIJ6WorkeriEEE9_M_invokeIJLm0ELm1EEEEDTclsr3stdE8__invokespcl10_S_declvalIXT_EEEEESt12_Index_tupleIJXspT_EEE /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:244:13 (out.bin+0x4b6eb6)
    #4 std::thread::_Invoker<std::tuple<Worker, int> >::operator()() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:253:11 (out.bin+0x4b6ca8)
    #5 std::thread::_State_impl<std::thread::_Invoker<std::tuple<Worker, int> > >::_M_run() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:196:13 (out.bin+0x4b5fe7)
    #6 <null> <null> (libstdc++.so.6+0xbf5c2)

  As if synchronized via sleep:
    #0 nanosleep <null> (out.bin+0x43565e)
    #1 void std::this_thread::sleep_for<long, std::ratio<1l, 1000l> >(std::chrono::duration<long, std::ratio<1l, 1000l> > const&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:379:9 (out.bin+0x4b7798)
    #2 Worker::operator()(int) /Users/dummy/projects/cmake-experiment/flawed-program.cpp:15:11 (out.bin+0x4b74ef)
    #3 void std::__invoke_impl<void, Worker, int>(std::__invoke_other, Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:60:14 (out.bin+0x4b7421)
    #4 std::__invoke_result<Worker, int>::type std::__invoke<Worker, int>(Worker&&, int&&) /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/bits/invoke.h:95:14 (out.bin+0x4b7026)
    #5 _ZNSt6thread8_InvokerISt5tupleIJ6WorkeriEEE9_M_invokeIJLm0ELm1EEEEDTclsr3stdE8__invokespcl10_S_declvalIXT_EEEEESt12_Index_tupleIJXspT_EEE /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:244:13 (out.bin+0x4b6eb6)
    #6 std::thread::_Invoker<std::tuple<Worker, int> >::operator()() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:253:11 (out.bin+0x4b6ca8)
    #7 std::thread::_State_impl<std::thread::_Invoker<std::tuple<Worker, int> > >::_M_run() /usr/bin/../lib/gcc/x86_64-redhat-linux/8/../../../../include/c++/8/thread:196:13 (out.bin+0x4b5fe7)
    #8 <null> <null> (libstdc++.so.6+0xbf5c2)

  Location is stack of main thread.

  Location is global '??' at 0x7ffdb2ac4000 ([stack]+0x00000001db78)

  Thread T2 (tid=22982, running) created by main thread at:
    #0 pthread_create <null> (out.bin+0x425eba)
    #1 std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_delete<std::thread::_State> >, void (*)()) <null> (libstdc++.so.6+0xbf8a8)
    #2 main /Users/dummy/projects/cmake-experiment/flawed-program.cpp:27:34 (out.bin+0x4b3304)

  Thread T1 (tid=22981, finished) created by main thread at:
    #0 pthread_create <null> (out.bin+0x425eba)
    #1 std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_delete<std::thread::_State> >, void (*)()) <null> (libstdc++.so.6+0xbf8a8)
    #2 main /Users/dummy/projects/cmake-experiment/flawed-program.cpp:27:34 (out.bin+0x4b3304)

SUMMARY: ThreadSanitizer: data race /Users/dummy/projects/cmake-experiment/flawed-program.cpp:16:17 in Worker::operator()(int)
==================
 result   = 304
ThreadSanitizer: reported 1 warnings

SUMMARY: ThreadSanitizer: data race /Users/... ..../flawed-program.cpp:16:17 in Worker::operator()(int)

acc = acc + x * x;

#include <iostream>
#include <thread>
#include <chrono>
#include <thread>
#include <vector>
#include <mutex>

using namespace std::chrono_literals;

struct Worker
{
    int& acc;
    // Requires <mutex> header
    std::mutex& m;

    Worker(int& acc, std::mutex& m): acc(acc), m(m) { }

    void operator()(int x)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        // --- Start of critical section ---- //
        m.lock();   // Curren thread acquire locks
        acc = acc + x * x;
        m.unlock(); // Current thread releases lock
        // --- End of critical section ---- //
    }
};


int main()
{
    // Shared resource
    int result = 0;
    std::mutex m;

    std::vector<std::thread> thread_list{};
    for(int i = 1; i <= 10; i++)
    {
        thread_list.push_back( std::thread{Worker(result, m), i} );
    }

    for(auto& t: thread_list) { t.join(); }

    std::cout << " result   = " << result << std::endl;

    return 0;
}

$ g++ thread2.cpp -o thread2.bin -std=c++1z -Wall -Wextra -O0 -g -lpthread 

$ ./thread2.bin 
 result   = 385

$ ./thread2.bin 
 result   = 385

$ ./thread2.bin 
 result   = 385

$ ./thread2.bin 
 result   = 385


$ ./thread2.bin 
 result   = 385

void operator()(int x)
{
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    // --- Start of critical section ---- //
    // Acquires lock 
    std::lock_guard<std::mutex> mutex_guard(m);
    acc = acc + x * x;
    // --- End of critical section ---- //
} // Releases lock here, when the mutex_guard goes out of scope and is destroyed. 

#include <iostream>
#include <thread>
#include <chrono>
#include <thread>
#include <vector>
#include <mutex>
#include <atomic>

using namespace std::chrono_literals;

struct Worker
{
    std::atomic<int>& acc;
    // Requires <mutex> header
    std::mutex& m;

    Worker(std::atomic<int>& acc, std::mutex& m)
       : acc(acc), m(m) { }

    void operator()(int x)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        acc += x * x;
    }
};

int main()
{
    // Shared resource
    std::atomic<int> result = 0;
    std::mutex m;

    std::vector<std::thread> thread_list{};
    for(int i = 1; i <= 10; i++)
    {
        thread_list.push_back( std::thread{Worker(result, m), i} );
    }

    for(auto& t: thread_list) { t.join(); }

    std::cout << " result   = " << result << std::endl;

    return 0;
}

$ ./thread3.bin 
 result   = 385

$ ./thread3.bin 
 result   = 385

$ ./thread3.bin 
 result   = 385

$ ./thread3.bin 
 result   = 385

  double heavy_computation(doube input)
  {
      // do something .... heavy calculation 
      return output; 
  }

 // Return value ignored. 
 auto th = std::thread( &heavy_computation, 10);

// ... ... ... 

 th.join();

 double output = 0.0;

 // Return value ignored. 
 auto th = std::thread( &[&]
          { 
             // perform heavy computation // 
              .... 
             /* set output */
             output =  .... 
          }, 10);

// ... ... ... 

 th.join();
 printf(" output = %f", output);

void heavy_computation(doube input, double& output1, double* output2)
{
    // do something .... heavy calculation 
    output1 =  .... ;
    *output2 = "Something else ...."; 
}

double      output1 = 0.0; 
std::string output2 = "";

auto th = std::thread(&heavy_computation, std::ref(output1), &output2);
th.join();

std::cout << " output1 = " << output1 " ; " << " output2 = " << output2 << "\n";

auto do_something = [](int input)
{
   // ....  .... // 
   if(input < 0 ){ throw std::logic_error("Invalid input"); }
   //.... ... ... ... //
};

// DOES NOT WORK =>>> C++ runtime calls std::terminate !!!
try
{
   auto th = std::thread(do_something, -10);
} catch(std::logic_error& ex)
{
   std::cout << " Error: " << ex.what() << "\n".;
}

std::exception_ptr exptr = nullptr;

auto do_something = [&](int input)
{
  try {
   // ....  .... // 
   if(input < 0 ){ throw std::logic_error("Invalid input"); }
   //.... ... ... ... //

  // Catch all exceptions 
  } catch(...) { 
      exception_ptr = std::current_exception();
  }
};

try
{
   auto th = std::thread(do_something, -10);
   th.join(); 
   std::rethrow_exception(exptr);
} catch(std::logic_error& ex)
{
   std::cout << " Error: " << ex.what() << "\n".;
}

// Experiment returning results from threads
//---------------------------------------------
#include <iostream>
#include <chrono>
#include <thread>
#include <vector>
#include <functional>

using namespace std::chrono_literals;
namespace cr = std::chrono;

struct Timer
{
    decltype(cr::steady_clock::now()) start;

    Timer(){  start = cr::steady_clock::now();  }

    ~Timer()
    {
        auto end = cr::steady_clock::now();
        auto duration = cr::duration_cast<cr::seconds>(end - start);
        std::cout << " [TRACE] Elapsed time: " << duration.count()
                  << std::endl;
    }
};


double heavy_computation(int input)
{
    using namespace std::string_literals;

    std::this_thread::sleep_for(4s);
    if(input < 0) {
        throw std::runtime_error("Error: invalid input: "s + std::to_string(input));
    }
    return input * 3.815 + 5.6;
}


int main(int argc, char* argv[])
{
    std::vector<int>         inputs{2, 10, 5};
    std::vector<double>      results(3);
    std::vector<std::thread> threads;
    threads.reserve(3);

    // Note: A std::thread will never return anything from a function
    // passed as argument, even if has non-void return type.
    // The only to return a value from a thread is to set an
    // argument passed by pointer or reference or set a global object.
    auto thread_adapter = [&](size_t index, int input)
    {
        results[index] = heavy_computation(input);
    };

    std::puts("\n ===== Experiment 1 - Return value from threads (Parallel Computing) ==== \n");

    {   // Shows elapsed time at end of this scope
        auto timer = Timer{};

        for(size_t i = 0; i < 3; i++) {
            threads.push_back( std::thread(thread_adapter, i, inputs[i]) );
        }

        std::puts(" [INFO] Waiting for thread completion");

        // Wait for the completion of all threads
        for(size_t i = 0; i < 3; i++) { threads[i].join(); }

        // Show results
        for(size_t i = 0; i < 3; i++) {
            std::printf(" Result[%lu] = %f\n", i, results[i]);
        }
    }

    ///---------------------------------------------------------------//
    std::puts("\n ===== Dealing with exceptions from threads ==== \n");

    double output = 0.0;
    std::exception_ptr exptr = nullptr;

    auto thread_adapter2 = [&](int input)
    {
        std::puts(" [TRACE] Inside thread_adapter2");

        // Note: if an exception is not caught,
        // it will not be propagated to the parent thread.
        // Instead, the C++ runtime will call std::terminate
        // causing abonormal temrination.
        try
        {
            output = heavy_computation(input);
        } catch (...)
        {
            // Catch all exceptions from the computation that
            // should run in a new thread and set a shared memory variable
            // set only once by this thread.
            exptr = std::current_exception();
        }
    };

    auto th = std::thread(thread_adapter2, -10);
    std::puts(" [TRACE] Waiting thread_adapter2 thread termination.");
    th.join();

    if(!exptr)
    {
        std::cout << " [INFO] Result of thread_adapter2 is: " << output << "\n";

    } else
    {
        try
        {
            std::rethrow_exception(exptr);
        } catch (std::runtime_error& err) {
            std::cout << " [ERROR] " << err.what() << "\n";
        }
    }

    return 0;
}

$ g++ return-thread.cpp -o out.bin -std=c++1z -O0 -g -lpthread

$ ./out.bin 

 ===== Experiment 1 - Return value from threads (Parallel Computing) ==== 

 [INFO] Waiting for thread completion
 Result[0] = 13.230000
 Result[1] = 43.750000
 Result[2] = 24.675000
 [TRACE] Elapsed time: 4

 ===== Dealing with exceptions from threads ==== 

 [TRACE] Waiting thread_adapter2 thread termination.
 [TRACE] Inside thread_adapter2
 [ERROR] Error: invalid input: -10

// Note: simplified pseudo-signature => 

// Overload 1: 
template<typename Callable, typename ... Args>
auto async( Function&& f, Args&&... args ) -> std::future<Return>;

// Overload 2: 
template<typename Return, typename Callable, typename ... Args>
auto async( std::launch policy, Function&& f, Args&&... args ) -> std::future<Return>;

 double heavy_computation(int size, std::string label, bool flag)
 { 
   // Sleep blocking the current thread for 10 seconds.
   std::this_thread::sleep_for(10s);
   // ... ... ... .. 
   return output; 
}

 std::future<double> futA = std::async(std::launch::async, heavy_computation, 100, "calc1", false);

 auto futB = std::async(std::launch::async, [](int n)
            { 
               // .... heavy CPU bound computation ... // 
               double output = ...; 
                 ... ... ... 
               return output;
            }, 200);

// Block current thread - similar to std::thread::join()
double resultA = futA.get(); 
double resultB = futB.get(); 

std::cout << " resultA = " << resultA << " ; " << resultB << "\n";

// Wait for the computation completion blocking current thread.
futB.wait(); 

 double computation(int input)
 { 
   ... .... ... ... 

   if(bad_input(input)) { 
       throw std::runtime_error("Invalid input, try again ..."); 
   }

   ... .... ... ... 
   return result; 
}

 // Type: std::future<double>
 auto afuture = std::async(std::launch::async, computation, 10);

 try
 { 
   double result = afuture.get(); 
   std::cout << " Result = " << result << "\n";
 } catch(std::runtime_error& ex)
 {
   std::cout << " [ERROR] " << ex.what() << "\n":
 }

  double computation(int input)
  { 
    ... .... ... ... 
    // Sleep blocking this thread for 8 seconds 
    // for simulating a heavy CPU-bound computation.
    std::this_thread::sleep_for(8s);
    ... .... ... ... 
    return result; 
 }

 std::vector<std::future<double>> futures; 
 futures.reserver(3);
 std::vector<int> inputs {10, 20, 25};
 std::vector<double> outputs; 

for(auto& x: inputs){ 
   futures.push_back( std::async(std::launch::async, computation, x);   ) 
};

// Block current thread waiting all computations finish. 
// If every computation takes 8 seconds, the total time waiting 
// will be 8 seconds since they are all run in parallel. 
for(auto& fut: futures)
{
   double x = fut.get(); 
   outputs.push_back(x);
   std::cout << "Result = " << x << "\n";
}

#include <iostream>
#include <thread>
#include <chrono>

#include <queue>
#include <vector>

// --- Concurrency Headers ---- //
#include <thread>
#include <mutex>
#include <future>
#include <type_traits>
#include <optional>

namespace cr = std::chrono;
using namespace std::chrono_literals;

#define LOG_FUNCTION_ENTRY() \
          std::cerr << " [TRACE-ENTRY] Function = " << __FUNCTION__  \
          << " ; threadID " << std::this_thread::get_id() << std::endl


struct TimeCounter
{


    decltype(std::chrono::steady_clock::now()) start;
    TimeCounter(){  start = cr::steady_clock::now();  }

    void report(std::string const& label)
    {
        auto end = cr::steady_clock::now();
        auto duration = cr::duration_cast<cr::seconds>(end - start);
        std::cout << " [TRACE] Elapsed time: {" << label << "} = " << duration.count()
                  << std::endl;
    }
};

std::mutex mu;

double expensive_computationA(int n)
{
    {   /* --- Start of critical section --- */
        // Protect std::cout from race condition
        std::lock_guard<std::mutex> guard(mu);
        LOG_FUNCTION_ENTRY();
    }   /* --- End of critical section --- */
    std::this_thread::sleep_for(4s);
    return 4.0 * n + 5;
}

double expensive_computationB(int n)
{
    {   /* --- Start of critical section --- */
        // Protect std::cout from race condition
        std::lock_guard<std::mutex> guard(mu);
        LOG_FUNCTION_ENTRY();
    }   /* --- End of critical section --- */
    std::this_thread::sleep_for(8s);
    return 2.515 * n + 15.8714;
}

std::string
computatio_with_exception(int n)
{
    LOG_FUNCTION_ENTRY();
    std::this_thread::sleep_for(1s);
    if(n < 0) { throw std::logic_error(" Error: not allowed negative N"); }
    return "Hello world";
}



int main(int argc, char** argv)
{

    std::cout << " Number of hardware threads = "
              << std::thread::hardware_concurrency() << std::endl;

#if 1

    LOG_FUNCTION_ENTRY();

    std::puts("\n ==== EXPERIMENT 1 >>  without std::future => Run Serially ==== ");
    std::puts("-----------------------------------------------------------------\n");
    {
        TimeCounter tc;
        double resultA = expensive_computationA(10);
        double resultB = expensive_computationB(10);

        std::cout << " [TRACE] Completed =>> resultA = " << resultA
                  << " ; resultB = " << resultB
                  << std::endl;
        tc.report("EXPERIMENT1");
    }

    std::puts("\n ==== EXPERIMENT 2 >> with std::futures - Run in Parallel ===== ");
    std::puts("-----------------------------------------------------------------\n");
    {
        TimeCounter tc;


        // Function runs in another thread when the future object
        // is instantiated.
        // The default policy of std::async is (async | deferred) which
        // would let the C++ runtime decide when run the thread.
        std::future<double> futA = std::async(std::launch::async
                                              , &expensive_computationA, 10);

        // Function runs only when the std::future<T>::get() method is called.
        // The function expensive_computationB is not run at the moment futB object
        // is created.
        auto futB = std::async(std::launch::async, expensive_computationB, 10);

        // Protect std::cout from race condition (aka data race)
        {
            std::lock_guard<std::mutex> guard(mu);
            std::cout << " [TRACE] Waiting futures results " << std::endl;
        }

        // Calls to .get() method blocks the current thread and waits
        // for the completion of the computations wrapped in the future object.
        double resultB = futB.get();
        double resultA = futA.get();

        std::cout << " [TRACE] Completed =>> resultA = " << resultA
                  << " ; resultB = " << resultB
                  << std::endl;

        tc.report("EXPERIMENT2");

        // The method can only be used only once!!
        // otherwise, it calls std::termiante causing abnormal termination.
        //---------------------------------
        // double resultAA = futA.get(); // DO NOT!
    }

    std::puts("\n ==== EXPERIMENT 3 >> Exceptions handling ====================== ");
    std::puts("-----------------------------------------------------------------\n");
    {
        std::cout << "Future Created" << "\n";
        auto fut1 = std::async(computatio_with_exception, -10);

        try {
            auto result1 = fut1.get();
            std::cout << " [INFO] Result1 = " << result1 << "\n";
        } catch(std::logic_error& ex)
        {
            std::cerr << " [ERROR] " << ex.what() << "\n";
        }

    }

#endif

    return 0;
}

clang++ future1.cpp -o future1.bin -std=c++1z -lpthread -Wall -Wextra -O0 -g

 $ ./future1.bin 

 Number of hardware threads = 4

 ==== EXPERIMENT 1 >>  without std::future => Run Serially ==== 
-----------------------------------------------------------------

 [TRACE-ENTRY] Function = main ; threadID 139662702782272
 [TRACE-ENTRY] Function = expensive_computationA ; threadID 139662702782272
 [TRACE-ENTRY] Function = expensive_computationB ; threadID 139662702782272
 [TRACE] Completed =>> resultA = 45 ; resultB = 41.0214
 [TRACE] Elapsed time: {EXPERIMENT1} = 12

 ==== EXPERIMENT 2 >> with std::futures - Run in Parallel ===== 
-----------------------------------------------------------------

 [TRACE] Waiting futures results 
 [TRACE-ENTRY] Function = expensive_computationA ; threadID 139662684923648
 [TRACE-ENTRY] Function = expensive_computationB ; threadID 139662676530944
 [TRACE] Completed =>> resultA = 45 ; resultB = 41.0214
 [TRACE] Elapsed time: {EXPERIMENT2} = 8

 ==== EXPERIMENT 3 >> Exceptions handling ====================== 
-----------------------------------------------------------------

Future Created
 [TRACE-ENTRY] Function = computatio_with_exception ; threadID 139662684923648
 [ERROR]  Error: not allowed negative N

// Message queue of fixed size 10 
fixed_message_queue<double> buffer(10);

auto consumer_thread = thread([&]{
   while(buffer.is_closed()){
      sleep(2);
      // Block current thread if the queue is empty 
      double next = buffer.take();   
      print("Received message x = %.3f", next);
   }
});

auto producer_thread = thread([&]{
   for(double x = 0.0; x < 100.0; x += 2.0) {
      sleep(10);
      double y = 4 * x + 10;
      // Block current thread if the queue is full (has 10 elements)
      double next = buffer.put(x);   
      print("Received message x = %.3f", next);
   }
   // Indicates that messages will no longer be sent 
   queue.close();
});

#include <iostream>
#include <thread>
#include <chrono>

#include <queue>
#include <vector>
#include <functional>

// --- Concurrency Headers ---- //
#include <thread> // threads
#include <mutex>  // mutex, lock_guard, unique_lock
#include <future> // conditional_variables

using namespace std::chrono_literals;

template<typename T>
class MessageQueue 
{
    using guard = std::lock_guard<std::mutex>;

    std::queue<T>           m_queue;
    std::condition_variable m_cvar;
    mutable std::mutex      m_mutex;

public:
    MessageQueue(){}

    // Called by producer thread
    void push(T& value)
    {
        auto g = guard(m_mutex);
        m_queue.push(value);
        m_cvar.notify_one();
    }

    void push(T&& value)
    {
        auto g = guard(m_mutex);
        m_queue.push(value);
        m_cvar.notify_one();
    }


    // Called by consumer thread.
    // blocks consumer thread making
    // it sleep when the queue is empty
    void pop()
    {
        auto g = guard(m_mutex);
        m_queue.pop();
    }

    // Called by consumer thread
    T& front()
    {
        auto lock = std::unique_lock<std::mutex>(m_mutex);
        // The second argument of m_cvar.wait, a lambda function is
        // passed as predicate to prevent spurious thread wake up.
        m_cvar.wait(lock, [&]{ return !m_queue.empty(); });
        T& result = m_queue.front();
        lock.unlock();
        return result;
    }

    // Called by consumer thread
    const T& front() const
    {
        auto lock = std::unique_lock<std::mutex>(m_mutex);
        m_cvar.wait(lock, [&]{ return !m_queue.empty(); });
        T& result = m_queue.front();
        lock.unlock();
        return result;
    }

    bool empty() const
    {
        auto g = guard(m_mutex);
        return m_queue.front();
    }

    std::size_t size() const
    {
        auto g = guard(m_mutex);
        return  m_queue.size();
    }

};



int main(int argc, char* argv[])
{

    MessageQueue<double> queue;

    // Mutex used for protecting std::cout from frace condition
    std::mutex output_mutex;
    std::atomic_bool finished = false;

    auto consumer_thread = std::thread( [&]
    {
      {
          auto guard = std::lock_guard(output_mutex);
          std::cout << " [CONSUMER] Consumer thread started." << "\n";
      }
        while(true)
        {
            auto value = queue.front();
            {
                auto guard = std::lock_guard(output_mutex);
                std::cout << " [CONSUMER] Wait input ..." << "\n";
                std::cout << " [CONSUMER] Received value = " << value << "\n";
            }
            queue.pop();
            if(!queue.empty() && finished) { break; }
        }
    });

     // Protect cout from race condition
    {
        auto guard = std::lock_guard(output_mutex);
        std::cout << " [TRACE] Start interactive SHELL OK." << "\n";
    }

    // The producer thread is the current one (main thread)
    for(int i = 0; i < 5; i++)
    {
        std::this_thread::sleep_for(1s);
        double x = i * i - 4 * i + 10.0 ;
        {
            auto guard = std::lock_guard(output_mutex);
            std::cout << " [PRODUCER] => Sending number: " << x << "\n";
        }
        queue.push(x);
    }

    finished = true;
    consumer_thread.join();
    return 0;
}

$ clang++ message_queue.cpp -o out.bin -std=c++1z -O0 -Wall -g -lpthread

$ out.bin  
 [TRACE] Start interactive SHELL OK.
 [CONSUMER] Consumer thread started.
 [PRODUCER] => Sending number: 10
 [CONSUMER] Wait input ...
 [CONSUMER] Received value = 10
 [PRODUCER] => Sending number: 7
 [CONSUMER] Wait input ...
 [CONSUMER] Received value = 7
 [PRODUCER] => Sending number: 6
 [CONSUMER] Wait input ...
 [CONSUMER] Received value = 6
 [PRODUCER] => Sending number: 7
 [CONSUMER] Wait input ...
 [CONSUMER] Received value = 7
 [PRODUCER] => Sending number: 10
 [CONSUMER] Wait input ...
 [CONSUMER] Received value = 10

#include <iostream>
#include <functional>
#include <vector>
#include <string>

// Concurrency headers
#include <deque>
#include <queue>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <chrono>
#include <cassert>

class ThreadPool
{
public:
    using Task = std::function<void ()>;

    // Create a thread pool with N worker threads
    ThreadPool(size_t n)
    {
        std::cout << " [TRACE] Created thread pool with N = "
                  << n << " worker threads" << std::endl;

        for(size_t i = 0; i < n; ++i) { this->add_worker(); }
    }

    ThreadPool(): ThreadPool(std::thread::hardware_concurrency()){ }

    ~ThreadPool()
    {
        std::cout << " [TRACE] <Destructor> Waiting for termination of all threads"
                  << std::endl;
        this->shutdown();
        std::cout << " [TRACE] <Destructor> All Worker threads were shutdown. OK"
                  << std::endl;
    }

    // Return the number of worker threads
    size_t size() const
    {
        return m_workers.size();
    }

    // Submit a task to the thread pool
    void submit(Task task)
    {
        auto guard = std::lock_guard<std::mutex>{m_qmutex};
        m_queue.push(task);
        // Wake up a single worker thread signaling it.
        m_qvar.notify_one();
    }

    // Force ThreadPool shutdown
    void shutdown()
    {
        // Set the flag to false for indicating to worker threads that
        // there nor further tasks(jobs) will be submited.
        m_is_active = false;
        // Signal all threads, wake up all worker threads
        m_qvar.notify_all();
        // Wait for completion of all threads.
        for(auto& worker: m_workers) { worker.join(); }
        // Remove all worker threads
        m_workers.clear();
    }

    // Block the calling thread and waits for the completion of all tasks.
    void wait_tasks()
    {
        auto lock = std::unique_lock<std::mutex>{m_qmutex};
        m_qempty.wait(lock, [&]{ return m_queue.empty(); } );
        lock.unlock();
    }

    // Pop a task from queue => blocking the calling thread
    // if the queue is empty.
    Task pop_task()
    {
        auto lock = std::unique_lock<std::mutex>{m_qmutex};
        // Block current thread if the queue is empty or until
        // the condition variable is signaled from the queue
        // thread.
        m_qvar.wait(lock, [&]{ return !m_queue.empty(); });
        auto task = m_queue.front();
        m_queue.pop();

        // Wake up thread that called .wait_tasks()
        if(m_queue.empty()){ m_qempty.notify_all(); }

        // lock.unlock();
        return task;
    }

    // Add new worker thread
    void add_worker()
    {

        auto th = std::thread([this]{
            while(true)
            {
                // std::puts(" [INFO] Waiting task");
                auto task = pop_task();
                task();

                // Shutdown worker thread when the flag m_is_active
                // is set to false and the queue is empty
                auto guard = std::lock_guard<std::mutex>{m_qmutex};
                if(!m_is_active && m_queue.empty()){ return; }
            }

        });

        std::cout << " [TRACE] <add_worker> - Created worker thread ID = "
                  << th.get_id() << std::endl;

        m_workers.push_back(std::move(th));

    } // --- End of add_worker() --- //

private:
    // Flag indicating that the thread pool is active (running)
    std::atomic<bool>       m_is_active = true;
    // Task queue (message queue)
    std::queue<Task>        m_queue;
    // Worker threads
    std::deque<std::thread> m_workers;
    // Mutex for protecting the queue
    std::mutex              m_qmutex;
    // Condition variable for signaling that
    // there are tasks in the queue
    std::condition_variable m_qvar;

    // Condition variable for signaling that the task queue is empty
    std::condition_variable m_qempty;
};


int main(int argc, char** argv)
{
    using namespace std::chrono_literals;


    if(argc < 3){
        std::cerr<< " Error: invalid command line switch." << std::endl;
        return EXIT_FAILURE;
    }

    auto cmd = std::string(argv[1]);
    auto size = std::stoi(argv[2]);
    assert(size > 0);

    if(cmd == "serial")
    {
        for(int i = 0; i < size; i++)
        {
                std::this_thread::sleep_for(std::chrono::seconds(2));
                std::cout << " Task id = " << i
                          << " thread id = " << std::this_thread::get_id()
                          << " completed OK. \n";
        }
        return EXIT_FAILURE;
    }

    if(cmd == "parallel")
    {
        // Get thread pool size from environemnt variable
        const char* thsize = std::getenv("THREAD_POOL_SIZE");

        // std::cout << " thsize = " << thsize << std::endl;

        int n_workers = thsize != nullptr
                            ? std::stoi(thsize) : std::thread::hardware_concurrency();

        ThreadPool thp(n_workers);
        // Mutex used for protecting std::cout from data race
        std::mutex cout_mutex;

        // Run
        for(int i = 0; i < size; i++)
        {
            thp.submit([=, &cout_mutex]{
                std::this_thread::sleep_for(std::chrono::seconds(2));

                {
                    std::lock_guard<std::mutex> guard(cout_mutex);
                    std::cout << " Task id = " << i
                              << " thread id = " << std::this_thread::get_id()
                              << " completed OK. \n";
                }

            });
        }
        std::cout << " [TRACE] <main> Waiting tasks execution ..." << "\n";
    }

    return 0;
}

$ clang++ thread-pool-simple.cpp -o out.bin -std=c++1z -lpthread -O0 -g -Wall -Wextra -fsanitize=thread -fsanitize=undefined

$ time ./out.bin serial 20
 Task id = 0 thread id = 140386475086592 completed OK. 
 Task id = 1 thread id = 140386475086592 completed OK. 
 Task id = 2 thread id = 140386475086592 completed OK. 
 Task id = 3 thread id = 140386475086592 completed OK. 
  ... ...   ... ...   ... ...   ... ...   ... ... 
  ... ...   ... ...   ... ...   ... ...   ... ... 
 Task id = 17 thread id = 140386475086592 completed OK. 
 Task id = 18 thread id = 140386475086592 completed OK. 
 Task id = 19 thread id = 140386475086592 completed OK. 

real    0m40.086s
user    0m0.037s
sys     0m0.045s

$ time ./out.bin parallel 20
 [TRACE] Created thread pool with N = 4 worker threads
 [TRACE] <add_worker> - Created worker thread ID = 139760855148288
 [TRACE] <add_worker> - Created worker thread ID = 139760846755584
 [TRACE] <add_worker> - Created worker thread ID = 139760836278016
 [TRACE] <add_worker> - Created worker thread ID = 139760827885312
 [TRACE] <main> Waiting tasks execution ...
 [TRACE] <Destructor> Waiting for termination of all threads
 Task id = 0 thread id = 139760846755584 completed OK. 
 Task id = 3 thread id = 139760827885312 completed OK. 
 Task id = 1 thread id = 139760855148288 completed OK. 
 Task id = 2 thread id = 139760836278016 completed OK. 

 ... ... ...      ... ... ...      ... ... ...      ... ... ...  

 Task id = 17 thread id = 139760827885312 completed OK. 
 Task id = 18 thread id = 139760836278016 completed OK. 
 Task id = 19 thread id = 139760846755584 completed OK. 
 [TRACE] <Destructor> All Worker threads were shutdown. OK

real    0m10.103s
user    0m0.050s
sys     0m0.062s