Is there a class accept a socket?

[woodywuuu]

I want to use asyncssh as server in a child process. Parent process use socketserver.ForkingTCPServer.

Like this:

server = socketserver.ForkingTCPServer((host, port), SSHHandler)
server.serve_forever()

class SSHHandler(socketserver.StreamRequestHandler)

    def handle(self):
        # accept a ssh socket, make a ssh connection and do something
        async_ssh_server(socket=self.request)

[ronf]

With socketserver, you wouldn't even have an asyncio event loop running yet, and having one running when trying to fork is probably a bad idea, so I think you'd actually have to wait until after you were in the handle() function of your ForkingTCPServer and create a new event loop for each new connection. Even if you could get this to work, you'll lose almost all the benefits of asyncio with such a design.

Is there any way to replace the use of socketserver in the parent process, where instead of server.serve_forever(), you'd start an event loop with an asyncssh server created in it and then call run_forever() on that event loop? See the example server at https://asyncssh.readthedocs.io/en/latest/#simple-server for what this could look like. That way, at least all of the SSH connections handled on a particular listening port would all run in that single parent process, instead of creating a new child process for each one.

[ghostbody]

It seems like a "single process multiple coroutines" model. Is it possible to make use of multiple processes here?

With socketserver, you wouldn't even have an asyncio event loop running yet, and having one running when trying to fork is probably a bad idea, so I think you'd actually have to wait until after you were in the handle() function of your ForkingTCPServer and create a new event loop for each new connection. Even if you could get this to work, you'll lose almost all the benefits of asyncio with such a design.

Is there any way to replace the use of socketserver in the parent process, where instead of server.serve_forever(), you'd start an event loop with an asyncssh server created in it and then call run_forever() on that event loop? See the example server at https://asyncssh.readthedocs.io/en/latest/#simple-server for what this could look like. That way, at least all of the SSH connections handled on a particular listening port would all run in that single parent process, instead of creating a new child process for each one.

[ronf]

In order to take advantage of multiple processes, you'd need for each process to create its own aysncio event loop after it is forked, and to get maximum benefit from that you'd really want to create a pool of such processes of roughly the size of the number of CPU cores on the system. With the forking socket server, you'll end up with a separate process for each connection, which can lead to performance problems if the number of connections is much larger than the number of available cores.

I must admit that I haven't spent a lot of time trying to get Python asyncio working with a pool of processes. If you truly don't need to share any state across connections, it's actually probably easier to just run multiple instances of your server. You can allow multiple instances to listen on the same port if you add reuse_port=True to the arguments when setting up your AsyncSSH listener, and at least on Linux it seems like connections are automatically spread across your multiple processes when you do that. So, up front you'd fork off some number of processes based on available cores on your system and then for each child process you'd do something similar to the simple server link I listed above, but with reuse_port=True added.

One caveat to this: With I tried this on macOS, it seemed to always prefer the first listener you set up. Other processes can listen on the same port, but until the first listener is closed the other listeners won't get any connections. So, if you need this to work on macOS, a different approach might be required where you pass an accepted socket between processes. There's no portable way to do that in Python, though. Anything you did there would be end up being OS-specific, if it was possible at all.

[ronf]

I spent some time looking into this further, and while I still have my doubts about its usefulness, I've decided to go ahead and provide a sock argument to the top-level AsyncSSH functions which is passed through to the underlying calls to asyncio.create_connection() and asyncio.create_server(), similar to other arguments such as family and flags.

Calls to connect(), connect_reverse(), get_server_host_key() and get_server_auth_methods() should be given a connected socket as a sock argument, while calls to listen() and listen_reverse() should be given a bound socket which is ready to accept new inbound connections.

If you wish to handle incoming connections without having AsyncSSH do the listening and accepting, you should be able to call asyncssh.connect_reverse() with the sock argument of an established inbound TCP connection and have it start up an SSH server running on that established connection. As noted above, you'll need to start a new event loop for each new process you fork, and then call AsyncSSH from within that loop with the already-connected socket.

I still still believe that using ForkingTCPServer could be a problem if there's any chance you'd get a large number of simultaneous inbound connections, but I've decided to provide an option for passing in sock to be consistent with what's available in asyncio, and applications can decide if they want to use it or not.

A first cut of this support is now available in the "develop" branch (commits 0db7e59 and 7ec0e74). If I get a bit more time, I'll try to post a more complete example here.

[ronf]

Here's some example code for starting up an SSH server on an already accepted socket (named sock in this example):

    asyncssh.connect_reverse(
        sock=sock, server_host_keys=['ssh_host_key'],
        authorized_client_keys='ssh_user_ca',
        process_factory=handle_client)

The other arguments are similar to the "simple server" example. You'll need to have a server host private key in 'ssh_host_key' and authorized client keys (or certificates) in 'ssh_user_ca', and then define handle_client() to handle each new incoming SSH session. It take an SSHServerProcess as its only argument.

Here's a more complete example which implements an "echo server" for each new SSH session:

import asyncio, asyncssh, socketserver

async def handle_client(process):
    while True:
        line = await process.stdin.readline()

        if line:
            process.stdout.write(line)
        else:
            break

    process.exit(0)

async def start_server(sock):
    async with await asyncssh.connect_reverse(
            sock=sock, server_host_keys=['ssh_host_key'],
            authorized_client_keys='ssh_user_ca',
            process_factory=handle_client) as conn:
        await conn.wait_closed()

class ConnHandler(socketserver.BaseRequestHandler):
    def handle(self):
        asyncio.run(start_server(self.request))

server = socketserver.ForkingTCPServer((HOST, PORT), ConnHandler)
server.serve_forever()

This will create a separate process for each incoming SSH connection, with its own asyncio event loop. If a client opens multiple SSH sessions on the same SSH connection (such as when using ControlMaster), though, all of those sessions will be handled by the same forked process, in the event loop associated with that process.

[ronf]

To make things a bit more obvious, I've decided to add two new top-level methods to AsyncSSH called run_client() and run_server(). These take a required argument of a connected socket as their first argument and do not accept host, port, or other arguments to connect() and connect_reverse() which are mutually exclusive with passing in sock.

With these new functions, the start_server() method in the above example becomes:

async def start_server(sock):
    async with await asyncssh.run_server(
            sock, server_host_keys=['ssh_host_key'],
            authorized_client_keys='ssh_user_ca',
            process_factory=handle_client) as conn:
        await conn.wait_closed()

[ronf]

These changes are now available in AsyncSSH 2.12.0.

[ghostbody]

That's awesome, man!