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