From cc89f207d7d71b329e5b65cac6e0532e7a15b776 Mon Sep 17 00:00:00 2001
From: Ruud Overeem <overeem@astron.nl>
Date: Wed, 9 Jan 2013 09:17:09 +0000
Subject: [PATCH] Task #1418: Added UDP port code from Apertif.
---
MAC/GCF/TM/include/GCF/TM/GCF_TCPPort.h | 20 ++--
MAC/GCF/TM/src/Port/GCF_RawPort.cc | 5 +-
MAC/GCF/TM/src/PortImpl/GCF_ETHRawPort.cc | 57 +++------
MAC/GCF/TM/src/PortImpl/GCF_TCPPort.cc | 103 ++++++++++++----
MAC/GCF/TM/src/PortImpl/GTM_ETHSocket.cc | 78 ++++--------
MAC/GCF/TM/src/PortImpl/GTM_File.cc | 9 +-
MAC/GCF/TM/src/PortImpl/GTM_FileHandler.cc | 14 +--
.../TM/src/PortImpl/GTM_TCPServerSocket.cc | 72 ++++++-----
MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.h | 3 +-
MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.cc | 113 ++++++++++++------
MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.h | 13 +-
MAC/GCF/TM/src/ServiceBroker/GTM_SBTCPPort.cc | 1 +
.../TM/src/ServiceBroker/ServiceBrokerTask.cc | 85 +++++++++----
13 files changed, 340 insertions(+), 233 deletions(-)
diff --git a/MAC/GCF/TM/include/GCF/TM/GCF_TCPPort.h b/MAC/GCF/TM/include/GCF/TM/GCF_TCPPort.h
index 2440ac1af01..36c3fe5f19a 100644
--- a/MAC/GCF/TM/include/GCF/TM/GCF_TCPPort.h
+++ b/MAC/GCF/TM/include/GCF/TM/GCF_TCPPort.h
@@ -52,10 +52,11 @@ class GCFTCPPort : public GCFRawPort
public:// consturctors && destructors
/// params see constructor of GCFPortInterface
GCFTCPPort (GCFTask& task,
- const string& name,
- TPortType type,
- int protocol,
- bool transportRawData = false);
+ const string& name,
+ TPortType type,
+ int protocol,
+ bool transportRawData = false,
+ bool useUDP = false);
/** default constructor
* GCFPortInterface params are:
@@ -75,7 +76,7 @@ public:// consturctors && destructors
const string& name,
TPortType type,
int protocol,
- bool transportRawData = false);
+ bool transportRawData = false);
// open/close methods
virtual bool open ();
@@ -83,8 +84,8 @@ public:// consturctors && destructors
// send/recv functions
virtual ssize_t send (GCFEvent& event);
- virtual ssize_t recv (void* buf,
- size_t count);
+ virtual ssize_t send (void* buf, size_t count); // for RAW ports
+ virtual ssize_t recv (void* buf, size_t count);
// Special auto-open mode that does the retries itself
// nrRetries : -1 = infinite ; How often to retry the open when it fails.
@@ -106,6 +107,9 @@ public:// consturctors && destructors
string getHostName();
unsigned int getPortNumber();
+ // support of UDP
+ void useUDP(bool useIt) { itsUseUDP = useIt; }
+
private:
/// copying is not allowed.
GCFTCPPort (const GCFTCPPort&);
@@ -120,6 +124,7 @@ private:
void _handleConnect();
void _handleDisconnect();
+ GCFEvent::TResult _recvUDPevent();
// ----- Data Members -----
protected:
@@ -130,6 +135,7 @@ private:
TPeerAddr _addr;
string _host;
unsigned int _portNumber;
+ bool itsUseUDP;
bool itsFixedPortNr;
bool itsAutoOpen;
unsigned int itsAutoOpenTimer;
diff --git a/MAC/GCF/TM/src/Port/GCF_RawPort.cc b/MAC/GCF/TM/src/Port/GCF_RawPort.cc
index b9676238e9c..14ecd90bfa4 100644
--- a/MAC/GCF/TM/src/Port/GCF_RawPort.cc
+++ b/MAC/GCF/TM/src/Port/GCF_RawPort.cc
@@ -305,11 +305,8 @@ GCFEvent::TResult GCFRawPort::recvEvent()
// Note:
string GCFRawPort::getRealName() const
{
-// return (isSlave() ? _pMaster->getName() : _name);
-
GCFPort* thePort = (isSlave() ? _pMaster : (GCFPort*)(this));
- return (thePort->usesModernServiceName() ?
- thePort->makeServiceName() : thePort->getName());
+ return (thePort->usesModernServiceName() ? thePort->makeServiceName() : thePort->getName());
}
} // namespace TM
diff --git a/MAC/GCF/TM/src/PortImpl/GCF_ETHRawPort.cc b/MAC/GCF/TM/src/PortImpl/GCF_ETHRawPort.cc
index a5fcfcf88ec..458507c68c9 100644
--- a/MAC/GCF/TM/src/PortImpl/GCF_ETHRawPort.cc
+++ b/MAC/GCF/TM/src/PortImpl/GCF_ETHRawPort.cc
@@ -74,98 +74,80 @@ bool GCFETHRawPort::close()
bool GCFETHRawPort::open()
{
- if (isConnected())
- {
+ if (isConnected()) {
LOG_WARN("already connected");
return false;
}
- else if (MSPP == getType())
- {
+ else if (MSPP == getType()) {
LOG_ERROR(formatString (
"ETH raw ports can not act as a MSPP (%s).",
getRealName().c_str()));
return false;
}
// check for ifname
- if (_ifname == "")
- {
- try
- {
+ if (_ifname == "") {
+ try {
string ifNameParam = formatString(
PARAM_ETH_IFNAME,
getTask()->getName().c_str(),
getRealName().c_str());
_ifname += globalParameterSet()->getString(ifNameParam);
}
- catch (...)
- {
+ catch (...) {
LOG_ERROR("no interface name specified");
return false;
}
}
- if (_destMacStr == "")
- {
- try
- {
+ if (_destMacStr == "") {
+ try {
string destMacParam = formatString(
PARAM_ETH_MACADDR,
getTask()->getName().c_str(),
getRealName().c_str());
_destMacStr += globalParameterSet()->getString(destMacParam);
}
- catch (...)
- {
+ catch (...) {
LOG_ERROR("no destination mac adres is specified");
return false;
}
}
- if (_ethertype == 0x0000)
- {
- try
- {
+ if (_ethertype == 0x0000) {
+ try {
string ethertypeParam = formatString(
PARAM_ETH_ETHERTYPE,
getTask()->getName().c_str(),
getRealName().c_str());
_ethertype += globalParameterSet()->getInt32(ethertypeParam);
}
- catch (...)
- {
+ catch (...) {
// is optional so no problem.
}
}
- if (!_pSocket)
- {
- if (isSlave())
- {
+ if (!_pSocket) {
+ if (isSlave()) {
LOG_ERROR(formatString (
"Port %s is not initialised.",
getRealName().c_str()));
return false;
}
- else
- {
+ else {
_pSocket = new GTMETHSocket(*this);
}
}
- if (_pSocket->open(_ifname.c_str(), _destMacStr.c_str(), _ethertype) < 0)
- {
- if (SAP == getType())
- {
+ if (_pSocket->open(_ifname.c_str(), _destMacStr.c_str(), _ethertype) < 0) {
+ if (SAP == getType()) {
setState(S_DISCONNECTING);
schedule_disconnected();
}
- else
- {
+ else {
return false;
}
}
- else
- {
+ else {
setState(S_CONNECTING);
schedule_connected();
}
@@ -178,8 +160,7 @@ ssize_t GCFETHRawPort::send(GCFEvent& e)
ASSERT(_pSocket);
- if (!isConnected())
- {
+ if (!isConnected()) {
LOG_ERROR(formatString (
"Port '%s' on task '%s' not connected! Event not sent!",
getRealName().c_str(),
diff --git a/MAC/GCF/TM/src/PortImpl/GCF_TCPPort.cc b/MAC/GCF/TM/src/PortImpl/GCF_TCPPort.cc
index d2cbc5f33dd..17123057310 100644
--- a/MAC/GCF/TM/src/PortImpl/GCF_TCPPort.cc
+++ b/MAC/GCF/TM/src/PortImpl/GCF_TCPPort.cc
@@ -35,6 +35,8 @@
#include "GTM_TCPServerSocket.h"
#include <errno.h>
+const uint UDP_BUFFER_SIZE = 1600;
+
namespace LOFAR {
namespace GCF {
using namespace SB;
@@ -48,13 +50,15 @@ GCFTCPPort::GCFTCPPort(GCFTask& task,
const string& name,
TPortType type,
int protocol,
- bool transportRawData)
+ bool transportRawData,
+ bool useUDP)
: GCFRawPort(task, name, type, protocol, transportRawData),
_pSocket (0),
_addrIsSet (false),
_addr (),
_host (myHostname(false)),
_portNumber (0),
+ itsUseUDP (useUDP),
itsFixedPortNr (false),
itsAutoOpen (false),
itsAutoOpenTimer (0),
@@ -65,10 +69,10 @@ GCFTCPPort::GCFTCPPort(GCFTask& task,
_broker (0)
{
if (SPP == getType() || MSPP == getType()) {
- _pSocket = new GTMTCPServerSocket(*this, (MSPP == type));
+ _pSocket = new GTMTCPServerSocket(*this, (MSPP == type), itsUseUDP);
}
else if (SAP == getType()) {
- _pSocket = new GTMTCPSocket(*this);
+ _pSocket = new GTMTCPSocket(*this, itsUseUDP);
}
}
@@ -82,6 +86,7 @@ GCFTCPPort::GCFTCPPort()
_addr (),
_host (myHostname(false)),
_portNumber (0),
+ itsUseUDP (false),
itsFixedPortNr (false),
itsAutoOpen (false),
itsAutoOpenTimer (0),
@@ -124,6 +129,7 @@ void GCFTCPPort::init(GCFTask& task,
_portNumber = 0;
_host = myHostname(false);
_addrIsSet = false;
+ itsUseUDP = false;
itsFixedPortNr = false;
if (_pSocket) {
delete _pSocket;
@@ -152,17 +158,20 @@ bool GCFTCPPort::open()
// allocate a TCP socket when not done before.
if (!_pSocket) {
+ LOG_DEBUG_STR("No socket yet during 'open', creating it");
if (isSlave()) {
LOG_ERROR(formatString ("Port %s not initialised.", makeServiceName().c_str()));
return (false);
}
if ((getType() == SPP) || (getType() == MSPP)) {
- _pSocket = new GTMTCPServerSocket(*this, (MSPP == getType()));
+ _pSocket = new GTMTCPServerSocket(*this, (MSPP == getType()), itsUseUDP);
+ ASSERTSTR(_pSocket, "Could not create GTMTCPServerSocket for port " << getName());
}
else {
ASSERTSTR (SAP == getType(), "Unknown TPCsocket type " << getType());
- _pSocket = new GTMTCPSocket(*this);
+ _pSocket = new GTMTCPSocket(*this, itsUseUDP);
+ ASSERTSTR(_pSocket, "Could not create GTMTCPSocket for port " << getName());
_pSocket->setBlocking(false);
}
}
@@ -328,6 +337,9 @@ void GCFTCPPort::_handleConnect()
//
GCFEvent::TResult GCFTCPPort::dispatch(GCFEvent& event)
{
+ LOG_TRACE_FLOW("GCFTCPPort::dispatch");
+
+ // autoOpen timer event?
if (event.signal == F_TIMER) {
GCFTimerEvent *TEptr = static_cast<GCFTimerEvent*>(&event);
if (TEptr->arg == &itsAutoOpenTimer) { // Max auto open time reached?
@@ -351,9 +363,46 @@ GCFEvent::TResult GCFTCPPort::dispatch(GCFEvent& event)
}
}
+ // UDP specialty?
+ if (itsUseUDP && event.signal == F_DATAIN && !isTransportRawData()) {
+ return (_recvUDPevent());
+ }
+
+ // fallback to default dispatcher.
return(GCFRawPort::dispatch(event)); // call dispatch from parent (RawPort).
}
+GCFEvent::TResult GCFTCPPort::_recvUDPevent()
+{
+ LOG_TRACE_FLOW("GCFTCPPort::recvUDPevent()");
+
+ char event_buf[UDP_BUFFER_SIZE];
+ GCFEvent* newEvent = new GCFEvent;
+ ASSERTSTR(newEvent, "Could not allocate memory for a GCFEvent class");
+
+ // read the bytes from the socket (must be done in one time)
+ size_t nrBytes = _pSocket->recv(event_buf, UDP_BUFFER_SIZE);
+ ASSERTSTR(nrBytes >= GCFEvent::sizeSignal + GCFEvent::sizeLength, "UPD packet too small: " << nrBytes);
+
+ // copy some fields of the Event structure
+ memcpy(&newEvent->signal, event_buf, GCFEvent::sizeSignal);
+ memcpy(&newEvent->length, event_buf + GCFEvent::sizeSignal, GCFEvent::sizeLength);
+
+ // check if payload matches nr of bytes received.
+ ASSERTSTR(newEvent->length == nrBytes - GCFEvent::sizeSignal - GCFEvent::sizeLength,
+ "Length (" << newEvent->length << ") doesn't match number is received bytes(" <<
+ nrBytes - GCFEvent::sizeSignal - GCFEvent::sizeLength <<")");
+
+ // dispatch the event to the task
+ newEvent->_buffer = event_buf; // attach buffer to event
+ if (_pTask->doEvent(*newEvent, *this) == GCFEvent::NEXT_STATE) {
+ LOG_TRACE_STAT_STR("GCFITCPort::dispatch:Task returned NEXT_STATE, queing " << eventName(*newEvent));
+ _pTask->queueTaskEvent(*newEvent, *this);
+ }
+ newEvent->_buffer = 0;
+ delete newEvent;
+ return (GCFEvent::HANDLED);
+}
//
// serviceRegistered(resultToReturn, portNr)
@@ -443,38 +492,38 @@ void GCFTCPPort::serviceGone()
//
ssize_t GCFTCPPort::send(GCFEvent& e)
{
- ssize_t written = 0;
-
ASSERT(_pSocket);
if (!isConnected()) {
- LOG_ERROR(formatString (
- "Port '%s' on task '%s' not connected! Event not sent!",
+ LOG_ERROR(formatString ("Port '%s' on task '%s' not connected! Event not sent!",
getRealName().c_str(), getTask()->getName().c_str()));
return 0;
}
- if (MSPP == getType()) {
+ if (getType() == MSPP && !itsUseUDP) {
return 0; // no messages can be send by this type of port
}
-#if 0
- unsigned int packSize;
- void* buf = e.pack(packSize);
-#else
e.pack();
char* buf = e.packedBuffer();
uint packSize = e.bufferSize();
-#endif
- LOG_TRACE_STAT(formatString (
- "Sending event '%s' for task '%s' on port '%s'",
- eventName(e).c_str(),
- getTask()->getName().c_str(),
- getRealName().c_str()));
+ LOG_TRACE_STAT(formatString ("Sending event '%s' for task '%s' on port '%s'",
+ eventName(e).c_str(), getTask()->getName().c_str(), getRealName().c_str()));
- if ((written = _pSocket->send(buf, packSize)) != (ssize_t) packSize) {
- LOG_ERROR_STR("Could only send " << written << " of " << packSize << " bytes");
+ return (send(buf, packSize));
+}
+
+//
+// send(buf, size) for RAW ports
+//
+ssize_t GCFTCPPort::send(void* buf, size_t count)
+{
+ ASSERT(_pSocket);
+
+ ssize_t written;
+ if ((written = _pSocket->send(buf, count)) != (ssize_t) count) {
+ LOG_ERROR_STR("Could only send " << written << " of " << count << " bytes");
setState(S_DISCONNECTING);
LOG_TRACE_COND("write: state = DISCONNECTING");
_handleDisconnect();
@@ -494,9 +543,8 @@ ssize_t GCFTCPPort::recv(void* buf, size_t count)
ASSERT(_pSocket);
if (!isConnected()) {
- LOG_ERROR(formatString (
- "Port '%s' on task '%s' not connected! Can't read device!",
- getRealName().c_str(), getTask()->getName().c_str()));
+ LOG_ERROR(formatString ("Port '%s' on task '%s' not connected! Can't read device!",
+ getRealName().c_str(), getTask()->getName().c_str()));
return 0;
}
@@ -555,6 +603,7 @@ void GCFTCPPort::setAddr(const TPeerAddr& addr)
_addrIsSet = ((_addr.taskname != "") && (_addr.portname != ""));
_deviceNameMask = formatString("%s:%s", _addr.taskname.c_str(),
_addr.portname.c_str());
+ LOG_DEBUG_STR("AddrIsSet:" << (_addrIsSet?"Y":"N"));
}
//
@@ -569,6 +618,10 @@ bool GCFTCPPort::accept(GCFTCPPort& port)
return (false);
}
+ if (itsUseUDP) {
+ return (false);
+ }
+
GTMTCPServerSocket* pProvider = (GTMTCPServerSocket*)_pSocket;
if (port._pSocket == 0) {
port._pSocket = new GTMTCPSocket(port);
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_ETHSocket.cc b/MAC/GCF/TM/src/PortImpl/GTM_ETHSocket.cc
index ffdddddba50..6f8942c7129 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_ETHSocket.cc
+++ b/MAC/GCF/TM/src/PortImpl/GTM_ETHSocket.cc
@@ -51,12 +51,9 @@
#include <linux/filter.h>
#endif
-namespace LOFAR
-{
- namespace GCF
- {
- namespace TM
- {
+namespace LOFAR {
+ namespace GCF {
+ namespace TM {
GTMETHSocket::GTMETHSocket(GCFETHRawPort& port) :
GTMFile(port)
@@ -152,37 +149,25 @@ int GTMETHSocket::open(const char* ifname,
// open the raw socket
socketFD = socket(PF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
- if (socketFD < 0)
- {
- LOG_ERROR(LOFAR::formatString (
- "open(PF_PACKET): %s",
- strerror(errno)));
+ if (socketFD < 0) {
+ LOG_ERROR(LOFAR::formatString ( "open(PF_PACKET): %s", strerror(errno)));
return socketFD;
}
// make large send/recv buffers
int val = 262144;
- if (::setsockopt(socketFD, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val)) < 0)
- {
- LOG_WARN(LOFAR::formatString (
- "setsockopt(SO_RCVBUF): %s",
- strerror(errno)));
+ if (::setsockopt(socketFD, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val)) < 0) {
+ LOG_WARN(LOFAR::formatString ( "setsockopt(SO_RCVBUF): %s", strerror(errno)));
}
- if (::setsockopt(socketFD, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val)) < 0)
- {
- LOG_WARN(LOFAR::formatString (
- "setsockopt(SO_SNDBUF): %s",
- strerror(errno)));
+ if (::setsockopt(socketFD, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val)) < 0) {
+ LOG_WARN(LOFAR::formatString ( "setsockopt(SO_SNDBUF): %s", strerror(errno)));
}
// find MAC address for specified interface
struct ifreq ifr;
strncpy(ifr.ifr_name, ifname, IFNAMSIZ-1);
- if (::ioctl(socketFD, SIOCGIFHWADDR, &ifr) < 0)
- {
- LOG_FATAL(LOFAR::formatString (
- "ioctl(SIOCGIFHWADDR): %s",
- strerror(errno)));
+ if (::ioctl(socketFD, SIOCGIFHWADDR, &ifr) < 0) {
+ LOG_FATAL(LOFAR::formatString ( "ioctl(SIOCGIFHWADDR): %s", strerror(errno)));
::close(socketFD);
return -1;
}
@@ -191,17 +176,13 @@ int GTMETHSocket::open(const char* ifname,
unsigned int hx;
char macPart[3];
// print MAC addres for source
- for (int i = 0; i < ETH_ALEN; i++)
- {
+ for (int i = 0; i < ETH_ALEN; i++) {
hx = ifr.ifr_hwaddr.sa_data[i] & 0xff;
snprintf(macPart, sizeof macPart, "%02x", hx);
macAddress += macPart;
if (i < ETH_ALEN - 1) macAddress += ':';
}
- LOG_DEBUG(LOFAR::formatString (
- "SRC (%s) HWADDR: %s",
- ifname,
- macAddress.c_str()));
+ LOG_DEBUG(LOFAR::formatString ( "SRC (%s) HWADDR: %s", ifname, macAddress.c_str()));
// convert HWADDR string to sll_addr
convertCcp2sllAddr(destMacStr, destMac);
@@ -212,10 +193,7 @@ int GTMETHSocket::open(const char* ifname,
memcpy((char*)(&mac_filter_insn[3].k) + 2, destMac, sizeof(__u16));
mac_filter_insn[3].k = htonl(mac_filter_insn[3].k);
filter.filter = mac_filter_insn;
- if (::setsockopt(socketFD,
- SOL_SOCKET, SO_ATTACH_FILTER,
- &filter, sizeof(struct sock_fprog)) < 0)
- {
+ if (::setsockopt(socketFD, SOL_SOCKET, SO_ATTACH_FILTER, &filter, sizeof(struct sock_fprog)) < 0) {
LOG_ERROR("setsockopt(SO_ATTACH_FILTER) failed");
::close(socketFD);
return -1;
@@ -223,16 +201,13 @@ int GTMETHSocket::open(const char* ifname,
// print MAC address for destination
macAddress = "";
- for (int i = 0; i < ETH_ALEN; i++)
- {
+ for (int i = 0; i < ETH_ALEN; i++) {
hx = destMac[i] & 0xff;
snprintf(macPart, sizeof macPart, "%02x", hx);
macAddress += macPart;
if (i < ETH_ALEN - 1) macAddress += ':';
}
- LOG_DEBUG(LOFAR::formatString (
- "DEST HWADDR: %s",
- macAddress.c_str()));
+ LOG_DEBUG(LOFAR::formatString ( "DEST HWADDR: %s", macAddress.c_str()));
// fill in packet header for sending messages
struct ethhdr* hdr = (struct ethhdr*)_sendPacket;
@@ -242,10 +217,8 @@ int GTMETHSocket::open(const char* ifname,
// get interface index number
strncpy(ifr.ifr_name, ifname, IFNAMSIZ-1);
- if (ioctl(socketFD, SIOCGIFINDEX, &ifr) < 0)
- {
- LOG_FATAL(LOFAR::formatString (
- "ioctl(SIOCGIFINDEX)"));
+ if (ioctl(socketFD, SIOCGIFINDEX, &ifr) < 0) {
+ LOG_FATAL(LOFAR::formatString ( "ioctl(SIOCGIFINDEX)"));
::close(socketFD);
return -1;
}
@@ -259,12 +232,8 @@ int GTMETHSocket::open(const char* ifname,
_sockaddr.sll_protocol = htons(ETH_P_ALL);
_sockaddr.sll_ifindex = ifindex;
_sockaddr.sll_hatype = ARPHRD_ETHER;
- if (::bind(socketFD, (struct sockaddr*)&_sockaddr,
- sizeof(struct sockaddr_ll)) < 0)
- {
- LOG_FATAL(LOFAR::formatString (
- "GCFETHRawPort::open; bind : %s",
- strerror(errno)));
+ if (::bind(socketFD, (struct sockaddr*)&_sockaddr, sizeof(struct sockaddr_ll)) < 0) {
+ LOG_FATAL(LOFAR::formatString ( "GCFETHRawPort::open; bind : %s", strerror(errno)));
::close(socketFD);
return -1;
}
@@ -276,8 +245,8 @@ int GTMETHSocket::open(const char* ifname,
so.mr_alen = 0;
memset(&so.mr_address, 0, sizeof(so.mr_address));
if (::setsockopt(socketFD, SOL_PACKET, PACKET_ADD_MEMBERSHIP,
- (void*)&so, sizeof(struct packet_mreq)) < 0)
- {
+ (void*)&so, sizeof(struct packet_mreq)) < 0) {
+ // TODO
}
//
@@ -313,8 +282,7 @@ void GTMETHSocket::convertCcp2sllAddr(const char* destMacStr,
sscanf(destMacStr, "%x:%x:%x:%x:%x:%x",
&hx[0], &hx[1], &hx[2], &hx[3], &hx[4], &hx[5]);
- for (int i = 0; i < ETH_ALEN; i++)
- {
+ for (int i = 0; i < ETH_ALEN; i++) {
destMac[i] = (char)hx[i];
}
}
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_File.cc b/MAC/GCF/TM/src/PortImpl/GTM_File.cc
index 5e49950b516..a4366947118 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_File.cc
+++ b/MAC/GCF/TM/src/PortImpl/GTM_File.cc
@@ -53,10 +53,8 @@ GTMFile::GTMFile(GCFRawPort& port) :
GTMFile::~GTMFile()
{
close();
-
GTMFileHandler::release();
_pHandler = 0;
-
itsScheduler = 0;
}
@@ -65,20 +63,17 @@ bool GTMFile::close()
bool result(true);
if (_fd > -1) {
+ ASSERT(_pHandler);
_pHandler->deregisterFile(*this);
-
result = (::close(_fd) == 0);
-
if (!result) {
LOG_ERROR(formatString ( "::close, error: %s", strerror(errno)));
-
// there is nothing we can do at this point, since we cannot know
// whether the fd is still valid.
}
_fd = -1;
}
-
return result;
}
@@ -129,7 +124,7 @@ void GTMFile::setBlocking(bool blocking) const
int flags = fcntl(_fd, F_GETFL);
- if (blocking) {
+ if (blocking) {
fcntl(_fd, F_SETFL, flags | O_NONBLOCK);
} else {
fcntl(_fd, F_SETFL, flags & ~O_NONBLOCK);
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_FileHandler.cc b/MAC/GCF/TM/src/PortImpl/GTM_FileHandler.cc
index 5f4ffcb9cd2..a4dce1f9eb8 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_FileHandler.cc
+++ b/MAC/GCF/TM/src/PortImpl/GTM_FileHandler.cc
@@ -92,21 +92,21 @@ void GTMFileHandler::workProc()
fd_set testFDs = _readFDs;
testFiles = _files;
- // map keys are sorted low-to-high, so last key is highest
- int maxfd = testFiles.empty() ? 0 : testFiles.rbegin()->first;
+ // map keys are sorted low-to-high, so last key is highest
+ int maxfd = testFiles.empty() ? 0 : testFiles.rbegin()->first;
- // wait for any file to be readable, or for our timeout
+ // wait for any file to be readable, or for our timeout
result = ::select(maxfd + 1, &testFDs, (fd_set *) 0, (fd_set *) 0, &select_timeout);
if (result >= 0) {
- for (TFiles::iterator i = testFiles.begin(); i != testFiles.end() && _running; ++i) {
- int fd = i->first;
- GTMFile *file = i->second;
+ for (TFiles::iterator i = testFiles.begin(); i != testFiles.end() && _running; ++i) {
+ int fd = i->first;
+ GTMFile *file = i->second;
if (FD_ISSET(fd, &testFDs)) {
file->doWork();
}
- }
+ }
}
}
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.cc b/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.cc
index 29d6d8ae0dd..ecec7ec8eea 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.cc
+++ b/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.cc
@@ -39,8 +39,8 @@ namespace LOFAR {
namespace GCF {
namespace TM {
-GTMTCPServerSocket::GTMTCPServerSocket(GCFTCPPort& port, bool isProvider) :
- GTMTCPSocket(port),
+GTMTCPServerSocket::GTMTCPServerSocket(GCFTCPPort& port, bool isProvider, bool useUDP) :
+ GTMTCPSocket(port, useUDP),
_isProvider(isProvider),
_pDataSocket(0)
{
@@ -53,6 +53,8 @@ GTMTCPServerSocket::~GTMTCPServerSocket()
bool GTMTCPServerSocket::open(unsigned int portNumber)
{
+ LOG_TRACE_COND_STR("open:fd=" << _fd << ", port=" << _port.getName());
+
if (_fd != -1) {
if (_pDataSocket != 0) {
_pDataSocket->close();
@@ -61,7 +63,7 @@ bool GTMTCPServerSocket::open(unsigned int portNumber)
return (false);
}
- int socketFD = ::socket(AF_INET, SOCK_STREAM, 0);
+ int socketFD = ::socket(AF_INET, itsUseUDP ? SOCK_DGRAM : SOCK_STREAM, 0);
if (socketFD == -1) {
LOG_WARN(formatString ( "::socket, error: %s", strerror(errno)));
return (false);
@@ -69,7 +71,7 @@ bool GTMTCPServerSocket::open(unsigned int portNumber)
unsigned int val = 1;
struct linger lin = { 1,1 };
- if (::setsockopt(socketFD, SOL_SOCKET, SO_REUSEADDR, (const void*)&val, sizeof val) < 0) {
+ if (::setsockopt(socketFD, SOL_SOCKET, SO_REUSEADDR, (const void*)&val, sizeof(val)) < 0) {
LOG_WARN(formatString (
"Error on setting socket options SO_REUSEADDR: %s",
strerror(errno)));
@@ -77,7 +79,7 @@ bool GTMTCPServerSocket::open(unsigned int portNumber)
return false;
}
- if (::setsockopt(socketFD, SOL_SOCKET, SO_LINGER, (const void*)&lin, sizeof lin) < 0) {
+ if (::setsockopt(socketFD, SOL_SOCKET, SO_LINGER, (const void*)&lin, sizeof(lin)) < 0) {
LOG_WARN(formatString (
"Error on setting socket options SO_LINGER: %s",
strerror(errno)));
@@ -95,10 +97,12 @@ bool GTMTCPServerSocket::open(unsigned int portNumber)
return (false);
}
- if (::listen(socketFD, 5) == -1) {
- LOG_WARN(formatString ( "::listen, error: %s", strerror(errno)));
- ::close(socketFD);
- return (false);
+ if (!itsUseUDP) {
+ if (::listen(socketFD, 5) == -1) {
+ LOG_WARN(formatString ( "::listen, error: %s", strerror(errno)));
+ ::close(socketFD);
+ return (false);
+ }
}
// set non-blocking
@@ -127,6 +131,12 @@ bool GTMTCPServerSocket::open(unsigned int portNumber)
void GTMTCPServerSocket::doWork()
{
LOG_TRACE_FLOW("GTMTCPServerSocket::doWork()");
+ if (itsUseUDP) {
+ LOG_TRACE_FLOW("doWork: socket is UDP, assuming data-in event");
+ GCFEvent dataInEvent(F_DATAIN);
+ forwardEvent(dataInEvent);
+ return;
+ }
if (_isProvider) {
GCFEvent acceptReqEvent(F_ACCEPT_REQ);
@@ -139,7 +149,7 @@ void GTMTCPServerSocket::doWork()
_pDataSocket = new GTMTCPSocket(pPort);
ASSERT (_pDataSocket->getFD() < 0);
- if (_accept(*_pDataSocket)) {
+ if (_accept(*_pDataSocket)) {
GCFEvent connectedEvent(F_CONNECTED);
forwardEvent(connectedEvent);
}
@@ -157,11 +167,11 @@ void GTMTCPServerSocket::doWork()
//
bool GTMTCPServerSocket::accept(GTMFile& newSocket)
{
- if (_isProvider && _pDataSocket == 0) {
- return _accept(newSocket);
- } else {
- return false;
- }
+ if (!itsUseUDP && _isProvider && _pDataSocket == 0) {
+ return (_accept(newSocket));
+ }
+
+ return (false);
}
//
@@ -173,20 +183,20 @@ bool GTMTCPServerSocket::_accept(GTMFile& newSocket)
{
bool result;
- struct sockaddr_in clientAddress;
- socklen_t clAddrLen = sizeof clientAddress;
- int newSocketFD;
+ struct sockaddr_in clientAddress;
+ socklen_t clAddrLen = sizeof clientAddress;
+ int newSocketFD;
- /* loop to handle transient errors */
- int retries = MAX_ACCEPT_RETRIES;
- while ((newSocketFD = ::accept(_fd, (struct sockaddr*) &clientAddress, &clAddrLen)) < 0
- && (EINTR == errno || EWOULDBLOCK == errno || EAGAIN == errno) && --retries > 0)
- /*noop*/;
+ /* loop to handle transient errors */
+ int retries = MAX_ACCEPT_RETRIES;
+ while ((newSocketFD = ::accept(_fd, (struct sockaddr*) &clientAddress, &clAddrLen)) < 0
+ && (EINTR == errno || EWOULDBLOCK == errno || EAGAIN == errno) && --retries > 0)
+ /*noop*/;
- result = (newSocket.setFD(newSocketFD) > 0);
- if (!result) {
- LOG_WARN(formatString ("::accept, error: %s", strerror(errno)));
- }
+ result = (newSocket.setFD(newSocketFD) > 0);
+ if (!result) {
+ LOG_WARN(formatString ("::accept(%d), error:(%d)=%s", _fd, errno, strerror(errno)));
+ }
return result;
}
@@ -209,6 +219,10 @@ bool GTMTCPServerSocket::close()
ssize_t GTMTCPServerSocket::send(void* buf, size_t count)
{
+ if (itsUseUDP) {
+ return GTMTCPSocket::send(buf, count);
+ }
+
if (!_isProvider && _pDataSocket != 0) {
return _pDataSocket->send(buf, count);
}
@@ -217,6 +231,10 @@ ssize_t GTMTCPServerSocket::send(void* buf, size_t count)
ssize_t GTMTCPServerSocket::recv(void* buf, size_t count, bool raw)
{
+ if (itsUseUDP) {
+ return GTMTCPSocket::recv(buf, count, raw);
+ }
+
if (!_isProvider && _pDataSocket != 0) {
return _pDataSocket->recv(buf, count, raw);
}
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.h b/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.h
index d7b3dee0141..a0e053934e4 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.h
+++ b/MAC/GCF/TM/src/PortImpl/GTM_TCPServerSocket.h
@@ -42,7 +42,8 @@ class GTMTCPServerSocket : public GTMTCPSocket
public:
// constructors, destructors and default operators
GTMTCPServerSocket (GCFTCPPort& port,
- bool isProvider = false);
+ bool isProvider = false,
+ bool useUDP = false);
virtual ~GTMTCPServerSocket ();
// GTMTCPServerSocket specific member methods
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.cc b/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.cc
index 7467994e223..4e267e3240c 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.cc
+++ b/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.cc
@@ -33,6 +33,7 @@
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
+#include <sys/ioctl.h>
#include <sys/select.h>
#include <fcntl.h>
#include <netdb.h>
@@ -43,12 +44,38 @@ namespace LOFAR {
namespace GCF {
namespace TM {
-GTMTCPSocket::GTMTCPSocket(GCFTCPPort& port) :
- GTMFile(port),
- _connecting(false)
+GTMTCPSocket::GTMTCPSocket(GCFTCPPort& port, bool useUDP) :
+ GTMFile (port),
+ itsUseUDP (useUDP),
+ itsConnecting(false)
{
}
+void GTMTCPSocket::doWork()
+{
+ LOG_TRACE_FLOW("GTMTCPSocket::doWork()");
+ unsigned long bytesRead = 0;
+
+ if (ioctl(_fd, FIONREAD, &bytesRead) > -1) {
+ if (bytesRead == 0 && !itsUseUDP) { // We need the test on UDP :-(
+ GCFEvent discoEvent(F_DISCONNECTED);
+ itsScheduler->queueEvent(0, discoEvent, &_port);
+ }
+ else {
+ GCFEvent dataEvent(F_DATAIN);
+ itsScheduler->queueEvent(0, dataEvent, &_port);
+ }
+ }
+ else {
+ ASSERT(_port.getTask());
+ LOG_FATAL(LOFAR::formatString ("%s(%s): Error in 'ioctl' on socket fd %d: %s",
+ _port.getTask()->getName().c_str(),
+ _port.getName().c_str(),
+ _fd,
+ strerror(errno)));
+ }
+}
+
//
// send(buf, count)
//
@@ -62,7 +89,13 @@ ssize_t GTMTCPSocket::send(void* buf, size_t count)
ssize_t countLeft(count);
ssize_t written(0);
do {
- written = ::write(_fd, ((char*)buf) + (count - countLeft), countLeft);
+ if (itsUseUDP) {
+ written = sendto(_fd, ((char*)buf) + (count - countLeft), countLeft, 0,
+ (struct sockaddr*) &itsTCPaddr, sizeof(itsTCPaddr));
+ }
+ else {
+ written = ::write(_fd, ((char*)buf) + (count - countLeft), countLeft);
+ }
if (written == 0) { // is it a disconnect?
return (0);
}
@@ -96,7 +129,12 @@ ssize_t GTMTCPSocket::recv(void* buf, size_t count, bool raw)
ssize_t countLeft(count);
ssize_t received(0);
+ socklen_t addrLen = sizeof(itsTCPaddr);
do {
+ if (itsUseUDP) {
+ return (recvfrom(_fd, (char*)buf, count, 0, (struct sockaddr*) &itsTCPaddr, &addrLen));
+ }
+
received = ::read(_fd, ((char*)buf) + (count - countLeft), countLeft);
if (received == 0) { // is it a disconnect?
return(0);
@@ -127,7 +165,7 @@ bool GTMTCPSocket::open(unsigned int /*portNumber*/)
return (true);
}
- _fd = ::socket(AF_INET, SOCK_STREAM, 0);
+ _fd = ::socket(AF_INET, itsUseUDP ? SOCK_DGRAM : SOCK_STREAM, 0);
if (_fd < 0) {
LOG_WARN(formatString ( "::socket, error: %s", strerror(errno)));
close();
@@ -151,22 +189,21 @@ int GTMTCPSocket::connect(unsigned int portNumber, const string& host)
LOG_TRACE_COND_STR(_port.getName() << ":connect(" << portNumber << "," << host << "),fd=" << _fd);
// try to resolve hostaddress/name
- struct sockaddr_in serverAddr;
struct hostent *hostinfo;
hostinfo = gethostbyname(host.c_str());
ASSERTSTR(hostinfo, _port.getName() << ":hostname " << host << " could not be resolved, error = " << errno);
// try to connect
- serverAddr.sin_family = AF_INET;
- serverAddr.sin_addr = *(struct in_addr *) *hostinfo->h_addr_list;
- serverAddr.sin_port = htons(portNumber);
+ itsTCPaddr.sin_family = AF_INET;
+ itsTCPaddr.sin_addr = *(struct in_addr *) *hostinfo->h_addr_list;
+ itsTCPaddr.sin_port = htons(portNumber);
errno = 0;
- if (!_connecting) {
+ if (!itsConnecting) {
// create a new connection
- if ((::connect(_fd, (struct sockaddr *)&serverAddr, sizeof(struct sockaddr_in)) == 0)) {
+ if ((::connect(_fd, (struct sockaddr *)&itsTCPaddr, sizeof(struct sockaddr_in)) == 0)) {
// connect succesfull, register filedescriptor
- setFD(_fd);
+ setFD(_fd);
return (1);
}
@@ -174,11 +211,11 @@ int GTMTCPSocket::connect(unsigned int portNumber, const string& host)
if (errno != EINPROGRESS) {
// serious error
LOG_WARN_STR(_port.getName() << ":connect(" << host << "," << portNumber << "), error: " << strerror(errno));
- close();
+ close();
return (-1);
}
- _connecting = true;
+ itsConnecting = true;
} else {
// poll an existing connection
@@ -192,7 +229,7 @@ int GTMTCPSocket::connect(unsigned int portNumber, const string& host)
switch(::select(_fd + 1, NULL, &fds, NULL, &timeout)) {
case 0:
// no data available
- break;
+ break;
case -1:
// serious error
@@ -201,31 +238,31 @@ int GTMTCPSocket::connect(unsigned int portNumber, const string& host)
return (-1);
default:
- // data available OR connection error
- int so_error;
- socklen_t slen = sizeof so_error;
- if (getsockopt(_fd, SOL_SOCKET, SO_ERROR, &so_error, &slen) < 0) {
- // serious error
- LOG_WARN_STR(_port.getName() << ":getsockopt(" << host << "," << portNumber << "), error: " << strerror(errno));
- close();
- return (-1);
- }
-
- if (so_error == 0) {
- // connect succesfull, register filedescriptor
- setFD(_fd);
- return 1;
- }
-
- // connection failure
- LOG_WARN_STR(_port.getName() << ":connect(" << host << "," << portNumber << "), error: " << strerror(errno));
- close();
- return (-1);
- }
- }
+ // data available OR connection error
+ int so_error;
+ socklen_t slen = sizeof so_error;
+ if (getsockopt(_fd, SOL_SOCKET, SO_ERROR, &so_error, &slen) < 0) {
+ // serious error
+ LOG_WARN_STR(_port.getName() << ":getsockopt(" << host << "," << portNumber << "), error: " << strerror(errno));
+ close();
+ return (-1);
+ }
+
+ if (so_error == 0) {
+ // connect succesfull, register filedescriptor
+ setFD(_fd);
+ return 1;
+ }
+
+ // connection failure
+ LOG_WARN_STR(_port.getName() << ":connect(" << host << "," << portNumber << "), error: " << strerror(errno));
+ close();
+ return (-1);
+ } // switch
+ } // case
LOG_DEBUG_STR(_port.getName() << ": still waiting for connection");
- return(0);
+ return (0);
}
} // namespace TM
diff --git a/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.h b/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.h
index fa354a0a9f2..7980bbc256f 100644
--- a/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.h
+++ b/MAC/GCF/TM/src/PortImpl/GTM_TCPSocket.h
@@ -25,6 +25,7 @@
#include "GTM_File.h"
#include <Common/lofar_string.h>
+#include <netinet/in.h>
namespace LOFAR {
namespace GCF {
@@ -42,7 +43,7 @@ class GTMTCPSocket : public GTMFile
{
public:
// constructors, destructors and default operators
- GTMTCPSocket (GCFTCPPort& port);
+ GTMTCPSocket (GCFTCPPort& port, bool useUDP = false);
// GTMTCPSocket specific member methods
// open/connect methods
@@ -54,13 +55,21 @@ public:
virtual ssize_t send (void* buf, size_t count);
virtual ssize_t recv (void* buf, size_t count, bool raw = false);
+protected:
+ virtual void doWork();
+
+ // --- datamembers ---
+ bool itsUseUDP;
+
private:
GTMTCPSocket ();
/// Don't allow copying of the GTMTCPSocket object.
GTMTCPSocket (const GTMTCPSocket&);
GTMTCPSocket& operator= (const GTMTCPSocket&);
- bool _connecting;
+ // --- datamembers ---
+ struct sockaddr_in itsTCPaddr;
+ bool itsConnecting;
};
diff --git a/MAC/GCF/TM/src/ServiceBroker/GTM_SBTCPPort.cc b/MAC/GCF/TM/src/ServiceBroker/GTM_SBTCPPort.cc
index b9ec143458c..04466c4e7dc 100644
--- a/MAC/GCF/TM/src/ServiceBroker/GTM_SBTCPPort.cc
+++ b/MAC/GCF/TM/src/ServiceBroker/GTM_SBTCPPort.cc
@@ -68,6 +68,7 @@ bool GTMSBTCPPort::open()
if (!_pSocket) {
_pSocket = new GTMTCPSocket(*this);
+ ASSERTSTR(_pSocket, "Could not create GTMTCPSocket for SBtask");
_pSocket->setBlocking(false);
}
diff --git a/MAC/GCF/TM/src/ServiceBroker/ServiceBrokerTask.cc b/MAC/GCF/TM/src/ServiceBroker/ServiceBrokerTask.cc
index 5e46d47170b..8efb555fbbb 100644
--- a/MAC/GCF/TM/src/ServiceBroker/ServiceBrokerTask.cc
+++ b/MAC/GCF/TM/src/ServiceBroker/ServiceBrokerTask.cc
@@ -203,13 +203,16 @@ void ServiceBrokerTask::deletePort(GCFTCPPort& aPort)
// clean up all action that refer to this port
actionList_t tmpActionList; // NOTE: 'erase' reorders the elements of a list!!!
tmpActionList.swap(itsActionList);
-
- for (ALiter iter = tmpActionList.begin(); iter != tmpActionList.end(); ++iter) {
+ ALiter end = tmpActionList.end();
+ ALiter iter = tmpActionList.begin();
+ while (iter != end) {
LOG_TRACE_COND_STR("deletePort checking: " << iter->print());
if (iter->pPort != &aPort) { // copy others only
itsActionList.push_back(*iter);
}
+ iter++;
}
+ tmpActionList.clear();
// unregister service of this port if any
_deleteService(aPort);
@@ -300,12 +303,21 @@ unsigned short ServiceBrokerTask::_registerAction(Action action)
//
void ServiceBrokerTask::_reRegisterServices(GCFPortInterface* brokerPort)
{
- for (SMiter iter = itsServiceMap.begin(); iter != itsServiceMap.end(); ++iter) {
+ // nothing to do?
+ if (itsServiceMap.empty()) {
+ return;
+ }
+
+ SMiter end = itsServiceMap.end();
+ SMiter iter = itsServiceMap.begin();
+ while (iter != end) {
SBReregisterServiceEvent request;
request.seqnr = 0;
request.servicename = iter->second.servicename;
request.portnumber = iter->second.portNr;
brokerPort->send(request);
+
+ iter++;
}
}
@@ -314,18 +326,23 @@ void ServiceBrokerTask::_reRegisterServices(GCFPortInterface* brokerPort)
//
void ServiceBrokerTask::_doActionList(const string& hostname)
{
+ // nothing to do?
+ if (itsActionList.empty()) {
+ return;
+ }
_printActionList();
// Note: while processing the list, the list grows. Therefore we use actionsLeft.
actionList_t tmpActionList; // NOTE: 'erase' reorders the elements of a list!!!
-
tmpActionList.swap(itsActionList);
-
- for (ALiter iter = tmpActionList.begin(); iter != tmpActionList.end(); ++iter) {
+ ALiter end = tmpActionList.end();
+ ALiter iter = tmpActionList.begin();
+ while (iter != end) {
LOG_TRACE_COND_STR("doActionList checking: " << iter->print());
// only process the actions for this host
if (iter->hostname != hostname) {
itsActionList.push_back(*iter); // restore in original list.
+ iter++;
continue;
}
@@ -344,7 +361,9 @@ void ServiceBrokerTask::_doActionList(const string& hostname)
ASSERTSTR(false, "Unknown action in actionlist: " << iter->type
<< ":" << iter->servicename << "@" << iter->hostname);
}
+ iter++;
}
+ tmpActionList.clear();
}
//
@@ -352,15 +371,22 @@ void ServiceBrokerTask::_doActionList(const string& hostname)
//
void ServiceBrokerTask::_lostBroker(const string& hostname)
{
+ // nothing to do?
+ if (itsActionList.empty()) {
+ return;
+ }
+
// Note: while processing the list, the list grows. Therefore we use actionsLeft.
actionList_t tmpActionList; // NOTE: 'erase' reorders the elements of a list!!!
tmpActionList.swap(itsActionList);
-
- for (ALiter iter = tmpActionList.begin(); iter != tmpActionList.end(); ++iter) {
+ ALiter end = tmpActionList.end();
+ ALiter iter = tmpActionList.begin();
+ while (iter != end) {
LOG_TRACE_COND_STR("_lostBroker checking: " << iter->print());
// only process the actions for this host
if (iter->hostname != hostname) {
itsActionList.push_back(*iter); // restore in original list.
+ iter++;
continue;
}
@@ -390,7 +416,9 @@ void ServiceBrokerTask::_lostBroker(const string& hostname)
ASSERTSTR(false, "Unknown action in actionlist: " << iter->type
<< ":" << iter->servicename << "@" << iter->hostname);
}
+ iter++;
}
+ tmpActionList.clear();
}
//
@@ -421,14 +449,17 @@ ServiceBrokerTask::BMiter ServiceBrokerTask::_getBroker(const string& hostname)
//
ServiceBrokerTask::ALiter ServiceBrokerTask::_findAction(uint16 seqnr)
{
- for (ALiter iter = itsActionList.begin(); iter != itsActionList.end(); ++iter) {
+ ALiter end = itsActionList.end();
+ ALiter iter = itsActionList.begin();
+ while (iter != end) {
LOG_TRACE_COND_STR("_findAction checking: " << iter->print());
if (iter->seqnr == seqnr) {
return (iter);
}
- }
- return itsActionList.end();
+ iter++;
+ }
+ return (iter);
}
//
@@ -438,12 +469,13 @@ void ServiceBrokerTask::_reconnectBrokers()
{
LOG_DEBUG("_reconnectBrokers()");
- // keep a next pointer because we're erasing elements
- for (BMiter iter = itsBrokerMap.begin(), iter_next = iter; iter != itsBrokerMap.end(); iter = iter_next) {
- ++iter_next;
+ BMiter end = itsBrokerMap.end();
+ BMiter iter = itsBrokerMap.begin();
+ while (iter != end) {
// ready when broker is connected
if (iter->second.port->isConnected()) {
+ iter++;
continue;
}
@@ -451,19 +483,21 @@ void ServiceBrokerTask::_reconnectBrokers()
if (--(iter->second.nRetries) > 0) {
iter->second.port->open(); // will result in F_CONN or F_DISCONN
_checkActionList(iter->first);
+ iter++;
}
else {
LOG_ERROR_STR("ServiceBroker on host " << iter->first << " is unreachable!");
_lostBroker(iter->first);
-
+ BMiter tmp = iter;
+ iter++;
// remove broker except when its the SB on my host and some services were registered there.
- if (iter->first == myHostname(false) && !itsServiceMap.empty()) {
- iter->second.nRetries = MAX_RECONNECT_RETRIES; // keep trying.
- iter->second.port->open(); // might result in F_CONN or F_DISCONN
+ if (tmp->first == myHostname(false) && !itsServiceMap.empty()) {
+ tmp->second.nRetries = MAX_RECONNECT_RETRIES; // keep trying.
+ tmp->second.port->open(); // might result in F_CONN or F_DISCONN
}
else {
- LOG_DEBUG_STR("Removing servicebroker for " << iter->first << " from brokermap");
- itsBrokerMap.erase(iter);
+ LOG_DEBUG_STR("Removing servicebroker for " << tmp->first << " from brokermap");
+ itsBrokerMap.erase(tmp);
}
}
}
@@ -478,9 +512,11 @@ void ServiceBrokerTask::_printActionList()
return;
}
+ ALiter end = itsActionList.end();
+ ALiter iter = itsActionList.begin();
string typeName;
LOG_TRACE_FLOW_STR("ActionList at " << time(0));
- for (ALiter iter = itsActionList.begin(); iter != itsActionList.end(); ++iter) {
+ while (iter != end) {
switch (iter->type) {
case SB_REGISTER_SERVICE: typeName = "Register"; break;
case SB_UNREGISTER_SERVICE: typeName = "Unregister"; break;
@@ -490,6 +526,7 @@ void ServiceBrokerTask::_printActionList()
LOG_DEBUG_STR(typeName << " " << iter->servicename << "@" << iter->hostname <<
"(" << iter->timestamp << "(" << iter->seqnr << "))");
+ ++iter;
}
}
@@ -503,9 +540,11 @@ void ServiceBrokerTask::_checkActionList(const string& hostname)
actionList_t tmpActionList; // NOTE: 'erase' reorders the elements of a list!!!
tmpActionList.swap(itsActionList);
+ ALiter end = tmpActionList.end();
+ ALiter iter = tmpActionList.begin();
time_t currentTime = time(0);
// check for which actions we are late.
- for (ALiter iter = tmpActionList.begin(); iter != tmpActionList.end(); ++iter) {
+ while (iter != end) {
LOG_TRACE_COND_STR("_checkActionList checking: " << iter->print());
bool actionHandled(false);
if (iter->hostname == hostname) {
@@ -536,7 +575,9 @@ void ServiceBrokerTask::_checkActionList(const string& hostname)
if (!actionHandled) {
itsActionList.push_back(*iter); // keep non-expired action
}
+ iter++;
}
+ tmpActionList.clear();
}
//
--
GitLab