Configuring IceStorm for subscribers/publishers over udp correctly
MirkoWaechter
Member Mirko WaechterOrganization: University of KarlsruheProject: ArmarX Robot Framework ✭
in Help Center
Hi,
I am having trouble configuring IceStorm and the publishers and subscribers correctly to send the messages over udp.
It works with tcp.
I have started IceStorm with IceStorm.Publish.Endpoints=udp. Is that needed?
This is the publisher:
This is the receiver:
Where do I need to insert "udp" to let publisher and/or the subscriber use udp?
thanks
Mirko
I am having trouble configuring IceStorm and the publishers and subscribers correctly to send the messages over udp.
It works with tcp.
I have started IceStorm with IceStorm.Publish.Endpoints=udp. Is that needed?
This is the publisher:
[COLOR=#000000] intmain(intargc,char*argv[])
{
Ice::CommunicatorPtrcommunicator;
communicator=Ice::initialize(argc,argv);
Ice::ObjectPrxobj=communicator->[I]stringToProxy[/I]("IceStorm/TopicManager");
IceStorm::TopicManagerPrxtopicManager=IceStorm::TopicManagerPrx::checkedCast(obj);
IceStorm::TopicPrxtopic;
while(!topic){
try{
topic=topicManager->retrieve("Weather");
}catch(constIceStorm::NoSuchTopic&){
try{
topic=topicManager->create("Weather");
}catch(constIceStorm::TopicExists&){
//Anotherclientcreatedthetopic.
}
}
}
Ice::ObjectPrxpub=topic->getPublisher()->ice_oneway();
Test::mytopicPrxmonitor=Test::mytopicPrx::uncheckedCast(pub);
while(true){
doubletime=IceUtil::Time::now().toMicroSecondsDouble();
monitor->reportTime(time);
std::cout<<time<<std::endl;
usleep(1000000);
}
}[/COLOR]
This is the receiver:
[COLOR=#000000] classMyTopicI:virtualpublicTest::mytopic{
public:
//mytopicinterface
public:
void[I]reportTime[/I](Ice::Doublet,constIce::Current&)
{
Ice::Doubletime=IceUtil::Time::now().toMicroSecondsDouble();
std::cout<<"t:"<<(time-t)<<std::endl;
}
};
typedefIceInternal::Handle<MyTopicI>MyTopicIPtr;
intmain(intargc,char*argv[])
{
Ice::CommunicatorPtrcommunicator;
communicator=Ice::initialize(argc,argv);
Ice::ObjectPrxobj=communicator->[I]stringToProxy[/I]("IceStorm/TopicManager");
IceStorm::TopicManagerPrxtopicManager=IceStorm::TopicManagerPrx::checkedCast(obj);
Ice::ObjectAdapterPtradapter=communicator->[I]createObjectAdapterWithEndpoints[/I]("MonitorAdapter","udp");
MyTopicIPtrmonitor=newMyTopicI;
Ice::ObjectPrxproxy=adapter->[I]addWithUUID[/I](monitor)->ice_oneway();
adapter->[I]activate[/I]();
IceStorm::TopicPrxtopic;
try{
topic=topicManager->retrieve("Weather");
IceStorm::QoSqos;
topic->subscribeAndGetPublisher(qos,proxy);
}
catch(constIceStorm::NoSuchTopic&){
//Error!Notopicfound!
std::cerr<<"notopicfound";
}
communicator->[I]waitForShutdown[/I]();
topic->unsubscribe(proxy);
}[/COLOR]
Where do I need to insert "udp" to let publisher and/or the subscriber use udp?
thanks
Mirko
0
Comments
For your subscriber and publisher proxies you should be calling ice_datagram(), not ice_oneway().
Regards
Is it possible to have UDP and TCP publishers on the same IceStorm?
The property IceStorm.Publish.Endpoints=udp seems to fix all publishers to udp.
When using udp the publish rate is around 110k Hz on my pc and the receive rate is 22k Hz, so a lot is dropped, which is acceptable at that rate.
When I now insert a sleep of 1ms into the publisher the publish rate of course goes down to around 1k Hz.
But also the receive rate goes down, to 100Hz.
Why could that be?
Another strange thing:
I am measuring they delay of the topic message.
When using tcp the delay is around 0.5ms.
When using udp the delay goes up to 12ms. And without a sleep in the sender it goes up to a few seconds.
Edit: With UDP it is completely unexplainable to me.
If I reset Ice completely and insert a 1000ms sleep it works for the first start of publisher and subscriber.
When I restart both, the receiving rate goes down to 50 and the delays is a few seconds. Also there builds up a backlog from the publisher (of course, because the publisher publishes with 1k Hz but the receiver only gets the data at 50 Hz).
Yes, it's possible to use both UDP and TCP for publishing, just specify [FONT=courier new, courier, monospace]IceStorm.Publish.Endpoints=udp,tcp[/FONT][FONT=arial, helvetica, sans-serif] for the publish endpoints.[/FONT]
[FONT=arial, helvetica, sans-serif]Which Ice version do you use and on which platform? Do you have a small test case we could use to reproduce what you're observing?
Cheers,
Benoit.[/FONT]
I did that, but then I get the following Exception for a tcp connection as well as for a udp: terminate called after throwing an instance of 'Ice::NoEndpointException'
what(): Reference.cpp:1663: Ice::NoEndpointException:
no suitable endpoint available for proxy `IceStorm/TopicManager -t -e 1.0'
My version is Ice 3.5.0 and Ubuntu 13.
My version is Ice 3.5.0 and Ubuntu 13.
This is the code:
Subscriber:
[COLOR=#000000]class MyTopicI : virtual public Test::mytopic {[/COLOR] [COLOR=#000000]public:[/COLOR] [COLOR=#000000] MyTopicI() :i(0){}[/COLOR] [COLOR=#000000] // mytopic interface[/COLOR] [COLOR=#000000]public:[/COLOR] [COLOR=#000000] void [I]reportTime[/I](Ice::Double t, const Ice::Current &)[/COLOR] [COLOR=#000000] {[/COLOR] [COLOR=#000000] static IceUtil::Time start = IceUtil::Time::now();[/COLOR] [COLOR=#000000] Ice::Double time = IceUtil::Time::now().toMicroSecondsDouble();[/COLOR] [COLOR=#000000]// if (i % 100 == 0)[/COLOR] [COLOR=#000000] {[/COLOR] [COLOR=#000000] std::cout << i << " rate: " << i/(IceUtil::Time::now()-start).toSecondsDouble() << std::endl;[/COLOR] [COLOR=#000000] std::cout << "t: " << (time-t)/1000/1000 << std::endl;[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] i++;[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] int i;[/COLOR] [COLOR=#000000]};[/COLOR] [COLOR=#000000]typedef IceInternal::Handle<MyTopicI> MyTopicIPtr;[/COLOR] [COLOR=#000000]int main(int argc, char* argv[])[/COLOR] [COLOR=#000000]{[/COLOR] [COLOR=#000000] Ice::CommunicatorPtr communicator;[/COLOR] [COLOR=#000000] communicator = Ice::initialize(argc, argv);[/COLOR] [COLOR=#000000] Ice::ObjectPrx obj = communicator->[I]stringToProxy[/I]("IceStorm/TopicManager");[/COLOR] [COLOR=#000000] IceStorm::TopicManagerPrx topicManager = IceStorm::TopicManagerPrx::checkedCast(obj);[/COLOR] [COLOR=#000000] Ice::ObjectAdapterPtr adapter = communicator->[I]createObjectAdapterWithEndpoints[/I]("MonitorAdapter", "udp");[/COLOR] [COLOR=#000000] MyTopicIPtr monitor = new MyTopicI;[/COLOR] [COLOR=#000000] Ice::ObjectPrx proxy = adapter->[I]addWithUUID[/I](monitor)->ice_datagram();[/COLOR] [COLOR=#000000] adapter->[I]activate[/I]();[/COLOR] [COLOR=#000000] IceStorm::TopicPrx topic;[/COLOR] [COLOR=#000000] try {[/COLOR] [COLOR=#000000] topic = topicManager->retrieve("Weather");[/COLOR] [COLOR=#000000] IceStorm::QoS qos;[/COLOR] [COLOR=#000000] topic->subscribeAndGetPublisher(qos, proxy);[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] catch (const IceStorm::NoSuchTopic&) {[/COLOR] [COLOR=#000000] // Error! No topic found![/COLOR] [COLOR=#000000] std::cerr<< "no topic found";[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] communicator->[I]waitForShutdown[/I]();[/COLOR] [COLOR=#000000] topic->unsubscribe(proxy);[/COLOR] [COLOR=#000000]}[/COLOR]Publisher:[COLOR=#000000]int main(int argc, char* argv[])[/COLOR] [COLOR=#000000]{[/COLOR] [COLOR=#000000] Ice::CommunicatorPtr communicator;[/COLOR] [COLOR=#000000] communicator = Ice::initialize(argc, argv);[/COLOR] [COLOR=#000000] Ice::ObjectPrx obj = communicator->[I]stringToProxy[/I]("IceStorm/TopicManager");[/COLOR] [COLOR=#000000] IceStorm::TopicManagerPrx topicManager = IceStorm::TopicManagerPrx::checkedCast(obj);[/COLOR] [COLOR=#000000] IceStorm::TopicPrx topic;[/COLOR] [COLOR=#000000] while (!topic) {[/COLOR] [COLOR=#000000] try {[/COLOR] [COLOR=#000000] topic = topicManager->retrieve("Weather");[/COLOR] [COLOR=#000000] } catch (const IceStorm::NoSuchTopic&) {[/COLOR] [COLOR=#000000] try {[/COLOR] [COLOR=#000000] topic = topicManager->create("Weather");[/COLOR] [COLOR=#000000] } catch (const IceStorm::TopicExists&) {[/COLOR] [COLOR=#000000] // Another client created the topic.[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000] Ice::ObjectPrx pub = topic->getPublisher()->ice_datagram();[/COLOR] [COLOR=#000000] Test::mytopicPrx monitor = Test::mytopicPrx::uncheckedCast(pub);[/COLOR] [COLOR=#000000] IceUtil::Time start = IceUtil::Time::now();[/COLOR] [COLOR=#000000] long i = 0;[/COLOR] [COLOR=#000000] while (true) {[/COLOR] [COLOR=#000000] double time = IceUtil::Time::now().toMicroSecondsDouble();[/COLOR] [COLOR=#000000] monitor->reportTime(time);[/COLOR] [COLOR=#000000] //std::cout << time << std::endl;[/COLOR] [COLOR=#000000] if (i % 10000 == 0)[/COLOR] [COLOR=#000000] std::cout << i << " rate: " << i/(IceUtil::Time::now()-start).toSecondsDouble() << std::endl;[/COLOR] [COLOR=#000000] i++;[/COLOR] [COLOR=#000000]// sleep(1);[/COLOR] [COLOR=#000000] usleep(1000);[/COLOR] [COLOR=#000000] }[/COLOR] [COLOR=#000000]}[/COLOR]I suppose the exception comes from the publisher process. An Ice::NoEndpointException indicates that the IceGrid node failed to start the IceBox server. Can you check the error message in the IceGrid node log file or on the standard error output where you started the IceGrid node?
Any chance you could try your test case with the latest Ice 3.6.1 version?
Cheers,
Benoit.
!! 11/07/15 01:21:42.897 IceStorm: error: ServiceManager: exception while starting service IceStorm:
ObjectAdapterI.cpp:1239: Ice::EndpointParseException:
error while parsing endpoint `invalid object adapter endpoint `udp,tcp''
I can try to compile Ice 3.6.1 and check it.
Sorry, I meant udp:tcp for the endpoints not udp,tcp.
Cheers,
Benoit.
It sounds like you checked out the master branch instead of the 3.6 branch.
To checkout the 3.6 branch in your existing ice repository: git checkout -b 3.6 origin/3.6
Cheers,
Benoit.
With branch 3.6 it works.
The problem with 3.6 is not the same.
The performance does not get worse on restart now, but it degrades over time:
t is the delay in ms and rate is the report frequency.
I tried your test case on Ubuntu 15.04 with Ice 3.6.1 and I'm not seeing this issue. The delay to receive the event remains stable. I tried restarting the publisher and changing the frequency of the updates too. Any chance you could try on another platform to see if you get the same problem?
Note that the rate computation in the test case you provided in the post above isn't correct. I also recommend to use an average for "t" instead of just taking snapshots every 1000 samples.
Cheers,
Benoit.
I'll try to setup something with Ubuntu 15.04 when I have some time.
What's wrong with the rate computation? I cannot find any error right now. It is the average rate starting from the first reportTime() call.
Here's the code you provided:
void reportTime(Ice::Double t, const Ice::Current&) { static IceUtil::Time start = IceUtil::Time::now(); Ice::Double time = IceUtil::Time::now().toMicroSecondsDouble(); // if (i % 100 == 0) { std::cout << i << " rate: " << i/(IceUtil::Time::now()-start).toSecondsDouble() << std::endl; std::cout << "t: " << (time-t)/1000/1000 << std::endl; } i++; }The start variable is a local variable so the rate isn't computed based on the first reportTime call but on the call being dispatched.
Cheers,
Benoit
Cheers,
Benoit.