#region Disclaimer / License // Copyright (C) 2010, Kenneth Skovhede // http://www.hexad.dk, opensource@hexad.dk // // This library is free software; you can redistribute it and/or // modify it under the terms of the GNU Lesser General Public // License as published by the Free Software Foundation; either // version 2.1 of the License, or (at your option) any later version. // // This library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU // Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public // License along with this library; if not, write to the Free Software // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA // #endregion using System; using System.Collections.Generic; using System.Text; namespace Duplicati.Library.Main { /// /// Internal class that ensures retry operations, tracks statistics and performs asyncronous operations /// internal class BackendWrapper : IDisposable { /// /// The actual backend that this class is wrapping /// private Duplicati.Library.Interface.IBackend m_backend; /// /// The statistics gathering object /// private CommunicationStatistics m_statistics; /// /// The set of options used by Duplicati /// private Options m_options; /// /// A list of pending async operations /// private Queue> m_pendingOperations; /// /// This event is set after an item has been removed from the queue /// private System.Threading.AutoResetEvent m_asyncItemProcessed; /// /// This event is set after an item has been placed into the queue /// private System.Threading.AutoResetEvent m_asyncItemReady; /// /// A flag used to signal the termination of the async thread /// private volatile bool m_asyncTerminate = false; /// /// The lock for items in the processing queue /// private object m_queuelock; /// /// A value describing if the backend wrapper is performing asynchronous operations /// private bool m_async; /// /// The thread performing asynchronous operations /// private System.Threading.Thread m_workerThread; /// /// The exception, if any, encountered by the worker thread /// private Exception m_workerException; /// /// The number of manifests uploaded asynchronously /// private int m_manifestUploads = 0; /// /// Temporary variable for progress reporting /// private string m_statusmessage; /// /// The cache filename strategy used /// /// A cache filename strategy object public static FilenameStrategy CreateCacheFilenameStrategy() { return new FilenameStrategy("dpl", "_", true); } /// /// The filename strategy used to generate and parse filenames /// private FilenameStrategy m_filenamestrategy; /// /// A local version of the cache filename strategy /// private FilenameStrategy m_cachefilenamestrategy = BackendWrapper.CreateCacheFilenameStrategy(); /// /// A list of items that should be removed /// private List m_orphans; /// /// The progress reporting event /// public event RSync.RSyncDir.ProgressEventDelegate ProgressEvent; /// /// An event that is raised after an async item has been uploaded, used to pause the uploads /// public event EventHandler AsyncItemProcessedEvent; /// /// If encryption is used, this is the instance that performs it /// private Duplicati.Library.Interface.IEncryption m_encryption = null; /// /// The number of bytes added to a file when encrypted /// private long m_encryptionSizeOverhead = 0; /// /// Gets the number of bytes added to a file when encrypted and transfered /// public long FileSizeOverhead { get { return m_encryptionSizeOverhead; } } /// /// Class to represent hash failures /// public class HashMismathcException : Exception { /// /// Default constructor, sets a generic string as the message /// public HashMismathcException() : base() { } /// /// Constructor with non-default message /// /// The exception message public HashMismathcException(string message) : base(message) { } /// /// Constructor with non-default message and inner exception details /// /// The exception message /// The exception that caused this exception public HashMismathcException(string message, Exception innerException) : base(message, innerException) { } } /// /// Constructs a new BackendWrapper /// /// The statistics logging module, may be null /// The url to the backend to wrap /// A set of backend options public BackendWrapper(CommunicationStatistics statistics, string backend, Options options) { m_statistics = statistics; m_options = options; m_filenamestrategy = new FilenameStrategy(m_options); m_backend = Duplicati.Library.DynamicLoader.BackendLoader.GetBackend(backend, m_options.RawOptions); if (m_backend == null) throw new Exception(string.Format(Strings.BackendWrapper.BackendNotFoundError, m_backend)); if (!m_options.NoEncryption) { if (string.IsNullOrEmpty(m_options.Passphrase)) throw new Exception(Strings.BackendWrapper.PassphraseMissingError); m_encryption = DynamicLoader.EncryptionLoader.GetModule(m_options.EncryptionModule, m_options.Passphrase, m_options.RawOptions); m_encryptionSizeOverhead = m_encryption.SizeOverhead(m_options.VolumeSize); } if (m_options.AutoCleanup) m_orphans = new List(); if (!string.IsNullOrEmpty(m_options.SignatureCachePath) && !System.IO.Directory.Exists(m_options.SignatureCachePath)) System.IO.Directory.CreateDirectory(m_options.SignatureCachePath); m_async = m_options.AsynchronousUpload; if (m_async) { //If we are using async operations, the entire class is actually threadsafe, //utilizing a common exclusive lock on all operations. But the implementation does //not prevent starvation, so it should not be called by multiple threads. m_pendingOperations = new Queue>(); m_asyncItemProcessed = new System.Threading.AutoResetEvent(false); m_asyncItemReady = new System.Threading.AutoResetEvent(false); m_workerThread = new System.Threading.Thread(ProcessQueue); m_workerThread.Name = "AsyncUploaderThread"; m_queuelock = new object(); m_workerThread.Start(); } } public void AddOrphan(BackupEntryBase entry) { if (m_orphans != null) m_orphans.Add(entry); else Logging.Log.WriteMessage(string.Format(Strings.BackendWrapper.PartialFileFoundMessage, entry.Filename), Duplicati.Library.Logging.LogMessageType.Warning); } public ManifestEntry GetBackupSet(string timelimit) { if (string.IsNullOrEmpty(timelimit)) timelimit = "now"; return GetBackupSet(Core.Timeparser.ParseTimeInterval(timelimit, DateTime.Now)); } /// /// Gets the manifest entry that represents the most recent full backup, with a list of incrementals in the chain. /// /// The oldest allowed time for a backup, set to DateTime.Now to get the most recent entry /// The manifest entry with incrementals that contains all changes up to the timelimit public ManifestEntry GetBackupSet(DateTime timelimit) { List backups = GetBackupSets(); if (backups.Count == 0) throw new Exception(Strings.BackendWrapper.NoBackupsFoundError); ManifestEntry bestFit = backups[0]; List additions = new List(); foreach (ManifestEntry be in backups) if (be.Time <= timelimit) { bestFit = be; additions.Clear(); foreach (ManifestEntry bex in be.Incrementals) if (bex.Time <= timelimit) additions.Add(bex); } if (bestFit.Volumes.Count == 0) throw new Exception(Strings.BackendWrapper.FilenameParseError); bestFit.Incrementals.Clear(); bestFit.Incrementals.AddRange(additions); return bestFit; } /// /// Parses the filelist into a list of full backups, each with a chain of incrementals attached /// /// The list of filenames found on the backend /// A list of full backups public List SortAndPairSets(List files) { Sorter sortHelper = new Sorter(); files.Sort(sortHelper); for(int i = 1; i < files.Count; i++) if (files[i].Name == files[i - 1].Name) { if (m_statistics != null) m_statistics.LogWarning(string.Format(Strings.BackendWrapper.DuplicateFileEntryWarning, files[i].Name), null); files.RemoveAt(i); i--; } List incrementals = new List(); List fulls = new List(); Dictionary> signatures = new Dictionary>(); Dictionary> contents = new Dictionary>(); //First we parse all files into their respective classes foreach (Duplicati.Library.Interface.IFileEntry fe in files) { BackupEntryBase be = m_filenamestrategy.ParseFilename(fe); if (be == null) { if (m_statistics != null && m_statistics.VerboseErrors && !fe.IsFolder) m_statistics.LogWarning(string.Format(Strings.BackendWrapper.UnmatchedFilenameWarning, fe.Name), null); continue; //Non-duplicati files } if (!string.IsNullOrEmpty(be.EncryptionMode) && Array.IndexOf(DynamicLoader.EncryptionLoader.Keys, be.EncryptionMode) < 0) continue; if (be is PayloadEntryBase && Array.IndexOf(DynamicLoader.CompressionLoader.Keys,((PayloadEntryBase)be).Compression) < 0) continue; if (be is ManifestEntry) { if (((ManifestEntry)be).IsFull) fulls.Add((ManifestEntry)be); else incrementals.Add((ManifestEntry)be); } else if (be is ContentEntry) { string key = be.TimeString; if (!contents.ContainsKey(key)) contents[key] = new List(); contents[key].Add((ContentEntry)be); } else if (be is SignatureEntry) { string key = be.TimeString; if (!signatures.ContainsKey(key)) signatures[key] = new List(); signatures[key].Add((SignatureEntry)be); } else throw new Exception(string.Format(Strings.BackendWrapper.InvalidEntryTypeError, be.GetType().FullName)); } fulls.Sort(sortHelper); incrementals.Sort(sortHelper); //Pair up the manifests in primary/alternate pairs foreach(List mfl in new List[] {fulls, incrementals}) for (int i = 0; i < mfl.Count - 1; i++) { if (mfl[i].TimeString == mfl[i + 1].TimeString && mfl[i].IsPrimary != mfl[i + 1].IsPrimary) { if (mfl[i].IsPrimary) { mfl[i].Alternate = mfl[i + 1]; mfl.RemoveAt(i + 1); } else { mfl[i + 1].Alternate = mfl[i]; mfl.RemoveAt(i); } } } //Attach volumes to the full backups for(int i = 0; i < fulls.Count; i++) { ManifestEntry be = fulls[i]; string key = be.TimeString; if (contents.ContainsKey(key) && signatures.ContainsKey(key)) { List signature = signatures[key]; List content = contents[key]; signature.Sort(sortHelper); content.Sort(sortHelper); int volCount = Math.Min(content.Count, signature.Count); for (int j = 0; j < volCount; j++) if (signature[0].Volumenumber == (j + 1) && content[0].Volumenumber == (j + 1)) { be.Volumes.Add(new KeyValuePair(signature[0], content[0])); signature.RemoveAt(0); content.RemoveAt(0); } } fulls[i] = SwapManifestAlternates(be); } //Attach volumes to the incrementals, and attach incrementals to the fulls int index = 0; foreach (ManifestEntry be in incrementals) { string key = be.TimeString; if (contents.ContainsKey(key) && signatures.ContainsKey(key)) { List signature = signatures[key]; List content = contents[key]; signature.Sort(sortHelper); content.Sort(sortHelper); int volCount = Math.Min(content.Count, signature.Count); for (int j = 0; j < volCount; j++) if (signature[0].Volumenumber == (j + 1) && content[0].Volumenumber == (j + 1)) { be.Volumes.Add(new KeyValuePair(signature[0], content[0])); signature.RemoveAt(0); content.RemoveAt(0); } } if (index >= fulls.Count || be.Time <= fulls[index].Time) { if (m_orphans == null) Logging.Log.WriteMessage(string.Format(Strings.BackendWrapper.OrphanIncrementalFoundMessage, be.Filename), Duplicati.Library.Logging.LogMessageType.Warning); else m_orphans.Add(be); continue; } else { while (index < fulls.Count - 1 && be.Time > fulls[index + 1].Time) index++; fulls[index].Incrementals.Add(SwapManifestAlternates(be)); } } foreach (List lb in contents.Values) foreach (ContentEntry be in lb) AddOrphan(be); foreach (List lb in signatures.Values) foreach (SignatureEntry be in lb) AddOrphan(be); if (m_statistics != null) { foreach (ManifestEntry me in fulls) { if (me.Volumes.Count == 0) m_statistics.LogWarning(string.Format(Strings.BackendWrapper.EmptyManifestWarning, me.Filename), null); foreach (ManifestEntry me2 in me.Incrementals) if (me.Volumes.Count == 0) m_statistics.LogWarning(string.Format(Strings.BackendWrapper.EmptyManifestWarning, me.Filename), null); } } return fulls; } private ManifestEntry SwapManifestAlternates(ManifestEntry m) { if (m.Volumes.Count % 2 != 1 && m.Alternate != null) { m.Alternate.Incrementals.Clear(); m.Alternate.Incrementals.AddRange(m.Incrementals); m.Alternate.Volumes.Clear(); m.Alternate.Volumes.AddRange(m.Volumes); m.Incrementals.Clear(); m.Volumes.Clear(); m.Alternate.Alternate = m; return m.Alternate; } return m; } public List GetBackupSets() { using (new Logging.Timer("Getting and sorting filelist from " + m_backend.DisplayName)) return SortAndPairSets((List)ProtectedInvoke("ListInternal")); } public void Put(BackupEntryBase remote, string filename) { if (!m_async) PutInternal(remote, filename); else { bool waitForCompletion; lock (m_queuelock) { if (m_workerException != null) throw m_workerException; m_pendingOperations.Enqueue(new KeyValuePair( remote, filename )); m_asyncItemReady.Set(); //The *3 is there because a volume consists of 3 files (signature, content and manifest) waitForCompletion = m_options.AsynchronousUploadLimit > 0 && m_pendingOperations.Count > (m_options.AsynchronousUploadLimit * 3); } while (waitForCompletion) { m_asyncItemProcessed.WaitOne(1000 * 5, false); lock (m_queuelock) { if (m_workerException != null) throw m_workerException; waitForCompletion = m_options.AsynchronousUploadLimit > 0 && m_pendingOperations.Count > (m_options.AsynchronousUploadLimit * 3); } } } } public void Get(BackupEntryBase remote, string filename, string filehash) { ProtectedInvoke("GetInternal", remote, filename, filehash); } public void Delete(BackupEntryBase remote) { ProtectedInvoke("DeleteInternal", remote); } /// /// This method invokes another method in such a way that no more than one thread is ever /// executing on the same backend at any time. /// /// The method to invoke /// The methods arguments /// The return value of the invoked function private object ProtectedInvoke(string methodname, params object[] arguments) { System.Reflection.MethodInfo method = this.GetType().GetMethod(methodname); if (method == null) method = this.GetType().GetMethod(methodname, System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); if (method == null) throw new Exception(string.Format(Strings.BackendWrapper.FunctionLookupError, methodname)); //If the code is not async, just invoke it if (!m_async) { try { return method.Invoke(this, arguments); } catch (System.Reflection.TargetInvocationException tex) { //Unwrap those annoying invocation exceptions if (tex.InnerException != null) throw tex.InnerException; else throw; } } else { while (true) { //Get the lock lock (m_queuelock) { //If the lock is free, just run, but keep the lock if (m_workerThread == null || !m_workerThread.IsAlive || m_pendingOperations.Count == 0) { try { return method.Invoke(this, arguments); } catch (System.Reflection.TargetInvocationException tex) { //Unwrap those annoying invocation exceptions if (tex.InnerException != null) throw tex.InnerException; else throw; } } } //Otherwise, wait for the worker to signal completion //We wait without holding the lock, because the worker cannot signal completion while //we hold the lock m_workerThread.Join(1000); //Now re-acquire the lock, and re-check the state of the event } } } private void DeleteSignatureCacheCopy(BackupEntryBase entry) { if (!string.IsNullOrEmpty(m_options.SignatureCachePath)) { string cacheFilename = m_cachefilenamestrategy.GenerateFilename(entry); if (System.IO.File.Exists(cacheFilename)) try { System.IO.File.Delete(cacheFilename); } catch { } } } public void DeleteOrphans() { if (m_orphans == null) return; foreach (BackupEntryBase be in m_orphans) { Logging.Log.WriteMessage(string.Format(Strings.BackendWrapper.RemovingLeftoverFileMessage, be.Filename), Duplicati.Library.Logging.LogMessageType.Information); if (m_options.Force) { if (m_statistics != null) m_statistics.LogWarning(string.Format(Strings.BackendWrapper.RemoveOrphanFileWarning, be.Filename), null); m_backend.Delete(be.Filename); DeleteSignatureCacheCopy(be); } if (be is ManifestEntry) foreach (KeyValuePair bex in ((ManifestEntry)be).Volumes) { Logging.Log.WriteMessage(string.Format(Strings.BackendWrapper.RemovingLeftoverFileMessage, bex.Key.Filename), Duplicati.Library.Logging.LogMessageType.Information); Logging.Log.WriteMessage(string.Format(Strings.BackendWrapper.RemovingLeftoverFileMessage, bex.Value.Filename), Duplicati.Library.Logging.LogMessageType.Information); if (m_options.Force) { if (m_statistics != null) { m_statistics.LogWarning(string.Format(Strings.BackendWrapper.RemoveOrphanFileWarning, bex.Key.Filename), null); m_statistics.LogWarning(string.Format(Strings.BackendWrapper.RemoveOrphanFileWarning, bex.Value.Filename), null); } m_backend.Delete(bex.Key.Filename); m_backend.Delete(bex.Value.Filename); DeleteSignatureCacheCopy(bex.Key); } } } if (!m_options.Force && m_orphans.Count > 0) Logging.Log.WriteMessage(Strings.BackendWrapper.FilesNotForceRemovedMessage, Duplicati.Library.Logging.LogMessageType.Information); } private List ListInternal() { int retries = m_options.NumberOfRetries; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; return m_backend.List(); } catch (System.Threading.ThreadAbortException) { throw; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message, ex); retries--; if (retries > 0 && m_options.RetryDelay.Ticks > 0) System.Threading.Thread.Sleep(m_options.RetryDelay); } } while (retries > 0); throw new Exception(string.Format(Strings.BackendWrapper.FileListingError, lastEx.Message), lastEx); } private void DeleteInternal(BackupEntryBase remote) { int retries = m_options.NumberOfRetries; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; m_backend.Delete(remote.Filename); lastEx = null; } catch (System.Threading.ThreadAbortException) { throw; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message, ex); retries--; if (retries > 0 && m_options.RetryDelay.Ticks > 0) System.Threading.Thread.Sleep(m_options.RetryDelay); } } while (lastEx != null && retries > 0); if (lastEx != null) throw new Exception(string.Format(Strings.BackendWrapper.FileDeleteError, lastEx.Message), lastEx); if (remote is SignatureEntry && !string.IsNullOrEmpty(m_options.SignatureCachePath)) { try { if (System.IO.File.Exists(System.IO.Path.Combine(m_options.SignatureCachePath, m_cachefilenamestrategy.GenerateFilename(remote)))) System.IO.File.Delete(System.IO.Path.Combine(m_options.SignatureCachePath, m_cachefilenamestrategy.GenerateFilename(remote))); } catch {} } } private void GetInternal(BackupEntryBase remote, string filename, string filehash) { int retries = m_options.NumberOfRetries; Exception lastEx = null; m_statusmessage = string.Format(Strings.BackendWrapper.StatusMessageDownloading, remote.Filename); do { try { if (!string.IsNullOrEmpty(m_options.SignatureCachePath) && remote is SignatureEntry) { string cachefilename = System.IO.Path.Combine(m_options.SignatureCachePath, m_cachefilenamestrategy.GenerateFilename(remote)); if (System.IO.File.Exists(cachefilename)) if (filehash != null && Core.Utility.CalculateHash(cachefilename) == filehash) { //TODO: Don't copy, but just return it as write protected System.IO.File.Copy(cachefilename, filename, true); //TODO: Warn on hash mismatch? return; } } Core.TempFile tempfile = null; try { if (!string.IsNullOrEmpty(remote.EncryptionMode)) tempfile = new Duplicati.Library.Core.TempFile(); else tempfile = new Duplicati.Library.Core.TempFile(filename); m_statistics.NumberOfRemoteCalls++; if (m_backend is Duplicati.Library.Interface.IStreamingBackend && !m_options.DisableStreamingTransfers) { using (System.IO.FileStream fs = System.IO.File.Open(tempfile, System.IO.FileMode.Create, System.IO.FileAccess.Write, System.IO.FileShare.None)) using (Core.ProgressReportingStream pgs = new Duplicati.Library.Core.ProgressReportingStream(fs, remote.Fileentry.Size)) using (Core.ThrottledStream ts = new Duplicati.Library.Core.ThrottledStream(pgs, m_options.MaxUploadPrSecond, m_options.MaxDownloadPrSecond)) { pgs.Progress += new Duplicati.Library.Core.ProgressReportingStream.ProgressDelegate(pgs_Progress); ts.Callback += new Duplicati.Library.Core.ThrottledStream.ThrottledStreamCallback(ThrottledStream_Callback); ((Duplicati.Library.Interface.IStreamingBackend)m_backend).Get(remote.Filename, ts); } } else { if (!m_async && ProgressEvent != null) ProgressEvent(50, m_statusmessage); m_backend.Get(remote.Filename, tempfile); if (!m_async && ProgressEvent != null) ProgressEvent(100, m_statusmessage); } if (!string.IsNullOrEmpty(remote.EncryptionMode)) { try { using (Library.Interface.IEncryption enc = DynamicLoader.EncryptionLoader.GetModule(remote.EncryptionMode, m_options.Passphrase, m_options.RawOptions)) enc.Decrypt(tempfile, filename); } catch (Exception ex) { //If we fail here, make sure that we throw a crypto exception if (ex is System.Security.Cryptography.CryptographicException) throw; else throw new System.Security.Cryptography.CryptographicException(ex.Message, ex); } tempfile.Dispose(); //Remove the encrypted file //Wrap the new file as a temp file tempfile = new Duplicati.Library.Core.TempFile(filename); } if (filehash != null && Core.Utility.CalculateHash(tempfile) != filehash) throw new HashMismathcException(string.Format(Strings.BackendWrapper.HashMismatchError, remote.Filename, filehash, Core.Utility.CalculateHash(tempfile))); if (!string.IsNullOrEmpty(m_options.SignatureCachePath) && remote is SignatureEntry) { string cachefilename = System.IO.Path.Combine(m_options.SignatureCachePath, m_cachefilenamestrategy.GenerateFilename(remote)); System.IO.File.Copy(tempfile, cachefilename); } lastEx = null; tempfile.Protected = true; //Don't delete it } finally { try { if (tempfile != null) tempfile.Dispose(); } catch { } } } catch (System.Threading.ThreadAbortException) { throw; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message, ex); retries--; if (retries > 0 && m_options.RetryDelay.Ticks > 0) System.Threading.Thread.Sleep(m_options.RetryDelay); } } while (lastEx != null && retries > 0); if (lastEx != null) if (lastEx is HashMismathcException) throw lastEx; else throw new Exception(string.Format(Strings.BackendWrapper.FileDownloadError, lastEx.Message), lastEx); m_statistics.NumberOfBytesDownloaded += new System.IO.FileInfo(filename).Length; } private void PutInternal(BackupEntryBase remote, string filename) { string remotename = GenerateFilename(remote); m_statusmessage = string.Format(Strings.BackendWrapper.StatusMessageUploading, remotename, Core.Utility.FormatSizeString(new System.IO.FileInfo(filename).Length)); string encryptedFile = filename; try { if (m_encryption != null) { using (Core.TempFile tf = new Duplicati.Library.Core.TempFile()) //If exception is thrown, tf will be deleted { m_encryption.Encrypt(filename, tf); tf.Protected = true; //Done, keep file encryptedFile = tf; } } int retries = m_options.NumberOfRetries; bool success = false; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; if (m_backend is Library.Interface.IStreamingBackend && !m_options.DisableStreamingTransfers) { #if DEBUG_THROTTLE DateTime begin = DateTime.Now; #endif using (System.IO.FileStream fs = System.IO.File.Open(encryptedFile, System.IO.FileMode.Open, System.IO.FileAccess.Read, System.IO.FileShare.Read)) using (Core.ProgressReportingStream pgs = new Duplicati.Library.Core.ProgressReportingStream(fs, fs.Length)) { pgs.Progress += new Duplicati.Library.Core.ProgressReportingStream.ProgressDelegate(pgs_Progress); using (Core.ThrottledStream ts = new Core.ThrottledStream(pgs, m_options.MaxUploadPrSecond, m_options.MaxDownloadPrSecond)) { ts.Callback += new Duplicati.Library.Core.ThrottledStream.ThrottledStreamCallback(ThrottledStream_Callback); ((Library.Interface.IStreamingBackend)m_backend).Put(remotename, ts); } } #if DEBUG_THROTTLE TimeSpan duration = DateTime.Now - begin; long size = new System.IO.FileInfo(encryptedFile).Length; Console.WriteLine("Transferred " + Core.Utility.FormatSizeString(size) + " in " + duration.TotalSeconds.ToString() + ", yielding : " + ((size / (double)1024.0) / duration.TotalSeconds) + " kb/s"); #endif } else { if (ProgressEvent != null) ProgressEvent(50, m_statusmessage); m_backend.Put(remotename, encryptedFile); if (ProgressEvent != null) ProgressEvent(50, m_statusmessage); } if (remote is SignatureEntry && !string.IsNullOrEmpty(m_options.SignatureCachePath)) System.IO.File.Copy(filename, System.IO.Path.Combine(m_options.SignatureCachePath, m_cachefilenamestrategy.GenerateFilename(remote)), true); if (remote is ManifestEntry) { if (m_queuelock != null) { lock (m_queuelock) m_manifestUploads++; } else m_manifestUploads++; } success = true; lastEx = null; } catch (System.Threading.ThreadAbortException) { throw; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message, ex); retries--; if (retries > 0 && m_options.RetryDelay.Ticks > 0) System.Threading.Thread.Sleep(m_options.RetryDelay); } } while (!success && retries > 0); if (!success) throw new Exception(string.Format(Strings.BackendWrapper.FileUploadError, lastEx == null ? "" : lastEx.Message), lastEx); m_statistics.NumberOfBytesUploaded += new System.IO.FileInfo(filename).Length; } finally { //We delete the file here, because the backend leaves the file, //in the case where we use async methods try { if (System.IO.File.Exists(filename)) System.IO.File.Delete(filename); } catch { } try { if (System.IO.File.Exists(encryptedFile)) System.IO.File.Delete(encryptedFile); } catch { } try { if (ProgressEvent != null) ProgressEvent(-1, ""); } catch { } } } /// /// A callback from the throttled stream, used to change speed based on user adjustments /// /// The stream that raised the event void ThrottledStream_Callback(Core.ThrottledStream sender) { sender.ReadSpeed = m_options.MaxUploadPrSecond; sender.WriteSpeed = m_options.MaxDownloadPrSecond; } /// /// Internal helper to consistenly name remote files beyond what the filenamestrategy supports /// /// The entry to create a filename for /// A filename with extensions private string GenerateFilename(BackupEntryBase remote) { string remotename = m_filenamestrategy.GenerateFilename(remote); if (remote is ManifestEntry) remotename += ".manifest"; else remotename += "." + m_options.CompressionModule; if (m_encryption != null) remotename += "." + m_encryption.FilenameExtension; return remotename; } private void pgs_Progress(int progress) { if (ProgressEvent != null) ProgressEvent(progress, m_statusmessage); } /// /// Worker Thread entry that empties the request queue /// /// Unused required parameter private void ProcessQueue(object dummy) { try { lock (m_queuelock) m_workerException = null; while (!m_asyncTerminate) { KeyValuePair args = new KeyValuePair(null, null); while (args.Key == null) { if (m_asyncTerminate) return; //Obtain the lock for the queue lock (m_queuelock) if (m_pendingOperations.Count > 0) args = m_pendingOperations.Peek(); if (args.Key == null) m_asyncItemReady.WaitOne(1000, false); } //Pause if requested if (AsyncItemProcessedEvent != null) AsyncItemProcessedEvent(this, null); PutInternal(args.Key, args.Value); lock (m_queuelock) { m_pendingOperations.Dequeue(); m_asyncItemProcessed.Set(); } } } catch (Exception ex) { lock (m_queuelock) { m_workerException = ex; m_asyncItemProcessed.Set(); } } } /// /// Gets the number of uploads performed asynchronously /// public int ManifestUploads { get { if (m_queuelock != null) { lock (m_queuelock) return m_manifestUploads; } else return m_manifestUploads; } } /// /// This function attemtps to forcefully abort all ongoing async operations /// public void AbortAll() { //If we are not running async, just return if (!m_async) return; List> pending = new List>(); m_asyncTerminate = true; if (m_workerThread != null && m_workerThread.IsAlive) m_workerThread.Join(1000); lock (m_queuelock) { if (m_workerThread != null && m_workerThread.IsAlive) m_workerThread.Abort(); //Extract all unprocessed items while (m_pendingOperations.Count > 0) pending.Add(m_pendingOperations.Dequeue()); } //Clean up all temporary files from pending operations foreach (KeyValuePair p in pending) try { if (System.IO.File.Exists(p.Value)) System.IO.File.Delete(p.Value); } catch { } } /// /// This function suspends the calling thread until the /// ongoing transfer has completed, then disables asynchronous /// transfers and returns the non-transferred items /// public List> ExtractPendingUploads() { List> work = null; while (true) { lock (m_queuelock) { if (m_workerException != null) throw m_workerException; //On the first run, we empty the queue and signal the stop if (work == null) { m_asyncTerminate = true; work = new List>(); if (m_pendingOperations.Count > 1) { while (m_pendingOperations.Count != 0) work.Add(m_pendingOperations.Dequeue()); //The top entry is being completed by the thread m_pendingOperations.Enqueue(work[0]); work.RemoveAt(0); } } //When the thread completes, disable asynchronous transfers and return the unfinished work if (m_workerThread == null || !m_workerThread.IsAlive) { m_async = false; return work; } } m_workerThread.Join(1000 * 5); } } private void DisposeInternal() { if (m_backend != null) { m_backend.Dispose(); m_backend = null; } } #region IDisposable Members public void Dispose() { if (m_async && m_queuelock != null) AbortAll(); if (m_backend != null) ProtectedInvoke("DisposeInternal"); if (m_encryption != null) m_encryption.Dispose(); } #endregion } }