#!/usr/bin/env perl
#
# $Id: ntred 4558 2011-02-24 17:11:09Z stepanek $
#
# This is ntred - a hub/client for btred servers
#
# NTrEd
# Copyright 2003-2008 Petr Pajas and Zdenek Zabokrtsky, All rights reserved.
# See documentation at the end of program (or run ntred --man)
#

use strict;
use Getopt::Long;
use Pod::Usage;
use IO::Socket;
use IO::Select;
use Cwd;

use vars qw($all_trees $all_nodes $no_all_trees $no_all_nodes
	    $all_nonhidden_nodes $macro_file $allow_no_trees
	    $code $hubname $hubport $init $nocheck $noservers $nohub
	    $client_key_file $verbose $srvlist $kill $ps_kill $ps_break $filelist
	    $srvlistfile $request_quit $request_dump $request_upload
	    $safe_mode $help $usage $manpage $ssh $qsub $run_local $server_debug
	    $maxservers $max_server_retries $request_reload $request_reload_changed
	    $request_load $request_close $glob $reffiles_to_save
	    $request_list $request_list_changed $request_save $request_list_secondary
	    $request_save_changed $strip_prefix $add_prefix
	    $strip_suffix $append_suffix $output_format $term_encoding
	    $config_file $start_context @preload_modules $btred $break
	    $request_reload_macros $quiet $really_quiet %optmap
	    $keep_empty_servers $ntred_config_file
	    $disable_extensions $enable_extensions $request_list_extensions
	    $version $print_version $listedfiles $regexp_files $list_options
            $olddist $count_files $logdir $userlogin $no_secondary $include_macro
	   );

$version='$Revision: 4558 $ ';

$usage=1 unless @ARGV;
$ssh='ssh -o ConnectTimeout=10';
$qsub='qsub';
#$max_server_retries = 10;

use POSIX qw(setlocale);

## Passing -o args... -- to TredMacro
Getopt::Long::Configure ("bundling");
%optmap=("all-trees|T"    => \$all_trees,
	 "all-nodes|N"    => \$all_nodes,
	 "no-all-trees"    => \$no_all_trees,
	 "no-all-nodes"    => \$no_all_nodes,
	 "all-nonhidden-nodes|H"  => \$all_nonhidden_nodes,
	 "help|h"         => \$help,
	 "usage|u"        => \$usage,
	 "man"            => \$manpage,
	 "quiet|q"        => \$quiet,
	 "really-quiet|Q" => \$really_quiet,
	 "macro-file|m=s" => \$macro_file,
	 "include-macro-file|I=s" => \$include_macro,
	 "list-extensions" => \$request_list_extensions,
	 "enable-extensions=s" => \$enable_extensions,
	 "disable-extensions=s" => \$disable_extensions,
	 "execute|e=s"    => \$code,
	 "context|t=s" => \$start_context,
	 "hub=s"          => \$hubname,
	 "port=i"         => \$hubport,
	 "init|i"           => \$init,
	 "qsub=s"           => \$qsub,
	 "ssh=s"           => \$ssh,
	 "local"           => \$run_local,
	 "btred=s"           => \$btred,
	 "kill|k"           => \$kill,
	 "ps-kill"           => \$ps_kill,
	 "break"           => \$break,
	 "ps-break"           => \$ps_break,
	 "quit"           => \$request_quit,
	 "reload-files"    => \$request_reload,
	 "reload-changed-files"    => \$request_reload_changed,
	 "reload-macros"   => \$request_reload_macros,
	 "close-files"     => \$request_close,
	 "load-files"      => \$request_load,
	 "no-secondary-files|Y" => \$no_secondary,
	 "list-files"      => \$request_list,
	 "list-secondary-files"      => \$request_list_secondary,
	 "count-files"      => \$count_files,
	 "list-changed-files" => \$request_list_changed,
	 "save-files"      => \$request_save,
	 "save-changed-files" => \$request_save_changed,
	 "knit|K=s" => \$reffiles_to_save,
	 "dump-files"    => \$request_dump,
	 "upload-file"    => \$request_upload,
	 "strip-prefix|p=s" => \$strip_prefix,
	 "add-prefix|r=s" => \$add_prefix,
	 "strip-suffix|s=s" => \$strip_suffix,
	 "add-suffix|a=s" => \$append_suffix,
	 "output-format|f=s" => \$output_format,
	 "config-file|c=s" => \$config_file,
	 'option|O=s@' => \$TrEd::Config::override_options,
	 "ntred-config-file=s" => \$ntred_config_file,
	 "servers=s"    => \$srvlist,
	 "safe-mode|F" => \$safe_mode,
	 "serverlist=s"    => \$srvlistfile,
	 "max-servers=i"    => \$maxservers,
	 "filelist|l=s"    => \$filelist,
	 "glob|g"          => \$glob,
	 "listed-files|L"    => \$listedfiles,
	 "regexp-files|R=s"    => \$regexp_files,
	 "no-server-check" => \$nocheck,
	 "no-server-start" => \$noservers,
	 "server-debug|D" => \$server_debug,
	 "no-hub-start"   => \$nohub,
	 "key-file=s"     => \$client_key_file,
	 "verbose|V"      => \$verbose,
	 "output-encoding|d=s" => \$term_encoding,
	 "preload-module|M=s" => \@preload_modules,
	 "version|v" => \$print_version,
         "list-options" => \$list_options,
	 "old-dist-method"        => \$olddist,
	 "logdir"        => \$logdir,
	 "keep-empty-servers|E"        => \$keep_empty_servers,
	 "allow-no-trees|0"        => \$allow_no_trees,
	 "max-retries=i"   => \$max_server_retries,
	);
GetOptions(%optmap) or $usage=1;

$ntred_config_file ||="$ENV{HOME}/.ntredrc";
if (-f $ntred_config_file) {
  my %any_opt = map { my $k = $_;
		      my $opts = $_;
		      $opts=~s/=.*//;
		      map { ( $_ => $k ) } split /\|/,$opts
		    } keys %optmap;
  open my $fh, '<',  $ntred_config_file;
  while (<$fh>) {
    chomp;
    s/#.*|^\s*|\s*$//g;
    if (/^([-A-Za-z_]+)\s*=\s*(\S.*)$/) {
      my $key = $any_opt{$1};
      if (defined $key ) {
	if (!defined ${$optmap{$key}}) {
	  ${$optmap{$key}} = $2;
	}
      } else {
	warn "Ignoring unknown option in $ntred_config_file: $1\n";
      }
    } elsif (/\S/) {
      warn "Ignoring bogus line in $ntred_config_file: $_\n";
    }
  }
  close $fh;
}

if ($print_version) {
  my $ver=$version;
  $ver=~s/\$//g;
  print "NTrEd $ver\nPerl: $]\nPlatform: $^O\n";
  exit;
}

($userlogin) = (getlogin() || ($^O ne 'MSWin32') && getpwuid($<) || 'unknown');
($userlogin =~ /\S/) || warn "Couldn\'t deterbmine user\'s login name\n";

$client_key_file ||= "$ENV{HOME}/.ntred_session_key";
$hubport ||= 1500;
$hubname ||= 'localhost'; # no remote connections by default

if ($usage) {
  pod2usage(-msg => 'ntred - controller/hub/client for remote btred servers');
#  exit 0;
}
if ($help) {
  pod2usage(-exitstatus => 0, -verbose => 1);
}
if ($manpage) {
  pod2usage(-exitstatus => 0, -verbose => 2);
}
if ($list_options) {
  foreach my $o (keys %optmap) {
    $o=~s/=.$//;
    foreach my $oo (split /\|/,$o) {
      print sort ((length($oo)>1) ? "--$oo\n" : "-$oo\n");
    }
  }
  exit;
}
if ($really_quiet) {
  $quiet = 1;
  open(STDERR, ">/dev/null") or die "Can't redirect STDERR to /dev/null: $!";
}

#my $IDENT = $init ? "NTRED-HUB($$)" : "NTRED-CLIENT($$)";

sub msg {
  return if $quiet;
  my $msg = join '',@_;
  return if $msg eq '';
  print STDERR (($init ? "NTRED-HUB($$): " : "NTRED-CLIENT($$): "), $msg);
}

# make given list (passes by reference) unique
sub make_uniq ($) { my %U; @{$_[0]} = grep {!exists($U{$_}) && ($U{$_}=1)} @{$_[0]} }

use Encode;

sub sign_md5 ($$) {
  my ($key,$message)=@_;
  Encode::_utf8_off($key);
  Encode::_utf8_off($message);
  TrEd::Cipher::block_md5(TrEd::Cipher::block_xor($key,$message))
}

sub send_request {
  my ($control, $server, $type, $key)=(shift,shift,shift,shift);
  my $request = "";
  return 0 unless $server;

  $server->print("$control REQUEST_TYPE=$type\n");
  $server->flush;
  foreach (@_) {
    chomp; print $server "$_\n";
    $request.="$_\n" if defined $key;
    $server->flush;
  }
  if (defined($key)) {
    $server->print("$control MD5_SIGNATURE ".sign_md5($key,$request)."\n");
  }
  $server->print("$control END_OF_REQUEST\n");
}
sub forward_response {
  my ($control, $server, $client)=@_;
  if (ref($client) eq 'SCALAR') {
    # store response to string
    while (<$server>) {
      if (/^\Q$control\E END_OF_SERVER_RESPONSE/) {
	return 1;
      }
      $$client.=$_;
    }
  } elsif (ref($client)) {
    # forward response to client
    while (<$server>) {
      print $client $_;
      if (/^\Q$control\E END_OF_SERVER_RESPONSE/) {
	$client->flush;
	return 1;
      }
    }
    $client->flush;
  } else {
    # spit response to stderr
    while (<$server>) {
      if (/^\Q$control\E END_OF_SERVER_RESPONSE/) {
	return 1;
      }
      msg "Server replies: $_";
    }
  }
  return 0;
}

sub end_response {
  my ($control,$client)=@_;
  print $client "$control END_OF_HUB_RESPONSE\n";
  $client->flush;
}

use File::Glob qw(:glob);
use FindBin;
my $libDir;
if (exists $ENV{TREDHOME}) {
  $libDir=$ENV{TREDHOME};
} elsif (-d "$FindBin::RealBin/tredlib") {
  $libDir="$FindBin::RealBin/tredlib";
} elsif (-d "$FindBin::RealBin/../lib/tredlib") {
  $libDir="$FindBin::RealBin/../lib/tredlib";
} elsif (-d "$FindBin::RealBin/../lib/tred") {
  $libDir="$FindBin::RealBin/../lib/tred";
}
msg "Trying $libDir\n" if ($libDir);
{
  unshift @INC,$libDir if (-d $libDir);
  my $libs = File::Spec->catfile($libDir,'libs');
  if (-d $libs) {
    push @INC, glob(File::Spec->catfile($libs,'*',''));
  }
}


my $control = '@-NTRED-@-PROTOCOL-@';
require TrEd::Cipher;
import TrEd::Cipher qw(generate_random_block block_to_hex hex_to_block
		       Negotiate Authentify save_block);


do {
  require TrEd::Config;
  $TrEd::Config::quiet=$quiet;
  $TrEd::Config::libDir=$libDir;
};

use File::Spec;
if (not File::Spec->can('rel2abs')) {
  die "The currently installed version of the File::Spec module doesn't provide rel2abs() method. Please upgrade it!\n";
} else {
  require File::Spec::Functions;
  import File::Spec::Functions qw(rel2abs);
}

if (defined $config_file) {
  @TrEd::Config::config_file_search_list=($config_file);
  # override any other possible config files
} else {
  TrEd::Config::set_default_config_file_search_list();
}
my $configFile=TrEd::Config::read_config();

require TrEd::Macros;
import TrEd::Macros;


$TrEd::Macros::defines{NTRED}=1;

sub kill_srv {
  my ($srv,$signal)=@_;
  if ($srv->{pid}) {
    $signal = 'TERM' unless defined($signal);
    #msg "Sending signal $signal to ssh connection to $srv->{host}.$srv->{port} (pid $srv->{pid})\n";
    kill $signal, $srv->{pid};
  } elsif ($srv->{protocol} eq 'qsub') {
    system('qdel',$srv->{fullname});
  }
}

if (defined($init) || defined($kill) || defined($ps_kill) || defined($ps_break) || defined($break)) {
  # get list of servers
  my @servers;

  if ($srvlist ne "") {
    @servers = split /\s*,\s*/,$srvlist;
  } else {
    $srvlistfile ||= $ENV{HOME}."/.ntred_serverlist";
    if (open(my $f, $srvlistfile)) {
      while (<$f>) {
	s/^\s+//;
	s/\s+$//;
	next if (/^#/ || !/\S/);
	chomp; push @servers,$_;
      }
      close $f;
    }
  }

  # --local implies we only want local servers
  if ($run_local) {
    use Sys::Hostname;
    chomp (my $host=hostname());
    @servers = grep { $_ =~ /^\s*(?:localhost|\Q$host\E):\d+\s*$/ } @servers;
    @servers=($host) unless (@servers);
  }
  make_uniq(\@servers);

  die "No servers given!\n" unless @servers;


  if (defined($ps_kill) || defined($ps_break) || defined($kill) || defined($break)) {
  # ==================================================
  # kill servers and start a hub
  # ==================================================
    msg "Killing all servers\n";
    my %hosts;
    my $signal = defined($kill || $ps_kill) ? '-TERM' : '-USR1';
    foreach my $server (@servers) {
      my $protocol = 'ssh';
      if ($server=~s{^\s*([a-z]+)://}{}) {
	$protocol = $1;
      }
      if ($protocol eq 'ssh') {
	my ($host)=split /:/,$server;
	next if $hosts{$host};
	$hosts{$host}=1;
	my $cmd;
	if (defined($ps_kill) or defined($ps_break)) {
	  $btred ||= '[b]tred';
	  $cmd = 'ps x -o pid,command --no-headers |sed "s/^[ \t]*//"|grep -e "'.$btred.'" | cut -f1 -d" " | xargs -r kill '.$signal;
	} else {
	  $btred ||= 'btred';
	  $cmd = "killall $signal $btred";
	}
	my $sh = $run_local ? $cmd : "$ssh '$host' '$cmd'";
	msg "$sh\n";
	system($sh);		#'$btred'
      } elsif ($protocol eq 'qsub') {
	my ($name,$port) = split /:/, $server, 2;
	my @jobs;
	if ($port =~ /(\d+)-(\d+)/) {
	  @jobs = map { "ntred-${name}-$_" } $1..$2;
	} else {
	  @jobs = "ntred-${name}-${port}";
	}
	system('qdel',@jobs);
      } else {
	warn "Unknown server protocol $protocol for server $protocol://$server; ignoring\n";
      }
    }
    msg "Killing hub\n";
    local $SIG{TERM}=sub {}; # protecting myself :-)
    system("killall ntred") if defined($kill);
    msg "Finished\n";
    exit 0;
  }
  # ==================================================
  # run servers and start a hub
  # ==================================================

  my @files = get_file_list();
  die "No files given!\n" unless $keep_empty_servers || @files;

  my $logdir ||= $ENV{NTRED_LOGDIR} || $ENV{TMP} || $ENV{TEMP} || "/tmp";
  my $tmp_dir = File::Temp->newdir("$ENV{HOME}/.ntred_d_XXXXX");

  my @running;
  my %hosts;
  # check that the server machines are actually running
  foreach my $server (@servers) {
    my $protocol = 'ssh';
    $server =~ s{\s+$}{};
    if ($server=~s{^\s*([a-z]+)://}{}) {
      $protocol = $1;
    }
    if ($protocol eq 'qsub') {
      my ($name,$port) = split /:/, $server,2;
      next if exists $hosts{$server};
      my @ports;
      if ($port =~ /(\d+)-(\d+)/) {
	@ports = ($1..$2);
      } else {
	@ports = ($port);
      }
      for my $p (@ports) {
	push @running, { port => $p, fullname => "ntred-${name}-${p}", protocol => $protocol };
	last if ($maxservers>0 && @running >= $maxservers);
      }
    } elsif ($protocol eq 'ssh') {
      my ($host,$port) = split /:/, $server;
      next if $host !~ /\S/;
      unless (exists $hosts{$host}) {
	$hosts{$host}=0;
	$port ||= 1600;
	if ($nocheck or $run_local) {
	  $hosts{$host}=1;
	} else {
	  msg "checking $host\n";
	  my $cmd = "$ssh $host 'sh -c \"echo -n RUNNING 2>/dev/null\"'";
	  msg "$cmd\n";
	  my $result=`$cmd`;
	  if ($result =~ /RUNNING$/) {
	    $hosts{$host}=1;
	    msg "Server $host - running\n";
	  } else {
	    msg "Server $host - down\n";
	    next;
	  }
	}
      }
      push @running, { host => $host, port => $port, fullname => "$host:$port", protocol => $protocol } if $hosts{$host};
    } else {
      warn "Unknown server protocol $protocol for server $protocol://$server; ignoring\n";
    }
    last if ($maxservers>0 && @running >= $maxservers);
  }

  die "No servers available - exit.\n" unless (@running);
  my $server_key;

  if ($noservers)  {
    msg "Expecting SESSION-KEY on stdin\n";
    # Read the server key from STDIN

    $server_key = readline(*STDIN);
    if ($server_key =~ /^\Q$control\E SESSION-KEY=([0-9A-Z]+)$/) {
      $server_key = $1;
      /(.)/; # clean-up :-)
    } else {
      die "Error: no SESSION-KEY given!\n";
    }
  } else {
    # Start servers

    # generate the session key for servers
    $server_key=generate_random_block();

    my $i = 0;
    $SIG{TERM} = sub {
      print STDERR "Caught SIGTERM\n";
      for my $srv (@running) {
	kill_srv($srv,'TERM') if $srv->{running};
      }
      undef $tmp_dir;
    };

    foreach my $srv (@running) {
      $i++;
      msg "Starting btred server [$i] on $srv->{host} port $srv->{port}\n";
      # as for now, we only make use of the secure pipe to the server
      # in order to pass it the session key

      # btred executable
      my $rb=$FindBin::RealBin;
      $rb=~s{/$}{};
      $btred ||= "$rb/btred";

      if ($srv->{protocol} eq 'qsub' or $srv->{port} =~ /-/) {
	$srv->{host_port_file}  = "$tmp_dir/ntred-server.$i";
      }

      my $cmd = "$btred -O -P ".
	  ($run_local ? '--server-host '.$srv->{host}.' ' : ()).
   	  ($safe_mode ? "-F " : "").
	  ($no_secondary ? "-Y " : "").
	  ($reffiles_to_save ne "" ? "-K $reffiles_to_save " : "").
	  ($server_debug ? "-D " : "").
	  (join "",map { "-M $_ " } @preload_modules).
	  ($macro_file ? "-m \"".rel2abs($macro_file)."\" " : "").
	  ($enable_extensions ? qq(--enable-extensions "$enable_extensions " ) : "").
	  ($disable_extensions ? qq(--disable-extensions "$disable_extensions " ) : "").
	  ($include_macro ? "-I \"".rel2abs($include_macro)."\" " : "").
	  ($term_encoding ? "-d $term_encoding " : "").
	  ($srv->{host_port_file} ? qq(--print-host-port '$srv->{host_port_file}' ) : '').
	  "-L $srv->{port} ";
      if ($run_local) {
	$cmd = "$cmd 2> $logdir/ntred-server-$i.log";
      } elsif ($srv->{protocol} eq 'qsub') {
	my $fh;
	my $start_script = qq{$tmp_dir/start_$i.sh};
	print STDERR "creating start script: $start_script\n";
	if (open my $fh, '>', $start_script) {
	  print $fh <<'EOF';
#!/bin/bash\n";
if [ -f $HOME/.bash_profile ]; then
  . $HOME/.bash_profile
elif [ -f $HOME/.profile ]; then
  . $HOME/.profile
fi
EOF
	  print $fh "echo '$control SESSION-KEY=".block_to_hex($server_key)."' | ".$cmd."\n";
	  print $fh "exit 0\n";
	  close $fh;

	  $cmd = "$qsub -N '$srv->{fullname}' -S /bin/bash -cwd -o '$tmp_dir/ntred-server-$i.out' -e '$tmp_dir/ntred-server-$i.log' '$start_script'";
	  print STDERR "$cmd\n";
	  if (system($cmd)==0) {
	    $srv->{running}=1;
	  } else {
	    warn "qsub failed!\n";
	  }
	} else {
	  warn "Cannot create qsub start script: $start_script\n";
	}
	next;
      } else {
	$cmd = "$ssh '$srv->{host}' '$cmd' 2> $logdir/ntred-server-$i.log";
      }
      if ($srv->{protocol} eq 'ssh') {
	# fork in order to avoid getting stuck waiting to write into a dead pipe
	$srv->{pid} = fork;
	if (!defined($srv->{pid})) {
	  die "Fatal error: couldn't fork: $!\n";
	} elsif ($srv->{pid} == 0) {
	  # this child is only responsible for starting the server and
	  # passing it the session key via a pipe
	  msg "$cmd\n";
	  my $ssh_fh;
	  if (open ($ssh_fh,"| ".$cmd)) {
	    $ssh_fh->autoflush(1);
	    $ssh_fh->print("$control SESSION-KEY=".block_to_hex($server_key)."\n");
	    close $ssh_fh;
	    exit 0;
	  } else {
	    exit 1;
	  }
	} else {
	  $srv->{running}=1;
	}
      }
    }
  }
  if ($nohub) {
    print "$control SESSION-KEY=".block_to_hex($server_key)."\n";
    exit;
  }

  # ==================================================
  # Operate as a hub
  # ==================================================
  # --------------------------------------------------------------
  # -------------- prepare socket for clients  -------------------

  msg "Reserving port $hubname:$hubport...\n";
  my $test_socket =  new IO::Socket::INET (LocalHost => $hubname,
					   LocalPort => $hubport,
					   Proto => 'tcp',
					   Listen => 0,
					   Reuse => 1,
					  );
  die $! unless $test_socket;
  msg "Opening connections to servers...\n";
  my $i = 0;
  my $queue_not_ready=0; # this is a countdown for all servers
 SERVER:
  foreach my $srv (@running) {
    $i++;
    my $retries = $max_server_retries || ($srv->{protocol} eq 'qsub' ? 20 : 10);
    next unless $srv->{running};
    my $not_ready = 0;
    msg "Opening connection to $srv->{fullname}\n";
    my $server;
    do {{
      if (!$srv->{stdout} and $srv->{host_port_file}) {
	while (!-f $srv->{host_port_file} or -z $srv->{host_port_file}) {
	  sleep 4;
	  $queue_not_ready++;
	  last if $queue_not_ready>=$retries;
	  msg("server $srv->{protocol}://$srv->{fullname} not yet ready\n");
	}
	my $ok = $queue_not_ready<$retries;
	msg(($ok ? "Found" : "Did not find")." $srv->{host_port_file}\n");
	my $fh;
	if ($ok and open($fh, '<', $srv->{host_port_file})) {
	  msg("Reading $tmp_dir/ntred-server.$i\n");
	  $srv->{stdout} = 1;
	  my $hostport = <$fh>;
	  msg("got $hostport\n");
	  chomp $hostport;
	  if ($hostport=~/^HOST=(.*) PORT=(.*)$/) {
	    $srv->{host} = $1;
	    $srv->{port} = $2;
	    msg("Updating host and port for server [$i]: $srv->{host}:$srv->{port}");
	  }
	  close $fh;
	} else {
	  msg("server [$i] $srv->{fullname} not ready - giving up.\n");
	  kill_srv($srv);
	  $srv->{socket} = undef;
	  $srv->{pid} = undef;
	  next SERVER;
	}
      }
      unless ($srv->{host}) {
	  kill_srv($srv);
	  $srv->{socket} = undef;
	  $srv->{pid} = undef;
	  next SERVER;
      }

      $server= new IO::Socket::INET (PeerAddr => $srv->{host},
				     PeerPort => $srv->{port},
				     Proto => 'tcp',
				    );
      if ($server and $server->connected) {
	unless (Negotiate($server_key,$server,$control)) {
	  warn "Authentification to $srv->{fullname} has been denied - exiting.\n".
	    "Hint: kill all servers with ntred --kill or ntred --ps-kill and try again.\n";
	  for (@running) {
	    kill_srv($_);
	  }
	  exit 1;
	}
      } 
      elsif ($not_ready>=$retries) {
	msg("server [$i] $srv->{fullname} not ready - giving up.\n");
	kill_srv($srv);
	$srv->{socket} = undef;
	$srv->{pid} = undef;
	next SERVER;
      }
      else {
	msg("$srv->{fullname} at $srv->{host}:$srv->{port} not ready\n");
	$not_ready++;
	sleep 2;
      }
    }} while (!$server);
    $srv->{socket} = $server;
    msg  " - connected to $srv->{fullname}\n";
  }
  @running = grep { defined($_->{socket}) } @running;
  unless (@running) {
    die "Couldn't connect to any btred server - exiting.\n";
  }
  msg "Connected to btred servers on the following hosts:\n";
  foreach  (@running) {
    msg " - $_->{fullname}\n"
  }

  if ($olddist) {
    # Benchmark the servers
    foreach my $server (@running) {
      send_request($control,$server->{socket},'BENCHMARK',$server_key);
    }
    foreach my $server (@running) {
      my $response;
      unless (forward_response($control,$server->{socket},\$response)) {
	msg "Communication with $server->{fullname} seems broken; skipping.\n";
	next;
      }
      chomp $response;
      msg "$server->{fullname} benchmark: $response\n";
      $server->{benchmark}=$response;
    }
  }

  # --------------------------------------------------------------
  # --------------- optimal file distribution  -------------------
  use Benchmark;
  my $t0 = Benchmark->new();
  my $state_flags = 
    [
      ($no_secondary ? ("$control STATE=NO_SECONDARY",1) : ()),
      ($allow_no_trees ? ("$control STATE=ALLOW_NO_TREES",1) : ())
    ];

  if ($olddist) {
    distribute_files($control,\@running,\@files,($keep_empty_servers ? 0 : 1),
		     $server_key,
		     $state_flags,
		    );
  } else {
    distribute_files2($control,\@running,\@files,($keep_empty_servers ? 0 : 1),
		      $server_key,
		      $state_flags,
		     );
  }
  my $t1 = Benchmark->new();
  msg ("Distribution took ",Benchmark::timestr(Benchmark::timediff($t1,$t0))."\n");
  # save client key
  save_block($server_key,$client_key_file);

  # --------------------------------------------------------------
  # -------------- prepare socket for clients  -------------------

  msg "Preparing client connection...\n";
  close $test_socket;
  my $listening_socket =  new IO::Socket::INET (LocalHost => $hubname,
						LocalPort => $hubport,
						Proto => 'tcp',
						Listen => 5,
						Reuse => 1,
					       );
  die $! unless $listening_socket;
  msg "Ready for client requests on $hubname:$hubport\n\n";

  # -------------------------------------------------------------------
  # ------- main cycle: processing requests from clients --------------

  my $request_number;

  sub sigpipe {
    msg "Communication to client interrupted (SIGPIPE).\n";
  }
  $SIG{PIPE} =\&sigpipe;

  while (my $client = $listening_socket->accept()) {
    unless (Authentify($server_key,$client,$control)==1) {
      $client->close();
      next;
    }
    #  $newclient->timeout(5);
    $request_number++;

    # ---------- reading the request ----------------
    msg "Receiving request $request_number from a client...\n";
    my $request;
    my $md5_request;
#     my $request_type;
#     while (<$client>) {		#
#       $request.=$_;
#       if (/^\Q$control\E REQUEST_TYPE=(.*)/) {
# 	$request_type=$1;
#       } elsif (/^\Q$control\E END_OF_REQUEST/) {
# 	last;
#       }
#     }

    my %request;
    my $state;
    # --- reading request from hub ---
    while (<$client>) {
      $request.=$_;
      if (/^\Q$control\E REQUEST_TYPE=(.*)/) {
	%request = (); # clear
	$request{REQUEST_TYPE}=$1;
      } elsif (/^\Q$control\E MD5_SIGNATURE (.*)$/) {
	$request{SIGNATURE} = $1;
      } elsif (/^\Q$control\E STATE=(.*)/) {
	$md5_request.=$_;
	$state=$1;
	$request{"STATE_".$state}.="";
      } elsif (!/^\Q$control\E/) {
	$md5_request.=$_;
	$request{"STATE_".$state}.=$_;
      } elsif (/\Q$control\E END_OF_REQUEST/) {
	$state='END_OF_REQUEST';
	last
      }
    }

    my $t0 = new Benchmark;
    if ($request{REQUEST_TYPE} eq 'QUIT') {
      $SIG{PIPE} = sub {};
    }

    if ($request{REQUEST_TYPE} eq 'HUB_DISTRIBUTE') {
      @files = split /\n/,$request{STATE_FILELIST};
      my $md5 = sign_md5($server_key,$md5_request);
      if ($md5 eq $request{SIGNATURE}) {
	#shift @files; pop @files; # get rid of the controls
	my $state_flags = 
	  [
	    ($request{STATE_NO_SECONDARY} ? ("$control STATE=NO_SECONDARY",1) : ()),
	    ($request{STATE_ALLOW_NO_TREES} ? ("$control STATE=ALLOW_NO_TREES",1) : ())
	   ];

	if ($olddist) {
	  distribute_files($control,\@running,\@files,0,$server_key,$state_flags);
	} else {
	  distribute_files2($control,\@running,\@files,0,$server_key,$state_flags);
	}
      } else {
	print $client "$control HUB_RESPONSE_STDERR\n";
	print $client "Request MD5 signature failed!\n";
      }
      end_response($control,$client);
    } else {
    #print STDERR "Request: $request\n\n";
    # ---------- distributing the request -------------
      msg "Distributing the request to the servers...\n";
      my $server_no = 1;
      foreach my $server (@running) {
	next unless $server->{socket};
	msg " - to $server->{fullname}\n";
	$server->{socket}->print("$control SERVER_NO=$server_no\n" );
	$server->{socket}->print($request);
	$server->{socket}->flush;
	$server_no ++;
      }

      # ----------- reading and forwarding the response ---------------
      msg "Waiting for server responses and forwarding them to the client...\n";
      foreach my $server (@running) {
	next unless $server->{socket};
	msg " - from $server->{fullname}\n";
	print $client "$control Response from $server->{fullname}\n";
	unless (forward_response($control,$server->{socket},$client)) {
	  print $client "$control HUB_RESPONSE_STDERR\n";
	  print $client "Warning: server $server->{fullname} seems down!\n";
	  msg "Warning: server $server->{fullname} seems down!\n";
	}
      }
      end_response($control,$client);
      msg "The processing of request $request_number completely finished.\n";
      my $t1 = new Benchmark;
      my $time = timediff($t1, $t0) if ($t1 and $t0);
      msg("Request processed in ",timestr($time),"\n");

      msg "Waiting for a new client request.\n\n";
      if ($request{REQUEST_TYPE} eq 'QUIT') {
	msg "Quitting on client's request.\n";
	exit;
      }
    }
  }
  msg "Closing hub.\n";
} else {
  # ==================================================
  # Operate as a client
  # ==================================================

  $term_encoding = 'utf8' if !defined($term_encoding) and setlocale('LC_CTYPE')=~/UTF-?8/i;

  # Read the session key for authentification to hub
  my $key;
  do {
    open my $tmp, $client_key_file;
    $key = readline($tmp); chomp $key;
    $key = hex_to_block($key);
    close $tmp;
  };

  my $control='@-NTRED-@-PROTOCOL-@';	    # @@@ marks the beginning of control lines

  # ---------- composing and testing the request -----------

  if ($macro_file || $include_macro) {
    #msg("Reading macros from $macro_file ...\n");
    $TrEd::Macros::defaultMacroFile='/dev/null'; # don't read this one
    #$TrEd::Config::macroDebug=1;
    TrEd::Macros::read_macros(rel2abs($macro_file),$libDir) if $macro_file; # read macros
    TrEd::Macros::read_macros(rel2abs($include_macro),$libDir) if $include_macro; # read macros
    #msg("done.\n");
    if ($TrEd::Macros::exec_code=~/^\S*[nb]tred\s(.*)$/) {
      local @ARGV=split /\s+/,$1;
      msg "Applying macro options: @ARGV\n";
      local $optmap{"all-trees|T"} if $no_all_trees;
      local $optmap{"all-nodes|N"} if $no_all_nodes;
      local $optmap{"all-nonhidden-nodes|H"} if $no_all_nodes;
      GetOptions (map {$_ => $optmap{$_}}
		  "all-trees|T",
		  "all-nodes|N",
		  "all-nonhidden-nodes|H",
		  $code ? () : ("execute|e=s"),
		  $start_context ? () : ("context|t=s"),
		 );
    }
  }

  $code ||= 'autostart()';

  if ($all_nonhidden_nodes) {
    $code = "while (\$this) { $code ; \$this=\$this->following_visible(\$grp->{FSFile}->FS)}";
  } elsif ($all_nodes) {
    $code = "while (\$this) { $code ; \$this=\$this->following }";
  }
  if ($all_trees) {
    $code = "if (\$root) { do {{ $code }} while TredMacro::NextTree() }";
  }

  $code =
    qq(\n#line 1 ").fastcwd().qq(/{ntred -e}"\n)
    .$code;

  my $request="";
  my $request_type;
  if ($request_quit) {
    $request_type="QUIT";
  } elsif ($request_reload_changed) {
    $request_type="RELOAD_CHANGED";
    $request="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
    $request.="$control STATE=ALLOW_NO_TREES\n1\n" if ($allow_no_trees);
  } elsif ($request_reload) {
    $request_type="RELOAD";
    my @files =  get_files();
    foreach (@files) {
      die "Error: File name contains tree suffix: $_\n" if
	/(##?[0-9A-Z]+(?:\.[0-9]+)?)$/;
    }
    $request.="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
    $request.="$control STATE=ALLOW_NO_TREES\n1\n" if ($allow_no_trees);
    $request.="$control STATE=REGEXP_FILES\n"._enc2bytes('utf8',$regexp_files)."\n" if ($regexp_files);
    $request.="$control STATE=FILELIST\n"._enc2bytes('utf8',join "\n",@files)."\n" if ($listedfiles || @files);
  } elsif ($request_reload_macros) {
    $request_type="RELOAD_MACROS";
    if ($macro_file) {
      $request="$control STATE=Filename\n".$macro_file."\n";
    }
    if ($enable_extensions) {
      $request="$control STATE=ENABLE_EXTENSIONS\n".$enable_extensions."\n";
    }
    if ($disable_extensions) {
      $request="$control STATE=DISABLE_EXTENSIONS\n".$disable_extensions."\n";
    }
  } elsif ($request_list_extensions) {
    $request_type="LIST_EXTENSIONS";
  } elsif ($count_files) {
    $request_type="COUNT";
  } elsif ($request_list) {
    $request_type="LIST";
    $request.="$control STATE=REGEXP_FILES\n"._enc2bytes('utf8',$regexp_files)."\n" if ($regexp_files);
    if ($listedfiles) {
      my @files =  get_files();
      foreach (@files) {
	die "Error: File name contains tree suffix: $_\n" if
	  /(##?[0-9A-Z]+(?:\.[0-9]+)?)$/;
      }
      $request.="$control STATE=FILELIST\n"._enc2bytes('utf8',join "\n",@files)."\n";
    }
  } elsif ($request_list_secondary) {
    $request_type="LIST_SECONDARY";
  } elsif ($request_list_changed) {
    $request_type="LIST_CHANGED";
    $request="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
  } elsif ($request_save || $request_save_changed) {
    $request_type= $request_save ? "SAVE" : "SAVE_CHANGED";
    $request="$control STATE=Format\n".($output_format || "fs")."\n" if ($output_format ne "");
    $request.="$control STATE=Strip-prefix\n".$strip_prefix."\n" if ($strip_prefix ne "");
    $request.="$control STATE=Strip-suffix\n".$strip_suffix."\n" if ($strip_suffix ne "");
    $request.="$control STATE=Prefix\n".$add_prefix."\n" if ($add_prefix ne "");
    $request.="$control STATE=Suffix\n".$append_suffix."\n" if ($append_suffix ne "");
    $request.="$control STATE=Knit\n".$reffiles_to_save."\n" if ($reffiles_to_save ne "");
    $request.="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
    $request.="$control STATE=REGEXP_FILES\n"._enc2bytes('utf8',$regexp_files)."\n" if ($regexp_files);
    msg($request,"\n");
    if ($listedfiles) {
      my @files =  get_files();
      foreach (@files) {
	die "Error: File name contains tree suffix: $_\n" if
	  /(##?[0-9A-Z]+(?:\.[0-9]+)?)$/;
      }
      $request.="$control STATE=FILELIST\n"._enc2bytes('utf8',join "\n",@files)."\n";
    }
  } elsif ($request_close) {
    $request_type="CLOSE";
    $request.="$control STATE=REGEXP_FILES\n"._enc2bytes('utf8',$regexp_files)."\n" if ($regexp_files);
    if ($listedfiles) {
      my @files =  get_files();
      foreach (@files) {
	die "Error: File name contains tree suffix: $_\n" if
	  /(##?[0-9A-Z]+(?:\.[0-9]+)?)$/;
      }
      $request.="$control STATE=FILELIST\n"._enc2bytes('utf8',join "\n",@files)."\n";
    }
  } elsif ($request_load) {
    $request_type="HUB_DISTRIBUTE";
    $request.="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
    $request.="$control STATE=ALLOW_NO_TREES\n1\n" if ($allow_no_trees);
    $request.="$control STATE=FILELIST\n";
    my @files = get_file_list();
    $request.=join "\n", @files;
  } elsif ($request_dump) {
    $request_type="DUMP";
    msg "DUMP @ARGV\n";
    my @files = get_file_list();
    $request=join("\n", @files)."\n";
    $request.="$control STATE=format\nStorable-Base64\n";
    $request.="$control STATE=NO_SECONDARY\n1\n" if ($no_secondary);
  } elsif ($request_upload) {
    die "Too few parameters for --upload!" unless @ARGV;
    die "Too many parameters for --upload!" if @ARGV>1;
    $request_type="UPLOAD";
    $request.="$control STATE=Filename\n".$ARGV[0]."\n";
    $request.="$control STATE=FSFile\n";
    my @fs=<STDIN>;
    $request.=join("",@fs)."\n";
  } else {
    $request_type="EVAL_CODE";
    my @files =  get_files();

    msg "ARGS: @ARGV\n";
    $request="$control STATE=ENCODING\n"._enc2bytes('utf8',$term_encoding)."\n" if $term_encoding;
    $request.="$control STATE=CODE\n"._enc2bytes('utf8',_enc2utf8($term_encoding || 'utf8',$code))."\n";
    $request.="$control STATE=QUIET\n1\n" if $quiet;
    $request.="$control STATE=CONTEXT\n"._enc2bytes('utf8',$start_context)."\n";
    $request.="$control STATE=REGEXP_FILES\n"._enc2bytes('utf8',$regexp_files)."\n" if ($regexp_files);
    $request.="$control STATE=FILELIST\n"._enc2bytes('utf8',join "\n",@files)."\n" if ($listedfiles || @files);
    $request.="$control STATE=Strip-prefix\n".$strip_prefix."\n" if ($strip_prefix ne "");
    $request.="$control STATE=Strip-suffix\n".$strip_suffix."\n" if ($strip_suffix ne "");
    $request.="$control STATE=Prefix\n".$add_prefix."\n" if ($add_prefix ne "");
    $request.="$control STATE=Suffix\n".$append_suffix."\n" if ($append_suffix ne "");
    $request.="$control STATE=MACRO\n".
      _enc2bytes("utf8",join("",@TrEd::Macros::macros)."\n");
    $request.="$control STATE=ARGS\n"._enc2bytes('utf8',_enc2utf8($term_encoding || 'utf8',join("\n",@ARGV)))."\n";
    msg "Code: $code\n\n";
  }

  # ---------- open hub connection ----------------
  msg "Opening connection to ntred-hub on $hubname:$hubport\n";
  my $hub= new IO::Socket::INET (PeerAddr => $hubname,
				 PeerPort => $hubport,
				 Proto => 'tcp'
				);
  die "Can't open socket to $hubname:$hubport. $!\n" unless $hub;
  $hub->autoflush(1);
  $hub->timeout(5);

  msg "Negotiating with the hub\n";

  die "Authentification to $hubname:$hubport has been denied.\n"
    unless Negotiate($key,$hub,$control);

  # --------- sending the request to the hub ---------------
  msg "Sending the request to the hub...\n";
  send_request($control,$hub,$request_type,$key,split /\n/,$request);
  # --------- receiving the response from the hub -----------
  msg "Waiting for responses forwarded by the hub...\n";
  my $first=1;
  my $real_response=0;
  my $out=\*STDERR;

  while (<$hub>) {
    if ($first) {
      msg "Receiving the server responses and printing it to STDOUT...\n";
      $first=0;
    }
    if (!/^\Q$control\E/o) {
      if ($real_response) {
	print $out $_;
      } else {
	msg "Hub send: $_";
      }
    } elsif (/^\Q$control\E Response from (\S+)/o) {
      msg "Receiving response from $1\n";
      $real_response=1;
    } elsif (/^\Q$control\E HUB_RESPONSE_STDERR/o) {
      $out = \*STDERR;
    } elsif (/^\Q$control\E SERVER_RESPONSE_STDOUT/o) {
      $out = \*STDOUT;
    } elsif (/^\Q$control\E SERVER_RESPONSE_STDERR/o) {
      $out = \*STDERR;
    } elsif (/^\Q$control\E SERVER_RESPONSE_STDOUT/o) {
      $out = \*STDOUT;
    } elsif (/^\Q$control\E END_OF_HUB_RESPONSE/) {
      last;
    } else {
      print STDERR $_ unless $quiet;
    }
  }

  msg "Done.\n";
}

# --------------------------------------------------------------
sub _enc2utf8 {
  my ($enc,$str)=@_;
  return $str if ($enc eq "" or $]<5.008);
  require Encode;
  return Encode::decode($enc,$str);
}

sub _enc2bytes {
  my ($enc,$str)=@_;
  return $str if ($enc eq "" or $]<5.008);
  require Encode;
  return Encode::encode($enc,$str);
}

# --------------- optimal file distribution  -------------------
sub distribute_files {
  my ($control,$servers,$files,$close_empty,$key,$state_flags) = @_;
  my @totalfilelist=@$files;
  my %benchmark_time;
  my %number_of_procs;
  my %power;

  my $totalpower;
  foreach my $server (@$servers) {
    ($benchmark_time{$server}, $number_of_procs{$server})=split /,/,$server->{benchmark};
    $power{$server}=$number_of_procs{$server}/($benchmark_time{$server}+1);
    $totalpower+=$power{$server};
  }

  my (%files,%filescnt);
  my $totalfilecnt=@totalfilelist;
  my $remaining_power=$totalpower;
  my $remaining_files=$totalfilecnt;
  foreach my $server (sort {$power{$a}<=>$power{$b}} @$servers) {
    $filescnt{$server}=
      sprintf("%0.0f", $remaining_files*$power{$server}/$remaining_power);
    $remaining_power-=$power{$server};
    $remaining_files-=$filescnt{$server};
  }

  # Distribute files
  my @servers_to_respond;
  foreach my $server (@$servers) {
    my @f=splice @totalfilelist,0,$filescnt{$server};
    if (@f) {
      push @servers_to_respond,$server;
      msg "Distributing ".scalar(@f)." files to $server->{fullname}\n";
      send_request($control,$server->{socket},'LOAD_FILES',$key,
		   @$state_flags,
		   "$control STATE=FILELIST", @f);
    } elsif ($close_empty) {
      msg "No files for server $server->{fullname}, closing it.\n";
      send_request($control,$server->{socket},'QUIT',$key,@f);
    }
  }
  msg "Please wait ...\n";
  # get responses
  foreach my $server (@servers_to_respond) {
    my $response;
    unless (forward_response($control,$server->{socket},\$response)) {
      msg "Communication with $server->{fullname} seems broken\n";
    }
    msg "$server->{fullname} $response\n";
  }
  @$servers = @servers_to_respond if $close_empty;
}


sub distribute_files2 {
  my ($control,$servers,$files,$close_empty,$key,$state_flags) = @_;
  my @totalfilelist=@$files;

  my $files_per_request = 16;
  $files_per_request /= 2 while ($files_per_request>=2 and
				 @$servers * $files_per_request > @totalfilelist);

  require IO::Select;
  my $select = IO::Select->new();
  # Distribute files
  my @servers_to_respond;
  my $all_files_count = @totalfilelist;
  my $distributed_files = 0;
  my $success = 0;
  foreach my $server (@$servers) {
    my @f=splice @totalfilelist,0,$files_per_request;
    $distributed_files += @f;
    $server->{last_batch} = @f;
    if (@f) {
      push @servers_to_respond,$server;
      msg "Distributing ".scalar(@f)." files to $server->{fullname} ".
	"(".int((100*$distributed_files)/$all_files_count)."%)".
	"\n";

      send_request($control,$server->{socket},'LOAD_FILES',$key,
		   @$state_flags,
		   "$control STATE=FILELIST", @f);

      $select->add($server->{socket});
    } elsif ($close_empty) {
      msg "No files for server $server->{fullname}, closing it.\n";
      send_request($control,$server->{socket},'QUIT',$key,@f);
      my $response;
      unless (forward_response($control,$server->{socket},\$response)) {
	msg "Communication with $server->{fullname} seems broken\n";
      }
      chomp $response;
      msg "$server->{fullname} $response\n";
    } else {
      print "Noop: $server->{fullname}\n";
    }
  }
  @$servers = @servers_to_respond if $close_empty;
  msg "Please wait ...\n";
  # get responses
  my @can_read;
  while (@can_read = $select->can_read()) {
    $files_per_request /= 2 while ($files_per_request>=2 and
				   @can_read * $files_per_request > @totalfilelist);
    foreach my $socket (@can_read) {
      my $response;
      my ($server) = grep { $_->{socket} eq $socket } @$servers;
      unless (forward_response($control,$socket,\$response)) {
	msg "Communication with $server->{fullname} seems broken\n";
      }
      chomp $response;
      msg "$server->{fullname} $response\n";
      $success+=$server->{last_batch} if ($response eq 'OK');
      my @f=splice @totalfilelist,0,$files_per_request;
      $distributed_files += @f;
      $server->{last_batch}= @f;
      if (@f) {
	msg "Distributing ".scalar(@f)." files to $server->{fullname} ".
	  "(".int((100*$distributed_files)/$all_files_count)."%)".
	  "\n";
	send_request($control,$server->{socket},'LOAD_FILES',$key,
		     @$state_flags,
		     "$control STATE=FILELIST", @f);


      } else {
	$select->remove($socket);
      }
    }
  }
  if (@totalfilelist) {
    msg "Error: ".scalar(@totalfilelist)." files left in the filelist ...oooops?\n";
  } else {
    if ($success == $all_files_count) {
      msg "All $all_files_count files have been successfully loaded :-)\n";
    } else {
      msg(("*"x75)."\n");
      msg "PROBLEM: servers reported errors/warnings while loading some files.\n";
      msg "Of $all_files_count files loaded, only $success were confirmed OK.\n";
      msg "Note that OK status is reported per-batch, not per-file, so more files may be OK.\n";
      msg "Anyway, up to ".($all_files_count-$success)." files had problems.\n";
      msg(("*"x75)."\n");
    }
  }
}

sub r2a {
  my ($fl,$file)=@_;
  if (defined($fl)) {
    return Treex::PML::ResolvePath($fl,$file);
  } else {
    return rel2abs($file);
  }
}

sub get_file_list {
  my @files;
  if (defined($filelist) or !scalar(@ARGV)) {
    $filelist ||= $ENV{HOME}."/.ntred_filelist";
    if (open(my $f, $filelist)) {
      my $fl = (($filelist =~ /\.fl$/i) ? Cwd::abs_path($filelist) : undef );
      my $first = 1;
      while (<$f>) {
	s/^\s+//;
	s/[\n\r]+$//;
	next if /^#/ or ($first and defined($fl) and !m{[/.]} and ! -f r2a($fl,$_));
	$first = 0;
	push @files, (m{^/|^[[:alnum:]]+://} ? $_ : r2a($fl,$_));
      }
      close $f;
    }
  }
  @files = (@files, map { m{^/|^[[:alnum:]]+://} ? $_ : rel2abs($_) }
	      map { $glob ? glob($_) : $_ } @ARGV);
  make_uniq(\@files);
  return @files;
}


sub get_files {
  my @files;
  if ($filelist) {
    local $_;
    if (open(my $f, $filelist)) {
      my $fl = (($filelist =~ /\.fl$/i) ? Cwd::abs_path($filelist) : undef );

      my $first = 1;
      while (<$f>) {
	s/^\s+//;
	s/[\n\r]+$//;
	next if /^#/ or ($first and defined($fl) and !m{[/.]} and ! -f r2a($fl,$_));
	$first = 0;
	s{^ntred://}{};
	s{@(\d+)##?1(\.\d+)?$}{##$1$2};
	push @files, (m{^/|^[[:alnum:]]+://} ? $_ : r2a($fl,$_));
      }
      close $f;
    }
  }
  if ($listedfiles) {
    local $_;
    while (@ARGV) {
      $_=shift @ARGV;
      last if ($_ eq '--');
      s{^ntred://}{};
      s{@(\d+)(?:##?1(\.\d+)?)?$}{##$1$2};
      push @files, (m{^/|^[[:alnum:]]+://} ? $_ : rel2abs($_));
    }
  }
  return @files;
}


__END__

=head1 ntred

ntred - controller/hub/client interface to a cluster of btred servers

=head1 SYNOPSIS

To query the servers:

  ntred [-m|-I macro-file] [-e code] [--hub hub] [--port port]
        [-N|-H] [-T] [--key-file file] [--filelist file-list ] [-L files]
        -- script-arguments

To start remote servers and a hub:

  ntred -i [--servers server[,server,...]] [--serverlist server-list]
           [--filelist file-list] [--max-servers num] 
           [--allow-no-trees] [--no-secondary-files]
           [--no-server-check] [--no-server-start] [--no-hub-start]
           [--old-dist-method] [--safe-mode] [--server-debug]
           [-m|-I macro-file] [--disable-extensions list] [--enable-extensions list]
           [-M module] [--btred path-to-btred]
           [--ssh ssh-command] [--local]
           [--key-file] [--hub hub] [--port port] [file [...]]

To close all remote servers and a hub:

  ntred --quit|--ps-quit

To kill all remote servers and a hub:

  ntred --kill|--ps-kill

To manage files on the servers:

  ntred --list-files|--list-changed-files

  ntred --reload-files [--disable-extensions list] [--enable-extensions list]

  ntred --reload-macros [-m macro-file]

  ntred --load-files [--filelist file-list] [file [...]]

  ntred --close-files

  ntred --save-files|--save-changed-files [-s strip-sfx]
       [-a append-sfx] [-p strip-prefix] [-r add-prefix] [-f out-fmt]

  ntred --dump-files [--filelist file-list] [file [...]]

  ntred --upload-file filename < fs-file

Get help:

  ntred -u          for usage (synopsis)
  ntred -h          for help
  ntred --man       for the manual page

=head1 DESCRIPTION

This program is able to start one or more btred servers on a set of
host machines over SSH, create a proxy hub to provide the
communication between the servers and a client, distribute given files
over the servers (provided the servers are able to load the files from
given filenames (eg. they share the files over NFS), query the servers
using a btred-macro and collect the answers.  It is highly recommended
to use some password-free authentication method for SSH (e.g. Kerberos
or C<ssh-agent>), so that password input is not required each time the
SSH connection is made (at least two per-host).

In the client mode, the standard output of the macro is printed to
STDOUT of the client. STDERR is reserved for debugging and information
messages as well as error messages caused by the macros on the
servers. The rest of the error output from a server is stored in a
file <logdir>/ntred-server-<host>.log (where <logdir> can be specified
using C<--logdir>, NTRED_LOGDIR or TMP or TEMP environment variable
and defaults to /tmp if none of the previous exists).

=head1 CONFIG FILE

Options for ntred can be specified on the command line or in a config
file (default is ~/.ntredrc, but --ntred-config-file option can be use
to set a custom config file). The format of this file is simple: each
non-empty line should be of the form:

 option = value

with leading and trailing white-space ignored and with all characters
following a '#' ignored. Option names for the config file are the same
as command-line options, but without leading - or --.

=head1 OPTIONS

=head2 QUERY MODE OPTIONS

  ntred [-m macro-file] [-e code] [--hub hub] [--port port]
        [-N|-H] [-T] [--key-file file] -- script-arguments

=over 8

=item B<--execute|-e> code

This is the query to be executed on the remote servers. It usually
either of a name of the macro defined in the file given as
C<--macro-file> or some Perl one-liner. If omitted, it defaults to
'autostart()' and a macro with this name must be defined in
the macro file provided..

=item B<--macro-file|-m> filename

A file containing macros that are to be sent to the servers and
preloaded on the servers before the particular query code is
evaluated. The query code to be evaluated must be specified in
C<--execute>.

Note that servers already have the set of macros they obtained on
startup (i.e. B<--init>): it is namely either the default set of
macros (C<tred.mac>) or the set specified via B<--macro-file>, and beside one of
these also the set specified via B<--include-macro-file>.

=item B<--include-macro-file|-I> filename

A file containing additional set of macros to be sent to the servers.
In query mode, this option behaves just like C<--macro-file> and is
only provided for compatibility with C<btred>. If both these options
are used, both sets of macros are loaded.

=item B<--hub> hostname

The hostname of the hub to connect to (defaults to localhost).

=item B<--port> port

The port on which the hub is listening (defaults to 1500).

=item B<--all-trees|-T>

Run the query code on all trees (wrapping the code into a
C<if ($root) { do {{ CODE }} while TredMacro::NextTree() }> loop).

=item B<--all-nodes|-N>

Run the query code on all nodes (you still must use --all-trees or -T
to process all trees) (wrapping the code into a
C<while ($this) { CODE ; $this=$this->following }";> loop).

=item B<--all-nonhidden-nodes|-H>

Run the query code on all nodes that are not hidden (you still must
use --all-trees or -T to process all trees). This wraps
the code into a
C<while ($this) { CODE ; $this=$this->following_visible(FS()) }";> loop).

=item B<--listed-files|-L>

Run the query code only on files specified on the command line
(provided they are already present on some server, i.e. this option
does I<not> make servers load files they don't already
have). Filenames may contain ordinary TrEd suffixes of the form ##tree
or ##tree.node to indicate that the processing should apply only on a
single tree (use in combination with -N) or a single node (use without
-T and -N). NTrEd URIs of the form ntred:// are also allowed. Relative
file-names are expanded according to the current working directory
before they are sent to ntred servers.

=item B<--regexp-files|-R> regular-expression

Run the query code only on files whose filenames match a given regular
expression.

=item B<--filelist|-l> filename

Like C<--listed-files|-L> but this time the files to be processed are
listed in the given file rather than on the command line. Both options
may be used together in which case the file-lists are joined.

=item B<--key-file>

This option may be used to provide a file with a session-key which is
necessary for the authentication to the running hub. This defaults to
`~/.ntred_session_key'.

=back

=head2 HUB AND SERVER MODE OPTIONS

  ntred -i [--servers server[,server,...]] [--serverlist server-list]
           [--filelist file-list] [--max-servers num]
           [--no-secondary-files] [--no-server-check]
           [--no-server-start] [--no-hub-start] [--old-dist-method]
           [--safe-mode] [--server-debug] [--max-retries num]
           [-m macro-file] [-M module] [--btred path-to-btred]
           [--ssh ssh-command] [--local] [--key-file file]
           [--hub hub] [--port port] [file [...]]

=over 8

=item B<--servers> list of hosts

A comma separated list of hosts to run btred servers. The hostname may
be optionally followed by a comma and a port number thus making it
possible to run several btred servers on one host. If the port number
is omitted, it defaults to 1600. See also C<--serverlist>.

Ports may be given as a range, e.g. 1600-1800; in this case btred
server will use the first port from the range that is free.

Special syntax can be used to start btred servers over SGE cluster
using the task queue (qsub and qdel commands must be in PATH):

  qsub://<job_prefix>:<port_range>

In this case ntred schedules one btred server task on the SGE queue
for each port in the range. It waits at most two seconds times
C<--max-retries> for the first server to start. Tasks of the servers
that are not running by that time are abandoned and deleted from the
queue.

=item B<--serverlist> filename

This provides more convenient way to specify servers by providing a
file containing a list of servers, one per line. If neither
C<--servers> nor C<--serverlist> is
provided, then the list of servers is read from ~/.ntred_serverlist.

=item B<--filelist> filename

A list of files to distribute between servers (one filename per line).
Additional files may be given as command-line arguments.

=item B<--no-secondary-files>

Don't load "secondary" files along with normal files (a file may
require other - secondary - file to load along with it; this is
typical for stand-off annotation where one tree is built upon
another).

=item B<--allow-no-trees|-0>

Allow files with no trees (normaly such files are considered broken).
Note: the short flag is -zero.

=item B<--max-servers> number

Limit number of servers to start even if the list of servers contains
more of them.

=item B<--max-retries> number

Specifies how many times the hub tries to connect to a btred server
before it gives up.

=item B<--no-server-check>

Skip an initial check for server hosts availability (a dummy attempt
for SSH connection).

=item B<--no-server-start>

Don't start new btred servers on the remote hosts. Instead, start a
hub and try to connect to the btred servers already running. This
requires a server-session key to be given on the standard input.

=item B<--no-hub-start>

Start servers on the remote hosts but don't start a hub. The server
session key required for authentication to the servers will be
printed on the standard output.

=item B<--safe-mode|-F>

Run btred servers in safe mode in which all macros are processed in a
Safe compartment whith some security restrictions.  This mode seems to
be likely to cause btred servers to suffer from memory leaks.

In the safe mode, only the following opcodes and opcode-sets are
allowed (see documentation for the Opcode Perl module):

  :base_core :base_mem :base_loop :base_math
  entereval caller dofile
  print entertry leavetry tie untie bless
  sprintf localtime gmtime sort require

plus :base_orig, but the following opcodes (which are forbidden):

  getppid getpgrp setpgrp getpriority setpriority pipe_op sselect
  select dbmopen dbmclose tie untie

=item B<--server-debug>

Run btred server with -D flag for some more debugging information.

=item B<--macro-file|-m> filename

A file containing the default set of macros to be prepared on btred
servers. The file (with exactly the same path) must be visible from
all server hosts.

=item B<--include-macro-file|-I> filename

A file containing additional set of macros to be prepared on btred
servers. This option is typically used instead of C<--macro-file> to
load macros from B<both> C<filename> B<and> the default macro set from
(C<tred.mac>). C<--macro-file> can still be used in combination with
B<--include-macro-file> to supply a replacement for C<tred.mac>.

=item B<--enable-extensions> list

Give a comma-separated list of installed TrEd extension names to
temporarily enable if disabled in the extension configuration.
Use '*' to enable all currently enabled extensions.

This option can only be used with C<--init> or C<--reload-macros>.

=item B<--disable-extensions> list

Give a comma-separated list of installed TrEd extension names to
temporarily disable if enabled in the extension configuration.
Use '*' to disable all currently enabled extensions.

This option can only be used with C<--init> or C<--reload-macros>.

=item B<--key-file>

Allows to specify a file where the the session-key for a client's
authentication will be stored. It defaults to ~/.ntred_session_key.

=item B<--terminal-encoding|-d> encoding

Automatically applies a given character encoding to all stdout output
operations on the servers and command-line arguments.
Only works with Perl >= 5.8.

=item B<--hub> hostname

The hostname of the local machine the hub will listen on (defaults to
localhost but a machine's hostname may be given to allow remote access
to the hub).

=item B<--port> port

The port number the hub will be listening on (defaults to 1500).

=item B<--preload-module|-M> module-name

This option is passed to the btred command line when starting a btred
server. It makes btred preload a given Perl module at btred startup so
as it is available to all macros (DOES NOT WORK WITH RESTRICTED MODE).
This option may be specified more than once with different modules.

=item B<--old-dist-method>

Use old benchmark-based distribution method.

=back

=head2 HUB CONTROL OPTIONS

  ntred --list-files|--list-changed-files

  ntred --reload-files [-filelist file-list] [--listed-files file [...]]

  ntred --reload-changed-files

  ntred --reload-macros [-m macro_file]

  ntred --load-files [--filelist file-list] [file [...]]

  ntred --close-files

  ntred --save-files|--save-changed-files [-s strip-sfx]
       [-a append-sfx] [-p strip-prefix] [-r add-prefix] [-f out-fmt] [--knit]

  ntred --quit

  ntred --kill|--ps-kill [--servers server[,server,...]] [--serverlist server-list]

  ntred --break|--ps-break

  ntred --dump-files [--filelist file-list] [file [...]]

  ntred --upload-file filename < fs-file


=over 8

=item B<--init|-i>

Start remote btred servers and a hub. See L</HUB AND SERVER MODE OPTIONS>.

=item B<--keep-empty-servers|-E>

Keep alive those servers which do not get any file during the initial
file distribution (by default, unused servers are automatically
closed).

=item B<--quit>

Sends the hub and all the servers a command to quit.

=item B<--break>

Tells ntred nub to send USR1 a signal to all running btred servers.
Upon receiving that signal, the servers should stop current processing
and return to the request awaiting state.

=item B<--ps-kill>

Similar to C<--break> but tries to identify btred server processes by
looking at the output of the system command C<ps x -o pid,command>. This may help if
C<killall -USR1 btred> doesn't work.


=item B<--kill>

Runs C<killall -9 ntred> on the local machine and and C<killall -9 btred> on
the server hosts listed with --servers, --serverlist, or in
~/.ntred_serverlist.

=item B<--ps-kill>

Similar to C<--kill> but tries to identify btred server processes by
looking at the output of the system command C<ps x -o pid,command>. This may help if
C<killall -TERM btred> doesn't work.

=item B<--list-files>

List all files currently open on servers.

=item B<--list-changed-files>

List all files that have been changed by some macro. Note, that a macro
has to claim that the file was changed by setting
C<$TredMacro::FileChanged> variable to 1, otherwise the btred server
would never notice.

=item B<--listed-files|-L>

Apply request only on listed files (currently only works for queries
and C<--reload-files> request).

=item B<--count-files>

Query number of files open on each server.

=item B<--reload-files>

Send a command to the btred servers to reload all open files.  If
C<--filelist> or C<--listed-files> options are given, reload only
files occuring in the given lists (all other files remain intact in
servers' memory).

=item B<--reload-changed-files>

Send a command to the btred servers to reload files that have been
modified since (re)loaded.

=item B<--reload-macros>

Send command to the btred servers to reload the initial macro file.
If C<-m> (C<--macro-file>) is specified, the servers use the given
macro-file instead of the original one (specified when initializing
btred servers). Note, that the file (with exactly the same path) must
be visible from all server hosts.

Options C<--disable-extensions> and C<--enable-extensions> can be used
to define a set of extensions to use.

=item B<--load-files>

Send a command to the hub to distribute files given on the
command-line or those specified using C<--filelist> of in
~/.ntred_filelist to the servers.

Warning: Do not load files already in servers' memory, otherwise you
may get duplicates (depending on to which server the files get
distributed). Use C<--list-files> to see which files are already loaded.

See also C<--reload-files>.

=item B<--close-files>

Send a command to the btred servers to close all open files.

=item B<--save-files>

Send a command to the servers to save all open files.  The filenames
of the saved files may be modified using
C<--add-prefix>, C<--strip-prefix>,
C<--strip-suffix>,
C<--append-suffix>.

=item B<--save-changed-files>

Same as C<--save-files> except that only files that have been changed
by some macro will be saved. Note, that a macro has to claim that the
file was changed by setting C<$TredMacro::FileChanged> variable to 1,
otherwise the btred server would never notice. See also
C<--list-changed-files>.

=item B<--knit|K> ALL|NONE|name1,name2,...

If a file is saved, save/update also listed types of reffiles the file
pulled data from. For the moment, this only makes sense with the
C<PML> backend which supports so called B<knitting>, i.e. a method to
pull certain data from external resources and push it back (with all
changes) to the original position in the resource when saving the
file. This option allows to list the types of resources (in PML the
types are the reference names listed in the PML schema) which should
be saved. Default is C<NONE>. This type of resources doesn't include
so called secondary files.

=item B<--dump-files>

Dumps given files or trees to standard output in FS format as they are
in memory of the btred servers. Individual dumps are separated with
`//FSEND' preceded and followed by two newline characters.  To output
a single tree, follow the filename with C<##n> suffix where C<n> is
the absolute position of the tree in the file (starting from one).
The following example shows how C<csplit> command can be used to
save individual dumps into separate files:

  ntred --dump <files> | csplit -z -f out -b '%d.fs' - '/\/\/FSEND/+2' '{*}'

To merge these separate files into one huge FS file, use

  any2any -m hugeout.fs out*.fs

=item B<--upload-file>

This command is kind of a reverse to dump. It takes a filename from
the first command line argument and a FS file from the standard input
and sends them to the btred servers. The server possessing the file
with the given filename replaces its own in-memory copy of the file
with the one provided on the standard input. To update a single tree,
follow the filename with C<##n> suffix where C<n> is the absolute
position of the tree in the file (starting from one).

=item B<--no-secondary-files>

Ignore "secondary" files even if loaded. This only affects some
commands, such as B<--list-changed>, in which particular case is means
that a file, whose secondary file was changed, is not reported as
changed unless the (primary) file itself was marked as changed.

=item B<--allow-no-trees|-0>

Allow files with no trees (normaly such files are considered broken).
Note: the short flag is -zero.

=back

=head2 SAVING OPTIONS

=over 8

=item B<--strip-prefix|-p> regexp

Remove strings matching given regexp from the beginning of filenames
before saving.

=item B<--add-prefix|-r> prefix

Prepend output filenames with the given prefix when saving.

=item B<--strip-suffix|-s> regexp

Strip strings matching given regexp from the end of filenames
when saving.

=item B<--add-suffix|-a> suffix

Append given suffix to the filenames when saving.

=item B<--output-format|-f> [fs|csts|trxml|tei|storable]

Format to use for files being saved.

=item B<--no-secondary-files>

Don't save "secondary" files (not even if changed). Normally,
secondary files (if loaded) are saved along with their primary files
(the exactly same file-name prefix/suffix processing and format apply
to both the primary and secondary files).

=back

=head2 GENERAL OPTIONS

=over 8

=item B<--glob|-g>

Apply Perl C<glob> function on the filename patterns given on the
command-line.  This expands possible wild-card patterns on each of the
filename command-line argument as the standard Unix shell /bin/csh
would do. This can not only help in a situation where the shell used
doesn't support wildcard expansion, but can also be used to reduce the
number of the command-line arguments passed to the process in cases
where the argument list would after the shell-expansion exceed a
system limit. Note, that currently expansion is performed on the
client regardless of the type of request. This may change in the
future versions.

=item B<--usage|-u>

Print a brief help message on usage and exits.

=item B<--help|-h>

Prints the help page and exits.

=item B<--man>

Displays the help as manual page.

=item B<--quiet|-q>

Suppress all NTRED-CLIENT/NTRED-HUB messages on error output.

=item B<--really-quiet|-Q>

Redirect all std error output to /dev/null.


=item B<--ssh> command

This option may be used to specify the ssh/rsh command to use to
connect to remote servers. Default value is `ssh -o ConnectTimeout=10'.

=item B<--local>

Run btred servers locally. It ignores all non-local entries in the
.ntred_serverlist (i.e. entries not matching local machine's
hostname). There still may be more BTrEd instances, provided they use
different ports. In this case, SSH is not used.

Use --servers localhost:XY (where XY is port number)
if you wish to run a single server on the loopback interface.

=item B<--btred> command

This option may be used to specify the command to use to start a btred
server on a remote host. The command must accept any btred parameters.

=back

=head1 SECURITY ISSUES

USE AT YOUR OWN RISK. IF SECURITY IS A CRITICAL ISSUE OR IF IN DOUBT,
DON'T USE IT AT ALL.

Why is security an issue here? Because btred servers execute almost
arbitrary Perl code provided by the client. In the C<--unrestricted>
mode such code may contain arbitrary commands such as system() or
open(). It is therefore desirable that the servers are not open for
all parties.

The following precautions have been taken to lower the potential
security risks:

1) Both btred servers and hub require an authorization based on
   verification of a MD5 signature of a random data block (generated
   by the server in case of the hub-to-btred-server communication and
   by the hub in case of the client-to-hub communication) xor-ed with
   an authorization key known to both parties. Although the
   communication is unencrypted, the client must with each
   request send a MD5 checksum of the request XORed with the secret
   authorization key.  Only requests whose signature is verified
   by the server, are responded to.

2) There may be only one connection from a hub to a server.  As soon as
   it is closed, the server terminates.

3) If the servers are started by the hub itself (using C<--init>) the
   authorization key is created by the hub and is passed to the btred
   server via a ssh encrypted pipe.

4) For the client's disposal, the authorization key is stored in
   user's home directory as C<~/.ntred_session_key> with permissions
   set to 600 (only user can read or write). This theoretically
   (depending on the general security of the system) limits the access
   to the hub (and thus to the servers) to the user running the hub
   only. It may, though, be obviously abused from the local root
   account to execute arbitrary perl code on all btred server
   hosts. This might especially be undesirable if the hub runs on a
   machine whose administrator would normaly have no user access to
   the machines running btred servers. Another possible security issue
   might arise if user's home directory is on a remote NFS server, so
   that NFS intervenes accessing the key file. Since NFS uses an
   unencrypted protocol, network sniffing techniques may be used to
   obtain the authorization key and hence run arbitrary code on btred
   hosts. If such situations are likely to happen (e.g. in a large
   network) it is advisable to use a different location for the
   authorization key (see C<--key-file>), e.g. C</tmp>.

5) It is possible to restrict Perl code evaluated on the servers to a
   safer compartment, where some critical Perl commands are disabled.
   In some cases, these restrictions may not be sufficient, in other
   they may be too strict. Some memory leaks can appear when Safe
   compartment is used. See C<--safe-mode> above for more discussion.

6) Unless C<--hub> option is used, the hub runs on localhost and as
   such is not (under normal circumstances) open for connections from
   the outside world. If you are considering making the hub listen on
   a non-local interface, note that it is a much better option to
   configure a secure SSH tunnel.

=head1 FILES

C<~/.ntred_serverlist>  - default list of servers to use

C<~/.ntred_filelist>    - default list of files to load on servers

C<~/.ntred_session_key> - client/hub session key

=head1 LICENSE

This software is distributed under GPL - The General Public Licence.
Full text of the GPL can be found in the LICENSE file distributed with
this program and also on-line at http://www.gnu.org/copyleft/gpl.html.

=head1 AUTHORS

Petr Pajas <pajas@matfyz.cz>

Zdenek Zabokrtsky <zabokrtsky@ufal.mff.cuni.cz>

Copyright 2003-2008 Petr Pajas and Zdenek Zabokrtsky, All rights reserved.

=cut
