Revision: 6421
http://playerstage.svn.sourceforge.net/playerstage/?rev=6421&view=rev
Author: thjc
Date: 2008-05-10 22:08:29 -0700 (Sat, 10 May 2008)
Log Message:
-----------
Merged 6410,6413 from 2-1 branch
dynamic reconnection of passthrough
Modified Paths:
--------------
code/player/trunk/libplayertcp/playertcp.cc
code/player/trunk/libplayertcp/playertcp.h
code/player/trunk/libplayertcp/remote_driver.cc
code/player/trunk/server/drivers/mixed/mricp/src/mricp_driver.cpp
code/player/trunk/server/drivers/position/nav200/sicknav200.cc
code/player/trunk/server/drivers/shell/passthrough.cc
code/player/trunk/utils/playerprop/playerprop.cpp
Modified: code/player/trunk/libplayertcp/playertcp.cc
===================================================================
--- code/player/trunk/libplayertcp/playertcp.cc 2008-05-11 04:57:19 UTC (rev
6420)
+++ code/player/trunk/libplayertcp/playertcp.cc 2008-05-11 05:08:29 UTC (rev
6421)
@@ -280,9 +280,12 @@
PLAYER_MSG3(1, "accepted TCP client %d on port %d, fd %d",
j, this->clients[j].port, this->clients[j].fd);
+ assert (this->clients[j].queue != NULL);
+
if(!have_lock)
Unlock();
+ assert (this->clients[j].queue != NULL);
return(this->clients[j].queue);
}
@@ -454,7 +457,6 @@
}
}
-
this->DeleteClients();
if(!have_lock)
Unlock();
@@ -477,6 +479,8 @@
num_deleted++;
}
}
+ /* removed this block of code as it does nothing as the loop above has
already killed them all
+ * is this what was intended here?
// Delete those connections that generated errors in this iteration
for(int i=0; i<this->num_clients; i++)
{
@@ -485,7 +489,7 @@
this->Close(i);
num_deleted++;
}
- }
+ }*/
this->num_clients -= num_deleted;
@@ -735,6 +739,13 @@
int
PlayerTCP::Write(bool have_lock)
{
+ if (have_lock)
+ {
+ int ret = pthread_mutex_trylock(&clients_mutex);
+ assert (ret == EBUSY);
+ }
+
+
if(!have_lock)
Lock();
@@ -747,7 +758,6 @@
}
}
-
this->DeleteClients();
if(!have_lock)
Unlock();
@@ -769,6 +779,7 @@
int
PlayerTCP::ReadClient(int cli)
{
+ assert(pthread_mutex_trylock(&clients_mutex) == EBUSY);
int numread;
playertcp_conn_t* client;
Modified: code/player/trunk/libplayertcp/playertcp.h
===================================================================
--- code/player/trunk/libplayertcp/playertcp.h 2008-05-11 04:57:19 UTC (rev
6420)
+++ code/player/trunk/libplayertcp/playertcp.h 2008-05-11 05:08:29 UTC (rev
6421)
@@ -136,13 +136,14 @@
/** Total size of @p decode_readbuffer */
int decode_readbuffersize;
- void Lock();
- void Unlock();
public:
PlayerTCP();
~PlayerTCP();
+ void Lock();
+ void Unlock();
+
static void InitGlobals(void);
pthread_t thread;
Modified: code/player/trunk/libplayertcp/remote_driver.cc
===================================================================
--- code/player/trunk/libplayertcp/remote_driver.cc 2008-05-11 04:57:19 UTC
(rev 6420)
+++ code/player/trunk/libplayertcp/remote_driver.cc 2008-05-11 05:08:29 UTC
(rev 6421)
@@ -373,11 +373,13 @@
if(this->ptcp->thread == pthread_self())
{
//this->ptcp->Read(0,true);
+ this->ptcp->Lock();
this->ptcp->ReadClient(this->queue);
+ this->ptcp->Unlock();
}
this->ProcessMessages();
if(this->ptcp->thread == pthread_self())
- this->ptcp->Write(true);
+ this->ptcp->Write(false);
}
int
Modified: code/player/trunk/server/drivers/mixed/mricp/src/mricp_driver.cpp
===================================================================
--- code/player/trunk/server/drivers/mixed/mricp/src/mricp_driver.cpp
2008-05-11 04:57:19 UTC (rev 6420)
+++ code/player/trunk/server/drivers/mixed/mricp/src/mricp_driver.cpp
2008-05-11 05:08:29 UTC (rev 6421)
@@ -329,7 +329,6 @@
void mricp_Register(DriverTable* table)
{
- puts("Driver added to table"); fflush(stdout);
table->AddDriver("mricp", MrIcp_Init);
}
Modified: code/player/trunk/server/drivers/position/nav200/sicknav200.cc
===================================================================
--- code/player/trunk/server/drivers/position/nav200/sicknav200.cc
2008-05-11 04:57:19 UTC (rev 6420)
+++ code/player/trunk/server/drivers/position/nav200/sicknav200.cc
2008-05-11 05:08:29 UTC (rev 6421)
@@ -578,7 +578,7 @@
return 0;
}
}
- if (Message::MatchMessage(hdr, PLAYER_MSGTYPE_REQ,
PLAYER_SET_STRPROP_REQ)) {
+ if (Message::MatchMessage(hdr, PLAYER_MSGTYPE_REQ,
PLAYER_SET_INTPROP_REQ)) {
player_intprop_req_t &req =
*reinterpret_cast<player_intprop_req_t*> (data);
if (strcmp("nearest", req.key) == 0) {
Modified: code/player/trunk/server/drivers/shell/passthrough.cc
===================================================================
--- code/player/trunk/server/drivers/shell/passthrough.cc 2008-05-11
04:57:19 UTC (rev 6420)
+++ code/player/trunk/server/drivers/shell/passthrough.cc 2008-05-11
05:08:29 UTC (rev 6421)
@@ -48,6 +48,14 @@
be reconfigured, since the change is done only in the server running the
passthrough
driver.
+The passthrough driver is also able to change its remote address at runtime.
To do
+this set the connect property to 0, and then change as needed the remote_host,
remote_port
+and remote_index properties. When you set connect to 1, it will connect to the
new address.
+Setting connect to -1 will trigger a disconnect followed by a connect allowing
for seamless
+transfer to a new remote device.
+
+Subscribed clients will have all requests nack'd while the driver is
disconnected.
+
@par Compile-time dependencies
- none
@@ -114,6 +122,9 @@
virtual int Setup();
virtual int Shutdown();
+ int ConnectRemote();
+ int DisconnectRemote();
+
virtual int ProcessMessage(QueuePointer &resp_queue, player_msghdr * hdr,
void * data);
private:
@@ -125,6 +136,15 @@
player_devaddr_t srcAddr;
//the device that this server connects to to get data
Device *srcDevice;
+
+ // properties
+ StringProperty RemoteHost;
+ IntProperty RemotePort;
+ IntProperty RemoteIndex;
+
+ IntProperty Connect;
+ int Connected;
+
};
@@ -140,7 +160,13 @@
PassThrough::PassThrough(ConfigFile* cf, int section)
- : Driver(cf, section) {
+ : Driver(cf, section),
+ RemoteHost("remote_host","",false,this,cf,section),
+ RemotePort("remote_port",-1,false,this,cf,section),
+ RemoteIndex("remote_index",-1,false,this,cf,section),
+ Connect("connect",1,false,this,cf,section),
+ Connected(0)
+{
memset(&srcAddr, 0, sizeof(player_devaddr_t));
memset(&dstAddr, 0, sizeof(player_devaddr_t));
@@ -175,7 +201,58 @@
PLAYER_MSG0(1,"PassThrough driver ready");
+ if (Connect)
+ {
+ int ret = ConnectRemote();
+ if (ret)
+ return ret;
+
+ }
+ StartThread();
+
+ return(0);
+}
+
+
+int PassThrough::Shutdown() {
+ PLAYER_MSG0(1,"Shutting PassThrough driver down");
+
+
+ StopThread();
+
+ DisconnectRemote();
+
+ PLAYER_MSG0(1,"PassThrough driver has been shutdown");
+
+ return(0);
+}
+
+int PassThrough::ConnectRemote()
+{
+ if (Connected)
+ return 0;
+
+ if (RemoteHost.GetValue()[0] != '\0')
+ {
+ PLAYER_MSG1(3,"Overriding remote hostname to %s",
RemoteHost.GetValue());
+ // assume it's a string containing a hostname or IP address
+ if(hostname_to_packedaddr(&srcAddr.host, RemoteHost.GetValue()) < 0)
+ {
+ PLAYER_ERROR1("name lookup failed for host \"%s\"",
RemoteHost.GetValue());
+ return -1;
+ }
+ }
+ if (RemotePort != -1)
+ {
+ PLAYER_MSG1(3,"Overriding remote robot to %d", RemotePort.GetValue());
+ srcAddr.robot = RemotePort;
+ }
+ if (RemoteIndex != -1)
+ {
+ PLAYER_MSG1(3,"Overriding remote index to %d", RemoteIndex.GetValue());
+ srcAddr.index = RemoteIndex;
+ }
srcDevice=deviceTable->GetDevice(srcAddr);
if (!srcDevice) {
@@ -188,28 +265,22 @@
PLAYER_ERROR("unable to subscribe to device");
return -1;
}
-
-
- StartThread();
-
- return(0);
+ Connected = 1;
+ return 0;
}
-
-int PassThrough::Shutdown() {
- PLAYER_MSG0(1,"Shutting PassThrough driver down");
-
-
- StopThread();
-
+int PassThrough::DisconnectRemote()
+{
+ if (Connected == 0)
+ return 0;
//Our clients disconnected, so let's disconnect from our SRC interface
srcDevice->Unsubscribe(this->InQueue);
+
+ Connected = 0;
+ return 0;
+}
- PLAYER_MSG0(1,"PassThrough driver has been shutdown");
- return(0);
-}
-
int PassThrough::ProcessMessage(QueuePointer & resp_queue,
player_msghdr * hdr,
void * data) {
@@ -217,6 +288,52 @@
bool inspected(false);
+ // let our properties through
+ if (Message::MatchMessage(hdr, PLAYER_MSGTYPE_REQ,
PLAYER_SET_STRPROP_REQ))
+ {
+ player_strprop_req_t req = *reinterpret_cast<player_strprop_req_t*>
(data);
+ if (strcmp("remote_host", req.key) == 0)
+ return -1;
+ }
+
+ if (Message::MatchMessage(hdr, PLAYER_MSGTYPE_REQ,
PLAYER_SET_INTPROP_REQ))
+ {
+ player_intprop_req_t req = *reinterpret_cast<player_intprop_req_t*>
(data);
+ if (strcmp("remote_port", req.key) == 0)
+ return -1;
+ if (strcmp("remote_index", req.key) == 0)
+ return -1;
+ if (strcmp("connect", req.key) == 0)
+ {
+ if (req.value == 0) // disconnect
+ {
+ DisconnectRemote();
+ }
+ else if (req.value == 1) // connect
+ {
+ ConnectRemote();
+ }
+ else if (req.value == -1) // reconnect (with new address if its
changed)
+ {
+ DisconnectRemote();
+ ConnectRemote();
+ }
+
+ return -1;
+ }
+ }
+
+ // silence warning etc while we are not connected
+ if (!Connected)
+ {
+ if (hdr->type == PLAYER_MSGTYPE_REQ)
+ {
+ Publish(dstAddr,resp_queue,PLAYER_MSGTYPE_RESP_NACK,hdr->subtype);
+ }
+ return 0;
+ }
+
+
PLAYER_MSG0(9,"PassThrough::ProcessMessage: Received a packet!");
if (Device::MatchDeviceAddress(hdr->addr,srcAddr) &&
@@ -273,7 +390,6 @@
inspected=true;
}
-
//Check if a packet went thorugh, without being forwarded
if (!inspected) {
static bool reported(false);
@@ -294,12 +410,10 @@
void PassThrough::Main() {
//The forwarding is done in the ProcessMessage method. Called once per
each message by ProcessMessages()
- while (true) {
- pthread_testcancel();
-
+ while (true)
+ {
+ InQueue->Wait();
ProcessMessages();
-
- usleep(100);
}
}
Modified: code/player/trunk/utils/playerprop/playerprop.cpp
===================================================================
--- code/player/trunk/utils/playerprop/playerprop.cpp 2008-05-11 04:57:19 UTC
(rev 6420)
+++ code/player/trunk/utils/playerprop/playerprop.cpp 2008-05-11 05:08:29 UTC
(rev 6421)
@@ -121,6 +121,12 @@
case PLAYER_FIDUCIAL_CODE:
deviceProxy = (ClientProxy*) new FiducialProxy (&client,
devIndex);
break;
+ case PLAYER_GRAPHICS2D_CODE:
+ deviceProxy = (ClientProxy*) new Graphics2dProxy (&client,
devIndex);
+ break;
+ case PLAYER_GRAPHICS3D_CODE:
+ deviceProxy = (ClientProxy*) new Graphics3dProxy (&client,
devIndex);
+ break;
case PLAYER_GRIPPER_CODE:
deviceProxy = (ClientProxy*) new GripperProxy (&client,
devIndex);
break;
This was sent by the SourceForge.net collaborative development platform, the
world's largest Open Source development site.
-------------------------------------------------------------------------
This SF.net email is sponsored by the 2008 JavaOne(SM) Conference
Don't miss this year's exciting event. There's still time to save $100.
Use priority code J8TL2D2.
http://ad.doubleclick.net/clk;198757673;13503038;p?http://java.sun.com/javaone
_______________________________________________
Playerstage-commit mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/playerstage-commit
