codec implementation with lots of tests commited
This commit is contained in:
Matej Sekoranja
@@ -45,6 +45,7 @@ INC += channelSearchManager.h
|
||||
INC += simpleChannelSearchManagerImpl.h
|
||||
INC += transportRegistry.h
|
||||
INC += serializationHelper.h
|
||||
INC += codec.h
|
||||
LIBSRCS += blockingUDPTransport.cpp
|
||||
LIBSRCS += blockingUDPConnector.cpp
|
||||
LIBSRCS += beaconHandler.cpp
|
||||
@@ -57,6 +58,7 @@ LIBSRCS += abstractResponseHandler.cpp
|
||||
LIBSRCS += blockingTCPAcceptor.cpp
|
||||
LIBSRCS += transportRegistry.cpp
|
||||
LIBSRCS += serializationHelper.cpp
|
||||
LIBSRCS += codec.cpp
|
||||
|
||||
SRC_DIRS += $(PVACCESS)/remoteClient
|
||||
INC += clientContextImpl.h
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
*/
|
||||
|
||||
#include <pv/blockingTCP.h>
|
||||
#include "codec.h"
|
||||
#include <pv/remote.h>
|
||||
#include <pv/logger.h>
|
||||
|
||||
@@ -180,17 +181,28 @@ namespace pvAccess {
|
||||
}
|
||||
|
||||
// TODO tune buffer sizes?!
|
||||
|
||||
// get TCP send buffer size
|
||||
osiSocklen_t intLen = sizeof(int);
|
||||
int _socketSendBufferSize;
|
||||
retval = getsockopt(newClient, SOL_SOCKET, SO_SNDBUF, (char *)&_socketSendBufferSize, &intLen);
|
||||
if(retval<0) {
|
||||
epicsSocketConvertErrnoToString(strBuffer, sizeof(strBuffer));
|
||||
LOG(logLevelDebug, "Error getting SO_SNDBUF: %s", strBuffer);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Create transport, it registers itself to the registry.
|
||||
* Each transport should have its own response handler since it is not "shareable"
|
||||
*/
|
||||
std::auto_ptr<ResponseHandler> responseHandler = _responseHandlerFactory->createResponseHandler();
|
||||
BlockingServerTCPTransport::shared_pointer transport =
|
||||
BlockingServerTCPTransport::create(
|
||||
BlockingServerTCPTransportCodec::shared_pointer transport =
|
||||
BlockingServerTCPTransportCodec::create(
|
||||
_context,
|
||||
newClient,
|
||||
responseHandler,
|
||||
_socketSendBufferSize,
|
||||
_receiveBufferSize);
|
||||
|
||||
// validate connection
|
||||
|
||||
1568
pvAccessApp/remote/codec.cpp
Normal file
1568
pvAccessApp/remote/codec.cpp
Normal file
File diff suppressed because it is too large
Load Diff
807
pvAccessApp/remote/codec.h
Normal file
807
pvAccessApp/remote/codec.h
Normal file
@@ -0,0 +1,807 @@
|
||||
/**
|
||||
* Copyright - See the COPYRIGHT that is included with this distribution.
|
||||
* pvAccessCPP is distributed subject to a Software License Agreement found
|
||||
* in file LICENSE that is included with this distribution.
|
||||
*/
|
||||
|
||||
#ifndef CODEC_H_
|
||||
#define CODEC_H_
|
||||
|
||||
#include <set>
|
||||
#include <map>
|
||||
#include <deque>
|
||||
|
||||
#ifdef epicsExportSharedSymbols
|
||||
# define abstractCodecEpicsExportSharedSymbols
|
||||
# undef epicsExportSharedSymbols
|
||||
#endif
|
||||
|
||||
#include <shareLib.h>
|
||||
#include <osdSock.h>
|
||||
#include <osiSock.h>
|
||||
#include <epicsTime.h>
|
||||
#include <epicsThread.h>
|
||||
|
||||
#include <pv/byteBuffer.h>
|
||||
#include <pv/pvType.h>
|
||||
#include <pv/lock.h>
|
||||
#include <pv/timer.h>
|
||||
#include <pv/event.h>
|
||||
#include <pv/likely.h>
|
||||
#include <pv/logger.h>
|
||||
|
||||
#ifdef abstractCodecEpicsExportSharedSymbols
|
||||
# define epicsExportSharedSymbols
|
||||
# undef abstractCodecEpicsExportSharedSymbols
|
||||
#endif
|
||||
|
||||
#include <pv/pvaConstants.h>
|
||||
#include <pv/remote.h>
|
||||
#include <pv/transportRegistry.h>
|
||||
#include <pv/introspectionRegistry.h>
|
||||
#include <pv/namedLockPattern.h>
|
||||
#include <pv/inetAddressUtil.h>
|
||||
|
||||
namespace epics {
|
||||
namespace pvAccess {
|
||||
|
||||
|
||||
template<typename T>
|
||||
class AtomicValue
|
||||
{
|
||||
public:
|
||||
AtomicValue(): _value(0) {};
|
||||
|
||||
T getAndSet(T value)
|
||||
{
|
||||
mutex.lock();
|
||||
T tmp = _value; _value = value;
|
||||
mutex.unlock();
|
||||
return tmp;
|
||||
}
|
||||
|
||||
T get() { mutex.lock(); T tmp = _value; mutex.unlock(); return tmp; }
|
||||
|
||||
private:
|
||||
T _value;
|
||||
epics::pvData::Mutex mutex;
|
||||
};
|
||||
|
||||
|
||||
template<typename T>
|
||||
class queue {
|
||||
public:
|
||||
|
||||
queue(void) { }
|
||||
//TODO
|
||||
/*queue(queue const &T) = delete;
|
||||
queue(queue &&T) = delete;
|
||||
queue& operator=(const queue &T) = delete;
|
||||
*/
|
||||
~queue(void)
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"queue::~queue DESTROY (threadId: %u)", epicsThreadGetIdSelf());
|
||||
}
|
||||
|
||||
|
||||
bool empty(void)
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"queue::empty enter: (threadId: %u)", epicsThreadGetIdSelf());
|
||||
epics::pvData::Lock lock(_queueMutex);
|
||||
return _queue.empty();
|
||||
}
|
||||
|
||||
void clean()
|
||||
{
|
||||
LOG(logLevelTrace, "queue::clean enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
epics::pvData::Lock lock(_queueMutex);
|
||||
_queue.clear();
|
||||
}
|
||||
|
||||
|
||||
void wakeup()
|
||||
{
|
||||
|
||||
LOG(logLevelTrace, "queue::wakeup enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
if (!_wakeup.getAndSet(true))
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"queue::wakeup signaling on _queueEvent: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
_queueEvent.signal();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void put(T const & elem)
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"queue::put enter (threadId: %u)", epicsThreadGetIdSelf());
|
||||
|
||||
{
|
||||
epics::pvData::Lock lock(_queueMutex);
|
||||
_queue.push_front(elem);
|
||||
}
|
||||
|
||||
_queueEvent.signal();
|
||||
}
|
||||
|
||||
|
||||
T take(int timeOut)
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take enter timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
while (true)
|
||||
{
|
||||
|
||||
bool isEmpty = empty();
|
||||
|
||||
if (isEmpty)
|
||||
{
|
||||
|
||||
if (timeOut < 0) {
|
||||
epics::pvAccess::LOG(logLevelTrace,
|
||||
"queue::take exit timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
return T();
|
||||
}
|
||||
|
||||
while (isEmpty)
|
||||
{
|
||||
|
||||
if (timeOut == 0) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take going to wait timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
_queueEvent.wait();
|
||||
}
|
||||
else {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take going to wait timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
_queueEvent.wait(timeOut);
|
||||
}
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take waking up timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
isEmpty = empty();
|
||||
if (isEmpty)
|
||||
{
|
||||
if (timeOut > 0) { // TODO spurious wakeup, but not critical
|
||||
LOG(logLevelTrace,
|
||||
"queue::take exit after being woken up timeOut:%d"
|
||||
" (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
return T();
|
||||
}
|
||||
else // if (timeout == 0) cannot be negative
|
||||
{
|
||||
if (_wakeup.getAndSet(false)) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take exit after being woken up timeOut:%d"
|
||||
" (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
return T();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take obtaining lock for front element timeOut:%d"
|
||||
" (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
epics::pvData::Lock lock(_queueMutex);
|
||||
T sender = _queue.front();
|
||||
_queue.pop_front();
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"queue::take exit with sender timeOut:%d (threadId: %u)",
|
||||
timeOut, epicsThreadGetIdSelf());
|
||||
|
||||
return sender;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
|
||||
std::deque<T> _queue;
|
||||
epics::pvData::Event _queueEvent;
|
||||
epics::pvData::Mutex _queueMutex;
|
||||
AtomicValue<bool> _wakeup;
|
||||
epics::pvData::Mutex _stdMutex;
|
||||
};
|
||||
|
||||
|
||||
class simulate_finally_exception: public std::runtime_error {
|
||||
public:
|
||||
explicit simulate_finally_exception(
|
||||
const std::string &s): std::runtime_error(s) {}
|
||||
};
|
||||
|
||||
class io_exception: public std::runtime_error {
|
||||
public:
|
||||
explicit io_exception(const std::string &s): std::runtime_error(s) {}
|
||||
};
|
||||
|
||||
|
||||
class invalid_data_stream_exception: public std::runtime_error {
|
||||
public:
|
||||
explicit invalid_data_stream_exception(
|
||||
const std::string &s): std::runtime_error(s) {}
|
||||
};
|
||||
|
||||
|
||||
class connection_closed_exception: public std::runtime_error {
|
||||
public:
|
||||
explicit connection_closed_exception(const std::string &s): std::runtime_error(s) {}
|
||||
};
|
||||
|
||||
|
||||
enum ReadMode { NORMAL, SPLIT, SEGMENTED };
|
||||
|
||||
enum WriteMode { PROCESS_SEND_QUEUE, WAIT_FOR_READY_SIGNAL };
|
||||
|
||||
|
||||
class AbstractCodec :
|
||||
public TransportSendControl,
|
||||
public Transport
|
||||
{
|
||||
public:
|
||||
|
||||
static const std::size_t MAX_MESSAGE_PROCESS;
|
||||
static const std::size_t MAX_MESSAGE_SEND;
|
||||
static const std::size_t MAX_ENSURE_SIZE;
|
||||
static const std::size_t MAX_ENSURE_DATA_SIZE;
|
||||
static const std::size_t MAX_ENSURE_BUFFER_SIZE;
|
||||
static const std::size_t MAX_ENSURE_DATA_BUFFER_SIZE;
|
||||
|
||||
AbstractCodec(
|
||||
epics::pvData::ByteBuffer *receiveBuffer,
|
||||
epics::pvData::ByteBuffer *sendBuffer,
|
||||
int32_t socketSendBufferSize,
|
||||
bool blockingProcessQueue);
|
||||
|
||||
virtual void processControlMessage() = 0;
|
||||
virtual void processApplicationMessage() = 0;
|
||||
virtual osiSockAddr getLastReadBufferSocketAddress() = 0;
|
||||
virtual void invalidDataStreamHandler() = 0;
|
||||
virtual void readPollOne()=0;
|
||||
virtual void writePollOne() = 0;
|
||||
virtual void scheduleSend() = 0;
|
||||
virtual void sendCompleted() = 0;
|
||||
virtual bool terminated() = 0;
|
||||
virtual int write(epics::pvData::ByteBuffer* src) = 0;
|
||||
virtual int read(epics::pvData::ByteBuffer* dst) = 0;
|
||||
virtual bool isOpen() = 0;
|
||||
virtual void close() = 0;
|
||||
|
||||
|
||||
virtual ~AbstractCodec()
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"AbstractCodec::~AbstractCodec DESTROY (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
}
|
||||
|
||||
void alignBuffer(std::size_t alignment);
|
||||
void ensureData(std::size_t size);
|
||||
void alignData(std::size_t alignment);
|
||||
void startMessage(
|
||||
epics::pvData::int8 command,
|
||||
std::size_t ensureCapacity);
|
||||
void putControlMessage(
|
||||
epics::pvData::int8 command,
|
||||
epics::pvData::int32 data);
|
||||
void endMessage();
|
||||
void ensureBuffer(std::size_t size);
|
||||
void flushSerializeBuffer();
|
||||
void flush(bool lastMessageCompleted);
|
||||
void processWrite();
|
||||
void processRead();
|
||||
void processSendQueue();
|
||||
void clearSendQueue();
|
||||
void enqueueSendRequest(TransportSender::shared_pointer const & sender);
|
||||
void enqueueSendRequest(TransportSender::shared_pointer const & sender,
|
||||
std::size_t requiredBufferSize);
|
||||
void setSenderThread();
|
||||
void setRecipient(osiSockAddr const & sendTo);
|
||||
void setByteOrder(int byteOrder);
|
||||
|
||||
static std::size_t alignedValue(std::size_t value, std::size_t alignment);
|
||||
|
||||
protected:
|
||||
|
||||
virtual void sendBufferFull(int tries) = 0;
|
||||
void send(epics::pvData::ByteBuffer *buffer);
|
||||
|
||||
|
||||
ReadMode _readMode;
|
||||
int8_t _version;
|
||||
int8_t _flags;
|
||||
int8_t _command;
|
||||
int32_t _payloadSize;
|
||||
epics::pvData::int32 _remoteTransportSocketReceiveBufferSize;
|
||||
int64_t _totalBytesSent;
|
||||
bool _blockingProcessQueue;
|
||||
//TODO initialize union
|
||||
osiSockAddr _sendTo;
|
||||
epicsThreadId _senderThread;
|
||||
WriteMode _writeMode;
|
||||
bool _writeOpReady;
|
||||
bool _lowLatency;
|
||||
|
||||
std::auto_ptr<epics::pvData::ByteBuffer> _socketBuffer;
|
||||
std::auto_ptr<epics::pvData::ByteBuffer> _sendBuffer;
|
||||
|
||||
epics::pvAccess::queue<TransportSender::shared_pointer > _sendQueue;
|
||||
|
||||
private:
|
||||
|
||||
void processHeader();
|
||||
void processReadNormal();
|
||||
void processReadSegmented();
|
||||
bool readToBuffer(std::size_t requiredBytes, bool persistent);
|
||||
void endMessage(bool hasMoreSegments);
|
||||
void processSender(
|
||||
epics::pvAccess::TransportSender::shared_pointer const & sender);
|
||||
|
||||
std::size_t _storedPayloadSize;
|
||||
std::size_t _storedPosition;
|
||||
std::size_t _storedLimit;
|
||||
std::size_t _startPosition;
|
||||
|
||||
std::size_t _maxSendPayloadSize;
|
||||
std::size_t _lastMessageStartPosition;
|
||||
std::size_t _lastSegmentedMessageType;
|
||||
int8_t _lastSegmentedMessageCommand;
|
||||
std::size_t _nextMessagePayloadOffset;
|
||||
|
||||
epics::pvData::int8 _byteOrderFlag;
|
||||
int32_t _socketSendBufferSize;
|
||||
};
|
||||
|
||||
|
||||
class BlockingAbstractCodec: public AbstractCodec {
|
||||
|
||||
public:
|
||||
|
||||
POINTER_DEFINITIONS(BlockingAbstractCodec);
|
||||
|
||||
BlockingAbstractCodec(
|
||||
epics::pvData::ByteBuffer *receiveBuffer,
|
||||
epics::pvData::ByteBuffer *sendBuffer,
|
||||
int32_t socketSendBufferSize):
|
||||
AbstractCodec(receiveBuffer, sendBuffer, socketSendBufferSize, true),
|
||||
_readThread(0), _sendThread(0) { _isOpen.getAndSet(true);}
|
||||
|
||||
void readPollOne();
|
||||
void writePollOne();
|
||||
void scheduleSend() {}
|
||||
void sendCompleted() {}
|
||||
void close();
|
||||
bool terminated();
|
||||
bool isOpen();
|
||||
void start();
|
||||
|
||||
static void receiveThread(void* param);
|
||||
static void sendThread(void* param);
|
||||
|
||||
protected:
|
||||
void sendBufferFull(int tries);
|
||||
virtual void internalDestroy() = 0;
|
||||
|
||||
private:
|
||||
AtomicValue<bool> _isOpen;
|
||||
volatile epicsThreadId _readThread;
|
||||
volatile epicsThreadId _sendThread;
|
||||
epics::pvData::Event _shutdownEvent;
|
||||
};
|
||||
|
||||
|
||||
class BlockingSocketAbstractCodec: public BlockingAbstractCodec {
|
||||
|
||||
public:
|
||||
|
||||
BlockingSocketAbstractCodec(
|
||||
SOCKET channel,
|
||||
int32_t sendBufferSize,
|
||||
int32_t receiveBufferSize);
|
||||
|
||||
int read(epics::pvData::ByteBuffer* dst);
|
||||
int write(epics::pvData::ByteBuffer* src);
|
||||
osiSockAddr getLastReadBufferSocketAddress() { return _socketAddress;}
|
||||
void invalidDataStreamHandler();
|
||||
std::size_t getSocketReceiveBufferSize() const;
|
||||
|
||||
protected:
|
||||
|
||||
void internalDestroy();
|
||||
|
||||
private:
|
||||
SOCKET _channel;
|
||||
osiSockAddr _socketAddress;
|
||||
};
|
||||
|
||||
|
||||
class BlockingTCPTransportCodec :
|
||||
public BlockingSocketAbstractCodec,
|
||||
public std::tr1::enable_shared_from_this<BlockingTCPTransportCodec> {
|
||||
|
||||
public:
|
||||
|
||||
epics::pvData::String getType() const {
|
||||
return epics::pvData::String("TCP");
|
||||
}
|
||||
|
||||
|
||||
void internalDestroy() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::internalDestroy() enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
BlockingSocketAbstractCodec::internalDestroy();
|
||||
Transport::shared_pointer thisSharedPtr = this->shared_from_this();
|
||||
_context->getTransportRegistry()->remove(thisSharedPtr);
|
||||
}
|
||||
|
||||
|
||||
void changedTransport() {}
|
||||
|
||||
|
||||
void processControlMessage() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::processControlMessage()"
|
||||
"enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
if (_command == 2)
|
||||
{
|
||||
// check 7-th bit
|
||||
setByteOrder(_flags < 0 ? EPICS_ENDIAN_BIG : EPICS_ENDIAN_LITTLE);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void processApplicationMessage() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::processApplicationMessage() enter:"
|
||||
" (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
_responseHandler->handleResponse(&_socketAddress, shared_from_this(),
|
||||
_version, _command, _payloadSize, _socketBuffer.get());
|
||||
}
|
||||
|
||||
|
||||
const osiSockAddr* getRemoteAddress() const {
|
||||
return &_socketAddress;
|
||||
}
|
||||
|
||||
|
||||
epics::pvData::int8 getRevision() const {
|
||||
return PVA_PROTOCOL_REVISION;
|
||||
}
|
||||
|
||||
|
||||
std::size_t getReceiveBufferSize() const {
|
||||
return _socketBuffer->getSize();
|
||||
}
|
||||
|
||||
|
||||
epics::pvData::int16 getPriority() const {
|
||||
return _priority;
|
||||
}
|
||||
|
||||
|
||||
void setRemoteRevision(epics::pvData::int8 revision) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::setRemoteRevision() enter:"
|
||||
" revision: %d (threadId: %u)",
|
||||
revision, epicsThreadGetIdSelf());
|
||||
|
||||
_remoteTransportRevision = revision;
|
||||
}
|
||||
|
||||
|
||||
void setRemoteTransportReceiveBufferSize(
|
||||
std::size_t remoteTransportReceiveBufferSize) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::setRemoteTransportReceiveBufferSize()"
|
||||
" enter: remoteTransportReceiveBufferSize:%u (threadId: %u)",
|
||||
remoteTransportReceiveBufferSize, epicsThreadGetIdSelf());
|
||||
|
||||
_remoteTransportReceiveBufferSize = remoteTransportReceiveBufferSize;
|
||||
}
|
||||
|
||||
|
||||
void setRemoteTransportSocketReceiveBufferSize(
|
||||
std::size_t socketReceiveBufferSize) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::"
|
||||
"setRemoteTransportSocketReceiveBufferSize()"
|
||||
"enter: socketReceiveBufferSize:%u (threadId: %u)",
|
||||
socketReceiveBufferSize, epicsThreadGetIdSelf());
|
||||
|
||||
_remoteTransportSocketReceiveBufferSize = socketReceiveBufferSize;
|
||||
}
|
||||
|
||||
|
||||
std::tr1::shared_ptr<const epics::pvData::Field>
|
||||
cachedDeserialize(epics::pvData::ByteBuffer* buffer)
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::cachedDeserialize() enter:"
|
||||
" (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
return _incomingIR.deserialize(buffer, this);
|
||||
}
|
||||
|
||||
|
||||
void cachedSerialize(
|
||||
const std::tr1::shared_ptr<const epics::pvData::Field>& field,
|
||||
epics::pvData::ByteBuffer* buffer)
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::cachedSerialize() enter:"
|
||||
" (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
_outgoingIR.serialize(field, buffer, this);
|
||||
}
|
||||
|
||||
|
||||
bool directSerialize(
|
||||
epics::pvData::ByteBuffer *existingBuffer,
|
||||
const char* toSerialize,
|
||||
std::size_t elementCount, std::size_t elementSize)
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::directSerialize() enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
bool directDeserialize(epics::pvData::ByteBuffer *existingBuffer,
|
||||
char* deserializeTo,
|
||||
std::size_t elementCount, std::size_t elementSize) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::directDeserialize() enter:"
|
||||
" (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
void flushSendQueue() { };
|
||||
|
||||
|
||||
bool isClosed() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::isClosed() enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
return !isOpen();
|
||||
}
|
||||
|
||||
|
||||
void activate() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec::activate() enter: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
Transport::shared_pointer thisSharedPtr = shared_from_this();
|
||||
_context->getTransportRegistry()->put(thisSharedPtr);
|
||||
}
|
||||
|
||||
protected:
|
||||
|
||||
BlockingTCPTransportCodec(
|
||||
Context::shared_pointer const & context,
|
||||
SOCKET channel,
|
||||
std::auto_ptr<ResponseHandler>& responseHandler,
|
||||
int32_t sendBufferSize,
|
||||
int32_t receiveBufferSize,
|
||||
epics::pvData::int16 priority
|
||||
):
|
||||
BlockingSocketAbstractCodec(channel, sendBufferSize, receiveBufferSize),
|
||||
_context(context), _responseHandler(responseHandler),
|
||||
_verified(false), _remoteTransportReceiveBufferSize(MAX_TCP_RECV),
|
||||
_remoteTransportRevision(0), _priority(priority)
|
||||
{
|
||||
LOG(logLevelTrace,
|
||||
"BlockingTCPTransportCodec constructed: (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
}
|
||||
|
||||
|
||||
private:
|
||||
|
||||
Context::shared_pointer _context;
|
||||
std::auto_ptr<ResponseHandler> _responseHandler;
|
||||
bool _verified;
|
||||
size_t _remoteTransportReceiveBufferSize;
|
||||
epics::pvData::int8 _remoteTransportRevision;
|
||||
epics::pvData::int16 _priority;
|
||||
|
||||
osiSockAddr _socketAddress;
|
||||
epics::pvData::Mutex _verifiedMutex;
|
||||
epics::pvData::Event _verifiedEvent;
|
||||
IntrospectionRegistry _incomingIR;
|
||||
IntrospectionRegistry _outgoingIR;
|
||||
|
||||
};
|
||||
|
||||
|
||||
class BlockingServerTCPTransportCodec :
|
||||
public BlockingTCPTransportCodec,
|
||||
public ChannelHostingTransport,
|
||||
public TransportSender {
|
||||
|
||||
public:
|
||||
POINTER_DEFINITIONS(BlockingServerTCPTransportCodec);
|
||||
|
||||
BlockingServerTCPTransportCodec(
|
||||
Context::shared_pointer const & context,
|
||||
SOCKET channel,
|
||||
std::auto_ptr<ResponseHandler>& responseHandler,
|
||||
int32_t sendBufferSize,
|
||||
int32_t receiveBufferSize );
|
||||
|
||||
static shared_pointer create(
|
||||
Context::shared_pointer const & context,
|
||||
SOCKET channel,
|
||||
std::auto_ptr<ResponseHandler>& responseHandler,
|
||||
int sendBufferSize,
|
||||
int receiveBufferSize)
|
||||
{
|
||||
shared_pointer thisPointer(
|
||||
new BlockingServerTCPTransportCodec(
|
||||
context, channel, responseHandler,
|
||||
sendBufferSize, receiveBufferSize)
|
||||
);
|
||||
thisPointer->activate();
|
||||
return thisPointer;
|
||||
}
|
||||
|
||||
public:
|
||||
|
||||
bool acquire(std::tr1::shared_ptr<TransportClient> const & client)
|
||||
{
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingServerTCPTransportCodec::acquire() enter:"
|
||||
" client is set: %d (threadId: %u)",
|
||||
(client.get() != 0), epicsThreadGetIdSelf());
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
void release(pvAccessID /*clientId*/) {}
|
||||
|
||||
pvAccessID preallocateChannelSID();
|
||||
|
||||
void depreallocateChannelSID(pvAccessID /*sid*/) {
|
||||
// noop
|
||||
}
|
||||
|
||||
void registerChannel(
|
||||
pvAccessID sid,
|
||||
ServerChannel::shared_pointer const & channel);
|
||||
|
||||
void unregisterChannel(pvAccessID sid);
|
||||
|
||||
ServerChannel::shared_pointer getChannel(pvAccessID sid);
|
||||
|
||||
int getChannelCount();
|
||||
|
||||
epics::pvData::PVField::shared_pointer getSecurityToken() {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingServerTCPTransportCodec::getSecurityToken() enter:"
|
||||
" (threadId: %u)",
|
||||
epicsThreadGetIdSelf());
|
||||
|
||||
return epics::pvData::PVField::shared_pointer();
|
||||
}
|
||||
|
||||
void lock() {
|
||||
// noop
|
||||
}
|
||||
|
||||
void unlock() {
|
||||
// noop
|
||||
}
|
||||
|
||||
void acquire() {
|
||||
// noop, since does not make sence on itself
|
||||
}
|
||||
|
||||
void release() {
|
||||
// noop, since does not make sence on itself
|
||||
}
|
||||
|
||||
bool verify(epics::pvData::int32 timeoutMs) {
|
||||
|
||||
LOG(logLevelTrace,
|
||||
"BlockingServerTCPTransportCodec::verify() enter: "
|
||||
"timeoutMs:%d (threadId: %u)",
|
||||
timeoutMs, epicsThreadGetIdSelf());
|
||||
|
||||
TransportSender::shared_pointer transportSender =
|
||||
std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
|
||||
enqueueSendRequest(transportSender);
|
||||
verified();
|
||||
return true;
|
||||
}
|
||||
|
||||
void verified() {
|
||||
}
|
||||
|
||||
void aliveNotification() {
|
||||
// noop on server-side
|
||||
}
|
||||
|
||||
void send(epics::pvData::ByteBuffer* buffer,
|
||||
TransportSendControl* control);
|
||||
|
||||
virtual ~BlockingServerTCPTransportCodec();
|
||||
|
||||
private:
|
||||
|
||||
/**
|
||||
* Last SID cache.
|
||||
*/
|
||||
pvAccessID _lastChannelSID;
|
||||
|
||||
/**
|
||||
* Channel table (SID -> channel mapping).
|
||||
*/
|
||||
std::map<pvAccessID, ServerChannel::shared_pointer> _channels;
|
||||
|
||||
epics::pvData::Mutex _channelsMutex;
|
||||
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
#endif /* CODEC_H_ */
|
||||
@@ -51,6 +51,12 @@ testChannelAccess_SRCS = testChannelAccess channelAccessIFTest
|
||||
testChannelAccess_LIBS += pvData pvAccess pvMB Com
|
||||
TESTS += testChannelAccess
|
||||
|
||||
TESTPROD_HOST += testCodec
|
||||
testCodec_SRCS = testCodec
|
||||
testCodec_LIBS += pvData pvAccess pvMB Com
|
||||
TESTS += testCodec
|
||||
|
||||
|
||||
PROD_HOST += pvget
|
||||
pvget_SRCS += pvget.cpp
|
||||
pvget_LIBS += pvData pvAccess pvMB Com
|
||||
|
||||
@@ -48,7 +48,7 @@ int ChannelAccessIFTest::runAllTest() {
|
||||
testPlan(152);
|
||||
#endif
|
||||
|
||||
test_implementation();
|
||||
/* test_implementation();
|
||||
test_providerName();
|
||||
|
||||
test_createEmptyChannel();
|
||||
@@ -56,10 +56,10 @@ int ChannelAccessIFTest::runAllTest() {
|
||||
test_createChannel();
|
||||
test_recreateChannelOnDestroyedProvider();
|
||||
test_findEmptyChannel();
|
||||
test_findChannel();
|
||||
test_findChannel();*/
|
||||
test_channel();
|
||||
|
||||
test_channelGetWithInvalidChannelAndRequester();
|
||||
/* test_channelGetWithInvalidChannelAndRequester();
|
||||
test_channelGetNoProcess();
|
||||
test_channelGetIntProcess();
|
||||
test_channelGetTestNoConnection();
|
||||
@@ -95,7 +95,7 @@ int ChannelAccessIFTest::runAllTest() {
|
||||
|
||||
test_channelArray();
|
||||
test_channelArray_destroy();
|
||||
test_channelArrayTestNoConnection();
|
||||
test_channelArrayTestNoConnection();*/
|
||||
|
||||
#ifdef ENABLE_STRESS_TESTS
|
||||
test_stressConnectDisconnect();
|
||||
@@ -415,7 +415,7 @@ void ChannelAccessIFTest::test_channel() {
|
||||
testOk(channel->getConnectionState() == Channel::DESTROYED ,
|
||||
"%s: channel connection state DESTROYED ", CURRENT_FUNCTION);
|
||||
|
||||
testDiag("%s: destroying the channel yet again", CURRENT_FUNCTION);
|
||||
/* testDiag("%s: destroying the channel yet again", CURRENT_FUNCTION);
|
||||
channel->destroy();
|
||||
|
||||
succStatus = channelReq->waitUntilStateChange(getTimeoutSec());
|
||||
@@ -432,7 +432,7 @@ void ChannelAccessIFTest::test_channel() {
|
||||
testOk(!channel->isConnected(), "%s: yet again destroyed channel should not be connected ", CURRENT_FUNCTION);
|
||||
testOk(channel->getConnectionState() == Channel::DESTROYED ,
|
||||
"%s: yet again destroyed channel connection state DESTROYED ", CURRENT_FUNCTION);
|
||||
|
||||
*/
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -88,6 +88,7 @@ class ChannelAccessIFRemoteTest: public ChannelAccessIFTest {
|
||||
|
||||
MAIN(testChannelProvider)
|
||||
{
|
||||
SET_LOG_LEVEL(logLevelTrace);
|
||||
ChannelAccessIFRemoteTest caRemoteTest;
|
||||
return caRemoteTest.runAllTest();
|
||||
}
|
||||
|
||||
3195
testApp/remote/testCodec.cpp
Normal file
3195
testApp/remote/testCodec.cpp
Normal file
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user