diff --git a/pvAccessApp/remote/blockingTCPTransport.cpp b/pvAccessApp/remote/blockingTCPTransport.cpp index 113cd48..a4778ec 100644 --- a/pvAccessApp/remote/blockingTCPTransport.cpp +++ b/pvAccessApp/remote/blockingTCPTransport.cpp @@ -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 diff --git a/pvAccessApp/remote/blockingUDPTransport.cpp b/pvAccessApp/remote/blockingUDPTransport.cpp index 3f53a8a..bb61ea9 100644 --- a/pvAccessApp/remote/blockingUDPTransport.cpp +++ b/pvAccessApp/remote/blockingUDPTransport.cpp @@ -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) { diff --git a/pvAccessApp/remote/channelSearchManager.cpp b/pvAccessApp/remote/channelSearchManager.cpp index e1c2358..c908e87 100644 --- a/pvAccessApp/remote/channelSearchManager.cpp +++ b/pvAccessApp/remote/channelSearchManager.cpp @@ -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); diff --git a/pvAccessApp/remote/remote.h b/pvAccessApp/remote/remote.h index 46f7a12..9f0c547 100644 --- a/pvAccessApp/remote/remote.h +++ b/pvAccessApp/remote/remote.h @@ -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, }; /** diff --git a/pvAccessApp/utils/inetAddressUtil.cpp b/pvAccessApp/utils/inetAddressUtil.cpp index 217c740..77eac47 100644 --- a/pvAccessApp/utils/inetAddressUtil.cpp +++ b/pvAccessApp/utils/inetAddressUtil.cpp @@ -15,6 +15,7 @@ #include #include #include +#include /* standard */ #include @@ -22,6 +23,8 @@ #include #include #include +#include +#include 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 (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 (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; } diff --git a/pvAccessApp/utils/inetAddressUtil.h b/pvAccessApp/utils/inetAddressUtil.h index 0450839..ee7f381 100644 --- a/pvAccessApp/utils/inetAddressUtil.h +++ b/pvAccessApp/utils/inetAddressUtil.h @@ -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); /** diff --git a/testApp/remote/testRemoteClientImpl.cpp b/testApp/remote/testRemoteClientImpl.cpp index 846d7c7..f917251 100644 --- a/testApp/remote/testRemoteClientImpl.cpp +++ b/testApp/remote/testRemoteClientImpl.cpp @@ -1,3 +1,4 @@ + /* testRemoteClientImpl.cpp */ /* Author: Matej Sekoranja Date: 2011.1.1 */ @@ -13,6 +14,8 @@ #include #include #include +#include +#include #include #include #include @@ -428,7 +431,38 @@ typedef std::map 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"<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 @@ -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 Channel. * @author Matej Sekoranja @@ -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(); // 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* 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(); diff --git a/testApp/utils/inetAddressUtilsTest.cpp b/testApp/utils/inetAddressUtilsTest.cpp index ce77ae4..5a594b7 100644 --- a/testApp/utils/inetAddressUtilsTest.cpp +++ b/testApp/utils/inetAddressUtilsTest.cpp @@ -9,7 +9,10 @@ #include #include + #include +#include + #include #include @@ -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: "<size()<size(); i++) { + cout<<"Broadcast address: "; + cout<at(i), false)<