X-Git-Url: https://code.grnet.gr/git/pithos-ms-client/blobdiff_plain/77f3c89760ee171030f6bc64b3d318b33458d0eb..1cc1e8c52a2712aacfb99eb535180126c9f85350:/trunk/Pithos.Core/Agents/NetworkAgent.cs?ds=sidebyside diff --git a/trunk/Pithos.Core/Agents/NetworkAgent.cs b/trunk/Pithos.Core/Agents/NetworkAgent.cs index e609dd6..f511614 100644 --- a/trunk/Pithos.Core/Agents/NetworkAgent.cs +++ b/trunk/Pithos.Core/Agents/NetworkAgent.cs @@ -1,15 +1,57 @@ -using System; +#region +/* ----------------------------------------------------------------------- + * + * + * Copyright 2011-2012 GRNET S.A. All rights reserved. + * + * Redistribution and use in source and binary forms, with or + * without modification, are permitted provided that the following + * conditions are met: + * + * 1. Redistributions of source code must retain the above + * copyright notice, this list of conditions and the following + * disclaimer. + * + * 2. Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following + * disclaimer in the documentation and/or other materials + * provided with the distribution. + * + * + * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS + * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED + * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR + * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR + * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF + * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED + * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN + * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + * + * The views and conclusions contained in the software and + * documentation are those of the authors and should not be + * interpreted as representing official policies, either expressed + * or implied, of GRNET S.A. + * + * ----------------------------------------------------------------------- + */ +#endregion + +using System; using System.Collections.Generic; -using System.ComponentModel; using System.ComponentModel.Composition; using System.Diagnostics; using System.Diagnostics.Contracts; using System.IO; using System.Linq; using System.Net; -using System.Text; +using System.Reflection; using System.Threading; using System.Threading.Tasks; +using Castle.ActiveRecord; using Pithos.Interfaces; using Pithos.Network; using log4net; @@ -19,51 +61,121 @@ namespace Pithos.Core.Agents [Export] public class NetworkAgent { + private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); + private Agent _agent; - [Import] + [System.ComponentModel.Composition.Import] + public DeleteAgent DeleteAgent { get; set; } + + [System.ComponentModel.Composition.Import] public IStatusKeeper StatusKeeper { get; set; } - - public IStatusNotification StatusNotification { get; set; } -/* - [Import] - public FileAgent FileAgent {get;set;} -*/ - /* public int BlockSize { get; set; } - public string BlockHash { get; set; }*/ + private IStatusNotification _statusNotification; + public IStatusNotification StatusNotification + { + get { return _statusNotification; } + set + { + _statusNotification = value; + DeleteAgent.StatusNotification = value; + Uploader.StatusNotification = value; + Downloader.StatusNotification = value; + } + } - private static readonly ILog Log = LogManager.GetLogger("NetworkAgent"); - private List _accounts=new List(); + [System.ComponentModel.Composition.Import] + public IPithosSettings Settings { get; set; } - public void Start(/*int blockSize, string blockHash*/) + private Uploader _uploader; + + [System.ComponentModel.Composition.Import] + public Uploader Uploader { -/* - if (blockSize<0) - throw new ArgumentOutOfRangeException("blockSize"); - if (String.IsNullOrWhiteSpace(blockHash)) - throw new ArgumentOutOfRangeException("blockHash"); - Contract.EndContractBlock(); -*/ + get { return _uploader; } + set + { + _uploader = value; + _uploader.UnpauseEvent = _unPauseEvent; + } + } -/* - BlockSize = blockSize; - BlockHash = blockHash; -*/ + private Downloader _downloader; + [System.ComponentModel.Composition.Import] + public Downloader Downloader + { + get { return _downloader; } + set + { + _downloader = value; + _downloader.UnpauseEvent = _unPauseEvent; + } + } + + [System.ComponentModel.Composition.Import] + public Selectives Selectives { get; set; } + + //The Proceed signals the poll agent that it can proceed with polling. + //Essentially it stops the poll agent to give priority to the network agent + //Initially the event is signalled because we don't need to pause + private readonly AsyncManualResetEvent _proceedEvent = new AsyncManualResetEvent(true); + private Agents.Selectives _selectives; + private bool _pause; + + public AsyncManualResetEvent ProceedEvent + { + get { return _proceedEvent; } + } + + private readonly AsyncManualResetEvent _unPauseEvent = new AsyncManualResetEvent(true); + + private CancellationTokenSource _currentOperationCancellation=new CancellationTokenSource(); + + public void CancelCurrentOperation() + { + //What does it mean to cancel the current upload/download? + //Obviously, the current operation will be cancelled by throwing + //a cancellation exception. + // + //The default behavior is to retry any operations that throw. + //Obviously this is not what we want in this situation. + //The cancelled operation should NOT bea retried. + // + //This can be done by catching the cancellation exception + //and avoiding the retry. + // + + //Have to reset the cancellation source - it is not possible to reset the source + //Have to prevent a case where an operation requests a token from the old source + var oldSource = Interlocked.Exchange(ref _currentOperationCancellation, new CancellationTokenSource()); + oldSource.Cancel(); + + } + + public void Start() + { + if (_agent != null) + return; + + if (Log.IsDebugEnabled) + Log.Debug("Starting Network Agent"); _agent = Agent.Start(inbox => { Action loop = null; loop = () => { + DeleteAgent.ProceedEvent.Wait(); + _unPauseEvent.Wait(); var message = inbox.Receive(); var process=message.Then(Process,inbox.CancellationToken); inbox.LoopAsync(process, loop); }; loop(); }); + } private async Task Process(CloudAction action) @@ -74,58 +186,98 @@ namespace Pithos.Core.Agents throw new ArgumentException("The action.AccountInfo is empty","action"); Contract.EndContractBlock(); - var accountInfo = action.AccountInfo; - using (log4net.ThreadContext.Stacks["NETWORK"].Push("PROCESS")) + + + using (ThreadContext.Stacks["Operation"].Push(action.ToString())) { - Log.InfoFormat("[ACTION] Start Processing {0}", action); - var localFile = action.LocalFile; var cloudFile = action.CloudFile; var downloadPath = action.GetDownloadPath(); try { + StatusNotification.SetPithosStatus(PithosStatus.LocalSyncing,"Processing"); + _proceedEvent.Reset(); + + var accountInfo = action.AccountInfo; - switch (action.Action) + if (action.Action == CloudActionType.DeleteCloud) + { + //Redirect deletes to the delete agent + DeleteAgent.Post((CloudDeleteAction)action); + } + if (DeleteAgent.IsDeletedFile(action)) { - case CloudActionType.UploadUnconditional: - await UploadCloudFile(action); - break; - case CloudActionType.DownloadUnconditional: - - await DownloadCloudFile(accountInfo, cloudFile,downloadPath); - break; - case CloudActionType.DeleteCloud: - DeleteCloudFile(accountInfo, cloudFile, cloudFile.Name); - break; - case CloudActionType.RenameCloud: - var moveAction = (CloudMoveAction)action; - RenameCloudFile(accountInfo, moveAction); - break; - case CloudActionType.MustSynch: - - if (!File.Exists(downloadPath)) - { - await DownloadCloudFile(accountInfo, cloudFile, downloadPath); - } - else - { - await SyncFiles(accountInfo, action); - } - break; + //Clear the status of already deleted files to avoid reprocessing + if (action.LocalFile != null) + StatusKeeper.ClearFileStatus(action.LocalFile.FullName); } - Log.InfoFormat("[ACTION] End Processing {0}:{1}->{2}", action.Action, action.LocalFile, + else + { + switch (action.Action) + { + case CloudActionType.UploadUnconditional: + //Abort if the file was deleted before we reached this point + var uploadAction = (CloudUploadAction) action; + ProcessChildUploads(uploadAction); + await Uploader.UploadCloudFile(uploadAction ,CurrentOperationCancelToken); + break; + case CloudActionType.DownloadUnconditional: + await Downloader.DownloadCloudFile(accountInfo, cloudFile, downloadPath, CurrentOperationCancelToken); + break; + case CloudActionType.RenameCloud: + var moveAction = (CloudMoveAction)action; + RenameCloudFile(accountInfo, moveAction); + break; + case CloudActionType.RenameLocal: + RenameLocalFile(accountInfo, action); + break; + case CloudActionType.MustSynch: + if (!File.Exists(downloadPath) && !Directory.Exists(downloadPath)) + { + await Downloader.DownloadCloudFile(accountInfo, cloudFile, downloadPath, CurrentOperationCancelToken); + } + else + { + await SyncFiles(accountInfo, action); + } + break; + } + } + Log.InfoFormat("End Processing {0}:{1}->{2}", action.Action, action.LocalFile, action.CloudFile.Name); } - catch (OperationCanceledException) +/* + catch (WebException exc) + { + Log.ErrorFormat("[WEB ERROR] {0} : {1} -> {2} due to exception\r\n{3}", action.Action, action.LocalFile, action.CloudFile, exc); + + + //Actions that resulted in server errors should be retried + var response = exc.Response as HttpWebResponse; + if (response != null && response.StatusCode >= HttpStatusCode.InternalServerError) + { + _agent.Post(action); + Log.WarnFormat("[REQUEUE] {0} : {1} -> {2}", action.Action, action.LocalFile, action.CloudFile); + } + } +*/ + catch (OperationCanceledException ex) + { + Log.WarnFormat("Cancelling [{0}]",ex); + } + catch (DirectoryNotFoundException) { - throw; + Log.ErrorFormat("{0} : {1} -> {2} failed because the directory was not found.\n Rescheduling a delete", + action.Action, action.LocalFile, action.CloudFile); + //Post a delete action for the missing file + Post(new CloudDeleteAction(action)); } - catch (FileNotFoundException exc) + catch (FileNotFoundException) { Log.ErrorFormat("{0} : {1} -> {2} failed because the file was not found.\n Rescheduling a delete", - action.Action, action.LocalFile, action.CloudFile, exc); + action.Action, action.LocalFile, action.CloudFile); //Post a delete action for the missing file Post(new CloudDeleteAction(action)); } @@ -135,782 +287,303 @@ namespace Pithos.Core.Agents action.Action, action.LocalFile, action.CloudFile, exc); _agent.Post(action); - } - } - } - - private async Task SyncFiles(AccountInfo accountInfo,CloudAction action) - { - if (accountInfo == null) - throw new ArgumentNullException("accountInfo"); - if (action==null) - throw new ArgumentNullException("action"); - if (action.LocalFile==null) - throw new ArgumentException("The action's local file is not specified","action"); - if (!Path.IsPathRooted(action.LocalFile.FullName)) - throw new ArgumentException("The action's local file path must be absolute","action"); - if (action.CloudFile== null) - throw new ArgumentException("The action's cloud file is not specified", "action"); - Contract.EndContractBlock(); - - var localFile = action.LocalFile; - var cloudFile = action.CloudFile; - var downloadPath=action.LocalFile.FullName.ToLower(); - - var cloudHash = cloudFile.Hash.ToLower(); - var localHash = action.LocalHash.Value.ToLower(); - var topHash = action.TopHash.Value.ToLower(); - - //Not enough to compare only the local hashes, also have to compare the tophashes - - //If any of the hashes match, we are done - if ((cloudHash == localHash || cloudHash == topHash)) - { - Log.InfoFormat("Skipping {0}, hashes match",downloadPath); - return; - } - - //The hashes DON'T match. We need to sync - var lastLocalTime = localFile.LastWriteTime; - var lastUpTime = cloudFile.Last_Modified; - - //If the local file is newer upload it - if (lastUpTime <= lastLocalTime) - { - //It probably means it was changed while the app was down - UploadCloudFile(action); - } - else - { - //It the cloud file has a later date, it was modified by another user or computer. - //We need to check the local file's status - var status = StatusKeeper.GetFileStatus(downloadPath); - switch (status) + } + finally { - case FileStatus.Unchanged: - //If the local file's status is Unchanged, we can go on and download the newer cloud file - await DownloadCloudFile(accountInfo,cloudFile,downloadPath); - break; - case FileStatus.Modified: - //If the local file is Modified, we may have a conflict. In this case we should mark the file as Conflict - //We can't ensure that a file modified online since the last time will appear as Modified, unless we - //index all files before we start listening. - case FileStatus.Created: - //If the local file is Created, it means that the local and cloud files aren't related, - // yet they have the same name. - - //In both cases we must mark the file as in conflict - ReportConflict(downloadPath); - break; - default: - //Other cases should never occur. Mark them as Conflict as well but log a warning - ReportConflict(downloadPath); - Log.WarnFormat("Unexcepted status {0} for file {1}->{2}", status, - downloadPath, action.CloudFile.Name); - break; + if (_agent.IsEmpty) + _proceedEvent.Set(); + UpdateStatus(PithosStatus.LocalComplete); } } } - private void ReportConflict(string downloadPath) + private void ProcessChildUploads(CloudUploadAction uploadAction) { - if (String.IsNullOrWhiteSpace(downloadPath)) - throw new ArgumentNullException("downloadPath"); - Contract.EndContractBlock(); + if (!uploadAction.IsCreation || !(uploadAction.LocalFile is DirectoryInfo)) + return; - StatusKeeper.SetFileOverlayStatus(downloadPath, FileOverlayStatus.Conflict); - var message = String.Format("Conflict detected for file {0}", downloadPath); - Log.Warn(message); - StatusNotification.NotifyChange(message, TraceLevel.Warning); + var dirInfo = uploadAction.LocalFile as DirectoryInfo; + + var account = uploadAction.AccountInfo; + var folderActions = from info in dirInfo.EnumerateDirectories("*", SearchOption.AllDirectories) + select + new CloudUploadAction(account, info, null, account.BlockSize, account.BlockHash, + uploadAction, true); + var fileActions = from info in dirInfo.EnumerateFiles("*", SearchOption.AllDirectories) + select + new CloudUploadAction(account, info, null, account.BlockSize, account.BlockHash, + uploadAction, true); + //Post folder actions first, to ensure the selective folders are updated + folderActions.ApplyAction(PostUploadAction); + fileActions.ApplyAction(PostUploadAction); } - public void Post(CloudAction cloudAction) + private void PostUploadAction(CloudUploadAction action) { - if (cloudAction == null) - throw new ArgumentNullException("cloudAction"); - if (cloudAction.AccountInfo==null) - throw new ArgumentException("The CloudAction.AccountInfo is empty","cloudAction"); - Contract.EndContractBlock(); - - //If the action targets a local file, add a treehash calculation - if (cloudAction.LocalFile != null) - { - var accountInfo = cloudAction.AccountInfo; - if (cloudAction.LocalFile.Length>accountInfo.BlockSize) - cloudAction.TopHash = new Lazy(() => Signature.CalculateTreeHashAsync(cloudAction.LocalFile, - accountInfo.BlockSize, accountInfo.BlockHash).Result - .TopHash.ToHashString()); - else - { - cloudAction.TopHash=new Lazy(()=> cloudAction.LocalHash.Value); - } - - } - _agent.Post(cloudAction); + var state = StatusKeeper.GetStateByFilePath(action.LocalFile.FullName); + if (state != null) + state.Delete(); + //StatusKeeper.SetFileState(action.LocalFile.FullName,FileStatus.Created,FileOverlayStatus.Normal,String.Empty); + state = FileState.CreateFor(action.LocalFile); + //StatusKeeper.SetFileStatus(); + state.FileStatus = FileStatus.Created; + state.OverlayStatus = FileOverlayStatus.Normal; + state.Create(); + action.FileState = state; + Post(action); } - class ObjectInfoByNameComparer:IEqualityComparer + private CancellationToken CurrentOperationCancelToken { - public bool Equals(ObjectInfo x, ObjectInfo y) - { - return x.Name.Equals(y.Name,StringComparison.InvariantCultureIgnoreCase); - } - - public int GetHashCode(ObjectInfo obj) - { - return obj.Name.ToLower().GetHashCode(); - } + get { return _currentOperationCancellation.Token; } } - - //Remote files are polled periodically. Any changes are processed - public Task ProcessRemoteFiles(DateTime? since=null) + private void UpdateStatus(PithosStatus status) { - return Task.Factory.StartNewDelayed(10000, () => - { - using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push("All accounts")) - { - //Next time we will check for all changes since the current check minus 1 second - //This is done to ensure there are no discrepancies due to clock differences - DateTime nextSince = DateTime.Now.AddSeconds(-1); - - var tasks=from accountInfo in _accounts - select ProcessAccountFiles(accountInfo, since); - var process=Task.Factory.Iterate(tasks); - - return process.ContinueWith(t => - { - if (t.IsFaulted) - { - Log.Error("Error while processing accounts"); - t.Exception.Handle(exc=> - { - Log.Error("Details:", exc); - return true; - }); - } - ProcessRemoteFiles(nextSince); - }); - } - }); + StatusNotification.SetPithosStatus(status); + //StatusNotification.Notify(new Notification()); } - public Task ProcessAccountFiles(AccountInfo accountInfo,DateTime? since=null) - { - if (accountInfo==null) + private void RenameLocalFile(AccountInfo accountInfo, CloudAction action) + { + if (accountInfo == null) throw new ArgumentNullException("accountInfo"); - if (String.IsNullOrWhiteSpace(accountInfo.AccountPath)) - throw new ArgumentException("The AccountInfo.AccountPath is empty","accountInfo"); + if (action == null) + throw new ArgumentNullException("action"); + if (action.LocalFile == null) + throw new ArgumentException("The action's local file is not specified", "action"); + if (!Path.IsPathRooted(action.LocalFile.FullName)) + throw new ArgumentException("The action's local file path must be absolute", "action"); + if (action.CloudFile == null) + throw new ArgumentException("The action's cloud file is not specified", "action"); Contract.EndContractBlock(); - - using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push(accountInfo.UserName)) + using (ThreadContext.Stacks["Operation"].Push("RenameLocalFile")) { - Log.Info("Scheduled"); - var client=new CloudFilesClient(accountInfo); - - var containers = client.ListContainers(accountInfo.UserName); - - CreateContainerFolders(accountInfo, containers); - - //Get the list of server objects changed since the last check - var listObjects = from container in containers - select Task>.Factory.StartNew(_ => - client.ListObjects(accountInfo.UserName, container.Name, since),container.Name); + //We assume that the local file already exists, otherwise the poll agent + //would have issued a download request - var listAll = Task.Factory.WhenAll(listObjects.ToArray()); - - - - //Get the list of deleted objects since the last check -/* - var listTrash = Task>.Factory.StartNew(() => - client.ListObjects(accountInfo.UserName, FolderConstants.TrashContainer, since)); + var currentInfo = action.CloudFile; + var previousInfo = action.CloudFile.Previous; + var fileAgent = FileAgent.GetFileAgent(accountInfo); - var listShared = Task>.Factory.StartNew(() => - client.ListSharedObjects(since)); + var previousRelativepath = previousInfo.RelativeUrlToFilePath(accountInfo.UserName); + var previousFile = fileAgent.GetFileSystemInfo(previousRelativepath); - var listAll = Task.Factory.TrackedSequence( - () => listObjects, - () => listTrash, - () => listShared); -*/ - - - - var enqueueFiles = listAll.ContinueWith(task => - { - if (task.IsFaulted) - { - //ListObjects failed at this point, need to reschedule - Log.ErrorFormat("[FAIL] ListObjects for{0} in ProcessRemoteFiles with {0}", accountInfo.UserName,task.Exception); - return; - } - using (log4net.ThreadContext.Stacks["SCHEDULE"].Push("Process Results")) - { - var dict=task.Result.ToDictionary(t=> t.AsyncState); - - //Get all non-trash objects. Remember, the container name is stored in AsyncState - var remoteObjects = from objectList in task.Result - where (string)objectList.AsyncState != "trash" - from obj in objectList.Result - select obj; - - var trashObjects = dict["trash"].Result; - //var sharedObjects = ((Task>) task.Result[2]).Result; - - //Items with the same name, hash may be both in the container and the trash - //Don't delete items that exist in the container - var realTrash = from trash in trashObjects - where !remoteObjects.Any(info => info.Hash == trash.Hash) - select trash; - ProcessDeletedFiles(accountInfo,realTrash); - - - var remote = from info in remoteObjects//.Union(sharedObjects) - let name = info.Name - where !name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase) && - !name.StartsWith(FolderConstants.CacheFolder +"/", StringComparison.InvariantCultureIgnoreCase) - select info; - - //Create a list of actions from the remote files - var allActions = ObjectsToActions(accountInfo,remote); - - //And remove those that are already being processed by the agent - var distinctActions = allActions - .Except(_agent.GetEnumerable(), new PithosMonitor.LocalFileComparer()) - .ToList(); - - //Queue all the actions - foreach (var message in distinctActions) - { - Post(message); - } - - //Report the number of new files - var remoteCount = distinctActions.Count(action=> - action.Action==CloudActionType.DownloadUnconditional); -/* - if ( remoteCount > 0) - StatusNotification.NotifyChange(String.Format("Processing {0} new files", remoteCount)); -*/ - - Log.Info("[LISTENER] End Processing"); - } - }); - - var log = enqueueFiles.ContinueWith(t => - { - if (t.IsFaulted) - { - Log.Error("[LISTENER] Exception", t.Exception); - } - else - { - Log.Info("[LISTENER] Finished"); - } - }); - return log; - } - } - - private static void CreateContainerFolders(AccountInfo accountInfo, IList containers) - { - var containerPaths = from container in containers - let containerPath = Path.Combine(accountInfo.AccountPath, container.Name) - where container.Name != FolderConstants.TrashContainer && !Directory.Exists(containerPath) - select containerPath; - - foreach (var path in containerPaths) - { - Directory.CreateDirectory(path); + //In every case we need to move the local file first + MoveLocalFile(accountInfo, previousFile, fileAgent, currentInfo); } } - //Creates an appropriate action for each server file - private IEnumerable ObjectsToActions(AccountInfo accountInfo,IEnumerable remote) + private void MoveLocalFile(AccountInfo accountInfo, FileSystemInfo previousFile, FileAgent fileAgent, + ObjectInfo currentInfo) { - if (remote==null) - throw new ArgumentNullException(); - Contract.EndContractBlock(); - var fileAgent = GetFileAgent(accountInfo); + var currentRelativepath = currentInfo.RelativeUrlToFilePath(accountInfo.UserName); + var newPath = Path.Combine(fileAgent.RootPath, currentRelativepath); - //In order to avoid multiple iterations over the files, we iterate only once - //over the remote files - foreach (var objectInfo in remote) + var isFile= (previousFile is FileInfo); + var previousFullPath = isFile? + FileInfoExtensions.GetProperFilePathCapitalization(previousFile.FullName): + FileInfoExtensions.GetProperDirectoryCapitalization(previousFile.FullName); + + using (NetworkGate.Acquire(previousFullPath, NetworkOperation.Renaming)) + using (NetworkGate.Acquire(newPath,NetworkOperation.Renaming)) + using (new SessionScope(FlushAction.Auto)) { - var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName); - //and remove any matching objects from the list, adding them to the commonObjects list - - if (fileAgent.Exists(relativePath)) - { - var localFile = fileAgent.GetFileInfo(relativePath); - var state = FileState.FindByFilePath(localFile.FullName); - //Common files should be checked on a per-case basis to detect differences, which is newer - - yield return new CloudAction(accountInfo,CloudActionType.MustSynch, - localFile, objectInfo, state, accountInfo.BlockSize, - accountInfo.BlockHash); - } + if (isFile) + (previousFile as FileInfo).MoveTo(newPath); else { - //If there is no match we add them to the localFiles list - //but only if the file is not marked for deletion - var targetFile = Path.Combine(accountInfo.AccountPath, relativePath); - var fileStatus = StatusKeeper.GetFileStatus(targetFile); - if (fileStatus != FileStatus.Deleted) - { - //Remote files should be downloaded - yield return new CloudDownloadAction(accountInfo,objectInfo); - } + (previousFile as DirectoryInfo).MoveTo(newPath); } + var state = StatusKeeper.GetStateByFilePath(previousFullPath); + state.FilePath = newPath; + state.SaveCopy(); + StatusKeeper.SetFileState(previousFullPath,FileStatus.Deleted,FileOverlayStatus.Deleted, "Deleted"); } } - private static FileAgent GetFileAgent(AccountInfo accountInfo) - { - return AgentLocator.Get(accountInfo.AccountPath); - } - - private void ProcessDeletedFiles(AccountInfo accountInfo,IEnumerable trashObjects) - { - var fileAgent = GetFileAgent(accountInfo); - foreach (var trashObject in trashObjects) - { - var relativePath = trashObject.RelativeUrlToFilePath(accountInfo.UserName); - //and remove any matching objects from the list, adding them to the commonObjects list - fileAgent.Delete(relativePath); - } - } - - - private void RenameCloudFile(AccountInfo accountInfo,CloudMoveAction action) + private async Task SyncFiles(AccountInfo accountInfo,CloudAction action) { - if (accountInfo==null) + if (accountInfo == null) throw new ArgumentNullException("accountInfo"); if (action==null) throw new ArgumentNullException("action"); - if (action.CloudFile==null) - throw new ArgumentException("CloudFile","action"); if (action.LocalFile==null) - throw new ArgumentException("LocalFile","action"); - if (action.OldLocalFile==null) - throw new ArgumentException("OldLocalFile","action"); - if (action.OldCloudFile==null) - throw new ArgumentException("OldCloudFile","action"); - Contract.EndContractBlock(); - - var newFilePath = action.LocalFile.FullName; - //The local file is already renamed - this.StatusKeeper.SetFileOverlayStatus(newFilePath, FileOverlayStatus.Modified); - - - var account = action.CloudFile.Account ?? accountInfo.UserName; - var container = action.CloudFile.Container; - - var client = new CloudFilesClient(accountInfo); - client.MoveObject(account, container, action.OldCloudFile.Name, container, action.CloudFile.Name); - - this.StatusKeeper.SetFileStatus(newFilePath, FileStatus.Unchanged); - this.StatusKeeper.SetFileOverlayStatus(newFilePath, FileOverlayStatus.Normal); - NativeMethods.RaiseChangeNotification(newFilePath); - } - - private void DeleteCloudFile(AccountInfo accountInfo, ObjectInfo cloudFile,string fileName) - { - if (accountInfo == null) - throw new ArgumentNullException("accountInfo"); - if (cloudFile==null) - throw new ArgumentNullException("cloudFile"); - - if (String.IsNullOrWhiteSpace(fileName)) - throw new ArgumentNullException("fileName"); - if (Path.IsPathRooted(fileName)) - throw new ArgumentException("The fileName should not be rooted","fileName"); - if (String.IsNullOrWhiteSpace(cloudFile.Container)) - throw new ArgumentException("Invalid container", "cloudFile"); + throw new ArgumentException("The action's local file is not specified","action"); + if (!Path.IsPathRooted(action.LocalFile.FullName)) + throw new ArgumentException("The action's local file path must be absolute","action"); + if (action.CloudFile== null) + throw new ArgumentException("The action's cloud file is not specified", "action"); Contract.EndContractBlock(); - - var fileAgent = GetFileAgent(accountInfo); - - using ( log4net.ThreadContext.Stacks["DeleteCloudFile"].Push("Delete")) + using (ThreadContext.Stacks["Operation"].Push("SyncFiles")) { - var info = fileAgent.GetFileInfo(fileName); - var fullPath = info.FullName.ToLower(); - this.StatusKeeper.SetFileOverlayStatus(fullPath, FileOverlayStatus.Modified); - var account = cloudFile.Account ?? accountInfo.UserName; - var container = cloudFile.Container ;//?? FolderConstants.PithosContainer; + //var localFile = action.LocalFile; + var cloudFile = action.CloudFile; + var downloadPath = action.LocalFile.GetProperCapitalization(); - var client = new CloudFilesClient(accountInfo); - client.DeleteObject(account, container, fileName); + var cloudHash = cloudFile.X_Object_Hash.ToLower(); + var previousCloudHash = cloudFile.PreviousHash == null?null: cloudFile.PreviousHash.ToLower(); + var localHash = action.TreeHash.Value.TopHash.ToHashString();// LocalHash.Value.ToLower(); + //var topHash = action.TopHash.Value.ToLower(); - this.StatusKeeper.ClearFileStatus(fullPath); - } - } + if(cloudFile.IsDirectory && action.LocalFile is DirectoryInfo) + { + Log.InfoFormat("Skipping folder {0} , exists in server", downloadPath); + return; + } - //Download a file. - private async Task DownloadCloudFile(AccountInfo accountInfo, ObjectInfo cloudFile , string localPath) - { - if (accountInfo == null) - throw new ArgumentNullException("accountInfo"); - if (cloudFile == null) - throw new ArgumentNullException("cloudFile"); - if (String.IsNullOrWhiteSpace(cloudFile.Account)) - throw new ArgumentNullException("cloudFile"); - if (String.IsNullOrWhiteSpace(cloudFile.Container)) - throw new ArgumentNullException("cloudFile"); - if (String.IsNullOrWhiteSpace(localPath)) - throw new ArgumentNullException("localPath"); - if (!Path.IsPathRooted(localPath)) - throw new ArgumentException("The localPath must be rooted", "localPath"); - Contract.EndContractBlock(); - - Uri relativeUrl = new Uri(cloudFile.Name, UriKind.Relative); + //At this point we know that an object has changed on the server and that a local + //file already exists. We need to decide whether the file has only changed on + //the server or there is a conflicting change on the client. + // - var url = relativeUrl.ToString(); - if (cloudFile.Name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase)) - return; + //If the hashes match, we are done + if (cloudFile != ObjectInfo.Empty && cloudHash == localHash) + { + Log.InfoFormat("Skipping {0}, hashes match", downloadPath); + return; + } - //Are we already downloading or uploading the file? - using (var gate=NetworkGate.Acquire(localPath, NetworkOperation.Downloading)) - { - if (gate.Failed) + //If the local and remote files have 0 length their hashes will not match + if (!cloudFile.IsDirectory && cloudFile.Bytes==0 && action.LocalFile is FileInfo && (action.LocalFile as FileInfo).Length==0 ) + { + Log.InfoFormat("Skipping {0}, files are empty", downloadPath); return; - //The file's hashmap will be stored in the same location with the extension .hashmap - //var hashPath = Path.Combine(FileAgent.CachePath, relativePath + ".hashmap"); - - var client = new CloudFilesClient(accountInfo); - var account = cloudFile.Account; - var container = cloudFile.Container; - - //Retrieve the hashmap from the server - var serverHash = await client.GetHashMap(account, container, url); - //If it's a small file - if (serverHash.Hashes.Count == 1 ) - //Download it in one go - await DownloadEntireFileAsync(accountInfo, client, cloudFile, relativeUrl, localPath, serverHash); - //Otherwise download it block by block - else - await DownloadWithBlocks(accountInfo,client, cloudFile, relativeUrl, localPath, serverHash); + } + + //The hashes DON'T match. We need to sync - if (cloudFile.AllowedTo == "read") + // If the previous tophash matches the local tophash, the file was only changed on the server. + if (localHash == previousCloudHash) + { + await Downloader.DownloadCloudFile(accountInfo, cloudFile, downloadPath, CurrentOperationCancelToken); + } + else { - var attributes=File.GetAttributes(localPath); - File.SetAttributes(localPath,attributes|FileAttributes.ReadOnly); + //If the previous and local hash don't match, there was a local conflict + //that was not uploaded to the server. We have a conflict + ReportConflictForMismatch(downloadPath); } - - //Now we can store the object's metadata without worrying about ghost status entries - StatusKeeper.StoreInfo(localPath, cloudFile); - } } - //Download a small file with a single GET operation - private async Task DownloadEntireFileAsync(AccountInfo accountInfo, CloudFilesClient client, ObjectInfo cloudFile, Uri relativeUrl, string localPath,TreeHash serverHash) + private void ReportConflictForMismatch(string downloadPath) { - if (client == null) - throw new ArgumentNullException("client"); - if (cloudFile==null) - throw new ArgumentNullException("cloudFile"); - if (relativeUrl == null) - throw new ArgumentNullException("relativeUrl"); - if (String.IsNullOrWhiteSpace(localPath)) - throw new ArgumentNullException("localPath"); - if (!Path.IsPathRooted(localPath)) - throw new ArgumentException("The localPath must be rooted", "localPath"); + if (String.IsNullOrWhiteSpace(downloadPath)) + throw new ArgumentNullException("downloadPath"); Contract.EndContractBlock(); - //If the file already exists - if (File.Exists(localPath)) - { - //First check with MD5 as this is a small file - var localMD5 = Signature.CalculateMD5(localPath); - var cloudHash=serverHash.TopHash.ToHashString(); - if (localMD5==cloudHash) - return; - //Then check with a treehash - var localTreeHash = Signature.CalculateTreeHash(localPath, serverHash.BlockSize, serverHash.BlockHash); - var localHash = localTreeHash.TopHash.ToHashString(); - if (localHash==cloudHash) - return; - } - - var fileAgent = GetFileAgent(accountInfo); - //Calculate the relative file path for the new file - var relativePath = relativeUrl.RelativeUriToFilePath(); - //The file will be stored in a temporary location while downloading with an extension .download - var tempPath = Path.Combine(fileAgent.CachePath, relativePath + ".download"); - //Make sure the target folder exists. DownloadFileTask will not create the folder - var tempFolder = Path.GetDirectoryName(tempPath); - if (!Directory.Exists(tempFolder)) - Directory.CreateDirectory(tempFolder); - - //Download the object to the temporary location - await client.GetObject(cloudFile.Account, cloudFile.Container, relativeUrl.ToString(), tempPath).ContinueWith(t => - { - t.PropagateExceptions(); - //Create the local folder if it doesn't exist (necessary for shared objects) - var localFolder = Path.GetDirectoryName(localPath); - if (!Directory.Exists(localFolder)) - Directory.CreateDirectory(localFolder); - //And move it to its actual location once downloading is finished - if (File.Exists(localPath)) - File.Replace(tempPath,localPath,null,true); - else - File.Move(tempPath,localPath); - //Notify listeners that a local file has changed - StatusNotification.NotifyChangedFile(localPath); - - }); + StatusKeeper.SetFileState(downloadPath,FileStatus.Conflict, FileOverlayStatus.Conflict,"File changed at the server"); + UpdateStatus(PithosStatus.HasConflicts); + var message = String.Format("Conflict detected for file {0}", downloadPath); + Log.Warn(message); + StatusNotification.NotifyChange(message, TraceLevel.Warning); } - //Download a file asynchronously using blocks - public async Task DownloadWithBlocks(AccountInfo accountInfo, CloudFilesClient client, ObjectInfo cloudFile, Uri relativeUrl, string localPath, TreeHash serverHash) + public void Post(CloudAction cloudAction) { - if (client == null) - throw new ArgumentNullException("client"); - if (cloudFile == null) - throw new ArgumentNullException("cloudFile"); - if (relativeUrl == null) - throw new ArgumentNullException("relativeUrl"); - if (String.IsNullOrWhiteSpace(localPath)) - throw new ArgumentNullException("localPath"); - if (!Path.IsPathRooted(localPath)) - throw new ArgumentException("The localPath must be rooted", "localPath"); - if (serverHash == null) - throw new ArgumentNullException("serverHash"); + if (cloudAction == null) + throw new ArgumentNullException("cloudAction"); + if (cloudAction.AccountInfo==null) + throw new ArgumentException("The CloudAction.AccountInfo is empty","cloudAction"); Contract.EndContractBlock(); - - var fileAgent = GetFileAgent(accountInfo); - - //Calculate the relative file path for the new file - var relativePath = relativeUrl.RelativeUriToFilePath(); - var blockUpdater = new BlockUpdater(fileAgent.CachePath, localPath, relativePath, serverHash); + DeleteAgent.ProceedEvent.Wait(); - - //Calculate the file's treehash - var treeHash = await Signature.CalculateTreeHashAsync(localPath, serverHash.BlockSize, serverHash.BlockHash); - - //And compare it with the server's hash - var upHashes = serverHash.GetHashesAsStrings(); - var localHashes = treeHash.HashDictionary; - for (int i = 0; i < upHashes.Length; i++) - { - //For every non-matching hash - var upHash = upHashes[i]; - if (!localHashes.ContainsKey(upHash)) - { - if (blockUpdater.UseOrphan(i, upHash)) - { - Log.InfoFormat("[BLOCK GET] ORPHAN FOUND for {0} of {1} for {2}", i, upHashes.Length, localPath); - continue; - } - Log.InfoFormat("[BLOCK GET] START {0} of {1} for {2}", i, upHashes.Length, localPath); - var start = i*serverHash.BlockSize; - //To download the last block just pass a null for the end of the range - long? end = null; - if (i < upHashes.Length - 1 ) - end= ((i + 1)*serverHash.BlockSize) ; - - //Download the missing block - var block = await client.GetBlock(cloudFile.Account, cloudFile.Container, relativeUrl, start, end); - - //and store it - blockUpdater.StoreBlock(i, block); - - - Log.InfoFormat("[BLOCK GET] FINISH {0} of {1} for {2}", i, upHashes.Length, localPath); - } - } - - //Want to avoid notifications if no changes were made - var hasChanges = blockUpdater.HasBlocks; - blockUpdater.Commit(); - - if (hasChanges) - //Notify listeners that a local file has changed - StatusNotification.NotifyChangedFile(localPath); - - Log.InfoFormat("[BLOCK GET] COMPLETE {0}", localPath); + if (cloudAction is CloudDeleteAction) + DeleteAgent.Post((CloudDeleteAction)cloudAction); + else + _agent.Post(cloudAction); } + - - private async Task UploadCloudFile(CloudAction action) + public IEnumerable GetEnumerable() { - if (action == null) - throw new ArgumentNullException("action"); - Contract.EndContractBlock(); - - try - { - if (action == null) - throw new ArgumentNullException("action"); - Contract.EndContractBlock(); - - var accountInfo=action.AccountInfo; - - var fileInfo=action.LocalFile; - - if (fileInfo.Extension.Equals("ignore", StringComparison.InvariantCultureIgnoreCase)) - return; - - var relativePath = fileInfo.AsRelativeTo(accountInfo.AccountPath); - if (relativePath.StartsWith(FolderConstants.OthersFolder)) - { - var parts=relativePath.Split('\\'); - var accountName = parts[1]; - var oldName = accountInfo.UserName; - var absoluteUri = accountInfo.StorageUri.AbsoluteUri; - var nameIndex=absoluteUri.IndexOf(oldName); - var root=absoluteUri.Substring(0, nameIndex); - - accountInfo = new AccountInfo - { - UserName = accountName, - AccountPath = Path.Combine(accountInfo.AccountPath, parts[0], parts[1]), - StorageUri = new Uri(root + accountName), - BlockHash=accountInfo.BlockHash, - BlockSize=accountInfo.BlockSize, - Token=accountInfo.Token - }; - } - - - var fullFileName = fileInfo.FullName; - using(var gate=NetworkGate.Acquire(fullFileName,NetworkOperation.Uploading)) - { - //Abort if the file is already being uploaded or downloaded - if (gate.Failed) - return; - - var cloudFile = action.CloudFile; - var account = cloudFile.Account ?? accountInfo.UserName; - - var client = new CloudFilesClient(accountInfo); - //Even if GetObjectInfo times out, we can proceed with the upload - var info = client.GetObjectInfo(account, cloudFile.Container, cloudFile.Name); - var cloudHash = info.Hash.ToLower(); - - var hash = action.LocalHash.Value; - var topHash = action.TopHash.Value; - - //If the file hashes match, abort the upload - if (hash == cloudHash || topHash ==cloudHash) - { - //but store any metadata changes - this.StatusKeeper.StoreInfo(fullFileName, info); - Log.InfoFormat("Skip upload of {0}, hashes match", fullFileName); - return; - } - - if (info.AllowedTo=="read") - return; - - //Mark the file as modified while we upload it - StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified); - //And then upload it - - //Upload even small files using the Hashmap. The server may already containt - //the relevant folder + return _agent.GetEnumerable(); + } - //First, calculate the tree hash - var treeHash = await Signature.CalculateTreeHashAsync(fileInfo.FullName, accountInfo.BlockSize, - accountInfo.BlockHash); - - await UploadWithHashMap(accountInfo,cloudFile,fileInfo,cloudFile.Name,treeHash); + public Task GetDeleteAwaiter() + { + return DeleteAgent.ProceedEvent.WaitAsync(); + } + public CancellationToken CancellationToken + { + get { return _agent.CancellationToken; } + } - //If everything succeeds, change the file and overlay status to normal - this.StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal); - } - //Notify the Shell to update the overlays - NativeMethods.RaiseChangeNotification(fullFileName); - StatusNotification.NotifyChangedFile(fullFileName); + public bool Pause + { + get { + return _pause; } - catch (AggregateException ex) - { - var exc = ex.InnerException as WebException; - if (exc == null) - throw ex.InnerException; - var response = exc.Response as HttpWebResponse; - if (response == null) - throw exc; - if (response.StatusCode == HttpStatusCode.Unauthorized) + set { + _pause = value; + if (_pause) + _unPauseEvent.Reset(); + else { - Log.Error("Not allowed to upload file", exc); - var message = String.Format("Not allowed to uplad file {0}", action.LocalFile.FullName); - StatusKeeper.SetFileState(action.LocalFile.FullName, FileStatus.Unchanged, FileOverlayStatus.Normal); - StatusNotification.NotifyChange(message, TraceLevel.Warning); - return; + _unPauseEvent.Set(); } - throw; } - } - public async Task UploadWithHashMap(AccountInfo accountInfo,ObjectInfo cloudFile,FileInfo fileInfo,string url,TreeHash treeHash) + + private void RenameCloudFile(AccountInfo accountInfo,CloudMoveAction action) { - if (accountInfo == null) + if (accountInfo==null) throw new ArgumentNullException("accountInfo"); - if (cloudFile==null) - throw new ArgumentNullException("cloudFile"); - if (fileInfo == null) - throw new ArgumentNullException("fileInfo"); - if (String.IsNullOrWhiteSpace(url)) - throw new ArgumentNullException(url); - if (treeHash==null) - throw new ArgumentNullException("treeHash"); - if (String.IsNullOrWhiteSpace(cloudFile.Container) ) - throw new ArgumentException("Invalid container","cloudFile"); + if (action==null) + throw new ArgumentNullException("action"); + if (action.CloudFile==null) + throw new ArgumentException("CloudFile","action"); + if (action.LocalFile==null) + throw new ArgumentException("LocalFile","action"); + if (action.OldLocalFile==null) + throw new ArgumentException("OldLocalFile","action"); + if (action.OldCloudFile==null) + throw new ArgumentException("OldCloudFile","action"); Contract.EndContractBlock(); - var fullFileName = fileInfo.FullName; + using (ThreadContext.Stacks["Operation"].Push("RenameCloudFile")) + { - var account = cloudFile.Account ?? accountInfo.UserName; - var container = cloudFile.Container ; + var newFilePath = action.LocalFile.FullName; - var client = new CloudFilesClient(accountInfo); - //Send the hashmap to the server - var missingHashes = await client.PutHashMap(account, container, url, treeHash); - //If the server returns no missing hashes, we are done - while (missingHashes.Count > 0) - { + //How do we handle concurrent renames and deletes/uploads/downloads? + //* A conflicting upload means that a file was renamed before it had a chance to finish uploading + // This should never happen as the network agent executes only one action at a time + //* A conflicting download means that the file was modified on the cloud. While we can go on and complete + // the rename, there may be a problem if the file is downloaded in blocks, as subsequent block requests for the + // same name will fail. + // This should never happen as the network agent executes only one action at a time. + //* A conflicting delete can happen if the rename was followed by a delete action that didn't have the chance + // to remove the rename from the queue. + // We can probably ignore this case. It will result in an error which should be ignored - var buffer = new byte[accountInfo.BlockSize]; - foreach (var missingHash in missingHashes) - { - //Find the proper block - var blockIndex = treeHash.HashDictionary[missingHash]; - var offset = blockIndex*accountInfo.BlockSize; - var read = fileInfo.Read(buffer, offset, accountInfo.BlockSize); + //The local file is already renamed + StatusKeeper.SetFileOverlayStatus(newFilePath, FileOverlayStatus.Modified).Wait(); - try - { - //And upload the block - await client.PostBlock(account, container, buffer, 0, read); - Log.InfoFormat("[BLOCK] Block {0} of {1} uploaded", blockIndex, fullFileName); - } - catch (Exception exc) - { - Log.ErrorFormat("[ERROR] uploading block {0} of {1}\n{2}", blockIndex, fullFileName, exc); - } - } + var account = action.CloudFile.Account ?? accountInfo.UserName; + var container = action.CloudFile.Container; - //Repeat until there are no more missing hashes - missingHashes = await client.PutHashMap(account, container, url, treeHash); + var client = new CloudFilesClient(accountInfo); + //TODO: What code is returned when the source file doesn't exist? + client.MoveObject(account, container, action.OldCloudFile.Name, container, action.CloudFile.Name); + + StatusKeeper.SetFileStatus(newFilePath, FileStatus.Unchanged); + StatusKeeper.SetFileOverlayStatus(newFilePath, FileOverlayStatus.Normal).Wait(); + NativeMethods.RaiseChangeNotification(newFilePath); } } - public void AddAccount(AccountInfo accountInfo) - { - if (!_accounts.Contains(accountInfo)) - _accounts.Add(accountInfo); - } + + }