reworked Queue, added TODO, corrected documentation, prepared execution of regular tasks in MainLoop

src/ebusd/bushandler.cpp

@@ -120,7 +120,7 @@ bool ScanRequest::notify(result_t result, SymbolString& slave)
 	}
 
 	// check for remaining secondary messages
-	if (m_messages.empty()) {
+	if (m_messages.empty()) { // TODO appears several times
 		logNotice(lf_bus, "scan completed, retrieved %d answers", m_scanResults->size());
 		return false;
 	}
@@ -153,8 +153,8 @@ result_t BusHandler::sendAndWait(SymbolString& master, SymbolString& slave)
 	ActiveBusRequest request(master, slave);
 
 	for (int sendRetries = m_failedSendRetries + 1; sendRetries >= 0; sendRetries--) {
-		m_nextRequests.add(&request);
+		m_nextRequests.push(&request);
-		bool success = m_finishedRequests.waitRemove(&request);
+		bool success = m_finishedRequests.remove(&request, true);
 		result = success ? request.m_result : RESULT_ERR_TIMEOUT;
 
 		if (result == RESULT_OK) {
@@ -238,7 +238,7 @@ result_t BusHandler::handleSymbol()
 		if (m_currentRequest != NULL)
 			setState(bs_ready, RESULT_ERR_TIMEOUT); // just to be sure an old BusRequest is cleaned up
 		if (m_remainLockCount == 0 && m_currentRequest == NULL) {
-			startRequest = m_nextRequests.next(false);
+			startRequest = m_nextRequests.peek();
 			if (startRequest == NULL && m_pollInterval > 0) { // check for poll/scan
 				time_t now;
 				time(&now);
@@ -254,7 +254,7 @@ result_t BusHandler::handleSymbol()
 						}
 						else {
 							startRequest = request;
-							m_nextRequests.add(request);
+							m_nextRequests.push(request);
 						}
 					}
 				}
@@ -625,13 +625,14 @@ result_t BusHandler::handleSymbol()
 
 	return RESULT_OK;
 }
+
 result_t BusHandler::setState(BusState state, result_t result, bool firstRepetition)
 {
 	if (m_currentRequest != NULL) {
 		if (result == RESULT_ERR_BUS_LOST && m_currentRequest->m_busLostRetries < m_busLostRetries) {
 			logDebug(lf_bus, "%s during %s, retry", getResultCode(result), getStateCode(m_state));
 			m_currentRequest->m_busLostRetries++;
-			m_nextRequests.add(m_currentRequest); // repeat
+			m_nextRequests.push(m_currentRequest); // repeat
 			m_currentRequest = NULL;
 		}
 		else if (state == bs_sendSyn || (result != RESULT_OK && !firstRepetition)) {
@@ -653,12 +654,12 @@ result_t BusHandler::setState(BusState state, result_t result, bool firstRepetit
 			);
 			if (restart) {
 				m_currentRequest->m_busLostRetries = 0;
-				m_nextRequests.add(m_currentRequest);
+				m_nextRequests.push(m_currentRequest);
 			}
 			else if (m_currentRequest->m_deleteOnFinish)
 				delete m_currentRequest;
 			else
-				m_finishedRequests.add(m_currentRequest);
+				m_finishedRequests.push(m_currentRequest);
 
 			m_currentRequest = NULL;
 		}
@@ -666,16 +667,16 @@ result_t BusHandler::setState(BusState state, result_t result, bool firstRepetit
 
 	if (state == bs_noSignal) { // notify all requests
 		m_response.clear(false); // notify with empty response
-		while ((m_currentRequest = m_nextRequests.remove(false)) != NULL) {
+		while ((m_currentRequest = m_nextRequests.pop()) != NULL) {
 			bool restart = m_currentRequest->notify(RESULT_ERR_NO_SIGNAL, m_response);
 			if (restart) { // should not occur with no signal
 				m_currentRequest->m_busLostRetries = 0;
-				m_nextRequests.add(m_currentRequest);
+				m_nextRequests.push(m_currentRequest);
 			}
 			else if (m_currentRequest->m_deleteOnFinish)
 				delete m_currentRequest;
 			else
-				m_finishedRequests.add(m_currentRequest);
+				m_finishedRequests.push(m_currentRequest);
 		}
 	}
 
@@ -832,7 +833,7 @@ result_t BusHandler::startScan(bool full)
 			delete request;
 			return result;
 		}
-		m_nextRequests.add(request);
+		m_nextRequests.push(request);
 	}
 	return RESULT_OK;
 }

src/ebusd/bushandler.h

@@ -25,7 +25,7 @@
 #include "symbol.h"
 #include "result.h"
 #include "device.h"
-#include "wqueue.h"
+#include "queue.h"
 #include "thread.h"
 #include <string>
 #include <vector>
@@ -429,13 +429,13 @@ private:
 	time_t m_lastPoll;
 
 	/** the queue of @a BusRequests that shall be handled. */
-	WQueue<BusRequest*> m_nextRequests;
+	Queue<BusRequest*> m_nextRequests;
 
 	/** the currently handled BusRequest, or NULL. */
 	BusRequest* m_currentRequest;
 
 	/** the queue of @a BusRequests that are already finished. */
-	WQueue<BusRequest*> m_finishedRequests;
+	Queue<BusRequest*> m_finishedRequests;
 
 	/** the offset of the next symbol that needs to be sent from the command or response,
 	 * (only relevant if m_request is set and state is @a bs_command or @a bs_response). */

src/ebusd/main.h

@@ -76,7 +76,8 @@ result_t loadConfigFiles(DataFieldTemplates* templates, MessageMap* messages, bo
  * Load the message definitions from a configuration file matching the scan result.
  * @param templates the @a DataFieldTemplates to load the necessary templates into, or NULL.
  * @param messages the @a MessageMap to load the messages into.
- * @param matchScan the scan @a Message for which to load the configuration file.
+ * @param master the scan master @a SymbolString for which to load the configuration file.
+ * @param slave the scan slave @a SymbolString for which to load the configuration file.
  * @return the result code.
  */
 result_t loadScanConfigFile(DataFieldTemplates* templates, MessageMap* messages, SymbolString& master, SymbolString& slave);

src/ebusd/mainloop.cpp

@@ -101,7 +101,11 @@ void MainLoop::run()
 		string result;
 
 		// pick the next message to handle
-		NetMessage* message = m_netQueue.remove();
+		NetMessage* message = m_netQueue.pop(5);
+		if (message==NULL) {
+			// TODO perform regular tasks
+			continue;
+		}
 		string request = message->getRequest();
 
 		time_t since, until;

src/ebusd/mainloop.h

@@ -58,7 +58,7 @@ public:
 	 * Add a client @a NetMessage to the queue.
 	 * @param message the client @a NetMessage to handle.
 	 */
-	void addMessage(NetMessage* message) { m_netQueue.add(message); }
+	void addMessage(NetMessage* message) { m_netQueue.push(message); }
 
 private:
 
@@ -80,8 +80,8 @@ private:
 	/** the created @a Network instance. */
 	Network* m_network;
 
-	/** the queue for @a NetMessage instances. */
+	/** the @a NetMessage @a Queue. */
-	WQueue<NetMessage*> m_netQueue;
+	Queue<NetMessage*> m_netQueue;
 
 	/** the path for HTML files served by the HTTP port. */
 	string m_htmlPath;

src/ebusd/network.cpp

@@ -129,7 +129,7 @@ void Connection::run()
 
 			// decode client data
 			if (message.add(data)) {
-				m_netQueue->add(&message);
+				m_netQueue->push(&message);
 
 				// wait for result
 				logDebug(lf_network, "[%05d] wait for result", getID());
@@ -153,7 +153,7 @@ void Connection::run()
 }
 
 
-Network::Network(const bool local, const uint16_t port, const uint16_t httpPort, WQueue<NetMessage*>* netQueue)
+Network::Network(const bool local, const uint16_t port, const uint16_t httpPort, Queue<NetMessage*>* netQueue)
 	: m_netQueue(netQueue), m_listening(false)
 {
 	if (local)

src/ebusd/network.h

@@ -22,7 +22,7 @@
 #define NETWORK_H_
 
 #include "tcpsocket.h"
-#include "wqueue.h"
+#include "queue.h"
 #include "notify.h"
 #include "thread.h"
 #include <string>
@@ -214,9 +214,9 @@ public:
 	 * Constructor.
 	 * @param socket the @a TCPSocket for communication.
 	 * @param isHttp whether this is a HTTP message.
-	 * @param netQueue the remote @a WQueue for handling @a NetMessage instances.
+	 * @param netQueue the reference to the @a NetMessage @a Queue.
 	 */
-	Connection(TCPSocket* socket, const bool isHttp, WQueue<NetMessage*>* netQueue)
+	Connection(TCPSocket* socket, const bool isHttp, Queue<NetMessage*>* netQueue)
 		: m_isHttp(isHttp), m_socket(socket), m_netQueue(netQueue)
 		{ m_id = ++m_ids; }
 
@@ -244,8 +244,8 @@ private:
 	/** the @a TCPSocket for communication. */
 	TCPSocket* m_socket;
 
-	/** the remote @a WQueue for handling @a NetMessage instances. */
+	/** the reference to the @a NetMessage @a Queue. */
-	WQueue<NetMessage*>* m_netQueue;
+	Queue<NetMessage*>* m_netQueue;
 
 	/** notification object for shutdown procedure. */
 	Notify m_notify;
@@ -270,9 +270,9 @@ public:
 	 * @param local true to accept connections only for local host.
 	 * @param port the port to listen for command line connections.
 	 * @param httpPort the port to listen for HTTP connections, or 0.
-	 * @param netQueue the remote queue for network messages.
+	 * @param netQueue the reference to the @a NetMessage @a Queue.
 	 */
-	Network(const bool local, const uint16_t port, const uint16_t httpPort, WQueue<NetMessage*>* netQueue);
+	Network(const bool local, const uint16_t port, const uint16_t httpPort, Queue<NetMessage*>* netQueue);
 
 	/**
 	 * destructor.
@@ -293,8 +293,8 @@ private:
 	/** the list of active @a Connection instances. */
 	list<Connection*> m_connections;
 
-	/** the @a MainLoop queue for transferring @a NetMessage instances. */
+	/** the reference to the @a NetMessage @a Queue. */
-	WQueue<NetMessage*>* m_netQueue;
+	Queue<NetMessage*>* m_netQueue;
 
 	/** the command line @a TCPServer instance. */
 	TCPServer* m_tcpServer;

src/lib/utils/queue.h

@@ -0,0 +1,161 @@
+/*
+ * Copyright (C) John Baier 2014-2015 <ebusd@ebusd.eu>
+ *
+ * This file is part of ebusd.
+ *
+ * ebusd is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ebusd is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ebusd. If not, see http://www.gnu.org/licenses/.
+ */
+
+#ifndef LIBUTILS_QUEUE_H_
+#define LIBUTILS_QUEUE_H_
+
+#include <list>
+#include <pthread.h>
+#include <errno.h>
+
+/** \file queue.h */
+
+using namespace std;
+
+/**
+ * Thread safe template class for queuing items.
+ * @param T the item type.
+ */
+template <typename T>
+class Queue
+{
+
+public:
+	/**
+	 * Constructor.
+	 */
+	Queue()
+	{
+		pthread_mutex_init(&m_mutex, NULL);
+		pthread_cond_init(&m_cond, NULL);
+	}
+
+	/**
+	 * Destructor.
+	 */
+	~Queue()
+	{
+		pthread_mutex_destroy(&m_mutex);
+		pthread_cond_destroy(&m_cond);
+	}
+
+private:
+
+	/**
+	 * Hidden copy constructor.
+	 * @param src the object to copy from.
+	 */
+	Queue(const Queue& src);
+
+public:
+
+	/**
+	 * Add an item to the end of queue.
+	 * @param item the item to add.
+	 */
+	void push(T item)
+	{
+		pthread_mutex_lock(&m_mutex);
+		m_queue.push_back(item);
+		pthread_cond_broadcast(&m_cond);
+		pthread_mutex_unlock(&m_mutex);
+	}
+
+	/**
+	 * Remove the first item from the queue optionally waiting for the queue being non-empty.
+	 * @param timeout the maximum time in seconds to wait for the queue being filled, or 0 for no wait.
+	 * @return the item, or NULL if no item is available within the specified time.
+	 */
+	T pop(int timeout=0)
+	{
+		T item;
+		pthread_mutex_lock(&m_mutex);
+		if (timeout>0) {
+			struct timespec t;
+			clock_gettime(CLOCK_REALTIME, &t);
+			t.tv_sec += timeout;
+			while (m_queue.empty()) {
+				if (pthread_cond_timedwait(&m_cond, &m_mutex, &t)==ETIMEDOUT)
+					break;
+			}
+		}
+		if (m_queue.empty())
+			item = NULL;
+		else {
+			item = m_queue.front();
+			m_queue.pop_front();
+		}
+		pthread_mutex_unlock(&m_mutex);
+		return item;
+	}
+
+	/**
+	 * Remove the specified item from the queue optionally waiting for it to appear in the queue.
+	 * @param item the item to remove and optionally wait for.
+	 * @param wait true to wait for the item to appear in the queue.
+	 * @return whether the item was removed.
+	 */
+	bool remove(T item, bool wait=false)
+	{
+		bool ret = false;
+		pthread_mutex_lock(&m_mutex);
+		do {
+			size_t oldSize = m_queue.size();
+			if (oldSize > 0) {
+				m_queue.remove(item);
+				if (m_queue.size() != oldSize) {
+					ret = true;
+					break;
+				}
+			}
+			pthread_cond_wait(&m_cond, &m_mutex);
+		} while (wait);
+		pthread_mutex_unlock(&m_mutex);
+		return ret;
+	}
+
+	/**
+	 * Return the first item in the queue without removing it.
+	 * @return the item, or NULL if no item is available.
+	 */
+	T peek()
+	{
+		T item;
+		pthread_mutex_lock(&m_mutex);
+		if (m_queue.empty())
+			item = NULL;
+		else
+			item = m_queue.front();
+		pthread_mutex_unlock(&m_mutex);
+		return item;
+	}
+
+private:
+	/** the queue itself */
+	list<T> m_queue;
+
+	/** mutex variable for exclusive lock */
+	pthread_mutex_t m_mutex;
+
+	/** condition variable for exclusive lock */
+	pthread_cond_t m_cond;
+
+};
+
+#endif // LIBUTILS_QUEUE_H_