No technical expert advice can be found here ;)

#!/usr/bin/perl
##
# This software is published under the Apchae 2.0 licenses.
# You may obtain a copy of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
#   
#   Author: Erik Scholtz 
#   Web: http://blog.elitecoderz.net
###
# We are strict, cauz we are elitecoderz!
use strict;
use threads qw(yield);
use threads::shared;
use Thread::Semaphore;
 
# No caching
$|=1;
 
################
# Configuration
my $debug = 1;																			# 0/1 to enable logging to the console or disable
my $path = '/path/to/your/data/';				# Path to sync
my @cmds;																				# Syncer commands that should be executed
$cmds[0] = '/usr/local/bin/rsync -raz --progress --size-only /path/to/symboliclink/data/<!--target--> rsync:///shareName/<!--target-->';
$cmds[1] = '/usr/local/bin/rsync -raz --progress --size-only /path/to/symboliclink/data/<!--target--> rsync:///shareName/<!--target-->';
$cmds[2] = '/usr/local/bin/rsync -raz --progress --size-only /path/to/symboliclink/data/<!--target--> rsync:///shareName/<!--target-->';
 
###############################################################################
# DO NOT CHANGE ANYTHING BELOW THIS LINE, UNLESS YOU KNOW WHAT YOU ARE DOING! #
###############################################################################
 
###
# Set Threads yield
threads->yield();
 
# SetUp some thread-shared variables
my $commands :shared;
$commands = &share([]);
 
my $run :shared;
$run = &share({});
$run->{'status'} = 1;
 
my $sema :shared;
$sema = &share({});
 
# Local array where all syncer-threads are stored
my @threads;
 
# Create a thread for each syncer
my $maxid = -1;
for (my $i=0;$i<=$#cmds;$i++) {
	print "Starting syncer $i\n" if $debug;
	$sema->{$i} = Thread::Semaphore->new(0);
	my $syncer = threads->create('syncJob',$run,$sema,$i,$commands,$path,$cmds[$i],$debug);
	push(@threads,$syncer);
	$maxid = $i;
}
 
# Create the Checker thread, which cleanup the jobs and ensures the function of all syncers
$sema->{'checker'} = Thread::Semaphore->new(0);
my $syncer = threads->create('JobChecker',$run,$sema,$commands,$maxid,$debug);
push(@threads,$syncer);
 
# Create the audit thread
print "Starting audit\n" if $debug;
my $auditthread = threads->create('audit',$run,$sema,$commands,$path,$maxid,$debug);
print "Waiting for audi to terminatet\n" if $debug;
$auditthread->join();			# If the audit-thread gets joinable, we have to terminate everything
 
# Terminate all threads and cleanup 
$run->{'status'} = 0;
while ($#threads >=0 ) {
	my $worker = shift(@threads);
	print "Shutdown of syncer ...\n" if $debug;
	$worker->join();
}
print "Shutdown clean completed\n" if $debug;
exit(0);
 
########################################################################################################################################
########################################################################################################################################
 
####################################################################################################
# audit thread
sub audit {
	my $r = shift;
	my $sp = shift;
	my $c = shift;
	my $p = shift;
	my $m = shift;
	my $d = shift;
 
	print "       audit started ...\n" if $d;
	# open listener on the audit device
	open(STATUS, "/usr/sbin/praudit /dev/auditpipe |") || die "can't fork: $!";
	while (<STATUS>) {
		my $line = $_;
		last if ($line eq '' || $r->{'status'}<=0);		# Terminate if audit terminated
		if ($line =~ /path,$p(.+)/) {					# Check if the changed file is in the observed path
			my $file = $1;
			print "Change detected on file: $file\n" if $d;
			my $hash :shared;							# Create a command for the syncers
			$hash = &share({});
			$hash->{'file'} = $file;
			$hash->{'status'} = '';
			$hash->{'time'} = '';
			for (my $j=0;$j<=$m;$j++) {					# init job done charta
				$hash->{$j} = 'no';
			}
			if (1) {
				lock($c);
				push(@{$c},$hash);
				print "Added new job for $file\n" if $d;
			}
			for (my $j=0;$j<=$m;$j++) {					# wakeup syncers
				$sp->{$j}->up();
			}
			$sp->{'checker'}->up();
		}
	}
	close STATUS || die "audit not closed correctly: $! $?";	
	print "       audit terminated ...\n" if $d;
	return(0);
}
 
####################################################################################################
# syncer thread
sub syncJob {
	my $r = shift;
	my $sp = shift;
	my $id = shift;
	my $c = shift;
	my $p = shift;
	my $e = shift;
	my $d = shift;
 
	print "       syncer $id started ...\n" if $d;
 
	while ($r->{'status'}>0) {
		if ($#{$c}>=0) {													# if there are any jobs to be done
			for (my $j=0; $j<=$#{$c}; $j++) {
				next if ($c->[$j]->{$id} eq 'ok');							# if my job is already done skip this job and check next
				my $file = $c->[$j]->{'file'};
				if (-e $p.$file)  {											# check if the file is existing
					$c->[$j]->{$id} = 'working';							# mark this job as being worked on
					my $dif = 1;
					while ($dif>0) {										# check if the file is in upload and changes size within 1,5 secs
						print "Checking Filesize ...\n" if $d;
						my $ssize = -s $p.$file;
						sleep(1.5);
						my $eesize = -s $p.$file;
						$dif = $eesize - $ssize;
						print "Checking Filesize $ssize - $eesize = $dif\n" if $d;
					}
					my $cm = $e;
					$cm =~ s/<!--target-->/$file/g;
					system($cm);											# rsync to other server
				}
				lock($c);
				$c->[$j]->{$id} = 'ok';										# mark job as done for me
			}
		}
		$sp->{$id}->down();
	}
	print "       syncer $id terminated ...\n" if $d;
	return(0);
}
 
####################################################################################################
# checker thread that checks if all jobs are done
sub JobChecker {
	my $r = shift;
	my $sp = shift;
	my $c = shift;
	my $m = shift;
	my $d = shift;
 
	print "       checker started ...\n" if $d;
 
	while ($r->{'status'}>0) {
		while ($#{$c} >= 0 && $r->{'status'}>0) {
			print "    Checker loop ...\n" if $d;
			my $rem = 0;
			foreach my $job (@{$c}) {						# loop through all jobs
				my $mem = 'ok';
				for (my $j=0;$j<=$m;$j++) {					# check job done charta
					if ($job->{$j} eq 'no') {				# job not handled
						$mem = 'no' if ($mem ne 'working');	# job not handled (may never override a job in progress state)
					} elsif ($job->{$j} eq 'working') {		# job in progress (always overrides not handled)
						$mem = 'working';
					}
				}
				# Job not completed
				if ($mem eq 'no') {							
					if ($job->{'time'} eq '') {				# Set timestamp to know, how long this job is already waiting
						$job->{'time'} = time;
					} else {								# Job already got a timestamp
						my $watch = time - $job->{'time'};
						print "Job age: $watch\n" if $d;
						if (time - $job->{'time'} > 300) {	# Job has waited for more than 5 minutes. terminate program
							print "TIME FOR JOB EXCEEDED - shutting down syncer";
							$r->{'status'} = 0;
							for (my $j=0;$j<=$m;$j++) {		# wakeup syncers
								$sp->{$j}->up();
								$sp->{'checker'}->up();		# wakeup ourself
							}
						}
					}
				} elsif ($mem eq 'working') {				# job in progress - just actualize the timestamp
					$job->{'time'} = time;
				} else {
					$job->{'status'} = 'complete';			# job is completely done and is marked for being removed
					$rem = 1;
				}
			}
			# Job to remove available
			if ($rem > 0) {
				lock($c);									# lock the command-queue
				my @arr;
				for (my $j=0;$j<=$#{$c};$j++) {				# store all not handled jobs / drop completed jobs
					my $ex = shift(@{$c});
					if ($ex->{'status'} ne 'complete') {
						push(@arr,$ex);
					}
				}
				for (my $j=0;$j<=$#arr;$j++) {				# put all stored (not finished) jobs back into the command queue
					push(@{$c},$arr[$j]);
				}
			}
			print "    Checker reloop ...\n" if $d;
			sleep(1);
		}
		print "    Checker sleeping (".$#{$c}.")...\n" if $d;
		$sp->{'checker'}->down();
	}
 
	print "       checker terminated ...\n" if $d;
	return(0);
}