From 6b0f58b2917b604f3e75f85c895b195b9e351f65 Mon Sep 17 00:00:00 2001 From: Roland Jax Date: Fri, 7 Nov 2014 20:30:38 +0100 Subject: [PATCH] integration from bus into ebusloop continued. --- src/ebusd/ebusloop.cpp | 614 +++++++++++++++++++++++------------- src/ebusd/ebusloop.h | 32 +- src/ebusd/main.cpp | 8 +- src/lib/ebus/buscommand.cpp | 4 +- src/lib/ebus/port.h | 5 +- src/lib/utils/wqueue.h | 14 + 6 files changed, 435 insertions(+), 242 deletions(-) diff --git a/src/ebusd/ebusloop.cpp b/src/ebusd/ebusloop.cpp index ef3a1098..e429086d 100644 --- a/src/ebusd/ebusloop.cpp +++ b/src/ebusd/ebusloop.cpp @@ -24,43 +24,27 @@ extern LogInstance& L; extern Appl& A; -EBusLoop::EBusLoop(Commands* commands) : m_commands(commands), m_stop(false) +EBusLoop::EBusLoop(Commands* commands) + : m_commands(commands), m_stop(false), m_busLocked(false), m_priorRetry(false) { m_port = new Port(A.getParam("p_device"), A.getParam("p_nodevicecheck")); - m_port->open(); if (m_port->isOpen() == false) L.log(bus, error, "can't open %s", A.getParam("p_device")); - m_dump = new Dump(A.getParam("p_dumpfile"), A.getParam("p_dumpsize")); - m_dumpState = A.getParam("p_dump"); m_logRawData = A.getParam("p_lograwdata"); - //~ m_deviceName = A.getParam("p_device"); + m_pollInterval = A.getParam("p_pollinterval"); - //~ m_bus = new Bus(m_deviceName, - //~ A.getParam("p_nodevicecheck"), - //~ A.getParam("p_recvtimeout"), - //~ A.getParam("p_dumpfile"), - //~ A.getParam("p_dumpsize"), - //~ A.getParam("p_dump")); + m_recvTimeout = A.getParam("p_recvtimeout"); - //~ m_retries = A.getParam("p_retries"); -//~ - //~ m_lookbusretries = A.getParam("p_lookbusretries"); -//~ - //~ m_pollInterval = A.getParam("p_pollinterval"); -//~ + m_sendRetries = A.getParam("p_sendretries"); - - //~ m_bus->connect(); - - //~ if (m_bus->isConnected() == false) - //~ L.log(bus, error, "can't open %s", m_deviceName.c_str()); + m_lockRetries = A.getParam("p_lockretries"); } EBusLoop::~EBusLoop() @@ -70,23 +54,35 @@ EBusLoop::~EBusLoop() delete m_port; delete m_dump; - //~ m_bus->disconnect(); -//~ - //~ if (m_bus->isConnected() == true) - //~ L.log(bus, error, "error during disconnect."); -//~ - //~ delete m_bus; } void* EBusLoop::run() { - bool busLock = false; + int sendRetries = 0; + int lockRetries = 0; + + // polling + time_t pollStart, pollEnd; + time(&pollStart); + double pollDelta; for (;;) { if (m_port->isOpen() == true) { - unsigned char byte; ssize_t numBytes; + // add poll command - timer reached + if (m_commands->sizePolDB() > 0) { + // check polling delta + time(&pollEnd); + pollDelta = difftime(pollEnd, pollStart); + + // add new polling command to send + if (pollDelta >= m_pollInterval) { + addPollCommand(); + time(&pollStart); + } + } + // read device - no timeout needed (AUTO-SYN) numBytes = m_port->recv(0); @@ -95,46 +91,63 @@ void* EBusLoop::run() continue; } - for (int i = 0; i < numBytes; i++) { - - // fetch byte - byte = recvByte(); - - // collect cycle data - if (byte != SYN) - m_sstr.push_back(byte, true, false); - - // unlock bus - if (byte == SYN && busLock == true) { - busLock = false; - L.log(bus, trace, " bus unlocked"); - } - - // analyse cycle data - if (byte == SYN && m_sstr.size() > 0) { - - analyseCycData(m_sstr); - - if (m_sstr.size() == 1) { - busLock = true; - L.log(bus, trace, " bus locked"); - } - - m_sstr.clear(); - } - } + // cycle bytes + collectCycData(numBytes); // send command - if (m_sstr.size() == 0 && busLock == false - && m_sendBuffer.size() > 0) { - // TODO: sendCommand.... + if (m_sstr.size() == 0 && m_busLocked == false && m_sendBuffer.size() > 0) { + // acquire Bus + int busResult = acquireBus(); + + // send bus command + if (busResult == RESULT_BUS_ACQUIRED) { + BusCommand* busCommand = sendCommand(); + L.log(bus, trace, " %s", busCommand->getMessageStr().c_str()); + + if (busCommand->isErrorResult() == true) { + if (sendRetries < m_sendRetries) { + sendRetries++; + L.log(bus, trace, " send retry %d", sendRetries); + busCommand->setResult(std::string(), RESULT_OK); + } + else { + sendRetries = 0; + if (busCommand->isPoll() == true) + delete m_sendBuffer.remove(); + else + busCommand->sendSignal(); + } + } + else { + sendRetries = 0; + if (busCommand->isPoll() == true) { + m_commands->storePolData(busCommand->getMessageStr().c_str()); // TODO use getResult() + delete busCommand; + } + else + busCommand->sendSignal(); + } + + } + else { + L.log(bus, trace, " acquire bus failed"); + if (lockRetries >= m_lockRetries) { + L.log(bus, event, " lock bus failed"); + BusCommand* busCommand = m_sendBuffer.remove(); + if (busCommand->isPoll() == true) + delete busCommand; + else + busCommand->sendSignal(); + + lockRetries = 0; + } + else + lockRetries++; + + } + } - // poll command - timer reached - if (m_sstr.size() == 0 && busLock == false - && m_sendBuffer.size() > 0) { - // TODO: pollCommand.... - } } else { // TODO: define max reopen @@ -158,7 +171,7 @@ void* EBusLoop::run() return NULL; } -unsigned char EBusLoop::recvByte() +unsigned char EBusLoop::fetchByte() { unsigned char byte; @@ -174,11 +187,48 @@ unsigned char EBusLoop::recvByte() return byte; } -void EBusLoop::analyseCycData(SymbolString data) const +void EBusLoop::collectCycData(const int numRecv) { - L.log(bus, trace, "%s", data.getDataStr().c_str()); + // cycle bytes + for (int i = 0; i < numRecv; i++) { - int index = m_commands->storeCycData(data.getDataStr()); + // fetch byte + unsigned char byte = fetchByte(); + + // collect cycle data + if (byte != SYN) + m_sstr.push_back(byte, true, false); + + // unlock bus + if (byte == SYN && m_busLocked == true) { + m_busLocked = false; + L.log(bus, trace, " bus unlocked"); + } + + // analyse cycle data + if (byte == SYN && m_sstr.size() > 0) { + + analyseCycData(); + + if (m_sstr.size() == 1) { + if (m_priorRetry == true) + m_priorRetry = false; + else { + m_busLocked = true; + L.log(bus, trace, " bus locked"); + } + } + + m_sstr.clear(); + } + } +} + +void EBusLoop::analyseCycData() +{ + L.log(bus, trace, "%s", m_sstr.getDataStr().c_str()); + + int index = m_commands->storeCycData(m_sstr.getDataStr()); if (index == -1) { L.log(bus, debug, " command not found"); @@ -198,174 +248,292 @@ void EBusLoop::analyseCycData(SymbolString data) const } } -/* -void* EBusLoop::run() +void EBusLoop::addPollCommand() { - int busResult; - int retries = 0; - int lookbusretries = 0; - bool busCommandActive = false; + int index = m_commands->nextPolCommand(); + if (index < 0) { + L.log(bus, error, "polling index out of range"); + } + else { + // TODO: implement as methode from class commands? + std::string tmp; + tmp += (*m_commands)[index][1]; + tmp += " "; + tmp += (*m_commands)[index][2]; + L.log(bus, event, " polling [%d] %s", index, tmp.c_str()); - // polling - time_t pollStart, pollEnd; - time(&pollStart); - double pollDelta = 0.0; + std::string ebusCommand(A.getParam("p_address")); + ebusCommand += m_commands->getEbusCommand(index); + std::transform(ebusCommand.begin(), ebusCommand.end(), ebusCommand.begin(), tolower); - for (;;) { - if (m_bus->isConnected() == true) { + BusCommand* busCommand = new BusCommand(ebusCommand, true); + L.log(bus, trace, " msg: %s", ebusCommand.c_str()); - // work on bus - busResult = m_bus->proceed(); + addBusCommand(busCommand); + } +} - // new cyc message arrived - if (busResult == RESULT_SYN || busResult == RESULT_BUS_LOCKED) { - SymbolString data = m_bus->getCycData(); +int EBusLoop::acquireBus() +{ + unsigned char recvByte, sendByte; + ssize_t numRecv, numSend; - if (data.size() == 0 && m_logAutoSyn == true) - L.log(bus, trace, "aa"); + sendByte = m_sendBuffer.next()->getCommand()[0]; - if (data.size() != 0) { - L.log(bus, trace, "%s", data.getDataStr().c_str()); + // send QQ + numSend = m_port->send(&sendByte); + if (numSend <= 0) { + L.log(bus, trace, " ERR_SEND: send error"); + return RESULT_ERR_SEND; + } - int index = m_commands->storeCycData(data.getDataStr()); + // receive 1 byte - must be QQ + numRecv = m_port->recv(0); - if (index == -1) { - L.log(bus, debug, " command not found"); + if (numRecv < 0) { + L.log(bus, trace, " ERR_DEVICE: generic device error"); + return RESULT_ERR_DEVICE; + } - } else if (index == -2) { - L.log(bus, debug, " no commands defined"); + if (numRecv == 1) { + // fetch byte + recvByte = fetchByte(); - } else if (index == -3) { - L.log(bus, debug, " search skipped - string too short"); - - } else { - std::string tmp; - tmp += (*m_commands)[index][1]; - tmp += " "; - tmp += (*m_commands)[index][2]; - L.log(bus, event, " cycle [%d] %s", index, tmp.c_str()); - } - } - - if (busResult == RESULT_BUS_LOCKED) - L.log(bus, trace, "bus locked"); - } - - // add new bus command to send - if (busResult == RESULT_SYN && busCommandActive == false && m_sendBuffer.size() != 0) { - BusCommand* busCommand = m_sendBuffer.remove(); - L.log(bus, debug, " msg: %s", busCommand->getCommand().getDataStr().c_str()); - m_bus->addCommand(busCommand); - L.log(bus, debug, " addCommand success"); - busCommandActive = true; - } - - // add new polling command - if (m_commands->sizePolDB() > 0) { - // check polling delta - time(&pollEnd); - pollDelta = difftime(pollEnd, pollStart); - - // add new polling command to send - if (busResult == RESULT_SYN && busCommandActive == false && pollDelta >= m_pollInterval) { - L.log(bus, trace, "polling Intervall reached"); - - int index = m_commands->nextPolCommand(); - if (index < 0) { - L.log(bus, error, "polling index out of range"); - time(&pollStart); - continue; - } - - std::string tmp; - tmp += (*m_commands)[index][1]; - tmp += " "; - tmp += (*m_commands)[index][2]; - L.log(bus, event, " polling [%d] %s", index, tmp.c_str()); - - std::string ebusCommand(A.getParam("p_address")); - ebusCommand += m_commands->getEbusCommand(index); - std::transform(ebusCommand.begin(), ebusCommand.end(), ebusCommand.begin(), tolower); - - BusCommand* busCommand = new BusCommand(ebusCommand, true); - L.log(bus, trace, " msg: %s", ebusCommand.c_str()); - - m_bus->addCommand(busCommand); - L.log(bus, debug, " addCommand success"); - busCommandActive = true; - - time(&pollStart); - } - - } - - // send bus command - if (busResult == RESULT_BUS_ACQUIRED && busCommandActive == true) { - L.log(bus, trace, " getBus success"); - lookbusretries = 0; - BusCommand* busCommand = m_bus->sendCommand(); - L.log(bus, trace, " %s", busCommand->getMessageStr().c_str()); - - if (busCommand->isErrorResult() == true && retries < m_retries) { - retries++; - L.log(bus, trace, " retry number: %d", retries); - busCommand->setResult(std::string(), RESULT_OK); - m_bus->addCommand(busCommand); - } else { - retries = 0; - if (busCommand->isPoll() == true) { - // only save correct results - if (busCommand->isErrorResult() == false) - m_commands->storePolData(busCommand->getMessageStr().c_str()); // TODO use getResult() - - delete busCommand; - } else { - busCommand->sendSignal(); - } - - busCommandActive = false; - } - } - - // get bus retry - if (busResult == RESULT_BUS_PRIOR_RETRY) - L.log(bus, trace, " getBus prior retry"); - - if (busResult == RESULT_ERR_BUS_LOST) { - L.log(bus, trace, " getBus failure"); - if (lookbusretries >= m_lookbusretries) { - L.log(bus, event, " getBus failed - command deleted"); - BusCommand* busCommand = m_bus->delCommand(); - if (busCommand->isPoll() == true) { - delete busCommand; - } else { - busCommand->sendSignal(); - } - lookbusretries = 0; - busCommandActive = false; - }else { - lookbusretries++; - } - } - - if (busResult == RESULT_ERR_SEND) - L.log(bus, event, " getBus send error"); - - } else { - sleep(10); - m_bus->connect(); - - if (m_bus->isConnected() == false) - L.log(bus, error, "can't open %s", m_deviceName.c_str()); + // compare sent and received byte + if (sendByte == recvByte) { + L.log(bus, trace, " bus acquired"); + return RESULT_BUS_ACQUIRED; } - if (m_stop == true) { - m_bus->disconnect(); - return NULL; + // collect cycle data + if (recvByte != SYN) + m_sstr.push_back(recvByte, true, false); + + // compare prior nibble for retry + if ((sendByte & 0x0F) == (recvByte & 0x0F)) { + m_priorRetry = true; + L.log(bus, trace, " bus prior retry"); + return RESULT_BUS_PRIOR_RETRY; + } + + L.log(bus, trace, " ERR_BUS_LOST: lost bus arbitration"); + return RESULT_ERR_BUS_LOST; + } + + // cycle bytes + collectCycData(numRecv); + + L.log(bus, trace, " ERR_BUS_LOST: lost bus arbitration"); + return RESULT_ERR_BUS_LOST; +} + +BusCommand* EBusLoop::sendCommand() +{ + unsigned char recvByte; + std::string result; + SymbolString slaveData; + int retval = RESULT_OK; + + BusCommand* busCommand = m_sendBuffer.next(); + + // send ZZ PB SB NN Dx CRC + SymbolString command = busCommand->getCommand(); + for (size_t i = 1; i < command.size(); i++) { + retval = sendByte(command[i]); + if (retval < 0) + goto on_exit; + } + + // BC -> send SYN + if (busCommand->getType() == broadcast) { + sendByte(SYN); + goto on_exit; + } + + // receive ACK + retval = recvSlaveAck(recvByte); + if (retval < 0) + goto on_exit; + + // is slave ACK negative? + if (recvByte == NAK) { + + // send QQ ZZ PB SB NN Dx CRC again + for (size_t i = 0; i < command.size(); i++) { + retval = sendByte(command[i]); + if (retval < 0) + goto on_exit; + } + + // receive ACK + retval = recvSlaveAck(recvByte); + if (retval < 0) + goto on_exit; + + // is slave ACK negative? + if (recvByte == NAK) { + sendByte(SYN); + L.log(bus, trace, " ERR_NAK: NAK received"); + retval = RESULT_ERR_NAK; + goto on_exit; } } - return NULL; + // MM -> send SYN + if (busCommand->getType() == masterMaster) { + sendByte(SYN); + goto on_exit; + } + + // receive NN, Dx, CRC + retval = recvSlaveData(slaveData); + + // are calculated and received CRC equal? + if (retval == RESULT_ERR_CRC) { + + // send NAK + retval = sendByte(NAK); + if (retval < 0) + goto on_exit; + + // receive NN, Dx, CRC + slaveData.clear(); + retval = recvSlaveData(slaveData); + + // are calculated and received CRC equal? + if (retval == RESULT_ERR_CRC) { + + // send NAK + retval = sendByte(NAK); + if (retval >= 0) + retval = RESULT_ERR_CRC; + } + } + + if (retval < 0) + goto on_exit; + + // send ACK + retval = sendByte(ACK); + if (retval == -1) { + L.log(bus, trace, " ERR_ACK: ACK error"); + retval = RESULT_ERR_ACK; + goto on_exit; + } + + // MS -> send SYN + sendByte(SYN); + +on_exit: + + // empty receive buffer + while (m_port->size() != 0) + recvByte = fetchByte(); + + busCommand->setResult(slaveData, retval); + + if (retval == RESULT_OK) + return m_sendBuffer.remove(); + else + return busCommand; + +} + +int EBusLoop::sendByte(const unsigned char sendByte) +{ + unsigned char recvByte; + ssize_t numRecv, numSend; + + numSend = m_port->send(&sendByte); + + // receive 1 byte - must be equal + numRecv = m_port->recv(RECV_TIMEOUT); + + if (numSend != numRecv) { + L.log(bus, trace, " ERR_EXTRA_DATA: received bytes > sent bytes"); + return RESULT_ERR_EXTRA_DATA; + } + + recvByte = fetchByte(); + + if (sendByte != recvByte) { + L.log(bus, trace, " ERR_SEND: send error"); + return RESULT_ERR_SEND; + } + + return RESULT_OK; +} + +int EBusLoop::recvSlaveAck(unsigned char& recvByte) +{ + ssize_t numRecv; + + // receive ACK + numRecv = m_port->recv(m_recvTimeout); + + if (numRecv > 1) { + L.log(bus, trace, " ERR_EXTRA_DATA: received bytes > sent bytes"); + return RESULT_ERR_EXTRA_DATA; + } + else if (numRecv < 0) { + L.log(bus, trace, " ERR_TIMEOUT: read timeout"); + return RESULT_ERR_TIMEOUT; + } + + recvByte = fetchByte(); + + // is received byte SYN? + if (recvByte == SYN) { + L.log(bus, trace, " ERR_SYN: SYN received"); + return RESULT_ERR_SYN; + } + + return RESULT_OK; +} + +int EBusLoop::recvSlaveData(SymbolString& result) +{ + unsigned char recvByte, calcCrc = 0; + ssize_t numRecv; + size_t NN = 0; + bool updateCrc = true; + int retval = 0; + + for (size_t i = 0, needed = 1; i < needed; i++) { + numRecv = m_port->recv(RECV_TIMEOUT); + if (numRecv < 0) { + L.log(bus, trace, " ERR_TIMEOUT: read timeout"); + return RESULT_ERR_TIMEOUT; + } + + recvByte = fetchByte(); + retval = result.push_back(recvByte, true, updateCrc); + if (retval < 0) + return retval; + + if (retval == RESULT_IN_ESC) + needed++; + else if (result.size() == 1) { // NN received + NN = result[0]; + needed += NN; + } + else if (NN > 0 && result.size() == 1+NN) {// all data received + updateCrc = false; + calcCrc = result.getCRC(); + needed++; + } + } + + if (retval == RESULT_IN_ESC) { + L.log(bus, trace, " ERR_ESC: invalid escape sequence received"); + return RESULT_ERR_ESC; + } + + if (updateCrc == true || calcCrc != result[result.size()-1]) { + L.log(bus, trace, " ERR_CRC: CRC error"); + return RESULT_ERR_CRC; + } + + return RESULT_OK; } -*/ diff --git a/src/ebusd/ebusloop.h b/src/ebusd/ebusloop.h index 245d01f6..d7b4dcdd 100644 --- a/src/ebusd/ebusloop.h +++ b/src/ebusd/ebusloop.h @@ -20,7 +20,6 @@ #ifndef EBUSLOOP_H_ #define EBUSLOOP_H_ -//~ #include "bus.h" #include "commands.h" #include "port.h" #include "dump.h" @@ -28,6 +27,9 @@ #include "wqueue.h" #include "thread.h" +/** the maximum time [us] allowed for retrieving a byte from an addressed slave */ +#define RECV_TIMEOUT 10000 + using namespace libebus; @@ -43,7 +45,6 @@ public: void addBusCommand(BusCommand* busCommand) { m_sendBuffer.add(busCommand); } - //~ void dump(const bool dumpState) { m_bus->setDumpState(dumpState); } void dump(const bool dumpState) { m_dumpState = dumpState; } void newCommands(Commands* commands) { m_commands = commands; } @@ -59,19 +60,26 @@ private: bool m_stop; - SymbolString m_sstr; - - //~ std::string m_deviceName; - //~ bool m_noDeviceCheck; - //~ Bus* m_bus; + bool m_busLocked; + bool m_priorRetry; WQueue m_sendBuffer; - //~ int m_retries; - //~ int m_lookbusretries; - //~ double m_pollInterval; + SymbolString m_sstr; - unsigned char recvByte(); - void analyseCycData(SymbolString data) const; + double m_pollInterval; + long m_recvTimeout; + int m_sendRetries; + int m_lockRetries; + + unsigned char fetchByte(); + void collectCycData(const int numRecv); + void analyseCycData(); + void addPollCommand(); + int acquireBus(); + BusCommand* sendCommand(); + int sendByte(const unsigned char sendByte); + int recvSlaveAck(unsigned char& recvByte); + int recvSlaveData(SymbolString& result); }; diff --git a/src/ebusd/main.cpp b/src/ebusd/main.cpp index cd456d02..7fef5379 100644 --- a/src/ebusd/main.cpp +++ b/src/ebusd/main.cpp @@ -47,12 +47,12 @@ void define_args() "disable valid ebus device test\n", Appl::type_bool, Appl::opt_none); - A.addItem("p_retries", Appl::Param(2), "r", "retries", - "\tnumber retries send ebus command (2)", + A.addItem("p_sendretries", Appl::Param(2), "s", "sendretries", + "number retries send ebus command (2)", Appl::type_int, Appl::opt_mandatory); - A.addItem("p_lookbusretries", Appl::Param(2), "", "lookbusretries", - "number retries to look ebus (2)", + A.addItem("p_lockretries", Appl::Param(2), "", "lockretries", + "number retries to lock ebus (2)", Appl::type_int, Appl::opt_mandatory); A.addItem("p_recvtimeout", Appl::Param(15000), "", "recvtimeout", diff --git a/src/lib/ebus/buscommand.cpp b/src/lib/ebus/buscommand.cpp index e7c35dd0..ac822f63 100644 --- a/src/lib/ebus/buscommand.cpp +++ b/src/lib/ebus/buscommand.cpp @@ -54,9 +54,9 @@ const std::string BusCommand::getMessageStr() result += "00"; result += m_result.getDataStr(); result += "00"; - } else { - result = "success"; } + else + result = "success"; } else result = "error: "+std::string(getResultCodeCStr()); diff --git a/src/lib/ebus/port.h b/src/lib/ebus/port.h index bb45ca4f..7dc8e53a 100644 --- a/src/lib/ebus/port.h +++ b/src/lib/ebus/port.h @@ -31,6 +31,9 @@ namespace libebus /** available device types. */ enum DeviceType { SERIAL, NETWORK }; +/** max bytes write to bus. */ +#define MAX_WRITE_SIZE 1 + /** max size of receive buffer. */ #define MAX_READ_SIZE 100 @@ -218,7 +221,7 @@ public: * @param nbytes number of bytes to send. * @return number of written bytes or -1 if an error has occured. */ - ssize_t send(const unsigned char* buffer, size_t nbytes) + ssize_t send(const unsigned char* buffer, size_t nbytes = MAX_WRITE_SIZE) { return m_device->sendBytes(buffer, nbytes); } /** diff --git a/src/lib/utils/wqueue.h b/src/lib/utils/wqueue.h index 15cca9a0..0509b87e 100644 --- a/src/lib/utils/wqueue.h +++ b/src/lib/utils/wqueue.h @@ -64,6 +64,20 @@ public: return item; } + T next() + { + pthread_mutex_lock(&m_mutex); + + while (m_queue.size() == 0) + pthread_cond_wait(&m_cond, &m_mutex); + + T item = m_queue.front(); + + pthread_mutex_unlock(&m_mutex); + + return item; + } + int size() { pthread_mutex_lock(&m_mutex);