Compiler Explorer

Source code

#include "co_manager.hpp"
#include "co_task.hpp"
#include "co_chrono.hpp"
 
#include "co_event.hpp"
#include "co_queue.hpp"
#include "co_mutex.hpp"
#include "co_timer.hpp"

int counter {0};
bool finish{false};

Coro task_1 () {

// инстанцируем объекты синхронизаци
    mutex_t mutex;
    task2_event_t event;
    task1_timer_t timer;

while(true){

// пробуем захватить мьютекс
        // если неуспешно - приостанавливаемся и ждем
        // сигнала об освобождении
        co_await mutex.get<CoPrio::low>();
        
        // после захвата мьютекса возобновляемся
        puts(__func__);

// освобождаем мьютекс
        mutex.give();

// заводим таймер на 1000мс и приостанавливаемся
        co_await timer.get<CoPrio::high>(1000);

// посылаем эвент для task_2
        event.give();
    }
}

Coro task_2 (){

task2_event_t event;
    task3_queue_t queue;

while(true){

// ожидаем эвент от task_1
        co_await event.get<CoPrio::normal>();

puts(__func__);

// создаем переменную для очереди
        co_payload_t pld{ base_t{10} };

// помещаем ее в очередь для использования в task_3
        queue.give(pld);
    }
}

Coro task_3 (){

task3_queue_t queue;

while(true){

// ожидаем сигнала от очереди
        co_await queue.get<CoPrio::normal>();

puts(__func__);

// выгружаем очередь
        do {
            counter += queue.unload().base_val();
        } while ( queue.ready() );

finish = true;
    }
}

int main(){
    
    mutex_t mutex;
    // захватываем мьютекс
    mutex.get<CoPrio::low>();

// запускаем задачи
	task_1();
	task_2();
	task_3();

// запускаем диспетчер, убеждаемся, что пока
    // мьютекс захвачен, задачи приостановлены
    CoManager::run();

// освобждаем мьютекс
    mutex.give();
	
    // выполняем диспетчер и службу таймеров
    // пока задача task_3 не выставит флаг finish в true
    while(not finish) {
        CoChrono::check_if_expired();
        CoManager::run();
    }

// проверяем значение, переданное через очередь
    printf("%d\n", counter);
    // выводим объем аллоцированной под задачи памяти
    printf("memory allocated: %lu\n", CoAlloc::check_memory());

return 0;
}

Source code

#ifndef INC_CO_MUTEX_HPP_
#define INC_CO_MUTEX_HPP_

#pragma once

#include "critical_section.hpp"
#include "co_proxy.hpp"

template<typename CS>
class CoMutexImpl{

public:

CoMutexRet get_mutex() noexcept{

puts(__func__);

CS critical_section;

bool taking_success = !mutex;

if (taking_success) mutex = true;

return {&mutex, taking_success};
    }

void give_mutex() noexcept{

puts(__func__);

mutex = false;
    }

private:

bool mutex{false};
};

template<co_act_t A>
class CoMutex: public CoProxy<CoMutex<A>>{

public:

constexpr co_act_t give_impl() const noexcept{

instance().give_mutex();

return A;
    };

template<CoPrio P>
    constexpr co_proxy_t<A, P> get_impl() const noexcept{

auto[m_ptr, ts] = instance().get_mutex();

return {.mutex_ptr = m_ptr, .taking_success = ts,};
    }

friend class CoProxy<CoMutex<A>>;

private:

using mutex_impl_t = CoMutexImpl<co_critical_t>;

[[gnu::always_inline]] mutex_impl_t& instance() const noexcept{
        static mutex_impl_t mutex_impl;
        return mutex_impl;
    }
};

#define CO_MUTEX(n)   using n = CoProxy<CoMutex<__COUNTER__>>

/*  USER SECTION START  */
CO_MUTEX(mutex_t);
/*  USER SECTION END  */

#endif /* INC_CO_MUTEX_HPP_ */

Source code

#ifndef CO_PROXY_HPP_
#define CO_PROXY_HPP_

#pragma once

#include "co_util.hpp"
#include "co_manager.hpp"

template<typename T>
class CoProxy : public CoManager {

public:

using derived_ptr = T*;

template<typename ...Args>
	constexpr void give(Args&& ...args) noexcept {

co_act_t action = 
            derived()->give_impl(co_detail::forward<Args>(args) ...);

set_action(action);
	}

template<CoPrio P, typename ...Args>
    constexpr auto get (Args&& ...args) noexcept {
        
        return derived()->template get_impl<P>(co_detail::forward<Args>(args) ...);
	}

constexpr bool ready() noexcept {
       
        return derived()->ready_impl();
	}

constexpr auto unload() noexcept {

return derived()->unload_impl();
	}

private:

constexpr derived_ptr derived() noexcept{
        
		return static_cast<derived_ptr>(this);
	}
};

template<co_act_t ID, CoPrio P>
struct CoProxyRet final : public CoManager {

bool* mutex_ptr{nullptr};
    bool taking_success{false};
};

template<co_act_t ID, CoPrio P>
using co_proxy_t = CoProxyRet<ID, P>;

#endif /* CO_PROXY_HPP_ */

Source code

#ifndef INC_CO_TASK_HPP_
#define INC_CO_TASK_HPP_

#pragma once

#include <coroutine>
#include <limits>

#include "co_proxy.hpp"
#include "co_alloc.hpp"

struct Coro {

using promise_type = Coro;

CoSync sync {
        .mutex{.mutex_ptr = nullptr, .taking_success = false},
        .address{nullptr},
	    .state{CoState::stopped},
        .prio{CoPrio::lowest},
        .id{ indexer_t{}() },
        .size{0},
        .expected{std::numeric_limits<co_act_t>::max()},
    };

auto get_return_object() const noexcept {
        puts(__func__);

return Coro{};
	}

std::suspend_never initial_suspend() noexcept {
        puts(__func__);
            
        sync.state = CoState::running;
        sync.size = CoAlloc::get_current_size();

return {};
    }

std::suspend_never final_suspend() const noexcept { 
        puts(__func__);
        return {};
    }

void return_void() const noexcept { puts(__func__); }

void unhandled_exception() {
        puts(__func__);
        std::abort();
    }

template<co_act_t ID, CoPrio P>
    auto yield_value ( co_proxy_t<ID, P> p) const noexcept {
        puts(__func__);

struct Awaitable{

co_proxy_t<ID, P> proxy;

bool await_ready () const noexcept {

if (not proxy.mutex_ptr) return false;

return proxy.taking_success;
            }

void await_suspend (std::coroutine_handle<promise_type> coro) noexcept {
                puts(__func__);

co_sync_t sync = &coro.promise().sync;

sync->address = coro.address();

if ( sync->mutex.mutex_ptr )
                    *( (sync->mutex).mutex_ptr ) = false;

sync->mutex = {
                    .mutex_ptr = proxy.mutex_ptr,
                    .taking_success = proxy.taking_success
                };
                sync->state = CoState::suspended;
                sync->prio = P;
                sync->expected = ID;

decltype(proxy)::store_sync(sync);
            }

auto await_resume () noexcept {

puts(__func__);

return 0;
			}
		};
            
		return Awaitable{p};
	}

template<co_act_t ID, CoPrio P>
    auto await_transform(co_proxy_t<ID, P> p) const noexcept {
        return yield_value<ID, P>(p);
    }

void* operator new(std::size_t sz){
        puts(__func__);
        return CoAlloc::allocate(sz);
	}

void operator delete( void* p){
        puts(__func__);
        CoAlloc::deallocate(p);
	}
};

#endif /* INC_CO_TASK_HPP_ */

Source code

#include <coroutine>
#include <limits>

#include "critical_section.hpp"
#include "co_queue_impl.hpp"
#include "co_manager.hpp"

static bool mutex_take (co_sync_t sync);
static void co_resume (co_sync_t sync);

using sync_queue_t = CoQueueImpl<co_sync_t, CoParam::CORO_TASK_NUM, co_critical_t>;

co_sync_t co_repo[CoParam::CORO_TASK_NUM];
sync_queue_t co_queue_repo[CoPrio::num];

co_sync_t current{nullptr};
base_t current_memory;

void CoManager::set_action(co_act_t act) noexcept{

for (auto sync : co_repo){

if (nullptr == sync) continue;

if(act == sync->expected)
            co_queue_repo[sync->prio].push(sync);
    }
};

void CoManager::store_sync(co_sync_t s) noexcept{

co_assert(s->id < CoParam::CORO_TASK_NUM, __func__);

co_repo[s->id] = s;

current_memory += s->size;
}

void CoManager::run(){

for (auto& queue : co_queue_repo){

while ( not queue.is_empty() ) {

co_sync_t sync = queue.pop();

if ( nullptr == current ||
                 current->state != CoState::running  ){

current = sync;
            
                co_resume(current);

} else if (CoState::running == current->state   &&
                        sync->prio < current->prio          ){
            
                current->state = CoState::blocked;

co_sync_t preemted = current;

current = sync;

co_resume(current);

preemted->state = CoState::running;

current = preemted;
            }
        }
    }
}

static void co_resume (co_sync_t sync){

if( not mutex_take(sync) ) return; 
    
    sync->state = CoState::running;

sync->expected = std::numeric_limits<co_act_t>::max();

co_detail::enable_irq();

std::coroutine_handle<>::from_address(sync->address).resume();

co_detail::disable_irq();
}

static bool mutex_take (co_sync_t sync){
    
    bool *mutex_ptr = sync->mutex.mutex_ptr;

if ( mutex_ptr && (*mutex_ptr) ) return false;

if (mutex_ptr) *mutex_ptr = true;

return true;
}

Source code

#ifndef INC_CO_QUEUE_HPP_
#define INC_CO_QUEUE_HPP_

#pragma once

#include "co_proxy.hpp"
#include "co_variant.hpp"
#include "co_queue_impl.hpp"

/*  USER SECTION START  */
using co_payload_t = CoVariant;

/*  USER SECTION END  */

template<co_act_t A>
class CoQueue: public CoProxy<CoQueue<A>>{

public:

using co_queue_t = CoQueueImpl<co_payload_t, CoParam::CORO_QUEUE_SIZE, co_critical_t>;

constexpr co_act_t give_impl(const co_payload_t& payload) const noexcept{

instance().push(payload);

return A;
    }

template<CoPrio P>
    constexpr co_proxy_t<A, P> get_impl() const noexcept{
        return {};
    }

constexpr bool ready_impl() const noexcept{
         return !instance().is_empty();
    }

constexpr co_payload_t unload_impl() const noexcept {
        return instance().pop();
    }

friend class CoProxy<CoQueue<A>>;

private:

[[gnu::always_inline]] co_queue_t& instance() const noexcept{
        static  co_queue_t queue_impl;
        return queue_impl;
    }
};

#define CO_QUEUE(q)   using q = CoProxy<CoQueue<__COUNTER__>>

/*  USER SECTION START  */

CO_QUEUE(task3_queue_t);

/*  USER SECTION end  */

#endif /* INC_CO_QUEUE_HPP_ */