code: def _maybe_connect(self, to_pid, callback=None):
      
  
      callback = stack_context.wrap(callback or (lambda stream: None))
  
      def streaming_callback(data):
        # we are not guaranteed to get an acknowledgment, but log and discard bytes if we do.
        log.info('Received %d bytes from %s, discarding.' % (len(data), to_pid))
        log.debug('  data: %r' % (data,))
  
      def on_connect(exit_cb, stream):
        log.info('Connection to %s established' % to_pid)
        with self._connection_callbacks_lock:
          self._connections[to_pid] = stream
        self.__dispatch_on_connect_callbacks(to_pid, stream)
        self.__loop.add_callback(
            stream.read_until_close,
            exit_cb,
            streaming_callback=streaming_callback)
  
      create = False
      with self._connection_callbacks_lock:
        stream = self._connections.get(to_pid)
        callbacks = self._connection_callbacks.get(to_pid)
  
        if not stream:
          self._connection_callbacks[to_pid].append(callback)
  
          if not callbacks:
            create = True
  
      if stream:
        self.__loop.add_callback(callback, stream)
        return
  
      if not create:
        return
  
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
      if not sock:
        raise self.SocketError('Failed opening socket')
  
      stream = IOStream(sock, io_loop=self.__loop)
      stream.set_nodelay(True)
      stream.set_close_callback(partial(self.__on_exit, to_pid, b'reached end of stream'))
  
      connect_callback = partial(on_connect, partial(self.__on_exit, to_pid), stream)
  
      log.info('Establishing connection to %s' % to_pid)
  
      stream.connect((to_pid.ip, to_pid.port), callback=connect_callback)
  
      if stream.closed():
        raise self.SocketError('Failed to initiate stream connection')
  
      log.info('Maybe connected to %s' % to_pid)
docs: Asynchronously establish a connection to the remote pid.