epics/pvAccess
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

Merge pull request #144 from mdavidsaver/idletimeout

Browse Source 
TCP Idle Timeout
This commit is contained in:
mdavidsaver
2019-06-26 15:57:37 -07:00
committed by GitHub
parent a2c106378a 05bf0fc453
commit 9eb5843def
21 changed files with 360 additions and 475 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
TODO
View File

@@ -6,6 +6,5 @@
- improve searching of channel with server address specified
- void transportUnresponsive() is not implemented (also in Java)
- complete authNZ (callback on right change)
- request event on disconnect/destroy, etc.?

6
pvtoolsSrc/pvlist.cpp
View File

@@ -120,6 +120,10 @@ bool processSearchResponse(osiSockAddr const & responseFrom, ByteBuffer & receiv
// second byte version
int8 version = receiveBuffer.getByte();
if(version == 0) {
// 0 -> 1 included incompatible changes
return false;
}
// only data for UDP
int8 flags = receiveBuffer.getByte();
@@ -333,7 +337,7 @@ bool discoverServers(double timeOut)
ByteBuffer sendBuffer(buffer, sizeof(buffer)/sizeof(char));
sendBuffer.putByte(PVA_MAGIC);
sendBuffer.putByte(PVA_VERSION);
sendBuffer.putByte(PVA_CLIENT_PROTOCOL_REVISION);
sendBuffer.putByte((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG) ? 0x80 : 0x00); // data + 7-bit endianess
sendBuffer.putByte((int8_t)CMD_SEARCH); // search
sendBuffer.putInt(4+1+3+16+2+1+2); // "zero" payload

9
src/pva/pv/pvaConstants.h
View File

@@ -7,6 +7,8 @@
#ifndef PVACONSTANTS_H_
#define PVACONSTANTS_H_
#include <compilerDependencies.h>
#ifdef epicsExportSharedSymbols
# define pvaConstantsepicsExportSharedSymbols
# undef epicsExportSharedSymbols
@@ -26,11 +28,14 @@ namespace pvAccess {
/** PVA protocol magic number */
const epics::pvData::int8 PVA_MAGIC = static_cast<epics::pvData::int8>(0xCA);
const epics::pvData::int8 PVA_SERVER_PROTOCOL_REVISION = 2;
const epics::pvData::int8 PVA_CLIENT_PROTOCOL_REVISION = 2;
/** PVA protocol revision (implemented by this library). */
const epics::pvData::int8 PVA_PROTOCOL_REVISION = 1;
const epics::pvData::int8 PVA_PROTOCOL_REVISION EPICS_DEPRECATED = 1;
/** PVA version signature used to report this implementation version in header. */
const epics::pvData::int8 PVA_VERSION = PVA_PROTOCOL_REVISION;
const epics::pvData::int8 PVA_VERSION EPICS_DEPRECATED = 1;
/** Default PVA server port. */
const epics::pvData::int32 PVA_SERVER_PORT = 5075;

16
src/remote/abstractResponseHandler.cpp
View File

@@ -4,8 +4,6 @@
* in file LICENSE that is included with this distribution.
*/
#include <sstream>
#include <osiSock.h>
#include <pv/byteBuffer.h>
@@ -15,9 +13,6 @@
#include <pv/remote.h>
#include <pv/hexDump.h>
using std::ostringstream;
using std::hex;
using namespace epics::pvData;
namespace epics {
@@ -44,13 +39,10 @@ void ResponseHandler::handleResponse(osiSockAddr* responseFrom,
char ipAddrStr[48];
ipAddrToDottedIP(&responseFrom->ia, ipAddrStr, sizeof(ipAddrStr));
ostringstream prologue;
prologue<<"Message [0x"<<hex<<(int)command<<", v0x"<<hex;
prologue<<(int)version<<"] received from "<<ipAddrStr<<" on "<<transport->getRemoteName();
hexDump(prologue.str(), _description,
(const int8*)payloadBuffer->getArray(),
payloadBuffer->getPosition(), static_cast<int>(payloadSize));
std::cerr<<"Message [0x"<<std::hex<<(int)command<<", v0x"<<std::hex
<<int(version)<<"] received from "<<ipAddrStr<<" on "<<transport->getRemoteName()
<<" : "<<_description<<"\n"
<<HexDump(*payloadBuffer, payloadSize).limit(0xffff);
}
}
}

17
src/remote/beaconHandler.cpp
View File

@@ -37,8 +37,8 @@ void BeaconHandler::beaconNotify(osiSockAddr* /*from*/, int8 remoteTransportRevi
PVFieldPtr /*data*/)
{
bool networkChanged = updateBeacon(remoteTransportRevision, timestamp, guid, sequentalID, changeCount);
if (networkChanged)
changedTransport();
// TODO: reduce search timers
(void)networkChanged;
}
bool BeaconHandler::updateBeacon(int8 /*remoteTransportRevision*/, TimeStamp* /*timestamp*/,
@@ -83,19 +83,6 @@ bool BeaconHandler::updateBeacon(int8 /*remoteTransportRevision*/, TimeStamp* /*
return false;
}
void BeaconHandler::changedTransport()
{
TransportRegistry::transportVector_t transports;
_context.lock()->getTransportRegistry()->toArray(transports, &_responseFrom);
// notify all
for (TransportRegistry::transportVector_t::iterator iter(transports.begin()), end(transports.end());
iter != end; iter++)
{
(*iter)->changedTransport();
}
}
}
}

48
src/remote/blockingUDPTransport.cpp
View File

@@ -9,6 +9,8 @@
#include <ws2tcpip.h>
#endif
#include <sstream>
#include <sys/types.h>
#include <cstdio>
@@ -25,6 +27,7 @@
#include <pv/inetAddressUtil.h>
#include <pv/logger.h>
#include <pv/likely.h>
#include <pv/hexDump.h>
using namespace epics::pvData;
using namespace std;
@@ -113,6 +116,16 @@ void BlockingUDPTransport::close() {
close(true);
}
void BlockingUDPTransport::ensureData(std::size_t size) {
if (_receiveBuffer.getRemaining() >= size)
return;
std::ostringstream msg;
msg<<"no more data in UDP packet : "
<<_receiveBuffer.getPosition()<<":"<<_receiveBuffer.getLimit()
<<" for "<<size;
throw std::underflow_error(msg.str());
}
void BlockingUDPTransport::close(bool waitForThreadToComplete) {
{
Lock guard(_mutex);
@@ -194,7 +207,7 @@ void BlockingUDPTransport::flushSendQueue()
void BlockingUDPTransport::startMessage(int8 command, size_t /*ensureCapacity*/, int32 payloadSize) {
_lastMessageStartPosition = _sendBuffer.getPosition();
_sendBuffer.putByte(PVA_MAGIC);
_sendBuffer.putByte(PVA_VERSION);
_sendBuffer.putByte((_clientServerWithEndianFlag&0x40) ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION);
_sendBuffer.putByte(_clientServerWithEndianFlag);
_sendBuffer.putByte(command); // command
_sendBuffer.putInt(payloadSize);
@@ -255,13 +268,18 @@ void BlockingUDPTransport::run() {
try {
processBuffer(thisTransport, fromAddress, &_receiveBuffer);
} catch(std::exception& e) {
LOG(logLevelError,
"an exception caught while in UDP receiveThread at %s:%d: %s",
__FILE__, __LINE__, e.what());
} catch (...) {
LOG(logLevelError,
"unknown exception caught while in UDP receiveThread at %s:%d.",
__FILE__, __LINE__);
if(IS_LOGGABLE(logLevelError)) {
char strBuffer[64];
sockAddrToDottedIP(&fromAddress.sa, strBuffer, sizeof(strBuffer));
size_t epos = _receiveBuffer.getPosition();
// of course _receiveBuffer _may_ have been modified during processing...
_receiveBuffer.setPosition(RECEIVE_BUFFER_PRE_RESERVE);
_receiveBuffer.setLimit(RECEIVE_BUFFER_PRE_RESERVE+bytesRead);
std::cerr<<"Error on UDP RX "<<strBuffer<<" -> "<<_remoteName<<" at "<<epos<<" : "<<e.what()<<"\n"
<<HexDump(_receiveBuffer).limit(256u);
}
}
}
} else {
@@ -323,6 +341,10 @@ bool BlockingUDPTransport::processBuffer(Transport::shared_pointer const & trans
// second byte version
int8 version = receiveBuffer->getByte();
if(version==0) {
// 0 -> 1 included incompatible changes
return false;
}
int8 flags = receiveBuffer->getByte();
if (flags & 0x80)
@@ -558,6 +580,8 @@ void initializeUDPTransports(bool serverFlag,
{
BlockingUDPConnector connector(serverFlag);
const int8_t protoVer = serverFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION;
//
// Create UDP transport for sending (to all network interfaces)
//
@@ -568,7 +592,7 @@ void initializeUDPTransports(bool serverFlag,
anyAddress.ia.sin_port = htons(0);
anyAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
sendTransport = connector.connect(responseHandler, anyAddress, PVA_PROTOCOL_REVISION);
sendTransport = connector.connect(responseHandler, anyAddress, protoVer);
if (!sendTransport)
{
THROW_BASE_EXCEPTION("Failed to initialize UDP transport.");
@@ -708,7 +732,7 @@ void initializeUDPTransports(bool serverFlag,
listenLocalAddress.ia.sin_addr.s_addr = node.addr.ia.sin_addr.s_addr;
BlockingUDPTransport::shared_pointer transport = connector.connect(
responseHandler, listenLocalAddress, PVA_PROTOCOL_REVISION);
responseHandler, listenLocalAddress, protoVer);
if (!transport)
continue;
listenLocalAddress = transport->getRemoteAddress();
@@ -742,7 +766,7 @@ void initializeUDPTransports(bool serverFlag,
bcastAddress.ia.sin_port = htons(listenPort);
bcastAddress.ia.sin_addr.s_addr = node.bcast.ia.sin_addr.s_addr;
transport2 = connector.connect(responseHandler, bcastAddress, PVA_PROTOCOL_REVISION);
transport2 = connector.connect(responseHandler, bcastAddress, protoVer);
if (transport2)
{
/* The other wrinkle is that nothing should be sent from this second
@@ -802,7 +826,7 @@ void initializeUDPTransports(bool serverFlag,
#else
anyAddress,
#endif
PVA_PROTOCOL_REVISION);
protoVer);
if (!localMulticastTransport)
throw std::runtime_error("Failed to bind UDP socket.");

2
src/remote/channelSearchManager.cpp
View File

@@ -206,7 +206,7 @@ void ChannelSearchManager::initializeSendBuffer()
// new buffer
m_sendBuffer.clear();
m_sendBuffer.putByte(PVA_MAGIC);
m_sendBuffer.putByte(PVA_VERSION);
m_sendBuffer.putByte(PVA_CLIENT_PROTOCOL_REVISION);
m_sendBuffer.putByte((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG) ? 0x80 : 0x00); // data + 7-bit endianess
m_sendBuffer.putByte(CMD_SEARCH);
m_sendBuffer.putInt(4+1+3+16+2+1); // "zero" payload

222
src/remote/codec.cpp
View File

@@ -16,6 +16,7 @@
#include <epicsTime.h>
#include <epicsThread.h>
#include <epicsVersion.h>
#include <errlog.h>
#include <pv/byteBuffer.h>
#include <pv/pvType.h>
@@ -94,7 +95,7 @@ AbstractCodec::AbstractCodec(
_remoteTransportSocketReceiveBufferSize(MAX_TCP_RECV), _totalBytesSent(0),
_senderThread(0),
_writeMode(PROCESS_SEND_QUEUE),
_writeOpReady(false),_lowLatency(false),
_writeOpReady(false),
_socketBuffer(bufSizeSelect(receiveBufferSize)),
_sendBuffer(bufSizeSelect(sendBufferSize)),
//PRIVATE
@@ -147,7 +148,12 @@ void AbstractCodec::processHeader() {
int8_t magicCode = _socketBuffer.getByte();
// version
_version = _socketBuffer.getByte();
int8_t ver = _socketBuffer.getByte();
if(_version!=ver) {
// enable timeout if both ends support
_version = ver;
setRxTimeout(getRevision()>1);
}
// flags
_flags = _socketBuffer.getByte();
@@ -159,12 +165,12 @@ void AbstractCodec::processHeader() {
_payloadSize = _socketBuffer.getInt();
// check magic code
if (magicCode != PVA_MAGIC)
if (magicCode != PVA_MAGIC || _version==0)
{
LOG(logLevelError,
"Invalid header received from the client at %s:%d: %s.,"
" disconnecting...",
__FILE__, __LINE__, inetAddressToString(*getLastReadBufferSocketAddress()).c_str());
"Invalid header received from the client : %s %02x%02x%02x%02x disconnecting...",
inetAddressToString(*getLastReadBufferSocketAddress()).c_str(),
unsigned(magicCode), unsigned(_version), unsigned(_flags), unsigned(_command));
invalidDataStreamHandler();
throw invalid_data_stream_exception("invalid header received");
}
@@ -185,12 +191,6 @@ void AbstractCodec::processReadNormal() {
return;
}
/*
hexDump("Header", (const int8*)_socketBuffer.getArray(),
_socketBuffer.getPosition(), PVA_MESSAGE_HEADER_SIZE);
*/
// read header fields
processHeader();
bool isControl = ((_flags & 0x01) == 0x01);
@@ -597,7 +597,7 @@ void AbstractCodec::startMessage(
PVA_MESSAGE_HEADER_SIZE + ensureCapacity + _nextMessagePayloadOffset);
_lastMessageStartPosition = _sendBuffer.getPosition();
_sendBuffer.putByte(PVA_MAGIC);
_sendBuffer.putByte(PVA_VERSION);
_sendBuffer.putByte(_clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION);
_sendBuffer.putByte(
(_lastSegmentedMessageType | _byteOrderFlag | _clientServerFlag)); // data message
_sendBuffer.putByte(command); // command
@@ -618,7 +618,7 @@ void AbstractCodec::putControlMessage(
std::numeric_limits<size_t>::max(); // TODO revise this
ensureBuffer(PVA_MESSAGE_HEADER_SIZE);
_sendBuffer.putByte(PVA_MAGIC);
_sendBuffer.putByte(PVA_VERSION);
_sendBuffer.putByte(_clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION);
_sendBuffer.putByte((0x01 | _byteOrderFlag | _clientServerFlag)); // control message
_sendBuffer.putByte(command); // command
_sendBuffer.putInt(data); // data
@@ -797,14 +797,6 @@ void AbstractCodec::send(ByteBuffer *buffer)
//int p = buffer.position();
int bytesSent = write(buffer);
/*
if (IS_LOGGABLE(logLevelTrace)) {
hexDump(std::string("AbstractCodec::send WRITE"),
(const int8 *)buffer->getArray(),
buffer->getPosition(), buffer->getRemaining());
}
*/
if (bytesSent < 0)
{
// connection lost
@@ -934,10 +926,7 @@ void AbstractCodec::enqueueSendRequest(
processSender(sender);
if (_sendBuffer.getPosition() > 0)
{
if (_lowLatency)
flush(true);
else
scheduleSend();
scheduleSend();
}
}
else
@@ -1132,6 +1121,10 @@ void BlockingTCPTransportCodec::receiveThread()
*/
Transport::shared_pointer ptr(this->shared_from_this());
// initially enable timeout for all clients to weed out
// impersonators (security scanners?)
setRxTimeout(true);
while (this->isOpen())
{
try {
@@ -1183,6 +1176,27 @@ void BlockingTCPTransportCodec::sendThread()
_sendQueue.clear();
}
void BlockingTCPTransportCodec::setRxTimeout(bool ena)
{
double timeout = !ena ? 0.0 : std::max(0.0, _context->getConfiguration()->getPropertyAsDouble("EPICS_PVA_CONN_TMO", 30.0));
#ifdef _WIN32
DWORD timo = DWORD(timeout*1000); // in milliseconds
#else
timeval timo;
timo.tv_sec = unsigned(timeout);
timo.tv_usec = (timeout-timo.tv_sec)*1e6;
#endif
int ret = setsockopt(_channel, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timo, sizeof(timo));
if(ret==-1) {
int err = SOCKERRNO;
static int lasterr;
if(err!=lasterr) {
errlogPrintf("%s: Unable to set RX timeout: %d\n", _socketName.c_str(), err);
lasterr = err;
}
}
}
void BlockingTCPTransportCodec::sendBufferFull(int tries) {
// TODO constants
@@ -1302,28 +1316,35 @@ int BlockingTCPTransportCodec::read(epics::pvData::ByteBuffer* dst) {
// NOTE: do not log here, you might override SOCKERRNO relevant to recv() operation above
/*
if (IS_LOGGABLE(logLevelTrace)) {
hexDump(std::string("READ"),
(const int8 *)(dst->getArray()+pos), bytesRead);
}
*/
if(unlikely(bytesRead<=0)) {
if (bytesRead<0)
{
int socketError = SOCKERRNO;
// TODO SOCK_ENOBUFS, for read?
// interrupted or timeout
if (socketError == SOCK_EINTR ||
socketError == EAGAIN ||
socketError == SOCK_EWOULDBLOCK)
continue;
}
if(unlikely(bytesRead==0)) {
return -1; // 0 means connection loss for blocking transport, notify codec by returning -1
} else if(unlikely(bytesRead<0)) {
int err = SOCKERRNO;
if(err==SOCK_EINTR) {
// interrupted by signal. Retry
continue;
} else if(err==SOCK_EWOULDBLOCK || err==EAGAIN || err==SOCK_EINPROGRESS
|| err==SOCK_ETIMEDOUT
|| err==SOCK_ECONNABORTED || err==SOCK_ECONNRESET
) {
// different ways of saying timeout.
// Linux: EAGAIN or EWOULDBLOCK, or EINPROGRESS
// WIN32: WSAETIMEDOUT
// others that RSRV checks for, but may not need to, ECONNABORTED, ECONNRESET
// Note: with windows, after ETIMEOUT leaves the socket in an undefined state.
// so it must be closed. (cf. SO_RCVTIMEO)
return -1;
} else {
// some other (fatal) error
errlogPrintf("%s : Connection closed with RX socket error %d\n", _socketName.c_str(), err);
return -1;
}
}
dst->setPosition(dst->getPosition() + bytesRead);
@@ -1493,7 +1514,7 @@ void BlockingServerTCPTransportCodec::send(ByteBuffer* buffer,
ensureBuffer(PVA_MESSAGE_HEADER_SIZE);
buffer->putByte(PVA_MAGIC);
buffer->putByte(PVA_VERSION);
buffer->putByte(PVA_SERVER_PROTOCOL_REVISION);
buffer->putByte(
0x01 | 0x40 | ((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG)
? 0x80 : 0x00)); // control + server + endian
@@ -1675,24 +1696,25 @@ BlockingClientTCPTransportCodec::BlockingClientTCPTransportCodec(
int16_t priority ) :
BlockingTCPTransportCodec(false, context, channel, responseHandler,
sendBufferSize, receiveBufferSize, priority),
_connectionTimeout(heartbeatInterval*1000),
_unresponsiveTransport(false),
_verifyOrEcho(true)
_connectionTimeout(heartbeatInterval),
_verifyOrEcho(true),
sendQueued(true) // don't start sending echo until after auth complete
{
// initialize owners list, send queue
acquire(client);
// use immediate for clients
//setFlushStrategy(DELAYED);
// setup connection timeout timer (watchdog) - moved to start() method
epicsTimeGetCurrent(&_aliveTimestamp);
}
void BlockingClientTCPTransportCodec::start()
{
TimerCallbackPtr tcb = std::tr1::dynamic_pointer_cast<TimerCallback>(shared_from_this());
_context->getTimer()->schedulePeriodic(tcb, _connectionTimeout, _connectionTimeout);
// add some randomness to our timer phase
double R = float(rand())/RAND_MAX; // [0, 1]
// shape a bit
R = R*0.5 + 0.5; // [0.5, 1.0]
_context->getTimer()->schedulePeriodic(tcb, _connectionTimeout/2.0*R, _connectionTimeout/2.0);
BlockingTCPTransportCodec::start();
}
@@ -1707,46 +1729,22 @@ BlockingClientTCPTransportCodec::~BlockingClientTCPTransportCodec() {
void BlockingClientTCPTransportCodec::callback() {
epicsTimeStamp currentTime;
epicsTimeGetCurrent(&currentTime);
_mutex.lock();
// no exception expected here
double diff = epicsTimeDiffInSeconds(&currentTime, &_aliveTimestamp);
_mutex.unlock();
if(diff>((3*_connectionTimeout)/2)) {
unresponsiveTransport();
}
// use some k (3/4) to handle "jitter"
else if(diff>=((3*_connectionTimeout)/4)) {
// send echo
TransportSender::shared_pointer transportSender = std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
enqueueSendRequest(transportSender);
void BlockingClientTCPTransportCodec::callback()
{
{
Guard G(_mutex);
if(sendQueued) return;
sendQueued = true;
}
// send echo
TransportSender::shared_pointer transportSender = std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
enqueueSendRequest(transportSender);
}
#define EXCEPTION_GUARD(code) try { code; } \
catch (std::exception &e) { LOG(logLevelError, "Unhandled exception caught from code at %s:%d: %s", __FILE__, __LINE__, e.what()); } \
catch (...) { LOG(logLevelError, "Unhandled exception caught from code at %s:%d.", __FILE__, __LINE__); }
void BlockingClientTCPTransportCodec::unresponsiveTransport() {
Lock lock(_mutex);
if(!_unresponsiveTransport) {
_unresponsiveTransport = true;
TransportClientMap_t::iterator it = _owners.begin();
for(; it!=_owners.end(); it++) {
ClientChannelImpl::shared_pointer client = it->second.lock();
if (client)
{
EXCEPTION_GUARD(client->transportUnresponsive());
}
}
}
}
bool BlockingClientTCPTransportCodec::acquire(ClientChannelImpl::shared_pointer const & client) {
Lock lock(_mutex);
if(isClosed()) return false;
@@ -1820,48 +1818,18 @@ void BlockingClientTCPTransportCodec::release(pvAccessID clientID) {
}
}
void BlockingClientTCPTransportCodec::aliveNotification() {
Lock guard(_mutex);
epicsTimeGetCurrent(&_aliveTimestamp);
if(_unresponsiveTransport) responsiveTransport();
}
void BlockingClientTCPTransportCodec::responsiveTransport() {
Lock lock(_mutex);
if(_unresponsiveTransport) {
_unresponsiveTransport = false;
Transport::shared_pointer thisSharedPtr = shared_from_this();
TransportClientMap_t::iterator it = _owners.begin();
for(; it!=_owners.end(); it++) {
ClientChannelImpl::shared_pointer client = it->second.lock();
if (client)
{
EXCEPTION_GUARD(client->transportResponsive(thisSharedPtr));
}
}
}
}
void BlockingClientTCPTransportCodec::changedTransport() {
_outgoingIR.reset();
Lock lock(_mutex);
TransportClientMap_t::iterator it = _owners.begin();
for(; it!=_owners.end(); it++) {
ClientChannelImpl::shared_pointer client = it->second.lock();
if (client)
{
EXCEPTION_GUARD(client->transportChanged());
}
}
}
void BlockingClientTCPTransportCodec::send(ByteBuffer* buffer,
TransportSendControl* control) {
if(_verifyOrEcho) {
TransportSendControl* control)
{
bool voe;
{
Guard G(_mutex);
sendQueued = false;
voe = _verifyOrEcho;
_verifyOrEcho = false;
}
if(voe) {
/*
* send verification response message
*/

4
src/remote/pv/beaconHandler.h
View File

@@ -101,10 +101,6 @@ private:
ServerGUID const &guid,
epics::pvData::int16 sequentalID,
epics::pvData::int16 changeCount);
/**
* Changed transport (server restarted) notify.
*/
void changedTransport();
};
}

13
src/remote/pv/blockingUDP.h
View File

@@ -97,14 +97,6 @@ public:
// noop for UDP (limited by 64k; MAX_UDP_SEND for PVA)
}
virtual void aliveNotification() OVERRIDE FINAL {
// noop
}
virtual void changedTransport() OVERRIDE FINAL {
// noop
}
virtual bool verify(epics::pvData::int32 /*timeoutMs*/) OVERRIDE FINAL {
// noop
return true;
@@ -135,10 +127,7 @@ public:
virtual void close() OVERRIDE FINAL;
virtual void ensureData(std::size_t size) OVERRIDE FINAL {
if (_receiveBuffer.getRemaining() < size)
throw std::underflow_error("no more data in UDP packet");
}
virtual void ensureData(std::size_t size) OVERRIDE FINAL;
virtual void alignData(std::size_t alignment) OVERRIDE FINAL {
_receiveBuffer.align(alignment);

57
src/remote/pv/codec.h
View File

@@ -240,12 +240,19 @@ public:
return _sendQueue.empty();
}
epics::pvData::int8 getRevision() const {
epicsGuard<epicsMutex> G(_mutex);
int8_t myver = _clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION;
return myver < _version ? myver : _version;
}
protected:
virtual void sendBufferFull(int tries) = 0;
void send(epics::pvData::ByteBuffer *buffer);
void flushSendBuffer();
virtual void setRxTimeout(bool ena) {}
ReadMode _readMode;
int8_t _version;
@@ -259,7 +266,6 @@ protected:
epicsThreadId _senderThread;
WriteMode _writeMode;
bool _writeOpReady;
bool _lowLatency;
epics::pvData::ByteBuffer _socketBuffer;
epics::pvData::ByteBuffer _sendBuffer;
@@ -289,7 +295,9 @@ private:
std::size_t _nextMessagePayloadOffset;
epics::pvData::int8 _byteOrderFlag;
epics::pvData::int8 _clientServerFlag;
protected:
const epics::pvData::int8 _clientServerFlag;
private:
const size_t _socketSendBufferSize;
public:
@@ -327,7 +335,7 @@ public:
virtual void waitJoin() OVERRIDE FINAL;
virtual bool terminated() OVERRIDE FINAL;
virtual bool isOpen() OVERRIDE FINAL;
void start();
virtual void start();
virtual int read(epics::pvData::ByteBuffer* dst) OVERRIDE FINAL;
virtual int write(epics::pvData::ByteBuffer* src) OVERRIDE FINAL;
@@ -363,12 +371,6 @@ public:
return _socketName;
}
epics::pvData::int8 getRevision() const {
epicsGuard<epicsMutex> G(_mutex);
return PVA_PROTOCOL_REVISION < _version
? PVA_PROTOCOL_REVISION : _version;
}
virtual std::size_t getReceiveBufferSize() const OVERRIDE FINAL {
return _socketBuffer.getSize();
@@ -435,6 +437,8 @@ private:
void sendThread();
protected:
virtual void setRxTimeout(bool ena) OVERRIDE FINAL;
virtual void sendBufferFull(int tries) OVERRIDE FINAL;
/**
@@ -515,8 +519,6 @@ public:
virtual void release(pvAccessID /*clientId*/) OVERRIDE FINAL {}
virtual void changedTransport() OVERRIDE {}
pvAccessID preallocateChannelSID();
void depreallocateChannelSID(pvAccessID /*sid*/) {}
@@ -554,10 +556,6 @@ public:
BlockingTCPTransportCodec::verified(status);
}
virtual void aliveNotification() OVERRIDE FINAL {
// noop on server-side
}
void authNZInitialize(const std::string& securityPluginName,
const epics::pvData::PVStructure::shared_pointer& data);
@@ -642,7 +640,7 @@ public:
public:
void start();
virtual void start() OVERRIDE FINAL;
virtual ~BlockingClientTCPTransportCodec() OVERRIDE FINAL;
@@ -656,10 +654,6 @@ public:
virtual void release(pvAccessID clientId) OVERRIDE FINAL;
virtual void changedTransport() OVERRIDE FINAL;
virtual void aliveNotification() OVERRIDE FINAL;
virtual void send(epics::pvData::ByteBuffer* buffer,
TransportSendControl* control) OVERRIDE FINAL;
@@ -685,34 +679,17 @@ private:
/**
* Connection timeout (no-traffic) flag.
*/
double _connectionTimeout;
/**
* Unresponsive transport flag.
*/
bool _unresponsiveTransport;
/**
* Timestamp of last "live" event on this transport.
*/
epicsTimeStamp _aliveTimestamp;
const double _connectionTimeout;
bool _verifyOrEcho;
/**
* Unresponsive transport notify.
*/
void unresponsiveTransport();
// are we queued to send verify or echo?
bool sendQueued;
/**
* Notifies clients about disconnect.
*/
void closedNotifyClients();
/**
* Responsive transport notify.
*/
void responsiveTransport();
};
}

10
src/remote/pv/remote.h
View File

@@ -222,11 +222,6 @@ public:
// TODO enum
virtual void setByteOrder(int byteOrder) = 0;
/**
* Notification that transport has changed.
*/
virtual void changedTransport() = 0;
/**
* Enqueue send request.
* @param sender
@@ -250,11 +245,6 @@ public:
*/
virtual bool verify(epics::pvData::int32 timeoutMs) = 0;
/**
* Notification transport that is still alive.
*/
virtual void aliveNotification() = 0;
/**
* Close transport.
*/

12
src/remote/security.cpp
View File

@@ -95,9 +95,15 @@ struct CAPlugin : public pva::AuthenticationPlugin
{
std::tr1::shared_ptr<SimpleSession> sess(new SimpleSession(user)); // no init data
if(server) {
peer->identified = true;
peer->account = data->getSubFieldT<pvd::PVString>("user")->get();
peer->aux = pvd::getPVDataCreate()->createPVStructure(data); // clone to ensure it won't be modified
pvd::PVString::shared_pointer user;
if(data)
user = data->getSubField<pvd::PVString>("user");
if(user) {
peer->account = user->get();
peer->identified = !peer->account.empty();
peer->aux = pvd::getPVDataCreate()->createPVStructure(data); // clone to ensure it won't be modified
}
control->authenticationCompleted(pvd::Status::Ok, peer);
}
return sess;

46
src/remoteClient/clientContextImpl.cpp
View File

@@ -2998,10 +2998,10 @@ public:
{
if (command < 0 || command >= (int8)m_handlerTable.size())
{
// TODO remove debug output
char buf[100];
sprintf(buf, "Invalid (or unsupported) command %d, its payload", command);
hexDump(buf, (const int8*)(payloadBuffer->getArray()), payloadBuffer->getPosition(), payloadSize);
if(IS_LOGGABLE(logLevelError)) {
std::cerr<<"Invalid (or unsupported) command: "<<std::hex<<(int)(0xFF&command)<<"\n"
<<HexDump(*payloadBuffer, payloadSize).limit(256u);
}
return;
}
// delegate
@@ -3597,7 +3597,7 @@ public:
// NOTE: calls channelConnectFailed() on failure
static ServerGUID guid = { { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 } };
// m_addresses[ix] is modified by the following
searchResponse(guid, PVA_PROTOCOL_REVISION, &m_addresses[ix]);
searchResponse(guid, PVA_CLIENT_PROTOCOL_REVISION, &m_addresses[ix]);
}
virtual void timerStopped() OVERRIDE FINAL {
@@ -3675,13 +3675,6 @@ public:
reportChannelStateChange();
}
virtual void transportChanged() OVERRIDE FINAL {
// initiateSearch();
// TODO
// this will be called immediately after reconnect... bad...
}
virtual Transport::shared_pointer checkAndGetTransport() OVERRIDE FINAL
{
Lock guard(m_channelMutex);
@@ -3711,35 +3704,6 @@ public:
return m_transport;
}
virtual void transportResponsive(Transport::shared_pointer const & /*transport*/) OVERRIDE FINAL {
Lock guard(m_channelMutex);
if (m_connectionState == DISCONNECTED)
{
updateSubscriptions();
// reconnect using existing IDs, data
connectionCompleted(m_serverChannelID/*, accessRights*/);
}
}
virtual void transportUnresponsive() OVERRIDE FINAL {
/*
{
Lock guard(m_channelMutex);
if (m_connectionState == CONNECTED)
{
// TODO 2 types of disconnected state - distinguish them otherwise disconnect will handle connection loss right
setConnectionState(DISCONNECTED);
// ... PVA notifies also w/ no access rights callback, although access right are not changed
}
}
// should be called without any lock hold
reportChannelStateChange();
*/
}
/**
* Set connection state and if changed, notifies listeners.
* @param newState state to set.

3
src/remoteClient/pv/clientContextImpl.h
View File

@@ -53,10 +53,7 @@ public:
virtual Transport::shared_pointer checkAndGetTransport() = 0;
virtual Transport::shared_pointer checkDestroyedAndGetTransport() = 0;
virtual Transport::shared_pointer getTransport() = 0;
virtual void transportUnresponsive() =0;
virtual void transportChanged() =0;
virtual void transportClosed() =0;
virtual void transportResponsive(Transport::shared_pointer const & /*transport*/) =0;
static epics::pvData::Status channelDestroyed;
static epics::pvData::Status channelDisconnected;

11
src/server/pv/responseHandlers.h
View File

@@ -93,20 +93,23 @@ public:
class EchoTransportSender : public TransportSender {
public:
EchoTransportSender(osiSockAddr* echoFrom) {
EchoTransportSender(osiSockAddr* echoFrom, size_t payloadSize, epics::pvData::ByteBuffer& payloadBuffer) {
memcpy(&_echoFrom, echoFrom, sizeof(osiSockAddr));
toEcho.resize(payloadSize);
payloadBuffer.getArray(&toEcho[0], payloadSize);
}
virtual ~EchoTransportSender() {}
virtual void send(epics::pvData::ByteBuffer* /*buffer*/, TransportSendControl* control) OVERRIDE FINAL {
control->startMessage(CMD_ECHO, 0);
virtual void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) OVERRIDE FINAL {
control->startMessage(CMD_ECHO, toEcho.size(), toEcho.size());
control->setRecipient(_echoFrom);
// TODO content
buffer->putArray<char>(&toEcho[0], toEcho.size());
}
private:
osiSockAddr _echoFrom;
std::vector<char> toEcho;
};
/****************************************************************************************/

15
src/server/responseHandlers.cpp
View File

@@ -155,16 +155,13 @@ void ServerResponseHandler::handleResponse(osiSockAddr* responseFrom,
{
if(command<0||command>=(int8)m_handlerTable.size())
{
LOG(logLevelDebug,
LOG(logLevelError,
"Invalid (or unsupported) command: %x.", (0xFF&command));
// TODO remove debug output
std::ostringstream name;
name<<"Invalid PVA header "<<hex<<(int)(0xFF&command);
name<<", its payload buffer";
hexDump(name.str(), (const int8*)payloadBuffer->getArray(),
payloadBuffer->getPosition(), payloadSize);
if(IS_LOGGABLE(logLevelError)) {
std::cerr<<"Invalid (or unsupported) command: "<<std::hex<<(int)(0xFF&command)<<"\n"
<<HexDump(*payloadBuffer, payloadSize).limit(256u);
}
return;
}
@@ -231,7 +228,7 @@ void ServerEchoHandler::handleResponse(osiSockAddr* responseFrom,
transport, version, command, payloadSize, payloadBuffer);
// send back
TransportSender::shared_pointer echoReply(new EchoTransportSender(responseFrom));
TransportSender::shared_pointer echoReply(new EchoTransportSender(responseFrom, payloadSize, *payloadBuffer));
transport->enqueueSendRequest(echoReply);
}

144
src/utils/hexDump.cpp
View File

@@ -7,102 +7,92 @@
#include <iostream>
#include <sstream>
#include <string>
#include <iomanip>
#include <algorithm>
#include <pv/byteBuffer.h>
#define epicsExportSharedSymbols
#include <pv/hexDump.h>
using namespace epics::pvData;
using std::string;
using std::stringstream;
using std::endl;
using std::cout;
namespace epics {
namespace pvAccess {
/// Byte to hexchar mapping.
static const char lookup[] = {
'0', '1', '2', '3', '4', '5', '6', '7',
'8', '9', 'A', 'B', 'C', 'D', 'E', 'F'
};
HexDump::HexDump(const char* buf, size_t len)
:buf(buf)
,buflen(len)
,_limit(1024u)
,_groupBy(4u)
,_perLine(16u)
{}
/// Get hex representation of byte.
string toHex(int8 b) {
string sb;
int upper = (b>>4)&0x0F;
sb += lookup[upper];
int lower = b&0x0F;
sb += lookup[lower];
sb += ' ';
return sb;
HexDump::HexDump(const pvData::ByteBuffer& bb,
size_t size, size_t offset)
:buf(bb.getBuffer() + bb.getPosition())
,buflen(bb.getRemaining())
,_limit((size_t)-1)
,_groupBy(4u)
,_perLine(16u)
{
if(offset > buflen)
offset = buflen;
buf += offset;
buflen -= offset;
if(buflen > size)
buflen = size;
}
/// Get ASCII representation of byte, dot if non-readable.
char toAscii(int8 b) {
if(b>(int8)31&&b<(int8)127)
return (char)b;
else
return '.';
HexDump::~HexDump() {}
static
size_t ilog2(size_t val)
{
size_t ret = 0;
while(val >>= 1)
ret++;
return ret;
}
void hexDump(std::string const & name, const int8 *bs, int len) {
hexDump(name, bs, 0, len);
static
size_t bits2bytes(size_t val)
{
// round up to next multiple of 8
val -= 1u;
val |= 7u;
val += 1u;
// bits -> bytes
val /= 8u;
return val;
}
void hexDump(std::string const & name, const int8 *bs, int start, int len) {
hexDump("", name, bs, start, len);
}
epicsShareFunc
std::ostream& operator<<(std::ostream& strm, const HexDump& hex)
{
size_t len = std::min(hex.buflen, hex._limit);
// find address width in hex chars
// find bit width, rounded up to 8 bits, divide down to bytes
size_t addrwidth = bits2bytes(ilog2(len))*2u;
void hexDump(std::string const & prologue, string const & name, const int8 *bs,
int start, int len) {
for(size_t i=0; i<len; i++)
{
unsigned val = hex.buf[i]&0xff;
size_t col = i%hex._perLine;
stringstream header;
header<<prologue<<endl<<"Hexdump ["<<name<<"] size = "<<len;
string out(header.str());
string chars;
for(int i = start; i<(start+len); i++) {
if(((i-start)%16)==0) {
out += chars;
out += '\n';
chars.erase();
if(col==0) {
// first address of this row
strm<<"0x"<<std::hex<<std::setw(addrwidth)<<std::setfill('0')<<i;
}
chars += toAscii(bs[i]);
out += toHex(bs[i]);
if(((i-start)%4)==3) {
chars += ' ';
out += ' ';
if(col%hex._groupBy == 0) {
strm<<' ';
}
strm<<std::hex<<std::setw(2)<<std::setfill('0')<<val;
if(col+1u==hex._perLine && i+1u!=len)
strm<<'\n';
}
if(len%16!=0) {
int pad = 0;
int delta_bytes = 16-(len%16);
//rest of line (no of bytes)
//each byte takes two chars plus one ws
pad = delta_bytes*3;
//additional whitespaces after four bytes
pad += (delta_bytes/4);
pad++;
for(int i = 0; i<pad; i++)
chars.insert(0, " ");
}
out += chars;
cout<<out<<endl;
if(len==hex._limit && len<hex.buflen)
strm<<" ...";
strm<<'\n';
return strm;
}
}

51
src/utils/pv/hexDump.h
View File

@@ -7,6 +7,8 @@
#ifndef HEXDUMP_H_
#define HEXDUMP_H_
#include <ostream>
#ifdef epicsExportSharedSymbols
# define hexDumpEpicsExportSharedSymbols
# undef epicsExportSharedSymbols
@@ -22,35 +24,34 @@
#include <shareLib.h>
namespace epics {
namespace pvData {
class ByteBuffer;
}
namespace pvAccess {
/**
* Output a buffer in hex format.
* @param name name (description) of the message.
* @param bs buffer to dump
* @param len first bytes (length) to dump.
*/
epicsShareFunc void hexDump(std::string const & name, const epics::pvData::int8 *bs, int len);
class epicsShareClass HexDump {
const char* buf;
size_t buflen;
size_t _limit;
unsigned _groupBy;
unsigned _perLine;
public:
HexDump(const char* buf, size_t len);
explicit HexDump(const pvData::ByteBuffer& buf, size_t size=(size_t)-1, size_t offset=0u);
~HexDump();
/**
* Output a buffer in hex format.
* @param[in] name name (description) of the message.
* @param[in] bs buffer to dump
* @param[in] start dump message using given offset.
* @param[in] len first bytes (length) to dump.
*/
epicsShareFunc void hexDump(std::string const & name, const epics::pvData::int8 *bs, int start, int len);
//! safety limit on max bytes printed
inline HexDump& limit(size_t n=(size_t)-1) { _limit = n; return *this; }
//! insert a space after this many bytes
inline HexDump& bytesPerGroup(size_t n=(size_t)-1) { _groupBy = n; return *this; }
//! start a new line after this many bytes
inline HexDump& bytesPerLine(size_t n=(size_t)-1) { _perLine = n; return *this; }
/**
* Output a buffer in hex format.
* @param[in] prologue string to prefixed to debug output, can be <code>null</code>
* @param[in] name name (description) of the message.
* @param[in] bs buffer to dump
* @param[in] start dump message using given offset.
* @param[in] len first bytes (length) to dump.
*/
epicsShareFunc void hexDump(std::string const & prologue, std::string const & name,
const epics::pvData::int8 *bs, int start, int len);
friend std::ostream& operator<<(std::ostream& strm, const HexDump& hex);
};
epicsShareFunc
std::ostream& operator<<(std::ostream& strm, const HexDump& hex);
}
}

128
testApp/remote/testCodec.cpp
View File

@@ -372,8 +372,6 @@ public:
void setRemoteTransportReceiveBufferSize(
std::size_t remoteTransportReceiveBufferSize) {}
void changedTransport() {}
void flushSendQueue() { };
bool verify(epics::pvData::int32 timeoutMs) {
@@ -382,8 +380,6 @@ public:
void verified(epics::pvData::Status const &) {}
void aliveNotification() {}
void authNZMessage(epics::pvData::PVStructure::shared_pointer const & data) {}
@@ -483,7 +479,7 @@ private:
TestCodec codec(DEFAULT_BUFFER_SIZE,DEFAULT_BUFFER_SIZE);
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x01);
codec._readBuffer->put((int8_t)0x23);
codec._readBuffer->putInt(0x456789AB);
@@ -506,8 +502,8 @@ private:
PVAMessage header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(codec._invalidDataStreamCount == 0,
"%s: codec._invalidDataStreamCount == 0",
@@ -523,13 +519,13 @@ private:
// two at the time, app and control
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x00);
codec._readBuffer->put((int8_t)0x20);
codec._readBuffer->putInt(0x00000000);
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x81);
codec._readBuffer->put((int8_t)0xEE);
codec._readBuffer->putInt(0xDDCCBBAA);
@@ -553,8 +549,8 @@ private:
// app, no payload
header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x00,
"%s: header._flags == 0x00", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x20,
@@ -565,8 +561,8 @@ private:
// control
header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x81,
"%s: header._flags == 0x81", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0xEE,
@@ -585,7 +581,7 @@ private:
TestCodec codec(DEFAULT_BUFFER_SIZE,DEFAULT_BUFFER_SIZE);
codec._readBuffer->put((int8_t)00);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x01);
codec._readBuffer->put((int8_t)0x23);
codec._readBuffer->putInt(0x456789AB);
@@ -621,7 +617,7 @@ private:
{
TestCodec codec(DEFAULT_BUFFER_SIZE,DEFAULT_BUFFER_SIZE);
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put(invalidFlagsValues[i]);
codec._readBuffer->put((int8_t)0x23);
//codec._readBuffer->putInt(0);
@@ -653,7 +649,7 @@ private:
TestCodec codec(DEFAULT_BUFFER_SIZE,DEFAULT_BUFFER_SIZE);
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x80);
codec._readBuffer->put((int8_t)0x23);
codec._readBuffer->putInt(0x456789AB);
@@ -682,7 +678,7 @@ private:
TestCodec codec(DEFAULT_BUFFER_SIZE,DEFAULT_BUFFER_SIZE);
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x01);
codec._readBuffer->flip();
@@ -724,8 +720,8 @@ private:
// app, no payload
PVAMessage header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x01,
"%s: header._flags == 0x01", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x23,
@@ -744,7 +740,7 @@ private:
codec._readPayload = true;
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x80);
codec._readBuffer->put((int8_t)0x23);
codec._readBuffer->putInt(1); // size
@@ -790,7 +786,7 @@ private:
codec._readPayload = true;
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x80);
codec._readBuffer->put((int8_t)0x23);
const int32_t payloadSize1 = 2;
@@ -806,7 +802,7 @@ private:
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x80);
codec._readBuffer->put((int8_t)0x45);
@@ -912,7 +908,7 @@ private:
_codec._readBuffer->put(PVA_MAGIC);
_codec._readBuffer->put(PVA_VERSION);
_codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
_codec._readBuffer->put((int8_t)0x80);
_codec._readBuffer->put((int8_t)0x45);
_codec._readBuffer->putInt(_payloadSize2);
@@ -958,7 +954,7 @@ private:
// 1st
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -971,7 +967,7 @@ private:
// 2nd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -983,14 +979,14 @@ private:
// control in between
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x81);
codec._readBuffer->put((int8_t)0xEE);
codec._readBuffer->putInt(0xDDCCBBAA);
// 3rd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1002,7 +998,7 @@ private:
// 4t (last)
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xA0);
codec._readBuffer->put((int8_t)0x01);
@@ -1057,8 +1053,8 @@ private:
msg = codec._receivedControlMessages[0];
testOk(msg._version == PVA_VERSION,
"%s: msg._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(msg._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: msg._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(msg._flags == (int8_t)0x81,
"%s: msg._flags == 0x81", CURRENT_FUNCTION);
testOk(msg._command == (int8_t)0xEE,
@@ -1078,7 +1074,7 @@ private:
// 1st
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -1091,7 +1087,7 @@ private:
// 2nd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
// invalid flag, should be 0xB0
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -1104,14 +1100,14 @@ private:
// control in between
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x81);
codec._readBuffer->put((int8_t)0xEE);
codec._readBuffer->putInt(0xDDCCBBAA);
// 3rd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1123,7 +1119,7 @@ private:
// 4t (last)
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xA0);
codec._readBuffer->put((int8_t)0x01);
@@ -1172,7 +1168,7 @@ private:
// 1st
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -1190,7 +1186,7 @@ private:
// 2nd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1209,7 +1205,7 @@ private:
// 3rd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1227,7 +1223,7 @@ private:
// 4t (last)
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xA0);
codec._readBuffer->put((int8_t)0x01);
@@ -1339,7 +1335,7 @@ private:
// 1st
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -1357,7 +1353,7 @@ private:
// 2nd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1377,7 +1373,7 @@ private:
// 3rd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xA0);
codec._readBuffer->put((int8_t)0x01);
@@ -1499,8 +1495,8 @@ private:
PVAMessage header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x01),
"%s: header._flags == 0x(0|8)1", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x23,
@@ -1537,8 +1533,8 @@ private:
// app, no payload
header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x00,
"%s: header._flags == 0x00", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x20,
@@ -1549,8 +1545,8 @@ private:
// control
header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x81,
"%s: header._flags == 0x81", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0xEE,
@@ -1878,7 +1874,7 @@ private:
codec._disconnected = true;
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x01);
codec._readBuffer->put((int8_t)0x23);
codec._readBuffer->putInt(0x456789AB);
@@ -1953,7 +1949,7 @@ private:
// 1st
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0x90);
codec._readBuffer->put((int8_t)0x01);
@@ -1971,7 +1967,7 @@ private:
// 2nd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xB0);
codec._readBuffer->put((int8_t)0x01);
@@ -1991,7 +1987,7 @@ private:
// 3rd
codec._readBuffer->put(PVA_MAGIC);
codec._readBuffer->put(PVA_VERSION);
codec._readBuffer->put(PVA_CLIENT_PROTOCOL_REVISION);
codec._readBuffer->put((int8_t)0xA0);
codec._readBuffer->put((int8_t)0x01);
@@ -2156,8 +2152,8 @@ private:
// app, no payload
PVAMessage header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x00),
"%s: header._flags == 0x(0|8)0", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x20,
@@ -2168,8 +2164,8 @@ private:
// control
header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x01),
"%s: header._flags == 0x(0|8)1", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0xEE,
@@ -2293,8 +2289,8 @@ private:
// app, no payload
PVAMessage header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x00),
"%s: header._flags == 0x(0|8)0", CURRENT_FUNCTION);
testOk(header._command == 0x20,
@@ -2306,8 +2302,8 @@ private:
// control
header = codec._receivedControlMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x01),
"%s: header._flags == 0x(0|8)1", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0xEE,
@@ -2363,8 +2359,8 @@ private:
header = codec._receivedControlMessages[1];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x01),
"%s: header._flags == 0x(0|8)1", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0xEE,
@@ -2443,8 +2439,8 @@ private:
// app
PVAMessage header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)0x80,
"%s: header._flags == 0x80", CURRENT_FUNCTION);
testOk(header._command == (int8_t)0x12,
@@ -2640,8 +2636,8 @@ private:
// app
PVAMessage header = codec._receivedAppMessages[0];
testOk(header._version == PVA_VERSION,
"%s: header._version == PVA_VERSION", CURRENT_FUNCTION);
testOk(header._version == PVA_CLIENT_PROTOCOL_REVISION,
"%s: header._version == PVA_CLIENT_PROTOCOL_REVISION", CURRENT_FUNCTION);
testOk(header._flags == (int8_t)((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00) | 0x10),
"%s: header._flags == (int8_t)(0x(0|8)0 | 0x10)",
CURRENT_FUNCTION);

20
testApp/utils/testHexDump.cpp
View File

@@ -1,6 +1,9 @@
#include <epicsUnitTest.h>
#include <sstream>
#include <testMain.h>
#include <pv/pvUnitTest.h>
#include <pv/hexDump.h>
using namespace epics::pvData;
@@ -8,19 +11,16 @@ using namespace epics::pvAccess;
MAIN(testHexDump)
{
testPlan(3);
testPlan(1);
testDiag("Tests for hexDump");
char TO_DUMP[] = "pvAccess dump test\0\1\2\3\4\5\6\254\255\256";
char TO_DUMP[] = "pvAccess dump test\0\1\2\3\4\5\6\xfd\xfe\xff";
hexDump("test", (int8*)TO_DUMP, 18+9);
testPass("Entire array");
std::ostringstream msg;
msg<<HexDump(TO_DUMP, sizeof(TO_DUMP)-1);
hexDump("only text", (int8*)TO_DUMP, 18);
testPass("Only text");
hexDump("22 byte test", (int8*)TO_DUMP, 22);
testPass("22 bytes test");
testEqual(msg.str(), "0x00 70764163 63657373 2064756d 70207465\n"
"0x10 73740001 02030405 06fdfeff\n");
return testDone();
}