diff --git a/src/ebusd/bushandler.cpp b/src/ebusd/bushandler.cpp index 5c527cb5..8951a03d 100644 --- a/src/ebusd/bushandler.cpp +++ b/src/ebusd/bushandler.cpp @@ -27,6 +27,7 @@ #include #include #include +#include using namespace std; @@ -54,6 +55,62 @@ const char* getStateCode(BusState state, int sendPos) { } +BusRequest::BusRequest(SymbolString& master, SymbolString& slave) + : m_master(master), m_slave(slave), m_finished(false) +{ + pthread_mutex_init(&m_mutex, NULL); + pthread_cond_init(&m_cond, NULL); +} + +BusRequest::~BusRequest() +{ + pthread_mutex_destroy(&m_mutex); + pthread_cond_destroy(&m_cond); +} + +bool BusRequest::wait(int timeout) +{ + struct timespec t; + clock_gettime(CLOCK_REALTIME, &t); + t.tv_sec += timeout; + int result = 0; + + pthread_mutex_lock(&m_mutex); + + while (m_finished == false && result == 0) + result = pthread_cond_timedwait(&m_cond, &m_mutex, &t); + + if (result == 0 && m_finished == false) + result = 1; + + pthread_mutex_unlock(&m_mutex); + + return result == 0; +} + +void BusRequest::notify(bool finished) +{ + pthread_mutex_lock(&m_mutex); + + m_finished = finished; + + pthread_mutex_unlock(&m_mutex); +} + + +result_t BusHandler::sendAndWait(SymbolString& master, SymbolString& slave) +{ + BusRequest* request = new BusRequest(master, slave); + + m_requests.add(request); + bool result = request->wait(5); + if (result == false) + m_requests.remove(request); + delete request; + + return result == true ? RESULT_OK : RESULT_ERR_TIMEOUT; +} + void BusHandler::run() { result_t result = RESULT_OK; @@ -82,14 +139,37 @@ result_t BusHandler::receiveSymbol() { long timeout; ssize_t count; + unsigned char sentSymbol = SYN; + BusRequest* startRequest = NULL; if (m_state == bs_skip) timeout = 0; - else if (m_state == bs_ready) - timeout = SYN_TIMEOUT; - else if (m_sendPos >= 0) + else if (m_sendPos >= 0) { timeout = SLAVE_RECV_TIMEOUT; - else + if (m_sendPos+1 < m_request->m_master.size()) { + m_sendPos++; + sentSymbol = m_request->m_master[m_sendPos]; + if (m_port->send(&sentSymbol) != 1) { + sentSymbol = SYN; // try again later // TODO error: send failed, abort send + m_request->notify(false); + m_request = NULL; + m_sendPos = -1; + } + } + } + else { timeout = SYN_TIMEOUT; + if (m_state == bs_ready && m_request == NULL) { + startRequest = m_requests.next(false); + if (startRequest != NULL) { + // initiate arbitration + sentSymbol = startRequest->m_master[0]; + if (m_port->send(&sentSymbol) != 1) { + sentSymbol = SYN; // try again later // TODO error: send failed + startRequest = NULL; + } + } + } + } count = m_port->recv(timeout, 1); @@ -119,12 +199,24 @@ result_t BusHandler::receiveSymbol() case bs_ready: if (symbol == ESC) return setState(bs_skip, RESULT_ERR_ESC); - + if (m_sendPos < 0 && sentSymbol != SYN) { + // check arbitration + if (symbol == sentSymbol) { // arbitration successful + if (m_requests.remove(startRequest) == false) { + sentSymbol = SYN; // try again later // TODO error: send failed, abort send + } else { + m_request = startRequest; + m_sendPos = 0; + } + } else { // arbitration lost + sentSymbol = SYN; // try again later // TODO error: lost arbitration + } + } result = m_command.push_back(symbol); if (result < RESULT_OK) return setState(bs_skip, result); - return setState(bs_command, result); + return setState(bs_command, RESULT_OK); case bs_command: headerLen = 4; @@ -152,7 +244,7 @@ result_t BusHandler::receiveSymbol() }*/ return setState(bs_commandAck, RESULT_OK); } - return result; + return RESULT_OK; case bs_commandAck: if (symbol == ESC) @@ -201,7 +293,7 @@ result_t BusHandler::receiveSymbol() }*/ return setState(bs_responseAck, RESULT_OK); } - return result; + return RESULT_OK; case bs_responseAck: if (symbol == ESC) diff --git a/src/ebusd/bushandler.h b/src/ebusd/bushandler.h index 89ebdb1c..8323a020 100644 --- a/src/ebusd/bushandler.h +++ b/src/ebusd/bushandler.h @@ -25,10 +25,12 @@ #include "symbol.h" #include "result.h" #include "port.h" +#include "wqueue.h" #include "thread.h" #include #include #include +#include using namespace std; @@ -68,6 +70,50 @@ enum MessageDirection { md_undefined, }; +class BusHandler; + +class BusRequest +{ + friend class BusHandler; +public: + + /** + * @brief Constructor. + */ + BusRequest(SymbolString& master, SymbolString& slave); + + /** + * @brief Destructor. + */ + virtual ~BusRequest(); + + /** + * @brief Wait for notification. + * @return the result code. + */ + bool wait(int timeout); + + /** + * @brief Notify all waiting threads. + */ + void notify(bool finished); + +private: + + SymbolString& m_master; + + SymbolString& m_slave; + + bool m_finished; + + /** a mutex for wait/notify. */ + pthread_mutex_t m_mutex; + + /** a mutex condition for wait/notify. */ + pthread_cond_t m_cond; + +}; + /** * @brief Handles input from and output to the bus with respect to the ebus protocol. @@ -87,13 +133,20 @@ public: unsigned char ownSlaveAddress) : m_port(port), m_messages(messages), m_ownMasterAddress(ownMasterAddress), m_ownSlaveAddress(ownSlaveAddress), m_state(bs_skip), m_repeat(false), - m_sendPos(-1), m_commandCrcValid(false), m_responseCrcValid(false) {} + m_sendPos(-1), m_commandCrcValid(false), m_responseCrcValid(false), m_request(NULL) {} /** * @brief Destructor. */ virtual ~BusHandler() {} + /** + * @brief Send a message on the bus and wait for the answer. + * @param master the @a SymbolString with the master data to send. + * @param slave the @a SymbolString that will be filled with retrieved slave data. + */ + result_t sendAndWait(SymbolString& master, SymbolString& slave); + /** * @brief Main thread entry. */ @@ -157,6 +210,10 @@ private: /** whether the response CRC is valid. */ bool m_responseCrcValid; + WQueue m_requests; + + BusRequest* m_request; + };