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

 
run = true 

function main()
   local function GetMilliseconds()  -- Милисекунды с 1970г.
      local dt = os.sysdate()
      return os.time(dt) * 1000 + dt.ms
   end
   --  Коды инициализации скрипта --

   local pause = 20 
   local TT = GetMilliseconds()
  
   -- Основной цикл скрипта (while run do)--
   while run do
      --  Выполнение пользовательского кода --
      
      -- Адаптивное ожидание (динамически зависящее от длительности пользовательского цикла обработки)
      local tt = GetMilliseconds() - TT  -- Длительность пользовательского цикла обработки
      tt = tt >= pause and 1 or pause - tt 
      sleep(tt)
      TT = GetMilliseconds()  -- Начало обработки пользовательского кода
   end

end

message('Вариант 1: "qty,price,datetime.sec,datetime.mcs" ======')
SearchItems("orders", 0, getNumberOf("orders") - 1,
function(qty, price, sec, mcs)
    message(tostring(sec))
    message(tostring(mcs))
    message(tostring(qty))
    message(tostring(price))
    return false
end,
"qty,price,datetime.sec,datetime.mcs")
------------- Результат Вариант 1 --
--  Вариант 1: "qty,price,datetime.sec,datetime.mcs" ======
--  0
--  nil
--  14.0
--  21.905
-------------------------------------

message('Вариант 2: "datetime.sec, datetime.mcs, qty, price" ======')
SearchItems("orders", 0, getNumberOf("orders") - 1,
function(sec, mcs, qty, price)
    message(tostring(sec))
    message(tostring(mcs))
    message(tostring(qty))
    message(tostring(price))
    return false
end,
"datetime.sec, datetime.mcs, qty, price")
------------- Результат Вариант 1 
--  Вариант 2: "datetime.sec, datetime.mcs, qty, price" ======
--  0
--  14.0
--  21.905
--  nil
------

local ord = getItem('orders', 0)
message('ord.datetime.mcs : ' .. tostring(ord.datetime.mcs))
--  ord.datetime.mcs : 0
----------------------------

if not collectgarbage('isrunning') then
      message('Где то был остановлен сбор мусора (collectgarbage("stop")). Разобраться где. ', 3)
end

//  ==============  Вариант реализации очередей событий QUIK (Lock-free, ~300 строк) ============
    //   Параметр Q_MAX определяет количество очередей в схеме очередей. Схем обработки очередей можно 
    // создать несколько. 
    //   Данные, передаваемые в информационных очередях, строки.
    // 1. Формат записи данных - текст описания таблицы Lua: 
    //     {name = <Имя коллбека>, tbl = {<Таблица для формирования вызова функции пользователя>}}
    //    После чтения выполнять десереализацию текста таблицы Lua и дальше обрабатывать с учетом значения name.
    //   name обеспечивает возможность группировка коллбеков по очередям.
    // 2.  Коллбеки можно, но необязательно, сгруппировать и группы распределить по очередям с учетом того,
    //   что при чтении обработка начинается с 0-й очереди. Параметры очередей size_max следует задать 
    //   такими, чтобы в них помещались строки их данных.
    //     Для коллбеков пользовательских таблиц QUIK, наверное, имеет смысл использовать все таки отдельную 
    //   очередь.
    // 3. Параметр FILLING_THRESHOLD - порог заполненности очередей для выдачи предупреждения о высоком уровне 
    // заполненности очередей.
    // 4. После первого цикла любой очереди, память ее созданных элементов переиспользуется
    //  (кроме чтения указателя нет затрат на управление ее памятью).
    // 5. Если для читающего потока в какой то его очереди возникнет переполнение, то об этом
    // выдается сообщение.
    // 6. Реализованы режимы записи в очереди: 0 - без сигналов (более эффективный); 1 - с сигналами.
    // ----------------------------------------------------------------------
  //                 Результат теста (оценка эффективности реализации очередей)
  // Режим записи в очереди: без сигналов.
  // Писатель выполняет в цикле запись в 10 очередей. Циклов записи : 1000
  //  T - время выполнения потоков с учетом их пауз в млс.
  //-->Писатель. Пауза:  1.  T (млс.): 1977. Обработано записей: 10000. Время ЦП (млс.): 3.4494
  //Читатель 3. Пауза:  50.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.3174
  //Читатель 1. Пауза:   5.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.1746
  //Читатель 4. Пауза: 100.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.2129
  //Читатель 2. Пауза:  10.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.2194
  // -------------------------------------------------------------------------------

//                              Краткое ТЗ на разработку очередей 
// 1) Создаются несколько циклических информационных очередей с разными типами данных и общим объектом 
// управления ими.
// 2) В очереди пишет данные один поток, а читают эти очереди несколько потоков.
// 3) Дополнительно создается специальная служебная циклическая очередь: битовые шкалы непустых информационных 
// очередей, записываемых пишущим потоком, после записи в информационные очереди.
// 4) У читающих потоков должны быть свои указатели чтения циклических очередей и их локальные битовые шкалы 
// непустых очередей, хранящиеся в их объекте доступа к информационным очередям.
// 5) Читающие потоки циклически, выполнив свои коды переходят в состояние ожидания сигнала появления данных 
// в их очередях или истечения заданного интервала времени на их общем объекте ожидания.
// 6) При записи в любую очередь пишущий поток после записи в информационные очереди записывает битовые шкалы 
// непустых очередей в служебную очередь шкал непустых очередей и [активирует в режиме 1] все читающие потоки.
// 7) Каждый читающий поток, при пробуждении, читает все появившиеся записи в служебной очереди шкал непустых 
// очередей и формирует свою локальную шкалу непустых очередей и далее, на ее основании, читает без 
// синхронизации, непустые очереди с признаком : состояние прочитанных очередей в виде размера непрочитанных 
// данных.
// 8) Реализовать эффективное управление памятью при передаче данных в этих очередях.

//                       C++20 (используется std::countr_zero) ---
#include <shlobj.h>
#include <windows.h>>
#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <mutex>
   //#include <condition_variable>
   //#include <atomic>
   //#include <bitset>
   //#include <intrin.h>
#pragma comment(lib, "winmm.lib")    // Установить точность таймера 1 мс

// Параметры тестирования очередей ---
const int N_MAX = 2000;       // Количество циклов записи при тестировании
//#define win32_API           // В реализации очередей используется win32_API
const int QUEUE_RECORDING_MODE = 0; // Режим записи в очереди: 0 - без сигналов; 1 - с сигналами
const INT64 Q_SIZE = 64;      // Длина информационной очереди по умолчанию 
const INT64 Q_SIZE_SL = 256;  // Длина служебной очереди шкал непустых информационных очередей
const int Q_MAX = 128;        // Максимальное количество очередей
const int Количество_очередей = 120; // Количество используемых очередей ( <= Q_MAX)
const int LEN_STR_MAX = 4096; // Размер буфера потоков чтения (максимум size_max_str всех используемых очередей) 
const double FILLING_THRESHOLD = 0.4; // Порог заполненности очередей для выдачи предупреждения об уровне заполненности очередей

// Системное время (синхронизированное). Точность 0,1 мкс.---
static INT64 T_OS_high_mls()
{
   FILETIME lpSystemTimeAsFileTime;
   GetSystemTimePreciseAsFileTime(&lpSystemTimeAsFileTime);  // Системное время (синхронизированное)
   INT64 tt = lpSystemTimeAsFileTime.dwHighDateTime;
   tt = (tt << 32) + lpSystemTimeAsFileTime.dwLowDateTime;
   return tt;
}

// Строка-объект информационной очереди (переиспользуемая) --
struct q_str {
   INT64 tail = 0;      // Указатель записи ---
   INT64 size_max = 0;  // Максимальный размер памяти строки (с учетом символа конца строки '\0')
   INT64 size = 0;      // Текущий размер памяти строки (с учетом символа конца строки '\0')
   int n_q = 0;         // Номер очереди  
   char* str = NULL;    // Память строки (размер size_max) ----
   int state = 0;       // Состояние (не используется, но на всякий случай): 1 - занята пишущим потоком; 2 - в очереди.
};

// Параметры очереди, задаваемые при ее инициализации --
struct q_parm {
   std::string name_Queue = "";     // имя очереди
   INT64 q_size = 64;               // длина очереди
   INT64 size_max_str = 64;         // максимальная память строки в информационной очереди
};

// Шаблонная циклическая очередь (Lock-free)
// Используется для служебной и информационных очередей  
template<typename T>
class InfoQueue {
public:
   std::vector<T> buffer;
   INT64 tail = 0;                     // пишет только один писатель
   std::string name_Queue = "";        // имя очереди (можно менять при начальной инициализации)
   INT64 q_size = 64;                  // размер очереди (можно менять при начальной инициализации)
   INT64 size_max_str = 64;            // максимальная память строки в очереди (можно менять при начальной инициализации)
   InfoQueue(INT64 s = Q_SIZE) : buffer(s), q_size(s) {}

   // Изменение длины очереди ---
   void q_size_set(int s) { 
      buffer.reserve(s); q_size = s;
   }

   // запись в очередь --
   bool push(T val) {
      buffer[tail++ % q_size] = val;
      return true;
   }

   // Количества доступных элементов очереди для конкретного читателя
   // local_head - указатель чтения (читающего потока)
   // Если отрицательное значение, то: количество пропущенных при чтении записей очереди --
   INT64 get_pending_count(INT64 local_head) const {
      INT64 t = tail - local_head;
      return (t <= q_size ? t : q_size - t);
   }
};

// Копировать строку-объект в буфер читающего потока для последующей обработки ---
void q_str_copy(q_str* s, q_str* s1, InfoQueue<q_str*>* q)
{
   //s1->size_max = s->size_max;   // ! Задана при создании буфере --
   s1->size = s->size;
   s1->n_q = s->n_q;
   if (s1->str == NULL) throw std::runtime_error("*** Ошибка: нет памяти для копирования");
   //  s1->size - размер памяти строки (с учетом символа конца строки '\0')
   //  Длина строки: (s1->size - 1)  
   memcpy(s1->str, s->str, s1->size);  s1->tail = s->tail;   // ! Обязательно в таком порядке --
   s1->state = s->state;
   // Дополнительный контроль переполнение очереди (после копирования в буфер потока чтения)
   if ((q->tail - s1->tail) > q->q_size) std::cout << " *** Ошибка переполнение очереди (при копировании) " << s1->n_q << "\n";
}

// Объявления (определяются позже)
struct ReaderAccess;
struct Manager;
//-------------------------------
// Информационные очереди событий --
struct DataCluster {
   //  Деструктор (! выход из блока кода, в котором определен DataCluster 
   // только при завершения потоков, использующих его) --
   ~DataCluster() { 
      for (int i = 0; i < Q_MAX; ++i) {
         INT64 size = q[i].buffer.size();
         INT64 N = (q[i].tail >= size) ? size : q[i].tail;  // Буфер может быть неполным ---
         for (int j = 0; j < N; ++j) {
            q_str* q_s = q[i].buffer[1]; 
            delete[]q_s->str;
            delete q_s;
         }
      }
   }

   int count_q = 64;  // Количество используемых информационных очередей (<= Q_MAX) --
   // При количестве очередей > 64 один бит шкалы непустых очередей соотносится 
   // к нескольким ((count_q - 1) / 64 + 1) последовательным очередям, в которых, 
   // возможно, есть записи.
   UINT32 Queue_grouping = 1; 
   InfoQueue<q_str*> q[Q_MAX];

   // Задать количество используемых очередей и их параметры
   void set_parm_q(int n, q_parm* parm) {
      if (n > 0 && n <= Q_MAX)
         count_q = n; 
      else 
         throw std::runtime_error(" Количество используемых очередей > Q_MAX или <= 0");   /* #### ошибка*/
      Queue_grouping = (count_q - 1) / 64 + 1;
      if (parm != NULL)
         for (int i = 0; i < count_q; ++i) {
            q[i].name_Queue = parm->name_Queue;
            q[i].q_size = parm->q_size;
            q[i].q_size_set(parm->q_size);
            q[i].size_max_str = parm->size_max_str;
         }
   }

   // Функция формирования маски группировки очередей (многозначное отображение очередей на шкалу непустых)
   uint64_t q_mask(int q_i) { return (uint64_t)1 << (q_i / Queue_grouping); }

   //  Запрос памяти потоком писателем под запись очереди --
   // Запрос памяти системы и формирование строки-объекта -----
   q_str* get_q_s(int q_i) {
      INT64 size_max_str = q[q_i].size_max_str;
      char* str = new char[size_max_str];
      str[size_max_str - 1] = '\0';
      q_str* q_s = new q_str;
      q_s->str = str;
      q_s->size_max = size_max_str;
      q_s->size = 0;
      q_s->n_q = q_i;
      return q_s;
   }

   // Запрос строки-объекта потоком записи для формирования строки 
   q_str* get_str(int q_i) {
      q_str* q_s;
      INT64 tail = q[q_i].tail;
      INT64 t = tail % q[q_i].q_size;
      if (q[q_i].buffer[t] != NULL) {   //  Получение памяти из очереди --
         q_s = q[q_i].buffer[t]; 
         q_s->tail = tail;
      }
      else {   // Запрос памяти у системы --
         q_s = get_q_s(q_i);
         q_s->tail = tail;
      }
      return q_s;
   }

   //  Инициализация используемых очередей строками-объектами --
   // #### Вряд ли стоит использовать ---
   void initialization_q_str() {
      if (count_q > 0) {
         for (int i = 0; i < count_q; ++i) {
            int size = q[i].q_size;
            for (int j = 0; j < size; ++j) {
               if (q[i].buffer[j] == NULL)
                  q[i].buffer[j] = get_q_s(i);
            }
         }
      }
   }

   // Начальная инициализация указателей чтения в потоке-читателе на основе указателей 
   // записи в очереди (при начальной инициализации потока)
   // Объявление.
   void init_access(ReaderAccess* access, Manager* mgr, DataCluster* dc);

   // запись в очередь --
   bool push(int q_n, q_str* val) {
      if (q_n < 0 || q_n >= count_q) throw std::runtime_error("*** Ошибка: нет такой очереди");
      q[q_n].buffer[q[q_n].tail++ % q[q_n].q_size] = val;
      return true;
   }
};

//struct ReaderAccess;   // Используется в Manager
#ifndef win32_API
// Объект управления потоками чтения 
struct Manager {
   INT64 signal_counter = 0;              // Счетчик сигналов записи
   int Выдавать_сигналы = 0;          // 0 -  без выдачи сигнала; 1 - выдача сигнала
   InfoQueue<uint64_t> service_q{ Q_SIZE_SL }; // Служебная очередь битовых шкал
   //--------- condition_variable --
   std::mutex mtx;                        // Для синхронизации  при ожидании             
   std::condition_variable cv;            // Общий объект ожидания 

   void pulse(uint64_t mask) {
      service_q.push(mask); ++signal_counter;  
      if (Выдавать_сигналы == 1)
         cv.notify_all(); 
   }

   void wait(ReaderAccess* access, int tt);  // !! Реализация после struct ReaderAccess (иначе недоступны поля)
};
#else // !win32_API
   // Вариант 2. #### Этот вариант работает, но предыдущий лучше --
   struct Manager {        
      INT64 signal_counter = 0;              // Счетчик сигналов записи
      int Выдавать_сигналы = 0;          // 0 -  без выдачи сигнала; 1 - выдача сигнала
      InfoQueue<uint64_t> service_q{ Q_SIZE_SL }; // Служебная очередь битовых шкал
      //--------- condition_variable --
      CRITICAL_SECTION cs;
      CONDITION_VARIABLE cv;
      Manager() {
         InitializeCriticalSection(&cs); 
         InitializeConditionVariable(&cv);
      }
      bool dataReady = false;

      void pulse(uint64_t mask) {
         service_q.push(mask);
         ++signal_counter;
         EnterCriticalSection(&cs);      //  #### ?
         if (Выдавать_сигналы == 1) {
            dataReady = true;
            WakeAllConditionVariable(&cv); // Будим тех, кто еще спит
         }
         LeaveCriticalSection(&cs);      //  #### ?
      }
   
      void wait(ReaderAccess* access, int tt);  // !! Реализация после struct ReaderAccess (иначе недоступны поля)
   };
#endif // !win32_API

// Объект доступа к очередям для читающих потоков --
struct ReaderAccess {
   INT64 h[Q_MAX] = { 0 };        // Локальные указатели чтения очередей
   INT64 h_service = 0;           // Указатель чтения служебной очереди
   uint64_t local_bitmask = 0;    // Локальная шкала непустых очередей потока-читателя
   INT64 signal_counter = 0;      // Локальный счетчик сигналов записи 
   //  -------- Статистика очередей ---------
   INT64 NN[Q_MAX] = { 0 };          // Коичество обращений к непустым очередям
   double statistics[Q_MAX] = { 0 }; // Сумма относительной заполненности очередей (среднее = statistics [i]/ (NN[i]))
   double q_filling_threshold = 0.4; // Порог заполненности очередей для выдачи предупреждения --
   int period_agr = 9;               // Период скользящей 
   //  ------- Данные снимка состояния очередей потока чтения ---
   int q_len[Q_MAX] = { 0 };   // Количество непустых непрочитанных записей в очередях
   int q_n[Q_MAX] = { 0 };     // Номера непустых очередей
   int q_p = 0;                //  Длина векторов q_n и q_len

   // Снимок состояния очередей  потока-читателя ("мгновенный", чтобы меньше было повторных чтений очередей).
   void snapshot_queue_state(DataCluster& dc, Manager& mgr) {
      q_p = 0;  // Сброс длины векторов q_n и q_len
      // Формирование шкалы непустых очередей потока-читателя и снимка состояния очередей потока-читателя
      INT64 count = mgr.service_q.get_pending_count(h_service);
      if (count > 0) {
         // Формируем общую локальную шкалу непустых очередей потока-читателя --
         local_bitmask = 0;
         uint64_t mask_val;
         for (int i = 0; i < count; ++i) {
            mask_val = mgr.service_q.buffer[h_service % mgr.service_q.q_size];
            local_bitmask |= mask_val; 
            ++h_service;
         }
         // Формирование снимка состояния очередей потока-читателя (на основе его шкалы) --
         int pos_old = 0;
         int Queue_grouping = dc.Queue_grouping;
         while (local_bitmask) {   //  Эффективная обработка шкалы непустых очередей --
            ////  ! Способ определения позиции младшего разряда для старых стандартов (MSVC)  -----
            // unsigned long pos3 = 0;
            // _BitScanForward64(&pos3, local_bitmask);  // позиция младшего разряда с 1
            ////  -------------------------------------------------
            int Сдвиг = std::countr_zero(local_bitmask); // C++20 (позиция младшего разряда с 1)
            int pos = Сдвиг + pos_old;
            pos_old = pos + 1;
            pos *= Queue_grouping;
            for (int i = 0; i < Queue_grouping; ++i) {  // обработка групп очередей разряда шкалы --
               INT64 q_count = dc.q[pos + i].get_pending_count(h[pos + i]);
               if (q_count > 0) {
                  int p = q_p;
                  q_n[p] = pos + i;
                  q_len[p] = q_count;
                  //  Статистика заполненности очереди (относительная, приблизительно средне скользящая) 
                  if (++NN[pos + i] < period_agr) {
                     statistics[pos + i] += (double)q_count / (double)dc.q[pos + i].q_size;
                  }
                  else {
                     statistics[pos + i] += (double)q_count / (double)dc.q[pos + i].q_size - statistics[pos + i] / period_agr;
                  }
                  // --------------------------------------------------------------------------------
                  ++q_p;
               }
               else
                  if (q_count < 0)  std::cout << " *** Ошибка переполнение очереди (при чтении) q = " << pos + i << ": q_count = " << q_count << "\n";  // #### Ошибка переполнение очереди
            }
            local_bitmask >>= Сдвиг + 1;
         }
      }
      else if (count < 0)  std::cout << " *** -- Ошибка переполнение служебной очереди шкал --\n";
      //std::cout << ".  Шкала : " << std::bitset<64>(local_bitmask) << "\n";
   }

   // Чтение строки-объекта очереди --
   // q_n - номер очереди
   q_str* get_q_str(int q_n, DataCluster& dc) {
      q_str* val = dc.q[q_n].buffer[h[q_n] % dc.q[q_n].q_size];
      ++h[q_n];
      return val;
   }
};

// ! Реализация члена DataCluster (расположена после struct ReaderAccess, Manager и DataCluster 
// чтобы в методе были доступны их поля)
// Начальная инициализация указателей чтения в потоке-читателе на основе указателей 
// записи в очереди (при начальной инициализации потока)
void DataCluster::init_access(ReaderAccess* access, Manager* mgr, DataCluster* dc) {
   int count_q = dc->count_q;
   if (count_q <= 0)  throw std::runtime_error("*** Ошибка: не задано количество используемых очередей");
   // Получение указателя служебной очереди
   access->h_service = mgr->service_q.tail;
   // Получение указателей информационных очередей 
   for (int i = 0; i < count_q; ++i) access->h[i] = q[i].tail;
}

// ! Реализация члена Manager (расположена после struct ReaderAccess чтобы в методе были 
// доступны ее поля)
// Пауза
#ifndef win32_API
void Manager::wait(ReaderAccess* access, int dt) {
   if (access->signal_counter == signal_counter) {// не было сигнала записи при отсутствии ожидания в потоке чтения
      std::unique_lock<std::mutex> lock(mtx); 
      cv.wait_for(lock, std::chrono::milliseconds(2)) == std::cv_status::timeout;
   }
   access->signal_counter = signal_counter;   // Обрабатываемый сигнал записи в потоке чтения
}
#else       // !win32_API
   // Вариант 2. #### -------
   void Manager::wait(ReaderAccess* access, int dt) {
      if (access->signal_counter == signal_counter) {// не было сигнала записи при отсутствии ожидания в потоке чтения
         EnterCriticalSection(&cs);
            // Пытаемся дождаться условия в течение заданного timeout
            if (!dataReady) {
               if (!SleepConditionVariableCS(&cv, &cs, 2)) {
                  if (GetLastError() == ERROR_TIMEOUT) {
                     //std::cout << "Thread [" << tt << "ms] TIMED OUT!" << std::endl;
                  }
               }
               else {
                  //std::cout << "Thread [" << tt << "ms] GOT SIGNAL!" << std::endl;
               }
               dataReady = false;
            }
         LeaveCriticalSection(&cs);
      }
      access->signal_counter = signal_counter;   // Обрабатываемый сигнал записи в потоке чтения
   }
#endif // !win32_API

//  Тестирование очередей событий  ------------------------------------------------
// Формат записываемых строк общий для всех очередей: тексты таблиц Lua (сереализация).
// Читающий поток выполняет десереализацию полученных строк и запуск Lua-функций обработки событий.
//    ----------------------------------------------------
//                          Функция потока-писателя
void producer(DataCluster& dc, Manager& mgr) {
   std::cout << "Писатель. Запись выполняется в 10 очередей. Циклов записи: " << N_MAX << std::endl;
   std::cout << "T - время выполнения потоков с учетом их пауз." << std::endl;
   // -------------------------------------
   INT64 tt = T_OS_high_mls();  // Для подсчета времени ЦП --
   INT64 TT = tt;               // Для замера времени выполнения  ---
   INT64 TCP = 0;               // Время ЦП  ---
   INT64 NRQ = 0;               // Количество обработанных записей
   int dt = 1;
   int q_n_test = 0;
   for (int i = 1; i <= N_MAX; ++i) {
      tt = T_OS_high_mls() - tt;
      TCP += tt;
      //std::this_thread::sleep_for(std::chrono::milliseconds(dt));   // Пауза потока записи
      Sleep(dt);   // И так можно 
      tt = T_OS_high_mls();
      //  %%%%%% Подготовка данных для записи в очереди в виде C-строк

      //   Запись строк в очереди --
      uint64_t mask = 0;  // шкала непустых очередей
      // Запись в разные очереди
      // !!  rec->size - размер памяти строки (с учетом символа конца строки '\0')
      //     Длина строки: (rec->size - 1) 
      const char* source = "{name = 'OnCleanUp',tbl = {  ....] }"; 
      int sz = strlen(source) + 1;
      q_n_test %= 2; 
      int q_n_test1 = q_n_test * 4;   // Для записи в разные очереди 
      q_str* rec = dc.get_str(0 + q_n_test1); memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(0 + q_n_test1, rec))  mask |= dc.q_mask(0 + q_n_test1);  ++NRQ;
      rec = dc.get_str(1 + q_n_test1);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(1 + q_n_test1, rec))  mask |= dc.q_mask(1 + q_n_test1);  ++NRQ;
      rec = dc.get_str(2 + q_n_test1);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(2 + q_n_test1, rec))  mask |= dc.q_mask(2 + q_n_test1);  ++NRQ;
      rec = dc.get_str(3);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(3 + q_n_test1, rec))  mask |= dc.q_mask(3 + q_n_test1);  ++NRQ;
      ++q_n_test;

      rec = dc.get_str(40);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(40, rec))  mask |= dc.q_mask(40);  ++NRQ;
      rec = dc.get_str(41);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(41, rec))  mask |= dc.q_mask(41);  ++NRQ;

      rec = dc.get_str(40);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(40, rec))  mask |= dc.q_mask(40);  ++NRQ;
      rec = dc.get_str(41);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(41, rec))  mask |= dc.q_mask(41);  ++NRQ;
      rec = dc.get_str(42);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(42, rec))  mask |= dc.q_mask(42);  ++NRQ;
      rec = dc.get_str(100); memcpy(rec->str, "шшшшшш", 7); rec->size = 7;
      if (i >= N_MAX) rec->state= -1;   // Признак читателям завершить поток ---
      if (dc.push(100, rec)) mask |= dc.q_mask(100);  ++NRQ;   // "шшшшшш"
      // std::cout << "[P] Данные записаны, маска: " << std::bitset<64>(mask) << std::endl;  // ####
      // Запись маски и активация потоков
      mgr.pulse(mask);
   }
   std::stringstream out;   // можно использовать <<
   out << "--->   Писатель. Пауза:" << dt
      << ". T (млс.):"
      << (int)((T_OS_high_mls() - TT) * 0.0001) << ". Обработано записей:" << NRQ
      << ". Время ЦП (млс.):" << TCP * 0.0001 << "\n";
   std::string result = out.str();    // Получаем std::string
   std::cout << result;
}

// Заглушка функции обработки событий --
//   Выполняет десереализацию полученных строк и запуск Lua-функций обработки событий
// в зависимости от name (вида очереди)
//  1) !! Память q_s возвращать не надо
//  2) !! q_s->size - размер памяти строки (с учетом символа конца строки '\0')
//        Длина строки: (q_s->size - 1) 
void event_handling(q_str* q_s, std::string name) {

}

//  Функция потоков-читателей --
//  filling_threshold - порог заполненности очередей (<текущее количество записей очереди> / <размер очереди>) 
//  для выдачи предупреждений: "Превышен порог заполненности очередей" 
//  dt - пауза (млс.)   
void consumer(int id, DataCluster& dc, Manager& mgr, double filling_threshold, int dt) {
   ReaderAccess access;
   access.q_filling_threshold = filling_threshold;
   // Инициализация access.h и access->h_service с учетом текущего состояния очередей
   dc.init_access(&access, &mgr, &dc);  // "Встраивание" потока чтения в очереди
   // Буфер строки-объекта для дальнейшей обработки - q_str_c --
   const char str_copy[LEN_STR_MAX] = {'\0'};
   q_str q_str_c;  q_str_c.str = (char*)str_copy; q_str_c.size_max = LEN_STR_MAX;
   // -------------------------------------
   INT64 tt = T_OS_high_mls();
   INT64 TT = tt;               // Для замера времени выполнения  ---
   dt = dt > 1 ? dt : 1;
   INT64 DT = dt * 10000;
   INT64 DTT = 0;
   INT64 TCP = 0;               // Для замера времени ЦП обработки очередей
   INT64 NRQ = 0;               // Количество обработанных записей
   INT64 NACT = 0;              // Количество активаций потока чтения

   while (true) {                            // Пользовательский цикл обработки 
      tt = T_OS_high_mls();
      DTT = T_OS_high_mls() + DT;

      while ((T_OS_high_mls() - DTT) < 0) {  // Цикл обработки очерелей событий
         tt = (T_OS_high_mls() - tt);  // время обработки потока в цикле --
         TCP += tt;
         // Пауза 
         mgr.wait(&access, dt);
         tt = T_OS_high_mls();
         ++NACT;
         //  ------------------------------------------------------------------
         //   Получение снимка состояния очередей потока
         access.snapshot_queue_state(dc, mgr);

         //  Обработка снимка состояния очередей потока: 
         //  1) access.q_p - вектор номеров непустых очередей 
         //  2) access.q_len - вектор длин необработанных очередей
         //  3) access.q_p - длина векторов 1), 2)
         int q_p = access.q_p;
         // Просмотр и обработка вектора непустых очередей потока чтения
         for (int i = 0; i < q_p; ++i) {
            int q_n = access.q_n[i];
            int count = access.q_len[i];  // Количество записей в очереди --
            //   ------- Сбор статистики ----
            double statistics_agr = access.statistics[q_n] / (access.NN[q_n] < access.period_agr ? access.NN[q_n] : access.period_agr);
            if (statistics_agr > access.q_filling_threshold)  //  #### Предупреждающее сообщение --
               std::cout << "[Поток: " << id << "] Очередь "
               << q_n << ".  Заполненность (средняя): " << statistics_agr
               << ".  *** Превышен порог заполненности очередей: " << access.q_filling_threshold << "\n";

            ////   Отладочная печать 
            //if (count > 0 || count < 0)
            //   std::cout << "[Поток: " << id << "] Очередь "
            //   << q_n << ".  Заполненность (средняя): " << statistics_agr
            //   << ".  Количество: " << count << ". Ук чтения " << access.h[q_n] << "\n";
            //  --------------------
            INT64 q_size = dc.q[q_n].q_size;

            // Чтение и обработка очередей 
            while (count--) {
               q_str* val_q = access.get_q_str(q_n, dc);
               //   Копировать запись очереди в буфер обработки q_str_c (типа q_str), 
               // созданный в стеке --
               q_str_copy(val_q, &q_str_c, &dc.q[q_n]);  val_q = NULL;   // ! далее использовать только q_str_c
               event_handling(&q_str_c, dc.q[q_n].name_Queue);   //  Вызов функции обработки записи очереди
               ++NRQ;   // Количество обработанных записей 

               //                    Результат теста 
               // Завершение потока при тестировании --
               if (q_str_c.state == -1) {   // Признак читателям завершить поток ---
                  std::stringstream out;    // можно использовать <<
                  out << "Читатель " << id << ". Пауза:" << dt
                     << ". T (млс.):"
                     << (int)((T_OS_high_mls() - TT) * 0.0001) << ". Обработано записей:" << NRQ
                     << ". Циклов обработки:" << NACT
                     << ". Время ЦП (млс.):" << TCP * 0.0001 << "\n";
                  std::string result = out.str();    // Получаем std::string
                  std::cout << result;
                  return;
               }

               //  Отладочная печать 
               //const char* val_str = q_str_c.str; 
               //if (access.h[q_n] == N_MAX) {
               //   std::cout << " Поток " << id << ". Очередь " << q_n << " -> Val(str): " << val_str << ". Указатель чтения "
               //      << access.h[q_n] << ". Интервал (млс.) " << (T_OS_high_mls() - TT) * 0.0001 << "\n";
               //}
            }
         }
      }
      //// %%%%%%%%%%%% Обработка остальных кодов (пользователя) в потоке чтения -- 
      //if (dt > 50)
      //   std::cout << " Поток " << "\n";
   }
}

//   ---------------------  Тест очередей  -------------------------
int main() {
   SetConsoleCP(1251);                // Кодировка 1251 (! #include <shlobj.h>)
   setlocale(LC_ALL, "");             // #### Русификация вывода ввода (работает) -----
   timeBeginPeriod(1); // Установить точность 1 мс
   // ==================================
#ifdef win32_API
   std::cout << "В реализации используется win32_API" << std::endl;
#endif
   DataCluster dc;        // Очереди ---
   //  1. Для каждой очереди должен быть задан размер памяти для максимальной строки, передаваемой 
   //     в ней (с учетом признака конца строки '\0'). 
   //  2. В вызове метода set_parm_q второй параметр: вектор типа q_parm * с параметрами очередей
   //  3. Порядок элементов в векторе определяет приоритет обработки очередей (0 -> 1 ...)
   q_parm* Параметры_очередей = NULL;   // Параметры_очередей[Количество_очередей] 
   dc.set_parm_q(Количество_очередей, Параметры_очередей);
   //dc.initialization_q_str();    // #### Вряд ли стоит использовать ---
   Manager mgr;                    // Объект управления потоками чтения 
   mgr.Выдавать_сигналы = QUEUE_RECORDING_MODE; // Режим записи в очереди: 0 - без сигналов;  1 - с сигналами
   std::cout << "Режим записи в очереди: " << (mgr.Выдавать_сигналы == 0 ? "без сигналов" : "с сигналами") << std::endl;

   //              Читающие потоки --
   // Последний параметр интервал ожидания (млс.)
   std::thread c1(consumer, 1, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 5);
   std::thread c2(consumer, 2, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 10);
   std::thread c3(consumer, 3, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 50);
   std::thread c4(consumer, 4, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 100);
   //              Пишущий поток --
   std::thread p(producer, std::ref(dc), std::ref(mgr));

   //  подключение читающего потока к обработке очередей в произвольный момент времени. 
   const int Задержка_запуска_читателя = 700;
   std::cout << "Задержка_запуска_читателя 5 на " << Задержка_запуска_читателя << " млс." << std::endl;
   Sleep(Задержка_запуска_читателя);
   std::thread c5(consumer, 5, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 1);

   p.join(); c1.join(); c2.join(); c3.join(); c4.join(); c5.join();
   timeEndPeriod(1); // Освободить точность 1 млс.
   char nazv_sh[100];
   std::cout << "Введите что-нибудь:" << std::endl;
   std::cin >> nazv_sh;
   return 0;
}

 // запись в очередь --
   bool push(T val) {
       buffer[tail % q_size] = val;
      ++tail;
      return true;
  }

//  ==============  Вариант реализации очередей событий QUIK (Lock-free, ~300 строк) ============
    //   Параметр Q_MAX определяет количество очередей в схеме очередей. Схем обработки очередей можно 
    // создать несколько. 
    //   Данные, передаваемые в информационных очередях, строки.
    // 1. Формат записи данных - текст описания таблицы Lua: 
    //     {name = <Имя коллбека>, tbl = {<Таблица для формирования вызова функции пользователя>}}
    //    После чтения выполнять десереализацию текста таблицы Lua и дальше обрабатывать с учетом значения name.
    //   name обеспечивает возможность группировка коллбеков по очередям.
    // 2.  Коллбеки можно, но необязательно, сгруппировать и группы распределить по очередям с учетом того,
    //   что при чтении обработка начинается с 0-й очереди. Параметры очередей size_max следует задать 
    //   такими, чтобы в них помещались строки их данных.
    //     Для коллбеков пользовательских таблиц QUIK, наверное, имеет смысл использовать все таки отдельную 
    //   очередь.
    // 3. Параметр FILLING_THRESHOLD - порог заполненности очередей для выдачи предупреждения о высоком уровне 
    // заполненности очередей.
    // 4. После первого цикла любой очереди, память ее созданных элементов переиспользуется
    //  (кроме чтения указателя нет затрат на управление ее памятью).
    // 5. Если для читающего потока в какой то его очереди возникнет переполнение, то об этом
    // выдается сообщение.
    // ----------------------------------------------------------------------
  //                 Результат теста (оценка эффективности реализации очередей)
  // Писатель выполняет в цикле запись в 10 очередей. Циклов записи : 1000
  //  T - время выполнения потоков с учетом их пауз в млс.
  //-->Писатель. Пауза:  1.  T (млс.): 1977. Обработано записей: 10000. Время ЦП (млс.): 0.0686
  //Читатель 3. Пауза:  50.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.3174
  //Читатель 1. Пауза:   5.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.1746
  //Читатель 4. Пауза: 100.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.2129
  //Читатель 2. Пауза:  10.  T (млс.): 2023. Обработано записей: 10000. Время ЦП (млс.): 1.2194
  // -------------------------------------------------------------------------------

//                              Краткое ТЗ на разработку очередей 
// 1) Создаются несколько циклических информационных очередей с разными типами данных и общим объектом 
// управления ими.
// 2) В очереди пишет данные один поток, а читают эти очереди несколько потоков.
// 3) Дополнительно создается специальная служебная циклическая очередь: битовые шкалы непустых информационных 
// очередей, записываемых пишущим потоком, после записи в информационные очереди.
// 4) У читающих потоков должны быть свои указатели чтения циклических очередей и их локальные битовые шкалы 
// непустых очередей, хранящиеся в их объекте доступа к информационным очередям.
// 5) Читающие потоки циклически, выполнив свои коды переходят в состояние ожидания сигнала появления данных 
// в их очередях или истечения заданного интервала времени на их общем объекте ожидания.
// 6) При записи в любую очередь пишущий поток после записи в информационные очереди записывает битовые шкалы 
// непустых очередей в служебную очередь шкал непустых очередей и активирует все читающие потоки.
// 7) Каждый читающий поток, при активации, читает все появившиеся записи в служебной очереди шкал непустых 
// очередей и формирует свою локальную шкалу непустых очередей и далее, на ее основании, читает без 
// синхронизации, непустые очереди с признаком : состояние прочитанных очередей в виде размера непрочитанных 
// данных.
// 8) Реализовать эффективное управление памятью при передаче данных в этих очередях.

//                       C++20 (используется std::countr_zero) ---
#include <shlobj.h>
#include <windows.h>>
#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <mutex>
   //#include <condition_variable>
   //#include <atomic>
   //#include <bitset>
   //#include <intrin.h>
#pragma comment(lib, "winmm.lib")    // Установить точность таймера 1 мс

// Параметры тестирования очередей ---
const int N_MAX = 1000;       // Количество циклов записи при тестировании
const INT64 Q_SIZE = 64;      // Длина информационной очереди по умолчанию 
const INT64 Q_SIZE_SL = 1024; // Длина служебной очереди шкал непустых информационных очередей
const int Q_MAX = 128;        // Максимальное количество очередей
const int Количество_очередей = 120; // Количество используемых очередей ( <= Q_MAX)
const int LEN_STR_MAX = 4096; // Размер буфера потоков чтения (максимум size_max_str всех используемых очередей) 
const double FILLING_THRESHOLD = 0.4; // Порог заполненности очередей для выдачи предупреждения об уровне заполненности очередей

// Системное время (синхронизированное). Точность 0,1 мкс.---
static INT64 T_OS_high_mls()
{
   FILETIME lpSystemTimeAsFileTime;
   GetSystemTimePreciseAsFileTime(&lpSystemTimeAsFileTime);  // Системное время (синхронизированное)
   INT64 tt = lpSystemTimeAsFileTime.dwHighDateTime;
   tt = (tt << 32) + lpSystemTimeAsFileTime.dwLowDateTime;
   return tt;
}

// Строка-объект информационной очереди (переиспользуемая) --
struct q_str {
   INT64 tail = 0;      // Указатель записи ---
   INT64 size_max = 0;  // Максимальный размер памяти строки (с учетом символа конца строки '\0')
   INT64 size = 0;      // Текущий размер памяти строки (с учетом символа конца строки '\0')
   int n_q = 0;         // Номер очереди  
   char* str = NULL;    // Память строки (размер len_max) ----
   int state = 0;       // Состояние (не используется, но на всякий случай): 1 - занята пишущим потоком; 2 - в очереди.
};

// Параметры очереди, задаваемые при ее инициализации --
struct q_parm {
   std::string name_Queue = "";     // имя очереди
   INT64 q_size = 0;                // длина очереди
   INT64 size_max_str = 64;         // максимальная память строки в информационной очереди
};

// Шаблонная циклическая очередь (Lock-free)
// Используется для информационных очередей и служебной
template<typename T>
class InfoQueue {
public:
   std::vector<T> buffer;
   INT64 tail = 0;                     // пишет только один писатель
   std::string name_Queue = "";        // имя очереди (можно менять при начальной инициализации)
   INT64 q_size = 64;                  // размер очереди (можно менять при начальной инициализации)
   INT64 size_max_str = 64;            // максимальная память строки в очереди (можно менять при начальной инициализации)
   InfoQueue(INT64 s = Q_SIZE) : buffer(s), q_size(s) {}

   // Изменение длины очереди ---
   void q_size_set(int s) { 
      buffer.reserve(s); q_size = s;
   }

   // запись в очередь --
   bool push(T val) {
      if (tail < q_size)
         buffer[tail % q_size] = val;
      ++tail;
      return true;
   }

   // Количества доступных элементов для конкретного читателя
   // local_head - указатель чтения (читающего потока)
   // Если отрицательное значение, то: количество пропущенных при чтении записей очереди --
   INT64 get_pending_count(INT64 local_head) const {
      INT64 t = tail - local_head;
      return (t <= q_size ? t : q_size - t);
   }
};

// Копировать строку-объект в буфер читающего потока для последующей обработки ---
void q_str_copy(q_str* s, q_str* s1, InfoQueue<q_str*>* q)
{
   //s1->size_max = s->size_max;   // ! Задана при создании буфере --
   s1->size = s->size;
   s1->n_q = s->n_q;
   if (s1->str == NULL) throw std::runtime_error("*** Ошибка: нет памяти для копирования");
   //  s1->size - размер памяти строки (с учетом символа конца строки '\0')
   //  Длина строки: (s1->size - 1)  
   memcpy(s1->str, s->str, s1->size);  s1->tail = s->tail;   // ! Обязательно в таком порядке --
   s1->state = s->state;
   // Дополнительный контроль переполнение очереди (после копирования в буфер потока чтения)
   if ((q->tail - s1->tail) > q->q_size) std::cout << " *** Ошибка переполнение очереди (при копировании) " << s1->n_q << "\n";
}

// Информационные очереди событий --
struct DataCluster {
   //  Деструктор (! выход из блока кода, в котором определен DataCluster 
   // только при завершения потоков, использующих его) --
   ~DataCluster() { 
      for (int i = 0; i < Q_MAX; ++i) {
         INT64 size = q[i].buffer.size();
         INT64 N = (q[i].tail >= size) ? size : q[i].tail;  // Буфер может быть неполным ---
         for (int j = 0; j < N; ++j) {
            q_str* q_s = q[i].buffer[1]; 
            delete[]q_s->str;
            delete q_s;
         }
      }
   }

   int count_q = 64;  // Количество используемых информационных очередей (<= Q_MAX) --
   // При количестве очередей > 64 один бит шкалы непустых очередей соотносится 
   // к нескольким ((count_q - 1) / 64 + 1) последовательным очередям, в которых, 
   // возможно, есть записи.
   UINT32 Queue_grouping = 1; 
   InfoQueue<q_str*> q[Q_MAX];

   // Задать количество используемых очередей и их параметры
   void set_parm_q(int n, q_parm* parm) {
      if (n > 0 && n <= Q_MAX)
         count_q = n; 
      else 
         throw std::runtime_error(" Количество используемых очередей > Q_MAX или <= 0");   /* #### ошибка*/
      Queue_grouping = (count_q - 1) / 64 + 1;
      if (parm != NULL)
         for (int i = 0; i < count_q; ++i) {
            q[i].name_Queue = parm->name_Queue;
            q[i].q_size = parm->q_size;
            q[i].q_size_set(parm->q_size);
            q[i].size_max_str = parm->size_max_str;
         }
   }

   // Функция формирования маски группировки очередей (многозначное отображение очередей на шкалу непустых)
   uint64_t q_mask(int q_i) { return (uint64_t)1 << (q_i / Queue_grouping); }

   //  Запрос памяти потоком писателем под запись очереди --
   // Запрос памяти системы и формирование строки-объекта -----
   q_str* get_q_s(int q_i) {
      INT64 size_max_str = q[q_i].size_max_str;
      char* str = new char[size_max_str];
      str[size_max_str - 1] = '\0';
      q_str* q_s = new q_str;
      q_s->str = str;
      q_s->size_max = size_max_str;
      q_s->size = 0;
      q_s->n_q = q_i;
      return q_s;
   }

   // Запрос строки-объекта потоком записи для формирования строки 
   q_str* get_str(int q_i) {
      q_str* q_s;
      INT64 tail = q[q_i].tail;
      INT64 t = tail % q[q_i].q_size;
      if (q[q_i].buffer[t] != NULL) {   //  Получение памяти из очереди --
         q_s = q[q_i].buffer[t]; 
         q_s->tail = tail;
      }
      else {   // Запрос памяти у системы --
         q_s = get_q_s(q_i);
         q_s->tail = tail;
      }
      return q_s;
   }

   //  Инициализация используемых очередей строками-объектами --
   // #### Вряд ли стоит использовать ---
   void initialization_q_str() {
      if (count_q > 0) {
         for (int i = 0; i < count_q; ++i) {
            int size = q[i].q_size;
            for (int j = 0; j < size; ++j) {
               if (q[i].buffer[j] == NULL)
                  q[i].buffer[j] = get_q_s(i);
            }
         }
      }
   }

   // Начальная инициализация указателей чтения в потоке-читателе на основе указателей 
   // записи в очереди (при начальной инициализации потока)
   void init_access(INT64 *tail_v) {
      if (count_q <= 0)  throw std::runtime_error("*** Ошибка: не задано количество используемых очередей");
      for (int i = 0; i < count_q; ++i) tail_v[i] = q[i].tail;
   }

   // запись в очередь --
   bool push(int q_n, q_str* val) {
      if (q_n >= count_q) throw std::runtime_error("*** Ошибка: нет такой очереди");
      if (q[q_n].tail < q[q_n].q_size)
         q[q_n].buffer[q[q_n].tail % q[q_n].q_size] = val;
      ++q[q_n].tail;
      return true;
   }
};

struct ReaderAccess;   // Используется в Manager
// Объект управления потоками чтения 
struct Manager {
   INT64 signal_counter = 0;              // Счетчик сигналов записи
   std::mutex mtx;                        // Для синхронизации  при ожидании             
   std::condition_variable cv;            // Общий объект ожидания  
   InfoQueue<uint64_t> service_q{ Q_SIZE_SL }; // Служебная очередь битовых шкал

   void wait(ReaderAccess* access, int tt);  // !! Реализация после struct ReaderAccess (иначе недоступны поля)

   void pulse(uint64_t mask) { service_q.push(mask); ++signal_counter;  cv.notify_all(); }
   ////  ------- win32 API --
   //HANDLE cv = CreateEvent(NULL, FALSE, FALSE, NULL);  // Общий объект ожидания
   //void pulse() { ++signal_counter; PulseEvent(cv); }
private:
   INT64 tt = T_OS_high_mls();  // Для коррекции интервала ожидания --
};

// Объект доступа к очередям для читающих потоков --
struct ReaderAccess {
   INT64 h[Q_MAX] = { 0 };        // Локальные указатели чтения
   INT64 h_service = 0;           // Указатель чтения служебной очереди
   uint64_t local_bitmask = 0;    // Локальная шкала непустых очередей потока-читателя
   INT64 signal_counter = 0;      // Локальный счетчик сигналов записи 
   //  -------- Статистика очередей ---------
   INT64 NN[Q_MAX] = { 0 };          // Коичество обращений к непустым очередям
   double statistics[Q_MAX] = { 0 }; // Сумма относительной заполненности очередей (среднее = statistics [i]/ (NN[i]))
   double q_filling_threshold = 0.4; // Порог заполненности очередей для выдачи предупреждения --
   int period_agr = 9;               // Период скользящей 
   //  ------- Данные снимка состояния очередей потока чтения ---
   int q_len[Q_MAX] = { 0 };   // Количество непустых непрочитанных записей в очередях
   int q_n[Q_MAX] = { 0 };     // Номера непустых очередей
   int q_p = 0;                //  Длина векторов q_n и q_len

   // Снимок состояния очередей  потока-читателя ("мгновенный", чтобы меньше было повторных чтений очередей).
   void snapshot_queue_state(DataCluster& dc, Manager& mgr) {
      q_p = 0;  // Сброс длины векторов q_n и q_len
      // Формирование шкалы непустых очередей потока-читателя
      uint64_t mask_val;
      INT64 count = mgr.service_q.get_pending_count(h_service);
      if (count > 0) {
         // Формируем общую локальную шкалу непустых очередей потока-читателя --
         for (int i = 0; i < count; ++i) {
            mask_val = mgr.service_q.buffer[h_service % mgr.service_q.q_size];
            local_bitmask |= mask_val; 
            ++h_service;
         }
         // Формирование снимка состояния очередей потока-читателя (на основе его шкалы) --
         int pos_old = 0;
         int Queue_grouping = dc.Queue_grouping;
         while (local_bitmask) {   //  Эффективная обработка шкалы непустых очередей --
            ////  ! Способ определения позиции младшего разряда для старых стандартов (MSVC)  -----
            // unsigned long pos3 = 0;
            // _BitScanForward64(&pos3, local_bitmask);  // позиция младшего разряда с 1
            ////  -------------------------------------------------
            int Сдвиг = std::countr_zero(local_bitmask); // C++20 (позиция младшего разряда с 1)
            int pos = Сдвиг + pos_old;
            pos_old = pos + 1;
            pos *= Queue_grouping;
            for (int i = 0; i < Queue_grouping; ++i) {  // обработка групп очередей разряда шкалы --
               INT64 q_count = dc.q[pos + i].get_pending_count(h[pos + i]);
               if (q_count > 0) {
                  int p = q_p;
                  q_n[p] = pos + i;
                  q_len[p] = q_count;
                  //  Статистика заполненности очереди (относительная, приблизительно средне скользящая) 
                  if (++NN[pos + i] < period_agr) {
                     statistics[pos + i] += (double)q_count / (double)dc.q[pos + i].q_size;
                  }
                  else {
                     statistics[pos + i] += (double)q_count / (double)dc.q[pos + i].q_size - statistics[pos + i] / period_agr;
                  }
                  ++q_p;
               }
               else
                  if (q_count < 0)  std::cout << " *** Ошибка переполнение очереди (при чтении) q = " << pos + i << ": q_count = " << q_count << "\n";  // #### Ошибка переполнение очереди
            }
            local_bitmask >>= Сдвиг + 1;
         }
      }
      else if (count < 0)  std::cout << " *** -- Ошибка переполнение служебной очереди шкал --\n";
      //std::cout << ".  Шкала : " << std::bitset<64>(local_bitmask) << "\n";
   }

   // Чтение строки-объекта очереди --
   // q_n - номер очереди
   q_str* get_q_str(int q_n, DataCluster& dc) {
      q_str* val = dc.q[q_n].buffer[h[q_n] % dc.q[q_n].q_size];
      ++h[q_n];
      return val;
   }
};

// ! Реализация члена Manager (расположена после struct ReaderAccess чтобы в методе были 
// доступны ее поля)
// Пауза (подстраиваемая с учетом длительности цикла обработки)
void Manager::wait(ReaderAccess* access, int dt) {
   std::unique_lock<std::mutex> lock(mtx);
   if (access->signal_counter == signal_counter) {// не было сигнала записи при отсутствии ожидания в потоке чтения
      // Вычисление коррекции паузы с учетом времени обработки цикла --
      tt = (T_OS_high_mls() - tt) * 0.0001;  // время обработки потока в цикле (млс.) --
      tt = (dt - tt);
      tt = tt > 0 ? tt : 1;
      cv.wait_for(lock, std::chrono::milliseconds(tt)) == std::cv_status::timeout;
      //WaitForSingleObject(cv, tt); // вместо предыдущего оператора с комментированием первого оператора метода
   }
   tt = T_OS_high_mls();
   access->signal_counter = signal_counter;   // Обрабатываемый сигнал записи в потоке чтения
}

//  Тестирование очередей событий  ------------------------------------------------
//                          Функция потока-писателя
// Формат записываемых строк общий для всех очередей: тексты таблиц Lua (сереализация).
// Читающий поток выполняет десереализацию полученных строк и запуск Lua-функций обработки событий
void producer(DataCluster& dc, Manager& mgr) {
   std::cout << "Писатель. Запись выполняется в 10 очередей. Циклов записи: " << N_MAX << std::endl;
   std::cout << "T - время выполнения потоков с учетом их пауз." << std::endl;
   // -------------------------------------
   INT64 tt = T_OS_high_mls();  // Для коррекции интервала ожидания --
   INT64 TT = tt;               // Для замера времени выполнения  ---
   INT64 TCP = 0;               // Для замера времени ЦП  ---
   INT64 NRQ = 0;               // Количество обработанных записей
   int dt = 1;
   for (int i = 1; i <= N_MAX; ++i) {
      tt = T_OS_high_mls() - tt;
      TCP += tt;
      std::this_thread::sleep_for(std::chrono::milliseconds(dt));   // Пауза потока записи
      //Sleep(dt);   // И так можно 
      //  %%%%%% Подготовка данных для записи в очереди в виде C-строк

      //   Запись строк в очереди --
      uint64_t mask = 0;  // шкала непустых очередей
      //// Запись в разные очереди
      // !!  rec->size - размер памяти строки (с учетом символа конца строки '\0')
      //     Длина строки: (rec->size - 1) 
      const char* source = "{name = 'OnCleanUp',tbl = {  ....] }"; int sz = strlen(source) + 1;
      q_str* rec = dc.get_str(0); memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(0, rec))  mask |= dc.q_mask(0);  ++NRQ;
      rec = dc.get_str(1);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(1, rec))  mask |= dc.q_mask(1);  ++NRQ;
      rec = dc.get_str(2);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(2, rec))  mask |= dc.q_mask(2);  ++NRQ;
      rec = dc.get_str(3);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(3, rec))  mask |= dc.q_mask(3);  ++NRQ;

      rec = dc.get_str(40);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(40, rec))  mask |= dc.q_mask(40);  ++NRQ;
      rec = dc.get_str(41);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(41, rec))  mask |= dc.q_mask(41);  ++NRQ;

      rec = dc.get_str(40);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(40, rec))  mask |= dc.q_mask(40);  ++NRQ;
      rec = dc.get_str(41);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(41, rec))  mask |= dc.q_mask(41);  ++NRQ;
      rec = dc.get_str(42);  memcpy(rec->str, source, sz); rec->size = sz;
      if (dc.push(42, rec))  mask |= dc.q_mask(42);  ++NRQ;
      rec = dc.get_str(100); memcpy(rec->str, "шшшшшш", 7); rec->size = 7;
      if (i >= N_MAX) rec->state= -1;   // Признак читателям завершить поток ---
      if (dc.push(100, rec)) mask |= dc.q_mask(100);  ++NRQ;   // "шшшшшш"
      // std::cout << "[P] Данные записаны, маска: " << std::bitset<64>(mask) << std::endl;  // ####
      // Запись маски и активация потоков
      mgr.pulse(mask);
      tt = T_OS_high_mls();
   }
   std::stringstream out;   // можно использовать <<
   out << "--->   Писатель. Пауза:" << dt
      << ". T (млс.):"
      << (int)((T_OS_high_mls() - TT) * 0.0001) << ". Обработано записей:" << NRQ
      << ". Время ЦП (млс.):" << TCP * 0.0001 << "\n";
   std::string result = out.str();    // Получаем std::string
   std::cout << result;
}

// Заглушка функции обработки событий. 
//   Выполняет десереализацию полученных строк и запуск Lua-функций обработки событий
// в зависимости от name (вида очереди)
//  1) !! Память q_s возвращать не надо
//  2) !! q_s->size - размер памяти строки (с учетом символа конца строки '\0')
//        Длина строки: (q_s->size - 1) 
void event_handling(q_str* q_s, std::string name) {

}

//  Функция потоков-читателей --
//  filling_threshold - порог заполненности очередей (<текущее количество записей очереди> / <размер очереди>) 
//  для выдачи предупреждений: "Превышен порог заполненности очередей" 
//  dt - пауза (млс.)   
void consumer(int id, DataCluster& dc, Manager& mgr, double filling_threshold, int dt) {
   ReaderAccess access;
   access.q_filling_threshold = filling_threshold;
   dc.init_access(access.h);  // Инициализация access.h (указателей) потока чтения с учетом текущего состояния очередей --
   // Буфер строки-объекта для дальнейшей обработки - q_str_c --
   const char str_copy[LEN_STR_MAX] = {'\0'};
   q_str q_str_c;  q_str_c.str = (char*)str_copy; q_str_c.size_max = LEN_STR_MAX;
   // -------------------------------------
   INT64 tt = T_OS_high_mls();  // Для коррекции интервала ожидания --
   INT64 TT = tt;               // Для замера времени выполнения  ---
   INT64 TCP = 0;               // Для замера времени ЦП  ---
   INT64 NRQ = 0;               // Количество обработанных записей
   while (true) {
      tt = (T_OS_high_mls() - tt);  // время обработки потока в цикле --
      TCP += tt;
      // Пауза (подстраиваемая с учетом длительности цикла обработки) 
      // с ожиданием сигнала записи в одну из информационных очередей --
      mgr.wait(&access, dt);
      tt = T_OS_high_mls(); 
      //  ------------------------------------------------------------------
      //   Получение снимка состояния очередей потока ("мгновенный", 
      //  чтобы меньше было повторных чтений очередей).
      access.snapshot_queue_state(dc, mgr);

      //  Обработка снимка состояния очередей потока: 
      //  1) access.q_p - вектор номеров непустых очередей 
      //  2) access.q_len - вектор длин необработанных очередей
      //  3) access.q_p - длина векторов 1), 2)
      int q_p = access.q_p;
      // Просмотр и обработка вектора непустых очередей потока чтения
      for (int i = 0; i < q_p; ++i) {     
         int q_n = access.q_n[i];
         int count = access.q_len[i];  // Количество записей в очереди --
         //   ------- Сбор статистики ----
         double statistics_agr = access.statistics[q_n] / (access.NN[q_n] < access.period_agr ? access.NN[q_n] : access.period_agr);
         if (statistics_agr > access.q_filling_threshold)  //  #### Предупреждающее сообщение --
            std::cout << "[Поток: " << id << "] Очередь "
            << q_n << ".  Заполненность (средняя): " << statistics_agr
            << ".  *** Превышен порог заполненности очередей: " << access.q_filling_threshold << "\n";

         ////   Отладочная печать 
         //if (count > 0 || count < 0)
         //   std::cout << "[Поток: " << id << "] Очередь "
         //   << q_n << ".  Заполненность (средняя): " << statistics_agr
         //   << ".  Количество: " << count << ". Ук чтения " << access.h[q_n] << "\n";
         //  --------------------
         INT64 q_size = dc.q[q_n].q_size;
         // Чтение и обработка очередей 
         while (count--) {
            q_str* val_q = access.get_q_str(q_n, dc);
            //   Копировать запись очереди в буфер обработки q_str_c (типа q_str), 
            // созданный в стеке --
            q_str_copy(val_q, &q_str_c, &dc.q[q_n]);  val_q = NULL;   // ! далее использовать только q_str_c
            event_handling(&q_str_c, dc.q[q_n].name_Queue);   //  Вызов функции обработки записи очереди
            ++NRQ;   // Количество обработанных записей 

            //                    Результат теста 
            // Завершение потока при тестировании --
            if (q_str_c.state == -1) {   // Признак читателям завершить поток ---
               std::stringstream out;    // можно использовать <<
               out << "Читатель " << id << ". Пауза:" << dt
                  << ". T (млс.):"
                  << (int)((T_OS_high_mls() - TT) * 0.0001) << ". Обработано записей:" << NRQ
                  << ". Время ЦП (млс.):" << TCP * 0.0001 << "\n";
               std::string result = out.str();    // Получаем std::string
               std::cout << result;
               return;
            }

            //  Отладочная печать 
            //const char* val_str = q_str_c.str; 
            //if (access.h[q_n] == N_MAX) {
            //   std::cout << " Поток " << id << ". Очередь " << q_n << " -> Val(str): " << val_str << ". Указатель чтения "
            //      << access.h[q_n] << ". Интервал (млс.) " << (T_OS_high_mls() - TT) * 0.0001 << "\n";
            //}
            //  ----------------------------------------------------

            //// %%%%%%%%%%%% Обработка остальных данных в потоке чтения --      
         }
      }
   }
}

//   ---------------------  Тест очередей  -------------------------
int main() {
   SetConsoleCP(1251);                // Кодировка 1251 (! #include <shlobj.h>)
   setlocale(LC_ALL, "");             // #### Русификация вывода ввода (работает) -----
   timeBeginPeriod(1); // Установить точность 1 мс
   // ==================================
   DataCluster dc;        // Очереди ---
   //  1. Для каждой очереди должен быть задан размер памяти для максимальной строки, передаваемой 
   //     в ней (с учетом признака конца строки '\0'). 
   //  2. В вызове метода set_parm_q второй параметр: вектор типа q_parm * с параметрами очередей
   //  3. Порядок элементов в векторе определяет приоритет обработки очередей (0 -> 1 ...)
   q_parm* Параметры_очередей = NULL;   // Параметры_очередей[Количество_очередей] 
   dc.set_parm_q(Количество_очередей, Параметры_очередей);
   //dc.initialization_q_str();    // #### Вряд ли стоит использовать ---
   Manager mgr;                    // Объект управления потоками чтения 

   //              Читающие потоки --
   // Последний параметр интервал ожидания (млс.)
   std::thread c1(consumer, 1, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 5);
   std::thread c2(consumer, 2, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 10);
   std::thread c3(consumer, 3, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 50);
   std::thread c4(consumer, 4, std::ref(dc), std::ref(mgr), FILLING_THRESHOLD, 100);
   //              Пишущий поток --
   std::thread p(producer, std::ref(dc), std::ref(mgr));

   p.join(); c1.join(); c2.join(); c3.join(); c4.join();
   timeEndPeriod(1); // Освободить точность
   char nazv_sh[100];
   std::cout << "Введите что-нибудь:" << std::endl;
   std::cin >> nazv_sh;
   return 0;
}

// Это сложная задача, требующая использования атомарных операций (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