code.grnet.gr Git - pithos-ms-client/blob - trunk/Pithos.Core/Agents/PollAgent.cs

   1 #region\r
   2 /* -----------------------------------------------------------------------\r
   3  * <copyright file="PollAgent.cs" company="GRNet">\r
   4  * \r
   5  * Copyright 2011-2012 GRNET S.A. All rights reserved.\r
   6  *\r
   7  * Redistribution and use in source and binary forms, with or\r
   8  * without modification, are permitted provided that the following\r
   9  * conditions are met:\r
  10  *\r
  11  *   1. Redistributions of source code must retain the above\r
  12  *      copyright notice, this list of conditions and the following\r
  13  *      disclaimer.\r
  14  *\r
  15  *   2. Redistributions in binary form must reproduce the above\r
  16  *      copyright notice, this list of conditions and the following\r
  17  *      disclaimer in the documentation and/or other materials\r
  18  *      provided with the distribution.\r
  19  *\r
  20  *\r
  21  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS\r
  22  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED\r
  23  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR\r
  24  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR\r
  25  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\r
  26  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\r
  27  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF\r
  28  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED\r
  29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT\r
  30  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN\r
  31  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE\r
  32  * POSSIBILITY OF SUCH DAMAGE.\r
  33  *\r
  34  * The views and conclusions contained in the software and\r
  35  * documentation are those of the authors and should not be\r
  36  * interpreted as representing official policies, either expressed\r
  37  * or implied, of GRNET S.A.\r
  38  * </copyright>\r
  39  * -----------------------------------------------------------------------\r
  40  */\r
  41 #endregion\r
  42 \r
  43 using System.Collections.Concurrent;\r
  44 using System.ComponentModel.Composition;\r
  45 using System.Diagnostics;\r
  46 using System.Diagnostics.Contracts;\r
  47 using System.IO;\r
  48 using System.Linq.Expressions;\r
  49 using System.Reflection;\r
  50 using System.Security.Cryptography;\r
  51 using System.Threading;\r
  52 using System.Threading.Tasks;\r
  53 using System.Threading.Tasks.Dataflow;\r
  54 using Castle.ActiveRecord;\r
  55 using Pithos.Interfaces;\r
  56 using Pithos.Network;\r
  57 using log4net;\r
  58 \r
  59 namespace Pithos.Core.Agents\r
  60 {\r
  61     using System;\r
  62     using System.Collections.Generic;\r
  63     using System.Linq;\r
  64 \r
  65     /*public class PollRequest\r
  66     {\r
  67         public DateTime? Since { get; set; }\r
  68         public IEnumerable<string> Batch { get; set; }\r
  69     }*/\r
  70 \r
  71 \r
  72     /// <summary>\r
  73     /// PollAgent periodically polls the server to detect object changes. The agent retrieves a listing of all\r
  74     /// objects and compares it with a previously cached version to detect differences. \r
  75     /// New files are downloaded, missing files are deleted from the local file system and common files are compared\r
  76     /// to determine the appropriate action\r
  77     /// </summary>\r
  78     [Export]\r
  79     public class PollAgent\r
  80     {\r
  81         private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);\r
  82 \r
  83         [System.ComponentModel.Composition.Import]\r
  84         public IStatusKeeper StatusKeeper { get; set; }\r
  85 \r
  86         [System.ComponentModel.Composition.Import]\r
  87         public IPithosSettings Settings { get; set; }\r
  88 \r
  89         [System.ComponentModel.Composition.Import]\r
  90         public NetworkAgent NetworkAgent { get; set; }\r
  91 \r
  92         [System.ComponentModel.Composition.Import]\r
  93         public Selectives Selectives { get; set; }\r
  94 \r
  95         public IStatusNotification StatusNotification { get; set; }\r
  96 \r
  97         private CancellationTokenSource _currentOperationCancellation = new CancellationTokenSource();\r
  98 \r
  99         public void CancelCurrentOperation()\r
 100         {\r
 101             //What does it mean to cancel the current upload/download?\r
 102             //Obviously, the current operation will be cancelled by throwing\r
 103             //a cancellation exception.\r
 104             //\r
 105             //The default behavior is to retry any operations that throw.\r
 106             //Obviously this is not what we want in this situation.\r
 107             //The cancelled operation should NOT bea retried. \r
 108             //\r
 109             //This can be done by catching the cancellation exception\r
 110             //and avoiding the retry.\r
 111             //\r
 112 \r
 113             //Have to reset the cancellation source - it is not possible to reset the source\r
 114             //Have to prevent a case where an operation requests a token from the old source\r
 115             var oldSource = Interlocked.Exchange(ref _currentOperationCancellation, new CancellationTokenSource());\r
 116             oldSource.Cancel();\r
 117 \r
 118         }\r
 119 \r
 120         public bool Pause\r
 121         {\r
 122             get {\r
 123                 return _pause;\r
 124             }\r
 125             set {\r
 126                 _pause = value;                \r
 127                 if (!_pause)\r
 128                     _unPauseEvent.Set();\r
 129                 else\r
 130                 {\r
 131                     _unPauseEvent.Reset();\r
 132                 }\r
 133             }\r
 134         }\r
 135 \r
 136         private bool _firstPoll = true;\r
 137 \r
 138         //The Sync Event signals a manual synchronisation\r
 139         private readonly AsyncManualResetEvent _syncEvent = new AsyncManualResetEvent();\r
 140 \r
 141         private readonly AsyncManualResetEvent _unPauseEvent = new AsyncManualResetEvent(true);\r
 142 \r
 143         private readonly ConcurrentDictionary<string, DateTime> _lastSeen = new ConcurrentDictionary<string, DateTime>();\r
 144         private readonly ConcurrentDictionary<Uri, AccountInfo> _accounts = new ConcurrentDictionary<Uri,AccountInfo>();\r
 145 \r
 146         //private readonly ActionBlock<PollRequest>  _pollAction;\r
 147 \r
 148         readonly HashSet<string> _knownContainers = new HashSet<string>();\r
 149 \r
 150         \r
 151         /// <summary>\r
 152         /// Start a manual synchronization\r
 153         /// </summary>\r
 154         public void SynchNow(IEnumerable<string> paths=null)\r
 155         {\r
 156             _batchQueue.Enqueue(paths);\r
 157             _syncEvent.Set();                \r
 158 \r
 159             //_pollAction.Post(new PollRequest {Batch = paths});\r
 160         }\r
 161 \r
 162         readonly ConcurrentQueue<IEnumerable<string>> _batchQueue=new ConcurrentQueue<IEnumerable<string>>();\r
 163 \r
 164         /// <summary>\r
 165         /// Remote files are polled periodically. Any changes are processed\r
 166         /// </summary>\r
 167         /// <param name="since"></param>\r
 168         /// <returns></returns>\r
 169         public  void PollRemoteFiles(DateTime? since = null)\r
 170         {\r
 171             if (Log.IsDebugEnabled)\r
 172                 Log.DebugFormat("Polling changes after [{0}]",since);\r
 173 \r
 174             Debug.Assert(Thread.CurrentThread.IsBackground, "Polling Ended up in the main thread!");\r
 175 \r
 176             //GC.Collect();\r
 177 \r
 178             using (ThreadContext.Stacks["Retrieve Remote"].Push("All accounts"))\r
 179             {\r
 180                 //If this poll fails, we will retry with the same since value\r
 181                 var nextSince = since;\r
 182                 try\r
 183                 {\r
 184                     _unPauseEvent.Wait();\r
 185                     UpdateStatus(PithosStatus.PollSyncing);\r
 186 \r
 187                     var accountBatches=new Dictionary<Uri, IEnumerable<string>>();\r
 188                     IEnumerable<string> batch = null;\r
 189                     if (_batchQueue.TryDequeue(out batch) && batch != null)\r
 190                         foreach (var account in _accounts.Values)\r
 191                         {\r
 192                             var accountBatch = batch.Where(path => path.IsAtOrBelow(account.AccountPath));\r
 193                             accountBatches[account.AccountKey] = accountBatch;\r
 194                         }\r
 195 \r
 196                     var tasks = new List<Task<DateTime?>>();\r
 197                     foreach(var accountInfo in _accounts.Values)\r
 198                     {\r
 199                         IEnumerable<string> accountBatch ;\r
 200                         accountBatches.TryGetValue(accountInfo.AccountKey,out accountBatch);\r
 201                         var t=ProcessAccountFiles (accountInfo, accountBatch, since);\r
 202                         tasks.Add(t);\r
 203                     }\r
 204 \r
 205                     var nextTimes=TaskEx.WhenAll(tasks.ToList()).Result;\r
 206 \r
 207                     _firstPoll = false;\r
 208                     //Reschedule the poll with the current timestamp as a "since" value\r
 209 \r
 210                     if (nextTimes.Length>0)\r
 211                         nextSince = nextTimes.Min();\r
 212                     if (Log.IsDebugEnabled)\r
 213                         Log.DebugFormat("Next Poll at [{0}]",nextSince);\r
 214                 }\r
 215                 catch (Exception ex)\r
 216                 {\r
 217                     Log.ErrorFormat("Error while processing accounts\r\n{0}", ex);\r
 218                     //In case of failure retry with the same "since" value\r
 219                 }\r
 220 \r
 221                 UpdateStatus(PithosStatus.PollComplete);\r
 222                 //The multiple try blocks are required because we can't have an await call\r
 223                 //inside a finally block\r
 224                 //TODO: Find a more elegant solution for reschedulling in the event of an exception\r
 225                 try\r
 226                 {\r
 227                     //Wait for the polling interval to pass or the Sync event to be signalled\r
 228                     nextSince = WaitForScheduledOrManualPoll(nextSince).Result;\r
 229                 }\r
 230                 finally\r
 231                 {\r
 232                     //Ensure polling is scheduled even in case of error\r
 233                     TaskEx.Run(()=>PollRemoteFiles(nextSince));\r
 234                     //_pollAction.Post(new PollRequest {Since = nextSince});\r
 235                 }\r
 236             }\r
 237         }\r
 238 \r
 239         /// <summary>\r
 240         /// Wait for the polling period to expire or a manual sync request\r
 241         /// </summary>\r
 242         /// <param name="since"></param>\r
 243         /// <returns></returns>\r
 244         private async Task<DateTime?> WaitForScheduledOrManualPoll(DateTime? since)\r
 245         {\r
 246             var sync = _syncEvent.WaitAsync();\r
 247             var wait = TaskEx.Delay(TimeSpan.FromSeconds(Settings.PollingInterval));\r
 248 \r
 249             var signaledTask = await TaskEx.WhenAny(sync, wait).ConfigureAwait(false);\r
 250             \r
 251             //Pausing takes precedence over manual sync or awaiting\r
 252             _unPauseEvent.Wait();\r
 253             \r
 254             //Wait for network processing to finish before polling\r
 255             var pauseTask=NetworkAgent.ProceedEvent.WaitAsync();\r
 256             await TaskEx.WhenAll(signaledTask, pauseTask).ConfigureAwait(false);\r
 257 \r
 258             //If polling is signalled by SynchNow, ignore the since tag\r
 259             if (sync.IsCompleted)\r
 260             {                \r
 261                 _syncEvent.Reset();\r
 262                 return null;\r
 263             }\r
 264             return since;\r
 265         }\r
 266 \r
 267         \r
 268 \r
 269         public async Task<DateTime?> ProcessAccountFiles(AccountInfo accountInfo, IEnumerable<string> accountBatch, DateTime? since = null)\r
 270         {\r
 271             if (accountInfo == null)\r
 272                 throw new ArgumentNullException("accountInfo");\r
 273             if (String.IsNullOrWhiteSpace(accountInfo.AccountPath))\r
 274                 throw new ArgumentException("The AccountInfo.AccountPath is empty", "accountInfo");\r
 275             Contract.EndContractBlock();\r
 276 \r
 277 \r
 278             using (ThreadContext.Stacks["Retrieve Remote"].Push(accountInfo.UserName))\r
 279             {\r
 280 \r
 281                 await NetworkAgent.GetDeleteAwaiter().ConfigureAwait(false);\r
 282 \r
 283                 Log.Info("Scheduled");\r
 284                 var client = new CloudFilesClient(accountInfo);\r
 285 \r
 286                 //We don't need to check the trash container\r
 287                 var containers = client.ListContainers(accountInfo.UserName)\r
 288                     .Where(c=>c.Name!="trash")\r
 289                     .ToList();\r
 290 \r
 291 \r
 292                 CreateContainerFolders(accountInfo, containers);\r
 293 \r
 294                 //The nextSince time fallback time is the same as the current.\r
 295                 //If polling succeeds, the next Since time will be the smallest of the maximum modification times\r
 296                 //of the shared and account objects\r
 297                 var nextSince = since;\r
 298 \r
 299                 try\r
 300                 {\r
 301                     //Wait for any deletions to finish\r
 302                     await NetworkAgent.GetDeleteAwaiter().ConfigureAwait(false);\r
 303                     //Get the poll time now. We may miss some deletions but it's better to keep a file that was deleted\r
 304                     //than delete a file that was created while we were executing the poll                    \r
 305 \r
 306                     //Get the list of server objects changed since the last check\r
 307                     //The name of the container is passed as state in order to create a dictionary of tasks in a subsequent step\r
 308                     var listObjects = (from container in containers\r
 309                                        select Task<IList<ObjectInfo>>.Factory.StartNew(_ =>\r
 310                                              client.ListObjects(accountInfo.UserName, container.Name, since), container.Name)).ToList();\r
 311 \r
 312                     var listShared = Task<IList<ObjectInfo>>.Factory.StartNew(_ => \r
 313                         client.ListSharedObjects(_knownContainers,since), "shared");\r
 314                     listObjects.Add(listShared);\r
 315                     var listTasks = await Task.Factory.WhenAll(listObjects.ToArray()).ConfigureAwait(false);\r
 316 \r
 317                     using (ThreadContext.Stacks["SCHEDULE"].Push("Process Results"))\r
 318                     {\r
 319                         var dict = listTasks.ToDictionary(t => t.AsyncState);\r
 320 \r
 321                         //Get all non-trash objects. Remember, the container name is stored in AsyncState\r
 322                         var remoteObjects = (from objectList in listTasks\r
 323                                             where (string)objectList.AsyncState != "trash"\r
 324                                             from obj in objectList.Result\r
 325                                             orderby obj.Bytes ascending \r
 326                                             select obj).ToList();\r
 327                         \r
 328                         //Get the latest remote object modification date, only if it is after\r
 329                         //the original since date                        \r
 330                         nextSince = GetLatestDateAfter(nextSince, remoteObjects);\r
 331 \r
 332                         var sharedObjects = dict["shared"].Result;\r
 333 \r
 334                         //DON'T process trashed files\r
 335                         //If some files are deleted and added again to a folder, they will be deleted\r
 336                         //even though they are new.\r
 337                         //We would have to check file dates and hashes to ensure that a trashed file\r
 338                         //can be deleted safely from the local hard drive.\r
 339                         /*\r
 340                         //Items with the same name, hash may be both in the container and the trash\r
 341                         //Don't delete items that exist in the container\r
 342                         var realTrash = from trash in trashObjects\r
 343                                         where\r
 344                                             !remoteObjects.Any(\r
 345                                                 info => info.Name == trash.Name && info.Hash == trash.Hash)\r
 346                                    8     select trash;\r
 347                         ProcessTrashedFiles(accountInfo, realTrash);\r
 348 */\r
 349 \r
 350                         var cleanRemotes = (from info in remoteObjects.Union(sharedObjects)\r
 351                                             let name = info.Name??""\r
 352                                             where !name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase) &&\r
 353                                                   !name.StartsWith(FolderConstants.CacheFolder + "/",\r
 354                                                                    StringComparison.InvariantCultureIgnoreCase)\r
 355                                             select info).ToList();\r
 356 \r
 357                         if (_firstPoll)\r
 358                             StatusKeeper.CleanupOrphanStates();\r
 359                         \r
 360                         var differencer = _differencer.PostSnapshot(accountInfo, cleanRemotes);\r
 361                         var currentRemotes = differencer.Current.ToList();\r
 362                         StatusKeeper.CleanupStaleStates(accountInfo, currentRemotes);\r
 363 \r
 364                         //var filterUris = Selectives.SelectiveUris[accountInfo.AccountKey];\r
 365 \r
 366                         //May have to wait if the FileAgent has asked for a Pause, due to local changes\r
 367                         await _unPauseEvent.WaitAsync().ConfigureAwait(false);\r
 368 \r
 369                         //Get the local files here                        \r
 370                         var agent = AgentLocator<FileAgent>.Get(accountInfo.AccountPath);                                                \r
 371                         var files = LoadLocalFileTuples(accountInfo, accountBatch);\r
 372 \r
 373                         var states = FileState.Queryable.ToList();                        \r
 374                         \r
 375                         var infos = (from remote in currentRemotes\r
 376                                     let path = remote.RelativeUrlToFilePath(accountInfo.UserName)\r
 377                                     let info=agent.GetFileSystemInfo(path)\r
 378                                     select Tuple.Create(info.FullName,remote))\r
 379                                     .ToList();\r
 380 \r
 381                         var token = _currentOperationCancellation.Token;\r
 382 \r
 383                         var tuples = MergeSources(infos, files, states).ToList();\r
 384 \r
 385                         //Process only the changes in the batch file, if one exists\r
 386                         var stateTuples = accountBatch==null?tuples:tuples.Where(t => accountBatch.Contains(t.FilePath));\r
 387                         foreach (var tuple in stateTuples.Where(s=>!s.Locked))\r
 388                         {\r
 389                             await _unPauseEvent.WaitAsync().ConfigureAwait(false);\r
 390 \r
 391                             //Set the Merkle Hash\r
 392                             //SetMerkleHash(accountInfo, tuple);\r
 393 \r
 394                             await SyncSingleItem(accountInfo, tuple, agent, token).ConfigureAwait(false);\r
 395 \r
 396                         }\r
 397 \r
 398 \r
 399                         //On the first run\r
 400 /*\r
 401                         if (_firstPoll)\r
 402                         {\r
 403                             MarkSuspectedDeletes(accountInfo, cleanRemotes);\r
 404                         }\r
 405 */\r
 406 \r
 407 \r
 408                         Log.Info("[LISTENER] End Processing");\r
 409                     }\r
 410                 }\r
 411                 catch (Exception ex)\r
 412                 {\r
 413                     Log.ErrorFormat("[FAIL] ListObjects for {0} in ProcessRemoteFiles with {1}", accountInfo.UserName, ex);\r
 414                     return nextSince;\r
 415                 }\r
 416 \r
 417                 Log.Info("[LISTENER] Finished");\r
 418                 return nextSince;\r
 419             }\r
 420         }\r
 421 /*\r
 422 \r
 423         private static void SetMerkleHash(AccountInfo accountInfo, StateTuple tuple)\r
 424         {\r
 425             //The Merkle hash for directories is that of an empty buffer\r
 426             if (tuple.FileInfo is DirectoryInfo)\r
 427                 tuple.C = MERKLE_EMPTY;\r
 428             else if (tuple.FileState != null && tuple.MD5 == tuple.FileState.ETag)\r
 429             {\r
 430                 //If there is a state whose MD5 matches, load the merkle hash from the file state\r
 431                 //insteaf of calculating it\r
 432                 tuple.C = tuple.FileState.Checksum;                              \r
 433             }\r
 434             else\r
 435             {\r
 436                 tuple.Merkle = Signature.CalculateTreeHashAsync((FileInfo)tuple.FileInfo, accountInfo.BlockSize, accountInfo.BlockHash,1,progress);\r
 437                 //tuple.C=tuple.Merkle.TopHash.ToHashString();                \r
 438             }\r
 439         }\r
 440 */\r
 441 \r
 442         private IEnumerable<FileSystemInfo> LoadLocalFileTuples(AccountInfo accountInfo,IEnumerable<string> batch )\r
 443         {\r
 444             using (ThreadContext.Stacks["Account Files Hashing"].Push(accountInfo.UserName))\r
 445             {\r
 446                 var batchPaths = (batch==null)?new List<string>():batch.ToList();\r
 447                 IEnumerable<FileSystemInfo> localInfos=AgentLocator<FileAgent>.Get(accountInfo.AccountPath)\r
 448                                                         .EnumerateFileSystemInfos();\r
 449                 if (batchPaths.Count>0)\r
 450                     localInfos= localInfos.Where(fi => batchPaths.Contains(fi.FullName));\r
 451 \r
 452                 return localInfos;\r
 453             }\r
 454         }\r
 455 \r
 456         /// <summary>\r
 457         /// Wait and Pause the agent while waiting\r
 458         /// </summary>\r
 459         /// <param name="backoff"></param>\r
 460         /// <returns></returns>\r
 461         private async Task PauseFor(int backoff)\r
 462         {\r
 463 \r
 464             Pause = true;\r
 465             await TaskEx.Delay(backoff).ConfigureAwait(false);\r
 466             Pause = false;\r
 467         }\r
 468 \r
 469         private async Task SyncSingleItem(AccountInfo accountInfo, StateTuple tuple, FileAgent agent, CancellationToken token)\r
 470         {\r
 471             Log.DebugFormat("Sync [{0}] C:[{1}] L:[{2}] S:[{3}]", tuple.FilePath, tuple.C, tuple.L, tuple.S);\r
 472 \r
 473             try\r
 474             {\r
 475 \r
 476                 var localFilePath = tuple.FilePath;\r
 477                 //Don't use the tuple info, it may have been deleted\r
 478                 var localInfo = FileInfoExtensions.FromPath(localFilePath);\r
 479 \r
 480 \r
 481                 var isUnselectedRootFolder = agent.IsUnselectedRootFolder(tuple.FilePath);\r
 482 \r
 483                 //Unselected root folders that have not yet been uploaded should be uploaded and added to the \r
 484                 //selective folders\r
 485 \r
 486                 if (!Selectives.IsSelected(accountInfo, localFilePath) &&\r
 487                     !(isUnselectedRootFolder && tuple.ObjectInfo == null))\r
 488                     return;\r
 489 \r
 490                 // Local file unchanged? If both C and L are null, make sure it's because \r
 491                 //both the file is missing and the state checksum is not missing\r
 492                 if (tuple.C == tuple.L /*&& (localInfo.Exists || tuple.FileState == null)*/)\r
 493                 {\r
 494                     //No local changes\r
 495                     //Server unchanged?\r
 496                     if (tuple.S == tuple.L)\r
 497                     {\r
 498                         // No server changes\r
 499                         //Has the file been renamed on the server?\r
 500                         MoveForServerMove(accountInfo, tuple);\r
 501                     }\r
 502                     else\r
 503                     {\r
 504                         //Different from server\r
 505                         //Does the server file exist?\r
 506                         if (tuple.S == null)\r
 507                         {\r
 508                             //Server file doesn't exist\r
 509                             //deleteObjectFromLocal()\r
 510                             using (\r
 511                                 StatusNotification.GetNotifier("Deleting local {0}", "Deleted local {0}",\r
 512                                                                Path.GetFileName(localFilePath)))\r
 513                             {\r
 514                                 StatusKeeper.SetFileState(localFilePath, FileStatus.Deleted,\r
 515                                                           FileOverlayStatus.Deleted, "");\r
 516                                 using (NetworkGate.Acquire(localFilePath, NetworkOperation.Deleting))\r
 517                                 {\r
 518                                     agent.Delete(localFilePath);\r
 519                                 }\r
 520                                 //updateRecord(Remove C, L)\r
 521                                 StatusKeeper.ClearFileStatus(localFilePath);\r
 522                             }\r
 523                         }\r
 524                         else\r
 525                         {\r
 526                             //Server file exists\r
 527                             //downloadServerObject() // Result: L = S\r
 528                             //If the file has moved on the server, move it locally before downloading\r
 529                             using (\r
 530                                 StatusNotification.GetNotifier("Downloading {0}", "Downloaded {0}",\r
 531                                                                Path.GetFileName(localFilePath)))\r
 532                             {\r
 533                                 var targetPath = MoveForServerMove(accountInfo, tuple);\r
 534 \r
 535                                 StatusKeeper.SetFileState(targetPath, FileStatus.Modified, FileOverlayStatus.Modified,\r
 536                                                           "");\r
 537 \r
 538                                 await\r
 539                                     NetworkAgent.Downloader.DownloadCloudFile(accountInfo, tuple.ObjectInfo, targetPath,\r
 540                                                                               token)\r
 541                                         .ConfigureAwait(false);\r
 542                                 //updateRecord( L = S )\r
 543                                 StatusKeeper.UpdateFileChecksum(targetPath, tuple.ObjectInfo.ETag,\r
 544                                                                 tuple.ObjectInfo.X_Object_Hash);\r
 545 \r
 546                                 StatusKeeper.StoreInfo(targetPath, tuple.ObjectInfo);\r
 547                             }\r
 548 \r
 549                             /*\r
 550                                                             StatusKeeper.SetFileState(targetPath, FileStatus.Unchanged,\r
 551                                                                                       FileOverlayStatus.Normal, "");\r
 552                                 */\r
 553                         }\r
 554                     }\r
 555                 }\r
 556                 else\r
 557                 {\r
 558                     //Local changes found\r
 559 \r
 560                     //Server unchanged?\r
 561                     if (tuple.S == tuple.L)\r
 562                     {\r
 563                         //The FileAgent selective sync checks for new root folder files\r
 564                         if (!agent.Ignore(localFilePath))\r
 565                         {\r
 566                             if ((tuple.C == null || !localInfo.Exists) && tuple.ObjectInfo != null)\r
 567                             {\r
 568                                 //deleteObjectFromServer()\r
 569                                 DeleteCloudFile(accountInfo, tuple);\r
 570                                 //updateRecord( Remove L, S)                  \r
 571                             }\r
 572                             else\r
 573                             {\r
 574                                 //uploadLocalObject() // Result: S = C, L = S                        \r
 575                                 var progress = new Progress<double>(d =>\r
 576                                     StatusNotification.Notify(new StatusNotification(String.Format("Merkle Hashing for Upload {0:p} of {1}", d, localInfo.Name))));\r
 577 \r
 578                                 //Debug.Assert(tuple.FileState !=null);\r
 579                                 var action = new CloudUploadAction(accountInfo, localInfo, tuple.FileState,\r
 580                                                                    accountInfo.BlockSize, accountInfo.BlockHash,\r
 581                                                                    "Poll", isUnselectedRootFolder,progress);\r
 582                                 using (\r
 583                                     StatusNotification.GetNotifier("Uploading {0}", "Uploaded {0}",\r
 584                                                                    Path.GetFileName(localFilePath)))\r
 585                                 {\r
 586                                     await NetworkAgent.Uploader.UploadCloudFile(action, token).ConfigureAwait(false);\r
 587                                 }\r
 588 \r
 589                                 //updateRecord( S = C )\r
 590                                 //State updated by the uploader\r
 591 \r
 592                                 if (isUnselectedRootFolder)\r
 593                                 {\r
 594                                     ProcessChildren(accountInfo, tuple, agent, token);\r
 595                                 }\r
 596                             }\r
 597                         }\r
 598                     }\r
 599                     else\r
 600                     {\r
 601                         if (tuple.C == tuple.S)\r
 602                         {\r
 603                             // (Identical Changes) Result: L = S\r
 604                             //doNothing()\r
 605                             \r
 606                             //Don't update anything for nonexistend server files\r
 607                             if (tuple.S != null)\r
 608                             {\r
 609                                 //Detect server moves\r
 610                                 var targetPath = MoveForServerMove(accountInfo, tuple);\r
 611                                 StatusKeeper.StoreInfo(targetPath, tuple.ObjectInfo);\r
 612                             }\r
 613                             else\r
 614                             {\r
 615                                 //At this point, C==S==NULL and we have a stale state (L)\r
 616                                 //Log the stale tuple for investigation\r
 617                                 Log.WarnFormat("Stale tuple detected FilePathPath:[{0}], State:[{1}], LocalFile:[{2}]", tuple.FilePath, tuple.FileState, tuple.FileInfo);\r
 618 \r
 619                                 //And remove it\r
 620                                 if (!String.IsNullOrWhiteSpace(tuple.FilePath))\r
 621                                     StatusKeeper.ClearFileStatus(tuple.FilePath);\r
 622                             }\r
 623                         }\r
 624                         else\r
 625                         {\r
 626                             if ((tuple.C == null || !localInfo.Exists) && tuple.ObjectInfo != null)\r
 627                             {\r
 628                                 //deleteObjectFromServer()\r
 629                                 DeleteCloudFile(accountInfo, tuple);\r
 630                                 //updateRecord(Remove L, S)                  \r
 631                             }\r
 632                                 //If both the local and server files are missing, the state is stale\r
 633                             else if (!localInfo.Exists && (tuple.S == null || tuple.ObjectInfo == null))\r
 634                             {\r
 635                                 StatusKeeper.ClearFileStatus(localInfo.FullName);\r
 636                             }\r
 637                             else\r
 638                             {\r
 639                                 ReportConflictForMismatch(localFilePath);\r
 640                                 //identifyAsConflict() // Manual action required\r
 641                             }\r
 642                         }\r
 643                     }\r
 644                 }\r
 645             }\r
 646             catch (Exception exc)\r
 647             {\r
 648                 //In case of error log and retry with the next poll\r
 649                 Log.ErrorFormat("[SYNC] Failed for file {0}. Will Retry.\r\n{1}",tuple.FilePath,exc);\r
 650 \r
 651                 \r
 652             }\r
 653         }\r
 654 \r
 655         private string MoveForServerMove(AccountInfo accountInfo, StateTuple tuple)\r
 656         {\r
 657             if (tuple.ObjectInfo == null)\r
 658                 return null;\r
 659             var relativePath = tuple.ObjectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
 660             var serverPath = Path.Combine(accountInfo.AccountPath, relativePath);\r
 661             \r
 662             //Compare Case Insensitive\r
 663             if (String.Equals(tuple.FilePath ,serverPath,StringComparison.InvariantCultureIgnoreCase)) \r
 664                 return serverPath;\r
 665 \r
 666             if (tuple.FileInfo.Exists)\r
 667             {\r
 668                 using (StatusNotification.GetNotifier("Moving local {0}", "Moved local {0}", Path.GetFileName(tuple.FilePath)))\r
 669                 using(NetworkGate.Acquire(tuple.FilePath,NetworkOperation.Renaming))\r
 670                 {\r
 671                     var fi = tuple.FileInfo as FileInfo;\r
 672                     if (fi != null)\r
 673                         fi.MoveTo(serverPath);\r
 674                     var di = tuple.FileInfo as DirectoryInfo;\r
 675                     if (di != null)\r
 676                         di.MoveTo(serverPath);\r
 677                 }\r
 678                 StatusKeeper.StoreInfo(serverPath, tuple.ObjectInfo);\r
 679             }\r
 680             else\r
 681             {\r
 682                 Debug.Assert(false, "File does not exist");\r
 683             }\r
 684             return serverPath;\r
 685         }\r
 686 \r
 687         private void DeleteCloudFile(AccountInfo accountInfo, StateTuple tuple)\r
 688         {\r
 689             using (StatusNotification.GetNotifier("Deleting server {0}", "Deleted server {0}", Path.GetFileName(tuple.FilePath)))\r
 690             {\r
 691 \r
 692                 StatusKeeper.SetFileState(tuple.FilePath, FileStatus.Deleted,\r
 693                                           FileOverlayStatus.Deleted, "");\r
 694                 NetworkAgent.DeleteAgent.DeleteCloudFile(accountInfo, tuple.ObjectInfo);\r
 695                 StatusKeeper.ClearFileStatus(tuple.FilePath);\r
 696             }\r
 697         }\r
 698 \r
 699         private void ProcessChildren(AccountInfo accountInfo, StateTuple tuple, FileAgent agent, CancellationToken token)\r
 700         {\r
 701 \r
 702             var dirInfo = tuple.FileInfo as DirectoryInfo;\r
 703             var folderTuples = from folder in dirInfo.EnumerateDirectories("*", SearchOption.AllDirectories)\r
 704                                select new StateTuple(folder);\r
 705             var fileTuples = from file in dirInfo.EnumerateFiles("*", SearchOption.AllDirectories)\r
 706                              select new StateTuple(file);\r
 707             \r
 708             //Process folders first, to ensure folders appear on the sever as soon as possible\r
 709             folderTuples.ApplyAction(async t =>await SyncSingleItem(accountInfo, t, agent, token).ConfigureAwait(false));\r
 710             \r
 711             fileTuples.ApplyAction(async t => await SyncSingleItem(accountInfo, t, agent, token).ConfigureAwait(false));\r
 712         }\r
 713 \r
 714 \r
 715         /*\r
 716          *                 //Use the queue to retry locked file hashing\r
 717                 var fileQueue = new ConcurrentQueue<FileSystemInfo>(localInfos);\r
 718                 \r
 719 \r
 720                 var results = new List<Tuple<FileSystemInfo, string>>();\r
 721                 var backoff = 0;\r
 722                 while (fileQueue.Count > 0)\r
 723                 {\r
 724                     FileSystemInfo file;\r
 725                     fileQueue.TryDequeue(out file);\r
 726                     using (ThreadContext.Stacks["File"].Push(file.FullName))\r
 727                     {\r
 728                         try\r
 729                         {\r
 730                             //Replace MD5 here, do the calc while syncing individual files\r
 731                             string hash ;\r
 732                             if (file is DirectoryInfo)\r
 733                                 hash = MD5_EMPTY;\r
 734                             else\r
 735                             {\r
 736                                 //Wait in case the FileAgent has requested a Pause\r
 737                                 await _unPauseEvent.WaitAsync().ConfigureAwait(false);\r
 738                                 \r
 739                                 using (StatusNotification.GetNotifier("Hashing {0}", "", file.Name))\r
 740                                 {\r
 741                                     hash = ((FileInfo)file).ComputeShortHash(StatusNotification);\r
 742                                     backoff = 0;\r
 743                                 }\r
 744                             }                            \r
 745                             results.Add(Tuple.Create(file, hash));\r
 746                         }\r
 747                         catch (IOException exc)\r
 748                         {\r
 749                             Log.WarnFormat("[HASH] File in use, will retry [{0}]", exc);\r
 750                             fileQueue.Enqueue(file);\r
 751                             //If this is the only enqueued file                            \r
 752                             if (fileQueue.Count != 1) continue;\r
 753                             \r
 754                             \r
 755                             //Increase delay\r
 756                             if (backoff<60000)\r
 757                                 backoff += 10000;\r
 758                             //Pause Polling for the specified time\r
 759                         }\r
 760                         if (backoff>0)\r
 761                             await PauseFor(backoff).ConfigureAwait(false);\r
 762                     }\r
 763                 }\r
 764 \r
 765                 return results;\r
 766 \r
 767          */\r
 768         private IEnumerable<StateTuple> MergeSources(\r
 769             IEnumerable<Tuple<string, ObjectInfo>> infos, \r
 770             IEnumerable<FileSystemInfo> files, \r
 771             IEnumerable<FileState> states)\r
 772         {\r
 773             var tuplesByPath = files.ToDictionary(f => f.FullName, f => new StateTuple {FileInfo = f}); new Dictionary<string, StateTuple>();\r
 774 \r
 775             //For files that have state\r
 776             foreach (var state in states)\r
 777             {\r
 778                 StateTuple hashTuple;\r
 779                 if (tuplesByPath.TryGetValue(state.FilePath, out hashTuple))\r
 780                 {\r
 781                     hashTuple.FileState = state;\r
 782                     UpdateMD5(hashTuple);\r
 783                 }\r
 784                 else\r
 785                 {\r
 786                     var fsInfo = FileInfoExtensions.FromPath(state.FilePath);\r
 787                     hashTuple = new StateTuple {FileInfo = fsInfo, FileState = state};\r
 788                     tuplesByPath[state.FilePath] = hashTuple;\r
 789                 }\r
 790             }\r
 791             //for files that don't have state\r
 792             foreach (var tuple in tuplesByPath.Values.Where(t => t.FileState == null))\r
 793             {\r
 794                 UpdateMD5(tuple);\r
 795             }\r
 796 \r
 797             var tuplesByID = tuplesByPath.Values\r
 798                 .Where(tuple => tuple.FileState != null && tuple.FileState.ObjectID!=null)\r
 799                 .ToDictionary(tuple=>tuple.FileState.ObjectID,tuple=>tuple);//new Dictionary<Guid, StateTuple>();\r
 800 \r
 801             foreach (var info in infos)\r
 802             {\r
 803                 StateTuple hashTuple;\r
 804                 var filePath = info.Item1;\r
 805                 var objectInfo = info.Item2;\r
 806                 var objectID = objectInfo.UUID;\r
 807 \r
 808                 if (tuplesByID.TryGetValue(objectID, out hashTuple))\r
 809                 {\r
 810                     hashTuple.ObjectInfo = objectInfo;                    \r
 811                 }\r
 812                 else if (tuplesByPath.TryGetValue(filePath, out hashTuple))\r
 813                 {\r
 814                     hashTuple.ObjectInfo = objectInfo;\r
 815                 }\r
 816                 else\r
 817                 {\r
 818                     var fsInfo = FileInfoExtensions.FromPath(filePath);\r
 819                     hashTuple= new StateTuple {FileInfo = fsInfo, ObjectInfo = objectInfo};\r
 820                     tuplesByPath[filePath] = hashTuple;\r
 821                     tuplesByID[objectInfo.UUID] = hashTuple;\r
 822                 }\r
 823             }\r
 824             Debug.Assert(tuplesByPath.Values.All(t => t.HashesValid()));\r
 825             return tuplesByPath.Values;\r
 826         }\r
 827 \r
 828         private void  UpdateMD5(StateTuple hashTuple)\r
 829         {\r
 830             \r
 831             try\r
 832             {\r
 833                 var hash = Signature.MD5_EMPTY;\r
 834                 if (hashTuple.FileInfo is FileInfo)\r
 835                 {\r
 836                     var file = hashTuple.FileInfo as FileInfo;\r
 837                     var stateDate = hashTuple.NullSafe(h => h.FileState).NullSafe(s => s.LastWriteDate) ??\r
 838                                     DateTime.MinValue;\r
 839                     if (file.LastWriteTime - stateDate < TimeSpan.FromSeconds(1) &&\r
 840                         hashTuple.FileState.LastLength == file.Length)\r
 841                     {\r
 842                         hash = hashTuple.FileState.LastMD5;\r
 843                     }\r
 844                     else\r
 845                     {\r
 846                         //Modified, must calculate hash\r
 847                         hash = file.ComputeShortHash(StatusNotification);\r
 848                         StatusKeeper.UpdateLastMD5(file, hash);\r
 849                     }\r
 850                 }\r
 851                 hashTuple.C = hash;\r
 852                 hashTuple.MD5 = hash;\r
 853             }\r
 854             catch (IOException)\r
 855             {\r
 856                 hashTuple.Locked = true;\r
 857             }            \r
 858         }\r
 859 \r
 860         /// <summary>\r
 861         /// Returns the latest LastModified date from the list of objects, but only if it is before\r
 862         /// than the threshold value\r
 863         /// </summary>\r
 864         /// <param name="threshold"></param>\r
 865         /// <param name="cloudObjects"></param>\r
 866         /// <returns></returns>\r
 867         private static DateTime? GetLatestDateBefore(DateTime? threshold, IList<ObjectInfo> cloudObjects)\r
 868         {\r
 869             DateTime? maxDate = null;\r
 870             if (cloudObjects!=null &&  cloudObjects.Count > 0)\r
 871                 maxDate = cloudObjects.Max(obj => obj.Last_Modified);\r
 872             if (maxDate == null || maxDate == DateTime.MinValue)\r
 873                 return threshold;\r
 874             if (threshold == null || threshold == DateTime.MinValue || threshold > maxDate)\r
 875                 return maxDate;\r
 876             return threshold;\r
 877         }\r
 878 \r
 879         /// <summary>\r
 880         /// Returns the latest LastModified date from the list of objects, but only if it is after\r
 881         /// the threshold value\r
 882         /// </summary>\r
 883         /// <param name="threshold"></param>\r
 884         /// <param name="cloudObjects"></param>\r
 885         /// <returns></returns>\r
 886         private static DateTime? GetLatestDateAfter(DateTime? threshold, IList<ObjectInfo> cloudObjects)\r
 887         {\r
 888             DateTime? maxDate = null;\r
 889             if (cloudObjects!=null &&  cloudObjects.Count > 0)\r
 890                 maxDate = cloudObjects.Max(obj => obj.Last_Modified);\r
 891             if (maxDate == null || maxDate == DateTime.MinValue)\r
 892                 return threshold;\r
 893             if (threshold == null || threshold == DateTime.MinValue || threshold < maxDate)\r
 894                 return maxDate;\r
 895             return threshold;\r
 896         }\r
 897 \r
 898         readonly AccountsDifferencer _differencer = new AccountsDifferencer();\r
 899         private bool _pause;\r
 900         \r
 901         const string MERKLE_EMPTY = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";\r
 902 \r
 903 \r
 904         private void ReportConflictForMismatch(string localFilePath)\r
 905         {\r
 906             if (String.IsNullOrWhiteSpace(localFilePath))\r
 907                 throw new ArgumentNullException("localFilePath");\r
 908             Contract.EndContractBlock();\r
 909 \r
 910             StatusKeeper.SetFileState(localFilePath, FileStatus.Conflict, FileOverlayStatus.Conflict, "File changed at the server");\r
 911             UpdateStatus(PithosStatus.HasConflicts);\r
 912             var message = String.Format("Conflict detected for file {0}", localFilePath);\r
 913             Log.Warn(message);\r
 914             StatusNotification.NotifyChange(message, TraceLevel.Warning);\r
 915         }\r
 916 \r
 917 \r
 918         /// <summary>\r
 919         /// Notify the UI to update the visual status\r
 920         /// </summary>\r
 921         /// <param name="status"></param>\r
 922         private void UpdateStatus(PithosStatus status)\r
 923         {\r
 924             try\r
 925             {\r
 926                 StatusNotification.SetPithosStatus(status);\r
 927                 //StatusNotification.Notify(new Notification());\r
 928             }\r
 929             catch (Exception exc)\r
 930             {\r
 931                 //Failure is not critical, just log it\r
 932                 Log.Warn("Error while updating status", exc);\r
 933             }\r
 934         }\r
 935 \r
 936         private static void CreateContainerFolders(AccountInfo accountInfo, IEnumerable<ContainerInfo> containers)\r
 937         {\r
 938             var containerPaths = from container in containers\r
 939                                  let containerPath = Path.Combine(accountInfo.AccountPath, container.Name)\r
 940                                  where container.Name != FolderConstants.TrashContainer && !Directory.Exists(containerPath)\r
 941                                  select containerPath;\r
 942 \r
 943             foreach (var path in containerPaths)\r
 944             {\r
 945                 Directory.CreateDirectory(path);\r
 946             }\r
 947         }\r
 948 \r
 949         public void AddAccount(AccountInfo accountInfo)\r
 950         {\r
 951             //Avoid adding a duplicate accountInfo\r
 952             _accounts.TryAdd(accountInfo.AccountKey, accountInfo);\r
 953         }\r
 954 \r
 955         public void RemoveAccount(AccountInfo accountInfo)\r
 956         {\r
 957             AccountInfo account;\r
 958             _accounts.TryRemove(accountInfo.AccountKey, out account);\r
 959 \r
 960             SnapshotDifferencer differencer;\r
 961             _differencer.Differencers.TryRemove(accountInfo.AccountKey, out differencer);\r
 962         }\r
 963        \r
 964     }\r
 965 }\r