Recipe 17.19 Managing Multiple Streams of Input

17.19.1 Problem

The next input to your program could be coming from any number of filehandles, but you don't know which. You've tried using select( ), but the need to then do unbuffered I/O is more than you can deal with (and it's making your code very difficult to follow).

17.19.2 Solution

Use the IO::Multiplex module from CPAN. It calls a mux_input( ) function when input is received over a socket, and handles input and output buffering for you:

use IO::Multiplex;
$mux = IO::Multiplex->new( );
$mux->add($FH1);
$mux->add($FH2); # ... and so on for all the filehandles to manage
$mux->set_callback_object(_ _PACKAGE_ _);  # or an object
$mux->Loop( );

sub mux_input {
  my ($package, $mux, $fh, $input) = @_;
  # $input is ref to the filehandle's input buffer
  # ...
}

17.19.3 Discussion

Although you can use select to manage input coming at you from multiple directions, there are many tricks and traps. For example, you can't use <> to read a line of input, because you never know whether the client has sent a full line yet (or will ever finish sending a line). You can't print to a socket without the risk of the output buffer being full and your process blocking. You need to use non-blocking I/O and maintain your own buffers, and, consequently, life rapidly becomes unmanageably complex.

Fortunately, we have a way of hiding complexity: modules. The IO::Multiplex module from CPAN takes care of non-blocking I/O and select for you. You tell it which filehandles to watch, and it tells you when new data arrives. You can even print to the filehandles, and it'll buffer and non-blockingly output it. An IO::Multiplex object manages a pool of filehandles.

Use the add method to tell IO::Multiplex to manage a filehandle. This enables non-blocking I/O and disables the stdio buffering. When IO::Multiplex receives data on one of its managed filehandles, it calls a mux_input method on an object or class of your choosing. Specify where mux_input is by passing a package name (if your callback is a class method) or object value (if your callback is an object method) to the IO::Multiplex set_callback_object method. In the example in the Solution, we pass in the current package name so that IO::Multiplex will call the current package's mux_input method.

Your mux_input callback is called with four parameters: the object or package name that you gave to set_callback_object, the IO::Multiplex object that dispatched the callback, the filehandle from which data was received, and a reference to the input buffer. The callback should delete data from the buffer once it has been processed. For example, to process line by line:

sub mux_input {
  my ($obj, $mux, $fh, $buffer) = @_;
  my ($line) = $$buffer =~ s{^(.*)\n}{  } or return;
  # ...
}

The IO::Multiplex module also takes care of accepting incoming connections on server sockets. Once you have a socket bound and listening (see Recipe 17.2), pass it to the listen method of an IO::Multiplex object:

use IO::Socket;
$server = IO::Socket::INET->new(LocalPort => $PORT, Listen => 10)
  or die $@;
$mux->listen($server);

When new incoming connections are accepted, the mux_connection callback is called. There are other callbacks, such as for full and partial closure of a filehandle, timeouts, and so on. For a full list of the methods you can use to control an IO::Multiplex object and a full list of the callbacks, see the IO::Multiplex documentation.

Example 17-7 is a rudimentary chat server that uses IO::Multiplex. It listens on port 6901 of the local host address and implements a very rudimentary chat protocol. Every client (see Example 17-8) has a "name," which they can change by sending a line that looks like /nick newname. Every other incoming line of text is sent out to all connected machines, prefaced with the name of the client that sent it.

To test this out, run the server in one window, then start a few clients in other windows. Type something into one and see what appears in the others.

Example 17-7. chatserver

  #!/usr/bin/perl -w
  # chatserver - very simple chat server
  use IO::Multiplex;
  use IO::Socket;
  use strict;
  my %Name;
  my $Server = IO::Socket::INET->new(LocalAddr => "localhost:6901",
                                     Listen   => 10, Reuse => 1,
                                     Proto    => 'tcp') or die $@;
  my $Mux = IO::Multiplex->new( );
  my $Person_Counter = 1;
  $Mux->listen($Server);
  $Mux->set_callback_object(_ _PACKAGE_ _);
  $Mux->loop( );
  exit;
  sub mux_connection {
    my ($package, $mux, $fh) = @_;
    $Name{$fh} = [ $fh, "Person " . $Person_Counter++ ];
  }
  sub mux_eof {
    my ($package, $mux, $fh) = @_;
    delete $Name{$fh};
  }
  sub mux_input {
    my ($package, $mux, $fh, $input) = @_;
    my $line;
    my $name;
    $$input =~ s{^(.*)\n+}{  } or return;
    $line = $1;
    if ($line =~ m{^/nick\s+(\S+)\s*}) {
      my $oldname = $Name{$fh};
      $Name{$fh} = [ $fh, $1 ];
      $line = "$oldname->[1] is now known as $1";
    } else {
      $line = "<$Name{$fh}[1]> $line";
    }
    foreach my $conn_struct (values %Name) {
      my $conn = $conn_struct->[0];
      $conn->print("$line\n");
    }
  }

Example 17-8. chatclient

  #!/usr/bin/perl -w
  # chatclient - client for the chat server
  use IO::Multiplex;
  use IO::Socket;
  use strict;
  my $sock = IO::Socket::INET->new(PeerAddr => "localhost:6901",
                                   Proto    => "tcp") or die $@;
  my $Mux = IO::Multiplex->new( );
  $Mux->add($sock);
  $Mux->add(*STDIN);
  $Mux->set_callback_object(_ _PACKAGE_ _);
  $Mux->loop( );
  exit;
  sub mux_input {
    my ($package, $mux, $fh, $input) = @_;
    my $line;
    $line = $$input;
    $$input = "";
    if (fileno($fh) =  = fileno(STDIN)) {
      print $sock $line;
    } else {
      print $line;
    }
  }

17.19.4 See Also

The documentation for the CPAN module IO::Multiplex; Recipe 17.1, Recipe 17.2, Recipe 17.20, and Recipe 17.21