/*
std::binary_semaphore, std::counting_semaphore
https://en.cppreference.com/w/cpp/thread/counting_semaphore
ZPR
Author: Błażej Ejzak
*/

#include <chrono>
#include <iostream>
#include <format>
#include <semaphore>
#include <atomic>
#include <thread>
#include <vector>
#include <array>
#include <queue>
#include <condition_variable>
#include <functional>
#include <chrono>

using namespace std::literals;

/*
Comparison of Semaphore (C++ 20) specific approach with Condition Variable (C++ 11) approach.
This shows the behaviour of threads waiting on semaphores.
*/
class Comparison {
    static constexpr size_t NUMTHREADS = 4;
    std::vector<std::jthread> threads;

    std::binary_semaphore csem{ 1 };

    std::mutex cvMutex;
    std::condition_variable cv;
    int count = 0;
    bool signal = true;

    // C++ 20 semaphore approach (processes waiting on acquire() might be scheduled or not).
    void ThreadSem(int id) {
        std::cout << "[thread " << id << " ] awaits signal\n";
        csem.acquire();
        std::cout << "[thread " << id << " ] got signal\n";
        std::this_thread::sleep_for(4s);
        csem.release();
    }

    // C++ 11 condition_variable approach (waiting processes scheduled in a queue).
    void ThreadCVScheduled(int id) {
        {
            std::cout << "[thread " << id << " ] awaits signal\n";
            std::unique_lock<std::mutex> lock(cvMutex);
            cv.wait(lock, [this, id]() { return id == count; });
        }
        std::cout << "[thread " << id << " ] got signal\n";

        std::this_thread::sleep_for(std::chrono::seconds(4));

        {
            std::lock_guard<std::mutex> lock(cvMutex);
            count++;
            cv.notify_all();
        }
    }

    // C++ 11 condition_variable approach (waiting processes not scheduled in a queue).
    void ThreadCVNotScheduled(int id) {
        {
            std::cout << "[thread " << id << " ] awaits signal\n";
            std::unique_lock<std::mutex> lock(cvMutex);
            cv.wait(lock, [this]() { return signal; });
            signal = false;
        }
        std::cout << "[thread " << id << " ] got signal\n";

        std::this_thread::sleep_for(std::chrono::seconds(4));

        {
            std::lock_guard<std::mutex> lock(cvMutex);
            signal = true;
            cv.notify_all();
        }
    }

    void runThreads(std::function<void(int)> fun) {
        for (int i = 0; i < NUMTHREADS; i++) {
            threads.push_back(std::jthread(fun, i));
            std::this_thread::sleep_for(1s);
        }
        threads.clear(); // Join all threads;
    }

public:
    void compare() {
        std::cout << "Semaphore" << std::endl;
        runThreads(std::bind(&Comparison::ThreadSem, this, std::placeholders::_1));
        std::cout << "Condition Variable with Scheduled policy" << std::endl;
        runThreads(std::bind(&Comparison::ThreadCVScheduled, this, std::placeholders::_1));
        std::cout << "Condition Variable without Scheduled policy" << std::endl;
        runThreads(std::bind(&Comparison::ThreadCVNotScheduled, this, std::placeholders::_1));
    }
};

/*
Readers writers problem - readers-preference.
Implementation of article from https://en.wikipedia.org/wiki/Readers%E2%80%93writers_problem
*/
template<size_t readersCount>
class WritersReadersProblem {
    std::binary_semaphore rds;
    std::binary_semaphore wrts;

    std::chrono::high_resolution_clock::time_point start;
    std::chrono::seconds duration;

    uint16_t readcount;
    std::string message;

    void writer(const std::string& txt) {
        while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start) < duration) {
            wrts.acquire();

            // Writing...
            std::cout << txt << std::endl;
            message = txt;
            std::this_thread::sleep_for(100ms);

            wrts.release();

            std::this_thread::sleep_for(100ms);
        }
    }

    void reader(const std::string& id) {
        while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start) < duration) {
            rds.acquire();

            if (++readcount == 1)
                wrts.acquire();

            rds.release();

            std::cout << std::format("I am {} and started reading ", id) << std::endl;
            std::this_thread::sleep_for(100ms);
            std::cout << std::format("I am {} and finished reading -> the message is {}", id, message) << std::endl;
            rds.acquire();

            if (--readcount == 0)
                wrts.release();

            rds.release();
            std::this_thread::sleep_for(100ms);
        }
    }
public:
    WritersReadersProblem(const std::chrono::seconds duration) : duration(duration), readcount(0), rds(1), wrts(1) {}

    void runSimulation() {
        start = std::chrono::high_resolution_clock::now();
        std::array<std::jthread, readersCount> readers;
        std::array<std::jthread, 2> writers = {
            std::jthread(&WritersReadersProblem::writer, this, "Thread1: abcdefghijklmnopqrstuvw"),
            std::jthread(&WritersReadersProblem::writer, this, "Thread2: loremipsum loremipsum")
        };

        for (size_t i = 0; i < readersCount; i++) {
            readers[i] = std::jthread(&WritersReadersProblem::reader, this, std::format("Reader {}", i));
        }
    }
};

/*
Many threads try to acquire message from the queue, one continously sends messages to the queue.

bare try_acquire checks immediately if sem can be incremented - returns true if it was
incremented successfully otherwise returns false - nonblocking command.
try_acquire_for is the same as try_acquire but blocks at maximum for the given period of time - returns true
immediately if thread managed to decrement sem during this period of time, otherwise returns false.
try_acquire_until is the same as try_acquire_for but at maximum it blocks until it reaches some point in time.
*/
template<size_t numRecv>
class MessageReception {
    std::queue<std::string> messageQueue;
    std::binary_semaphore sem;

    std::chrono::high_resolution_clock::time_point start;
    std::chrono::seconds duration;

    void send() {
        std::chrono::milliseconds t = 750ms;    // here define timestamp if using try_acquire_until()
        while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start) < duration) {
            // try_acquire_until(timestamp)
            // try_acquire()
            if (sem.try_acquire_for(t)) {
                if (!messageQueue.empty()) {
                    std::cout << std::format("Resource acquired in less than {}, message popped: {}", t, messageQueue.front()) << std::endl;
                    messageQueue.pop();
                }
                else {
                    std::cout << std::format("Resource acquired in less than {} -> Nothing to pop from the queue.", t) << std::endl;
                }
                sem.release();
            }
            else {
                // Do some other things;
                std::cout << "Other threads are adding to the queue -> Doing other things." << std::endl;
                std::this_thread::sleep_for(100ms); // job
            }
            std::this_thread::sleep_for(20ms);
        }
    }

    void receive(const std::string& message) {
        while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::high_resolution_clock::now() - start) < duration) {
            sem.acquire();
            std::this_thread::sleep_for(500ms); // job
            std::cout << message << std::endl;
            messageQueue.push(message);
            sem.release();
            std::this_thread::sleep_for(500ms); // job
        }
    }
public:
    MessageReception(std::chrono::seconds duration) : duration(duration), sem(1) {}

    void runSimulation() {
        start = std::chrono::high_resolution_clock::now();
        std::array<std::jthread, numRecv> receivers;
        for (size_t i = 0; i < numRecv; i++) {
            receivers[i] = std::jthread(&MessageReception::receive, this, std::format("Receiver {}.", i));
        }
        std::jthread sender(&MessageReception::send, this);
    }
};

/*
Example which shows the limitation of number of threads which pass through counting semaphore without blocking.
We have 3 variables in counters array. 6 threads want to increment one o the varaiables by 100000000 but only
4 threads at the time are allowed to increment values. After the thread finished incrementing it releases the sem
and let's one of the blocked thread do the same job.
*/
class Counter {
    std::queue<std::string> messageQueue;
    std::counting_semaphore<4> sem;

    const static size_t numResources = 3;
    const static size_t numThreads = 6;
    std::array<std::atomic<uint32_t>, numResources> counters;
    std::vector<std::jthread> threads;

    void receive(size_t id) {
        uint32_t number;
        sem.acquire();
        std::cout << std::format("ID = {} started.", id) << std::endl;
        for (size_t i = 0; i < 100000000; i++) {
            counters[id % numResources]++;
        }
        number = counters[id % numResources];
        std::cout << std::format("ID = {} finished with {}", id, number) << std::endl;
        sem.release();
    }
public:
    Counter() : sem(4), counters{}, threads(numThreads) {}

    void runSimulation() {
        for (size_t i = 0; i < numThreads; i++) {
            threads[i] = std::jthread(&Counter::receive, this, i);
        }
        threads.clear();
    }
};

int main() {
    Comparison basicExample;
    std::cout << "Simulation 1" << std::endl;
    basicExample.compare();

    WritersReadersProblem<3> program(5s);
    std::cout << "Simulation 2" << std::endl;
    program.runSimulation();

    MessageReception<3> msgRec(5s);
    std::cout << "Simulation 3" << std::endl;
    msgRec.runSimulation();

    std::cout << "Simulation 4" << std::endl;
    Counter cntr;
    cntr.runSimulation();

    // summary of most important features:
    // max() is an operation which returns the internal counter's maximum possible value. It is greater than or equal to n where:
    // std::counting_semaphore<n> sem{1};

    std::cout << std::endl << "Other commands:" << std::endl;

    std::binary_semaphore s1{ 0 };
    std::cout << s1.max() << std::endl;

    std::counting_semaphore<7> s2{ 3 };
    std::cout << s2.max() << std::endl;

    // initial value of semaphore belongs to <0, max()> .
    // release(stride) is an operation which increments the semaphore value by stride - cannot increment over max(). Stride equals 1 by default.
    // acquire() is an operation which decrements the semaphore (cannot decrement under 0 but blocks until sem is greater than 0 and can successfully decrement).
    // std::counting_semaphore and std::binary_semaphore aren't owned by the given thread as it is 
    // in std::mutex case. One thread can increment semaphore and other can decrement at the same time without
    // undefined behaviour.
}