In-stream compression now working in copy.

lib/BackRest/File.pm

@@ -699,6 +699,72 @@ sub move
 #     }
 # }
 
+####################################################################################################################################
+# EXISTS - Checks for the existence of a file, but does not imply that the file is readable/writeable.
+#
+# Return: true if file exists, false otherwise
+####################################################################################################################################
+sub exists
+{
+    my $self = shift;
+    my $strPathType = shift;
+    my $strPath = shift;
+
+    # Set operation variables
+    my $strPathOp = $self->path_get($strPathType, $strPath);
+
+    # Set operation and debug strings
+    my $strOperation = OP_FILE_EXISTS;
+    my $strDebug = "${strPathType}:${strPathOp}";
+
+    # Run remotely
+    if ($self->is_remote($strPathType))
+    {
+        # Build param hash
+        my %oParamHash;
+
+        $oParamHash{path} = ${strPathOp};
+
+        # Build debug string
+        $strDebug = "${strOperation}: remote (" . $self->{oRemote}->command_param_string(\%oParamHash) . "): " . $strDebug;
+        &log(DEBUG, $strDebug);
+
+        # Execute the command
+        return $self->{oRemote}->command_execute($strOperation, \%oParamHash, true, $strDebug) eq "Y";
+    }
+    # Run locally
+    else
+    {
+        # Build debug string
+        $strDebug = "${strOperation}: local: " . $strDebug;
+        &log(DEBUG, ${strDebug});
+
+        # Stat the file/path to determine if it exists
+        my $oStat = lstat($strPathOp);
+
+        # Evaluate error
+        if (!defined($oStat))
+        {
+            # If the error is not entry missing, then throw error
+            if (!$!{ENOENT})
+            {
+                if ($strPathType eq PATH_ABSOLUTE)
+                {
+                    confess &log(ERROR, $!, COMMAND_ERR_FILE_READ);
+                }
+                else
+                {
+                    confess &log(ERROR, "${strDebug}: " . $!, COMMAND_ERR_FILE_READ);
+                }
+            }
+
+            return false;
+        }
+    }
+
+    return true;
+}
+
 ####################################################################################################################################
 # COPY
 #
@@ -724,12 +790,12 @@ sub copy
     my $strPermission = shift;
 
     # Set defaults
+    my $bIsCompressed = false;
     $bCompress = defined($bCompress) ? $bCompress : defined($self->{bCompress}) ? $self->{bCompress} : true;
     $bIgnoreMissingSource = defined($bIgnoreMissingSource) ? $bIgnoreMissingSource : false;
     $bPathCreate = defined($bPathCreate) ? $bPathCreate : false;
 
     # Set working variables
-    my $strErrorPrefix = "File->copy";
     my $bSourceRemote = $self->is_remote($strSourcePathType) || $strSourcePathType eq PIPE_STDIN;
     my $bDestinationRemote = $self->is_remote($strDestinationPathType) || $strDestinationPathType eq PIPE_STDOUT;
     my $strSourceOp = $strSourcePathType eq PIPE_STDIN ?
@@ -739,22 +805,22 @@ sub copy
     my $strDestinationTmpOp = $strDestinationPathType eq PIPE_STDOUT ?
         undef : $self->path_get($strDestinationPathType, $strDestinationFile, true);
 
-    # Determine if the file needs compression extension
+    # Set operation and debug string
-    if ($bCompress && $strDestinationOp !~ "^.*\.$self->{strCompressExtension}\$")
+    my $strOperation = "File->unknown";
-    {
+    my $strDebug = ($bSourceRemote ? " remote" : " local") . " ${strSourcePathType}" .
-        $strDestinationOp .= "." . $self->{strCompressExtension};
+                   (defined($strSourceFile) ? ":${strSourceFile}" : "") .
-    }
+                   " to" . ($bDestinationRemote ? " remote" : " local") . " ${strDestinationPathType}" .
-
+                   (defined($strDestinationFile) ? ":${strDestinationFile}" : "") .
-    # Output trace info
+                   ", compress = " . ($bCompress ? "true" : "false");
-#    &log(TRACE, "${strErrorPrefix}:" . ($bSourceRemote ? " remote" : " local") . " ${strSourcePathType}:${strSourceFile}" .
-#                " to" . ($bDestinationRemote ? " remote" : " local") . " ${strDestinationPathType}:${strDestinationFile}" .
-#                ", compress = " . ($bCompress ? "true" : "false"));
 
     # Open the source file
     my $hSourceFile;
 
     if (!$bSourceRemote)
     {
+        # Determine if the file is compressed
+        $bIsCompressed = $strSourceOp !~ "^.*\.$self->{strCompressExtension}\$";
+
         open($hSourceFile, "<", $strSourceOp)
             or confess &log(ERROR, "cannot open ${strSourceOp}: " . $!);
     }
@@ -764,6 +830,12 @@ sub copy
 
     if (!$bDestinationRemote)
     {
+        # Determine if the file needs compression extension
+        if ($bCompress && $strDestinationOp !~ "^.*\.$self->{strCompressExtension}\$")
+        {
+            $strDestinationOp .= "." . $self->{strCompressExtension};
+        }
+
         open($hDestinationFile, ">", $strDestinationTmpOp)
             or confess &log(ERROR, "cannot open ${strDestinationTmpOp}: " . $!);
     }
@@ -771,10 +843,8 @@ sub copy
     # If source or destination are remote
     if ($bSourceRemote || $bDestinationRemote)
     {
-#        print "got outside\n";
         # Build the command and open the local file
         my $hFile;
-        my $strOperation;
         my %oParamHash;
         my $hIn,
         my $hOut;
@@ -784,17 +854,26 @@ sub copy
         if ($bSourceRemote && !$bDestinationRemote)
         {
             $hOut = $hDestinationFile;
+            $strOperation = OP_FILE_COPY_OUT;
+
+            $bCompress = $bCompress ? undef : $bCompress;
 
             if ($strSourcePathType eq PIPE_STDIN)
             {
                 $strRemote = 'in';
                 $hIn = *STDIN;
+                
             }
             else
             {
                 $strRemote = 'in';
-                $strOperation = OP_FILE_COPY_OUT;
                 $oParamHash{source_file} = ${strSourceOp};
+
+                if (defined($bCompress))
+                {
+                    $oParamHash{compress} = $bCompress;
+                }
+                
                 $hIn = $self->{oRemote}->{hOut};
             }
         }
@@ -802,6 +881,7 @@ sub copy
         elsif (!$bSourceRemote && $bDestinationRemote)
         {
             $hIn = $hSourceFile;
+            $strOperation = OP_FILE_COPY_IN;
 
             if ($strDestinationPathType eq PIPE_STDOUT)
             {
@@ -811,14 +891,19 @@ sub copy
             else
             {
                 $strRemote = 'out';
-                $strOperation = OP_FILE_COPY_IN;
                 $oParamHash{destination_file} = ${strDestinationOp};
+
+                $bCompress = $bCompress ? undef : $bCompress;
+
+                if (defined($bCompress))
+                {
+                    $oParamHash{compress} = $bCompress;
+                }
+
                 $hOut = $self->{oRemote}->{hIn};
             }
             
-            # Build debug string
+            $bCompress = true;
-#            $strDebug = "${strOperation}: remote (" . $self->{oRemote}->command_param_string(\%oParamHash) . "): " . $strDebug;
-#            &log(DEBUG, $strDebug);
         }
         # Else source and destination are remote
         else
@@ -832,28 +917,25 @@ sub copy
             return false;
         }
 
-        # If an operation is defined then write it
+        # Build debug string
-        if (defined($strOperation))
+        $strDebug = "${strOperation}: " . (%oParamHash ? "remote (" .
-        {
+                    $self->{oRemote}->command_param_string(\%oParamHash) . ") :" : "") . $strDebug;
-            # Trace command
+        &log(DEBUG, $strDebug);
-            &log(TRACE, "${strErrorPrefix} operation:" . $strOperation);
 
-            # Execute the operation
+        # If an operation is defined then write it
+        if (%oParamHash)
+        {
             $self->{oRemote}->command_write($strOperation, \%oParamHash);
         }
 
         # Transfer the file
-#        print "binary xfer start\n";
+        $self->{oRemote}->binary_xfer($hIn, $hOut, $strRemote, $bCompress);
-        $self->{oRemote}->binary_xfer($hIn, $hOut, $strRemote);
-#        print "binary xfer stop\n";
 
-        if ($strSourcePathType ne PIPE_STDIN && $strDestinationPathType ne PIPE_STDOUT)
+        # If this is the controlling process then wait for OK from remote
+        if (%oParamHash)
         {
-            $self->{oRemote}->output_read(false, $strErrorPrefix);
+            $self->{oRemote}->output_read(false, $strDebug);
         }
-        
-        # Wait for process exit (and error)
-#        $self->wait_pid($pId, $strCommand, $hIn, $hOut, $hErr);
     }
     else
     {
@@ -890,7 +972,6 @@ sub copy
                 or confess &log(ERROR, "unable to set time for local ${strDestinationTmpOp}");
         }
 
-
         # Move the file from tmp to final destination
         $self->move(PATH_ABSOLUTE, $strDestinationTmpOp, PATH_ABSOLUTE, $strDestinationOp, $bPathCreate);
     }
@@ -1114,72 +1195,6 @@ sub list
     return @stryFileList;
 }
 
-####################################################################################################################################
-# EXISTS - Checks for the existence of a file, but does not imply that the file is readable/writeable.
-#
-# Return: true if file exists, false otherwise
-####################################################################################################################################
-sub exists
-{
-    my $self = shift;
-    my $strPathType = shift;
-    my $strPath = shift;
-
-    # Set operation variables
-    my $strPathOp = $self->path_get($strPathType, $strPath);
-
-    # Set operation and debug strings
-    my $strOperation = OP_FILE_EXISTS;
-    my $strDebug = "${strPathType}:${strPathOp}";
-
-    # Run remotely
-    if ($self->is_remote($strPathType))
-    {
-        # Build param hash
-        my %oParamHash;
-
-        $oParamHash{path} = ${strPathOp};
-
-        # Build debug string
-        $strDebug = "${strOperation}: remote (" . $self->{oRemote}->command_param_string(\%oParamHash) . "): " . $strDebug;
-        &log(DEBUG, $strDebug);
-
-        # Execute the command
-        return $self->{oRemote}->command_execute($strOperation, \%oParamHash, true, $strDebug) eq "Y";
-    }
-    # Run locally
-    else
-    {
-        # Build debug string
-        $strDebug = "${strOperation}: local: " . $strDebug;
-        &log(DEBUG, ${strDebug});
-
-        # Stat the file/path to determine if it exists
-        my $oStat = lstat($strPathOp);
-
-        # Evaluate error
-        if (!defined($oStat))
-        {
-            # If the error is not entry missing, then throw error
-            if (!$!{ENOENT})
-            {
-                if ($strPathType eq PATH_ABSOLUTE)
-                {
-                    confess &log(ERROR, $!, COMMAND_ERR_FILE_READ);
-                }
-                else
-                {
-                    confess &log(ERROR, "${strDebug}: " . $!, COMMAND_ERR_FILE_READ);
-                }
-            }
-
-            return false;
-        }
-    }
-
-    return true;
-}
-
 ####################################################################################################################################
 # REMOVE
 ####################################################################################################################################

lib/BackRest/Remote.pm

@@ -12,6 +12,8 @@ use Moose;
 use Net::OpenSSH;
 use File::Basename;
 use POSIX ":sys_wait_h";
+use IO::Compress::Gzip qw(gzip $GzipError);
+use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
  
 use lib dirname($0) . "/../lib";
 use BackRest::Exception;
@@ -22,7 +24,7 @@ use BackRest::Utility;
 ####################################################################################################################################
 use constant
 {
-    DEFAULT_BLOCK_SIZE  => 8192
+    DEFAULT_BLOCK_SIZE  => 4
 };
 
 ####################################################################################################################################
@@ -301,21 +303,74 @@ sub binary_xfer
     $strRemote = defined($strRemote) ? $strRemote : 'none';
 
     my $iBlockSize = $self->{iBlockSize};
-    my $iHeaderSize = 12;
     my $iBlockIn;
     my $iBlockOut;
     my $strBlockHeader;
     my $strBlock;
+    my $oGzip;
+    my $hPipeIn;
+    my $hPipeOut;
+    my $pId;
 
     if (!defined($hIn) || !defined($hOut))
     {
         confess &log(ASSERT, "hIn or hOut is not defined");
     }
 
+    if ($strRemote eq "out")
+    {
+        pipe $hPipeOut, $hPipeIn;
+        
+        # fork and exit the parent process
+        $pId = fork();
+        
+        if (!$pId)
+        {
+            close($hPipeOut);
+
+            gzip($hIn => $hPipeIn)
+                or die confess &log(ERROR, "unable to compress: " . $GzipError);
+                
+            close($hPipeIn);
+    
+            exit 0;
+        }
+        
+        close($hPipeIn);
+    
+        $hIn = $hPipeOut;
+    }
+    elsif ($strRemote eq "in" && defined($bCompress) && !$bCompress)
+    {
+        pipe $hPipeOut, $hPipeIn;
+
+        # fork and exit the parent process
+        $pId = fork();
+        
+        if (!$pId)
+        {
+            close($hPipeIn);
+
+            gunzip($hPipeOut => $hOut)
+                or die confess &log(ERROR, "unable to uncompress: " . $GunzipError);
+
+            close($hPipeOut);
+
+            exit 0;
+        }
+
+#        exit 0;
+        
+        close($hPipeOut);
+        
+        $hOut = $hPipeIn;
+    }
+
     while (1)
     {
         if ($strRemote eq 'in')
         {
+        
             $strBlockHeader = $self->read_line($hIn);
 
             if ($strBlockHeader !~ /^block [0-9]+$/)
@@ -325,6 +380,8 @@ sub binary_xfer
             
             $iBlockSize = trim(substr($strBlockHeader, index($strBlockHeader, " ") + 1));
 
+#            confess "found $iBlockSize to write";
+
             if ($iBlockSize != 0)
             {
                 $iBlockIn = sysread($hIn, $strBlock, $iBlockSize);
@@ -336,27 +393,65 @@ sub binary_xfer
             }
             else
             {
-                # 
-                # if ($strRemote eq 'in')
-                # {
-                #     confess &log(ERROR, "block size $iBlockSize");
-                # }
                 $iBlockIn = 0;
             }
         }
         else
-        {
-            if (!defined($bCompress))
         {
             $iBlockIn = sysread($hIn, $strBlock, $iBlockSize);
             
             if (!defined($iBlockIn))
             {
-                    confess "unable to read ${iBlockSize} bytes from remote: " . $!;
+                confess &log(ERROR, "unable to read");
-                }
             }
         }
 
+#         if (defined($bCompress))
+#         {
+#             if ($iBlockIn > 0)
+#             {
+#                 my $iBlockInTmp = $iBlockIn;
+#             
+# #                print "${iBlockIn} bytes to compress";
+#             
+#                 if ($bCompress)
+#                 {
+#                     $iBlockIn = $oGzip->syswrite($hIn, $iBlockInTmp);
+# 
+#                     if (!defined($iBlockIn))
+#                     {
+#                         confess &log(ERROR, "unable to compress stream: " . $GunzipError)
+#                     }
+#             
+#                     if ($iBlockIn != $iBlockInTmp)
+#                     {
+#                         confess &log(ERROR, "unable to read ${iBlockSize} bytes");
+#                     }
+#                 }
+#                 else
+#                 {
+#                     $iBlockIn = $oGzip->sysread($strBlock, 8192);
+#                     
+#                     if (!defined($iBlockIn))
+#                     {
+#                         confess &log(ERROR, "unable to read compressed stream: " . $GunzipError)
+#                     }
+#                     # $strBlock = $strBlockTmp;
+#                 }
+#             }
+#             else
+#             {
+#                 if ($bCompress)
+#                 {
+#                     $oGzip->flush()
+#                         or confess &log(ERROR, "unable to flush compressed stream");
+#                     $bDone = true;
+#                 }
+#             }
+# 
+#             $iBlockIn = length($strBlock);
+#         }
+
         if ($strRemote eq 'out')
         {
             $strBlockHeader = "block ${iBlockIn}\n";
@@ -382,12 +477,63 @@ sub binary_xfer
             {
                 confess "unable to write ${iBlockIn} bytes" . (defined($!) ? ": " . $! : "");
             }
+            
+#            $strBlock = undef;
+
+#             if ($bDone && $strRemote eq 'out')
+#             {
+#                 $strBlockHeader = "block 0\n";
+# 
+# #                print "wrote block header: ${strBlockHeader}"; # REMOVE!
+# 
+#                 $iBlockOut = syswrite($hOut, $strBlockHeader);
+# 
+#                 if (!defined($iBlockOut) || $iBlockOut != length($strBlockHeader))
+#                 {
+#                     confess "unable to write block header";
+#                 }
+#             }
         }
         else
         {
             last;
+            
+            # if ($strRemote eq 'in')
+            # {
+            #     confess "got out\n";
+            # }
+            #     
+            # if (defined($hPipeIn))
+            # {
+            #     close($hPipeIn)
+            # };
+            # 
+            # $bDone = true;
         }
     }
+
+    if (defined($pId))
+    {
+        if ($strRemote eq "out")
+        {
+            close($hPipeOut);
+        }
+        elsif ($strRemote eq "in" && defined($bCompress) && !$bCompress)
+        {
+#            confess "waiting for child";
+            close($hPipeIn);
+        }
+
+        waitpid($pId, 0);
+        my $iExitStatus = ${^CHILD_ERROR_NATIVE} >> 8;
+                
+        if ($iExitStatus != 0)
+        {
+            confess &log(ERROR, "compression/decompression child process returned " . $iExitStatus);
+        }
+    }
+    
+#    print "got out\n";
 }
 
 ####################################################################################################################################