Если бы я был архитектором QUIK

//  ==============  Вариант реализации очередей событий 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;
}