Synnefo » Pithos MS Client

#region

/* -----------------------------------------------------------------------

 * <copyright file="BlockUpdater.cs" company="GRNet">

 * 

 * Copyright 2011-2012 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 * </copyright>

 * -----------------------------------------------------------------------

 */

#endregion

using System;

using System.Collections.Concurrent;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Reflection;

using System.Threading;

using System.Threading.Tasks;

using OpenSSL.Crypto;

using Pithos.Network;

using Alpha = Alphaleonis.Win32.Filesystem;

namespace Pithos.Core.Agents

{

    class BlockUpdater

    {

        //TODO: Must clean orphaned blocks from the Cache folder.

        //

        //The Cache folder may have orphaned blocks. Blocks may be left in the Cache folder because:

        //1. A download was in progress when the application terminated. These blocks are needed to proceed 

        //  with partial download

        //2. The application terminated abnormally before the blocks were cleared after a download

        //3. The server file was deleted before the download completed.

        //

        //In #1, we need to keep the blocks. We need to detect the other cases and delete orphans

        //

        //Mitigations:

        // - Delete blocks with no corresponding state

        // - Check and delete possible orphans when a Deletion is detected

        // - Add Advanced command "Clear Cache"

        //

        //Need a better way to differentiate between cases #2, #3 and #1

        private static readonly log4net.ILog Log = log4net.LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);

        public string FilePath { get; private set; }

        public string RelativePath { get; private set; }

        public string CachePath { get; private set; }

        public TreeHash ServerHash { get; private set; }

        public string TempPath { get; private set; }

        public bool HasBlocks

        {

            get { return _blocks.Count>0; }            

        }

        readonly ConcurrentDictionary<long, string> _blocks = new ConcurrentDictionary<long, string>();

        readonly ConcurrentDictionary<string, string> _orphanBlocks = new ConcurrentDictionary<string, string>();

        [ContractInvariantMethod]

        private void Invariants()

        {

            Contract.Invariant(Path.IsPathRooted(CachePath));

            Contract.Invariant(Path.IsPathRooted(FilePath));

            Contract.Invariant(Path.IsPathRooted(TempPath));

            Contract.Invariant(!Path.IsPathRooted(RelativePath));

            Contract.Invariant(_blocks!=null);

            Contract.Invariant(_orphanBlocks!=null);

            Contract.Invariant(ServerHash!=null);

        }

        public BlockUpdater(string cachePath, string filePath, string relativePath,TreeHash serverHash)

        {   

            if (String.IsNullOrWhiteSpace(cachePath))

                throw new ArgumentNullException("cachePath");

            if (!Path.IsPathRooted(cachePath))

                throw new ArgumentException("The cachePath must be rooted", "cachePath");

            if (string.IsNullOrWhiteSpace(filePath))

                throw new ArgumentNullException("filePath");

            if (!Path.IsPathRooted(filePath))

                throw new ArgumentException("The filePath must be rooted", "filePath");

            if (string.IsNullOrWhiteSpace(relativePath))

                throw new ArgumentNullException("relativePath");

            if (Path.IsPathRooted(relativePath))

                throw new ArgumentException("The relativePath must NOT be rooted", "relativePath");

            if (serverHash == null)

                throw new ArgumentNullException("serverHash");

            Contract.EndContractBlock();

            CachePath=cachePath;

            FilePath = filePath;

            RelativePath=relativePath;

            ServerHash = serverHash;

            //The file will be stored in a temporary location while downloading with an extension .download

            TempPath = Path.Combine(CachePath, RelativePath + ".download");

            //Need to calculate the directory path because RelativePath may include folders

            var directoryPath = Path.GetDirectoryName(TempPath);            

            //directoryPath CAN be null if TempPath is a root path

            if (String.IsNullOrWhiteSpace(directoryPath))

                throw new ArgumentException("TempPath");

            //CachePath was absolute so directoryPath is absolute too

            Contract.Assume(Path.IsPathRooted(directoryPath));

            if (!Directory.Exists(directoryPath))

                Directory.CreateDirectory(directoryPath);

            LoadOrphans(directoryPath);

        }

        private void LoadOrphans(string directoryPath)

        {

            if (string.IsNullOrWhiteSpace(directoryPath))

                throw new ArgumentNullException("directoryPath");

            if (!Path.IsPathRooted(directoryPath))

                throw new ArgumentException("The directoryPath must be rooted", "directoryPath");

            if (ServerHash==null)

                throw new InvalidOperationException("ServerHash wasn't initialized");

            Contract.EndContractBlock();

            var fileNamename = Path.GetFileName(FilePath);

            var orphans = Directory.GetFiles(directoryPath, fileNamename + ".*");

            foreach (var orphan in orphans)

            {

                using (var hasher = new MessageDigestContext(MessageDigest.CreateByName(ServerHash.BlockHash)))                

                {

                    hasher.Init();

                    var buffer=File.ReadAllBytes(orphan);

                    //The server truncates nulls before calculating hashes, have to do the same

                    //Find the last non-null byte, starting from the end

                    var lastByteIndex = Array.FindLastIndex(buffer, buffer.Length-1, aByte => aByte != 0);

                    //lastByteIndex may be -1 if the file was empty. We don't want to use that block file

                    if (lastByteIndex >= 0)

                    {

                        byte[] block;

                        if (lastByteIndex == buffer.Length - 1)

                            block = buffer;

                        else

                        {

                            block=new byte[lastByteIndex+1];

                            Buffer.BlockCopy(buffer,0,block,0,lastByteIndex+1);

                        }

                        var binHash = hasher.Digest(block);

                        var hash = binHash.ToHashString();

                        _orphanBlocks[hash] = orphan;

                    }

                }

            }

        }

        public void Commit()

        {

            if (String.IsNullOrWhiteSpace(FilePath))

                throw new InvalidOperationException("FilePath is empty");

            if (String.IsNullOrWhiteSpace(TempPath))

                throw new InvalidOperationException("TempPath is empty");

            Contract.EndContractBlock();

            var info = Alpha.Volume.GetVolumeInformation(Alpha.Path.GetPathRoot(FilePath));            

            var isNTFS = info.FileSystemName.Equals("NTFS");                

            if (isNTFS && MS.WindowsAPICodePack.Internal.CoreHelpers.RunningOnVista)

            {

                ApplyBlocksWithTransaction();

            }

            else

            {

                ApplyBlocksWithCopy();

            }

            ClearBlocks();

        }

        private void ApplyBlocksWithTransaction()

        {

            var targetFile = FilePath;

            using (var tx = new Alpha.KernelTransaction())

            {

                //Retry if the file is still open

                    using (var stream = Alpha.File.Open(tx, targetFile, 

                        Alpha.FileMode.OpenOrCreate,Alpha.FileAccess.Write))

                    {

                        stream.SetLength(ServerHash.Bytes);

                        foreach (var block in _blocks)

                        {

                            var blockPath = block.Value;

                            var blockIndex = block.Key;

                            //Retry if we can't open the block immediatelly

                            for (int i = 0; i < 3; i++)

                            {

                                try

                                {

                                    using (var blockStream = File.OpenRead(blockPath))

                                    {

                                        long offset = blockIndex*ServerHash.BlockSize;

                                        stream.Seek(offset, SeekOrigin.Begin);

                                        blockStream.CopyTo(stream);

                                    }

                                    break;

                                }

                                catch (IOException)

                                {

                                    if (i>=2)

                                        throw;

                                    Thread.Sleep((i+1)*300);

                                }                                

                            }

                        }

                    }

                tx.Commit();

            }

        }

        private void ApplyBlocksWithCopy()

        {

            //Copy the file to a temporary location. Changes will be made to the

            //temporary file, then it will replace the original file

            if (File.Exists(FilePath))

                File.Copy(FilePath, TempPath, true);

            //Set the size of the file to the size specified in the treehash

            //This will also create an empty file if the file doesn't exist                        

            ApplyBlocks(TempPath);

            SwapFiles();

        }

        private void ApplyBlocks(string targetFile)

        {

            SetFileSize(targetFile, ServerHash.Bytes);

            //Update the temporary file with the data from the blocks

            using (var stream = File.OpenWrite(targetFile))

            {

                foreach (var block in _blocks)

                {

                    var blockPath = block.Value;

                    var blockIndex = block.Key;

                    using (var blockStream = File.OpenRead(blockPath))

                    {

                        long offset = blockIndex*ServerHash.BlockSize;

                        stream.Seek(offset, SeekOrigin.Begin);

                        blockStream.CopyTo(stream);

                    }

                }

            }

        }

        private void SwapFiles()

        {

            if (String.IsNullOrWhiteSpace(FilePath))

                throw new InvalidOperationException("FilePath is empty");

            if (String.IsNullOrWhiteSpace(TempPath))

                throw new InvalidOperationException("TempPath is empty");            

            Contract.EndContractBlock();

            if (File.Exists(FilePath))

                File.Replace(TempPath, FilePath, null, true);

            else

            {

                var targetDirectory = Path.GetDirectoryName(FilePath);

                if (!Directory.Exists(targetDirectory))

                    Directory.CreateDirectory(targetDirectory);

                File.Move(TempPath, FilePath);

            }

        }

        private void ClearBlocks()

        {

            if (Log.IsDebugEnabled)

                Log.DebugFormat("Clearing blocks for {0}",this.FilePath);

            //Get all the the block paths, orphan or not

            var paths= _blocks.Select(pair => pair.Value)

                          .Union(_orphanBlocks.Select(pair => pair.Value));

            foreach (var filePath in paths)

            {

                File.Delete(filePath);

            }

            File.Delete(TempPath);

            _blocks.Clear();

            _orphanBlocks.Clear();

        }

        //Change the file's size, possibly truncating or adding to it

        private  void SetFileSize(string filePath, long fileSize)

        {

            if (String.IsNullOrWhiteSpace(filePath))

                throw new ArgumentNullException("filePath");

            if (!Path.IsPathRooted(filePath))

                throw new ArgumentException("The filePath must be rooted", "filePath");

            if (fileSize < 0)

                throw new ArgumentOutOfRangeException("fileSize");

            Contract.EndContractBlock();

            using (var stream = File.Open(filePath, FileMode.OpenOrCreate, FileAccess.Write))

            {

                stream.SetLength(fileSize);

            }

        }

       /* //Check whether we should copy the local file to a temp path        

        private  bool ShouldCopy(string localPath, string tempPath)

        {

            //No need to copy if there is no file

            if (!File.Exists(localPath))

                return false;

            //If there is no temp file, go ahead and copy

            if (!File.Exists(tempPath))

                return true;

            //If there is a temp file and is newer than the actual file, don't copy

            var localLastWrite = File.GetLastWriteTime(localPath);

            var tempLastWrite = File.GetLastWriteTime(tempPath);

            //This could mean there is an interrupted download in progress

            return (tempLastWrite < localLastWrite);

        }*/

        public bool UseOrphan(long blockIndex, string blockHash)

        {

            string blockPath=null;

            if (_orphanBlocks.TryGetValue(blockHash,out blockPath))

            {

                _blocks[blockIndex] = blockPath;

                return true;

            }

            return false;

        }

        public Task StoreBlock(long blockIndex,byte[] buffer)

        {

            var blockPath = String.Format("{0}.{1:000000}", TempPath, blockIndex);

            _blocks[blockIndex] = blockPath;

            //Remove any orphan files

            if (File.Exists(blockPath))

                File.Delete(blockPath);

            return FileAsync.WriteAllBytes(blockPath, buffer);

        }

    }

}