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

merging

Browse Source
This commit is contained in:
Gasper Jansa
2011-01-09 21:45:08 +01:00
parent 3d59fef56e 2dcc2eb486
commit 948f706c4f
8 changed files with 375 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pvAccessApp/remote/blockingTCPTransport.cpp
+2 -2
View File
@@ -92,12 +92,12 @@ namespace epics {
_sendThreadRunning(false) {
_socketBuffer = new ByteBuffer(max(MAX_TCP_RECV
+MAX_ENSURE_DATA_BUFFER_SIZE, receiveBufferSize));
+MAX_ENSURE_DATA_BUFFER_SIZE, receiveBufferSize), EPICS_ENDIAN_BIG);
_socketBuffer->setPosition(_socketBuffer->getLimit());
_startPosition = _socketBuffer->getPosition();
// allocate buffer
_sendBuffer = new ByteBuffer(_socketBuffer->getSize());
_sendBuffer = new ByteBuffer(_socketBuffer->getSize(), EPICS_ENDIAN_BIG);
_maxPayloadSize = _sendBuffer->getSize()-2*CA_MESSAGE_HEADER_SIZE; // one for header, one for flow control
// get send buffer size
pvAccessApp/remote/blockingUDPTransport.cpp
+2 -2
View File
@@ -39,8 +39,8 @@ namespace epics {
_closed(false), _responseHandler(responseHandler),
_channel(channel), _sendAddresses(sendAddresses),
_ignoredAddresses(NULL), _sendTo(NULL), _receiveBuffer(
new ByteBuffer(MAX_UDP_RECV)), _sendBuffer(
new ByteBuffer(MAX_UDP_RECV)),
new ByteBuffer(MAX_UDP_RECV, EPICS_ENDIAN_BIG)), _sendBuffer(
new ByteBuffer(MAX_UDP_RECV, EPICS_ENDIAN_BIG)),
_lastMessageStartPosition(0), _readBuffer(
new char[MAX_UDP_RECV]), _mutex(new Mutex()),
_threadId(NULL) {
pvAccessApp/remote/channelSearchManager.cpp
+2 -2
View File
@@ -416,7 +416,7 @@ ChannelSearchManager::ChannelSearchManager(ClientContextImpl* context):
_sequenceNumber(0)
{
// create and initialize send buffer
_sendBuffer = new ByteBuffer(MAX_UDP_SEND);
_sendBuffer = new ByteBuffer(MAX_UDP_SEND, EPICS_ENDIAN_BIG);
initializeSendBuffer();
// TODO should be configurable
@@ -551,7 +551,7 @@ void ChannelSearchManager::initializeSendBuffer()
_sendBuffer->clear();
_sendBuffer->putShort(CA_MAGIC_AND_VERSION);
_sendBuffer->putByte((int8)0); // data
_sendBuffer->putByte((int8)3); // beacon
_sendBuffer->putByte((int8)3); // search
_sendBuffer->putInt(sizeof(int32)/sizeof(int8) + 1); // "zero" payload
_sendBuffer->putInt(_sequenceNumber);
pvAccessApp/remote/remote.h
+5 -4
View File
@@ -33,11 +33,12 @@ namespace epics {
enum MessageCommands {
CMD_BEACON = 0, CMD_CONNECTION_VALIDATION = 1, CMD_ECHO = 2,
CMD_SEARCH = 3, CMD_INTROSPECTION_SEARCH = 5,
CMD_CREATE_CHANNEL = 7, CMD_DESTROY_CHANNEL = 8, CMD_GET = 10,
CMD_PUT = 11, CMD_PUT_GET = 12, CMD_MONITOR = 13, CMD_ARRAY = 14,
CMD_SEARCH = 3, CMD_SEARCH_RESPONSE = 4,
CMD_INTROSPECTION_SEARCH = 5, CMD_CREATE_CHANNEL = 7,
CMD_DESTROY_CHANNEL = 8, CMD_GET = 10, CMD_PUT = 11,
CMD_PUT_GET = 12, CMD_MONITOR = 13, CMD_ARRAY = 14,
CMD_CANCEL_REQUEST = 15, CMD_PROCESS = 16, CMD_GET_FIELD = 17,
CMD_RPC = 20,
CMD_MESSAGE = 18, CMD_MULTIPLE_DATA = 19, CMD_RPC = 20,
};
/**
pvAccessApp/utils/inetAddressUtil.cpp
+119 -69
View File
@@ -15,6 +15,7 @@
#include <ellLib.h>
#include <epicsAssert.h>
#include <epicsException.h>
#include <errlog.h>
/* standard */
#include <vector>
@@ -22,6 +23,8 @@
#include <cstdlib>
#include <sstream>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <net/if.h>
using namespace std;
using namespace epics::pvData;
@@ -29,81 +32,128 @@ using namespace epics::pvData;
namespace epics {
namespace pvAccess {
/* copied from EPICS v3 ca/iocinf.cpp
* removeDuplicateAddresses ()
/* port of osiSockDiscoverBroadcastAddresses() in
* epics/base/src/libCom/osi/os/default/osdNetIntf.c
*/
void removeDuplicateAddresses(ELLLIST *pDestList, ELLLIST *pSrcList,
int silent) {
ELLNODE *pRawNode;
while((pRawNode = ellGet(pSrcList))) {
STATIC_ASSERT(offsetof(osiSockAddrNode, node)==0);
osiSockAddrNode *pNode =
reinterpret_cast<osiSockAddrNode *> (pRawNode);
osiSockAddrNode *pTmpNode;
if(pNode->addr.sa.sa_family==AF_INET) {
pTmpNode = (osiSockAddrNode *)ellFirst (pDestList); // X aCC 749
while(pTmpNode) {
if(pTmpNode->addr.sa.sa_family==AF_INET) {
if(pNode->addr.ia.sin_addr.s_addr
==pTmpNode->addr.ia.sin_addr.s_addr
&&pNode->addr.ia.sin_port
==pTmpNode->addr.ia.sin_port) {
if(!silent) {
char buf[64];
ipAddrToDottedIP(&pNode->addr.ia, buf,
sizeof(buf));
fprintf(
stderr,
"Warning: Duplicate EPICS CA Address list entry \"%s\" discarded\n",
buf);
}
free(pNode);
pNode = NULL;
break;
}
}
pTmpNode = (osiSockAddrNode *)ellNext (&pTmpNode->node); // X aCC 749
}
if(pNode) {
ellAdd(pDestList, &pNode->node);
}
}
else {
ellAdd(pDestList, &pNode->node);
}
}
}
InetAddrVector* getBroadcastAddresses(SOCKET sock) {
ELLLIST bcastList;
ELLLIST tmpList;
osiSockAddr addr;
static const unsigned nelem = 100;
int status;
struct ifconf ifconf;
struct ifreq* pIfreqList;
osiSockAddr* pNewNode;
ellInit ( &bcastList ); // X aCC 392
ellInit ( &tmpList ); // X aCC 392
InetAddrVector* retVector = new InetAddrVector();
addr.ia.sin_family = AF_UNSPEC;
osiSockDiscoverBroadcastAddresses(&bcastList, sock, &addr);
removeDuplicateAddresses(&tmpList, &bcastList, 1);
// forcePort ( &bcastList, port ); // if needed copy from ca/iocinf.cpp
int size = ellCount(&bcastList );
InetAddrVector* retVector = new InetAddrVector(size);
ELLNODE *pRawNode;
while((pRawNode = ellGet(&tmpList))) {
osiSockAddrNode *pNode =
reinterpret_cast<osiSockAddrNode *> (pRawNode);
osiSockAddr* posa = new osiSockAddr;
memcpy(posa, &(pNode->addr), sizeof(osiSockAddr));
retVector->push_back(posa);
free(pNode); // using free because it is allocated by calloc
/*
* use pool so that we avoid using too much stack space
*
* nelem is set to the maximum interfaces
* on one machine here
*/
pIfreqList = new ifreq[nelem];
if(!pIfreqList) {
errlogSevPrintf(errlogMajor,
"getBroadcastAddresses(): no memory to complete request");
return retVector;
}
// get number of interfaces
ifconf.ifc_len = nelem*sizeof(ifreq);
ifconf.ifc_req = pIfreqList;
status = ioctl(sock, SIOCGIFCONF, &ifconf);
if(status<0||ifconf.ifc_len==0) {
errlogSevPrintf(
errlogMinor,
"getBroadcastAddresses(): unable to fetch network interface configuration");
delete[] pIfreqList;
return retVector;
}
errlogPrintf("Found %d interfaces\n", ifconf.ifc_len);
for(int i = 0; i<=ifconf.ifc_len; i++) {
/*
* If its not an internet interface then dont use it
*/
if(pIfreqList[i].ifr_addr.sa_family!=AF_INET) continue;
status = ioctl(sock, SIOCGIFFLAGS, &pIfreqList[i]);
if(status) {
errlogSevPrintf(
errlogMinor,
"getBroadcastAddresses(): net intf flags fetch for \"%s\" failed",
pIfreqList[i].ifr_name);
continue;
}
/*
* dont bother with interfaces that have been disabled
*/
if(!(pIfreqList[i].ifr_flags&IFF_UP)) continue;
/*
* dont use the loop back interface
*/
if(pIfreqList[i].ifr_flags&IFF_LOOPBACK) continue;
pNewNode = new osiSockAddr;
if(pNewNode==NULL) {
errlogSevPrintf(errlogMajor,
"getBroadcastAddresses(): no memory available for configuration");
delete[] pIfreqList;
return retVector;
}
/*
* If this is an interface that supports
* broadcast fetch the broadcast address.
*
* Otherwise if this is a point to point
* interface then use the destination address.
*
* Otherwise CA will not query through the
* interface.
*/
if(pIfreqList[i].ifr_flags&IFF_BROADCAST) {
status = ioctl(sock, SIOCGIFBRDADDR, &pIfreqList[i]);
if(status) {
errlogSevPrintf(
errlogMinor,
"getBroadcastAddresses(): net intf \"%s\": bcast addr fetch fail",
pIfreqList->ifr_name);
delete pNewNode;
continue;
}
pNewNode->sa = pIfreqList[i].ifr_broadaddr;
}
#ifdef IFF_POINTOPOINT
else if(pIfreqList->ifr_flags&IFF_POINTOPOINT) {
status = ioctl(sock, SIOCGIFDSTADDR, &pIfreqList[i]);
if(status) {
errlogSevPrintf(
errlogMinor,
"getBroadcastAddresses(): net intf \"%s\": pt to pt addr fetch fail",
pIfreqList[i].ifr_name);
delete pNewNode;
continue;
}
pNewNode->sa = pIfreqList[i].ifr_dstaddr;
}
#endif
else {
errlogSevPrintf(
errlogMinor,
"getBroadcastAddresses(): net intf \"%s\": not point to point or bcast?",
pIfreqList[i].ifr_name);
delete pNewNode;
continue;
}
retVector->push_back(pNewNode);
}
delete[] pIfreqList;
return retVector;
}
pvAccessApp/utils/inetAddressUtil.h
+1 -3
View File
@@ -33,9 +33,7 @@ namespace epics {
/**
* returns a vector containing all the IPv4 broadcast addresses
* on this machine. IPv6 doesn't have a local broadcast address.
*
* TODO Check implementation/rewrite this
*/
*/
InetAddrVector* getBroadcastAddresses(SOCKET sock);
/**
testApp/remote/testRemoteClientImpl.cpp
+233 -40
View File
@@ -1,3 +1,4 @@
/* testRemoteClientImpl.cpp */
/* Author: Matej Sekoranja Date: 2011.1.1 */
@@ -13,6 +14,8 @@
#include <caConstants.h>
#include <timer.h>
#include <blockingUDP.h>
#include <blockingTCP.h>
#include <namedLockPattern.h>
#include <inetAddressUtil.h>
#include <hexDump.h>
#include <remote.h>
@@ -428,7 +431,38 @@ typedef std::map<pvAccessID, ResponseRequest*> IOIDResponseRequestMap;
class ClientContextImpl;
class DebugResponse : public ResponseHandler, private epics::pvData::NoDefaultMethods {
public:
/**
* @param context
*/
DebugResponse()
{
}
virtual ~DebugResponse() {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer)
{
char ipAddrStr[48];
std::cout << "ole" << std::endl;
ipAddrToA(&responseFrom->ia, ipAddrStr, sizeof(ipAddrStr));
std::cout << "ole2" << std::endl;
ostringstream prologue;
prologue<<"Message [0x"<<hex<<(int)command<<", v0x"<<hex;
prologue<<(int)version<<"] received from "<<ipAddrStr;
std::cout << "ole / " << prologue.str() << std::endl;
hexDump(prologue.str(), "received",
(const int8*)payloadBuffer->getArray(),
payloadBuffer->getPosition(), payloadSize);
}
};
/**
* CA response handler - main handler which dispatches responses to appripriate handlers.
@@ -458,7 +492,7 @@ class ClientResponseHandler : public ResponseHandler, private epics::pvData::NoD
* @param context
*/
ClientResponseHandler(ClientContextImpl* context) : m_context(context) {
static ResponseHandler* badResponse = 0; //new BadResponse(context);
static ResponseHandler* badResponse = new DebugResponse();
static ResponseHandler* dataResponse = 0; //new DataResponseHandler(context);
#define HANDLER_COUNT 28
@@ -497,6 +531,8 @@ class ClientResponseHandler : public ResponseHandler, private epics::pvData::NoD
Transport* transport, int8 version, int8 command,
int payloadSize, ByteBuffer* payloadBuffer)
{
int c = command+0;
std::cout << "received " << c << std::endl;
if (command < 0 || command >= HANDLER_COUNT)
{
// TODO context.getLogger().fine("Invalid (or unsupported) command: " + command + ".");
@@ -509,12 +545,28 @@ class ClientResponseHandler : public ResponseHandler, private epics::pvData::NoD
}
// delegate
m_handlerTable[command]->handleResponse(responseFrom, transport, version, command, payloadSize, payloadBuffer);
m_handlerTable[c]->handleResponse(responseFrom, transport, version, command, payloadSize, payloadBuffer);
}
};
class TCI : public TransportSendControl {
public:
virtual void flushSerializeBuffer() {
}
virtual void ensureBuffer(int size) {
}
virtual void startMessage(int8 command, int ensureCapacity){}
virtual void endMessage() {}
virtual void flush(bool lastMessageCompleted) {}
virtual void setRecipient(const osiSockAddr& sendTo) {}
};
#include <arrayFIFO.h>
@@ -548,19 +600,37 @@ class BaseSearchInstance : public SearchInstance
virtual void removeAndUnsetListOwnership() {}
virtual int getOwnerIndex() { return 0; }
virtual bool generateSearchRequestMessage(ByteBuffer* buffer, TransportSendControl* control) { return false; };
virtual bool generateSearchRequestMessage(ByteBuffer* requestMessage, TransportSendControl* control)
{
const int DATA_COUNT_POSITION = CA_MESSAGE_HEADER_SIZE + sizeof(int32)/sizeof(int8) + 1;
const int PAYLOAD_POSITION = sizeof(int16)/sizeof(int8) + 2;
int16 dataCount = requestMessage->getShort(DATA_COUNT_POSITION);
dataCount++;
if(dataCount >= MAX_SEARCH_BATCH_COUNT)
{
return false;
}
const string name = getChannelName();
// not nice...
const int addedPayloadSize = sizeof(int32)/sizeof(int8) + (1 + sizeof(int32)/sizeof(int8) + name.length());
if(requestMessage->getRemaining() < addedPayloadSize)
{
return false;
}
requestMessage->putInt(getChannelID());
SerializeHelper::serializeString(name, requestMessage, control);
requestMessage->putInt(PAYLOAD_POSITION, requestMessage->getPosition() - CA_MESSAGE_HEADER_SIZE);
requestMessage->putShort(DATA_COUNT_POSITION, dataCount);
return true;
};
};
class ChannelSearchManager { // tODO no default, etc.
public:
virtual void registerChannel(SearchInstance* channel) = 0;
virtual void unregisterChannel(SearchInstance* channel) = 0;
};
class BlockingTCPConnector;
class NamedLockPattern;
class BeaconHandlerImpl;
@@ -591,9 +661,52 @@ enum ContextState {
};
class ClientContextImpl : public ClientContext
class ClientContextImpl : public ClientContext,
public Context /* TODO */
{
class ChannelSearchManager { // tODO no default, etc.
ClientContextImpl* _context;
public:
ChannelSearchManager(ClientContextImpl* context):
_context(context) {
}
virtual void registerChannel(SearchInstance* channel) {
ByteBuffer sendBuffer(100, EPICS_ENDIAN_BIG);
// new buffer
sendBuffer.clear();
sendBuffer.putShort(CA_MAGIC_AND_VERSION);
sendBuffer.putByte((int8)0); // data
sendBuffer.putByte((int8)3); // search
sendBuffer.putInt(5); // "zero" payload
sendBuffer.putInt(0);
sendBuffer.putByte((int8)0);
sendBuffer.putShort((int16)0); // count
TCI tci;
channel->generateSearchRequestMessage(&sendBuffer, &tci);
std::cout << "sending..." << sendBuffer.getPosition() << " bytes." << std::endl;
_context->getSearchTransport()->send(&sendBuffer);
};
virtual void unregisterChannel(SearchInstance* channel) {};
};
/**
* Implementation of CAJ JCA <code>Channel</code>.
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
@@ -1196,7 +1309,6 @@ class ChannelImpl :
// TODO
}
virtual void getField(GetFieldRequester *requester,epics::pvData::String subField)
{
requester->getDone(getStatusCreate()->getStatusOK(),m_pvStructure->getSubField(subField)->getField());
@@ -1394,7 +1506,25 @@ class ChannelImpl :
Lock lock(&m_contextMutex);
return m_provider;
}
virtual Timer* getTimer()
{
Lock lock(&m_contextMutex);
return m_timer;
}
virtual TransportRegistry* getTransportRegistry()
{
Lock lock(&m_contextMutex);
return m_transportRegistry;
}
virtual BlockingUDPTransport* getSearchTransport()
{
Lock lock(&m_contextMutex);
return m_searchTransport;
}
virtual void initialize() {
Lock lock(&m_contextMutex);
@@ -1484,36 +1614,92 @@ class ChannelImpl :
void internalInitialize() {
m_timer = new Timer("pvAccess-client timer", lowPriority);
/* TODO
connector = new BlockingTCPConnector(this, receiveBufferSize, beaconPeriod);
transportRegistry = new TransportRegistry();
namedLocker = new NamedLockPattern();
*/
m_connector = new BlockingTCPConnector(this, m_receiveBufferSize, m_beaconPeriod);
m_transportRegistry = new TransportRegistry();
m_namedLocker = new NamedLockPattern<String>();
// setup UDP transport
initializeUDPTransport();
// TODO
// setup search manager
//channelSearchManager = new ChannelSearchManager(this);
m_channelSearchManager = new ChannelSearchManager(this);
}
void initializeUDPTransport() {
// TODO
}
/**
* Initialized UDP transport (broadcast socket and repeater connection).
*/
void initializeUDPTransport() {
// setup UDP transport
try
{
// where to bind (listen) address
osiSockAddr listenLocalAddress;
listenLocalAddress.ia.sin_family = AF_INET;
listenLocalAddress.ia.sin_port = htons(m_broadcastPort);
listenLocalAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
// where to send address
InetAddrVector* broadcastAddresses = getSocketAddressList("192.168.1.255", m_broadcastPort);
// TODO getBroadcastAddresses(broadcastPort)
/// TOD !!!! addresses !!!!! by pointer and not copied
BlockingUDPConnector* broadcastConnector = new BlockingUDPConnector(true, broadcastAddresses, true);
m_broadcastTransport = (BlockingUDPTransport*)broadcastConnector->connect(
0, new ClientResponseHandler(this),
listenLocalAddress, CA_MINOR_PROTOCOL_REVISION,
CA_DEFAULT_PRIORITY);
BlockingUDPConnector* searchConnector = new BlockingUDPConnector(false, broadcastAddresses, true);
// undefined address
osiSockAddr undefinedAddress;
undefinedAddress.ia.sin_family = AF_INET;
undefinedAddress.ia.sin_port = htons(0);
undefinedAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
m_searchTransport = (BlockingUDPTransport*)searchConnector->connect(
0, new ClientResponseHandler(this),
undefinedAddress, CA_MINOR_PROTOCOL_REVISION,
CA_DEFAULT_PRIORITY);
// set broadcast address list
if (!m_addressList.empty())
{
// if auto is true, add it to specified list
InetAddrVector* appendList = 0;
if (m_autoAddressList)
appendList = m_broadcastTransport->getSendAddresses();
InetAddrVector* list = getSocketAddressList(m_addressList, m_broadcastPort, appendList);
// TODO delete !!!!
if (list && list->size()) {
m_broadcastTransport->setBroadcastAddresses(list);
m_searchTransport->setBroadcastAddresses(list);
}
}
m_broadcastTransport->start();
m_searchTransport->start();
}
catch (...)
{
// TODO
}
}
void internalDestroy() {
// stop searching
/* TODO
if (m_channelSearchManager)
channelSearchManager->destroy();
*/
delete m_channelSearchManager; //->destroy();
// stop timer
if (m_timer)
delete m_timer;
//
// cleanup
//
@@ -1521,13 +1707,15 @@ class ChannelImpl :
// this will also close all CA transports
destroyAllChannels();
// close broadcast transport
/* TODO
// TODO destroy !!!
if (m_broadcastTransport)
m_broadcastTransport->destroy(true);
if (m_searchTransport != null)
m_searchTransport->destroy(true);
*/
delete m_broadcastTransport; //->destroy(true);
if (m_searchTransport)
delete m_searchTransport; //->destroy(true);
if (m_namedLocker) delete m_namedLocker;
if (m_transportRegistry) delete m_transportRegistry;
if (m_connector) delete m_connector;
m_provider->destroy();
delete m_version;
@@ -1779,7 +1967,7 @@ class ChannelImpl :
/**
* Context instance.
*/
NamedLockPattern* m_namedLocker;
NamedLockPattern<String>* m_namedLocker;
/**
* Context instance.
@@ -2077,13 +2265,17 @@ int main(int argc,char *argv[])
{
ClientContextImpl* context = new ClientContextImpl();
context->printInfo();
context->initialize();
context->printInfo();
epicsThreadSleep ( 1.0 );
ChannelFindRequesterImpl findRequester;
context->getProvider()->channelFind("something", &findRequester);
//ChannelFindRequesterImpl findRequester;
//context->getProvider()->channelFind("something", &findRequester);
ChannelRequesterImpl channelRequester;
Channel* channel = context->getProvider()->createChannel("test", &channelRequester);
Channel* channel = context->getProvider()->createChannel("structureArrayTest", &channelRequester);
channel->printInfo();
/*
GetFieldRequesterImpl getFieldRequesterImpl;
@@ -2122,6 +2314,7 @@ int main(int argc,char *argv[])
monitor->destroy();
*/
epicsThreadSleep ( 10.0 );
channel->destroy();
context->destroy();
testApp/utils/inetAddressUtilsTest.cpp
+11 -1
View File
@@ -9,7 +9,10 @@
#include <byteBuffer.h>
#include <pvType.h>
#include <epicsAssert.h>
#include <osiSock.h>
#include <iostream>
#include <cstring>
@@ -123,8 +126,15 @@ int main(int argc, char *argv[]) {
assert(strncmp(buff->getArray(), src, 16)==0);
cout<<"\nPASSED!\n";
// TODO add test for 'getBroadcastAddresses'
SOCKET socket = epicsSocketCreate(AF_INET, SOCK_STREAM, IPPROTO_TCP);
InetAddrVector* broadcasts = getBroadcastAddresses(socket);
cout<<"Broadcast addresses: "<<broadcasts->size()<<endl;
for(size_t i = 0; i<broadcasts->size(); i++) {
cout<<"Broadcast address: ";
cout<<inetAddressToString(broadcasts->at(i), false)<<endl;
}
delete broadcasts;
delete addr;
return 0;