WebSVN – DevTools – Diff – /dpkg_blat/trunk/src/blat/blatDaemon.pl

     filtered => 0,                      # Packages filtered out
     missing  => 0,                      # Packages missing
     transfer => 0,                      # Packages to transfer
     writable => 0,                      # Packages still writable - thus not transferred
     tagCount => 0,                      # Packages tagged to be transferred
+    tagDelCount => 0,                   # Packages tagged to be deleted
+                                        #
                                         # Expected from the Target
 #   Target.Hostname => '',              # Target Hostname
 #   Target.avail    => 0,               # Information from 'df' 1Kblocks
 #   Target.pcent    => 0,
     'packageFilter'   => {'default'   => undef  , 'fmt' => 'text'},
     'active'          => {'default'   => 1      , 'fmt' => 'bool'},
     'debug'           => {'default'   => 0      , 'fmt' => 'bool'},                 # Log to screen
     'txdetail'        => {'default'   => 0      , 'fmt' => 'bool'},
     'noTransfers'     => {'default'   => 0      , 'fmt' => 'bool'},                 # Debugging option to prevent transfers
+    'maxTarZips'      => {'default'   => 5      , 'fmt' => 'int'},
+    #
     #   Transfer via ssh
     #   Cannot be used in conjunction with S3Bucket
+    #
     my $txcount = $conf->{maxpackages};
+    #
     #   Transfer packages that we have identified
+    #
-    send_pkgs:
     while ( (my ($pname, $pvers)) = each %{$needPkgList} )
+    {
         while ( (my ($pver, $pdata) ) = each %{$pvers} )
+        {
-            if ( --$txcount <= 0 )
-            {
-                $logger->warn("Max transfer count exceeded: $needPkgListCount transfer remaining");
-                $lastReleaseScan = 0;
-                last send_pkgs;
-            }
-            if ( readConfig() )
-            {
-                $logger->warn("Config file changed");
-                $txcount = 0;
-                last send_pkgs;
-            }
-            transferPackage ($pname, $pver);
+            tagForTransfer($pname, $pver);
             $needPkgListCount--;
-            reapChildren();
+        }
+    }
+    #
     #   Delete packages that have been identified as excess
+    #
-    delete_pkgs:
     while ( (my ($pname, $pvers)) = each %{$excessPkgList} )
+    {
         while ( (my ($pver, $pdata) ) = each %{$pvers} )
+        {
+            tagForDelete ($pname, $pver, $pdata);
-            if ( --$txcount <= 0 )
+            $excessPkgListCount--;
+        }
+    }
+    #
+    #   Need to transmission remove tags for packages that don't need any more
+    #   ie: Tags that have since been made unnessesary
+    #
+    my $taggedPackages = getTaggedPackages();
+    if ($taggedPackages) {
+        #
+        #   Mark entries thay we still need
+        #
+        foreach my $pname ( keys %{$pkgList} )
+        {
+            foreach my $pver ( keys %{$pkgList->{$pname}} )
+            {
-                $logger->warn("Max transfer count exceeded: $excessPkgListCount deletion remaining");
+                if (exists $taggedPackages->{$pname}{$pver}) {
-                $lastReleaseScan = 0;
+                    $taggedPackages->{$pname}{$pver} = 1;
-                last delete_pkgs;
+                }
+            }
+        }
-            if ( readConfig() )
-            {
+        #
+        #   Sweep out entries that we don't need any more
+        foreach my $pname ( keys %{$taggedPackages} ) {
+            foreach my $pver ( keys %{$taggedPackages->{$pname}} ) {
-                $logger->warn("Config file changed");
+                unless ($taggedPackages->{$pname}{$pver}) {
+$logger->verbose("Untag: $pname $pver");
-                $txcount = 0;
+                    unTagForTransfer($pname, $pver);
-                last delete_pkgs;
+                }
+            }
-            deletePackage ($pname, $pver, $pdata);
-            $excessPkgListCount--;
-            reapChildren();
+        }
-    }
+    }
+    #
     #   Send package list to the target
+    #
     sendPackageList ($pkgList);
     $tagDirTime = 0;
     $releaseScanMode = 0;
+}
 #-------------------------------------------------------------------------------
+# Function        : tagForTransfer
+#
+# Description     : Tag a package to be transferred
+#
+# Inputs          : $pname
+#                   $pvers
+#
+# Returns         :
+#
+sub tagForTransfer
+{
+    my ($pname, $pver) = @_;
+    my $tag = "$pname::$pver";
+    my $myTag =  catfile($conf->{'tagdir'} , $tag);
+    unless (-f $myTag ) {
+        $logger->verbose2("tagForTransfer: $pname, $pver");
+        Utils::TouchFile($conf, $myTag);
+    }
+    unlink 'DELD::' . $myTag;
+    unlink 'DELF::' . $myTag;
+}
+#-------------------------------------------------------------------------------
+# Function        : unTagForTransfer
+#
+# Description     : Un Tag a package to be transferred
+#
+# Inputs          : $pname
+#                   $pvers
+#
+# Returns         :
+#
+sub unTagForTransfer
+{
+    my ($pname, $pver) = @_;
+    my $tag = "$pname::$pver";
+    my $myTag =  catfile($conf->{'tagdir'} , $tag);
+    if (-f $myTag ) {
+        $logger->verbose2("UnTagForTransfer: $pname, $pver");
+        unlink($myTag);
+    }
+}
+#-------------------------------------------------------------------------------
+# Function        : tagForDelete
+#
+# Description     : Tag a package to be transferred
+#                   Generate tags of the form
+#                       DELD::pname::pver   - Delayed Delete
+#                       DELF::pname::pver   - Forced Delete
+#
+# Inputs          : $pname
+#                   $pver
+#                   $pdata  - Type of delete
+#
+# Returns         :
+#
+sub tagForDelete
+{
+    my ($pname, $pver, $pdata) = @_;
+    my $tag = "$pname::$pver";
+    my $myTag =  catfile($conf->{'tagdir'} , $tag);
+    my $delType = $pdata->{FORCEDELETE}  ? 'D' : 'F';
+    my $myDelTag =  'DEL' . $delType . '::' . $myTag;
+    unless (-f $myDelTag ) {
+        $logger->verbose2("tagForDelete: $pname, $pver");
+        Utils::TouchFile($conf, $myDelTag);
+    }
+    unlink $myTag;
+}
+#-------------------------------------------------------------------------------
+# Function        : getTaggedPackages
+#
+# Description     : Determine the tagged packages
+#
+# Inputs          : None
+#
+# Returns         : Returns a pointer to a hash of tagged packages of the form
+#
+sub getTaggedPackages
+{
+    my $taggedPackages = {};
+    foreach ( glob ("$conf->{'tagdir'}/*::*"))
+    {
+        next if m~/DEL.::~;
+        m~.*/(.*)::(.*)~;
+        $taggedPackages->{$1}{$2} = 0;
+    }
+    return $taggedPackages;
+}
+#-------------------------------------------------------------------------------
 # Function        : sendPackageList
+#
 # Description     : Transfer package list to the target
+#
 # Inputs          : $pkgList            - Ref to hash of package names and versions
+    #
     #   Get list of things
+    #
     my %config;
-    if ($conf->{'allArchive'} )
-    {
+    #
-        $config{allArchive} = 1
+    #   Is Tag Processing active
-    }
-    elsif ($conf->{'allProjects'} )
+    #   Can configure blat to disable tag sync
-    {
+    #
-        $config{allProjects} = 1;
+    if ( $conf->{'tagMaxPackages'} > 0 )
-    }
-    else
+    {
+        if ($conf->{'allArchive'} )
+        {
+            $config{allArchive} = 1
+        }
+        elsif ($conf->{'allProjects'} )
+        {
+            $config{allProjects} = 1;
+        }
+        else
+        {
-        %{$config{projects}} = map { $_ => 1 } @projectList;
+            %{$config{projects}} = map { $_ => 1 } @projectList;
-        %{$config{releases}} = map { $_ => 1 } getReleaseList();
+            %{$config{releases}} = map { $_ => 1 } getReleaseList();
+        }
+    } else {
+        $config{disableTagTx} = 1
+    }
+    #
     #   Save data
+    #
 # Function        : processTags
+#
 # Description     : Process tags and send marked package versions to the target
 #                       Determine if new tags are present
 #                       Process each tag
+#
+#                       Two types of tag
+#                           Transfer Requests
+#                           Delete Request
+#                       Send packages before deleting packages
+#
 # Inputs          : None
+#
 # Returns         : Nothing
+#
 sub processTags
+{
+    #
-    #   Is Tag Processing active
-    #   Can configure blat to disable tag sync
-    #
-    if ( $conf->{'tagMaxPackages'} == 0 )
-    {
-        $logger->verbose2("processTags disabled");
-        return;
-    }
-    #
     #   Determine if new tags are present by examining the time
     #   that the directory was last modified.
+    #
     #   Allow for a forced scan to catch packages that did not transfer
     #   on the first attempt
+    #
     my $tagCount = 0;
+    my $delCount = 0;
     my ($mtime) = Utils::mtime($conf->{'tagdir'} );
     if ( ($mtime > $tagDirTime) || ($now > ($lastDirScan + $conf->{'forcedirscan'})) )
+    {
         $logger->verbose2("processTags: ",$conf->{'tagdir'}, $mtime - $tagDirTime, $now - $lastDirScan);
         $tagDirTime = $mtime;
         $lastDirScan = $now;
         my $txcount = $conf->{'tagMaxPackages'};
         my $dh;
         unless (opendir($dh, $conf->{'tagdir'}))
+        {
             $logger->warn ("can't opendir $conf->{'tagdir'}: $!");
             return;
+        #
         #   Process each entry
         #   Ignore those that start with a .
+        #
         my %tagPkgList;
+        my %deleteTags;
         while (my $tag = readdir($dh) )
+        {
             next if ( $tag =~ m~^\.~ );
             my $file = "$conf->{'tagdir'}/$tag";
             $logger->verbose3("processTags: $file");
             next unless ( -f $file );
             next if ( $tag  eq 'ReleaseList' );
+            if ( $tag  =~ m~^DEL(.)::(.+)::(.+)~) {
+                $deleteTags {$2}{$3}{file} = $file;
+                $deleteTags {$2}{$3}{mode} = $1 eq 'F' ? 1 : 0;
+                $delCount++;
+            }
             if ( $tag =~ m~(.+)::(.+)~  )
+            {
-                my $package = $1;
-                my $version = $2;
+                $tagPkgList{$1}{$2} = $file;
                 $tagCount++;
-                $tagPkgList{$package}{$version} = $file;
+            }
+        }
         $statistics{tagCount} = $tagCount;
+        $statistics{tagDelCount} = $delCount;
         closedir $dh;
+        #
+        #   Process delete requests after all transfers have occured
+        #
+        unless ($tagCount) {
+            delete_pkgs:
+            while ( (my ($pname, $pvers)) = each %deleteTags )
+            {
+                while ( (my ($pver, $pdata) ) = each %{$pvers} )
+                {
+                    if ( --$txcount <= 0 )
+                    {
+                        $logger->warn("Max tag transfer count exceeded: $delCount deletion remaining");
+                        $tagDirTime = 0;
+                        last delete_pkgs;
+                    }
+                    if ( readConfig() )
+                    {
+                        $logger->warn("Config file changed");
+                        $txcount = 0;
+                        $tagDirTime = 0;
+                        last delete_pkgs;
+                    }
+                    deletePackage ($pname, $pver, $pdata->{mode});
+                    unlink $pdata->{file};
+                    $delCount--;
+                    reapChildren();
+                }
+            }
+        }
+        #
         #   Process the packages located in the tags area
+        #
         send_tags:
         while ( (my ($package, $pvers)) = each %{tagPkgList} )
+        {
             while ( (my ($version, $file) ) = each %{$pvers} )
+            {
                 if ( --$txcount <= 0 )
+                {
                     $logger->warn("Max tag transfer count exceeded: $tagCount transfer remaining");
+                    $logger->warn("Max tag transfer count exceeded: $delCount deletion remaining");
                     $tagDirTime = 0;
                     last send_tags;
+                }
                 if ( readConfig() )
+        }
         LogTxError ($?);
+    }
+}
+#-------------------------------------------------------------------------------
+# Function        : calcZipRequests
+#
+# Description     : Determine the number of packages that I need that have
+#                   outstanding ZIP requests
+#
+#                   Purpose is to limit the number of ZIP requests that are
+#                   outstanding at any one time to reduce disc space
+#
+#                   ie: Iff there 1000 outstanding transfers, we don't need to zip all
+#                   of the packages right now.
+#
+# Inputs          :
+#
+# Returns         :
+#
+#
+sub calcZipRequests
+{
+    my $zipRequests = 0;
+    $conf->{'tagdir'} =~ m~^(.*)/~;
+    my $tagRoot = $1;
+    my $dh;
+    if ( opendir($dh, $conf->{'tagdir'}) )
+    {
+        while (my $tag = readdir($dh) )
+        {
+            next if ( $tag =~ m~^\.~ );
+            my $file = "$conf->{'tagdir'}/$tag";
+            $logger->verbose3("calcZipRequests: $file");
+            next unless ( -f $file );
+            next if ( $tag  eq 'ReleaseList' );
+            next if ( $tag =~ m~^DEL(.)::~);
+            my $zipTag = catfile($tagRoot, 'tarZip', $tag);
+            if (-f $zipTag ) {
+                $zipRequests++;
+            }
+        }
+        closedir $dh;
+    }
+    $logger->verbose("zipRequests: $zipRequests");
+    return $zipRequests;
+}
 #-------------------------------------------------------------------------------
 # Function        : transferPackage
+#
 # Description     : Transfer specified package to target system
     my $tzdir = catfile( $conf->{'dpkg_archive'} , '.dpkg_archive', 'tarStore' );
     my $tzfile = $pname . '__' . $pver . '.tgz';
     my $tzpath = catfile($tzdir, $tzfile);
     unless (-f $tzpath) {
-        $logger->verbose("transferPackage: tarZip not found - $tzpath");
+        $logger->verbose("transferPackage: tarZip not found - $tzfile");
         $conf->{'tagdir'} =~ m~^(.*)/~;
         my $tagRoot = $1;
         my $tag = "$pname::$pver";
         my $zipTag = catfile($tagRoot, 'tarZip', $tag);
         my $myTag =  catfile($conf->{'tagdir'} , $tag);
         $logger->logmsg("transferPackage. Request Zip: @_") unless ( -f $zipTag && -f $myTag );
         Utils::TouchFile($conf, $myTag) unless -f $myTag;
+        unless (-f $zipTag) {
+            if (calcZipRequests() < $conf->{maxTarZips} ) {
-        Utils::TouchFile($conf, $zipTag) unless -f $zipTag;
+                Utils::TouchFile($conf, $zipTag);
+            } else {
+                $logger->verbose("transferPackage: Max outstanding tarZip Requests");
+            }
+        }
         return 0;
+    }
     ###########################################################################
+#
 # Description     : Delete specified package to target system
+#
 # Inputs          : $pname          - Name of the package
 #                   $pver           - Package version
-#                   $pdata          - Hash of extra data
+#                   $mode           - 1 ForcedDelete 0: Tag for delayed Delete
+#
 # Returns         : true    - Package deleted
 #                   false   - Package not deleted
+#
 sub deletePackage
+{
-    my ($pname, $pver, $pdata ) = @_;
+    my ($pname, $pver, $mode ) = @_;
     my $rv = 0;
     my $cmdRv = 0;
     $logger->logmsg("deletePackage: $pname, $pver");
     if ($isS3Target) {
         #   Create the process pipe to delete the package
         #   ssh  ... "./delete_package ${rx_opts} \"$pname\" \"$pver\""
+        #
         unless ($conf->{'noTransfers'}) {
             my $ph;
-            my $flags = $pdata->{FORCEDELETE}  ? '' : ' -T';
+            my $flags = $mode  ? '-T' : '';
             my $tgt_cmd = "$conf->{'bindir'}/delete_package $flags \"$pname\" \"$pver\"";
             my $ssh_cmd = sshCmd($tgt_cmd);
             $logger->verbose2("deletePackage:tgt_cmd:$tgt_cmd");
             $logger->verbose2("deletePackage:ssh_cmd:$ssh_cmd");

Subversion Repositories DevTools

(root)/dpkg_blat/trunk/src/blat/blatDaemon.pl – Rev 7389 → 7394