Synnefo » Pithos MS Client

#region

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

 * <copyright file="BlockHashAlgorithms.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.Collections.Concurrent;

using System.Diagnostics.Contracts;

using System.IO;

using System.Reflection;

using System.Security.Cryptography;

using System.ServiceModel.Channels;

using System.Threading;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

namespace Pithos.Network

{

    using System;

    /// <summary>

    /// TODO: Update summary.

    /// </summary>

    public static class BlockHashAlgorithms

    {

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

/*

        public static Func<FileStream, int, string, ConcurrentDictionary<int, byte[]>, int, Task<ConcurrentDictionary<int, byte[]>>> CalculateBlockHash;

        public static Task<ConcurrentDictionary<int, byte[]>> CalculateBlockHashesRecursiveAsync(FileStream stream, int blockSize, string algorithm, ConcurrentDictionary<int, byte[]> hashes = null, int index = 0)

        {

            if (stream == null)

                throw new ArgumentNullException("stream");

            if (String.IsNullOrWhiteSpace(algorithm))

                throw new ArgumentNullException("algorithm");

            if (blockSize <= 0)

                throw new ArgumentOutOfRangeException("blockSize", "blockSize must be a value greater than zero ");

            if (index < 0)

                throw new ArgumentOutOfRangeException("index", "index must be a non-negative value");

            Contract.EndContractBlock();

            if (hashes == null)

                hashes = new ConcurrentDictionary<int, byte[]>();

            var buffer = new byte[blockSize];

            return stream.ReadAsync(buffer, 0, blockSize).ContinueWith(t =>

            {

                var read = t.Result;

                var nextTask = read == blockSize

                                    ? CalculateBlockHashesRecursiveAsync(stream, blockSize, algorithm, hashes, index + 1)

                                    : Task.Factory.StartNew(() => hashes);

                if (read > 0)

                    using (var hasher = HashAlgorithm.Create(algorithm))

                    {

                        //This code was added for compatibility with the way Pithos calculates the last hash

                        //We calculate the hash only up to the last non-null byte

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

                        var hash = hasher.ComputeHash(buffer, 0, lastByteIndex + 1);

                        hashes[index] = hash;

                    }

                return nextTask;

            }).Unwrap();

        }

        public static async Task<ConcurrentDictionary<int, byte[]>> CalculateBlockHashesAsync(FileStream stream, int blockSize, string algorithm, ConcurrentDictionary<int, byte[]> hashes = null, int index = 0)

        {

            if (stream == null)

                throw new ArgumentNullException("stream");

            if (String.IsNullOrWhiteSpace(algorithm))

                throw new ArgumentNullException("algorithm");

            if (blockSize <= 0)

                throw new ArgumentOutOfRangeException("blockSize", "blockSize must be a value greater than zero ");

            Contract.EndContractBlock();

            if (hashes == null)

                hashes = new ConcurrentDictionary<int, byte[]>();

            var buffer = new byte[blockSize];

            int read;

            while ((read = await stream.ReadAsync(buffer, 0, blockSize)) > 0)

            {

                using (var hasher = HashAlgorithm.Create(algorithm))

                {

                    //This code was added for compatibility with the way Pithos calculates the last hash

                    //We calculate the hash only up to the last non-null byte

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

                    var hash = hasher.ComputeHash(buffer, 0, lastByteIndex + 1);

                    hashes[index] = hash;

                }

                index += read;

            }

            return hashes;

        }

        public static async Task<ConcurrentDictionary<int, byte[]>> CalculateBlockHashesAgentAsync(FileStream stream, int blockSize, string algorithm, int parallelism)

        {

            if (stream == null)

                throw new ArgumentNullException("stream");

            if (String.IsNullOrWhiteSpace(algorithm))

                throw new ArgumentNullException("algorithm");

            if (blockSize <= 0)

                throw new ArgumentOutOfRangeException("blockSize", "blockSize must be a value greater than zero ");

            Contract.EndContractBlock();

            var hashes = new ConcurrentDictionary<int, byte[]>();

            var path = stream.Name;

            var size = stream.Length;

            Log.DebugFormat("Hashing [{0}] size [{1}]",path,size);

            var options = new ExecutionDataflowBlockOptions {BoundedCapacity = parallelism,MaxDegreeOfParallelism=parallelism};

            var hashBlock=new ActionBlock<Tuple<int,byte[]>>(input=>

                              {

                                  int idx = input.Item1;

                                  byte[] block = input.Item2;

                                  using (var hasher = HashAlgorithm.Create(algorithm))

                                  {

                                      //This code was added for compatibility with the way Pithos calculates the last hash

                                      //We calculate the hash only up to the last non-null byte

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

                                      var hash = hasher.ComputeHash(block, 0, lastByteIndex + 1);

                                      hashes[idx] = hash;

                                  }                                  

                              },options);

            var buffer = new byte[blockSize];

            int read;

            int index = 0;

            while ((read = await stream.ReadAsync(buffer, 0, blockSize)) > 0)

            {

                var block = new byte[read];

                Buffer.BlockCopy(buffer, 0, block, 0, read);

                await hashBlock.SendAsync(Tuple.Create(index, block));

                index += read;

            }

            hashBlock.Complete();

            await hashBlock.Completion;

            return hashes;

        } 

        public static async Task<ConcurrentDictionary<int, byte[]>> CalculateBlockHashesInPlace(FileStream stream, int blockSize, string algorithm, int parallelism)

        {

            if (stream == null)

                throw new ArgumentNullException("stream");

            if (String.IsNullOrWhiteSpace(algorithm))

                throw new ArgumentNullException("algorithm");

            if (blockSize <= 0)

                throw new ArgumentOutOfRangeException("blockSize", "blockSize must be a value greater than zero ");

            Contract.EndContractBlock();

            var hashes = new ConcurrentDictionary<int, byte[]>();

            var path = stream.Name;

            var size = stream.Length;

            Log.DebugFormat("Hashing [{0}] size [{1}]",path,size);

            var buffer = new byte[blockSize];

            var index = 0;

            using (var hasher = HashAlgorithm.Create(algorithm))

            {

                int read;

                while ((read = await stream.ReadAsync(buffer, 0, blockSize)) > 0)

                {

                    //This code was added for compatibility with the way Pithos calculates the last hash

                    //We calculate the hash only up to the last non-null byte

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

                    var hash = hasher.ComputeHash(buffer, 0, lastByteIndex + 1);

                    Log.DebugFormat("Hashed [{0}] [{1}/{2}] [{3:p}]", path, index,size,(double)index/size);

                    hashes[index] = hash;

                    index += read;

                }

            }

            return hashes;

        }

*/

        private static IBufferPool _bufferMgr;

        private static readonly object _syncRoot = new object();

        private static IBufferPool GetBufferManager(int blockSize,int parallelism)

        {

            lock (_syncRoot)

            {

                if (_bufferMgr == null)

                    _bufferMgr = new BufferPool(parallelism*blockSize, blockSize);

            }

            return _bufferMgr;

        }

        //public static async Task<ConcurrentDictionary<long, byte[]>> CalculateBlockHashesInPlacePFor(FileStream stream, int blockSize, string algorithm, int parallelism, Action<long, byte[], int> postAction,CancellationToken token, IProgress<double> progress)

        public static async Task<ConcurrentDictionary<long, byte[]>> CalculateBlockHashesInPlacePFor(FileStream stream, int blockSize, string algorithm, byte parallelism, CancellationToken token, IProgress<HashProgress> progress)

        {

            if (stream == null)

                throw new ArgumentNullException("stream");

            if (String.IsNullOrWhiteSpace(algorithm))

                throw new ArgumentNullException("algorithm");

            if (blockSize <= 0)

                throw new ArgumentOutOfRangeException("blockSize", "blockSize must be a value greater than zero ");

            Contract.EndContractBlock();

            var hashes = new ConcurrentDictionary<long, byte[]>();

            var path = stream.Name;

            var size = stream.Length;

            Log.DebugFormat("Hashing [{0}] size [{1}]", path, size);

            //TODO: Handle zero-length files

            if (size == 0)

            {

                var buf = new byte[0];

                using (var hasher = HashAlgorithm.Create(algorithm))

                {

                    hashes[0] = hasher.ComputeHash(buf);

                    return hashes;

                }

            }

            var blocks = size<blockSize?1:size/blockSize;

            var actualHashers = parallelism > blocks ? (byte)blocks : parallelism;

            var buffer = new byte[actualHashers][];

            var hashers = new HashAlgorithm[actualHashers];

            var bufferManager = GetBufferManager(blockSize, actualHashers);

            try

            {

                for (var i = 0; i < actualHashers; i++)

                {

                    buffer[i] = bufferManager.TakeBuffer(blockSize);// new byte[blockSize];

                    hashers[i] = HashAlgorithm.Create(algorithm);

                }

                var indices = new long[actualHashers];

                var bufferCount = new int[actualHashers];

                int read;

                int bufIdx = 0;

                long index = 0;

                //long block = 0;

                while ((read = await stream.ReadAsync(buffer[bufIdx], 0, blockSize).ConfigureAwait(false)) > 0)

                {

                    index += read;

                    indices[bufIdx] = index;

                    bufferCount[bufIdx] = read;

                    //postAction(block++, buffer[bufIdx], read);

                    //If we have filled the last buffer or if we have read from the last block,

                    //we can calculate the clocks in parallel

                    if (bufIdx == actualHashers - 1 || read < blockSize)

                    {

                        var options = new ParallelOptions {MaxDegreeOfParallelism = actualHashers,CancellationToken=token};

                        Parallel.For(0, bufIdx + 1, options,(idx,state) =>

                                                        {

                                                            //This code was added for compatibility with the way Pithos calculates the last hash

                                                            //We calculate the hash only up to the last non-null byte

                                                            options.CancellationToken.ThrowIfCancellationRequested();

                                                            var lastByteIndex = Array.FindLastIndex(buffer[idx],

                                                                                                    bufferCount[idx] - 1,

                                                                                                    aByte => aByte != 0);

                                                            var hasher = hashers[idx];

                                                            byte[] hash=hasher.ComputeHash(buffer[idx],0,lastByteIndex+1);

/*

                                                            if (buffer[idx].Length == lastByteIndex || lastByteIndex==-1)

                                                                hash = hasher.Digest(buffer[idx]);

                                                            else

                                                            {

                                                                var buf=new byte[lastByteIndex+1];

                                                                Buffer.BlockCopy(buffer[idx],0,buf,0,lastByteIndex+1);

                                                                hash = hasher.Digest(buf);

                                                            }

*/

                                                            var filePosition = indices[idx];

                                                            /*

                                                        Trace.TraceInformation("Hashed [{0}] [{1}/{2}] [{3:p}]", path,

                                                                                filePosition, size,

                                                                                (double)filePosition / size);

                            */

                                                            hashes[filePosition] = hash;

                                                            //Do not report for files smaller than 4MB

                                                            if (progress != null && stream.Length > 4*1024*1024)

                                                                progress.Report(new HashProgress((long)hashes.Count*blockSize,stream.Length));

                                                        });

                    }

                    bufIdx = (bufIdx +1)%actualHashers;

                }

            }

            finally

            {

                for (var i = 0; i < actualHashers; i++)

                {

                    if (hashers[i] != null)

                        hashers[i].Dispose();

                    bufferManager.ReturnBuffer(buffer[i]);

                }

            }

            return hashes;

        }

        static BlockHashAlgorithms()

        {

/*

            CalculateBlockHash = CalculateBlockHashesRecursiveAsync;

*/

        }

    }

}