title: How to make a TCP source reconnect in GNU radio?
body: <p>How do you make a TCP source server reconnect in GNU radio? Once you have connected a client once, it cannot accept new connections. Having studied <code>tcp.py</code> it is obvious that this is because the python part only calls <code>accept</code> once.</p>
  
  <p>This issue was asked before on a mailing list: <a href="https://lists.gnu.org/archive/html/discuss-gnuradio/2010-04/msg00501.html" rel="nofollow">https://lists.gnu.org/archive/html/discuss-gnuradio/2010-04/msg00501.html</a> but the answer was slightly unsatisfactory, because it was supposedly easy to do with some "dup magic".</p>
  
  <p>I managed to make a reconnecting socket using the file descriptor as an identifier in Python in the following way:</p>
  
  <pre><code>#!/usr/bin/env python
  
  import os
  import socket
  import sys
  import threading
  
  def get_server_socket_and_fd():
      s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      s.bind(('127.0.0.1', 9999))
      s.listen(1)
      print "waiting for client connection on port 9999"
      cs, address = s.accept()
      return s, os.dup(cs.fileno())
  
  class StoppableThread(threading.Thread):
      def __init__(self):
          super(StoppableThread, self).__init__()
          self._stop = threading.Event()
  
      def stop(self):
          self._stop.set()
  
      def stopped(self):
          return self._stop.isSet()
  
  class SocketWatcher(StoppableThread):
      def __init__(self, server_socket, fd):
          super(SocketWatcher, self).__init__()
          self.setDaemon(True)
          self._fd = fd
          self._server_socket = server_socket
  
      def run(self):
          print "WWW: starting watcher on fd = {}".format(self._fd)
          while not self.stopped():
              #print "WWW: creating socket object from fd = {}".format(self._fd)
              s = socket.fromfd(self._fd, socket.AF_INET, socket.SOCK_STREAM)
              while not self.stopped():
                  print "WWW: trying to peek for new data"
                  data = s.recv(1024, socket.MSG_PEEK)
                  print "WWW: msg peek returned: {} bytes".format(len(data))
                  if len(data) == 0:
                      print "WWW: EOF? closing socket"
                      s.close()
                      print "WWW: waiting for new connection..."
                      cs, address = self._server_socket.accept()
                      print "WWW: new connection! fileno = {}".format(cs.fileno())
                      print "WWW: duplicating this client socket fd into the old one"
                      os.dup2(cs.fileno(), self._fd)
                      break
          print "WWW: thread stopped, exiting run method"
  
  server_socket, client_fd = get_server_socket_and_fd()
  
  watcher = SocketWatcher(server_socket, client_fd)
  watcher.start()
  
  try:
      while True:
          s = socket.fromfd(client_fd, socket.AF_INET, socket.SOCK_STREAM)
          data = s.recv(1024)
          if len(data) &gt; 0:
              print "    data received: {} bytes".format(len(data))
              print repr(data)            
  
  except (KeyboardInterrupt, SystemExit):
      print "stopping program..."
      watcher.stop()
      sys.exit()
  </code></pre>
  
  <p>which works like expected: start python script, connect to the port, write things, close connection, open another connection, write more things, and notice that they continue to be printed.</p>
  
  <p>However, when I try to integrate this into GNU radio: it does not work. This is my best attempt:</p>
  
  <pre><code>#
  # Copyright 2009 Free Software Foundation, Inc.
  #
  # This file is part of GNU Radio
  #
  # GNU Radio is free software; you can redistribute it and/or modify
  # it under the terms of the GNU General Public License as published by
  # the Free Software Foundation; either version 3, or (at your option)
  # any later version.
  #
  # GNU Radio is distributed in the hope that it will be useful,
  # but WITHOUT ANY WARRANTY; without even the implied warranty of
  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  # GNU General Public License for more details.
  #
  # You should have received a copy of the GNU General Public License
  # along with GNU Radio; see the file COPYING.  If not, write to
  # the Free Software Foundation, Inc., 51 Franklin Street,
  # Boston, MA 02110-1301, USA.
  #
  
  from gnuradio import gr, blocks
  import socket
  import os
  import threading
  
  class StoppableThread(threading.Thread):
      def __init__(self):
          super(StoppableThread, self).__init__()
          self._stop = threading.Event()
  
      def stop(self):
          self._stop.set()
  
      def stopped(self):
          return self._stop.isSet()
  
  class SocketWatcher(StoppableThread):
      def __init__(self, server_socket, fd):
          super(SocketWatcher, self).__init__()
          self.setDaemon(True)
          self._fd = fd
          self._server_socket = server_socket
  
      def run(self):
          while not self.stopped():
              s = socket.fromfd(self._fd, socket.AF_INET, socket.SOCK_STREAM)
              while not self.stopped():
                  data = s.recv(1024, socket.MSG_PEEK)
                  if len(data) == 0:
                      print "EOF detected. Closing socket and waiting for new connection..."
                      s.close()
                      cs, address = self._server_socket.accept()
                      print "got new connection!"
                      os.dup2(cs.fileno(), self._fd)
                      break
  
  
  def _get_sock_fd(addr, port, server):
      """
      Get the file descriptor for the socket.
      As a client, block on connect, dup the socket descriptor.
      As a server, block on accept, dup the client descriptor.
  
      Args:
          addr: the ip address string
          port: the tcp port number
          server: true for server mode, false for client mode
  
      Returns:
          the file descriptor number
      """
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      if server:
          sock.bind((addr, port))
          sock.listen(1)
          clientsock, address = sock.accept()
          return os.dup(clientsock.fileno()), sock
      else:
          sock.connect((addr, port))
          return os.dup(sock.fileno()), sock
  
  class tcp_source(gr.hier_block2):
      def __init__(self, itemsize, addr, port, server=True):
          #init hier block
          gr.hier_block2.__init__(
              self, 'tcp_source',
              gr.io_signature(0, 0, 0),
              gr.io_signature(1, 1, itemsize),
          )
          if not server:
              raise NotImplementedError
          fd, server_socket = _get_sock_fd(addr, port, server)
          watcher = SocketWatcher(server_socket, fd)
          watcher.start()
          self.connect(blocks.file_descriptor_source(itemsize, fd), self)
  
  class tcp_sink(gr.hier_block2):
      def __init__(self, itemsize, addr, port, server=False):
          #init hier block
          gr.hier_block2.__init__(
              self, 'tcp_sink',
              gr.io_signature(1, 1, itemsize),
              gr.io_signature(0, 0, 0),
          )
          fd, _ = _get_sock_fd(addr, port, server)
          self.connect(self, blocks.file_descriptor_sink(itemsize, fd))
  </code></pre>
  
  <p>I do see on the command line:</p>
  
  <pre><code>EOF detected. Closing socket and waiting for new connection...
  got new connection!
  </code></pre>
  
  <p>which suggests that detecting EOF was successful, however, once I reconnect the connection is accepted, but whatever I write to the socket does not appear on the other end. I use a simple GNU radio program to test this, consisting of a TCP source (my version of it), a throttle and a TCP sink.</p>
  
  <pre><code>&lt;?xml version='1.0' encoding='utf-8'?&gt;
  &lt;?grc format='1' created='3.7.8'?&gt;
  &lt;flow_graph&gt;
    &lt;timestamp&gt;Fri Dec 18 14:18:32 2015&lt;/timestamp&gt;
    &lt;block&gt;
      &lt;key&gt;options&lt;/key&gt;
      &lt;param&gt;
        &lt;key&gt;author&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;window_size&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;category&lt;/key&gt;
        &lt;value&gt;Custom&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;comment&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;description&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_enabled&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_coordinate&lt;/key&gt;
        &lt;value&gt;(8, 8)&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_rotation&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;generate_options&lt;/key&gt;
        &lt;value&gt;qt_gui&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;id&lt;/key&gt;
        &lt;value&gt;top_block&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;max_nouts&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;realtime_scheduling&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;run_options&lt;/key&gt;
        &lt;value&gt;prompt&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;run&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;thread_safe_setters&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;title&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
    &lt;/block&gt;
    &lt;block&gt;
      &lt;key&gt;variable&lt;/key&gt;
      &lt;param&gt;
        &lt;key&gt;comment&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_enabled&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_coordinate&lt;/key&gt;
        &lt;value&gt;(8, 160)&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_rotation&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;id&lt;/key&gt;
        &lt;value&gt;samp_rate&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;value&lt;/key&gt;
        &lt;value&gt;32000&lt;/value&gt;
      &lt;/param&gt;
    &lt;/block&gt;
    &lt;block&gt;
      &lt;key&gt;blks2_tcp_sink&lt;/key&gt;
      &lt;param&gt;
        &lt;key&gt;addr&lt;/key&gt;
        &lt;value&gt;127.0.0.1&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;alias&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;comment&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;affinity&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_enabled&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_coordinate&lt;/key&gt;
        &lt;value&gt;(696, 125)&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_rotation&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;id&lt;/key&gt;
        &lt;value&gt;blks2_tcp_sink_0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;type&lt;/key&gt;
        &lt;value&gt;complex&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;server&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;port&lt;/key&gt;
        &lt;value&gt;9001&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;vlen&lt;/key&gt;
        &lt;value&gt;1&lt;/value&gt;
      &lt;/param&gt;
    &lt;/block&gt;
    &lt;block&gt;
      &lt;key&gt;blks2_tcp_source&lt;/key&gt;
      &lt;param&gt;
        &lt;key&gt;addr&lt;/key&gt;
        &lt;value&gt;127.0.0.1&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;alias&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;comment&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;affinity&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_enabled&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_coordinate&lt;/key&gt;
        &lt;value&gt;(344, 136)&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_rotation&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;id&lt;/key&gt;
        &lt;value&gt;blks2_tcp_source_0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;maxoutbuf&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;minoutbuf&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;server&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;type&lt;/key&gt;
        &lt;value&gt;complex&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;port&lt;/key&gt;
        &lt;value&gt;9000&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;vlen&lt;/key&gt;
        &lt;value&gt;1&lt;/value&gt;
      &lt;/param&gt;
    &lt;/block&gt;
    &lt;block&gt;
      &lt;key&gt;blocks_throttle&lt;/key&gt;
      &lt;param&gt;
        &lt;key&gt;alias&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;comment&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;affinity&lt;/key&gt;
        &lt;value&gt;&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_enabled&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_coordinate&lt;/key&gt;
        &lt;value&gt;(536, 112)&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;_rotation&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;id&lt;/key&gt;
        &lt;value&gt;blocks_throttle_0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;ignoretag&lt;/key&gt;
        &lt;value&gt;True&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;maxoutbuf&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;minoutbuf&lt;/key&gt;
        &lt;value&gt;0&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;samples_per_second&lt;/key&gt;
        &lt;value&gt;samp_rate&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;type&lt;/key&gt;
        &lt;value&gt;complex&lt;/value&gt;
      &lt;/param&gt;
      &lt;param&gt;
        &lt;key&gt;vlen&lt;/key&gt;
        &lt;value&gt;1&lt;/value&gt;
      &lt;/param&gt;
    &lt;/block&gt;
    &lt;connection&gt;
      &lt;source_block_id&gt;blks2_tcp_source_0&lt;/source_block_id&gt;
      &lt;sink_block_id&gt;blocks_throttle_0&lt;/sink_block_id&gt;
      &lt;source_key&gt;0&lt;/source_key&gt;
      &lt;sink_key&gt;0&lt;/sink_key&gt;
    &lt;/connection&gt;
    &lt;connection&gt;
      &lt;source_block_id&gt;blocks_throttle_0&lt;/source_block_id&gt;
      &lt;sink_block_id&gt;blks2_tcp_sink_0&lt;/sink_block_id&gt;
      &lt;source_key&gt;0&lt;/source_key&gt;
      &lt;sink_key&gt;0&lt;/sink_key&gt;
    &lt;/connection&gt;
  &lt;/flow_graph&gt;
  </code></pre>