Differences

This shows you the differences between two versions of the page.

--- courses:high_performance_computing:producer_consumer [2017/03/28 08:41] – created kel
+++ courses:high_performance_computing:producer_consumer [2025/04/27 06:51] (current) – kel
@@ Line 1: / Line 1: @@
-====== Использование pthread API ======
+====== Posix threads ======
+> [[https://classroom.github.com/a/w-966goN|GitHub-classroom]] для самостоятельно изучающих курс **не** в рамках университетских программ
 Ваша задача - реализовать классический паттерн producer-consumer с небольшими дополнительными условиями.
-Программа должна состоять из четырех потоков: один - главный, producer, consumer и interruptor. На стандартный ввод программе подается строка - список чисел произвольной длины.
+Программа должна состоять из 3+N потоков:
-  * Задача producer-потока - получить на вход список чисел, и по очереди использовать каждое значение из этого списка для обновления переменной разделяемой между потоками. После этого поток должен дождаться реакции consumer-потока, и продолжить обновление значений только после того как поток-consumer принял это изменение. Функция исполняющая код этого потока producer_routine должна принимать указатель на объект типа Value, и использовать его для обновления.
+  - Главный
-  * Задача consumer-потока отреагировать на каждое изменение переменной data и набирать сумму полученных значений. После того как достигнуто последнее обновление, функция потока должна вернуть результирующую сумму. Также этот поток должен защититься от попыток потока-interruptor его остановить. Функция исполняющая код этого потока consumer_routine должна принимать указатель на тот же объект типа Value, и читать из него обновления .
+  - producer
-  * Задача потока-interruptor проста: пока происходит процесс обновления значений, он должен постоянно пытаться остановить поток consumer. Как только поток producer произвел последнее обновление, этот поток завершается.
+  - interruptor
+  - N потоков consumer
+В файл, чтение которого уже реализовано в шаблоне кода в блокирующкем режиме, пишется список чисел, разделённых пробелом (читать можно до переноса строки). Длина списка чисел не задаётся - считывание происходит до перевода каретки.
+  * Задача producer-потока - получить на вход список чисел, и по очереди использовать каждое значение из этого списка для обновления переменной разделяемой между потоками
+  * Задача consumer-потоков отреагировать на уведомление от producer и набирать сумму полученных значений. Также этот поток должен защититься от попыток потока-interruptor его остановить. Дополнительные условия:
+    - Функция, исполняющая код этого потока consumer_routine, должна принимать указатель на объект/переменную, из которого будет читать обновления
+    - После суммирования переменной поток должен заснуть на случайное количество миллисекунд, верхний предел будет передан на вход приложения (0 миллисекунд также должно корректно обрабатываться). Вовремя сна поток не должен мешать другим потокам consumer выполнять свои задачи, если они есть
+    - Потоки consumer не должны дублировать вычисления друг с другом одних и тех же значений
+    - В качестве возвращаемого значения поток должен вернуть свою частичную посчитанную сумму
+  * Задача потока-interruptor проста: пока происходит процесс обновления значений, он должен постоянно пытаться остановить случайный поток consumer (вычисление случайного потока происходит перед каждой попыткой остановки). Как только поток producer произвел последнее обновление, этот поток завершается. В этом потоке можно выполнять вспомогательные действия которые помогут корректно обработать сигнал
+  * Функция run_threads должна запускать все потоки, дожидаться их выполнения, и возвращать результат общего суммирования
+  * Завершение приложения происходит или при считывании перевода каретки из файла или по посылке сигнала SIGTERM, обработку которого нужно также добавить. В обработчике сигнала можно вызывать только signal-safe функции https://man7.org/linux/man-pages/man7/signal-safety.7.html В случае, если в этот момент поток, читающий данные с файла, находится в режиме блокирующего чтения, - он также должен корректно завершиться. Полезно присмотреться к [[https://en.cppreference.com/w/cpp/utility/program/sig_atomic_t|std::sig_atomic_t]] - позволяет потокобезопасно и signal-safe обращаться к переменной такого типа.
+  * При сигнале в качестве вывода нужно выдавать посчитанную на этот момент сумму
-Функция run_threads должна запускать все три потока, дожидаться их выполнения, и возвращать результат потока-consumer.
+Нефункциональные требования:
+  - В нашем producer/consumer необходимо реализоваться потоковую обработку данных для этих целей запрещается загружать весь файл в ОЗУ, иначе на больших файлах обработка не будет помещаться в ограничения ОЗУ на вычислительном устройстве.
+  - При отсутствии входных данных ядра процессора не должны вхолостую есть такты
-Для обновления и получения значения следует использовать подготовленный класс Value. Следует создать один экземпляр Value, передать его аргументом в функции producer_routine и consumer_routine. Чтобы обновить значение, следует использовать метод update, чтобы получить - get
+Для обеспечения межпоточного взаимодействия допускается использование только pthread API. На вход приложения передаётся 2 аргумента при старте именно в такой последовательности:
+  - Число потоков consumer
+  - Верхний предел сна consumer в миллисекундах
-Для обеспечения межпоточного взаимодействия допускается использование только pthread API.
+В поток вывода должно попадать только результирующее значение, по умолчанию никакой отладочной или запросной информации выводиться не должно.
+В случае детектирования ошибок нужно выдавать не нулевой код возврата.
 <file cpp>
+#include <iostream>
+#include <fstream>
 #include <pthread.h>
-class Value {
-public:
-    Value() : _value(0) { }
-    void update(int value) {
-        _value = value;
-    }
-    int get() const {
-        return _value;
-    }
-private:
-    int _value;
-};
 void* producer_routine(void* arg) {
-  // Wait for consumer to start
+  // Wait for consumer to start.
+  // You should use this waiting only for debugging your code
+  // For the final solution please remove this waiting
   // Read data, loop through each value and update the value, notify consumer, wait for consumer to process
+  std::ifstream ifs("in.txt");
+  // ...
 }
 void* consumer_routine(void* arg) {
   // notify about start
-  // allocate value for result
+  // you should use this notification only for debugging your code
+  // for the final solution please remove this notification
   // for every update issued by producer, read the value and add to sum
-  // return pointer to result
+  // return pointer to result (for particular consumer)
 }
 void* consumer_interruptor_routine(void* arg) {
-  // wait for consumer to start
+  // wait for consumers to start
+  // you should use this waiting only for debugging your code
+  // for the final solution please remove this waiting
-  // interrupt consumer while producer is running
+  // interrupt random consumer while producer is running
 }
 int run_threads() {
-  // start 3 threads and wait until they're done
+  // start N threads and wait until they're done
-  // return sum of update values seen by consumer
+  // return aggregated sum of values
   return 0;