Synnefo » Pithos MS Client

// -----------------------------------------------------------------------

// <copyright file="BlockHashAlgorithms.cs" company="Microsoft">

// TODO: Update copyright text.

// </copyright>

// -----------------------------------------------------------------------

using System.Collections.Concurrent;

using System.Diagnostics.Contracts;

using System.IO;

using System.Security.Cryptography;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

namespace Pithos.Network

{

    using System;

    using System.Collections.Generic;

    using System.Linq;

    using System.Text;

    /// <summary>

    /// TODO: Update summary.

    /// </summary>

    public static class BlockHashAlgorithms

    {

        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)

            {

                //TODO: identify the value of index

                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 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;

        }

        static BlockHashAlgorithms()

        {

            CalculateBlockHash = CalculateBlockHashesRecursiveAsync;

        }

    }

}