This is Unofficial EPICS BASE Doxygen Site: epics::pvAccess::BlockingUDPTransport Class Reference

Definition at line 181 of file blockingUDP.h.

                                                                  {
         _sendBuffer.align(alignment);
     }

virtual void epics::pvAccess::BlockingUDPTransport::alignData ( std::size_t alignment )

inlinevirtual

Align buffer. Note that this takes care only current buffer alignment. If streaming protocol is used, care must be taken that entire stream is aligned.

Parameters

alignment size in bytes, must be power of two.

Definition at line 132 of file blockingUDP.h.

                                                                {
         _receiveBuffer.align(alignment);
     }

virtual void epics::pvAccess::BlockingUDPTransport::authNZMessage ( epics::pvData::PVStructure::shared_pointer const & data )

inlinevirtual

Pass data to the active security plug-in session.

Parameters

data	the data (any data), can be `null`.

Implements epics::pvAccess::Transport.

Definition at line 109 of file blockingUDP.h.

                                                                                                  {
         // noop
     }

virtual std::tr1::shared_ptr<const epics::pvData::Field> epics::pvAccess::BlockingUDPTransport::cachedDeserialize ( epics::pvData::ByteBuffer * buffer )

inlinevirtual

deserialize via cache

Parameters

buffer buffer to be deserialized from

Definition at line 193 of file blockingUDP.h.

     {
         // no cache
         // TODO
         return epics::pvData::getFieldCreate()->deserialize(buffer, this);
     }

virtual void epics::pvAccess::BlockingUDPTransport::cachedSerialize	(	const std::tr1::shared_ptr< const epics::pvData::Field > &	field,
		epics::pvData::ByteBuffer *	buffer
	)

inlinevirtual

Definition at line 185 of file blockingUDP.h.

     {
         // no cache
         field->serialize(buffer, this);
     }

void epics::pvAccess::BlockingUDPTransport::close ( )

virtual

Close transport.

Implements epics::pvAccess::Transport.

Definition at line 116 of file blockingUDPTransport.cpp.

                                  {
     close(true);
 }

virtual bool epics::pvAccess::BlockingUDPTransport::directDeserialize	(	epics::pvData::ByteBuffer *	existingBuffer,
		char *	deserializeTo,
		std::size_t	elementCount,
		std::size_t	elementSize
	)

inlinevirtual

Method for deserializing array data. Hook for supplying custom deserialization implementation. The deserialization implementation need not be provided. Returns true if method performs deserialization, false otherwise. This should only be used for arrays of primitive types. i.e. boolean, byte,..., double. It cannot be called for string, structure, or union arrays.

Parameters

existingBuffer	the existing buffer from the caller.
deserializeTo	location of data.
elementCount	number of elements.
elementSize	element size.

Returns: true if deserialization performed, else false.

Definition at line 142 of file blockingUDP.h.

     {
         return false;
     }

virtual bool epics::pvAccess::BlockingUDPTransport::directSerialize	(	epics::pvData::ByteBuffer *	existingBuffer,
		const char *	toSerialize,
		std::size_t	elementCount,
		std::size_t	elementSize
	)

inlinevirtual

Method for serializing primitive array data. Hook for supplying custom serialization implementation. The serialization implementation need not be provided. Returns true if method performs serialization, false otherwise. This should only be used for arrays of primitive types, i. e. boolean, byte,..., double. It cannot be called for string, structure, or union arrays.

Parameters

existingBuffer	the existing buffer from the caller.
toSerialize	location of data to be put into buffer.
elementCount	number of elements.
elementSize	element size.

Returns: true if serialization performed, else false.

Definition at line 136 of file blockingUDP.h.

     {
         return false;
     }

void epics::pvAccess::BlockingUDPTransport::endMessage ( )

virtual

Definition at line 217 of file blockingUDPTransport.cpp.

                                       {
     _sendBuffer.putInt(
         _lastMessageStartPosition+(sizeof(int16)+2),
         _sendBuffer.getPosition()-_lastMessageStartPosition-PVA_MESSAGE_HEADER_SIZE);
 }

void epics::pvAccess::BlockingUDPTransport::enqueueSendRequest ( TransportSender::shared_pointer const & sender )

virtual

Enqueue send request.

Parameters

sender

Implements epics::pvAccess::Transport.

Definition at line 186 of file blockingUDPTransport.cpp.

                                                                                           {
     Lock lock(_sendMutex);
 
     _sendToEnabled = false;
     _sendBuffer.clear();
     {
         epicsGuard<TransportSender> G(*sender);
         sender->send(&_sendBuffer, this);
     }
     endMessage();
     if(!_sendToEnabled)
         send(&_sendBuffer);
     else
         send(&_sendBuffer, _sendTo);
 }

virtual void epics::pvAccess::BlockingUDPTransport::ensureBuffer ( std::size_t size )

inlinevirtual

Make sure buffer has at least size bytes remaining. If not flush existing buffer and provide a new one.

Parameters

size	The number of bytes.

Definition at line 177 of file blockingUDP.h.

                                                           {
         // noop
     }

void epics::pvAccess::BlockingUDPTransport::ensureData ( std::size_t size )

virtual

Helper method. Ensures specified size of bytes, provides it if necessary.

Parameters

size	The number of bytes.

Definition at line 120 of file blockingUDPTransport.cpp.

                                                     {
     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());
 }

virtual void epics::pvAccess::BlockingUDPTransport::flush ( bool )

inlinevirtual

Definition at line 151 of file blockingUDP.h.

                                               {
         // noop since all UDP requests are sent immediately
     }

void epics::pvAccess::BlockingUDPTransport::flushSendQueue ( )

virtual

Flush send queue (sent messages).

Implements epics::pvAccess::Transport.

Definition at line 203 of file blockingUDPTransport.cpp.

 {
     // noop (note different sent addresses are possible)
 }

virtual void epics::pvAccess::BlockingUDPTransport::flushSerializeBuffer ( )

inlinevirtual

Done with this buffer. Flush it.

Definition at line 173 of file blockingUDP.h.

                                                        {
         // noop
     }

const osiSockAddr* epics::pvAccess::BlockingUDPTransport::getBindAddress ( ) const

inline

Get bind address.

Returns: bind address.

Definition at line 257 of file blockingUDP.h.

                                               {
         return &_bindAddress;
     }

const InetAddrVector& epics::pvAccess::BlockingUDPTransport::getIgnoredAddresses ( ) const

inline

Get list of ignored addresses.

Returns: ignored addresses.

Definition at line 219 of file blockingUDP.h.

                                                       {
         return _ignoredAddresses;
     }

const osiSockAddr& epics::pvAccess::BlockingUDPTransport::getLocalMulticastAddress ( ) const

inline

Definition at line 169 of file blockingUDP.h.

                                                         {
         return _localMulticastAddress;
     }

virtual epics::pvData::int16 epics::pvAccess::BlockingUDPTransport::getPriority ( ) const

inlinevirtual

Transport priority.

Returns: protocol priority.

Implements epics::pvAccess::Transport.

Definition at line 86 of file blockingUDP.h.

                                                                 {
         return PVA_DEFAULT_PRIORITY;
     }

virtual std::size_t epics::pvAccess::BlockingUDPTransport::getReceiveBufferSize ( ) const

inlinevirtual

Get receive buffer size.

Returns: receive buffer size.

Implements epics::pvAccess::Transport.

Definition at line 82 of file blockingUDP.h.

                                                                   {
         return _receiveBuffer.getSize();
     }

virtual const osiSockAddr& epics::pvAccess::BlockingUDPTransport::getRemoteAddress ( ) const

inlinevirtual

Implements epics::pvAccess::Transport.

Definition at line 70 of file blockingUDP.h.

                                                                        {
         return _remoteAddress;
     }

virtual const std::string& epics::pvAccess::BlockingUDPTransport::getRemoteName ( ) const

inlinevirtual

Implements epics::pvAccess::Transport.

Definition at line 74 of file blockingUDP.h.

                                                                   {
         return _remoteName;
     }

const InetAddrVector& epics::pvAccess::BlockingUDPTransport::getSendAddresses ( )

inline

Get list of send addresses.

Returns: send addresses.

Definition at line 249 of file blockingUDP.h.

                                              {
         return _sendAddresses;
     }

const InetAddrVector& epics::pvAccess::BlockingUDPTransport::getTappedNIF ( ) const

inline

Get list of tapped NIF addresses.

Returns: tapped NIF addresses.

Definition at line 235 of file blockingUDP.h.

                                                {
         return _tappedNIF;
     }

virtual std::string epics::pvAccess::BlockingUDPTransport::getType ( ) const

inlinevirtual

Get protocol type (tcp, udp, ssl, etc.).

Returns: protocol type.

Implements epics::pvAccess::Transport.

Definition at line 78 of file blockingUDP.h.

                                                      {
         return std::string("udp");
     }

bool epics::pvAccess::BlockingUDPTransport::hasLocalMulticastAddress ( ) const

inline

Definition at line 165 of file blockingUDP.h.

                                           {
         return _localMulticastAddressEnabled;
     }

bool epics::pvAccess::BlockingUDPTransport::isBroadcastAddress	(	const osiSockAddr *	address,
		const InetAddrVector &	broadcastAddresses
	)

inline

Definition at line 261 of file blockingUDP.h.

     {
         for (size_t i = 0; i < broadcastAddresses.size(); i++)
             if (broadcastAddresses[i].ia.sin_addr.s_addr == address->ia.sin_addr.s_addr)
                 return true;
         return false;
     }

virtual bool epics::pvAccess::BlockingUDPTransport::isClosed ( )

inlinevirtual

Check connection status.

Returns: true if connected.

Implements epics::pvAccess::Transport.

Definition at line 66 of file blockingUDP.h.

                                            {
         return _closed.get();
     }

void epics::pvAccess::BlockingUDPTransport::join	(	const osiSockAddr &	mcastAddr,
		const osiSockAddr &	nifAddr
	)

Definition at line 525 of file blockingUDPTransport.cpp.

 {
     struct ip_mreq imreq;
     memset(&imreq, 0, sizeof(struct ip_mreq));
 
     imreq.imr_multiaddr.s_addr = mcastAddr.ia.sin_addr.s_addr;
     imreq.imr_interface.s_addr = nifAddr.ia.sin_addr.s_addr;
 
     // join multicast group on the given interface
     int status = ::setsockopt(_channel, IPPROTO_IP, IP_ADD_MEMBERSHIP,
                               (char*)&imreq, sizeof(struct ip_mreq));
     if (status)
     {
         char errStr[64];
         epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
         throw std::runtime_error(
             string("Failed to join to the multicast group '") +
             inetAddressToString(mcastAddr) + "' on network interface '" +
             inetAddressToString(nifAddr, false) + "': " + errStr);
     }
 }

epics::pvAccess::BlockingUDPTransport::POINTER_DEFINITIONS ( BlockingUDPTransport )

virtual void epics::pvAccess::BlockingUDPTransport::release ( pvAccessID clientId )

inlinevirtual

Releases transport.

Parameters

client client (channel) releasing the transport

Implements epics::pvAccess::Transport.

Definition at line 205 of file blockingUDP.h.

205 {}

void epics::pvAccess::BlockingUDPTransport::run ( )

protectedvirtual

Definition at line 223 of file blockingUDPTransport.cpp.

                                {
     // This function is always called from only one thread - this
     // object's own thread.
 
     osiSockAddr fromAddress;
     osiSocklen_t addrStructSize = sizeof(sockaddr);
     Transport::shared_pointer thisTransport(internal_this);
 
     try {
 
         char* recvfrom_buffer_start = (char*)(_receiveBuffer.getBuffer()+RECEIVE_BUFFER_PRE_RESERVE);
         size_t recvfrom_buffer_len =_receiveBuffer.getSize()-RECEIVE_BUFFER_PRE_RESERVE;
         while(!_closed.get())
         {
             int bytesRead = recvfrom(_channel,
                                      recvfrom_buffer_start, recvfrom_buffer_len,
                                      0, (sockaddr*)&fromAddress,
                                      &addrStructSize);
 
             if(likely(bytesRead>=0)) {
                 // successfully got datagram
                 atomic::add(_totalBytesRecv, bytesRead);
                 bool ignore = false;
                 for(size_t i = 0; i <_ignoredAddresses.size(); i++)
                 {
                     if(_ignoredAddresses[i].ia.sin_addr.s_addr==fromAddress.ia.sin_addr.s_addr)
                     {
                         ignore = true;
                         if(pvAccessIsLoggable(logLevelDebug)) {
                             char strBuffer[64];
                             sockAddrToDottedIP(&fromAddress.sa, strBuffer, sizeof(strBuffer));
                             LOG(logLevelDebug, "UDP Ignore (%d) %s x- %s", bytesRead, _remoteName.c_str(), strBuffer);
                         }
                         break;
                     }
                 }
 
                 if(likely(!ignore)) {
                     if(pvAccessIsLoggable(logLevelDebug)) {
                         char strBuffer[64];
                         sockAddrToDottedIP(&fromAddress.sa, strBuffer, sizeof(strBuffer));
                         LOG(logLevelDebug, "UDP %s Rx (%d) %s <- %s", (_clientServerWithEndianFlag&0x40)?"Server":"Client", bytesRead, _remoteName.c_str(), strBuffer);
                     }
 
                     _receiveBuffer.setPosition(RECEIVE_BUFFER_PRE_RESERVE);
                     _receiveBuffer.setLimit(RECEIVE_BUFFER_PRE_RESERVE+bytesRead);
 
                     try {
                         processBuffer(thisTransport, fromAddress, &_receiveBuffer);
                     } catch(std::exception& e) {
                         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 {
 
                 int socketError = SOCKERRNO;
 
                 // interrupted or timeout
                 if (socketError == SOCK_EINTR ||
                         socketError == EAGAIN ||        // no alias in libCom
                         // windows times out with this
                         socketError == SOCK_ETIMEDOUT ||
                         socketError == SOCK_EWOULDBLOCK)
                     continue;
 
                 if (socketError == SOCK_ECONNREFUSED || // avoid spurious ECONNREFUSED in Linux
                         socketError == SOCK_ECONNRESET)     // or ECONNRESET in Windows
                     continue;
 
                 // log a 'recvfrom' error
                 if(!_closed.get())
                 {
                     char errStr[64];
                     epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
                     LOG(logLevelError, "Socket recvfrom error: %s.", errStr);
                 }
 
                 close(false);
                 break;
             }
 
         }
     } catch(...) {
         // TODO: catch all exceptions, and act accordingly
         close(false);
     }
 
     if (IS_LOGGABLE(logLevelTrace))
     {
         string threadName = "UDP-rx "+inetAddressToString(_bindAddress);
         LOG(logLevelTrace, "Thread '%s' exiting.", threadName.c_str());
     }
 }

bool epics::pvAccess::BlockingUDPTransport::send	(	const char *	buffer,
		size_t	length,
		const osiSockAddr &	address
	)

Definition at line 433 of file blockingUDPTransport.cpp.

 {
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "UDP Tx (%zu) %s -> %s.",
             length, _remoteName.c_str(), inetAddressToString(address).c_str());
     }
 
     int retval = sendto(_channel, buffer,
                         length, 0, &(address.sa), sizeof(sockaddr));
     if(unlikely(retval<0))
     {
         char errStr[64];
         epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
         LOG(logLevelDebug, "Socket sendto to %s error: %s.",
             inetAddressToString(address).c_str(), errStr);
         return false;
     }
     atomic::add(_totalBytesSent, length);
 
     return true;
 }

bool epics::pvAccess::BlockingUDPTransport::send	(	epics::pvData::ByteBuffer *	buffer,
		const osiSockAddr &	address
	)

Definition at line 456 of file blockingUDPTransport.cpp.

                                                                               {
 
     buffer->flip();
 
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "Sending %zu bytes %s -> %s.",
             buffer->getRemaining(), _remoteName.c_str(), inetAddressToString(address).c_str());
     }
 
     int retval = sendto(_channel, buffer->getBuffer(),
                         buffer->getLimit(), 0, &(address.sa), sizeof(sockaddr));
     if(unlikely(retval<0))
     {
         char errStr[64];
         epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
         LOG(logLevelDebug, "Socket sendto to %s error: %s.",
             inetAddressToString(address).c_str(), errStr);
         return false;
     }
     atomic::add(_totalBytesSent, buffer->getLimit());
 
     // all sent
     buffer->setPosition(buffer->getLimit());
 
     return true;
 }

bool epics::pvAccess::BlockingUDPTransport::send	(	epics::pvData::ByteBuffer *	buffer,
		InetAddressType	target = `inetAddressType_all`
	)

Definition at line 484 of file blockingUDPTransport.cpp.

                                                                           {
     if(_sendAddresses.empty()) return false;
 
     buffer->flip();
 
     bool allOK = true;
     for(size_t i = 0; i<_sendAddresses.size(); i++) {
 
         // filter
         if (target != inetAddressType_all)
             if ((target == inetAddressType_unicast && !_isSendAddressUnicast[i]) ||
                     (target == inetAddressType_broadcast_multicast && _isSendAddressUnicast[i]))
                 continue;
 
         if (IS_LOGGABLE(logLevelDebug))
         {
             LOG(logLevelDebug, "Sending %zu bytes %s -> %s.",
                 buffer->getRemaining(), _remoteName.c_str(), inetAddressToString(_sendAddresses[i]).c_str());
         }
 
         int retval = sendto(_channel, buffer->getBuffer(),
                             buffer->getLimit(), 0, &(_sendAddresses[i].sa),
                             sizeof(sockaddr));
         if(unlikely(retval<0))
         {
             char errStr[64];
             epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
             LOG(logLevelDebug, "Socket sendto to %s error: %s.",
                 inetAddressToString(_sendAddresses[i]).c_str(), errStr);
             allOK = false;
         }
         atomic::add(_totalBytesSent, buffer->getLimit());
     }
 
     // all sent
     buffer->setPosition(buffer->getLimit());
 
     return allOK;
 }

virtual void epics::pvAccess::BlockingUDPTransport::setByteOrder ( int byteOrder )

inlinevirtual

Set byte order.

Parameters

byteOrder byte order to set.

Implements epics::pvAccess::Transport.

Definition at line 114 of file blockingUDP.h.

                                                              {
         // called from receive thread... or before processing
         _receiveBuffer.setEndianess(byteOrder);
 
         // sync?!
         _sendBuffer.setEndianess(byteOrder);
     }

void epics::pvAccess::BlockingUDPTransport::setIgnoredAddresses ( const InetAddrVector & addresses )

inline

Set ignore list.

Parameters

address list of ignored addresses.

Definition at line 211 of file blockingUDP.h.

                                                               {
         _ignoredAddresses = addresses;
     }

void epics::pvAccess::BlockingUDPTransport::setLocalMulticastAddress ( const osiSockAddr & sendTo )

inline

Definition at line 160 of file blockingUDP.h.

                                                              {
         _localMulticastAddressEnabled = true;
         _localMulticastAddress = sendTo;
     }

void epics::pvAccess::BlockingUDPTransport::setMutlicastNIF	(	const osiSockAddr &	nifAddr,
		bool	loopback
	)

Definition at line 547 of file blockingUDPTransport.cpp.

 {
     // set the multicast outgoing interface
     int status = ::setsockopt(_channel, IPPROTO_IP, IP_MULTICAST_IF,
                               (char*)&nifAddr.ia.sin_addr, sizeof(struct in_addr));
     if (status)
     {
         char errStr[64];
         epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
         throw std::runtime_error(
             string("Failed to set multicast network interface '") +
             inetAddressToString(nifAddr, false) + "': " + errStr);
     }
 
     // send multicast traffic to myself too
     unsigned char mcast_loop = (loopback ? 1 : 0);
     status = ::setsockopt(_channel, IPPROTO_IP, IP_MULTICAST_LOOP,
                           (char*)&mcast_loop, sizeof(unsigned char));
     if (status)
     {
         char errStr[64];
         epicsSocketConvertErrnoToString(errStr, sizeof(errStr));
         throw std::runtime_error(
             string("Failed to enable multicast loopback on network interface '") +
             inetAddressToString(nifAddr, false) + "': " + errStr);
     }
 
 }

virtual void epics::pvAccess::BlockingUDPTransport::setRecipient ( const osiSockAddr & sendTo )

inlinevirtual

Definition at line 155 of file blockingUDP.h.

                                                                         {
         _sendToEnabled = true;
         _sendTo = sendTo;
     }

virtual void epics::pvAccess::BlockingUDPTransport::setRemoteTransportReceiveBufferSize ( std::size_t receiveBufferSize )

inlinevirtual

Set remote transport receive buffer size.

Parameters

receiveBufferSize receive buffer size.

Implements epics::pvAccess::Transport.

Definition at line 90 of file blockingUDP.h.

                                     {
         // noop for UDP (limited by 64k; MAX_UDP_SEND for PVA)
     }

virtual void epics::pvAccess::BlockingUDPTransport::setRemoteTransportSocketReceiveBufferSize ( std::size_t socketReceiveBufferSize )

inlinevirtual

Set remote transport socket receive buffer size.

Parameters

socketReceiveBufferSize remote socket receive buffer size.

Implements epics::pvAccess::Transport.

Definition at line 95 of file blockingUDP.h.

                                     {
         // noop for UDP (limited by 64k; MAX_UDP_SEND for PVA)
     }

void epics::pvAccess::BlockingUDPTransport::setSendAddresses	(	InetAddrVector &	addresses,
		std::vector< bool > &	address_types
	)

inline

Definition at line 270 of file blockingUDP.h.

                                                                                      {
         _sendAddresses.swap(addresses);
         _isSendAddressUnicast.swap(address_types);
     }

void epics::pvAccess::BlockingUDPTransport::setTappedNIF ( const InetAddrVector & addresses )

inline

Set tapped NIF list.

Parameters

NIF	address list to tap.

Definition at line 227 of file blockingUDP.h.

                                                        {
         _tappedNIF = addresses;
     }

void epics::pvAccess::BlockingUDPTransport::start ( )

Definition at line 101 of file blockingUDPTransport.cpp.

                                  {
 
     string threadName = "UDP-rx " + inetAddressToString(_bindAddress);
 
     if (IS_LOGGABLE(logLevelTrace))
     {
         LOG(logLevelTrace, "Starting thread: %s.", threadName.c_str());
     }
 
     _thread.reset(new epicsThread(*this, threadName.c_str(),
                                   epicsThreadGetStackSize(epicsThreadStackBig),
                                   epicsThreadPriorityMedium));
     _thread->start();
 }

void epics::pvAccess::BlockingUDPTransport::startMessage	(	epics::pvData::int8	command,
		std::size_t	ensureCapacity,
		epics::pvData::int32	payloadSize = `0`
	)

virtual