Merge branch 'enhanced_device' of github.com:john30/ebusd into enhanced_device

src/lib/ebus/device.cpp

@@ -40,6 +40,8 @@
 #include <cstdlib>
 #include <cstring>
 #include <fstream>
+#include <ios>
+#include <iomanip>
 #include "lib/ebus/data.h"
 
 namespace ebusd {
@@ -50,16 +52,64 @@ namespace ebusd {
 #define POLLRDHUP 0
 #endif
 
-Device::~Device() {
-  close();
+// ebusd enhanced protocol IDs:
+#define ENH_REQ_INIT ((uint8_t)0x0)
+#define ENH_RES_RESETTED ((uint8_t)0x0)
+#define ENH_REQ_SEND ((uint8_t)0x1)
+#define ENH_RES_RECEIVED ((uint8_t)0x1)
+#define ENH_REQ_START ((uint8_t)0x2)
+#define ENH_RES_STARTED ((uint8_t)0x2)
+#define ENH_RES_FAILED ((uint8_t)0xa)
+#define ENH_RES_ERROR_EBUS ((uint8_t)0xb)
+#define ENH_RES_ERROR_HOST ((uint8_t)0xc)
+
+// ebusd enhanced error codes for the ERROR_* responses
+#define ENH_ERR_FRAMING ((uint8_t)0x00)
+#define ENH_ERR_OVERRUN ((uint8_t)0x01)
+
+#define ENH_BYTE_FLAG ((uint8_t)0x80)
+#define ENH_BYTE_MASK ((uint8_t)0xc0)
+#define ENH_BYTE1 ((uint8_t)0xc0)
+#define ENH_BYTE2 ((uint8_t)0x80)
+#define makeEnhancedSequence(cmd, data) {(uint8_t)(ENH_BYTE1 | ((cmd)<<2) | (((data)&0xc0)>>6)), (uint8_t)(ENH_BYTE2 | ((data)&0x3f))}
+
+Device::Device(const char* name, bool checkDevice, unsigned int latency, bool readOnly, bool initialSend,
+    bool enhancedProto)
+  : m_name(name), m_checkDevice(checkDevice),
+    m_latency(HOST_LATENCY_MS+latency), m_readOnly(readOnly), m_initialSend(initialSend),
+    m_enhancedProto(enhancedProto), m_fd(-1), m_listener(nullptr), m_arbitrationMaster(SYN),
+    m_arbitrationCheck(false), m_bufSize(((MAX_LEN+1+3)/4)*4), m_bufLen(0), m_bufPos(0) {
+  m_buffer = reinterpret_cast<symbol_t*>(malloc(m_bufSize));
+  if (!m_buffer) {
+    m_bufSize = 0;
+  }
 }
 
-Device* Device::create(const char* name, bool checkDevice, bool readOnly, bool initialSend) {
+Device::~Device() {
+  close();
+  if (m_buffer) {
+    free(m_buffer);
+  }
+}
+
+Device* Device::create(const char* name, unsigned int extraLatency, bool checkDevice, bool readOnly, bool initialSend) {
+  bool enhanced = strncmp(name, "enh:", 4) == 0;
+  if (enhanced) {
+    name += 4;
+  }
   if (strchr(name, '/') == nullptr && strchr(name, ':') != nullptr) {
     char* in = strdup(name);
     bool udp = false;
     char* addrpos = in;
     char* portpos = strchr(addrpos, ':');
+    if (!enhanced && portpos >= addrpos+3 && strncmp(addrpos, "enh", 3) == 0) {
+      enhanced = true; // support enhtcp:<ip>:<port> and enhudp:<ip>:<port>
+      addrpos += 3;
+      if (portpos == addrpos) {
+        addrpos++;
+        portpos = strchr(addrpos, ':');
+      }
+    } // else: support enh:<ip>:<port> defaulting to TCP
     if (portpos == addrpos+3 && (strncmp(addrpos, "tcp", 3) == 0 || (udp=(strncmp(addrpos, "udp", 3) == 0)))) {
       addrpos += 4;
       portpos = strchr(addrpos, ':');
@@ -69,7 +119,7 @@ Device* Device::create(const char* name, bool checkDevice, bool readOnly, bool i
       return nullptr;  // invalid protocol or missing port
     }
     result_t result = RESULT_OK;
-    unsigned int port = parseInt(portpos+1, 10, 1, 65535, &result);
+    uint16_t port = (uint16_t)parseInt(portpos+1, 10, 1, 65535, &result);
     if (result != RESULT_OK) {
       free(in);
       return nullptr;  // invalid port
@@ -77,9 +127,31 @@ Device* Device::create(const char* name, bool checkDevice, bool readOnly, bool i
     *portpos = 0;
     char* hostOrIp = strdup(addrpos);
     free(in);
-    return new NetworkDevice(name, hostOrIp, port, readOnly, initialSend, udp);
+    return new NetworkDevice(name, hostOrIp, port, extraLatency, readOnly, initialSend, udp, enhanced);
   }
-  return new SerialDevice(name, checkDevice, readOnly, initialSend);
+  // support enh:/dev/<device>
+  return new SerialDevice(name, checkDevice, extraLatency, readOnly, initialSend, enhanced);
+}
+
+result_t Device::open() {
+  close();
+  return m_bufSize == 0 ? RESULT_ERR_DEVICE : RESULT_OK;
+}
+
+result_t Device::afterOpen() {
+  m_bufLen = 0;
+  if (m_enhancedProto) {
+    symbol_t buf[2] = makeEnhancedSequence(ENH_REQ_INIT, 0); // TODO define additional feature flags
+    if (::write(m_fd, buf, 2) != 2) {
+      return RESULT_ERR_SEND;
+    }
+    if (m_listener != nullptr) {
+      m_listener->notifyStatus(false, "resetting");
+    }
+  } else if (m_initialSend && !write(ESC)) {
+    return RESULT_ERR_SEND;
+  }
+  return RESULT_OK;
 }
 
 void Device::close() {
@@ -87,6 +159,7 @@ void Device::close() {
     ::close(m_fd);
     m_fd = -1;
   }
+  m_bufLen = 0;  // flush read buffer
 }
 
 bool Device::isValid() {
@@ -103,7 +176,7 @@ result_t Device::send(symbol_t value) {
   if (!isValid()) {
     return RESULT_ERR_DEVICE;
   }
-  if (m_readOnly || write(value) != 1) {
+  if (m_readOnly || !write(value)) {
     return RESULT_ERR_SEND;
   }
   if (m_listener != nullptr) {
@@ -112,74 +185,383 @@ result_t Device::send(symbol_t value) {
   return RESULT_OK;
 }
 
-result_t Device::recv(unsigned int timeout, symbol_t* value) {
+/**
+ * the maximum duration in milliseconds to wait for an enhanced sequence to complete after the first part was already
+ * retrieved: 2* (Start+8Bit+Stop+Extra @ 9600Bd)
+ */
+#define ENHANCED_COMPLETE_WAIT_DURATION 3
+
+
+bool Device::cancelRunningArbitration(ArbitrationState* arbitrationState) {
+  if (m_enhancedProto && m_arbitrationMaster != SYN) {
+    *arbitrationState = as_error;
+    m_arbitrationMaster = SYN;
+    m_arbitrationCheck = false;
+    write(SYN, true);
+    return true;
+  }
+  if (m_enhancedProto || m_arbitrationMaster == SYN) {
+    return false;
+  }
+  *arbitrationState = as_error;
+  m_arbitrationMaster = SYN;
+  m_arbitrationCheck = false;
+  return true;
+}
+
+result_t Device::recv(unsigned int timeout, symbol_t* value, ArbitrationState* arbitrationState) {
+  if (m_arbitrationMaster!=SYN) {
+    *arbitrationState = as_running;
+  }
   if (!isValid()) {
+    cancelRunningArbitration(arbitrationState);
     return RESULT_ERR_DEVICE;
   }
-  if (!available() && timeout > 0) {
-    int ret;
-    struct timespec tdiff;
+  bool repeat = false;
+  bool repeated = false;
+  timeout += m_latency;
+  do {
+    repeat = false;
+    bool isAvailable = available();
+    if (!isAvailable && timeout > 0) {
+      int ret;
+      struct timespec tdiff;
 
-    // set select timeout
-    tdiff.tv_sec = timeout/1000000;
-    tdiff.tv_nsec = (timeout%1000000)*1000;
+      // set select timeout
+      tdiff.tv_sec = timeout/1000;
+      tdiff.tv_nsec = (timeout%1000)*1000000;
 
 #ifdef HAVE_PPOLL
-    nfds_t nfds = 1;
-    struct pollfd fds[nfds];
+      nfds_t nfds = 1;
+      struct pollfd fds[nfds];
 
-    memset(fds, 0, sizeof(fds));
+      memset(fds, 0, sizeof(fds));
 
-    fds[0].fd = m_fd;
-    fds[0].events = POLLIN | POLLERR | POLLHUP | POLLRDHUP;
-    ret = ppoll(fds, nfds, &tdiff, nullptr);
-    if (ret >= 0 && fds[0].revents & (POLLERR | POLLHUP | POLLRDHUP)) {
-      ret = -1;
-    }
+      fds[0].fd = m_fd;
+      fds[0].events = POLLIN | POLLERR | POLLHUP | POLLRDHUP;
+      ret = ppoll(fds, nfds, &tdiff, nullptr);
+      if (ret >= 0 && fds[0].revents & (POLLERR | POLLHUP | POLLRDHUP)) {
+        ret = -1;
+      }
 #else
 #ifdef HAVE_PSELECT
-    fd_set readfds, exceptfds;
+      fd_set readfds, exceptfds;
 
-    FD_ZERO(&readfds);
-    FD_ZERO(&exceptfds);
-    FD_SET(m_fd, &readfds);
+      FD_ZERO(&readfds);
+      FD_ZERO(&exceptfds);
+      FD_SET(m_fd, &readfds);
 
-    ret = pselect(m_fd + 1, &readfds, nullptr, &exceptfds, &tdiff, nullptr);
-    if (ret >= 1 && FD_ISSET(m_fd, &exceptfds)) {
-      ret = -1;
-    }
+      ret = pselect(m_fd + 1, &readfds, nullptr, &exceptfds, &tdiff, nullptr);
+      if (ret >= 1 && FD_ISSET(m_fd, &exceptfds)) {
+        ret = -1;
+      }
 #else
-    ret = 1;  // ignore timeout if neither ppoll nor pselect are available
+      ret = 1;  // ignore timeout if neither ppoll nor pselect are available
 #endif
 #endif
-    if (ret == -1) {
-      close();
-      return RESULT_ERR_DEVICE;
+      if (ret == -1) {
+#ifdef DEBUG_RAW_TRAFFIC
+        fprintf(stdout, "poll error %d\n", errno);
+#endif
+        close();
+        cancelRunningArbitration(arbitrationState);
+        return RESULT_ERR_DEVICE;
+      }
+      if (ret == 0) {
+        return RESULT_ERR_TIMEOUT;
+      }
     }
-    if (ret == 0) {
+
+    // directly read byte from device
+    bool incomplete = false;
+    if (!read(value, isAvailable, arbitrationState, &incomplete)) {
+      if (!isAvailable && incomplete && !repeated) {
+        // for a two-byte transfer another poll is needed
+        repeat = true;
+        repeated = true;
+        timeout = m_latency+ENHANCED_COMPLETE_WAIT_DURATION;
+        continue;
+      }
       return RESULT_ERR_TIMEOUT;
     }
+  } while (repeat);
+  if (m_enhancedProto || *value != SYN || m_arbitrationMaster == SYN) {
+    if (m_listener != nullptr) {
+      m_listener->notifyDeviceData(*value, true);
+    }
+    if (!m_enhancedProto && m_arbitrationMaster != SYN) {
+      if (m_arbitrationCheck) {
+        *arbitrationState = *value == m_arbitrationMaster ? as_won : as_lost;
+        m_arbitrationMaster = SYN;
+        m_arbitrationCheck = false;
+      } else {
+        *arbitrationState = m_arbitrationMaster == SYN ? as_none : as_start;
+      }
+    }
+    return RESULT_OK;
   }
-
-  // directly read byte from device
-  ssize_t nbytes = read(value);
-  if (nbytes == 0) {
-    return RESULT_ERR_EOF;
-  }
-  if (nbytes < 0) {
-    close();
-    return RESULT_ERR_DEVICE;
-  }
+  // non-enhanced: arbitration executed by ebusd itself
+  bool wrote = write(m_arbitrationMaster);  // send as fast as possible
   if (m_listener != nullptr) {
     m_listener->notifyDeviceData(*value, true);
   }
+  if (!wrote) {
+    cancelRunningArbitration(arbitrationState);
+    return RESULT_OK;
+  }
+  if (m_listener != nullptr) {
+    m_listener->notifyDeviceData(m_arbitrationMaster, false);
+  }
+  m_arbitrationCheck = true;
+  *arbitrationState = as_running;
+  return RESULT_OK;
+}
+
+result_t Device::startArbitration(symbol_t masterAddress) {
+  if (m_arbitrationCheck) {
+    if (masterAddress != SYN) {
+      return RESULT_ERR_ARB_RUNNING; // should not occur
+    }
+    m_arbitrationCheck = false;
+    m_arbitrationMaster = SYN;
+    if (m_enhancedProto) {
+      // cancel running arbitration
+      if (!write(SYN, true)) {
+        return RESULT_ERR_SEND;
+      }
+    }
+    return RESULT_OK;
+  }
+  if (m_readOnly) {
+    return RESULT_ERR_SEND;
+  }
+  m_arbitrationMaster = masterAddress;
+  if (m_enhancedProto && masterAddress != SYN) {
+    if (!write(masterAddress, true)) {
+      m_arbitrationMaster = SYN;
+      return RESULT_ERR_SEND;
+    }
+    m_arbitrationCheck = true;
+  }
   return RESULT_OK;
 }
 
+bool Device::write(symbol_t value, bool startArbitration) {
+  if (m_enhancedProto) {
+    symbol_t buf[2] = makeEnhancedSequence(startArbitration ? ENH_REQ_START : ENH_REQ_SEND, value);
+    return ::write(m_fd, buf, 2) == 2;
+  }
+  return ::write(m_fd, &value, 1) == 1;
+}
+
+bool Device::available() {
+  if (m_bufLen <= 0) {
+    return false;
+  }
+  if (!m_enhancedProto) {
+    return true;
+  }
+  // peek into the received enhanced proto bytes to determine symbol availability
+  for (size_t pos = 0; pos < m_bufLen; pos++) {
+    symbol_t ch = m_buffer[(pos+m_bufPos)%m_bufSize];
+    if (!(ch&ENH_BYTE_FLAG)) {
+#ifdef DEBUG_RAW_TRAFFIC
+      fprintf(stdout, "raw avail direct\n");
+#endif
+      return true;
+    }
+    if ((ch&ENH_BYTE_MASK) == ENH_BYTE1) {
+      if (pos+1 >= m_bufLen) {
+        return false;
+      }
+      // peek into next byte to check if enhanced sequence is ok
+      ch = m_buffer[(pos+m_bufPos+1)%m_bufSize];
+      if (!(ch&ENH_BYTE_FLAG) || (ch&ENH_BYTE_MASK) != ENH_BYTE2) {
+#ifdef DEBUG_RAW_TRAFFIC
+        fprintf(stdout, "raw avail enhanced following bad\n");
+#endif
+        if (m_listener != nullptr) {
+          m_listener->notifyStatus(true, "unexpected available enhanced following byte 1");
+        }
+        // drop first byte of invalid sequence
+        m_bufPos = (m_bufPos + 1) % m_bufSize;
+        m_bufLen--;
+        pos--;
+        continue;
+      }
+#ifdef DEBUG_RAW_TRAFFIC
+      fprintf(stdout, "raw avail enhanced\n");
+#endif
+      return true;
+    }
+#ifdef DEBUG_RAW_TRAFFIC
+    fprintf(stdout, "raw avail enhanced bad\n");
+#endif
+    if (m_listener != nullptr) {
+      m_listener->notifyStatus(true, "unexpected available enhanced byte 2");
+    }
+    // skip byte from erroneous protocol
+    m_bufPos = (m_bufPos+1)%m_bufSize;
+    m_bufLen--;
+    pos--;
+  }
+  return false;
+}
+
+bool Device::read(symbol_t* value, bool isAvailable, ArbitrationState* arbitrationState, bool* incomplete) {
+  if (!isAvailable) {
+    if (m_bufLen > 0 && m_bufPos != 0) {
+      if (m_bufLen > m_bufSize / 2) {
+        // more than half of input buffer consumed is taken as signal that ebusd is too slow
+        m_bufLen = 0;
+        if (m_listener != nullptr) {
+          m_listener->notifyStatus(true, "buffer overflow");
+        }
+      } else {
+        size_t tail;
+        if (m_bufPos+m_bufLen > m_bufSize) {
+          // move wrapped tail away
+          tail = (m_bufPos+m_bufLen) % m_bufSize;
+          size_t head = m_bufLen-tail;
+          memmove(m_buffer+head, m_buffer, tail);
+        } else {
+          tail = 0;
+        }
+        // move head to first position
+        memmove(m_buffer, m_buffer + m_bufPos, m_bufLen - tail);
+      }
+    }
+    m_bufPos = 0;
+    // fill up the buffer
+    ssize_t size = ::read(m_fd, m_buffer + m_bufLen, m_bufSize - m_bufLen);
+    if (size <= 0) {
+      return false;
+    }
+#ifdef DEBUG_RAW_TRAFFIC
+    fprintf(stdout, "raw <");
+    for (int pos=0; pos<size; pos++) {
+      fprintf(stdout, " %2.2x", m_buffer[m_bufLen+pos]);
+    }
+    fprintf(stdout, "\n");
+#endif
+    m_bufLen += size;
+  }
+  if (!available()) {
+    if (incomplete) {
+      *incomplete = m_enhancedProto && m_bufLen > 0;
+    }
+    return false;
+  }
+  if (!m_enhancedProto) {
+    *value = m_buffer[m_bufPos];
+    m_bufPos = (m_bufPos+1)%m_bufSize;
+    m_bufLen--;
+    return true;
+  }
+  while (m_bufLen > 0) {
+    symbol_t ch = m_buffer[m_bufPos];
+    if (!(ch&ENH_BYTE_FLAG)) {
+      *value = ch;
+      m_bufPos = (m_bufPos+1)%m_bufSize;
+      m_bufLen--;
+      return true;
+    }
+    uint8_t kind = ch&ENH_BYTE_MASK;
+    if (kind == ENH_BYTE1 && m_bufLen<2) {
+      return false; // transfer not complete yet
+    }
+    m_bufPos = (m_bufPos+1)%m_bufSize;
+    m_bufLen--;
+    if (kind == ENH_BYTE2) {
+      if (m_listener != nullptr) {
+        m_listener->notifyStatus(true, "unexpected enhanced byte 2");
+      }
+      return false;
+    }
+    // kind is ENH_BYTE1
+    symbol_t ch2 = m_buffer[m_bufPos];
+    m_bufPos = (m_bufPos + 1) % m_bufSize;
+    m_bufLen--;
+    if ((ch2 & ENH_BYTE_MASK) != ENH_BYTE2) {
+      if (m_listener != nullptr) {
+        m_listener->notifyStatus(true, "missing enhanced byte 2");
+      }
+      return false;
+    }
+    symbol_t data = (symbol_t)(((ch&0x03)<<6) | (ch2&0x3f));
+    symbol_t cmd = (ch>>2)&0xf;
+    switch (cmd) {
+      case ENH_RES_STARTED:
+        *arbitrationState = as_won;
+        if (m_listener != NULL) {
+          m_listener->notifyDeviceData(data, false);
+        }
+        m_arbitrationMaster = SYN;
+        m_arbitrationCheck = false;
+        *value = data;
+        return true;
+      case ENH_RES_FAILED:
+        *arbitrationState = as_lost;
+        if (m_listener != NULL) {
+          m_listener->notifyDeviceData(m_arbitrationMaster, false);
+        }
+        m_arbitrationMaster = SYN;
+        m_arbitrationCheck = false;
+        *value = data;
+        return true;
+      case ENH_RES_RECEIVED:
+        *value = data;
+        return true;
+      case ENH_RES_RESETTED:
+        if (*arbitrationState != as_none) {
+          *arbitrationState = as_error;
+          m_arbitrationMaster = SYN;
+          m_arbitrationCheck = false;
+        }
+        // TODO define additional feature flags
+        if (m_listener != nullptr) {
+          m_listener->notifyStatus(false, "reset");
+        }
+        break;
+      case ENH_RES_ERROR_EBUS:
+      case ENH_RES_ERROR_HOST:
+        if (m_listener != nullptr) {
+          ostringstream stream;
+          stream << (cmd==ENH_RES_ERROR_EBUS ? "eBUS comm error: " : "host comm error: ");
+          switch (data) {
+            case ENH_ERR_FRAMING:
+              stream << "framing";
+              break;
+            case ENH_ERR_OVERRUN:
+              stream << "overrun";
+              break;
+            default:
+              stream << "unknown 0x" << std::setw(2) << std::setfill('0') << std::hex << static_cast<unsigned>(data);
+              break;
+          }
+          string str = stream.str();
+          m_listener->notifyStatus(true, str.c_str());
+        }
+        cancelRunningArbitration(arbitrationState);
+        break;
+      default:
+        if (m_listener != nullptr) {
+          ostringstream stream;
+          stream << "unexpected enhanced command 0x" << std::setw(2) << std::setfill('0') << std::hex << static_cast<unsigned>(cmd);
+          string str = stream.str();
+          m_listener->notifyStatus(true, str.c_str());
+        }
+        return false;
+    }
+  }
+  return false;
+}
+
 
 result_t SerialDevice::open() {
-  if (m_fd != -1) {
-    close();
+  result_t result = Device::open();
+  if (result != RESULT_OK) {
+    return result;
   }
   struct termios newSettings;
 
@@ -223,7 +605,7 @@ result_t SerialDevice::open() {
   // create new settings
   memset(&newSettings, 0, sizeof(newSettings));
 
-  cfsetspeed(&newSettings, B2400);
+  cfsetspeed(&newSettings, m_enhancedProto ? B9600 : B2400);
   newSettings.c_cflag |= (CS8 | CLOCAL | CREAD);
   newSettings.c_lflag &= ~(ICANON | ECHO | ECHOE | ISIG);  // non-canonical mode
   newSettings.c_iflag |= IGNPAR;  // ignore parity errors
@@ -237,7 +619,7 @@ result_t SerialDevice::open() {
   tcflush(m_fd, TCIFLUSH);
 
   // activate new settings of serial device
-  if (tcsetattr(m_fd, TCSAFLUSH, &newSettings)) {
+  if (tcsetattr(m_fd, TCSANOW, &newSettings)) {
     close();
     return RESULT_ERR_DEVICE;
   }
@@ -245,10 +627,7 @@ result_t SerialDevice::open() {
   // set serial device into blocking mode
   fcntl(m_fd, F_SETFL, fcntl(m_fd, F_GETFL) & ~O_NONBLOCK);
 
-  if (m_initialSend && write(ESC) != 1) {
-    return RESULT_ERR_SEND;
-  }
-  return RESULT_OK;
+  return afterOpen();
 }
 
 void SerialDevice::close() {
@@ -282,8 +661,9 @@ void SerialDevice::checkDevice() {
 #endif
 
 result_t NetworkDevice::open() {
-  if (m_fd != -1) {
-    close();
+  result_t result = Device::open();
+  if (result != RESULT_OK) {
+    return result;
   }
   struct sockaddr_in address;
   memset(reinterpret_cast<char*>(&address), 0, sizeof(address));
@@ -341,23 +721,7 @@ result_t NetworkDevice::open() {
     close();
     return RESULT_ERR_GENERIC_IO;
   }
-  if (m_bufSize == 0) {
-    m_bufSize = MAX_LEN+1;
-    m_buffer = reinterpret_cast<symbol_t*>(malloc(m_bufSize));
-    if (!m_buffer) {
-      m_bufSize = 0;
-    }
-  }
-  m_bufLen = 0;
-  if (m_initialSend && write(ESC) != 1) {
-    return RESULT_ERR_SEND;
-  }
-  return RESULT_OK;
-}
-
-void NetworkDevice::close() {
-  m_bufLen = 0;  // flush read buffer
-  Device::close();
+  return afterOpen();
 }
 
 void NetworkDevice::checkDevice() {
@@ -367,33 +731,4 @@ void NetworkDevice::checkDevice() {
   }
 }
 
-bool NetworkDevice::available() {
-  return m_buffer && m_bufLen > 0;
-}
-
-ssize_t NetworkDevice::write(symbol_t value) {
-  m_bufLen = 0;  // flush read buffer
-  return Device::write(value);
-}
-
-ssize_t NetworkDevice::read(symbol_t* value) {
-  if (available()) {
-    *value = m_buffer[m_bufPos];
-    m_bufPos = (m_bufPos+1)%m_bufSize;
-    m_bufLen--;
-    return 1;
-  }
-  if (m_bufSize > 0) {
-    ssize_t size = ::read(m_fd, m_buffer, m_bufSize);
-    if (size <= 0) {
-      return size;
-    }
-    *value = m_buffer[0];
-    m_bufPos = 1;
-    m_bufLen = size-1;
-    return size;
-  }
-  return Device::read(value);
-}
-
 }  // namespace ebusd

src/lib/ebus/device.h

@@ -40,6 +40,26 @@ namespace ebusd {
  * to a file and/or forwarding it to a logging function.
  */
 
+/** the transfer latency of the network device [ms]. */
+#define NETWORK_LATENCY_MS 10
+
+/** the latency of the host [ms]. */
+#ifdef __CYGWIN__
+#define HOST_LATENCY_MS 20
+#else
+#define HOST_LATENCY_MS 0
+#endif
+
+/** the arbitration state handled by @a Device. */
+enum ArbitrationState {
+  as_none,    //!< no arbitration in process
+  as_start,   //!< arbitration start requested
+  as_error,   //!< error while sending master address
+  as_running, //!< arbitration currently running (master address sent, waiting for reception)
+  as_lost,    //!< arbitration lost
+  as_won,     //!< arbitration won
+};
+
 /**
  * Interface for listening to data received on/sent to a device.
  */
@@ -56,6 +76,13 @@ class DeviceListener {
    * @param received @a true on reception, @a false on sending.
    */
   virtual void notifyDeviceData(symbol_t symbol, bool received) = 0;  // abstract
+
+  /**
+   * Called to notify a status message from the device.
+   * @param error true for an error message, false for an info message.
+   * @param message the message string.
+   */
+  virtual void notifyStatus(bool error, const char* message) = 0;  // abstract
 };
 
 
@@ -63,18 +90,20 @@ class DeviceListener {
  * The base class for accessing an eBUS.
  */
 class Device {
- public:
+ protected:
   /**
    * Construct a new instance.
    * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
-   * @param checkDevice whether to regularly check the device availability (only for serial devices).
+   * @param checkDevice whether to regularly check the device availability.
+   * @param latency the bus transfer latency in milliseconds.
    * @param readOnly whether to allow read access to the device only.
    * @param initialSend whether to send an initial @a ESC symbol in @a open().
+   * @param enhancedProto whether to use the ebusd enhanced protocol.
    */
-  Device(const char* name, bool checkDevice, bool readOnly, bool initialSend)
-    : m_name(name), m_checkDevice(checkDevice), m_readOnly(readOnly), m_initialSend(initialSend), m_fd(-1),
-      m_listener(nullptr) {}
+  Device(const char* name, bool checkDevice, unsigned int latency, bool readOnly, bool initialSend,
+      bool enhancedProto=false);
 
+ public:
   /**
    * Destructor.
    */
@@ -83,26 +112,33 @@ class Device {
   /**
    * Factory method for creating a new instance.
    * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   * @param extraLatency the extra bus transfer latency in milliseconds.
    * @param checkDevice whether to regularly check the device availability (only for serial devices).
    * @param readOnly whether to allow read access to the device only.
    * @param initialSend whether to send an initial @a ESC symbol in @a open().
    * @return the new @a Device, or nullptr on error.
    * Note: the caller needs to free the created instance.
    */
-  static Device* create(const char* name, bool checkDevice = true, bool readOnly = false,
-      bool initialSend = false);
+  static Device* create(const char* name, unsigned int extraLatency = 0, bool checkDevice = true,
+      bool readOnly = false, bool initialSend = false);
 
   /**
    * Get the transfer latency of this device.
-   * @return the transfer latency in microseconds.
+   * @return the transfer latency in milliseconds.
    */
-  virtual unsigned int getLatency() const { return 0; }
+  virtual unsigned int getLatency() const { return m_latency; }
 
   /**
    * Open the file descriptor.
    * @return the @a result_t code.
    */
-  virtual result_t open() = 0;  // abstract
+  virtual result_t open();
+
+  /**
+   * Has to be called by subclasses upon successful opening the device as last action in open().
+   * @return the @a result_t code.
+   */
+  result_t afterOpen();
 
   /**
    * Close the file descriptor if opened.
@@ -118,11 +154,27 @@ class Device {
 
   /**
    * Read a single byte from the device.
-   * @param timeout maximum time to wait for the byte in microseconds, or 0 for infinite.
+   * @param timeout maximum time to wait for the byte in milliseconds, or 0 for infinite.
    * @param value the reference in which the received byte value is stored.
+   * @param arbitrationState the reference in which the current @a ArbitrationState is stored on success. When set to
+   * @a as_won, the received byte is the master address that was successfully arbitrated with.
    * @return the result_t code.
    */
-  result_t recv(unsigned int timeout, symbol_t* value);
+  result_t recv(unsigned int timeout, symbol_t* value, ArbitrationState* arbitrationState);
+
+  /**
+   * Start the arbitration with the specified master address. A subsequent request while an arbitration is currently in
+   * checking state will always result in @a RESULT_ERR_DUPLICATE.
+   * @param masterAddress the master address, or @a SYN to cancel a previous arbitration request.
+   * @return the result_t code.
+   */
+  result_t startArbitration(symbol_t masterAddress);
+
+  /**
+   * Return whether the device is currently in arbitration.
+   * @return true when the device is currently in arbitration.
+   */
+  bool isArbitrating() const { return m_arbitrationMaster != SYN; };
 
   /**
    * Return the device name.
@@ -156,37 +208,54 @@ class Device {
   virtual void checkDevice() = 0;  // abstract
 
   /**
-   * Check whether a byte is available immediately (without waiting).
-   * @return true when a a byte is available immediately.
+   * Cancel a running arbitration.
+   * @param arbitrationState the reference in which @a as_error is stored when cancelled.
+   * @return true if it was cancelled, false if not.
    */
-  virtual bool available() { return false; }
+  bool cancelRunningArbitration(ArbitrationState* arbitrationState);
 
   /**
    * Write a single byte.
    * @param value the byte value to write.
-   * @return the number of bytes written, or -1 on error.
+   * @param startArbitration true to start arbitration.
+   * @return true on success, false on error.
    */
-  virtual ssize_t write(symbol_t value) { return ::write(m_fd, &value, 1); }
+  virtual bool write(symbol_t value, bool startArbitration=false);
+
+  /**
+   * Check whether a symbol is available for reading immediately (without waiting).
+   * @return true when a symbol is available for reading immediately.
+   */
+  virtual bool available();
 
   /**
    * Read a single byte.
    * @param value the reference in which the read byte value is stored.
-   * @return the number of bytes read, or -1 on error.
+   * @param isAvailable the result of the immediately preceding call to @a available().
+   * @param arbitrationState the variable in which to store the current/received arbitration state (mandatory for enhanced proto).
+   * @param incomplete the variable in which to store when a partial transfer needs another poll.
+   * @return true on success, false on error.
    */
-  virtual ssize_t read(symbol_t* value) { return ::read(m_fd, value, 1); }
+  virtual bool read(symbol_t* value, bool isAvailable, ArbitrationState* arbitrationState=nullptr, bool* incomplete=nullptr);
 
   /** the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network). */
   const char* m_name;
 
-  /** whether to regularly check the device availability (only for serial devices). */
+  /** whether to regularly check the device availability. */
   const bool m_checkDevice;
 
+  /** the bus transfer latency in milliseconds. */
+  const unsigned int m_latency;
+
   /** whether to allow read access to the device only. */
   const bool m_readOnly;
 
   /** whether to send an initial @a ESC symbol in @a open(). */
   const bool m_initialSend;
 
+  /** whether the device supports the ebusd enhanced protocol. */
+  const bool m_enhancedProto;
+
   /** the opened file descriptor, or -1. */
   int m_fd;
 
@@ -194,8 +263,27 @@ class Device {
  private:
   /** the @a DeviceListener, or nullptr. */
   DeviceListener* m_listener;
+
+  /** the arbitration master address to send when in arbitration, or @a SYN. */
+  symbol_t m_arbitrationMaster;
+
+  /** true when in arbitration and the next received symbol needs to be checked against the sent master address. */
+  bool m_arbitrationCheck;
+
+  /** the read buffer. */
+  symbol_t* m_buffer;
+
+  /** the read buffer size (multiple of 4). */
+  size_t m_bufSize;
+
+  /** the read buffer fill length. */
+  size_t m_bufLen;
+
+  /** the read buffer read position. */
+  size_t m_bufPos;
 };
 
+
 /**
  * The @a Device for directly connected serial interfaces (tty).
  */
@@ -204,12 +292,15 @@ class SerialDevice : public Device {
   /**
    * Construct a new instance.
    * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
-   * @param checkDevice whether to regularly check the device availability (only for serial devices).
+   * @param checkDevice whether to regularly check the device availability.
+   * @param extraLatency the extra bus transfer latency in milliseconds.
    * @param readOnly whether to allow read access to the device only.
    * @param initialSend whether to send an initial @a ESC symbol in @a open().
+   * @param enhancedProto whether to use the ebusd enhanced protocol.
    */
-  SerialDevice(const char* name, bool checkDevice, bool readOnly, bool initialSend)
-    : Device(name, checkDevice, readOnly, initialSend) {}
+  SerialDevice(const char* name, bool checkDevice, unsigned int extraLatency, bool readOnly, bool initialSend,
+      bool enhancedProto=false)
+    : Device(name, checkDevice, extraLatency, readOnly, initialSend, enhancedProto) {}
 
   // @copydoc
   result_t open() override;
@@ -239,48 +330,34 @@ class NetworkDevice : public Device {
    * @param address the socket address of the device.
    * @param hostOrIp the host name or IP address of the device.
    * @param port the TCP or UDP port of the device.
+   * @param extraLatency the extra bus transfer latency in milliseconds.
    * @param readOnly whether to allow read access to the device only.
    * @param initialSend whether to send an initial @a ESC symbol in @a open().
    * @param udp true for UDP, false to TCP.
+   * @param enhancedProto whether to use the ebusd enhanced protocol.
    */
-  NetworkDevice(const char* name, const char* hostOrIp, uint16_t port, bool readOnly, bool initialSend, bool udp)
-    : Device(name, true, readOnly, initialSend), m_hostOrIp(hostOrIp), m_port(port), m_udp(udp),
-      m_buffer(nullptr), m_bufSize(0), m_bufLen(0), m_bufPos(0) {}
+  NetworkDevice(const char* name, const char* hostOrIp, uint16_t port, unsigned int extraLatency, bool readOnly,
+      bool initialSend, bool udp, bool enhancedProto=false)
+    : Device(name, true, NETWORK_LATENCY_MS+extraLatency, readOnly, initialSend, enhancedProto),
+    m_hostOrIp(hostOrIp), m_port(port), m_udp(udp) {}
 
   /**
    * Destructor.
    */
-  virtual ~NetworkDevice() {
+  ~NetworkDevice() override {
     if (m_hostOrIp) {
       free((void*)m_hostOrIp);
     }
-    if (m_buffer) {
-      free(m_buffer);
-    }
   }
 
-  // @copydoc
-  unsigned int getLatency() const override { return 10000; }
-
   // @copydoc
   result_t open() override;
 
-  // @copydoc
-  void close() override;
 
  protected:
   // @copydoc
   void checkDevice() override;
 
-  // @copydoc
-  bool available() override;
-
-  // @copydoc
-  ssize_t write(symbol_t value) override;
-
-  // @copydoc
-  ssize_t read(symbol_t* value) override;
-
 
  private:
   /** the host name or IP address of the device. */
@@ -291,18 +368,6 @@ class NetworkDevice : public Device {
 
   /** true for UDP, false to TCP. */
   const bool m_udp;
-
-  /** the buffer memory, or nullptr. */
-  symbol_t* m_buffer;
-
-  /** the buffer size. */
-  size_t m_bufSize;
-
-  /** the buffer fill length. */
-  size_t m_bufLen;
-
-  /** the buffer read position. */
-  size_t m_bufPos;
 };
 
 }  // namespace ebusd

src/lib/ebus/result.cpp

@@ -44,6 +44,7 @@ const char* getResultCode(result_t resultCode) {
   case RESULT_ERR_DUPLICATE:    return "ERR: duplicate entry";
   case RESULT_ERR_DUPLICATE_NAME: return "ERR: duplicate name";
   case RESULT_ERR_BUS_LOST:     return "ERR: arbitration lost";
+  case RESULT_ERR_ARB_RUNNING:     return "ERR: arbitration running";
   case RESULT_ERR_CRC:          return "ERR: CRC error";
   case RESULT_ERR_ACK:          return "ERR: ACK error";
   case RESULT_ERR_NAK:          return "ERR: NAK received";

src/lib/ebus/result.h

@@ -56,15 +56,16 @@ enum result_t {
   RESULT_ERR_DUPLICATE_NAME = -17,  //!< duplicate entry (name)
 
   RESULT_ERR_BUS_LOST = -18,        //!< arbitration lost
-  RESULT_ERR_CRC = -19,             //!< CRC error
-  RESULT_ERR_ACK = -20,             //!< ACK error
-  RESULT_ERR_NAK = -21,             //!< NAK received
+  RESULT_ERR_ARB_RUNNING = -19,     //!< arbitration running
+  RESULT_ERR_CRC = -20,             //!< CRC error
+  RESULT_ERR_ACK = -21,             //!< ACK error
+  RESULT_ERR_NAK = -22,             //!< NAK received
 
-  RESULT_ERR_NO_SIGNAL = -22,       //!< no signal found on the bus
-  RESULT_ERR_SYN = -23,             //!< SYN received instead of answer
-  RESULT_ERR_SYMBOL = -24,          //!< wrong symbol received instead of sent symbol
+  RESULT_ERR_NO_SIGNAL = -23,       //!< no signal found on the bus
+  RESULT_ERR_SYN = -24,             //!< SYN received instead of answer
+  RESULT_ERR_SYMBOL = -25,          //!< wrong symbol received instead of sent symbol
 
-  RESULT_ERR_NOTAUTHORIZED = -25    //!< not authorized for this action
+  RESULT_ERR_NOTAUTHORIZED = -26    //!< not authorized for this action
 };
 
 

src/lib/ebus/symbol.h

@@ -72,19 +72,19 @@ using std::vector;
 typedef unsigned char symbol_t;
 
 /** escape symbol, either followed by 0x00 for the value 0xA9, or 0x01 for the value 0xAA. */
-#define ESC 0xA9
+#define ESC ((symbol_t)0xA9)
 
 /** synchronization symbol. */
-#define SYN 0xAA
+#define SYN ((symbol_t)0xAA)
 
 /** positive acknowledge symbol. */
-#define ACK 0x00
+#define ACK ((symbol_t)0x00)
 
 /** negative acknowledge symbol. */
-#define NAK 0xFF
+#define NAK ((symbol_t)0xFF)
 
 /** the broadcast destination address. */
-#define BROADCAST 0xFE
+#define BROADCAST ((symbol_t)0xFE)
 
 /**
  * Parse an unsigned int value.

src/lib/ebus/test/CMakeLists.txt

@@ -11,10 +11,6 @@ add_executable(test_filereader test_filereader.cpp)
 target_link_libraries(test_filereader ebus pthread)
 add_test(filereader test_filereader)
 
-add_executable(test_device test_device.cpp)
-target_link_libraries(test_device ebus pthread ${test_LIBS})
-add_test(device test_device)
-
 add_executable(test_symbol test_symbol.cpp)
 target_link_libraries(test_symbol ebus pthread)
 add_test(symbol test_symbol)

src/lib/ebus/test/Makefile.am

@@ -3,7 +3,6 @@ AM_CXXFLAGS = -I$(top_srcdir)/src \
 	      -Wno-unused-parameter
 
 noinst_PROGRAMS = test_filereader \
-		  test_device \
 		  test_symbol \
 		  test_data \
 		  test_message
@@ -11,9 +10,6 @@ noinst_PROGRAMS = test_filereader \
 test_filereader_SOURCES = test_filereader.cpp
 test_filereader_LDADD = ../libebus.a -lpthread
 
-test_device_SOURCES = test_device.cpp
-test_device_LDADD = ../libebus.a -lpthread
-
 test_symbol_SOURCES = test_symbol.cpp
 test_symbol_LDADD = ../libebus.a -lpthread
 
@@ -24,7 +20,6 @@ test_message_SOURCES = test_message.cpp
 test_message_LDADD = ../libebus.a -lpthread
 
 if CONTRIB
-test_device_LDADD += ../contrib/libebuscontrib.a
 test_data_LDADD += ../contrib/libebuscontrib.a
 test_message_LDADD += ../contrib/libebuscontrib.a
 endif

src/lib/utils/log.cpp

@@ -175,6 +175,9 @@ void closeLogFile() {
 }
 
 bool needsLog(const LogFacility facility, const LogLevel level) {
+  if (s_logFile == nullptr && !s_useSyslog) {
+    return false;
+  }
   return s_facilityLogLevel[facility] >= level;
 }
 

src/lib/utils/rotatefile.cpp

@@ -50,6 +50,22 @@ bool RotateFile::setEnabled(bool enabled) {
   if (enabled) {
     m_stream = fopen(m_fileName.c_str(), m_textMode ? "w" : "wb");
     m_fileSize = 0;
+#ifdef FORWARD_RAW_TTY
+    if (!m_textMode && isatty(fileno(m_stream)) == 1) {
+      int fd = fileno(m_stream);
+      struct termios newSettings;
+      memset(&newSettings, 0, sizeof(newSettings));
+
+      cfsetspeed(&newSettings, B2400);
+      newSettings.c_cflag |= (CS8 | CLOCAL);
+      newSettings.c_lflag &= ~(ICANON | ECHO | ECHOE | ISIG);  // non-canonical mode
+      newSettings.c_iflag |= IGNPAR;  // ignore parity errors
+      newSettings.c_oflag &= ~OPOST;
+
+      // activate new settings of serial device
+      tcsetattr(fd, TCSANOW, &newSettings);
+    }
+#endif
   }
   return true;
 }

src/lib/utils/thread.cpp

@@ -84,25 +84,54 @@ WaitThread::~WaitThread() {
 void WaitThread::stop() {
   pthread_mutex_lock(&m_mutex);
   pthread_cond_signal(&m_cond);
-  pthread_mutex_unlock(&m_mutex);
   Thread::stop();
+  pthread_mutex_unlock(&m_mutex);
 }
 
 bool WaitThread::join() {
-  pthread_mutex_lock(&m_mutex);
-  pthread_cond_signal(&m_cond);
-  pthread_mutex_unlock(&m_mutex);
+  stop();
   return Thread::join();
 }
 
 bool WaitThread::Wait(int seconds) {
+  pthread_mutex_lock(&m_mutex);
   struct timespec t;
   clockGettime(&t);
   t.tv_sec += seconds;
-  pthread_mutex_lock(&m_mutex);
   pthread_cond_timedwait(&m_cond, &m_mutex, &t);
   pthread_mutex_unlock(&m_mutex);
   return isRunning();
 }
 
+
+NotifiableThread::NotifiableThread()
+    : WaitThread(), m_notified(false) {
+}
+
+void NotifiableThread::notify() {
+  pthread_mutex_lock(&m_mutex);
+  pthread_cond_signal(&m_cond);
+  m_notified = true;
+  pthread_mutex_unlock(&m_mutex);
+}
+
+bool NotifiableThread::waitNotified(int millis) {
+  pthread_mutex_lock(&m_mutex);
+  if (!m_notified) {
+    struct timespec t;
+    clockGettime(&t);
+    t.tv_sec += millis / 1000000000;
+    t.tv_nsec += (millis % 1000000000) * 1000000;
+    if (t.tv_nsec > 1000000000) {
+      t.tv_sec++;
+      t.tv_nsec -= 1000000000;
+    }
+    pthread_cond_timedwait(&m_cond, &m_mutex, &t);
+  }
+  bool notified = m_notified;
+  m_notified = false;
+  pthread_mutex_unlock(&m_mutex);
+  return notified;
+}
+
 }  // namespace ebusd

src/lib/utils/thread.h

@@ -1,6 +1,6 @@
 /*
  * ebusd - daemon for communication with eBUS heating systems.
- * Copyright (C) 2014-2018 John Baier <ebusd@ebusd.eu>, Roland Jax 2012-2014 <ebusd@liwest.at>
+ * Copyright (C) 2014-2020 John Baier <ebusd@ebusd.eu>, Roland Jax 2012-2014 <ebusd@liwest.at>
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
@@ -26,7 +26,7 @@ namespace ebusd {
 /** \file lib/utils/thread.h */
 
 /**
- * wrapper class for pthread.
+ * Wrapper class for pthread.
  */
 class Thread {
  public:
@@ -82,7 +82,7 @@ class Thread {
   /**
    * Thread entry method to be overridden by derived class.
    */
-  virtual void run() = 0;
+  virtual void run() = 0;  // abstract
 
 
  private:
@@ -134,7 +134,7 @@ class WaitThread : public Thread {
   bool Wait(int seconds);
 
 
- private:
+ protected:
   /** the mutex for waiting. */
   pthread_mutex_t m_mutex;
 
@@ -143,6 +143,35 @@ class WaitThread : public Thread {
 };
 
 
+/**
+ * A @a WaitThread that can be waited on.
+ */
+class NotifiableThread : public WaitThread {
+ public:
+  /**
+   * Constructor.
+   */
+  NotifiableThread();
+
+  /**
+   * Notify another thread currently in @a wait().
+   */
+  void notify();
+
+  /**
+   * Wait for getting notified up to the specified amount of time.
+   * @param millis the maximum number of milliseconds to wait.
+   * @return true if @a notify() was called while waiting.
+   */
+  bool waitNotified(int millis);
+
+
+ private:
+  /** whether @a notify() was called while waiting. */
+  bool m_notified;
+};
+
+
 /**
  * A simple mutex.
  */