Echo Server does not work

[marek22k]

Hello,

I would like to build an EchoServer with Concurrent::Async. For this I have the following code:

require 'concurrent-ruby'
require 'socket'

class EchoServer
    include Concurrent::Async

    def initialize(port)
        @server = TCPServer.new('localhost', port)
        @server.do_not_reverse_lookup = true
    end

    def run
        puts "Accepting connections."
        loop do
            puts "Ready for new connection."
            async.handle_connection @server.accept
            puts "Connection delegated."
        end
    end

    def handle_connection(socket)
        puts "Handle new connection."
        _, port, host = socket.peeraddr
        puts "Incoming connection from #{host} port #{port}."
        while ! socket.closed?
            while ! socket.eof?
                IO::copy_stream(socket, socket)
            end
            socket.flush
        end
    end
end

server = EchoServer.new 6666
server.async.run

sleep 10000

However, the server does not work as desired. It accepts incoming connections. However, it never seems to reach the handle_connection function.
The only way I found to make the server work was to write server.run instead of server.async.run.

Is this a bug? Or do I have a thinking error somewhere here? Does anyone have any ideas about this?

[bensheldon]

Hi! Maybe I can help. Concurrent::Async only creates a single thread for tasks, which means that if an async'ed method calls another async'ed method, the second method will not start running until the first method completes.

concurrent-ruby/lib/concurrent-ruby/concurrent/async.rb

Lines 101 to 103 in 1982b92

	# External method calls should *always* use the `async` and `await` proxy
	# methods. When one method calls another method, the `async` proxy should
	# rarely be used and the `await` proxy should *never* be used.

You should not have async.run then subsequently call async.handle_connection because the latter method call will not start executing until the former method call completes. I think you should remove the async. from handle_connection, which will allow async.run to run your server in a background thread.

[marek22k]

Thanks for the explanation! The problem is that if I run handle_connection not asyncron, it is blocking, which means that I can have only one client at a time.

[bensheldon]

If you need more than 1 thread, I don't think Concurrent::Async is the right tool, because it will only ever allocate 1 task thread for your object.

I think what you want is to instantiate a thread pool (or use the global io pool, though it is unbounded) and then create Concurrent::Future / Promises that use that pool.

Something that looks like this (sorry, this is off the top of my head, so it might have some typos):

def initialize
  @thread_pool = Concurrent::FixedThreadPool.new(3)
end

def run
  Concurrent::Future.new(executor: @thread_pool) do
    loop do
      break unless @thread_pool.running?  
      socket = @server.accept
      Concurrent::Future.new(executor: @thread_pool, args: [socket]) do |thr_socket|
        handle_connection(thr_socket)
      end.execute
    end
  end.execute
end

def shutdown
  @thread_pool.shutdown
end

[marek22k]

Thanks, I'll give that a try!