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

for i = 1, 5000000000 do   end

LUA_API void lua_pushboolean (lua_State *L, int b) {
  lua_lock(L);
  if (b)
    setbtvalue(s2v(L->top));
  else
    setbfvalue(s2v(L->top));
  api_incr_top(L);
  lua_unlock(L);
}

 for i = 1, 5000000000 do   end

local sec_list      = {
    {'SPBFUT', 'MMU6'},
    {'SPBFUT', 'SiU6'},
    {'SPBFUT', 'RIU6'},
    {'SPBFUT', 'USDRUBF'},
    {'SPBFUT', 'CNYRUBF'},
    {'TQBR', 'SBER'},
    {'TQBR', 'GAZP'}
}

//  ==============  Вариант реализации очередей событий 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 = 128;  // Длина служебной очереди шкал непустых информационных очередей
const int Q_MAX = 64;         // Максимальное количество очередей
const int Количество_очередей = 64; // Количество используемых очередей ( <= 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 alignas(64) q_str {
   INT64 tail = -1;     // Указатель записи ---
   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 alignas(64) Queue {
public:
   std::vector<T> buffer;
   //volatile INT64 tail = 0;           
   std::atomic<INT64> tail{ 0 };       // пишет только один писатель
   std::string name_Queue = "";        // имя очереди (можно менять при начальной инициализации)
   INT64 q_size = 64;                  // размер очереди (можно менять при начальной инициализации)
   INT64 size_max_str = 64;            // максимальная память строки в очереди (можно менять при начальной инициализации)
   Queue(INT64 s = Q_SIZE) : buffer(s), q_size(s) {}

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

   // запись в очередь --
   bool push_v(T val) {
      buffer[tail.load(std::memory_order_relaxed) % q_size] = val;
      //++tail;
      tail.fetch_add(1, std::memory_order_relaxed);
      return true;
   }

   // Чтение из очереди --
   T get_v(INT64 h_v) {
      if (tail.load(std::memory_order_relaxed) - h_v <= 0) return NULL;
      T val = buffer[h_v % q_size];
      if (tail.load(std::memory_order_relaxed) - h_v >= q_size - 1) throw std::invalid_argument("**** Переполнение при чтении");
      return val;
   }

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

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

   int count_q = 64;  // Количество используемых информационных очередей (<= Q_MAX) --
   // При количестве очередей > 64 один бит шкалы непустых очередей соотносится 
   // к нескольким ((count_q - 1) / 64 + 1) последовательным очередям, в которых, 
   // возможно, есть записи.
   UINT32 Queue_grouping = 1; 
   Queue<q_str*> Queue_QUIK[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) {
            Queue_QUIK[i].name_Queue = parm->name_Queue;
            Queue_QUIK[i].q_size = parm->q_size;
            Queue_QUIK[i].q_size_set(parm->q_size);
            Queue_QUIK[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 = Queue_QUIK[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 t = Queue_QUIK[q_i].tail.load(std::memory_order_relaxed) % Queue_QUIK[q_i].q_size;
      if (Queue_QUIK[q_i].buffer[t] != NULL) {   //  Получение памяти из очереди --
         q_s = Queue_QUIK[q_i].buffer[t]; 
         q_s->tail = -1;
      }
      else {   // Запрос памяти у системы --
         q_s = get_q_s(q_i);
         q_s->tail = -1;
      }
      return q_s;
   }

   //  Инициализация используемых очередей строками-объектами --
   // #### Вряд ли стоит использовать ---
   void initialization_q_str() {
      if (count_q > 0) {
         for (int i = 0; i < count_q; ++i) {
            int size = Queue_QUIK[i].q_size;
            for (int j = 0; j < size; ++j) {
               if (Queue_QUIK[i].buffer[j] == NULL)
                  Queue_QUIK[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("*** Ошибка: нет такой очереди");
      INT64  tail_p = Queue_QUIK[q_n].tail.load(std::memory_order_relaxed);
      val->tail = tail_p;
      Queue_QUIK[q_n].buffer[tail_p % Queue_QUIK[q_n].q_size] = val;
      Queue_QUIK[q_n].tail.fetch_add(1, std::memory_order_relaxed);
      return true;
   }
};

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

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

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

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

// Объект доступа к очередям для читающих потоков --
struct alignas(64) ReaderAccess {
   INT64 h_data[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;
         //  -----
         INT64 q_size = mgr.service_q.q_size;
         for (int i = 0; i < count; ++i) {
            mask_val = mgr.service_q.buffer[h_service % q_size];
            if (mgr.service_q.tail.load(std::memory_order_relaxed) - h_service >= q_size - 1)  // ####
               std::cout << " *** Ошибка переполнение служебной очереди (при чтении) Queue_QUIK \n";  // #### Ошибка переполнение очереди
            ++h_service;
            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.Queue_QUIK[pos + i].get_pending_count(h_data[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.Queue_QUIK[pos + i].q_size;
                  }
                  else {
                     statistics[pos + i] += (double)q_count / (double)dc.Queue_QUIK[pos + i].q_size - statistics[pos + i] / period_agr;
                  }
                  // --------------------------------------------------------------------------------
                  ++q_p;
               }
               else
                  if (q_count < 0)  std::cout << " *** Ошибка переполнение очереди (при чтении) Queue_QUIK = " << pos + i << ": q_count = " << q_count << "\n";  // #### Ошибка переполнение очереди
            }
            local_bitmask >>= Сдвиг + 1;
         }
      }
      else 
         if (count < 0)  std::cout << " *** -- Ошибка переполнение служебной очереди шкал --\n";
   }

   // Чтение строки-объекта очереди --
   // q_n - номер очереди
   // q_str* s1  -- буфер чтения (структура в стеке)
   q_str* get_q_str(int q_n, q_str* s1, DataCluster& dc) {
      if (s1->str == NULL) throw std::runtime_error("*** Ошибка: нет памяти для копирования");
      INT64 tail_h = h_data[q_n];
      if (tail_h >= dc.Queue_QUIK[q_n].tail.load(std::memory_order_relaxed)) return NULL;   // нет записи --
      INT64 q_size = dc.Queue_QUIK[q_n].q_size;
      q_str* val = dc.Queue_QUIK[q_n].buffer[tail_h % q_size];
      if (val != NULL) {   // есть запись --
         s1->size = val->size;
         s1->n_q = val->n_q;
         //  s1->size - размер памяти строки (с учетом символа конца строки '\0')
         //  Длина строки: (s1->size - 1)  
         memcpy(s1->str, val->str, s1->size);  s1->tail = tail_h;   // ! Обязательно в таком порядке --
         s1->state = val->state;
         // Дополнительный контроль корректности чтения очереди (после копирования в буфер потока чтения)
         //  ! Чтение dc.Queue_QUIK[q_n].tail обязательно
         if (dc.Queue_QUIK[q_n].tail.load(std::memory_order_relaxed) - tail_h >= q_size - 1) std::cout << " *** Ошибка чтения очереди (переполнение) " << s1->n_q << "\n";
         ++h_data[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.load(std::memory_order_relaxed);
   // Получение указателей информационных очередей 
   for (int i = 0; i < count_q; ++i) access->h_data[i] = Queue_QUIK[i].tail.load(std::memory_order_relaxed);
}

// ! Реализация члена Manager (расположена после struct ReaderAccess чтобы в методе были 
// доступны ее поля)
// Пауза
#ifndef win32_API
void Manager::wait(ReaderAccess* access) {
   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) {
      if (access->signal_counter == signal_counter) {// не было сигнала записи при отсутствии ожидания в потоке чтения
         EnterCriticalSection(&cs);
            // Пытаемся дождаться условия в течение заданного timeout
               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;
               }
         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 + q_n_test1);  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(60);  memcpy(rec->str, "шшшшшш", 7); rec->size = 7;
      if (i >= N_MAX) rec->state= -1;   // Признак читателям завершить поток ---
      if (dc.push(60, rec))  mask |= dc.q_mask(60);  ++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_data и 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;
   INT64 TT = T_OS_high_mls();               // Для замера общего времени выполнения  ---
   INT64 DT = 2 * 10000;        //  
   INT64 DTT = 0;
   INT64 TCP = 0;               // Для замера времени ЦП обработки очередей
   INT64 NRQ = 0;               // Количество обработанных записей
   INT64 NACT = 0;              // Количество активаций потока чтения

   while (true) {                            // Пользовательский цикл обработки 

      DTT = T_OS_high_mls() + DT;
      while ((T_OS_high_mls() - DTT) < 0) {  // Цикл обработки очередей событий
         // Пауза 
         mgr.wait(&access);
         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_data[q_n] << "\n";
            //  --------------------

            INT64 q_size = dc.Queue_QUIK[q_n].q_size;
            // Чтение и обработка очередей 
            while (count-- > 0) {
               //   Копировать запись очереди в буфер обработки q_str_c (типа q_str), 
               // созданный в стеке --
               q_str* val_q = access.get_q_str(q_n, &q_str_c, dc);
               if (val_q == NULL) continue;
               event_handling(&q_str_c, dc.Queue_QUIK[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_data[q_n] == N_MAX) {
               //   std::cout << " Поток " << id << ". Очередь " << q_n << " -> Val(str): " << val_str << ". Указатель чтения "
               //      << access.h_data[q_n] << ". Интервал (млс.) " << (T_OS_high_mls() - TT) * 0.0001 << "\n";
               //}
            }
         }
         TCP += (T_OS_high_mls() - tt);  // время обработки очередей --
      }

      tt = T_OS_high_mls();   // Начало обработки пользовательского цикла --
      //// %%%%%%%%%%%% Обработка остальных кодов (пользователя) в потоке чтения -- 
      //if (dt > 50)
      //   std::cout << " Поток " << id << ". Интервал " << dt * 10000 - (T_OS_high_mls() - tt) << "\n";

      // Конец обработки пользовательского цикла --
      // Коррекция DT с учетом времени выполнения пользовательского цикла --
      DT = dt * 10000 - (T_OS_high_mls() - tt);
      DT = DT > 20000 ? DT : 20000;
   }
}

//   ---------------------  Тест очередей  -------------------------
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, 3);
   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;
}

----   Функции работы с очередями --
--- При использовании между потоком колбеков и потоком main очереди потокобезопасные. ---
--- Создать очередь 
function QueueNew()
   return  {first = 1, last = 0 }
end
---  Записать в очередь --
function QueuePush(self, v)
   if v == nil then error('!!! Ошибка. Параметр v функции QueuePush nil') end
   local last = self.last + 1 
   self[last] = v
   self.last = last
   return last - self.first + 1    -- количество элементов в очереди ---
end 
---  Читать очередь  --
function QueuePop(self)
   local  first =  self.first
   if  first > self.last then return nil   end 
   local v, v1  =  self[first], self[first + 1]
   self.first = first + 1 
   self[first] =  nil 
   v1 = type (v1) == 'string' and #v1 or 0
   return  v, v1   -- v1 -длина следующего элемента (если строка) ---
end
  
-- Текущий размер очереди ---
function QueueSize(self)
   return self.last - self.first + 1 
end 

res.last_rec    = 0  ---   tonumber((tostring(getInfoParam('LASTRECORDTIME') or ''):gsub(':', ''))) or 0
res.local_time = 0 --- (tonumber((tostring(getInfoParam('LOCALTIME') or ''):gsub(':', ''))) or 0)

res.last_rec    = 0  ---   tonumber((tostring(getInfoParam('LASTRECORDTIME') or ''):gsub(':', ''))) or 0
res.local_time = 0 --- (tonumber((tostring(getInfoParam('LOCALTIME') or ''):gsub(':', ''))) or 0)

res.last_rec   = tonumber((tostring(getInfoParam('LASTRECORDTIME') or ''):gsub(':', ''))) or 0
res.local_time = (tonumber((tostring(getInfoParam('LOCALTIME') or ''):gsub(':', ''))) or 0)

 
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