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 : Backend.IBackend { private Backend.IBackend m_backend; private CommunicationStatistics m_statistics; private Dictionary m_options; private int m_retries; private TimeSpan m_retrydelay; private bool m_asyncOperation; //These keep track of async operations private Queue m_pendingOperations; //TODO: Figure out if the linux implementation uses the pthreads model, where signals //are lost if there are no waiters at the signaling time private System.Threading.ManualResetEvent m_asyncWait; private object m_queuelock; public BackendWrapper(CommunicationStatistics statistics, Backend.IBackend backend, Dictionary options) { m_statistics = statistics; m_backend = backend; m_retries = 5; m_options = options; if (options.ContainsKey("number-of-retries")) int.TryParse(options["number-of-retries"], out m_retries); m_retrydelay = new TimeSpan(TimeSpan.TicksPerSecond * 10); if (options.ContainsKey("retry-delay")) try { Core.Timeparser.ParseTimeSpan(options["retry-delay"]); } catch { } m_asyncOperation = options.ContainsKey("asynchronous-upload"); if (m_asyncOperation) { //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_asyncWait = new System.Threading.ManualResetEvent(true); m_queuelock = new object(); } } #region IBackendInterface Members public string DisplayName { get { return m_backend.DisplayName; } } public string ProtocolKey { get { return m_backend.ProtocolKey; } } public bool SupportsStreaming { get { return m_backend is Backend.IStreamingBackend ? ((Backend.IStreamingBackend)m_backend).SupportsStreaming : false; } } public List List() { return (List)ProtectedInvoke("ListInternal"); } public void Put(string remotename, string filename) { if (!m_asyncOperation) PutInternal(remotename, filename); else { lock (m_queuelock) { m_pendingOperations.Enqueue(new string[] { remotename, filename }); if (m_asyncWait.WaitOne(0)) { m_asyncWait.Reset(); System.Threading.ThreadPool.QueueUserWorkItem(new System.Threading.WaitCallback(ProcessQueue)); } } } } public void Get(string remotename, string filename) { ProtectedInvoke("GetInternal", remotename, filename); } public void Delete(string remotename) { ProtectedInvoke("DeleteInternal", remotename); } #endregion /// /// 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("Failed to find function: " + methodname); //If the code is not async, just invoke it if (!m_asyncOperation) { 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_asyncWait.WaitOne(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_asyncWait.WaitOne(); //Now re-acquire the lock, and re-check the state of the event } } } private List ListInternal() { int retries = m_retries; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; return m_backend.List(); } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message); retries--; if (retries > 0 && m_retrydelay.Ticks > 0) System.Threading.Thread.Sleep(m_retrydelay); } } while (retries > 0); throw new Exception("Failed to retrieve file listing: " + lastEx.Message, lastEx); } private void DeleteInternal(string remotename) { int retries = m_retries; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; m_backend.Delete(remotename); lastEx = null; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message); retries--; if (retries > 0 && m_retrydelay.Ticks > 0) System.Threading.Thread.Sleep(m_retrydelay); } } while (lastEx != null && retries > 0); if (lastEx != null) throw new Exception("Failed to delete file: " + lastEx.Message, lastEx); } private void GetInternal(string remotename, string filename) { int retries = m_retries; Exception lastEx = null; do { try { m_statistics.NumberOfRemoteCalls++; if (!this.SupportsStreaming) m_backend.Get(remotename, filename); else using (System.IO.FileStream fs = System.IO.File.Open(filename, System.IO.FileMode.Create, System.IO.FileAccess.Write, System.IO.FileShare.None)) ((Backend.IStreamingBackend)m_backend).Get(remotename, ThrottleStream(fs, m_options)); lastEx = null; } catch (Exception ex) { lastEx = ex; m_statistics.LogError(ex.Message); retries--; if (retries > 0 && m_retrydelay.Ticks > 0) System.Threading.Thread.Sleep(m_retrydelay); } } while (lastEx != null && retries > 0); if (lastEx != null) throw new Exception("Failed to download file: " + lastEx.Message, lastEx); m_statistics.NumberOfBytesDownloaded += new System.IO.FileInfo(filename).Length; } private void PutInternal(string remotename, string filename) { try { int retries = m_retries; bool success = false; do { try { m_statistics.NumberOfRemoteCalls++; if (!this.SupportsStreaming) m_backend.Put(remotename, filename); else { #if DEBUG DateTime begin = DateTime.Now; long l; using (System.IO.FileStream fs = System.IO.File.Open(filename, System.IO.FileMode.Open, System.IO.FileAccess.Read, System.IO.FileShare.Read)) { l = fs.Length; ((Backend.IStreamingBackend)m_backend).Put(remotename, ThrottleStream(fs, m_options)); } TimeSpan duration = DateTime.Now - begin; Console.WriteLine("Transferred " + l.ToString() + " bytes in " + duration.TotalSeconds.ToString() + ", yielding : " + (l / (double)1024 / duration.TotalSeconds) + " kb/s"); #else using (System.IO.FileStream fs = System.IO.File.Open(filename, System.IO.FileMode.Open, System.IO.FileAccess.Read, System.IO.FileShare.Read)) ((Backend.IStreamingBackend)m_backend).Put(remotename, ThrottleStream(fs, m_options)); #endif } success = true; } catch (Exception ex) { m_statistics.LogError(ex.Message); retries--; if (retries > 0 && m_retrydelay.Ticks > 0) System.Threading.Thread.Sleep(m_retrydelay); } } while (!success && retries > 0); if (!success) throw new Exception("Failed to upload file"); 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 { } } } /// /// Worker Thread entry that empties the request queue /// /// Unused required parameter private void ProcessQueue(object dummy) { while (true) { string[] args; //Obtain the lock for the queue lock (m_queuelock) { if (m_pendingOperations.Count == 0) { //Signal that we are done m_asyncWait.Set(); return; } else args = m_pendingOperations.Dequeue(); } PutInternal(args[0], args[1]); } } private void DisposeInternal() { if (m_backend != null) { m_backend.Dispose(); m_backend = null; } } private Core.ThrottledStream ThrottleStream(System.IO.Stream basestream, Dictionary options) { return new Core.ThrottledStream(basestream, options.ContainsKey("max-upload-pr-second") ? Core.Sizeparser.ParseSize(options["max-upload-pr-second"], "kb") : 0, options.ContainsKey("max-download-pr-second") ? Core.Sizeparser.ParseSize(options["max-download-pr-second"], "kb") : 0); } private Core.ThrottledStream ThrottleStream(System.IO.Stream basestream, string upspeed, string downspeed) { return new Core.ThrottledStream(basestream, upspeed == null ? 0 : Core.Sizeparser.ParseSize(upspeed, "kb"), downspeed == null ? 0 : Core.Sizeparser.ParseSize(downspeed, "kb")); } #region IDisposable Members public void Dispose() { if (m_backend != null) ProtectedInvoke("DisposeInternal"); } #endregion } }