flecs - Concurrent Systems

main.cpp

#include <flecs.h>
#include <iostream>

#define THREAD_COUNT (8)
#define ENTITY_COUNT (THREAD_COUNT * 1000000)

#define ENABLE_LOG 0

struct Position {
    double x, y;
};

struct Velocity {
    double x, y;
};

struct Data {
    int data;
};

double random(double min, double max) {
    double r = (double)rand() / RAND_MAX;
    return min + r * (max - min);
}

int main(int, char *[]) {
    flecs::world ecs;
    ecs.set_threads(THREAD_COUNT);

    // An entity that can be used in a pair to identify data by thread
    flecs::entity thread_entities[THREAD_COUNT];
    for (size_t t = 0; t < THREAD_COUNT; t++)
        thread_entities[t] = ecs.entity();

    // Generate random component data
    Position* p = new Position[ENTITY_COUNT];
    Velocity* v = new Velocity[ENTITY_COUNT];
    Data** d = new Data*[THREAD_COUNT];
    for (size_t t = 0; t < THREAD_COUNT; t++)
    {
        d[t] = new Data[ENTITY_COUNT];
    }
    for (size_t i = 0; i < ENTITY_COUNT; i++)
    {
        p[i] = {random(-100, 100), random(-100, 100)};
        v[i] = {random(-1, 1), random(-1, 1)};
        for (size_t t = 0; t < THREAD_COUNT; t++)
        {
            d[t][i] = {rand()};
        }
    }

    // Create lots of entities
    ecs_bulk_desc_t desc{};
    desc.count = ENTITY_COUNT;
    desc.ids[0] = ecs.id<Position>();
    desc.ids[1] = ecs.id<Velocity>();
    void* data[2 + THREAD_COUNT]{ p, v };
    for (size_t t = 0; t < THREAD_COUNT; t++)
    {
        desc.ids[2 + t] = ecs_pair(ecs.id<Data>(), thread_entities[t]);
        data[2 + t] = d[t];
    }
    desc.data = data;
    ecs_bulk_init(ecs, &desc);

    // Clean up data used to initialize entities
    delete p;
    delete v;
    for (size_t t = 0; t < THREAD_COUNT; t++)
    {
        delete d[t];
    }
    delete d;

    // Create a multi-threaded system that matches Position, Velocity
    // This will process an equal amount of entities on each thread in parallel
    ecs.system<Position, const Velocity>()
        .multi_threaded()
        .each([](flecs::entity e, Position& p, const Velocity& v) {
#if ENABLE_LOG
            std::cout << "Parallel - Stage ID:" << e.world().get_stage_id() << ", Entity: " << e << std::endl;
#endif
            p.x += v.x;
            p.y += v.y;
        });

    // Create systems that will be executed on each worker thread concurrently
    flecs::system systems[THREAD_COUNT];
    for (size_t t = 0; t < THREAD_COUNT; t++)
    {
        systems[t] = ecs.system<Data>()
            .kind(0)
            .term_at(1).second(thread_entities[t])
            .each([t](flecs::entity e, Data& data) {
#if ENABLE_LOG
                std::cout << "Concurrent " << t << " - Stage ID:" << e.world().get_stage_id() << ", Entity: " << e << std::endl;
#endif
                data.data++;
             });
    }

    // Create a system that will execute on each thread in parallel
    // This system then executes different systems concurrently on each thread based on its stage id
    ecs.system<>()
        .multi_threaded()
        .iter([&systems](flecs::iter& it) {
            auto ecs = it.world();
#if ENABLE_LOG
            std::cout << "Concurrent Runner - Stage ID:" << ecs.get_stage_id() << std::endl;
#endif
            flecs::system_runner_fluent(ecs, systems[ecs.get_stage_id()], 0, 0, 0.0, nullptr);
        });

    ecs.progress();

    return 0;
}