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.Threading;\r
  49 using System.Threading.Tasks;\r
  50 using System.Threading.Tasks.Dataflow;\r
  51 using Castle.ActiveRecord;\r
  52 using Pithos.Interfaces;\r
  53 using Pithos.Network;\r
  54 using log4net;\r
  55 \r
  56 namespace Pithos.Core.Agents\r
  57 {\r
  58     using System;\r
  59     using System.Collections.Generic;\r
  60     using System.Linq;\r
  61     using System.Text;\r
  62 \r
  63     /// <summary>\r
  64     /// PollAgent periodically polls the server to detect object changes. The agent retrieves a listing of all\r
  65     /// objects and compares it with a previously cached version to detect differences. \r
  66     /// New files are downloaded, missing files are deleted from the local file system and common files are compared\r
  67     /// to determine the appropriate action\r
  68     /// </summary>\r
  69     [Export]\r
  70     public class PollAgent\r
  71     {\r
  72         private static readonly ILog Log = LogManager.GetLogger("PollAgent");\r
  73 \r
  74         [System.ComponentModel.Composition.Import]\r
  75         public IStatusKeeper StatusKeeper { get; set; }\r
  76 \r
  77         [System.ComponentModel.Composition.Import]\r
  78         public IPithosSettings Settings { get; set; }\r
  79 \r
  80         [System.ComponentModel.Composition.Import]\r
  81         public NetworkAgent NetworkAgent { get; set; }\r
  82 \r
  83         public IStatusNotification StatusNotification { get; set; }\r
  84 \r
  85         private bool _firstPoll = true;\r
  86 \r
  87         //The Sync Event signals a manual synchronisation\r
  88         private readonly AsyncManualResetEvent _syncEvent = new AsyncManualResetEvent();\r
  89 \r
  90         private ConcurrentDictionary<string, DateTime> _lastSeen = new ConcurrentDictionary<string, DateTime>();\r
  91         private readonly ConcurrentBag<AccountInfo> _accounts = new ConcurrentBag<AccountInfo>();\r
  92 \r
  93 \r
  94         /// <summary>\r
  95         /// Start a manual synchronization\r
  96         /// </summary>\r
  97         public void SynchNow()\r
  98         {            \r
  99             _syncEvent.Set();\r
 100         }\r
 101 \r
 102         //Remote files are polled periodically. Any changes are processed\r
 103         public async Task PollRemoteFiles(DateTime? since = null)\r
 104         {\r
 105             Debug.Assert(Thread.CurrentThread.IsBackground, "Polling Ended up in the main thread!");\r
 106 \r
 107             UpdateStatus(PithosStatus.Syncing);\r
 108             StatusNotification.Notify(new PollNotification());\r
 109 \r
 110             using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push("All accounts"))\r
 111             {\r
 112                 //If this poll fails, we will retry with the same since value\r
 113                 var nextSince = since;\r
 114                 try\r
 115                 {\r
 116                     //Next time we will check for all changes since the current check minus 1 second\r
 117                     //This is done to ensure there are no discrepancies due to clock differences\r
 118                     var current = DateTime.Now.AddSeconds(-1);\r
 119 \r
 120                     var tasks = from accountInfo in _accounts\r
 121                                 select ProcessAccountFiles(accountInfo, since);\r
 122 \r
 123                     await TaskEx.WhenAll(tasks.ToList());\r
 124 \r
 125                     _firstPoll = false;\r
 126                     //Reschedule the poll with the current timestamp as a "since" value\r
 127                     nextSince = current;\r
 128                 }\r
 129                 catch (Exception ex)\r
 130                 {\r
 131                     Log.ErrorFormat("Error while processing accounts\r\n{0}", ex);\r
 132                     //In case of failure retry with the same "since" value\r
 133                 }\r
 134 \r
 135                 UpdateStatus(PithosStatus.InSynch);\r
 136                 //Wait for the polling interval to pass or the Sync event to be signalled\r
 137                 nextSince = await WaitForScheduledOrManualPoll(nextSince);\r
 138 \r
 139                 TaskEx.Run(()=>PollRemoteFiles(nextSince));\r
 140 \r
 141             }\r
 142         }\r
 143 \r
 144         /// <summary>\r
 145         /// Wait for the polling period to expire or a manual sync request\r
 146         /// </summary>\r
 147         /// <param name="since"></param>\r
 148         /// <returns></returns>\r
 149         private async Task<DateTime?> WaitForScheduledOrManualPoll(DateTime? since)\r
 150         {\r
 151             var sync = _syncEvent.WaitAsync();\r
 152             var wait = TaskEx.Delay(TimeSpan.FromSeconds(Settings.PollingInterval), NetworkAgent.CancellationToken);\r
 153             var signaledTask = await TaskEx.WhenAny(sync, wait);\r
 154 \r
 155             //Wait for network processing to finish before polling\r
 156             var pauseTask=NetworkAgent.ProceedEvent.WaitAsync();\r
 157             await TaskEx.WhenAll(signaledTask, pauseTask);\r
 158 \r
 159             //If polling is signalled by SynchNow, ignore the since tag\r
 160             if (sync.IsCompleted)\r
 161             {\r
 162                 //TODO: Must convert to AutoReset\r
 163                 _syncEvent.Reset();\r
 164                 return null;\r
 165             }\r
 166             return since;\r
 167         }\r
 168 \r
 169         public async Task ProcessAccountFiles(AccountInfo accountInfo, DateTime? since = null)\r
 170         {\r
 171             if (accountInfo == null)\r
 172                 throw new ArgumentNullException("accountInfo");\r
 173             if (String.IsNullOrWhiteSpace(accountInfo.AccountPath))\r
 174                 throw new ArgumentException("The AccountInfo.AccountPath is empty", "accountInfo");\r
 175             Contract.EndContractBlock();\r
 176 \r
 177 \r
 178             using (log4net.ThreadContext.Stacks["Retrieve Remote"].Push(accountInfo.UserName))\r
 179             {\r
 180                 await NetworkAgent.GetDeleteAwaiter();\r
 181 \r
 182                 Log.Info("Scheduled");\r
 183                 var client = new CloudFilesClient(accountInfo);\r
 184 \r
 185                 var containers = client.ListContainers(accountInfo.UserName);\r
 186 \r
 187 \r
 188                 CreateContainerFolders(accountInfo, containers);\r
 189 \r
 190                 try\r
 191                 {\r
 192                     await NetworkAgent.GetDeleteAwaiter();\r
 193                     //Get the poll time now. We may miss some deletions but it's better to keep a file that was deleted\r
 194                     //than delete a file that was created while we were executing the poll                    \r
 195                     var pollTime = DateTime.Now;\r
 196 \r
 197                     //Get the list of server objects changed since the last check\r
 198                     //The name of the container is passed as state in order to create a dictionary of tasks in a subsequent step\r
 199                     var listObjects = (from container in containers\r
 200                                        select Task<IList<ObjectInfo>>.Factory.StartNew(_ =>\r
 201                                              client.ListObjects(accountInfo.UserName, container.Name, since), container.Name)).ToList();\r
 202 \r
 203                     var listShared = Task<IList<ObjectInfo>>.Factory.StartNew(_ => client.ListSharedObjects(since), "shared");\r
 204                     listObjects.Add(listShared);\r
 205                     var listTasks = await Task.Factory.WhenAll(listObjects.ToArray());\r
 206 \r
 207                     using (log4net.ThreadContext.Stacks["SCHEDULE"].Push("Process Results"))\r
 208                     {\r
 209                         var dict = listTasks.ToDictionary(t => t.AsyncState);\r
 210 \r
 211                         //Get all non-trash objects. Remember, the container name is stored in AsyncState\r
 212                         var remoteObjects = from objectList in listTasks\r
 213                                             where (string)objectList.AsyncState != "trash"\r
 214                                             from obj in objectList.Result\r
 215                                             select obj;\r
 216 \r
 217                         var trashObjects = dict["trash"].Result;\r
 218                         var sharedObjects = dict["shared"].Result;\r
 219 \r
 220                         //DON'T process trashed files\r
 221                         //If some files are deleted and added again to a folder, they will be deleted\r
 222                         //even though they are new.\r
 223                         //We would have to check file dates and hashes to ensure that a trashed file\r
 224                         //can be deleted safely from the local hard drive.\r
 225                         /*\r
 226                         //Items with the same name, hash may be both in the container and the trash\r
 227                         //Don't delete items that exist in the container\r
 228                         var realTrash = from trash in trashObjects\r
 229                                         where\r
 230                                             !remoteObjects.Any(\r
 231                                                 info => info.Name == trash.Name && info.Hash == trash.Hash)\r
 232                                         select trash;\r
 233                         ProcessTrashedFiles(accountInfo, realTrash);\r
 234 */\r
 235 \r
 236                         var cleanRemotes = (from info in remoteObjects.Union(sharedObjects)\r
 237                                             let name = info.Name\r
 238                                             where !name.EndsWith(".ignore", StringComparison.InvariantCultureIgnoreCase) &&\r
 239                                                   !name.StartsWith(FolderConstants.CacheFolder + "/",\r
 240                                                                    StringComparison.InvariantCultureIgnoreCase)\r
 241                                             select info).ToList();\r
 242 \r
 243                         var differencer = _differencer.PostSnapshot(accountInfo, cleanRemotes);\r
 244 \r
 245                         ProcessDeletedFiles(accountInfo, differencer.Deleted, pollTime);\r
 246 \r
 247                         //Create a list of actions from the remote files\r
 248                         var allActions = ChangesToActions(accountInfo, differencer.Changed)\r
 249                                         .Union(\r
 250                                         CreatesToActions(accountInfo, differencer.Created));\r
 251 \r
 252                         //And remove those that are already being processed by the agent\r
 253                         var distinctActions = allActions\r
 254                             .Except(NetworkAgent.GetEnumerable(), new PithosMonitor.LocalFileComparer())\r
 255                             .ToList();\r
 256 \r
 257                         //Queue all the actions\r
 258                         foreach (var message in distinctActions)\r
 259                         {\r
 260                             NetworkAgent.Post(message);\r
 261                         }\r
 262 \r
 263                         Log.Info("[LISTENER] End Processing");\r
 264                     }\r
 265                 }\r
 266                 catch (Exception ex)\r
 267                 {\r
 268                     Log.ErrorFormat("[FAIL] ListObjects for{0} in ProcessRemoteFiles with {1}", accountInfo.UserName, ex);\r
 269                     return;\r
 270                 }\r
 271 \r
 272                 Log.Info("[LISTENER] Finished");\r
 273 \r
 274             }\r
 275         }\r
 276 \r
 277         AccountsDifferencer _differencer = new AccountsDifferencer();\r
 278 \r
 279         /// <summary>\r
 280         /// Deletes local files that are not found in the list of cloud files\r
 281         /// </summary>\r
 282         /// <param name="accountInfo"></param>\r
 283         /// <param name="cloudFiles"></param>\r
 284         /// <param name="pollTime"></param>\r
 285         private void ProcessDeletedFiles(AccountInfo accountInfo, IEnumerable<ObjectInfo> cloudFiles, DateTime pollTime)\r
 286         {\r
 287             if (accountInfo == null)\r
 288                 throw new ArgumentNullException("accountInfo");\r
 289             if (String.IsNullOrWhiteSpace(accountInfo.AccountPath))\r
 290                 throw new ArgumentException("The AccountInfo.AccountPath is empty", "accountInfo");\r
 291             if (cloudFiles == null)\r
 292                 throw new ArgumentNullException("cloudFiles");\r
 293             Contract.EndContractBlock();\r
 294 \r
 295             //On the first run\r
 296             if (_firstPoll)\r
 297             {\r
 298                 //Only consider files that are not being modified, ie they are in the Unchanged state            \r
 299                 var deleteCandidates = FileState.Queryable.Where(state =>\r
 300                     state.FilePath.StartsWith(accountInfo.AccountPath)\r
 301                     && state.FileStatus == FileStatus.Unchanged).ToList();\r
 302 \r
 303 \r
 304                 //TODO: filesToDelete must take into account the Others container            \r
 305                 var filesToDelete = (from deleteCandidate in deleteCandidates\r
 306                                      let localFile = FileInfoExtensions.FromPath(deleteCandidate.FilePath)\r
 307                                      let relativeFilePath = localFile.AsRelativeTo(accountInfo.AccountPath)\r
 308                                      where\r
 309                                          !cloudFiles.Any(r => r.RelativeUrlToFilePath(accountInfo.UserName) == relativeFilePath)\r
 310                                      select localFile).ToList();\r
 311 \r
 312 \r
 313 \r
 314                 //Set the status of missing files to Conflict\r
 315                 foreach (var item in filesToDelete)\r
 316                 {\r
 317                     //Try to acquire a gate on the file, to take into account files that have been dequeued\r
 318                     //and are being processed\r
 319                     using (var gate = NetworkGate.Acquire(item.FullName, NetworkOperation.Deleting))\r
 320                     {\r
 321                         if (gate.Failed)\r
 322                             continue;\r
 323                         StatusKeeper.SetFileState(item.FullName, FileStatus.Conflict, FileOverlayStatus.Deleted);\r
 324                     }\r
 325                 }\r
 326                 UpdateStatus(PithosStatus.HasConflicts);\r
 327                 StatusNotification.NotifyConflicts(filesToDelete, String.Format("{0} local files are missing from Pithos, possibly because they were deleted", filesToDelete.Count));\r
 328                 StatusNotification.NotifyForFiles(filesToDelete, String.Format("{0} files were deleted", filesToDelete.Count), TraceLevel.Info);\r
 329             }\r
 330             else\r
 331             {\r
 332                 var deletedFiles = new List<FileSystemInfo>();\r
 333                 foreach (var objectInfo in cloudFiles)\r
 334                 {\r
 335                     var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
 336                     var item = GetFileAgent(accountInfo).GetFileSystemInfo(relativePath);\r
 337                     if (item.Exists)\r
 338                     {\r
 339                         if ((item.Attributes & FileAttributes.ReadOnly) == FileAttributes.ReadOnly)\r
 340                         {\r
 341                             item.Attributes = item.Attributes & ~FileAttributes.ReadOnly;\r
 342 \r
 343                         }\r
 344                         item.Delete();\r
 345                         DateTime lastDate;\r
 346                         _lastSeen.TryRemove(item.FullName, out lastDate);\r
 347                         deletedFiles.Add(item);\r
 348                     }\r
 349                     StatusKeeper.SetFileState(item.FullName, FileStatus.Deleted, FileOverlayStatus.Deleted);\r
 350                 }\r
 351                 StatusNotification.NotifyForFiles(deletedFiles, String.Format("{0} files were deleted", deletedFiles.Count), TraceLevel.Info);\r
 352             }\r
 353 \r
 354         }\r
 355 \r
 356         //Creates an appropriate action for each server file\r
 357         private IEnumerable<CloudAction> ChangesToActions(AccountInfo accountInfo, IEnumerable<ObjectInfo> changes)\r
 358         {\r
 359             if (changes == null)\r
 360                 throw new ArgumentNullException();\r
 361             Contract.EndContractBlock();\r
 362             var fileAgent = GetFileAgent(accountInfo);\r
 363 \r
 364             //In order to avoid multiple iterations over the files, we iterate only once\r
 365             //over the remote files\r
 366             foreach (var objectInfo in changes)\r
 367             {\r
 368                 var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
 369                 //and remove any matching objects from the list, adding them to the commonObjects list\r
 370                 if (fileAgent.Exists(relativePath))\r
 371                 {\r
 372                     //If a directory object already exists, we don't need to perform any other action                    \r
 373                     var localFile = fileAgent.GetFileSystemInfo(relativePath);\r
 374                     if (objectInfo.Content_Type == @"application/directory" && localFile is DirectoryInfo)\r
 375                         continue;\r
 376                     using (new SessionScope(FlushAction.Never))\r
 377                     {\r
 378                         var state = StatusKeeper.GetStateByFilePath(localFile.FullName);\r
 379                         _lastSeen[localFile.FullName] = DateTime.Now;\r
 380                         //Common files should be checked on a per-case basis to detect differences, which is newer\r
 381 \r
 382                         yield return new CloudAction(accountInfo, CloudActionType.MustSynch,\r
 383                                                      localFile, objectInfo, state, accountInfo.BlockSize,\r
 384                                                      accountInfo.BlockHash);\r
 385                     }\r
 386                 }\r
 387                 else\r
 388                 {\r
 389                     //Remote files should be downloaded\r
 390                     yield return new CloudDownloadAction(accountInfo, objectInfo);\r
 391                 }\r
 392             }\r
 393         }\r
 394 \r
 395         private IEnumerable<CloudAction> CreatesToActions(AccountInfo accountInfo, IEnumerable<ObjectInfo> creates)\r
 396         {\r
 397             if (creates == null)\r
 398                 throw new ArgumentNullException();\r
 399             Contract.EndContractBlock();\r
 400             var fileAgent = GetFileAgent(accountInfo);\r
 401 \r
 402             //In order to avoid multiple iterations over the files, we iterate only once\r
 403             //over the remote files\r
 404             foreach (var objectInfo in creates)\r
 405             {\r
 406                 var relativePath = objectInfo.RelativeUrlToFilePath(accountInfo.UserName);\r
 407                 //and remove any matching objects from the list, adding them to the commonObjects list\r
 408                 if (fileAgent.Exists(relativePath))\r
 409                 {\r
 410                     //If the object already exists, we probably have a conflict\r
 411                     //If a directory object already exists, we don't need to perform any other action                    \r
 412                     var localFile = fileAgent.GetFileSystemInfo(relativePath);\r
 413                     StatusKeeper.SetFileState(localFile.FullName, FileStatus.Conflict, FileOverlayStatus.Conflict);\r
 414                 }\r
 415                 else\r
 416                 {\r
 417                     //Remote files should be downloaded\r
 418                     yield return new CloudDownloadAction(accountInfo, objectInfo);\r
 419                 }\r
 420             }\r
 421         }\r
 422 \r
 423         private static FileAgent GetFileAgent(AccountInfo accountInfo)\r
 424         {\r
 425             return AgentLocator<FileAgent>.Get(accountInfo.AccountPath);\r
 426         }\r
 427 \r
 428         private void ProcessTrashedFiles(AccountInfo accountInfo, IEnumerable<ObjectInfo> trashObjects)\r
 429         {\r
 430             var fileAgent = GetFileAgent(accountInfo);\r
 431             foreach (var trashObject in trashObjects)\r
 432             {\r
 433                 var barePath = trashObject.RelativeUrlToFilePath(accountInfo.UserName);\r
 434                 //HACK: Assume only the "pithos" container is used. Must find out what happens when\r
 435                 //deleting a file from a different container\r
 436                 var relativePath = Path.Combine("pithos", barePath);\r
 437                 fileAgent.Delete(relativePath);\r
 438             }\r
 439         }\r
 440 \r
 441         private void UpdateStatus(PithosStatus status)\r
 442         {\r
 443             StatusKeeper.SetPithosStatus(status);\r
 444             StatusNotification.Notify(new Notification());\r
 445         }\r
 446 \r
 447         private static void CreateContainerFolders(AccountInfo accountInfo, IEnumerable<ContainerInfo> containers)\r
 448         {\r
 449             var containerPaths = from container in containers\r
 450                                  let containerPath = Path.Combine(accountInfo.AccountPath, container.Name)\r
 451                                  where container.Name != FolderConstants.TrashContainer && !Directory.Exists(containerPath)\r
 452                                  select containerPath;\r
 453 \r
 454             foreach (var path in containerPaths)\r
 455             {\r
 456                 Directory.CreateDirectory(path);\r
 457             }\r
 458         }\r
 459 \r
 460     }\r
 461 }\r