#! /usr/oracle/bin/oraperl
push(@INC,"/usr/local/lib/perl");
require "sys/ipc.ph";
require "sys/sem.ph";
require "sys/mode.ph";
require "sys/msg.ph";

# codes of message identifiers
$URL_QUEUE_SIZE=5000;
$URL=2;
$NEXT=3;
$ADD_ADDR=4;
# key for creating message queues and semaphores
$key=989690581;

#$filename = '/usr/oracle/search/tmp/urls.tmp';
$filename = '/home/T11D/tapulika/think/final_projs/gather/tmp/urls.tmp';
$|=1;

#$command = 'aixterm -l -lf /usr/oracle/search/tmp/aixterm.log -fn 6x10 -e rtload.sh';
$command = 'aixterm -l -lf /home/T11D/tapulika/think/final_projs/gather/tmp/aixterm.log -fn 6x10 -e rtload.sh';

# This procedure sends passed in argument number of url to gathering program
# It also marks these urls as send ( flag 1)

sub send_addrs
{
  local($number)=@_;
  local($nr_snd)=$number;
  local($csr,$msg);
  local(@url_rowids,@url_prots,@url_hosts,@url_paths,@url_depths,@url_fid,@url_hostids);
  while(1)
    {
      $csr=&ora_open($lda,
        "SELECT rowid,host_id,prot,host,path,depth,fid FROM search_urls WHERE flag='0' order by priority asc");
      for($i=0;$i<$number;$i++)
	{  
	  ($url_rowids[$i],$url_hostids[$i],$url_prots[$i],$url_hosts[$i],
	   $url_paths[$i],$url_depths[$i],$url_fid[$i])=&ora_fetch($csr);
	  last if $url_rowids[$i] eq '';
	}
      &ora_close($csr);
      last if $i!=0;
      {
#	sleep(30);
	local($arg,$tmp,$nummsg)=(' ' x 80,'',0);
	msgctl($msgid,&IPC_STAT,$arg);
	@tmp=unpack("S30",$arg);
	$nummsg=$tmp[19];
	last if $nummsg != 0;
#	&load_to_TR if -e $filename;
	sleep(10);
      }
    }
  $number=$i;
  for($i=0;$i<$number;$i++)
    {
      local($string);
      &ora_do($lda,
	"UPDATE search_urls SET flag='1'  WHERE rowid='$url_rowids[$i]' AND flag='0'");
      print "OUT: $url_hosts[$i] , $url_paths[$i] , $url_fids[$i]\n";
      $string=join("\0",$url_prots[$i],$url_hosts[$i],$url_paths[$i]) . "\0";
      $hostoffs=length($url_prots[$i])+1;
      $pathoffs=$hostoffs+length($url_hosts[$i])+1;
      $msg=pack("L6 a*",$URL,$url_fids[$i],$url_hostids[$i],$url_depths[$i],$hostoffs,$pathoffs,$string);
      msgsnd($msgid,$msg,0) || die $!;
    }
  &ora_commit($lda);
}

# This procedure loads data into TR database. It continues execution immediatelly,
# but it waits berore running next load for termination of a previous one.
# If rtload is interrupted database is reloaded
# after wait it also checks if it has connection with oracle. If Delete command cannot
# be executed it logson and tries to repeat oracle operation.

sub load_to_TR
{
  local($retval);
  if($child_pid!=0)
    {
      $?=0;
      wait;
      if(($? & 255) !=0)   # is it necessary to reload?
	{
	  while(($? & 255) != 0)
	    {
	      $?=0;
	      system($command);
	      print "Return code: $? \n";
	    }
	}
    }
  semop($id,pack("sss",0,-1,0)) || die "*******error $!\n";
  while(! (defined ($retval=&ora_do($lda,"DELETE FROM search_urls WHERE flag='3'"))))
    {
      &ora_logoff($lda);
      $lda=&ora_login('','think2','think2');
      print "SOMETHING WRONG\n";
    }
  &ora_commit($lda);
  &ora_do($lda,"UPDATE search_urls SET flag='3' WHERE flag='2'");
  &ora_commit($lda);
  system("mv $filename $filename.rtd");
  semop($id,pack("sss",0,1,0)) || die "*******error $!\n";
 FORK:
  {
    if($child_pid=fork)
      {
      }
    elsif(defined $child_pid)
      {
	exec($command);
	exit;
      }
    elsif($! =~ /No more process/)
      {
	sleep (10);
	redo FORK;
      }
    else
      {
	die "Can't fork: $!\n";
      }
  }
}


# this is main server of urls procedure. It receives maessages from gathering
# programs and if number of messages in message queue drops below certain
# number it starts sending urls
# It also check and if size of file to which data is loaded exceeds 1 MB it runs
# rtload application to load it to TR

sub server
{
  local($nr)=0;
  local($flag);
  $SIG{'TERM'}='ser_handler';
  $SIG{'INT'}='ser_handler';      
  while()
    {
      local($arg)=' ' x 80;
      local($nummsg,$msg,@tmp,$ref);
      local($flag,$path,$prot,$host,$path);
      $msg="\0" x 255;
      msgrcv($msgid,$msg,255,$NEXT,0);
      ($tmp,$flag,$host_id,$tmp,$tmp3,$tmp4,$ref)=unpack("L L L L L L a*",$msg);
      
      ($prot,$host,$path)=split("\0",$ref);
      
      if($prot ne '')
	{
	  if($flag==0)
	    {
	      &ora_do($lda,"UPDATE search_urls SET flag='2' WHERE host_id=$host_id AND path='$path'") ;
	      print "IN*OK: $prot://$host$path\n";
	    }
	  else
	    {
	      &ora_do($lda,"UPDATE search_urls SET flag='4' WHERE host=$host_id AND path='$path'") ;
	      print "IN*ER: $prot://$host$path\n";
	    }
	}

      &ora_commit($lda);
      msgctl($msgid,&IPC_STAT,$arg);
      @tmp=unpack("S30",$arg);
      $nummsg=$tmp[19];
      &send_addrs(120-$nummsg) if $nummsg < 60;
      &load_to_TR if ((-s $filename) > 1000);
    }
}

# initialization of semaphores and messages
# this module actually does not use msgida but it still opens this message queue
# for use by other modules.
# I just prefered to have all queues opened here

      
$id=semget($key,2,2, &IPC_CREAT | &S_IRUSR | &S_IWUSR) 
  || die "can't open semaphore";
semctl($id,0,&SETVAL,1) || die "can't set semaphore";
semctl($id,1,&SETVAL,$URL_QUEUE_SIZE) || die "can't set semaphore2";
$msgid=msgget($key, &IPC_CREAT | &S_IRUSR | &S_IWUSR) 
  || die "can't open message queue";
$msgida=msgget($key+1, &IPC_CREAT | &S_IRUSR | &S_IWUSR) 
  || die "can't open message queue";
$lda = &ora_login('','think2','think2');

# unmark all urls that were previously sent but have not been marked processed
&ora_do($lda,"UPDATE search_urls SET FLAG=0 WHERE FLAG=1");
&ora_commit($lda);
&server;


# this procedure handles interrupts
# it receives all messages that are in queue
# and marks records in urls tables accordingly to returned status
# Then it closes down all message queues and semaphores
# waits for loading process to finish and logsoff from oracle

sub ser_handler 
{
  print "shutting down\n";
  $SIG{'TERM'}='ser_handler';
  $SIG{'INT'}='ser_handler';
  while(msgrcv($msgid,$msg,255,$NEXT,&IPC_NOWAIT))
    {
      local($flag,$path,$prot,$host,$path);
      ($tmp,$flag,$tmp,$tmp,$tmp,$tmp,$msg)=(unpack("L6 a*",$msg));
      ($prot,$host,$path)=split("\0",$msg);
      if($msg ne '')
	{
	  if($flag==0)
	    {
	      &ora_do($lda,"UPDATE search_urls SET flag='2' WHERE host='$host' AND path='$path'") ;
	      #print "IN*OK: $msg\n";
	    }
	  else
	    {
	      &ora_do($lda,"UPDATE search_urls SET flag='4' WHERE host='$host' AND path='$path'") ;
	      #print "IN*ER: $msg\n";
	    }
	}

    };
  semctl($id,0,&IPC_RMID,0);
  print "semaphore closed\n";
  msgctl($msgid,&IPC_RMID,0);
  msgctl($msgida,&IPC_RMID,0);
  print "messages queues closed\n";
  wait();
  print "children terminated\n";
  &ora_logoff($lda);
  exit;
}
