From bd1c991c773c72dab90ea519d312c59c62ab8151 Mon Sep 17 00:00:00 2001 From: john30 Date: Sat, 31 Oct 2015 16:22:13 +0100 Subject: [PATCH] reworked Queue, added TODO, corrected documentation, prepared execution of regular tasks in MainLoop --- src/ebusd/bushandler.cpp | 25 +++--- src/ebusd/bushandler.h | 6 +- src/ebusd/main.h | 3 +- src/ebusd/mainloop.cpp | 6 +- src/ebusd/mainloop.h | 6 +- src/ebusd/network.cpp | 4 +- src/ebusd/network.h | 18 ++--- src/lib/utils/queue.h | 161 +++++++++++++++++++++++++++++++++++++++ 8 files changed, 198 insertions(+), 31 deletions(-) create mode 100644 src/lib/utils/queue.h diff --git a/src/ebusd/bushandler.cpp b/src/ebusd/bushandler.cpp index 69437afc..81c06381 100644 --- a/src/ebusd/bushandler.cpp +++ b/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); - bool success = m_finishedRequests.waitRemove(&request); + m_nextRequests.push(&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; } diff --git a/src/ebusd/bushandler.h b/src/ebusd/bushandler.h index 84ec7e99..87aa619a 100644 --- a/src/ebusd/bushandler.h +++ b/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 #include @@ -429,13 +429,13 @@ private: time_t m_lastPoll; /** the queue of @a BusRequests that shall be handled. */ - WQueue m_nextRequests; + Queue m_nextRequests; /** the currently handled BusRequest, or NULL. */ BusRequest* m_currentRequest; /** the queue of @a BusRequests that are already finished. */ - WQueue m_finishedRequests; + Queue 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). */ diff --git a/src/ebusd/main.h b/src/ebusd/main.h index d7e60089..8421c662 100644 --- a/src/ebusd/main.h +++ b/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); diff --git a/src/ebusd/mainloop.cpp b/src/ebusd/mainloop.cpp index edc02958..316cddcf 100644 --- a/src/ebusd/mainloop.cpp +++ b/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; diff --git a/src/ebusd/mainloop.h b/src/ebusd/mainloop.h index 7e5b8500..a4e7987f 100644 --- a/src/ebusd/mainloop.h +++ b/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. */ - WQueue m_netQueue; + /** the @a NetMessage @a Queue. */ + Queue m_netQueue; /** the path for HTML files served by the HTTP port. */ string m_htmlPath; diff --git a/src/ebusd/network.cpp b/src/ebusd/network.cpp index f5f4be72..710838df 100644 --- a/src/ebusd/network.cpp +++ b/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* netQueue) +Network::Network(const bool local, const uint16_t port, const uint16_t httpPort, Queue* netQueue) : m_netQueue(netQueue), m_listening(false) { if (local) diff --git a/src/ebusd/network.h b/src/ebusd/network.h index b90e3e1e..5db51ce9 100644 --- a/src/ebusd/network.h +++ b/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 @@ -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* netQueue) + Connection(TCPSocket* socket, const bool isHttp, Queue* 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. */ - WQueue* m_netQueue; + /** the reference to the @a NetMessage @a Queue. */ + Queue* 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* netQueue); + Network(const bool local, const uint16_t port, const uint16_t httpPort, Queue* netQueue); /** * destructor. @@ -293,8 +293,8 @@ private: /** the list of active @a Connection instances. */ list m_connections; - /** the @a MainLoop queue for transferring @a NetMessage instances. */ - WQueue* m_netQueue; + /** the reference to the @a NetMessage @a Queue. */ + Queue* m_netQueue; /** the command line @a TCPServer instance. */ TCPServer* m_tcpServer; diff --git a/src/lib/utils/queue.h b/src/lib/utils/queue.h new file mode 100644 index 00000000..2c4d5004 --- /dev/null +++ b/src/lib/utils/queue.h @@ -0,0 +1,161 @@ +/* + * Copyright (C) John Baier 2014-2015 + * + * 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 +#include +#include + +/** \file queue.h */ + +using namespace std; + +/** + * Thread safe template class for queuing items. + * @param T the item type. + */ +template +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 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_