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

   1 using System;\r
   2 using System.ComponentModel.Composition;\r
   3 using System.Diagnostics;\r
   4 using System.Diagnostics.Contracts;\r
   5 using System.IO;\r
   6 using System.Net;\r
   7 using System.Reflection;\r
   8 using System.Threading;\r
   9 using System.Threading.Tasks;\r
  10 using Pithos.Interfaces;\r
  11 using Pithos.Network;\r
  12 using log4net;\r
  13 \r
  14 namespace Pithos.Core.Agents\r
  15 {\r
  16     [Export(typeof(Uploader))]\r
  17     public class Uploader\r
  18     {\r
  19         private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);\r
  20 \r
  21         [Import]\r
  22         private IStatusKeeper StatusKeeper { get; set; }\r
  23 \r
  24         [Import]\r
  25         private IPithosSettings Settings { get; set; }\r
  26 \r
  27         public IStatusNotification StatusNotification { get; set; }\r
  28 \r
  29         \r
  30         //CancellationTokenSource _cts = new CancellationTokenSource();\r
  31         /*public void SignalStop()\r
  32         {\r
  33             _cts.Cancel();\r
  34         }*/\r
  35 \r
  36         public async Task UploadCloudFile(CloudUploadAction action,CancellationToken cancellationToken)\r
  37         {\r
  38             if (action == null)\r
  39                 throw new ArgumentNullException("action");\r
  40             Contract.EndContractBlock();\r
  41 \r
  42             using (ThreadContext.Stacks["Operation"].Push("UploadCloudFile"))\r
  43             {\r
  44                 try\r
  45                 {\r
  46                     await UnpauseEvent.WaitAsync().ConfigureAwait(false);\r
  47 \r
  48                     var fileInfo = action.LocalFile;\r
  49 \r
  50                     if (fileInfo.Extension.Equals("ignore", StringComparison.InvariantCultureIgnoreCase))\r
  51                         return;\r
  52 \r
  53                     if (!Selectives.IsSelected(action.AccountInfo, fileInfo) && !action.IsCreation)\r
  54                         return;\r
  55 \r
  56 \r
  57                     //Try to load the action's local state, if it is empty\r
  58                     if (action.FileState == null)\r
  59                         action.FileState = StatusKeeper.GetStateByFilePath(fileInfo.FullName);\r
  60 \r
  61                     TreeHash localTreeHash;\r
  62                     using (StatusNotification.GetNotifier("Merkle Hashing for Upload {0}", "Merkle Hashed for Upload {0}", fileInfo.Name))\r
  63                     {\r
  64                         //TODO: Load the stored treehash if appropriate\r
  65                         //TODO: WHO updates LastMD5?\r
  66 \r
  67                         var progress = new Progress<HashProgress>(d => StatusNotification.Notify(\r
  68                             new StatusNotification(String.Format("Merkle Hashing for Upload {0:p} of {1}", d.Percentage, fileInfo.Name))));\r
  69 \r
  70                         //If the action's Treehash is already calculated, use it instead of reprocessing\r
  71                         localTreeHash = action.TreeHash.IsValueCreated\r
  72                             ? action.TreeHash.Value \r
  73                             : StatusAgent.CalculateTreeHash(fileInfo, action.AccountInfo, action.FileState, Settings.HashingParallelism, cancellationToken, progress);\r
  74                     }\r
  75 \r
  76 \r
  77                     if (action.FileState != null)\r
  78                     {\r
  79                         /*\r
  80                                                 Log.WarnFormat("File [{0}] has no local state. It was probably created by a download action", fileInfo.FullName);\r
  81                                                 return;\r
  82                         */\r
  83 \r
  84 \r
  85                         var latestState = action.FileState;\r
  86 \r
  87                         //Do not upload files in conflict\r
  88                         if (latestState.FileStatus == FileStatus.Conflict)\r
  89                         {\r
  90                             Log.InfoFormat("Skipping file in conflict [{0}]", fileInfo.FullName);\r
  91                             return;\r
  92                         }\r
  93                         //Do not upload files when we have no permission\r
  94                         if (latestState.FileStatus == FileStatus.Forbidden)\r
  95                         {\r
  96                             Log.InfoFormat("Skipping forbidden file [{0}]", fileInfo.FullName);\r
  97                             return;\r
  98                         }\r
  99                     }\r
 100                     //Are we targeting our own account or a sharer account?\r
 101                     var relativePath = fileInfo.AsRelativeTo(action.AccountInfo.AccountPath);\r
 102                     var accountInfo = relativePath.StartsWith(FolderConstants.OthersFolder) \r
 103                                                   ? GetSharerAccount(relativePath, action.AccountInfo) \r
 104                                                   : action.AccountInfo;\r
 105 \r
 106 \r
 107 \r
 108                     var fullFileName = fileInfo.GetProperCapitalization();\r
 109                     using (var gate = NetworkGate.Acquire(fullFileName, NetworkOperation.Uploading))\r
 110                     {\r
 111                         //Abort if the file is already being uploaded or downloaded\r
 112                         if (gate.Failed)\r
 113                             return;\r
 114 \r
 115                         var cloudFile = action.CloudFile;\r
 116                         var account = cloudFile.Account ?? accountInfo.UserName;\r
 117                         try\r
 118                         {\r
 119 \r
 120                             var client = new CloudFilesClient(accountInfo);\r
 121 \r
 122                             //Even if GetObjectInfo times out, we can proceed with the upload            \r
 123                             var cloudInfo = await client.GetObjectInfo(account, cloudFile.Container, cloudFile.Name).ConfigureAwait(false);\r
 124 \r
 125                             //If this a shared file\r
 126                             if (!cloudFile.Account.Equals(action.AccountInfo.UserName,StringComparison.InvariantCultureIgnoreCase))\r
 127                             {\r
 128                                 \r
 129 /*\r
 130                                 if (!cloudInfo.IsWritable(action.AccountInfo.UserName))\r
 131                                 {\r
 132                                     MakeFileReadOnly(fullFileName);\r
 133                                     StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");\r
 134                                     return;\r
 135                                 }\r
 136 */\r
 137 \r
 138                                 //If this is a read-only file, do not upload changes\r
 139                                 if ( !cloudInfo.IsWritable(action.AccountInfo.UserName) ||\r
 140                                     //If the file is new, but we can't upload it\r
 141                                     (!cloudInfo.Exists && !client.CanUpload(account, cloudFile)) )\r
 142                                 {\r
 143                                     MakeFileReadOnly(fullFileName);\r
 144                                     StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");\r
 145                                     return;\r
 146                                 }\r
 147 \r
 148                             }\r
 149 \r
 150                             await UnpauseEvent.WaitAsync().ConfigureAwait(false);\r
 151 \r
 152                             fileInfo.Refresh();\r
 153                             //Does the file still exist or was it deleted/renamed?\r
 154                             if (fileInfo.Exists)\r
 155                             {\r
 156                                 if (fileInfo is DirectoryInfo)\r
 157                                 {\r
 158                                     //If the directory doesn't exist the Hash property will be empty\r
 159                                     if (String.IsNullOrWhiteSpace(cloudInfo.X_Object_Hash))\r
 160                                         //Go on and create the directory\r
 161                                         await\r
 162                                             client.PutObject(account, cloudFile.Container, cloudFile.Name, fullFileName,\r
 163                                                              Signature.MERKLE_EMPTY, ObjectInfo.CONTENT_TYPE_DIRECTORY);\r
 164                                     //If the upload is in response to a Folder create with Selective Sync enabled\r
 165                                     if (action.IsCreation)\r
 166                                     {\r
 167                                         //Add the folder to the Selected URls\r
 168                                         var selectiveUri = client.RootAddressUri.Combine(cloudFile.Uri);\r
 169                                         Selectives.AddUri(accountInfo, selectiveUri);\r
 170                                         Selectives.Save(accountInfo);\r
 171                                     }\r
 172                                 }\r
 173                                 else\r
 174                                 {\r
 175 \r
 176                                     var cloudHash = cloudInfo.X_Object_Hash.ToLower();\r
 177 \r
 178                                     string topHash;\r
 179                                     TreeHash treeHash;\r
 180                                     using (\r
 181                                         StatusNotification.GetNotifier("Hashing {0} for Upload", "Finished hashing {0}",\r
 182                                                                        fileInfo.Name))\r
 183                                     {\r
 184                                         treeHash = localTreeHash ?? action.TreeHash.Value;\r
 185                                         topHash = treeHash.TopHash.ToHashString();\r
 186                                     }\r
 187 \r
 188 \r
 189 \r
 190                                     //If the file hashes match, abort the upload\r
 191                                     if (cloudInfo != ObjectInfo.Empty && (topHash == cloudHash))\r
 192                                     {\r
 193                                         //but store any metadata changes \r
 194                                         StatusKeeper.StoreInfo(fullFileName, cloudInfo,treeHash);\r
 195                                         Log.InfoFormat("Skip upload of {0}, hashes match", fullFileName);\r
 196                                         return;\r
 197                                     }\r
 198 \r
 199 \r
 200                                     //Mark the file as modified while we upload it\r
 201                                     StatusKeeper.SetFileOverlayStatus(fullFileName, FileOverlayStatus.Modified);\r
 202                                     //And then upload it\r
 203 \r
 204                                     //Upload even small files using the Hashmap. The server may already contain\r
 205                                     //the relevant block                                \r
 206 \r
 207 \r
 208 \r
 209                                     await\r
 210                                         UploadWithHashMap(accountInfo, cloudFile, fileInfo as FileInfo, cloudFile.Name,\r
 211                                                           treeHash, cancellationToken).ConfigureAwait(false);\r
 212                                 }\r
 213 \r
 214                                 var currentInfo =await client.GetObjectInfo(cloudFile.Account, cloudFile.Container,\r
 215                                                                        cloudFile.Name).ConfigureAwait(false);\r
 216 \r
 217                                 StatusKeeper.StoreInfo(fullFileName, currentInfo, localTreeHash);\r
 218                                 //Ensure the status is cleared\r
 219                                 StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged,\r
 220                                                               FileOverlayStatus.Normal, "");\r
 221 /*\r
 222                                 //If there is no stored ObjectID in the file state, add it\r
 223                                 //TODO: Why not just update everything, then change the state?\r
 224                                 if (action.FileState == null || action.FileState.ObjectID == null)\r
 225                                 {\r
 226                                     \r
 227                                 }\r
 228                                 else\r
 229                                     //If everything succeeds, change the file and overlay status to normal\r
 230                                     StatusKeeper.SetFileState(fullFileName, FileStatus.Unchanged,\r
 231                                                               FileOverlayStatus.Normal, "");\r
 232 */\r
 233                             }\r
 234                             else\r
 235                             {\r
 236                                 StatusKeeper.ClearFileStatus(fullFileName);\r
 237                             }\r
 238                         }\r
 239                         catch (WebException exc)\r
 240                         {\r
 241                             var response = (exc.Response as HttpWebResponse);\r
 242                             if (response == null)\r
 243                                 throw;\r
 244                             if (response.StatusCode == HttpStatusCode.Forbidden)\r
 245                             {\r
 246                                 StatusKeeper.SetFileState(fileInfo.FullName, FileStatus.Forbidden, FileOverlayStatus.Conflict, "Forbidden");\r
 247                                 MakeFileReadOnly(fullFileName);\r
 248                             }\r
 249                             else\r
 250                                 //In any other case, propagate the error\r
 251                                 throw;\r
 252                         }\r
 253                     }\r
 254                     //Notify the Shell to update the overlays\r
 255                     NativeMethods.RaiseChangeNotification(fullFileName);\r
 256                     StatusNotification.NotifyChangedFile(fullFileName);\r
 257                 }\r
 258                 catch (AggregateException ex)\r
 259                 {\r
 260                     var exc = ex.InnerException as WebException;\r
 261                     if (exc == null)\r
 262                         throw ex.InnerException;\r
 263                     if (HandleUploadWebException(action, exc))\r
 264                         return;\r
 265                     throw;\r
 266                 }\r
 267                 catch (WebException ex)\r
 268                 {\r
 269                     if (HandleUploadWebException(action, ex))\r
 270                         return;\r
 271                     throw;\r
 272                 }\r
 273                 catch (Exception ex)\r
 274                 {\r
 275                     Log.Error("Unexpected error while uploading file", ex);\r
 276                     throw;\r
 277                 }\r
 278             }\r
 279         }\r
 280 \r
 281 \r
 282         private static void MakeFileReadOnly(string fullFileName)\r
 283         {\r
 284             var attributes = File.GetAttributes(fullFileName);\r
 285             //Do not make any modifications if not necessary\r
 286             if (attributes.HasFlag(FileAttributes.ReadOnly))\r
 287                 return;\r
 288             File.SetAttributes(fullFileName, attributes | FileAttributes.ReadOnly);            \r
 289         }\r
 290 \r
 291         private static AccountInfo GetSharerAccount(string relativePath, AccountInfo accountInfo)\r
 292         {\r
 293             var parts = relativePath.Split('\\');\r
 294             var accountName = parts[1];\r
 295             var oldName = accountInfo.UserName;\r
 296             var absoluteUri = accountInfo.StorageUri.AbsoluteUri;\r
 297             var nameIndex = absoluteUri.IndexOf(oldName, StringComparison.Ordinal);\r
 298             var root = absoluteUri.Substring(0, nameIndex);\r
 299 \r
 300             accountInfo = new AccountInfo\r
 301                               {\r
 302                                   UserName = accountName,\r
 303                                   AccountPath = Path.Combine(accountInfo.AccountPath, parts[0], parts[1]),\r
 304                                   StorageUri = new Uri(root + accountName),\r
 305                                   BlockHash = accountInfo.BlockHash,\r
 306                                   BlockSize = accountInfo.BlockSize,\r
 307                                   Token = accountInfo.Token\r
 308                               };\r
 309             return accountInfo;\r
 310         }\r
 311 \r
 312 \r
 313         public async Task UploadWithHashMap(AccountInfo accountInfo, ObjectInfo cloudFile, FileInfo fileInfo, Uri uri, TreeHash treeHash, CancellationToken token)\r
 314         {\r
 315             if (accountInfo == null)\r
 316                 throw new ArgumentNullException("accountInfo");\r
 317             if (cloudFile == null)\r
 318                 throw new ArgumentNullException("cloudFile");\r
 319             if (fileInfo == null)\r
 320                 throw new ArgumentNullException("fileInfo");\r
 321             if (uri==null)\r
 322                 throw new ArgumentNullException("uri");\r
 323             if (treeHash == null)\r
 324                 throw new ArgumentNullException("treeHash");\r
 325             if (cloudFile.Container==null)\r
 326                 throw new ArgumentException("Invalid container", "cloudFile");\r
 327             if (cloudFile.Container.IsAbsoluteUri)\r
 328                 throw new ArgumentException("Container URI must be relative", "cloudFile");\r
 329             Contract.EndContractBlock();\r
 330 \r
 331 \r
 332             if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))\r
 333                 return;\r
 334 \r
 335             var fullFileName = fileInfo.GetProperCapitalization();\r
 336 \r
 337             var account = cloudFile.Account ?? accountInfo.UserName;\r
 338             var container = cloudFile.Container;\r
 339 \r
 340             int block = 0;\r
 341 \r
 342             var client = new CloudFilesClient(accountInfo);\r
 343             //Send the hashmap to the server            \r
 344             var missingHashes = await client.PutHashMap(account, container, uri, treeHash).ConfigureAwait(false);\r
 345             ReportUploadProgress(fileInfo.Name, block, 0, missingHashes.Count, fileInfo.Length);\r
 346             //If the server returns no missing hashes, we are done\r
 347 \r
 348             client.UploadProgressChanged += (sender, args) =>\r
 349                                             ReportUploadProgress(fileInfo.Name, block, args.ProgressPercentage,\r
 350                                                                  missingHashes.Count, fileInfo.Length);\r
 351 \r
 352 \r
 353             while (missingHashes.Count > 0)\r
 354             {\r
 355                 block = 0;\r
 356 \r
 357                 if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))\r
 358                     return;\r
 359 \r
 360 \r
 361                 var buffer = new byte[accountInfo.BlockSize];\r
 362                 foreach (var missingHash in missingHashes)\r
 363                 {\r
 364                     if (await WaitOrAbort(accountInfo, cloudFile, token).ConfigureAwait(false))\r
 365                         return;\r
 366 \r
 367 \r
 368                     //Find the proper block\r
 369                     long blockIndex = treeHash.HashDictionary[missingHash];\r
 370                     long offset = blockIndex*accountInfo.BlockSize;\r
 371                     Debug.Assert(offset >= 0,\r
 372                                  String.Format("Negative Offset! BlockIndex {0} BlockSize {1}", blockIndex,\r
 373                                                accountInfo.BlockSize));\r
 374 \r
 375                     var read = fileInfo.Read(buffer, offset, accountInfo.BlockSize);\r
 376 \r
 377                     try\r
 378                     {\r
 379                         //And upload the block                \r
 380                         await client.PostBlock(account, container, buffer, 0, read,missingHash, token).ConfigureAwait(false);\r
 381                         token.ThrowIfCancellationRequested();\r
 382                         Log.InfoFormat("[BLOCK] Block {0} of {1} uploaded", blockIndex, fullFileName);\r
 383                     }\r
 384                     catch (TaskCanceledException)\r
 385                     {\r
 386                         throw new OperationCanceledException(token);\r
 387                     }\r
 388                     catch (Exception exc)\r
 389                     {\r
 390                         Log.Error(String.Format("Uploading block {0} of {1}", blockIndex, fullFileName), exc);\r
 391                     }\r
 392                     ReportUploadProgress(fileInfo.Name, block++, 100, missingHashes.Count, fileInfo.Length);\r
 393                 }\r
 394 \r
 395                 token.ThrowIfCancellationRequested();\r
 396                 //Repeat until there are no more missing hashes                \r
 397                 missingHashes = await client.PutHashMap(account, container, uri, treeHash).ConfigureAwait(false);\r
 398             }\r
 399 \r
 400             ReportUploadProgress(fileInfo.Name, missingHashes.Count, 0, missingHashes.Count, fileInfo.Length);\r
 401 \r
 402         }\r
 403 \r
 404         private async Task<bool> WaitOrAbort(AccountInfo account,ObjectInfo cloudFile, CancellationToken token)\r
 405         {\r
 406             token.ThrowIfCancellationRequested();\r
 407             await UnpauseEvent.WaitAsync().ConfigureAwait(false);\r
 408             var shouldAbort = !Selectives.IsSelected(account,cloudFile);\r
 409             if (shouldAbort)\r
 410                 Log.InfoFormat("Aborting [{0}]",cloudFile.Uri);\r
 411             return shouldAbort;\r
 412         }\r
 413 \r
 414         private void ReportUploadProgress(string fileName, int block, int blockPercentage, int totalBlocks, long fileSize)\r
 415         {\r
 416             StatusNotification.Notify(totalBlocks == 0\r
 417                                           ? new ProgressNotification(fileName, "Uploading", 1,blockPercentage, 1, fileSize)\r
 418                                           : new ProgressNotification(fileName, "Uploading", block, blockPercentage, totalBlocks, fileSize));\r
 419         }\r
 420 \r
 421 \r
 422         private bool HandleUploadWebException(CloudAction action, WebException exc)\r
 423         {\r
 424             var response = exc.Response as HttpWebResponse;\r
 425             if (response == null)\r
 426                 throw exc;\r
 427             if (response.StatusCode == HttpStatusCode.Unauthorized)\r
 428             {\r
 429                 Log.Error("Not allowed to upload file", exc);\r
 430                 var message = String.Format("Not allowed to uplad file {0}", action.LocalFile.FullName);\r
 431                 StatusKeeper.SetFileState(action.LocalFile.FullName, FileStatus.Unchanged, FileOverlayStatus.Normal, "");\r
 432                 StatusNotification.NotifyChange(message, TraceLevel.Warning);\r
 433                 return true;\r
 434             }\r
 435             return false;\r
 436         }\r
 437 \r
 438         [Import]\r
 439         public Selectives Selectives { get; set; }\r
 440 \r
 441         public AsyncManualResetEvent UnpauseEvent { get; set; }\r
 442     }\r
 443 }\r