| ... | @@ -752,56 +752,29 @@ fn SelectLeafContext(comptime Variant: type) type { | ... | @@ -752,56 +752,29 @@ fn SelectLeafContext(comptime Variant: type) type { |
| 752 | const Result = BatchResult(Variant); | 752 | const Result = BatchResult(Variant); |
| 753 | | 753 | |
| 754 | return struct { | 754 | return struct { |
| 755 | view: *const MultiSliceView, | 755 | leaves: []const u8, |
| 756 | batch_idx: usize, | 756 | batch_idx: usize, |
| 757 | start_offset: usize, | | |
| 758 | num_leaves: usize, | | |
| 759 | | 757 | |
| 760 | fn process(ctx: @This()) Result { | 758 | fn process(ctx: @This()) Result { |
| | 759 | const num_leaves = ctx.leaves.len / chunk_size; |
| 761 | var result: Result = .{ | 760 | var result: Result = .{ |
| 762 | .batch_idx = ctx.batch_idx, | 761 | .batch_idx = ctx.batch_idx, |
| 763 | .cv_len = ctx.num_leaves * cv_size, | 762 | .cv_len = num_leaves * cv_size, |
| 764 | .cvs = undefined, | 763 | .cvs = undefined, |
| 765 | }; | 764 | }; |
| 766 | | 765 | |
| 767 | var leaf_buffer: [bytes_per_batch]u8 align(cache_line_size) = undefined; | 766 | var i: usize = 0; |
| 768 | var leaves_processed: usize = 0; | 767 | while (i + optimal_vector_len <= num_leaves) : (i += optimal_vector_len) { |
| 769 | var byte_offset = ctx.start_offset; | 768 | processLeaves( |
| 770 | var cv_offset: usize = 0; | 769 | Variant, |
| 771 | const simd_batch_bytes = optimal_vector_len * chunk_size; | 770 | optimal_vector_len, |
| 772 | while (leaves_processed + optimal_vector_len <= ctx.num_leaves) { | 771 | ctx.leaves[i * chunk_size ..][0 .. optimal_vector_len * chunk_size], |
| 773 | if (ctx.view.tryGetSlice(byte_offset, byte_offset + simd_batch_bytes)) |leaf_data| { | 772 | result.cvs[i * cv_size ..][0 .. optimal_vector_len * cv_size], |
| 774 | var leaf_cvs: [optimal_vector_len * Variant.cv_size]u8 = undefined; | 773 | ); |
| 775 | processLeaves(Variant, optimal_vector_len, leaf_data, &leaf_cvs); | | |
| 776 | @memcpy(result.cvs[cv_offset..][0..leaf_cvs.len], &leaf_cvs); | | |
| 777 | } else { | | |
| 778 | ctx.view.copyRange(byte_offset, byte_offset + simd_batch_bytes, leaf_buffer[0..simd_batch_bytes]); | | |
| 779 | var leaf_cvs: [optimal_vector_len * Variant.cv_size]u8 = undefined; | | |
| 780 | processLeaves(Variant, optimal_vector_len, leaf_buffer[0..simd_batch_bytes], &leaf_cvs); | | |
| 781 | @memcpy(result.cvs[cv_offset..][0..leaf_cvs.len], &leaf_cvs); | | |
| 782 | } | | |
| 783 | leaves_processed += optimal_vector_len; | | |
| 784 | byte_offset += optimal_vector_len * chunk_size; | | |
| 785 | cv_offset += optimal_vector_len * cv_size; | | |
| 786 | } | 774 | } |
| 787 | | 775 | while (i < num_leaves) : (i += 1) { |
| 788 | while (leaves_processed < ctx.num_leaves) { | 776 | const leaf = MultiSliceView.init(ctx.leaves[i * chunk_size ..][0..chunk_size], &.{}, &.{}); |
| 789 | const leaf_end = byte_offset + chunk_size; | 777 | Variant.turboShakeToBuffer(&leaf, 0x0B, result.cvs[i * cv_size ..][0..cv_size]); |
| 790 | var cv_buffer: [64]u8 = undefined; | | |
| 791 | | | |
| 792 | if (ctx.view.tryGetSlice(byte_offset, leaf_end)) |leaf_data| { | | |
| 793 | const cv_slice = MultiSliceView.init(leaf_data, &[_]u8{}, &[_]u8{}); | | |
| 794 | Variant.turboShakeToBuffer(&cv_slice, 0x0B, cv_buffer[0..cv_size]); | | |
| 795 | } else { | | |
| 796 | ctx.view.copyRange(byte_offset, leaf_end, leaf_buffer[0..chunk_size]); | | |
| 797 | const cv_slice = MultiSliceView.init(leaf_buffer[0..chunk_size], &[_]u8{}, &[_]u8{}); | | |
| 798 | Variant.turboShakeToBuffer(&cv_slice, 0x0B, cv_buffer[0..cv_size]); | | |
| 799 | } | | |
| 800 | @memcpy(result.cvs[cv_offset..][0..cv_size], cv_buffer[0..cv_size]); | | |
| 801 | | | |
| 802 | leaves_processed += 1; | | |
| 803 | byte_offset += chunk_size; | | |
| 804 | cv_offset += cv_size; | | |
| 805 | } | 778 | } |
| 806 | | 779 | |
| 807 | return result; | 780 | return result; |
| ... | @@ -834,38 +807,20 @@ fn FinalLeafContext(comptime Variant: type) type { | ... | @@ -834,38 +807,20 @@ fn FinalLeafContext(comptime Variant: type) type { |
| 834 | }; | 807 | }; |
| 835 | } | 808 | } |
| 836 | | 809 | |
| 837 | fn ktMultiThreaded( | 810 | /// Hashes complete leaves on multiple threads and absorbs chaining values in order. |
| | 811 | fn absorbLeavesParallel( |
| 838 | comptime Variant: type, | 812 | comptime Variant: type, |
| 839 | allocator: Allocator, | 813 | allocator: Allocator, |
| 840 | io: Io, | 814 | io: Io, |
| 841 | view: *const MultiSliceView, | 815 | final_state: *Variant.StateType, |
| 842 | total_len: usize, | 816 | leaves: []const u8, |
| 843 | output: []u8, | 817 | ) error{ OutOfMemory, Canceled }!void { |
| 844 | ) !void { | | |
| 845 | comptime assert(bytes_per_batch % (optimal_vector_len * chunk_size) == 0); | 818 | comptime assert(bytes_per_batch % (optimal_vector_len * chunk_size) == 0); |
| | 819 | assert(leaves.len % chunk_size == 0); |
| 846 | | 820 | |
| 847 | const cv_size = Variant.cv_size; | 821 | const cv_size = Variant.cv_size; |
| 848 | const StateType = Variant.StateType; | | |
| 849 | const leaves_per_batch = bytes_per_batch / chunk_size; | 822 | const leaves_per_batch = bytes_per_batch / chunk_size; |
| 850 | const remaining_bytes = total_len - chunk_size; | 823 | const full_leaves = leaves.len / chunk_size; |
| 851 | const total_leaves = std.math.divCeil(usize, remaining_bytes, chunk_size) catch unreachable; | | |
| 852 | | | |
| 853 | var final_state = StateType.init(.{}); | | |
| 854 | | | |
| 855 | var first_chunk_buffer: [chunk_size]u8 = undefined; | | |
| 856 | if (view.tryGetSlice(0, chunk_size)) |first_chunk| { | | |
| 857 | final_state.update(first_chunk); | | |
| 858 | } else { | | |
| 859 | view.copyRange(0, chunk_size, &first_chunk_buffer); | | |
| 860 | final_state.update(&first_chunk_buffer); | | |
| 861 | } | | |
| 862 | | | |
| 863 | const padding = [_]u8{ 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 }; | | |
| 864 | final_state.update(&padding); | | |
| 865 | | | |
| 866 | const full_leaves = remaining_bytes / chunk_size; | | |
| 867 | const has_partial_leaf = (remaining_bytes % chunk_size) != 0; | | |
| 868 | const partial_leaf_size = if (has_partial_leaf) remaining_bytes % chunk_size else 0; | | |
| 869 | | 824 | |
| 870 | if (full_leaves > 0) { | 825 | if (full_leaves > 0) { |
| 871 | const total_batches = std.math.divCeil(usize, full_leaves, leaves_per_batch) catch unreachable; | 826 | const total_batches = std.math.divCeil(usize, full_leaves, leaves_per_batch) catch unreachable; |
| ... | @@ -893,14 +848,11 @@ fn ktMultiThreaded( | ... | @@ -893,14 +848,11 @@ fn ktMultiThreaded( |
| 893 | while (batches_spawned < total_batches and batches_spawned - next_to_process < max_concurrent) { | 848 | while (batches_spawned < total_batches and batches_spawned - next_to_process < max_concurrent) { |
| 894 | const batch_start_leaf = batches_spawned * leaves_per_batch; | 849 | const batch_start_leaf = batches_spawned * leaves_per_batch; |
| 895 | const batch_leaves = @min(leaves_per_batch, full_leaves - batch_start_leaf); | 850 | const batch_leaves = @min(leaves_per_batch, full_leaves - batch_start_leaf); |
| 896 | const start_offset = chunk_size + batch_start_leaf * chunk_size; | | |
| 897 | | 851 | |
| 898 | select_outstanding += 1; | 852 | select_outstanding += 1; |
| 899 | select.async(.batch, SelectLeafContext(Variant).process, .{SelectLeafContext(Variant){ | 853 | select.async(.batch, SelectLeafContext(Variant).process, .{SelectLeafContext(Variant){ |
| 900 | .view = view, | 854 | .leaves = leaves[batch_start_leaf * chunk_size ..][0 .. batch_leaves * chunk_size], |
| 901 | .batch_idx = batches_spawned, | 855 | .batch_idx = batches_spawned, |
| 902 | .start_offset = start_offset, | | |
| 903 | .num_leaves = batch_leaves, | | |
| 904 | }}); | 856 | }}); |
| 905 | batches_spawned += 1; | 857 | batches_spawned += 1; |
| 906 | } | 858 | } |
| ... | @@ -932,29 +884,6 @@ fn ktMultiThreaded( | ... | @@ -932,29 +884,6 @@ fn ktMultiThreaded( |
| 932 | | 884 | |
| 933 | assert(select_outstanding == 0); | 885 | assert(select_outstanding == 0); |
| 934 | } | 886 | } |
| 935 | | | |
| 936 | if (has_partial_leaf) { | | |
| 937 | var cv_buffer: [64]u8 = undefined; | | |
| 938 | var leaf_buffer: [chunk_size]u8 = undefined; | | |
| 939 | | | |
| 940 | const start_offset = chunk_size + full_leaves * chunk_size; | | |
| 941 | if (view.tryGetSlice(start_offset, start_offset + partial_leaf_size)) |leaf_data| { | | |
| 942 | const cv_slice = MultiSliceView.init(leaf_data, &[_]u8{}, &[_]u8{}); | | |
| 943 | Variant.turboShakeToBuffer(&cv_slice, 0x0B, cv_buffer[0..cv_size]); | | |
| 944 | } else { | | |
| 945 | view.copyRange(start_offset, start_offset + partial_leaf_size, leaf_buffer[0..partial_leaf_size]); | | |
| 946 | const cv_slice = MultiSliceView.init(leaf_buffer[0..partial_leaf_size], &[_]u8{}, &[_]u8{}); | | |
| 947 | Variant.turboShakeToBuffer(&cv_slice, 0x0B, cv_buffer[0..cv_size]); | | |
| 948 | } | | |
| 949 | final_state.update(cv_buffer[0..cv_size]); | | |
| 950 | } | | |
| 951 | | | |
| 952 | const n_enc = rightEncode(total_leaves); | | |
| 953 | final_state.update(n_enc.slice()); | | |
| 954 | const terminator = [_]u8{ 0xFF, 0xFF }; | | |
| 955 | final_state.update(&terminator); | | |
| 956 | | | |
| 957 | final_state.final(output); | | |
| 958 | } | 887 | } |
| 959 | | 888 | |
| 960 | /// Generic KangarooTwelve hash function builder. | 889 | /// Generic KangarooTwelve hash function builder. |
| ... | @@ -1129,6 +1058,30 @@ fn KTHash( | ... | @@ -1129,6 +1058,30 @@ fn KTHash( |
| 1129 | } | 1058 | } |
| 1130 | } | 1059 | } |
| 1131 | | 1060 | |
| | 1061 | /// Like `update`, but uses multiple threads for large inputs. |
| | 1062 | /// |
| | 1063 | /// Inputs smaller than 2 MiB are hashed on the current thread. |
| | 1064 | /// The allocator is only used for temporary buffers. |
| | 1065 | /// |
| | 1066 | /// After an error, the state is partially updated and must not be used anymore. |
| | 1067 | pub fn updateParallel(self: *Self, data: []const u8, allocator: Allocator, io: Io) error{ OutOfMemory, Canceled }!void { |
| | 1068 | if (data.len < large_file_threshold) { |
| | 1069 | return self.update(data); |
| | 1070 | } |
| | 1071 | |
| | 1072 | // Finish the first chunk or the current leaf, so that the rest starts on a leaf boundary. |
| | 1073 | const head_len = if (self.final_state == null or self.buffer_len > 0) chunk_size - self.buffer_len else 0; |
| | 1074 | self.update(data[0..head_len]); |
| | 1075 | self.flushPendingChunks(); |
| | 1076 | |
| | 1077 | const rest = data[head_len..]; |
| | 1078 | const full_leaves = rest.len / chunk_size; |
| | 1079 | try absorbLeavesParallel(Variant, allocator, io, &self.final_state.?, rest[0 .. full_leaves * chunk_size]); |
| | 1080 | self.num_leaves += full_leaves; |
| | 1081 | self.message_len += full_leaves * chunk_size; |
| | 1082 | self.update(rest[full_leaves * chunk_size ..]); |
| | 1083 | } |
| | 1084 | |
| 1132 | /// Finalize the hash and produce output. | 1085 | /// Finalize the hash and produce output. |
| 1133 | /// | 1086 | /// |
| 1134 | /// Unlike traditional hash functions, the output can be of any length. | 1087 | /// Unlike traditional hash functions, the output can be of any length. |
| ... | @@ -1256,27 +1209,15 @@ fn KTHash( | ... | @@ -1256,27 +1209,15 @@ fn KTHash( |
| 1256 | /// Hash with automatic parallelization for large inputs (>2MB). | 1209 | /// Hash with automatic parallelization for large inputs (>2MB). |
| 1257 | /// Automatically uses sequential processing for smaller inputs to avoid thread overhead. | 1210 | /// Automatically uses sequential processing for smaller inputs to avoid thread overhead. |
| 1258 | /// Allocator required for temporary buffers. IO object required for thread management. | 1211 | /// Allocator required for temporary buffers. IO object required for thread management. |
| 1259 | pub fn hashParallel(message: []const u8, out: []u8, options: Options, allocator: Allocator, io: Io) !void { | 1212 | pub fn hashParallel(message: []const u8, out: []u8, options: Options, allocator: Allocator, io: Io) error{ OutOfMemory, Canceled }!void { |
| 1260 | const custom = options.customization orelse &[_]u8{}; | | |
| 1261 | | | |
| 1262 | const custom_len_enc = rightEncode(custom.len); | | |
| 1263 | const view = MultiSliceView.init(message, custom, custom_len_enc.slice()); | | |
| 1264 | const total_len = view.totalLen(); | | |
| 1265 | | | |
| 1266 | // Single chunk case | | |
| 1267 | if (total_len <= chunk_size) { | | |
| 1268 | singleChunkFn(&view, 0x07, out); | | |
| 1269 | return; | | |
| 1270 | } | | |
| 1271 | | | |
| 1272 | // Use single-threaded processing if below threshold | 1213 | // Use single-threaded processing if below threshold |
| 1273 | if (total_len < large_file_threshold) { | 1214 | if (message.len < large_file_threshold) { |
| 1274 | ktSingleThreaded(Variant, &view, total_len, out); | 1215 | return hash(message, out, options); |
| 1275 | return; | | |
| 1276 | } | 1216 | } |
| 1277 | | 1217 | |
| 1278 | // Tree mode - multi-threaded processing | 1218 | var hasher = Self.init(options); |
| 1279 | try ktMultiThreaded(Variant, allocator, io, &view, total_len, out); | 1219 | try hasher.updateParallel(message, allocator, io); |
| | 1220 | hasher.final(out); |
| 1280 | } | 1221 | } |
| 1281 | }; | 1222 | }; |
| 1282 | } | 1223 | } |
| ... | @@ -1561,6 +1502,37 @@ test "KT256 sequential and parallel produce same output with customization" { | ... | @@ -1561,6 +1502,37 @@ test "KT256 sequential and parallel produce same output with customization" { |
| 1561 | try std.testing.expectEqualSlices(u8, &output_seq, &output_par); | 1502 | try std.testing.expectEqualSlices(u8, &output_seq, &output_par); |
| 1562 | } | 1503 | } |
| 1563 | | 1504 | |
| | 1505 | test "parallel hashing matches sequential hashing" { |
| | 1506 | const allocator = std.testing.allocator; |
| | 1507 | const io = std.testing.io; |
| | 1508 | |
| | 1509 | var prng = std.Random.DefaultPrng.init(std.testing.random_seed); |
| | 1510 | const message = try allocator.alloc(u8, 3 * large_file_threshold + 5 * chunk_size + 123); |
| | 1511 | defer allocator.free(message); |
| | 1512 | prng.random().bytes(message); |
| | 1513 | |
| | 1514 | const update_lens = [_]usize{ 100, large_file_threshold + chunk_size - 100, 3 * chunk_size, large_file_threshold + 77 }; |
| | 1515 | |
| | 1516 | inline for (.{ KT128, KT256 }) |KT| { |
| | 1517 | var expected: [64]u8 = undefined; |
| | 1518 | var actual: [64]u8 = undefined; |
| | 1519 | try KT.hash(message, &expected, .{ .customization = "test" }); |
| | 1520 | |
| | 1521 | try KT.hashParallel(message, &actual, .{ .customization = "test" }, allocator, io); |
| | 1522 | try std.testing.expectEqualSlices(u8, &expected, &actual); |
| | 1523 | |
| | 1524 | var hasher = KT.init(.{ .customization = "test" }); |
| | 1525 | var rest: []const u8 = message; |
| | 1526 | for (update_lens) |len| { |
| | 1527 | try hasher.updateParallel(rest[0..len], allocator, io); |
| | 1528 | rest = rest[len..]; |
| | 1529 | } |
| | 1530 | try hasher.updateParallel(rest, allocator, io); |
| | 1531 | hasher.final(&actual); |
| | 1532 | try std.testing.expectEqualSlices(u8, &expected, &actual); |
| | 1533 | } |
| | 1534 | } |
| | 1535 | |
| 1564 | /// Helper: Generate pattern data where data[i] = (i % 251) | 1536 | /// Helper: Generate pattern data where data[i] = (i % 251) |
| 1565 | fn generatePattern(allocator: Allocator, len: usize) ![]u8 { | 1537 | fn generatePattern(allocator: Allocator, len: usize) ![]u8 { |
| 1566 | const data = try allocator.alloc(u8, len); | 1538 | const data = try allocator.alloc(u8, len); |
Frank Denis <124872+jedisct1@users.noreply.github.com> 2026-10-03 16:46:41+02:00