Re: Problem handling 2 simultaneous socket streams

[zentara]

You might get some ideas from testing Simple threaded chat server. It has a simple method for making the chat echo to all clients, and you can probably modify it for your purposes manipulating the @clients list, or splitting @clients into 2, @clients1 and @clients2. It's not bulletproof code, but does demonstrate the flow thru threads.

[PhillyR]

Thanks for the advice. I switched over to using a combination of select and threads for managing the separate incoming streams and it is working properly.
How would I go about adding in reading STDIN to allow a user to type quit to end this program? I have unsuccessfully tried creating a new thread (prior to the while loop below) that monitored STDIN. Once again I was blocked.

#!/usr/bin/perl -w
# -----------------------------
use warnings;
use strict;
use IO::Socket;
use IO::Select;
use threads;
use threads::shared;
# -----------------------------



# -----------------------------
our @stream1_clients : shared;
our @stream2_clients : shared;
my $stream1_prt = 5555;
my $stream2_prt = 5556;
my $client1_prt = 1100;
my $client2_prt = 1101;
my $ip = '192.168.1.12';
# -----------------------------



# Create Stream 1 listen ------
my $stream1_lsn = new IO::Socket::INET (LocalAddr => $ip,
                                     LocalPort => $stream1_prt,
                                     Type      => SOCK_STREAM,
                                     Proto     => "tcp",
                                     Reuse     => 1,
                                     Listen    => 5)
or die "Stream 1 listen socket couldn't be created: $@\n";
# -----------------------------


# Create Client 1 listen ------
my $client1_lsn = new IO::Socket::INET (LocalAddr => $ip,
                                     LocalPort => $client1_prt,
                                     Type      => SOCK_STREAM,
                                     Proto     => "tcp",
                                     Reuse     => 1,
                                     Blocking  => 0, #part of issue?
                                     Listen    => 5)
or die "Client 1 listen socket couldn't be create: $@\n";
# -----------------------------



# Create Client 2 listen ------
my $client2_lsn = new IO::Socket::INET (LocalAddr => $ip,
                                     LocalPort => $client2_prt,
                                     Type      => SOCK_STREAM,
                                     Proto     => "tcp",
                                     Reuse     => 1,
                                     Blocking  => 0, #part of issue?
                                     Listen    => 5)
or die "Client 2 listen socket couldn't be create: $@\n";
# -----------------------------


# Add listen sockets to select
my $sockets = new IO::Select();
$sockets->add($stream1_lsn);
$sockets->add($client1_lsn);
$sockets->add($client2_lsn);
# -----------------------------


#<< SPLIT OFF A SEPARATE THREAD FOR STDIN??


#Go into infinite loop, handling connections - how do I end this?
while(my @ready = $sockets->can_read) { #got data
  foreach my $socket (@ready) { # find which socket sent data

    #New Stream1---------------------
    if ($socket == $stream1_lsn) {
      my $stream1_sock = $stream1_lsn->accept();  #accept stream 1
      $stream1_hst = $stream1_sock->peerhost;
      print "New stream1 from IP [$stream1_hst]\n";
      my $thr_stream1 = threads->new(\&stream1, $stream1_sock)->detach
+();
    }
    # -----------------------------


    #New client1 ------------------
    elsif ($socket == $client1_lsn) {
      my $client1_sock = $client1_lsn->accept();  #accept client 1
      $sockets->add($client1_sock);  #add socket to select
      my $fileno = fileno $client1_sock;
      push (@stream1_clients, $fileno);

      $client1_hst = $client1_sock->peerhost;
      print "New client for stream 1 @ IP [$client1_hst]\n";
    }
    # -----------------------------



    #New client2 ------------------
    elsif ($socket == $client2_lsn) {
      my $client2_sock = $client2_lsn->accept();  #accept client 2
      $sockets->add($client2_sock);  #add socket to select
      my $fileno2 = fileno $client2_sock;
      push(@stream2_clients, $fileno2);

      $client2_hst = $client2_sock->peerhost;
      print "New client for stream 2 @ IP [$client2_hst]\n";

      #Connection to second server running locally
      my $stream2_sock = new IO::Socket::INET(
                                  PeerAddr => 'localhost',
                                  PeerPort => $stream2_prt,
                                  Type     => SOCK_STREAM,
                                  Proto    => "tcp")
      or die "Stream2 socket couldn't be created: $@\n";
  
      my $thr_stream2 = threads->new(\&stream2, $stream2_sock)->detach
+();
    }
    # -----------------------------

  }
}



# Routine called by thread to send stream 1
sub stream1 {
  my ($lclient1) = @_;
  my $buf1;
  if ($lclient1->connected) {
    while(defined($buf1=<$lclient1>) {
      foreach $fn1 (@stream1_clients) {
        open my $fh1, ">&=fn1" or warn $! and die;
        binmode($fh1, ":raw");
        print $fh1 $buf1;
      }
    }
  }
  close ($lclient1);
}
# -----------------------------


# Routine called by thread to send stream 2
sub stream2 {
  my ($lclient2) = @_;
  my $buf2;
  if ($lclient2->connected) {
    while(defined($buf2=<$lclient2>) {
      foreach $fn2 (@stream2_clients) {
        open my $fh2, ">&=fn2" or warn $! and die;
        binmode($fh2, ":raw");
        print $fh2 $buf2;
      }
    }
  }
  close ($lclient2);
}
# -----------------------------


# sub STDIN_stream ??

[zentara]

How would I go about adding in reading STDIN to allow a user to type quit to end this program?

See Re: Ask for STDIN but don't pause for it and the last example. This is untested in your code, but something like this will work. Just add \*STDIN to your Select object and test for it.

# Add listen sockets to select
my $sockets = new IO::Select();
$sockets->add(\*STDIN );
$sockets->add($stream1_lsn);
$sockets->add($client1_lsn);
$sockets->add($client2_lsn);

.....

#Go into infinite loop, handling connections - how do I end this?
while(my @ready = $sockets->can_read) { #got data
  
foreach my $socket (@ready) { # find which socket sent data
     
    if ($socket == \*STDIN) {  
            #read stdin and detect a q;
            exit;  }

[PhillyR]

Adding \*STDIN to the select causes the while loop not to execute...