monitor example from weber split into seperate files

MonitorExample/CMakeLists.txt

@@ -0,0 +1,27 @@
+cmake_minimum_required(VERSION 3.14)
+
+project(MonitorExample LANGUAGES CXX)
+
+set(CMAKE_AUTOUIC ON)
+set(CMAKE_AUTOMOC ON)
+set(CMAKE_AUTORCC ON)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core)
+find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core)
+
+add_executable(MonitorExample
+  main.cpp
+  monitor.h monitor.cpp
+  producer.h producer.cpp
+  consumer.h consumer.cpp
+)
+target_link_libraries(MonitorExample Qt${QT_VERSION_MAJOR}::Core)
+
+include(GNUInstallDirs)
+install(TARGETS MonitorExample
+    LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
+    RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
+)

MonitorExample/consumer.cpp

@@ -0,0 +1,23 @@
+#include "consumer.h"
+
+template<typename T> QSemaphore Consumer<T>::numProducts;
+template<typename T> Monitor<T> *Consumer<T>::mon;
+template<> void (*Consumer<int>::consume)(int, int) = NULL;
+
+//---------------------------------------
+template<typename T> void Consumer<T>::initClass(int numP, Monitor<T> *m, void (*cons)(T, int)) {
+    numProducts.release(numP);
+    mon = m;
+    consume = cons;
+}
+//---------------------------------------
+
+template<typename T> void Consumer<T>::run() {
+    while (numProducts.tryAcquire()) {	// While not all numProducts are consumed:
+        T* aux = mon->canGet();			//   Get pointer to item in itemQ
+        T item = *aux;					//   Take the item out of itemQ
+        mon->doneGetting(aux);			//   Give info to producer threads
+        (*consume)(item,ID);				//   Consume one item
+    }
+}
+//---------------------------------------

MonitorExample/consumer.h

@@ -0,0 +1,22 @@
+#ifndef CONSUMER_H
+#define CONSUMER_H
+
+#include "monitor.h"
+#include <QThread>
+#include <QSemaphore>
+
+template<typename T>
+class Consumer : public QThread {
+private:
+    int ID;
+    static Monitor<T> *mon;
+    static QSemaphore numProducts;
+public:
+    static void (*consume)(T i, int);
+    static void initClass(int numP, Monitor<T> *m, void (*cons)(T, int));
+
+    Consumer<T>(int i) : ID(i) {}
+    void run();
+};
+
+#endif // CONSUMER_H

MonitorExample/main.cpp

@@ -0,0 +1,71 @@
+/*
+ ============================================================================
+ Author        : G. Barlas
+ Version       : 1.0, September 2015
+               : 1.1, 16.12.2019 - Comments added (H. Weber)
+               : 1.2, 27.01.2022 - Console output improved (H. Weber, S. Stahl)
+ License       : V 1.0 released under the GNU GPL 3.0
+ Description   : Producer/consumer example using a monitor.
+ To compile    : qmake monitor2ProdCons.pro; make
+ ============================================================================
+ */
+#include "monitor.h"
+#include "producer.h"
+#include "consumer.h"
+
+#include <iostream>
+#include <unistd.h>
+
+
+using namespace std;
+
+
+int produce(int id) {
+    QMutex prodBinSem;
+    static int i = 0;
+
+    prodBinSem.lock();
+    int j = i++;
+    prodBinSem.unlock();
+
+    printf("%3d produced by producer %d\n", j, id); // i to j
+    return j;
+}
+
+void consume(int i, int ID) {
+    printf("%3d consumed by consumer %d\n", i, ID);
+}
+
+int main(int argc, char *argv[]) {
+    Monitor<int> m;
+
+    int N = 4;		// Count of Producer objects (each is one QThread)
+    int M = 3;		// Count of Consumer objects (each is one QThread)
+    int numP = 10;	// Count of produced/consumed items for all Producer/Consumer threads combined
+
+    printf("%d items produced by %d producers and consumed by %d consumers\n", numP, N, M);
+
+    Producer<int>::initClass(numP, &m, &produce);
+    Consumer<int>::initClass(numP, &m, &consume);
+
+    Producer<int> *p[N];
+    Consumer<int> *c[M];
+    //Producers start producing
+    for (int i = 0; i < N; i++) {
+        p[i] = new Producer<int>(i);
+        p[i]->start();
+    }
+    //Consumer start consuming
+    for (int i = 0; i < M; i++) {
+        c[i] = new Consumer<int>(i);
+        c[i]->start();
+    }
+
+
+    for (int i = 0; i < N; i++)
+        p[i]->wait();
+    for (int i = 0; i < M; i++)
+        c[i]->wait();
+
+    return 0;
+}

MonitorExample/monitor.cpp

@@ -0,0 +1,65 @@
+#include "monitor.h"
+
+template<typename T>
+Monitor<T>::Monitor(int n) {
+    buffer = new T[n];
+    for(int i=0;i<n;i++)
+        emptySpotsQ.push(&buffer[i]);	// Initialize emptySpotsQ with buffer places
+}
+//-----------------------------------------
+
+template<typename T>
+Monitor<T>::~Monitor() {
+    delete []buffer;
+}
+//-----------------------------------------
+
+//producer tries to put an item:
+//  lock l
+//  wait for spot in putting queue
+//  get first spot in putting queue
+template<typename T>
+T* Monitor<T>::canPut() {
+    QMutexLocker ml(&l);			// Lock entry controlling mutex l
+    while (emptySpotsQ.size() == 0)	// If no free place in emptySpots queue:
+        full.wait(&l);				// Block until condition variable full is given free by mutex l
+    T *aux = emptySpotsQ.front();	// Get place in emptySpotsQ
+    emptySpotsQ.pop();				// ?? Free one place in emptySpotsQ ?? (documentation of class queue currently not available)
+    return aux;						// Return pointer to place in emptySpotsQ for item to be produced
+}
+//-----------------------------------------
+
+//consumer tries to take an item:
+//  lock l
+//  push produced item to itemQ
+//  wake waiting consumer
+template<typename T>
+T* Monitor<T>::canGet() {
+    QMutexLocker ml(&l);		// Lock entry controlling mutex l
+    while (itemQ.size() == 0)	// If no item in itemQ:
+        empty.wait(&l);			// Block until condition variable empty is given free by mutex l
+    T* temp = itemQ.front();	// Get item out of itemQ
+    itemQ.pop();				// ?? Free one place in itemQ ?? (documentation of class queue currently not available)
+    return temp;				// Return pointer to item in itemQ, which will be consumed
+}
+//-----------------------------------------
+
+//producer is done with producing and wants to put that item:
+//  lock l
+//  push produced item to itemQ
+//  wake waiting consumer
+template<typename T>
+void Monitor<T>::donePutting(T *x) {
+    QMutexLocker ml(&l);	// Lock entry controlling mutex l
+    itemQ.push(x);			// Push place with produced item into itemQ
+    empty.wakeOne();		// Notify one consumer thread waiting for condition variable empty
+}
+//-----------------------------------------
+
+template<typename T>
+void Monitor<T>::doneGetting(T *x) {
+    QMutexLocker ml(&l);	// Lock entry controlling mutex l
+    emptySpotsQ.push(x);	// Give free the buffer place which had hold the consumed item by pushing in into emptySpotsQ
+    full.wakeOne();			// Notify one producer thread waiting for condition variable empty
+}
+//************************************************************

MonitorExample/monitor.h

@@ -0,0 +1,31 @@
+#ifndef MONITOR_H
+#define MONITOR_H
+#include <QMutexLocker>
+#include <QWaitCondition>
+#include <queue>
+
+using namespace std;
+
+const int BUFFSIZE = 100;
+//************************************************************
+
+template<typename T>
+class Monitor {
+private:
+    QMutex l;
+    QWaitCondition full, empty;	// full: Condition variable for queue itemQ.
+                                // empty: Condition variable for queue emptySpotsQ.
+
+    queue<T *> emptySpotsQ;		// Queue of free item buffer places for items to be produced
+    queue<T *> itemQ;			// Queue of item buffer places with items to be consumed
+    T *buffer;					// Item buffer
+public:
+    T* canPut();
+    T* canGet();
+    void donePutting(T *x);
+    void doneGetting(T *x);
+    Monitor(int n = BUFFSIZE);
+    ~Monitor();
+};
+
+#endif // MONITOR_H

MonitorExample/producer.cpp

@@ -0,0 +1,25 @@
+#include "producer.h"
+
+//---------------------------------------
+template<typename T> QSemaphore Producer<T>::numProducts;
+template<typename T> Monitor<T> * Producer<T>::mon;
+template<> int (*Producer<int>::produce)(int) = NULL;
+//---------------------------------------
+
+template<typename T> void Producer<T>::initClass(int numP, Monitor<T> *m, T(*prod)(int)) {
+    mon = m;
+    numProducts.release(numP);
+    produce = prod;
+}
+//---------------------------------------
+
+template<typename T>
+void Producer<T>::run() {
+    while (numProducts.tryAcquire()) {	// While not all numProducts items are produced:
+        T item = (*produce)(ID);			//   Produce one item
+        T* aux = mon->canPut();			//   Get place for item in emptySpotsQ
+        *aux = item;					//   Put item into emptySpotsQ
+        mon->donePutting(aux);			//   Give info to consumer threads
+    }
+}
+//---------------------------------------

MonitorExample/producer.h

@@ -0,0 +1,21 @@
+#ifndef PRODUCER_H
+#define PRODUCER_H
+#include "monitor.h"
+#include <QThread>
+#include <QSemaphore>
+
+template<typename T>
+class Producer : public QThread {
+private:
+    static QSemaphore numProducts;
+    int ID;
+    static Monitor<T> *mon;
+public:
+    static T(*produce)(int);
+    static void initClass(int numP, Monitor<T> *m, T(*prod)(int));
+
+    Producer<T>(int i) : ID(i) {}
+    void run();
+};
+
+#endif // PRODUCER_H