7
\$\begingroup\$

I'm working on optimizing a class for decompressing Mass Effect 3's .sfar files. .sfar files are archives, and this function decompresses a file from inside an .sfar and writes it to the provided Stream.

Sometimes the file isn't compressed, and so just gets copied to the output stream. Most of the time though, the file is split into a lot of individually compressed blocks (and occasionally uncompressed ones).

I figured that I could speed up the process by decompressing the blocks in parallel. And it did result in a fairly significant speedup compared to the original single-threaded synchronous function. I'd like to know if this is a good way to have done it though. Is there anything I'm missing that could speed this up some more?

public async Task DecompressEntryAsync(int Index, Stream output)
{
    FileEntryStruct e = Files[Index];
    using (FileStream fs = new FileStream(FileName, FileMode.Open, FileAccess.Read, FileShare.None, 4096, useAsync: true))
    {
        fs.Seek(e.BlockOffsets[0], SeekOrigin.Begin);
        byte[] buff;
        if (e.BlockSizeIndex == 0xFFFFFFFF)
        {
            buff = new byte[e.RealUncompressedSize];
            await fs.ReadAsync(buff, 0, buff.Length).ConfigureAwait(continueOnCapturedContext: false);
            await output.WriteAsync(buff, 0, buff.Length).ConfigureAwait(continueOnCapturedContext: false);
        }
        else
        {
            uint count = 0;
            byte[] inputBlock;
            long left = e.RealUncompressedSize;
            List<Task<byte[]>> tasks = new List<Task<byte[]>>();
            while (left > 0)
            {
                uint compressedBlockSize = e.BlockSizes[count];
                if (compressedBlockSize == 0)
                    compressedBlockSize = Header.MaxBlockSize;
                if (compressedBlockSize == Header.MaxBlockSize || compressedBlockSize == left)
                {
                    left -= compressedBlockSize;
                    buff = new byte[compressedBlockSize];
                    await fs.ReadAsync(buff, 0, buff.Length).ConfigureAwait(continueOnCapturedContext: false);
                    tasks.Add(Task.FromResult(buff));
                }
                else
                {
                    var uncompressedBlockSize = Math.Min(left, Header.MaxBlockSize);
                    left -= uncompressedBlockSize;
                    if (compressedBlockSize < 5)
                    {
                        throw new Exception("compressed block size smaller than 5");
                    }
                    inputBlock = new byte[compressedBlockSize];

                    await fs.ReadAsync(inputBlock, 0, (int)compressedBlockSize).ConfigureAwait(continueOnCapturedContext: false);

                    tasks.Add(SevenZipHelper.DecompressAsync(inputBlock, (int)uncompressedBlockSize));
                }
                count++;
            }
            await Task.WhenAll(tasks).ConfigureAwait(continueOnCapturedContext: false);
            foreach (var task in tasks)
            {
                buff = task.Result;
                await output.WriteAsync(buff, 0, buff.Length).ConfigureAwait(continueOnCapturedContext: false);
            }
        } 
    }
}

Here's the SevenZipHelper.DecompressAsync function:

public static Task<byte[]> DecompressAsync(byte[] inputBytes, int outSize)
{
    return Task.Run(() =>
    {
        MemoryStream newInStream = new MemoryStream(inputBytes);

        Decoder decoder = new Decoder();

        newInStream.Seek(0, 0);

        byte[] properties2 = new byte[5];
        if (newInStream.Read(properties2, 0, 5) != 5)
            throw (new Exception("input .lzma is too short"));
        decoder.SetDecoderProperties(properties2);

        long compressedSize = newInStream.Length - newInStream.Position;
        MemoryStream newOutStream = new MemoryStream();
        decoder.Code(newInStream, newOutStream, compressedSize, outSize, null);
        if (newOutStream.Length != outSize)
            throw new Exception("Decompression Error");
        return newOutStream.ToArray();
    });
}
\$\endgroup\$
1
  • 2
    \$\begingroup\$ Basically, you are reading all blocks in //, but you are writing them in serial. You may eventually win some perfs by writing blocks as soon as they are decompressed. But, you may encounter hard times to write the blocks in the good order :=) \$\endgroup\$
    – thesyndarn
    Aug 3 '16 at 20:22
3
\$\begingroup\$

This is a perfect task for the TPL Dataflow library (by Microsoft). Available on Nuget.

Using dataflow you can post tasks serially, process them in parallel and complete them serially maintaining order. This achieves the goal in the comment by @thesyndarn. As soon as the first block is processed it will immediately written to the output. Blocks that finish processing out of order will wait for their turn to be written.

To set up the dataflow you need to define two blocks, a decompression block and an output writer block, and link them together:

using System.Threading.Tasks.Dataflow;
...
internal class InputBlock
{
    public const long Uncompressed = -1;

    public byte[] Data { get; }
    public long UncompressedSize { get; }
    public bool IsCompressed { get; }

    public InputBlock(byte[] data, long uncompressedSize)
    {
        Data = data;
        UncompressedSize = uncompressedSize;
        IsCompressed = uncompressedSize > 0;
    }
}
...
var decompressor = new TransformBlock<InputBlock, byte[]>(
    input => input.IsCompressed
        ? SevenZipHelper.Decompress(input.Data, input.UncompressedSize)
        : input.Data
    , new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded }
    );

var outputWriter = new ActionBlock<byte[]>(
    data => output.Write(data, 0, data.Length)
    );

decompressor.LinkTo(outputWriter, new DataflowLinkOptions { PropagateCompletion = true });
...
decompressor.Post(new InputBlock(compressedData, uncompressedBlockSize));

Post data to the decompressor, it will process it in parallel and pass it on to the outputWriter in the same order as it was posted.

This needs a synchronous Decompress:

public static byte[] Decompress(byte[] inputBytes, long decompressedSize)
{
    var compressed = new MemoryStream(inputBytes);
    var decoder = new Decoder();

    var properties2 = new byte[5];
    if (compressed.Read(properties2, 0, 5) != 5)
    {
        throw (new Exception("input .lzma is too short"));
    }

    decoder.SetDecoderProperties(properties2);

    var compressedSize = compressed.Length - compressed.Position;
    var decompressed = new MemoryStream();
    decoder.Code(compressed, decompressed, compressedSize, decompressedSize, null);

    if (decompressed.Length != decompressedSize)
        throw new Exception("Decompression Error");

    return decompressed.ToArray();
}

I (think I) made the variable names a little easier to read.

And now you just use the dataflow pipeline instead of firing off tasks:

public async Task DecompressEntryAsync(int index, Stream output)
{
    var entry = Files[index];
    using (var fs = new FileStream(FileName, FileMode.Open, FileAccess.Read, FileShare.None, 4096, useAsync: true))
    {
        fs.Seek(entry.BlockOffsets[0], SeekOrigin.Begin);

        var isUncompressed = entry.BlockSizeIndex == 0xFFFFFFFF;
        if (isUncompressed)
        {
            var uncompressed = new byte[entry.RealUncompressedSize];
            await fs.ReadAsync(uncompressed, 0, uncompressed.Length).ConfigureAwait(continueOnCapturedContext: false);
            await output.WriteAsync(uncompressed, 0, uncompressed.Length).ConfigureAwait(continueOnCapturedContext: false);

            return;
        }

        var decompressor = new TransformBlock<InputBlock, byte[]>(
            input => input.IsCompressed
                ? SevenZipHelper.Decompress(input.Data, input.UncompressedSize)
                : input.Data
            , new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded }
            );

        var outputWriter = new ActionBlock<byte[]>(
            data => output.Write(data, 0, data.Length)
            );

        decompressor.LinkTo(outputWriter, new DataflowLinkOptions { PropagateCompletion = true });

        uint count = 0;
        long left = entry.RealUncompressedSize;
        while (left > 0)
        {
            uint compressedBlockSize = entry.BlockSizes[count];
            if (compressedBlockSize == 0)
            {
                compressedBlockSize = Header.MaxBlockSize;
            }

            if (compressedBlockSize == Header.MaxBlockSize ||
                compressedBlockSize == left)
            {
                left -= compressedBlockSize;
                var uncompressedData = new byte[compressedBlockSize];
                await fs.ReadAsync(uncompressedData, 0, uncompressedData.Length).ConfigureAwait(continueOnCapturedContext: false);
                decompressor.Post(new InputBlock(uncompressedData, InputBlock.Uncompressed));
            }
            else
            {
                var uncompressedBlockSize = Math.Min(left, Header.MaxBlockSize);
                left -= uncompressedBlockSize;
                if (compressedBlockSize < 5)
                {
                    throw new Exception("compressed block size smaller than 5");
                }

                var compressedData = new byte[compressedBlockSize];
                await fs.ReadAsync(compressedData, 0, (int)compressedBlockSize).ConfigureAwait(continueOnCapturedContext: false);

                decompressor.Post(new InputBlock(compressedData, uncompressedBlockSize));
            }
            count++;
        }

        decompressor.Complete();
        await outputWriter.Completion;
    }
}
\$\endgroup\$
1
  • \$\begingroup\$ Thanks! On the 1.8 GB file I've been testing with, this sped up decompression from 6:15 to 5:45. TPL Dataflow looks like a really useful library. I'll definitely be using it more. \$\endgroup\$
    – SirCxyrtyx
    Aug 5 '16 at 17:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.