TGB (Все сообщения пользователя)

// Это сложная задача, требующая использования атомарных операций (std::atomic) и барьеров памяти для реализации Lock-Free взаимодействия.
// Для выполнения условия динамического управления очередями и уведомления без мьютексов, мы будем использовать std::atomic_flag или
// std::atomic<bool> для сигнализации (busy-wait или wait/notify в C++20) и атомарные указатели.
// Ниже представлен концептуальный код, реализующий эту логику:
//    C++20   ----
#include <iostream>
#include <vector>
#include <atomic>
#include <thread>
#include <memory>
#include <optional>

// Шаблонная циклическая очередь (Lock-Free для одного писателя и нескольких читателей со своими индексами)
template<typename T, size_t Capacity>
class CircularQueue {
    T buffer[Capacity];
    std::atomic<size_t> head{0}; // Указатель записи

public:
    void push(T item) {
        size_t h = head.load(std::memory_order_relaxed);
        buffer[h % Capacity] = item;
        head.store(h + 1, std::memory_order_release);
    }

    bool try_pop(size_t& reader_idx, T& out_item, size_t& pending_count) {
        size_t current_head = head.load(std::memory_order_acquire);
        if (reader_idx < current_head) {
            out_item = buffer[reader_idx % Capacity];
            reader_idx++;
            pending_count = current_head - reader_idx;
            return true;
        }
        pending_count = 0;
        return false;
    }
};

// Объект доступа для каждого читающего потока
struct ReaderContext {
    std::atomic<uint64_t> bitmask{0}; // Битовая шкала непустых очередей
    std::vector<size_t> read_indices;  // Свои указатели чтения
    std::atomic<bool> data_ready{false};

    void update_mask(size_t queue_idx) {
        bitmask.fetch_or(1ULL << queue_idx, std::memory_order_release);
        data_ready.store(true, std::memory_order_release);
        data_ready.notify_all(); // C++20 Lock-free уведомление
    }
};

class QueueManager {
    CircularQueue<int, 1024> int_queue;
    CircularQueue<float, 1024> float_queue;
    std::vector<std::shared_ptr<ReaderContext>> readers;

public:
    void add_reader(std::shared_ptr<ReaderContext> ctx) {
        ctx->read_indices.resize(2, 0); // 0 для int, 1 для float
        readers.push_back(ctx);
    }

    void write_int(int val) {
        int_queue.push(val);
        for (auto& r : readers) r->update_mask(0);
    }

    void write_float(float val) {
        float_queue.push(val);
        for (auto& r : readers) r->update_mask(1);
    }

    void reader_thread_func(std::shared_ptr<ReaderContext> ctx, int id) {
        while (true) {
            // Ожидание данных (Lock-free wait)
            ctx->data_ready.wait(false, std::memory_order_acquire);
            
            uint64_t mask = ctx->bitmask.load(std::memory_order_acquire);
            for (size_t i = 0; i < 2; ++i) {
                if (mask & (1ULL << i)) {
                    size_t pending = 0;
                    if (i == 0) {
                        int val;
                        while (int_queue.try_pop(ctx->read_indices[0], val, pending)) {
                            printf("[Reader %d] Int: %d (Pending: %zu)\n", id, val, pending);
                        }
                    } else {
                        float val;
                        while (float_queue.try_pop(ctx->read_indices[1], val, pending)) {
                            printf("[Reader %d] Float: %.2f (Pending: %zu)\n", id, val, pending);
                        }
                    }
                    // Сбрасываем бит, если данных больше нет
                    if (pending == 0) ctx->bitmask.fetch_and(~(1ULL << i));
                }
            }
            ctx->data_ready.store(false, std::memory_order_relaxed);
        }
    }
};

int main() {
    QueueManager manager;
    auto r1 = std::make_shared<ReaderContext>();
    auto r2 = std::make_shared<ReaderContext>();

    manager.add_reader(r1);
    manager.add_reader(r2);

    std::thread t1(&QueueManager::reader_thread_func, &manager, r1, 1);
    std::thread t2(&QueueManager::reader_thread_func, &manager, r2, 2);

    // Пишущий поток
    std::thread writer([&]() {
        for (int i = 0; i < 5; ++i) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            manager.write_int(100 + i);
            manager.write_float(1.5f * i);
        }
    });

    writer.join();
    t1.detach(); 
    t2.detach();

    return 0;
}
// Ключевые моменты реализации :
// Индивидуальные индексы : Каждый читатель хранит свои read_indices, что позволяет им двигаться по одной и 
// той же очереди независимо.
// Битовая шкала : Используется std::atomic<uint64_t>, где каждый бит соответствует ID очереди.Это позволяет 
// мгновенно проверить, где появились данные.
// Lock - Free Notify : Использован метод atomic::wait / notify_all(доступен с C++20).Он работает эффективнее 
// обычного yield, так как позволяет потоку "заснуть" без использования мьютекса в пользовательском 
// коде(на уровне ОС это может использовать фьютексы).
// Pending Count : Метод try_pop возвращает разницу между указателем записи и текущим индексом чтения, 
// выполняя условие получения "размера непрочитанных данных".

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

template <typename T, size_t Size>
class CircularQueue {
public:
    void push(T item) {
        buffer[head % Size] = item;
        head.fetch_add(1, std::memory_order_release);
    }

    // Каждый читатель передает свой локальный индекс
    bool try_read(size_t& reader_idx, T& out_item, size_t& pending_count) {
        size_t current_head = head.load(std::memory_order_acquire);
        if (reader_idx < current_head) {
            out_item = buffer[reader_idx % Size];
            reader_idx++;
            pending_count = current_head - reader_idx;
            return true;
        }
        pending_count = 0;
        return false;
    }

private:
    T buffer[Size];
    std::atomic<size_t> head{0};
};

// Контейнер для очередей разных типов
struct SharedData {
    CircularQueue<int, 100> int_queue;
    CircularQueue<double, 100> double_queue;
    std::mutex mtx;
    std::condition_variable cv;
    bool stop = false;
};

void writer(SharedData& data) {
    for (int i = 1; i <= 5; ++i) {
        {
            std::lock_guard<std::mutex> lock(data.mtx);
            data.int_queue.push(i * 10);
            data.double_queue.push(i * 1.1);
            std::cout << "[Writer] Pushed data set " << i << std::endl;
        }
        data.cv.notify_all(); // Активируем всех читателей
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    data.stop = true;
    data.cv.notify_all();
}

void reader(int id, SharedData& data) {
    size_t my_int_ptr = 0;
    size_t my_dbl_ptr = 0;
    
    while (true) {
        std::unique_lock<std::mutex> lock(data.mtx);
        data.cv.wait(lock, [&] { return data.stop || true; }); // Упрощенно: просыпаемся по сигналу

        int val;
        double dval;
        size_t pending;

        // Читаем из инт-очереди
        while (data.int_queue.try_read(my_int_ptr, val, pending)) {
            std::cout << "Reader " << id << " got INT: " << val << " (Left: " << pending << ")\n";
        }
        // Читаем из дабл-очереди
        while (data.double_queue.try_read(my_dbl_ptr, dval, pending)) {
            std::cout << "Reader " << id << " got DBL: " << dval << " (Left: " << pending << ")\n";
        }

        if (data.stop) break;
    }
}

int main() {
    SharedData data;
    std::thread w(writer, std::ref(data));
    std::thread r1(reader, 1, std::ref(data));
    std::thread r2(reader, 2, std::ref(data));

    w.join();
    r1.join();
    r2.join();
    return 0;
}

// Основные моменты реализации (от ИИ):
// Индивидуальные указатели: Переменные my_int_ptr и my_dbl_ptr находятся в стеке каждого читающего потока. Они отслеживают прогресс конкретного 
// читателя относительно общей «головы» (head) очереди.
// Шаблон очереди: Позволяет создавать экземпляры для int, double или любых других типов в одном контейнере SharedData.
// Синхронизация: std::condition_variable используется для мгновенной активации читателей пишущим потоком.
// Статус очереди: Метод try_read возвращает pending_count, что позволяет читателю знать, сколько данных еще осталось в буфере именно для него.

-- В самом начале вашего текста скрипта ---
is_Callback = false   --- Коллбеки вас не беспокоят --
 --- Во всех ваших коллбеках первая строка:
if not is_Callback then return end

--- В любом месте вашего скрипта.
--  Включить обработку коллбеков:
is_Callback = true
--  Выключить обработку коллбеков:
is_Callback = false
 

function OnInit_VPM()
    -- Логирование (ваш код)
    Logger.init(log_file)
    Logger.log("INIT", "", "Starting robot")
  ----    Весь остальной ваш код ---


end

function main()
      --- Инициализация (в начале main)--
      OnInit_VPM()
     
      --- Дальше ваш код ---

end

  <Определение функций обработки событий ровно так, как это сейчас>

---  Весь код скрипта без изменений, в том числе sleep(<Интервал>) --
-- Отличие только в реализации функции sleep.
-- При  ее вызове с заданным интервалом внутри нее выполняется (код на C++):
    WaitForSingleObject(<ОбъектОжидания>, <Интервал>); // здесь ожидание истечения <Интервала> или выдачи сигнала QUIK (неблокирующей функцией Pulse в служебном потоке) о записи в очереди новых событий.
  -- Когда sleep "срывается" с <ОбъектаОжидания>, то в ней анализируются очереди (это  можно сделать эффективно, используя битовую шкалу непустых очередей скрипта) и выполняются соответствующие <Функций обработки событий> с параметрами считанными из очередей.
  -- Функционально это не отличается от того, что реализовано сейчас, но выполняется в потоке пользователя main.
-- Тому, кому интересны детали выполнения sleep, может изменить в своем скрипте формат вызова sleep: вместо sleep(<Интервал>) строка
     local <Количество обработанных элементов очереди>, <Данные о потерях событий в очередях>  = sleep(<Интервал>)