Re^5: multithreaded tcp listener with IO::Socket

[BrowserUk]

This is a known problem. The solution is to pass the fileno( $socket ) bwtween the threads. As this is just an integer, it bypasses the restrictions on shared variables. Once you get the fileno in the destination thread, you can re-open the socket there and use it in the normal ways.

See Re: FileHandles and threads & Re^3: Passing globs between threads for some background and example code, plus some discussion of the caveats. The main problem with this approach is that you have to ensure that the socket returned by accept() does not go out of scope (causing the socket to be closed), before the thread has had chance to reopen it. That's easily achieved by pushing a copy of the socket handle into a (non-shared) array in the accept thread, as well as queueing it.

However, you then have to arrange for those copies to be cleaned up so that you do not leak handles. The method I've evolved to handle that, involves saving copies of the handles in the accept thread as the values in a non-shared hash, keyed by the fileno. Once the client thread has finished with the thread, it closes it's handle and queues the fileno back to the accept thread which use it to close the original handle as a part of the acceppt loop.

It sounds complicated, but is actually easier to code than describe. I'll clean up one of my test programs and post it later today.

[Random_Walk]

Hi BrowserUk,

I am still having a little trouble with my socket handles. I put the fileno an a queue and the treah picks it up OK. The problem comes down to this line in the handler thread
open my $socket, '<&=', $fileno;

The socket of course is opened read only but I need to write an 'OK' back to it when I have got the event. I tried
open my $socket, '+<&=', $fileno;

still no luck. Any idea how I can re-open it read/write ?

Thanks a million for your help, there is a $favourite_drink waiting for you any time you are in Amsterdam,
R.

[BrowserUk]

You need a dot ('.'), not a comma (',') (and don't forget some error handling :).

open my $socket, '+<&=' . $fileno or die $!;
........................^

For an explaination of the syntax and what it is doing, see perlopentut and the section entitled "Re-Opening Files (dups)".

Sorry for not posting the code I promised. I've been trying to work it into a proper module. It works, but need substantial extra testing and documentation. I'll post it below as it is in it's current state.

The usage will be something like this:

my $server = threads::Server->new(
     LocalPort => 9000,
     Pool => 5,
     Accept => sub {
         printf "Accepted connection from %s:%d on port:%d\n",
            $_->peerhost, $_->peerport, $_->sockport;
         return;
     },
    Thread => sub {
        my( $client, $Qout, @args ) = @_;
        while( <$client> ) {
            chomp;
            ##1 warnf "Got '%s'\n", $_;
            $Qout->enqueue( $_ );
            print $client 'Ack';
        }
    },
    Common => sub {
        my $Q = shift;
        while( $Q->dequeue() ) {
            #1 warnf "Processing '$_'";
        }
    },
);

$server->Run;

That needs explaination, documentation, a lot of work, and relies on another of my own modules (a ripoff of theDamian's Smart::Comments), but it might provide some ideas for you.

Read more... (5 kB)

[txemy]

hello, first of all thanks a lot for posting that code, it has been of great help for me.

Now, I found I need to make some changes for it to work. I'm starting a process to serve http request on a port. Client is doing basically this:

     $ua = LWP::UserAgent->new;
     my $req = HTTP::Request->new(GET =>   
         "http://localhost:8998/foo");
     $req->authorization_basic('fo', 'fi');
     my $res = $ua->request($req) ;

The thing is it looks I really need to close the socket in sub Run after the sub _thread opened the dequeued file_no. . In other words, instead of :

 while( $Qclean->pending ) {
             my $fno = $Qclean->dequeue();
            close delete $self->{ Clients }{ $fno };
        }

I need to do:

my $fno = $Qclean->dequeue();
close delete $self->{ Clients }{ $fno };

before letting it arrive to the "local $_ = $server->accept " again.

If I don't do that,then looks like the "close" of the fd I get in sub _thread does not really closes the socket? The thing is "$ua->request($req)" at the client does not return until I kill the server process if I don't actually close the socket in sub Run.