Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.lang.python > #81631 > unrolled thread

Asyncio problem, looking for advice.

Back to article view | Back to comp.lang.python

Contents

  Asyncio problem, looking for advice. Benjamin Risher <brisher777@gmail.com> - 2014-11-27 06:27 -0800
    Re: Asyncio problem, looking for advice. Akira Li <4kir4.1i@gmail.com> - 2014-11-28 15:10 +0300
      Re: Asyncio problem, looking for advice. Benjamin Risher <brisher777@gmail.com> - 2014-11-28 07:51 -0800
        Re: Asyncio problem, looking for advice. Akira Li <4kir4.1i@gmail.com> - 2014-11-29 00:04 +0300

#81631 — Asyncio problem, looking for advice.

Hello all,

I'm working on a project to learn asyncio and network programming.  What I'm trying to do is forward a connection from myself to another machine.  Kind of like an asynchronous python implementation of fpipe.  

In a nutshell:

1 --> start a server listening on localhost
2 --> connect to server 
3 --> server connects to a listening service (ssh for instance)
4 --> server handles both connections to pass traffic back and forth through each

What I have now *kind of* works.  It sends data back and forth, but when I ssh to localhost -p 12345, i never get the password prompt.  It looks like one packet hangs out and doesn't get sent from what I saw in tcpdump.  

Any help would be greatly appreciated.  


Here's a link to the same code as below, just with syntax highlighting etc...
http://pastebin.com/iLE4GZH3


import asyncio
import logging


class MyServer:
    def __init__(self, loop):
        self.server = None
        self.loop = loop
        self.clients = dict()
        self.log = logging.getLogger(__name__)

    def _accept_client(self, client_reader, client_writer):
        """ Client initially drops in here """
        task = asyncio.Task(self._handle_client(client_reader, client_writer))
        self.clients[task] = (client_reader, client_writer)

        def client_done(task):
            self.clients[task][1].close()  # closes the StreamWriter that was part of the task
            del self.clients[task]

        self.log.info("New connection.")
        task.add_done_callback(client_done)

    @asyncio.coroutine
    def _handle_client(self, client_reader, client_writer):
        """ Try to connect to port 22 and broker between 2 connections.

        :param client_reader: StreamReader object
        :param client_writer: StreamWriter object
        """
        reader, writer = yield from asyncio.async(asyncio.open_connection('localhost', 22))

        while True:
            client_data = yield from asyncio.async(self.read_data(client_reader))
            if client_data:
                writer.write(client_data)
                yield from writer.drain()
                
            server_data = yield from asyncio.async(self.read_data(reader))
            if server_data:
                client_writer.write(server_data)
                yield from client_writer.drain()


    @asyncio.coroutine
    def read_data(self, reader):
        data = None

        while True:
            print('top of while')
            data = yield from asyncio.Task(reader.read(2048))
            return data

    def start(self):
        """ start the server listening on 12345 """

        self.server = self.loop.run_until_complete(
            asyncio.streams.start_server(
                self._accept_client,                 # client_connected_callback
                'localhost',                         # host
                12345,                               # port
                loop=self.loop                       # loop
            )
        )

    def stop(self):
        if self.server:
            self.server.close()
            self.loop.run_until_complete(self.server.wait_closed())
            self.server = None

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    my_server = MyServer(loop)
    my_server.start()
    loop.run_forever()
    my_server.stop()

[toc] | [next] | [standalone]

#81677

Benjamin Risher <brisher777@gmail.com> writes:

> Hello all,
>
> I'm working on a project to learn asyncio and network programming.  What I'm trying to do is forward a connection from myself to another machine.  Kind of like an asynchronous python implementation of fpipe.
>
> In a nutshell:
>
> 1 --> start a server listening on localhost
> 2 --> connect to server
> 3 --> server connects to a listening service (ssh for instance)
> 4 --> server handles both connections to pass traffic back and forth through each
>
> What I have now *kind of* works.  It sends data back and forth, but when I ssh to localhost -p 12345, i never get the password prompt.  It looks like one packet hangs out and doesn't get sent from what I saw in tcpdump.
>
> Any help would be greatly appreciated.

Do you want to emulate `ssh -L 12345:localhost:22 <host>`?

> Here's a link to the same code as below, just with syntax highlighting etc...
> http://pastebin.com/iLE4GZH3

There are several issue e.g., unnecessary async(), deprecated Task()
calls but the main issue is that _handle_client() doesn't read
concurrently from the server while client writes to it.

You could use asyncio.wait() to run several tasks in parallel
[1]. Here's a forward-port.py example [2]:

  #!/usr/bin/env python3
  """Forward a local tcp port to host:port.
  
  Usage: %(prog)s local_port:host:port
  
  Example:
  
    $ python3 forward-port.py 26992:icanhazip.com:80 # start server
  
  and in another window:
  
    $ curl localhost:26992 # connect to it
  """
  import asyncio
  import logging
  import sys
  
  info = logging.getLogger('forward-port').info
  
  @asyncio.coroutine
  def copy_stream(reader, writer, bufsize=1<<16):
      while True:
          data = yield from reader.read(bufsize)
          if not data:
              break
          writer.write(data)
          yield from writer.drain()
      writer.close()
  
  def port_forwarder(host, port, *, loop):
      @asyncio.coroutine
      def forward(local_reader, local_writer):
          client = local_writer.get_extra_info('peername')
          info('connected client %s %s', *client)
          remote_reader, remote_writer = yield from asyncio.open_connection(host, port, loop=loop)
          yield from asyncio.wait([copy_stream(local_reader, remote_writer),
                                   copy_stream(remote_reader, local_writer)],
                                  loop=loop)
          info('disconnected client %s %s', *client)
  
      return forward
  
  # main
  logging.basicConfig(level=logging.INFO,
                      format="%(asctime)-15s %(message)s", datefmt="%F %T")
  if len(sys.argv) != 2:
      sys.exit(__doc__)
  local_port, host, port = sys.argv[1].split(':') # e.g., 12345:localhost:22
  
  loop = asyncio.get_event_loop()
  server = loop.run_until_complete(asyncio.start_server(port_forwarder(host, int(port), loop=loop),
                                                        'localhost', int(local_port), loop=loop))
  info('listening on: %s %s', *server.sockets[0].getsockname())
  for closing in range(2):
      try:
          loop.run_until_complete(server.wait_closed())
      except KeyboardInterrupt:
          if not closing:
              server.close()
              info('closing server')
      else:
          break
      info('done')
  loop.close()

[1]
https://docs.python.org/3/library/asyncio-task.html#example-parallel-execution-of-tasks

[2] http://pastebin.com/g08YaJyz


--
Akira

[toc] | [prev] | [next] | [standalone]

#81688

On Friday, November 28, 2014 6:12:20 AM UTC-6, Akira Li wrote:
> Benjamin Risher writes:
> 
> > Hello all,
> >
> > I'm working on a project to learn asyncio and network programming.  What I'm trying to do is forward a connection from myself to another machine.  Kind of like an asynchronous python implementation of fpipe.
> >
> > In a nutshell:
> >
> > 1 --> start a server listening on localhost
> > 2 --> connect to server
> > 3 --> server connects to a listening service (ssh for instance)
> > 4 --> server handles both connections to pass traffic back and forth through each
> >
> > What I have now *kind of* works.  It sends data back and forth, but when I ssh to localhost -p 12345, i never get the password prompt.  It looks like one packet hangs out and doesn't get sent from what I saw in tcpdump.
> >
> > Any help would be greatly appreciated.
> 
> Do you want to emulate `ssh -L 12345:localhost:22 <host>`?
> 
> > Here's a link to the same code as below, just with syntax highlighting etc...
> > http://pastebin.com/iLE4GZH3
> 
> There are several issue e.g., unnecessary async(), deprecated Task()
> calls but the main issue is that _handle_client() doesn't read
> concurrently from the server while client writes to it.
> 
> You could use asyncio.wait() to run several tasks in parallel

> [1]
> https://docs.python.org/3/library/asyncio-task.html#example-parallel-execution-of-tasks
> 
> [2] http://pastebin.com/g08YaJyz
> 
> 
> --
> Akira


Akira, 

First, thank you very much for your response.  It helps tremendously.  I have a question or two though, if you don't mind.  

You said Task() is deprecated, but it's not listed as such in the docs.  Is it just that it's preferred to use other methods instead of using Task() directly, or am I missing something?  

Also, just so I can wrap my head around things, I was trying to mix concurrent and procedural programming in the code I provided, correct?  Do you have any advice on how to avoid mixing types again in the future? 

Thank you again,
Ben

[toc] | [prev] | [next] | [standalone]

#81703

Benjamin Risher <brisher777@gmail.com> writes:

> On Friday, November 28, 2014 6:12:20 AM UTC-6, Akira Li wrote:
>> Benjamin Risher writes:
>> 
>> > Hello all,
>> >
>> > I'm working on a project to learn asyncio and network programming.
>> > What I'm trying to do is forward a connection from myself to
>> > another machine.  Kind of like an asynchronous python
>> > implementation of fpipe.
>> >
>> > In a nutshell:
>> >
>> > 1 --> start a server listening on localhost
>> > 2 --> connect to server
>> > 3 --> server connects to a listening service (ssh for instance)
>> > 4 --> server handles both connections to pass traffic back and forth through each
>> >
>> > What I have now *kind of* works.  It sends data back and forth,
>> > but when I ssh to localhost -p 12345, i never get the password
>> > prompt.  It looks like one packet hangs out and doesn't get sent
>> > from what I saw in tcpdump.
>> >
>> > Any help would be greatly appreciated.
>> 
>> Do you want to emulate `ssh -L 12345:localhost:22 <host>`?
>> 
>> > Here's a link to the same code as below, just with syntax highlighting etc...
>> > http://pastebin.com/iLE4GZH3
>> 
>> There are several issue e.g., unnecessary async(), deprecated Task()
>> calls but the main issue is that _handle_client() doesn't read
>> concurrently from the server while client writes to it.
>> 
>> You could use asyncio.wait() to run several tasks in parallel
>
>> [1]
>> https://docs.python.org/3/library/asyncio-task.html#example-parallel-execution-of-tasks
>> 
>> [2] http://pastebin.com/g08YaJyz
>> 


> Akira, 
>
> First, thank you very much for your response.  It helps tremendously.
> I have a question or two though, if you don't mind.
>
> You said Task() is deprecated, but it's not listed as such in the
> docs.  Is it just that it's preferred to use other methods instead of
> using Task() directly, or am I missing something?

asyncio is a provisional API [3], it may change without notice (though
it shouldn't without a reason). From asyncio docs [4]:

  Don’t directly create Task instances: use the async() function or the
  BaseEventLoop.create_task() method.

The reason is probably to support Trollius (asyncio for Python 2) [5].

> Also, just so I can wrap my head around things, I was trying to mix
> concurrent and procedural programming in the code I provided, correct?
> Do you have any advice on how to avoid mixing types again in the
> future?

In short, the main issue was that your code executed some parts
sequentially e.g., A then B:

  yield from A
  yield from B # this is not reached until A finishes

that needed to be run concurrently:

  yield from asyncio.wait([A, B])

or in general, if you don't need to wait the results:

  asyncio.async(A)
  asyncio.async(B) 

  # ... yield later, to pass the control to the event loop
    (Task._step/_fut_waiter dance ☯)

Ask, if you have any specific questions about the code
http://pastebin.com/g08YaJyz 

There is a bug at the end. info('done') should be outside the loop
(improper indent).

[3] https://www.python.org/dev/peps/pep-0411
[4] https://docs.python.org/3/library/asyncio-task.html#asyncio.Task
[5] https://code.google.com/p/tulip/issues/detail?id=185


--
Akira

[toc] | [prev] | [standalone]

Back to top | Article view | comp.lang.python

csiph-web