User-Level Multithreading

[Agagamand]

I've heard that Blake3 is possibly the only algorithm that supports parallel computation across multiple threads. But as far as I understand, in all implementations, I can't control the threads—I can't just run Parallel.ForEach and start the computation. However, I'd like to have that capability, especially since Blake3 is technically capable of providing it.

[Dissimilis]

Hi.
Did it understood correctly that you want to limit the usage of parallel threads? Or you want some deeper control like setting thread affinity?

[Agagamand]

I need more deep control.
Let's say I'm downloading a file using four threads. I'd like to calculate its hash simultaneously in each of the four threads so that, in the end, I get the same hash sum as I would if I were performing a standard hash calculation.

[Dissimilis]

Got it.
BLAKE3's tree lets you hash a byte range on its own and get back a 32-byte chaining value, and those CVs fold into exactly the root a single-threaded hash would produce. So your four download threads can each hash their own range as it arrives.
The one real constraint: a range can only be hashed independently if it lines up with the tree. Its size has to be a power-of-two number of 1024-byte chunks, and its start offset has to be a multiple of that size. Only the final range may be short. Splitting the file into four equal parts usually breaks this.

I can try implementing API around the primitives to get you this control. Are you willing to test it?

[Agagamand]

I'm a beginner at programming, but if need to test it, I'll give it a try.

I'm aware of the restrictions. I don't think it will be too difficult to comply with these requirements.

P.S.
The number of threads shouldn't be fixed at any specific number. I just used four as an example.

[Dissimilis]

Here's the shape I'd like to build. You pick one piece size, hash each piece on whatever thread it arrives on, and fold the results at the end:

using Blake3.Managed;

const int PieceSize = 1024 * 1024;      // 1 MiB
long fileLength = /* total size, from Content-Length */;
int pieceCount  = (int)((fileLength + PieceSize - 1) / PieceSize);

var cvs = new Blake3Cv[pieceCount];     // one 32-byte value per piece

// Each download thread calls this for the pieces it owns.
// Order does not matter, and threads never touch each other's slots.
void OnPieceDownloaded(int pieceIndex, ReadOnlySpan<byte> pieceBytes)
{
    cvs[pieceIndex] = Hasher.HashSubtree(pieceBytes, pieceIndex, PieceSize);
}

// Once every piece is in:
Hash hash = Hasher.FinalizeFromSubtrees(cvs, fileLength);
// same result as Hasher.Hash(wholeFile)

Two things to keep in mind:

Pass the right piece index. A piece's 32-byte value depends on where the piece sits in the file, not only on its bytes. Hashing piece 3's bytes with index 0 gives a wrong digest and no error. Passing the index is what keeps you on the tree, so you don't have to think about the alignment rules from my last message.

Every piece is exactly PieceSize, except the last one, which can be short. PieceSize has to be a power-of-two multiple of 1024 bytes (1024, 2048, 4096, and so on); 1 MiB is a good default. The number of pieces can be anything, four or thirty-seven both work.

If you use keyed or derive-key hashing, the same key goes to every piece and to the finalize call.

Does this sound like it meets your needs?

[Agagamand]

Yeah, that looks like what I need.

[Dissimilis]

This is now released in v1.4.0 on NuGet.

The final shape is a small context object rather than static methods on Hasher. It holds the piece size, the key or mode, and, when you know it, the total length, so it can tell you how many pieces there are and where each one starts, and it can check your pieces for you.

using Blake3.Managed;

long fileLength = /* from Content-Length */;

using var ctx = Blake3SubtreeContext.Create(pieceSize: 1024 * 1024, totalLength: fileLength);
var pieces = new Blake3Subtree[ctx.PieceCount];

// Each download thread calls this for the pieces it owns, in any order.
// ctx.GetPieceOffset(i) and ctx.GetPieceLength(i) tell you which byte range to request.
void OnPieceDownloaded(int pieceIndex, ReadOnlySpan<byte> pieceBytes)
{
    pieces[pieceIndex] = ctx.HashSubtree(pieceBytes, pieceIndex);
}

// Once every piece is in:
Hash hash = ctx.Finalize(pieces);   // same result as Hasher.Hash(wholeFile)

What it does for you:

• HashSubtree is safe to call from several threads at once. Any number of threads, any order.
• Piece size must be a power-of-two multiple of 1024 bytes. 1 MiB is a good default. The piece count can be anything.
• With the total length given, a piece of the wrong length is rejected as soon as you hash it, and Finalize rejects a set with a piece missing, stored at the wrong index, or hashed under a different key or piece size.
• CreateKeyed and CreateDeriveKey cover the other two modes, and Finalize has overloads for extended output of any length.
• If you do not know the total length up front, Create(pieceSize) works too. You lose the length checks, and in particular a file cut off exactly on a piece boundary would hash cleanly as the shorter file, so pass the length whenever you have it.

The one thing it cannot catch is bytes hashed under the wrong index. A piece's value depends on where it sits in the file, so that gives a wrong hash and no error.

Give it a try and let me know how it goes.

[Agagamand]

I can't seem to do it. What am I doing wrong?

long targetPartSize = totalSize / maxParts;
long partSize = Math.Max(1024L, (long)System.Numerics.BitOperations.RoundUpToPowerOf2((ulong)targetPartSize));

ConcurrentBag<ChunkInfo> parts = [];
long currentStart = 0;
int chunkIndex = 0;

while (currentStart < totalSize)
{
	long currentEnd = Math.Min(currentStart + partSize - 1, totalSize - 1);

	parts.Add(new(chunkIndex, currentStart, currentEnd));

	chunkIndex++;
	currentStart += partSize;
}

Blake3SubtreeContext blake3 = Blake3SubtreeContext.Create(pieceSize: (int)partSize, totalLength: totalSize);
Blake3Subtree[] blake3Part = new Blake3Subtree[blake3.PieceCount];

ParallelOptions options = new()
{
	MaxDegreeOfParallelism = maxParts,
	CancellationToken = cancellationToken
};

using FileStream mainFileStream = new(destinationPath, FileMode.Open, FileAccess.Write, FileShare.ReadWrite, 4096, FileOptions.Asynchronous);
Microsoft.Win32.SafeHandles.SafeFileHandle fileHandle = mainFileStream.SafeFileHandle;

await Parallel.ForEachAsync(parts, options, async (chunk, ct) =>
{
	byte[] buffer = ArrayPool<byte>.Shared.Rent(1048576);
	
	using HttpRequestMessage request = new(HttpMethod.Get, url);
	request.Headers.Range = new RangeHeaderValue(chunk.Start, chunk.End);
	
	using HttpResponseMessage response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct).ConfigureAwait(false);
	
	await using Stream contentStream = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
	
	int bytesRead;
	while ((bytesRead = await contentStream.ReadAsync(buffer, ct).ConfigureAwait(false)) > 0)
	{
		await RandomAccess.WriteAsync(fileHandle, buffer.AsMemory(0, bytesRead), currentStart, ct).ConfigureAwait(false);
		
		blake3Part[chunk.Number] = blake3.HashSubtree(buffer.AsSpan(0, bytesRead), chunk.Number);
	}

	ArrayPool<byte>.Shared.Return(buffer);
}

Blake3.Managed.Hash hashBlake3 = blake3.Finalize(blake3Part);
string result = hashBlake3.ToString();

[Dissimilis]

Thanks for the code, this makes the problem easy to spot. Two things need to change.

1. Don't make the piece size depend on the file size.

Right now you compute partSize = totalSize / maxParts. For a big file that gives a huge piece, and a piece cannot be bigger than 2 GB. When you cast a huge long to int you get 0 or a negative number, and Create throws.

Why 2 GB: HashSubtree takes the piece as a ReadOnlySpan<byte>, and .NET spans cannot be longer than int.MaxValue. It is not a BLAKE3 limitation. The proper fix is a per-piece hasher you can feed bit by bit as data arrives, so a piece never has to sit in one buffer and the size can be a long. That is planned for the next version; it would also let you keep your "hash inside the read loop" shape. Until then:

Just use a fixed piece size. 1 MiB is a good choice:

const int PieceSize = 1024 * 1024;
using var ctx = Blake3SubtreeContext.Create(PieceSize, totalSize);

A big file simply has more pieces. The number of parallel downloads is a separate setting: MaxDegreeOfParallelism = 4 means four pieces are downloading at once, and each worker keeps taking the next piece until none are left.

2. Hash each piece once, after the whole piece has arrived.

Your loop calls HashSubtree after every ReadAsync. But one ReadAsync only gives you part of the piece (however many bytes the network happened to deliver). The library expects the complete piece in one call, so it throws "Piece N must be exactly X bytes".

Fix: read the whole piece into a buffer first, then write it to the file and hash it once.

Putting both together, this is the whole download loop:

const int PieceSize = 1024 * 1024;

using var ctx = Blake3SubtreeContext.Create(PieceSize, totalSize);
var pieces = new Blake3Subtree[ctx.PieceCount];

var options = new ParallelOptions { MaxDegreeOfParallelism = maxParts, CancellationToken = cancellationToken };

await Parallel.ForEachAsync(Enumerable.Range(0, ctx.PieceCount), options, async (i, ct) =>
{
    long offset = ctx.GetPieceOffset(i);   // where this piece starts in the file
    int length = ctx.GetPieceLength(i);    // PieceSize, except the last piece may be shorter

    byte[] buffer = ArrayPool<byte>.Shared.Rent(length);
    try
    {
        using var request = new HttpRequestMessage(HttpMethod.Get, url);
        request.Headers.Range = new RangeHeaderValue(offset, offset + length - 1);

        using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, ct);
        await using var stream = await response.Content.ReadAsStreamAsync(ct);

        // Keeps reading until the whole piece is in the buffer.
        await stream.ReadExactlyAsync(buffer.AsMemory(0, length), ct);

        await RandomAccess.WriteAsync(fileHandle, buffer.AsMemory(0, length), offset, ct);
        pieces[i] = ctx.HashSubtree(buffer.AsSpan(0, length), i);
    }
    finally
    {
        ArrayPool<byte>.Shared.Return(buffer);
    }
});

Hash hash = ctx.Finalize(pieces);

You no longer need your own ChunkInfo list: ctx.PieceCount, ctx.GetPieceOffset(i) and ctx.GetPieceLength(i) give you everything.

One more bug in your original snippet, unrelated to hashing: RandomAccess.WriteAsync was given currentStart, which is the loop variable that had already counted up to the end of the file, so every download wrote to the same spot. The version above uses offset instead.

Let me know if it works.

[Agagamand]

If I split a 60 GB file into 1 MB chunks, it will generate too many requests to the server. The server might interpret this as a DDoS attack and ban me.

[Dissimilis]

Piece size and request size are two different things. You can keep a few large range requests, one per connection, and still hash in 1 MB pieces: each connection just finishes a piece every time 1 MB has passed through it. The server only sees your few big requests.

That said, v1.5.0 (released today) removes the reason for small pieces altogether. ctx.CreateSubtreeHasher(i) gives you a per-piece hasher you feed with Update on every read and complete with Finish, and the piece size is now a long, so one piece per connection works for a 60 GB file exactly as you first wrote it. Use GetPieceOffset and GetPieceLength for the Range header and the write offset so they always agree.