#-------------------------------------------------------------------------------
# xCopy package
# coded by Sanjit Rath
# For bugs and suggestions mail to sanjit.rath at gmail.com
# April - September 2006
#-------------------------------------------------------------------------------

package xcopy;

require v5.6.0; 
use warnings;
use strict;
use Config qw(%Config);

#version of the script 
my $VERSION = 0.1;

#check if the thread support is available
$Config{useithreads} or die "Recompile Perl with threads to run this program.";

use IO::File;
use File::Copy;
use threads;
use threads::shared;
use Thread::Queue;

sub BEGIN
{
	push( @INC, './tsk' );
}

use tsk::task;
use tsk::bucket;

sub new
{
	my ( $class, $numberOfThreads, $logFile, $errorLog ) = @_;
	my @taskList     = ();
	my @threadIdList = ();
	my $self = {
		_numberOfThreads => $numberOfThreads,
		_logFile         => defined($logFile) ? $logFile : "",
		_errorLog        => defined($errorLog) ? $errorLog : "",    #error log
		_taskListRef     => \@taskList,
		_bServerStarted  => 0,                 #server is yet to be started
		_expandThreadId  => undef,
		_threadIdListRef => \@threadIdList,    #list of worker thread id
	};
	
	bless $self, $class;
	return $self;
}

#logger queue of thread ids 
my $qLog = Thread::Queue->new();
#boolean value to stop logger thread 
my $stopLogger : shared = 0;
#thread object of logger thread 
my $thLog =(); 

#error queue of thread ids 
my $qError = Thread::Queue->new();
#boolean value to stop error thread
my $stopError : shared = 0;
#thread object of error thread 
my $thError = (); 

sub __startLogger
{
	my ($this) = @_;
	my $logOpened = 0;
	if ( $this->{_logFile} eq "" )
	{
		$stopLogger = -1;
		return 0;
	}
	else
	{
		if ( open( LOG, ">" . $this->{_logFile} ) )
		{
			$logOpened = 1;
		}
		else
		{
			print "Error: Couldn't open log file $this->{_logFile}\n";
		}
	}

	sub logger
	{
		while ( $stopLogger == 0 && $qLog->pending() > 0 )
		{
			my $__error_val = qLog->dqueue();
			print LOG $__error_val;
			print $__error_val;
		}	
		close(LOG);
	}

	my $tid = threads->create( \&logger, "" );
	unless ( defined($tid) )
	{
		die("Error: Failed to start logger theread\n");
	}
	
	$thLog = $tid;
	return 1; 
}

sub log
{
	my ( $this, $message ) = @_;
	if ( $stopLogger == -1 )
	{
		return;
	}
	$qLog->enqueue($message);
}

sub __startErrorLogger
{
	my ($this) = @_;
	my $logOpened = 0;
	if ( $this->{_errorLog} eq "" )
	{
		$stopError = -1;
		return 0;
	}
	else
	{
		if ( open( ERROR_LOG, ">" . $this->{_errorLog} ) )
		{
			$logOpened = 1;
		}
		else
		{
			print "Error: Couldn't open log file $this->{_errorLog}\n";
		}
	}

	sub error_logger
	{
		while ( $stopError == 0 && $qError->pending > 0 )
		{
			my $__error_val = qLog->dqueue;
			print ERROR_LOG $__error_val;
			print $__error_val;
		}
		
		close(ERROR_LOG);
	}

	my $tid = threads->create( \&error_logger, "" );
	unless ( defined($tid) )
	{
		die("Error: Failed to create error logger thread\n");
	}
	 
	$thError = $tid; 
	return 1; 
}

sub error
{
	my ( $this, $message ) = @_;
	if ( $stopError == -1 )
	{
		return;
	}
	$qError->enqueue($message);
}

#subroutine to stop both error logger and logger threads
sub __stopLoggers
{
	if ( $stopLogger == 0 )
	{
		$stopLogger = 1;
		if ( $qLog->pending() == 0 )
		{
			$qLog->enqueue("\n\n ** stopping logger **\n");
		}
	}

	if ( $stopError == 0 )
	{
		$stopError = 1;
		if ( $qError->pending() == 0 )
		{
			$qError->enqueue("\n ** ending error logger \n");
		}
	}
}

sub addTask
{
	my ( $this, $strSrcDir, $strDestDir, $strFlags ) = @_;
	unless ( defined($strSrcDir) || defined($strDestDir) || defined($strFlags) )
	{
		die("xcopy Error: undefined add task parameters @_ ");
		return;
	}

	my $tsk = new tsk::task( $strSrcDir, $strDestDir, $strFlags );
	push( @{ $this->{_taskListRef} }, $tsk );
}

sub __runSerialCopy
{
	my ($this) = @_;
	my @tasks = @{ $this->{'_taskListRef'} };

	sub __fileBrowser
	{
		my ( $dir, $tsk ) = @_;
		print "$dir \n";

		#try opening the directory
		unless ( opendir( DIRF, $dir ) )
		{
			$this->error("Can't open $dir\n");
			return;
		}

		my ( $dir_item, @dirs );
		foreach $dir_item ( sort readdir(DIRF) )
		{
			if ( $dir_item eq "." || $dir_item eq ".." )
			{
				next;
			}

			my $complete_path = "$dir/$dir_item";
			if ( -d $complete_path )
			{
				push( @dirs, $complete_path );

				#it is a directory
				my $dest_path = $tsk->destPath($complete_path);
				unless ($dest_path eq "" )
				{
					unless ( $this->__makeDir($dest_path) )
					{
						next;
					}
				}
			}
			else
			{
				#it is a file
				my $dest_path = $tsk->destPath($complete_path);
				unless ($dest_path eq "" )
				{
					unless ( $this->__makeDirForFile($dest_path) )
					{
						next;
					}
					unless ( copy( $complete_path, $dest_path ) )
					{
						$this->error("Error: Failed to copy $complete_path - > $dest_path \n");
					}
				}
			}
		}

		closedir(DIRF);

		$dir_item = "";
		foreach $dir_item (@dirs)
		{
			__fileBrowser( $dir_item, $tsk );
		}
	}

	#for each tasks in task list run file browser
	foreach my $tsk (@tasks)
	{
		__fileBrowser( $tsk->__taskSourceDir(), $tsk );
	}

}

sub __runParallelCopy
{
	my $this = shift;

	#inputs to the thread procedure
	# array of reference
	# 1 -> \@ list of files
	sub __threadProc
	{
		my ($listRef) = @_;
		unless ( ( ref $listRef ) eq "ARRAY" )
		{
			return 0;
		}

		my @list = @$listRef;
		foreach my $cpPair (@list)
		{
			my ( $src, $dest ) = $cpPair =~ m/(.+)|(.+)/g;
			unless ( $this->__makeDirForFile($dest) )
			{
				next;
			}
			if ( copy( $src, $dest ) )
			{
				$this->error("Error: Failed to copy $src - > $dest \n");
			}
			else
			{
				$this->log("$src - > $dest \n");
			}

		}
	}

	my $bucket = tsk::bucket->new(10);
	my @tasks  = @{ $this->{'_taskListRef'} };

	sub __fileBrowserP
	{
		my ( $dir, $tsk ) = @_;

		#try opening the directory
		unless ( opendir( DIRF, $dir ) )
		{
			$this->error("Can't open $dir\n");
			return;
		}

		my ( $dir_item, @dirs );
		foreach $dir_item ( sort readdir(DIRF) )
		{
			if ( $dir_item eq "." || $dir_item eq ".." )
			{
				next;
			}

			my $complete_path = "$dir\\$dir_item";
			if ( -d $complete_path )
			{
				push( @dirs, $complete_path );

				#it is a directory dont schedule in the job queue
				my $dest_path = $tsk->destPath($complete_path);
				if ( defined($dest_path) )
				{
					unless ( $this->__makeDir($dest_path) )
					{
						next;
					}
				}
			}
			else
			{

				#it is a file
				my $dest_path = $tsk->destPath($complete_path);
				if ( defined($dest_path) )
				{
					$bucket->add( $complete_path, $dest_path );
					if ( $bucket->items() == $bucket->size() )
					{
						while (threads->list() >= $this->{'_numberOfThreads'} )
						{
							sleep(2);

							#wait till the copy jobs are below
							#the number of permissible jobs
						}
						my @content = $bucket->content();
						$bucket->clean();
						
						unless(threads->create( \&__threadProc, \@content ))
						{
							die("Fatal: Couldn't create thred for parallel run"); 
						}
					}
				}
			}
		}

		closedir(DIRF);

		$dir_item = "";
		foreach $dir_item (@dirs)
		{
			__fileBrowserP( $dir_item, $tsk );
		}
	}

	#for each tasks in task list run file browser
	foreach my $tsk (@tasks)
	{
		__fileBrowserP( $tsk->__taskSourceDir(), $tsk);
	}

}

# sub to wait for threads to finish execution
# input nothing, returns after threads have finished
sub __waitForThreads
{
	my ($this) = @_;
	#check if the logger threads are running 
	if(($stopError == 0)||($stopLogger == 0)) 
	{
		$this->__stopLoggers(); 
	}
	
	foreach (threads->list())
	{
		threads->object($_)->join();
	}
}

#sub to make directory tree
#inputs directory
#returns true if directory is created false otherwise
# TODO __makeDir takes directory and creates the directory structure
sub __makeDir
{
	my ( $this, $dir ) = @_;

	#if the directory exist return
	if ( -d $dir )
	{
		return 1;
	}

	$dir =~ s/\\/\//g;
	my @comps = split( /\//, $dir );
	my $path = "";
	for ( my $i = 0 ; $i <= $#comps ; $i++ )
	{
		if ( $path eq "" )
		{
			$path = $comps[$i];
		}
		else
		{
			$path = "$path/$comps[$i]";
		}

		#create directory component if it doesnt exist
		unless ( -d $path )
		{
			unless ( mkdir($path, 0777) )
			{
				$this->error("Error: failed to make directory $path\n");
				return 0;
			}
		}
	}

	return ( -d $dir );
}

#sub to make directory tree out of FilePath
#input: File Path
#returns true if directory is created false otherwise
sub __makeDirForFile
{
	my ( $this, $file ) = @_;
	$file =~ s/\\/\//g;
	my @comps = split( /\//, $file );
	pop(@comps);
	my $dir = join( '/', @comps );

	#if the directory exist return
	if ( -d $dir )
	{
		return 1;
	}
	my $path = "";
	for ( my $i = 0 ; $i <= $#comps ; $i++ )
	{
		if ( $path eq "" )
		{
			$path = $comps[$i];
		}
		else
		{
			$path = "$path/$comps[$i]";
		}

		#create directory component if it doesnt exist
		unless ( -d $path )
		{
			unless ( mkdir($path) )
			{
				$this->error("Error: failed to make directory $path\n");
				return 0;
			}
		}
	}

	return ( -d $dir );
}

sub run
{
	if ( $#_ != 1 )
	{
		print "Error: wrong number of arguments to run\n";
		return;
	}

	my ( $this, $bucketSize ) = @_;

	#$this->__startLogger();
	#$this->__startErrorLogger();
	if ( $this->{_numberOfThreads} == 0 )
	{
		$this->__runSerialCopy();
	}
	else
	{
		$this->__runParallelCopy();
	}
	sleep(2);
	$this->__stopLoggers();
}

1;

__END__
=head1 NAME

Module xop::xcopy

This is a generic XCOPY implementation with platform independant standard 
features, in perl, with many improvements. It uses task concept, where the task is 
expanded to subtasks and each subtask is grouped as buckets, each buckets are executed
parallel or in serial as per the arguments to task. 

It is designed for very large copy of files typically used in SCM (Souce Code Management)
enviroments  like ClearCase, CVS and Oracle ADE, where time required for copy and 
accuracy is most critical. 

Features as of version 0.1  
  i.  Stable task execution
 ii.  Serial & Parallel XCOPY  
iii.  Log file generation for each task


=head1 DESCRIPTION
xop::xcopy 
Concepts: 
xopy works by thread and bucket concept, buckets represent a group of tasks, these
are executed by a single thread. This has been designed keeping in mind the following 
factors
  i. Prevent thread rush for acessing shared task: 
      there can be two approches for the problem 
       a) constant number of running thread, and variable number of file copy sub-tasks 
          if time required for the sub-tasks are small there is possibility of thread rush
          where most CPU time is consumed by the running threads
       b) constant number of running threads and constant number of file copy sub-tasks
          Here also if time required for copy task is small, there is potential thread 
          rush problem, secondly constant running threads consume CPU time
       c) Create a thread for a constant number of file copy sub-tasks (bucket) up a 
          constant number of threads (number of threads)
          This approach solves thread rush as well as most of the CPU time is given 
          for file copy sub-tasks 
          

Requirement: 
perl version 5.8 and higher although it will work with 5.6 and higher 
This is because of the improved thread in perl in higher versions. 


Usage: 

use xop::xcopy; 
my $cp = xop::xcopy->new(<number of threads>,<log>, <error log>);

 <number of threads>: if 0,1 initiates serial copy, no threads are created 
                      if >1, n, initiates parallel copy with 'n' threads running 
 < log >            : log file 
 < error log >      : error log file
 

$cp->addTask(<source dir>,<dest dir>,<xcopy flags>);
 
<source dir>: source directory content, current version doesnt support wild cards 
<dest dir>  : destination directory
<flags>     : currently being worked on, please leave this empty, ""  

possible values for flags 
set the flags
 -d:d/m/y copies files with modification time after the said date
 -s copies directories and subdirectoriesand files
 -e copies directories and subdirectories including empty ones
 -c continue copying even if error occures, default behaviour is to stop
    the script execution
 -i copyies files from directoriy tree to a destination directory. Here
   destination directory structure is not created
 -h copies system and hidden files
 -r overwrites read only files
 -t creates directory structure only
 -u copies files only if destination file exists
 -rx: simple wild card expression, possible values *, *.*, *<some>*  
 -prx: complex perl regular expression 

more tasks can be added to the xcopy objects 

$cp->run(<size of bucket>); 

<size of bucket> : number of task to be grouped for copy, this number is relevant if
                   running a parallel copy, ie <number of threads> is set to > 1
                   

=head1 README

If there is any text in this section, it will be extracted into
a separate README file.

=head1 PREREQUISITES
The module requires perl to be compiled with thread support. 

This script requires the C<strict> module.  
It also requires the following 
C<IO::File>
C<File::Copy>
C<threads>
C<threads::shared>
C<Thread::Queue>

=head1 COREQUISITES

=pod OSNAMES

any

=pod SCRIPT CATEGORIES

CPAN/Administrative
Fun/Educational

=head1 AUTHOR

Sanjit Rath (c) 2006, March - September 2006 

I am currently working on new version of scripts with almost weekly updates. 
Feel free to suggest new features, bugs @
sanjit [.] rath [@] Oracle.com 
sanjit [.] rath [@] Gmail.com 


=cut
