Fix/107

clients/rospy/src/rospy/impl/registration.py

@@ -306,6 +306,8 @@ def run(self):
                 if cond is not None:
                     cond.release()
 
+            get_topic_manager().check_all()
+
             #call _connect_topic on all URIs as it can check to see whether
             #or not a connection exists.
             if uris and not self.handler.done:

clients/rospy/src/rospy/topics.py

@@ -109,6 +109,10 @@ def isstring(s):
 # for interfacing with topics, while _TopicImpl implements the
 # underlying connection details. 
 
+if not hasattr(select, 'EPOLLRDHUP'):
+    select.EPOLLRDHUP = 0x2000
+
+
 class Topic(object):
     """Base class of L{Publisher} and L{Subscriber}"""
 
@@ -188,25 +192,28 @@ class Poller(object):
     on multiple platforms.  NOT thread-safe.
     """
     def __init__(self):
-        try:
+        if hasattr(select, 'epoll'):
+            self.poller = select.epoll()
+            self.add_fd = self.add_epoll
+            self.remove_fd = self.remove_epoll
+            self.error_iter = self.error_epoll_iter
+        elif hasattr(select, 'poll'):
             self.poller = select.poll()
             self.add_fd = self.add_poll
             self.remove_fd = self.remove_poll
             self.error_iter = self.error_poll_iter
-        except:
-            try:
-                # poll() not available, try kqueue
-                self.poller = select.kqueue()
-                self.add_fd = self.add_kqueue
-                self.remove_fd = self.remove_kqueue
-                self.error_iter = self.error_kqueue_iter
-                self.kevents = []
-            except:
-                #TODO: non-Noop impl for Windows
-                self.poller = self.noop
-                self.add_fd = self.noop
-                self.remove_fd = self.noop
-                self.error_iter = self.noop_iter
+        elif hasattr(select, 'kqueue'):
+            self.poller = select.kqueue()
+            self.add_fd = self.add_kqueue
+            self.remove_fd = self.remove_kqueue
+            self.error_iter = self.error_kqueue_iter
+            self.kevents = []
+        else:
+            #TODO: non-Noop impl for Windows
+            self.poller = self.noop
+            self.add_fd = self.noop
+            self.remove_fd = self.noop
+            self.error_iter = self.noop_iter
 
     def noop(self, *args):
         pass
@@ -228,6 +235,18 @@ def error_poll_iter(self):
             if event & (select.POLLHUP | select.POLLERR):
                 yield fd
 
+    def add_epoll(self, fd):
+        self.poller.register(fd, select.EPOLLHUP|select.EPOLLERR|select.EPOLLRDHUP)
+
+    def remove_epoll(self, fd):
+        self.poller.unregister(fd)
+
+    def error_epoll_iter(self):
+        events = self.poller.poll(0)
+        for fd, event in events:
+            if event & (select.EPOLLHUP | select.EPOLLERR | select.EPOLLRDHUP):
+                yield fd
+
     def add_kqueue(self, fd):
         self.kevents.append(select.kevent(fd))
 
@@ -439,6 +458,17 @@ def cleanup_cb_wrapper(s):
 
             return True
 
+    def check(self):
+        fds_to_remove = list(self.connection_poll.error_iter())
+        if fds_to_remove:
+            with self.c_lock:
+                new_connections = self.connections[:]
+                to_remove = [x for x in new_connections if x.fileno() in fds_to_remove]
+                for x in to_remove:
+                    rospydebug("removing connection to %s, connection error detected"%(x.endpoint_id))
+                    self._remove_connection(new_connections, x)
+                self.connections = new_connections
+
     def remove_connection(self, c):
         """
         Remove connection from topic.
@@ -1136,6 +1166,15 @@ def close_all(self):
                 t.close()
             self.pubs.clear()
             self.subs.clear()        
+
+
+    def check_all(self):
+        """
+        Check all registered publication and subscriptions.
+        """
+        with self.lock:
+            for t in chain(iter(self.pubs.values()), iter(self.subs.values())):
+                t.check()
 
     def _add(self, ps, rmap, reg_type):
         """

test/test_rospy/CMakeLists.txt

@@ -57,4 +57,5 @@ if(CATKIN_ENABLE_TESTING)
   add_rostest(test/rostest/latch.test)
   add_rostest(test/rostest/on_shutdown.test)
   add_rostest(test/rostest/sub_to_multiple_pubs.test)
+  add_rostest(test/rostest/latch_unsubscribe.test)
 endif()

test/test_rospy/nodes/listener_once.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+# Software License Agreement (BSD License)
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above
+#    copyright notice, this list of conditions and the following
+#    disclaimer in the documentation and/or other materials provided
+#    with the distribution.
+#  * Neither the name of Willow Garage, Inc. nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# Revision $Id: gossipbot.py 1013 2008-05-21 01:08:56Z sfkwc $
+
+## Simple talker demo that listens to std_msgs/Strings published
+## to the 'chatter' topic and shuts down afterwards
+
+from __future__ import print_function
+
+import rospy
+from std_msgs.msg import String
+
+
+def callback(data):
+    print(rospy.get_caller_id(), "I heard %s" % data.data)
+    rospy.signal_shutdown("Received %s, exiting now"  % data.data)
+
+
+def listener():
+    rospy.init_node('listener', anonymous=True)
+    rospy.sleep(rospy.get_param('delay', 0.0))
+    rospy.Subscriber("chatter", String, callback)
+    rospy.spin()
+
+
+if __name__ == '__main__':
+    listener()

test/test_rospy/package.xml

@@ -18,6 +18,7 @@
   <build_depend>test_rosmaster</build_depend>
 
   <test_depend>python-numpy</test_depend>
+  <test_depend>python-psutil</test_depend>
   <test_depend>rosbuild</test_depend>
   <test_depend>rosgraph</test_depend>
   <test_depend>rospy</test_depend>

test/test_rospy/test/rostest/latch_unsubscribe.test

@@ -0,0 +1,7 @@
+<launch>
+  <node name="listener_once_1" pkg="test_rospy" type="listener_once.py" output="screen" />
+  <node name="listener_once_2" pkg="test_rospy" type="listener_once.py" output="screen">
+    <param name="delay" value="1.0" type="double" />
+  </node>
+  <test test-name="test_latch_unsubscribe" pkg="test_rospy" type="test_latch_unsubscribe.py" />
+</launch>

test/test_rospy/test/rostest/test_latch_unsubscribe.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+# Software License Agreement (BSD License)
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above
+#    copyright notice, this list of conditions and the following
+#    disclaimer in the documentation and/or other materials provided
+#    with the distribution.
+#  * Neither the name of Willow Garage, Inc. nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+PKG = 'test_rospy'
+NAME = 'test_latch_unsubscribe'
+
+import os
+import sys
+import unittest
+
+import psutil
+
+from std_msgs.msg import String
+
+
+def _get_connections(process_info):
+    if hasattr(process_info, 'connections'):  # new naming
+        return process_info.connections()
+    elif hasattr(process_info, 'get_connections'):  # old naming
+        return process_info.get_connections()
+    raise AttributeError('Wrong psutil version?')
+
+
+def _get_connection_statii(process_info):
+    return (conn.status for conn in _get_connections(process_info))
+
+
+class TestLatch(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def test_latch(self):
+        import rospy
+        proc_info = psutil.Process(os.getpid())
+        self.assertNotIn('CLOSE_WAIT', _get_connection_statii(proc_info),
+                         'CLOSE_WAIT sockets already before the test. This '
+                         'should not happen at all.')
+
+        rospy.init_node(NAME)
+        pub = rospy.Publisher('chatter', String, latch=True)
+        pub.publish(String("hello"))
+        rospy.sleep(0.5)
+        self.assertNotIn('CLOSE_WAIT', _get_connection_statii(proc_info),
+                         'CLOSE_WAIT sockets after the subscriber exited. '
+                         '(#107)')
+        rospy.sleep(1.5)
+        # also check for a second subscriber
+        self.assertNotIn('CLOSE_WAIT', _get_connection_statii(proc_info),
+                         'CLOSE_WAIT sockets after the second subscriber '
+                         'exited. (#107)')
+
+
+if __name__ == '__main__':
+    import rostest
+    rostest.run(PKG, NAME, TestLatch, sys.argv)