From fb72e811d021d5480e180ad36bdbcc581f27ee32 Mon Sep 17 00:00:00 2001 From: Yutaka Sawada <60930312+Yutaka-Sawada@users.noreply.github.com> Date: Mon, 23 Oct 2023 10:54:28 +0900 Subject: [PATCH] Improve GPU function --- source/par2j/Command_par2j.txt | 6 +- source/par2j/common2.c | 4 +- source/par2j/create.c | 17 +- source/par2j/create.h | 1 + source/par2j/gf16.c | 40 +- source/par2j/lib_opencl.c | 49 +-- source/par2j/lib_opencl.h | 7 +- source/par2j/list.c | 12 +- source/par2j/md5_crc.c | 8 +- source/par2j/par2.c | 8 +- source/par2j/par2_cmd.c | 34 +- source/par2j/reedsolomon.c | 50 ++- source/par2j/reedsolomon.h | 10 + source/par2j/res_par2j.rc | 4 +- source/par2j/rs_decode.c | 649 ++++++++++++++++++++++----------- source/par2j/rs_encode.c | 640 +++++++++++++++++++++----------- source/par2j/source.cl | 27 +- source/par2j/version.h | 2 +- 18 files changed, 1022 insertions(+), 546 deletions(-) diff --git a/source/par2j/Command_par2j.txt b/source/par2j/Command_par2j.txt index 1485ecc..c229ff3 100644 --- a/source/par2j/Command_par2j.txt +++ b/source/par2j/Command_par2j.txt @@ -1,4 +1,4 @@ -[ par2j.exe - version 1.3.3.0 or later ] +[ par2j.exe - version 1.3.3.1 or later ] Type "par2j.exe" to see version, test integrity, and show usage below. @@ -367,10 +367,10 @@ The format is "/lc#" (# is from 1 to 32 as the number of using threads). 253: It uses 3/4 number of physical Cores. 254: It uses one less threads than number of physical Cores. 0: It uses the number of physical Cores. -255: It uses one more threads than number of physical Cores. +255: It tries to use more threads than number of physical Cores. You may set additional combinations; -+1024 to disable CLMUL (and use old SSSE3 code), ++1024 to disable CLMUL (and use slower SSSE3 code), +2048 to disable JIT (for SSE2), +4096 to disable SSSE3, +8192 to disable AVX2, diff --git a/source/par2j/common2.c b/source/par2j/common2.c index f3697da..1371ee2 100644 --- a/source/par2j/common2.c +++ b/source/par2j/common2.c @@ -1,5 +1,5 @@ // common2.c -// Copyright : 2023-09-23 Yutaka Sawada +// Copyright : 2023-10-13 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -1848,7 +1848,7 @@ int sqrt32(int num) /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ int cpu_num = 1; // CPU/Core 個数が制限されてる場合は、上位に本来の数を置く -// /arch:SSE2, +1=SSSE3, +2=SSE4.1, +4=SSE4.2, +8=CLMUL, +16=AVX2, +128=JIT(SSE2), +256=Old +// /arch:SSE2, +1=SSSE3, +2=SSE4.1, +4=SSE4.2, +8=CLMUL, +16=AVX2, +128=JIT(SSE2), +256=ALTMAPなし // 上位 16-bit = L2 cache サイズから計算した制限サイズ unsigned int cpu_flag = 0; unsigned int cpu_cache = 0; // 上位 16-bit = L3 cache の制限サイズ, 下位 16-bit = 同時処理数 diff --git a/source/par2j/create.c b/source/par2j/create.c index bb24f7a..113239b 100644 --- a/source/par2j/create.c +++ b/source/par2j/create.c @@ -1,5 +1,5 @@ // create.c -// Copyright : 2023-09-23 Yutaka Sawada +// Copyright : 2023-10-22 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -320,7 +320,7 @@ error_end: return off; } -#define MAX_MULTI_READ 4 // SSDで同時に読み込む最大ファイル数 +#define MAX_MULTI_READ 6 // SSDで同時に読み込む最大ファイル数 // SSD 上で複数ファイルのハッシュ値を同時に求めるバージョン int set_common_packet_multi( @@ -348,11 +348,9 @@ unsigned int time_start = GetTickCount(); memset(th, 0, sizeof(FILE_HASH_TH) * MAX_MULTI_READ); // Core数に応じてスレッド数を増やす if ((memory_use & 32) != 0){ // NVMe SSD - if (cpu_num >= 8){ // 8 ~ 16 Cores - multi_read = 4; - } else { // 3 Cores + Hyper-threading, or 4 ~ 7 Cores - multi_read = 3; - } + multi_read = (cpu_num + 2) / 3 + 1; // 3=2, 4~6=3, 7~9=4, 10~12=5, 13~=6 + if (multi_read > MAX_MULTI_READ) + multi_read = MAX_MULTI_READ; } else { // SATA SSD multi_read = 2; } @@ -1282,6 +1280,7 @@ int create_recovery_file_1pass( int footer_size, // 末尾パケットのバッファー・サイズ HANDLE *rcv_hFile, // 各リカバリ・ファイルのハンドル unsigned char *p_buf, // 計算済みのパリティ・ブロック + unsigned char *g_buf, // GPU用 (GPUを使わない場合は NULLにすること) unsigned int unit_size) { unsigned char *packet_header, hash[HASH_SIZE]; @@ -1438,6 +1437,10 @@ int create_recovery_file_1pass( // Recovery Slice packet は後から書き込む for (j = block_start; j < block_start + block_count; j++){ + if (g_buf != NULL){ // GPUを使った場合 + // CPUスレッドと GPUスレッドの計算結果を合わせる + galois_align_xor(g_buf + (size_t)unit_size * j, p_buf, unit_size); + } // パリティ・ブロックのチェックサムを検証する checksum16_return(p_buf, hash, unit_size - HASH_SIZE); if (memcmp(p_buf + unit_size - HASH_SIZE, hash, HASH_SIZE) != 0){ diff --git a/source/par2j/create.h b/source/par2j/create.h index f434950..239bf58 100644 --- a/source/par2j/create.h +++ b/source/par2j/create.h @@ -82,6 +82,7 @@ int create_recovery_file_1pass( int footer_size, // 末尾パケットのバッファー・サイズ HANDLE *rcv_hFile, // 各リカバリ・ファイルのハンドル unsigned char *p_buf, // 計算済みのパリティ・ブロック + unsigned char *g_buf, // GPU用 (GPUを使わない場合は NULLにすること) unsigned int unit_size); // 作成中のリカバリ・ファイルを削除する diff --git a/source/par2j/gf16.c b/source/par2j/gf16.c index 4f3e44c..754794d 100644 --- a/source/par2j/gf16.c +++ b/source/par2j/gf16.c @@ -71,7 +71,6 @@ extern unsigned int cpu_flag; // declared in common2.h // CPU によって使う関数を変更する際の仮宣言 //#define NO_SIMD // SIMD を使わない場合 -//#define NO_ALTMAP // SSSE3 や JIT(SSE2) の並び替えを使わない場合 (CLMULや32バイト単位は有効) int sse_unit; @@ -134,8 +133,11 @@ int galois_create_table(void) checksum16_altmap = checksum16; checksum16_return = checksum16; #ifndef NO_SIMD -#ifndef NO_ALTMAP - if (cpu_flag & 16){ // AVX2 対応なら + if (cpu_flag & 256){ // AVX2, SSSE3, JIT(SSE2) の並び替えを使わない場合 + // 将来的には AVX-512 などの命令に対応してもいい + //printf("\nWithout ALTMAP\n"); + //sse_unit = 32; + } else if (cpu_flag & 16){ // AVX2 対応なら //printf("\nUse AVX2 & ALTMAP\n"); sse_unit = 32; // 32, 64, 128 のどれでもいい galois_align_multiply = galois_align32avx_multiply; @@ -145,16 +147,14 @@ int galois_create_table(void) checksum16_altmap = checksum16_altmap32; checksum16_return = checksum16_return32; } else if (cpu_flag & 1){ // SSSE3 対応なら - if ((cpu_flag & 256) == 0){ // SSSE3 & ALTMAP を使う - //printf("\nUse SSSE3 & ALTMAP\n"); - sse_unit = 32; // 32, 64, 128 のどれでもいい - galois_align_multiply = galois_align32_multiply; - galois_align_multiply2 = galois_align32_multiply2; - galois_altmap_change = galois_altmap32_change; - galois_altmap_return = galois_altmap32_return; - checksum16_altmap = checksum16_altmap32; - checksum16_return = checksum16_return32; - } + //printf("\nUse SSSE3 & ALTMAP\n"); + sse_unit = 32; // 32, 64, 128 のどれでもいい + galois_align_multiply = galois_align32_multiply; + galois_align_multiply2 = galois_align32_multiply2; + galois_altmap_change = galois_altmap32_change; + galois_altmap_return = galois_altmap32_return; + checksum16_altmap = checksum16_altmap32; + checksum16_return = checksum16_return32; } else { // SSSE3 が利用できない場合 if ((cpu_flag & 128) && (jit_alloc() == 0)){ // JIT(SSE2) を使う //printf("\nUse JIT(SSE2) & ALTMAP\n"); @@ -167,7 +167,6 @@ int galois_create_table(void) checksum16_return = checksum16_return256; } } -#endif #endif return 0; @@ -2792,7 +2791,7 @@ void galois_align_xor( #endif } -// 16バイト境界のバッファー専用の掛け算 +// 16バイト境界のバッファー専用の掛け算 (ALTMAP しない) void galois_align16_multiply( unsigned char *r1, // Region to multiply (must be aligned by 16) unsigned char *r2, // Products go here @@ -2826,6 +2825,16 @@ void galois_align16_multiply( // 掛け算用のテーブルを常に作成する (32バイトだと少し遅くなる) #ifndef NO_SIMD +/* + // sse_unit が 32の倍数な時だけ + } else if (cpu_flag & 16){ // AVX2 対応なら + __declspec( align(32) ) unsigned char small_table[128]; + + create_eight_table_avx2(small_table, factor); + + gf16_avx2_block32u(r1, r2, len, small_table); +*/ + } else if (cpu_flag & 1){ // SSSE3 対応なら __declspec( align(16) ) unsigned char small_table[128]; @@ -2869,7 +2878,6 @@ void galois_align16_multiply( len -= 8; } #endif - } } diff --git a/source/par2j/lib_opencl.c b/source/par2j/lib_opencl.c index 419f4fe..471697a 100644 --- a/source/par2j/lib_opencl.c +++ b/source/par2j/lib_opencl.c @@ -1,5 +1,5 @@ // lib_opencl.c -// Copyright : 2023-09-23 Yutaka Sawada +// Copyright : 2023-10-22 Yutaka Sawada // License : GPL #ifndef _WIN32_WINNT @@ -102,7 +102,6 @@ API_clEnqueueNDRangeKernel gfn_clEnqueueNDRangeKernel; 入力 OpenCL_method : どのデバイスを選ぶか unit_size : ブロックの単位サイズ -chunk_size: 分割された断片サイズ src_max : ソース・ブロック個数 出力 @@ -112,7 +111,7 @@ OpenCL_method : 動作フラグいろいろ */ // 0=成功, 1～エラー番号 -int init_OpenCL(int unit_size, int chunk_size, int *src_max) +int init_OpenCL(int unit_size, int *src_max) { char buf[2048], *p_source; int err = 0, i, j; @@ -283,23 +282,12 @@ int init_OpenCL(int unit_size, int chunk_size, int *src_max) // printf("Shared Virtual Memory = 0x%I64X\n", param_value8); #endif - ret = fn_clGetDeviceInfo(device_id[j], CL_DEVICE_ADDRESS_BITS, sizeof(cl_uint), ¶m_value, NULL); - if (ret != CL_SUCCESS) - continue; ret = fn_clGetDeviceInfo(device_id[j], CL_DEVICE_MAX_MEM_ALLOC_SIZE, sizeof(cl_ulong), ¶m_value8, NULL); if (ret != CL_SUCCESS) continue; #ifdef DEBUG_OUTPUT - printf("ADDRESS_BITS = %d\n", param_value); printf("MAX_MEM_ALLOC_SIZE = %I64d MB\n", param_value8 >> 20); #endif - if (param_value == 32){ // CL_DEVICE_ADDRESS_BITS によって確保するメモリー領域の上限を変える - if (param_value8 > 0x30000000) // 768MB までにする - param_value8 = 0x30000000; - } else { // 64-bit OS でも 2GB までにする - if (param_value8 > 0x80000000) - param_value8 = 0x80000000; - } ret = fn_clGetDeviceInfo(device_id[j], CL_DEVICE_MAX_COMPUTE_UNITS, sizeof(cl_uint), &num_groups, NULL); if (ret != CL_SUCCESS) @@ -325,13 +313,13 @@ int init_OpenCL(int unit_size, int chunk_size, int *src_max) OpenCL_group_num = num_groups; // ワークグループ数は COMPUTE_UNITS 数にする alloc_max = (size_t)param_value8; - // AMD Radeon ではメモリー領域が全体の 1/4 とは限らない + // AMD や Intel の GPU ではメモリー領域が全体の 1/4 とは限らない ret = fn_clGetDeviceInfo(device_id[j], CL_DEVICE_GLOBAL_MEM_SIZE, sizeof(cl_ulong), ¶m_value8, NULL); if (ret == CL_SUCCESS){ #ifdef DEBUG_OUTPUT printf("GLOBAL_MEM_SIZE = %I64d MB\n", param_value8 >> 20); #endif - // 領域一個あたりのサイズは全体の 1/4 までにする + // 領域一個あたりのサイズは全体の 1/4 までにする（VRAMを使いすぎると不安定になる） param_value8 /= 4; if ((cl_ulong)alloc_max > param_value8) alloc_max = (size_t)param_value8; @@ -366,7 +354,7 @@ int init_OpenCL(int unit_size, int chunk_size, int *src_max) return (ret << 8) | 12; // 計算方式を選択する - if ((((cpu_flag & 0x101) == 1) || ((cpu_flag & 16) != 0)) && (sse_unit == 32)){ + if ((((cpu_flag & 0x101) == 1) || ((cpu_flag & 0x110) == 0x10)) && (sse_unit == 32)){ OpenCL_method = 2; // SSSE3 & ALTMAP または AVX2 ならデータの並び替え対応版を使う } else if (((cpu_flag & 128) != 0) && (sse_unit == 256)){ OpenCL_method = 4; // JIT(SSE2) は bit ごとに上位から 16バイトずつ並ぶ @@ -380,10 +368,10 @@ int init_OpenCL(int unit_size, int chunk_size, int *src_max) // work group 数が必要以上に多い場合は減らす if (OpenCL_method == 2){ // work item 一個が 8バイトずつ計算する、256個なら work group ごとに 2KB 担当する - data_size = chunk_size / 2048; + data_size = unit_size / 2048; } else { // work item 一個が 4バイトずつ計算する、256個なら work group ごとに 1KB 担当する - data_size = chunk_size / 1024; + data_size = unit_size / 1024; } if (OpenCL_group_num > data_size){ OpenCL_group_num = data_size; @@ -401,9 +389,9 @@ int init_OpenCL(int unit_size, int chunk_size, int *src_max) printf("src buf : %zd KB (%d blocks), possible\n", data_size >> 10, count); #endif - // 出力先はchunk 1個分だけあればいい + // 出力先は1ブロック分だけあればいい // CL_MEM_ALLOC_HOST_PTRを使えばpinned memoryになるらしい - data_size = (chunk_size + 63) & ~63; // cache line sizes (64 bytes) の倍数にする + data_size = unit_size; OpenCL_dst = gfn_clCreateBuffer(OpenCL_context, CL_MEM_WRITE_ONLY | CL_MEM_ALLOC_HOST_PTR, data_size, NULL, &ret); if (ret != CL_SUCCESS) return (ret << 8) | 13; @@ -700,12 +688,11 @@ int gpu_copy_blocks( } // ソース・ブロックを掛け算する -int gpu_multiply_chunks( +int gpu_multiply_blocks( int src_num, // Number of multiplying source blocks unsigned short *mat, // Matrix of numbers to multiply by unsigned char *buf, // Products go here - int offset, // Offset in each block - int length) // Byte length + int len) // Byte length { unsigned __int64 *vram, *src, *dst; size_t global_size, local_size; @@ -720,14 +707,6 @@ int gpu_multiply_chunks( ret = gfn_clSetKernelArg(OpenCL_kernel, 3, sizeof(int), &src_num); if (ret != CL_SUCCESS) return (ret << 8) | 103; - offset /= 4; // 4バイト整数単位にする - ret = gfn_clSetKernelArg(OpenCL_kernel, 4, sizeof(int), &offset); - if (ret != CL_SUCCESS) - return (ret << 8) | 104; - length /= 4; // 4バイト整数単位にする - ret = gfn_clSetKernelArg(OpenCL_kernel, 5, sizeof(int), &length); - if (ret != CL_SUCCESS) - return (ret << 8) | 105; // カーネル並列実行 local_size = 256; // テーブルやキャッシュのため、work item 数は 256に固定する @@ -738,18 +717,18 @@ int gpu_multiply_chunks( return (ret << 8) | 11; // 出力内容をホスト側に反映させる - vram = gfn_clEnqueueMapBuffer(OpenCL_command, OpenCL_dst, CL_TRUE, CL_MAP_READ, 0, length * 4, 0, NULL, NULL, &ret); + vram = gfn_clEnqueueMapBuffer(OpenCL_command, OpenCL_dst, CL_TRUE, CL_MAP_READ, 0, len, 0, NULL, NULL, &ret); if (ret != CL_SUCCESS) return (ret << 8) | 12; // 8バイトごとに XOR する (SSE2 で XOR しても速くならず) src = vram; dst = (unsigned __int64 *)buf; - while (length > 0){ + while (len > 0){ *dst ^= *src; dst++; src++; - length -= 2; + len -= 8; } // ホスト側でデータを変更しなくても、clEnqueueMapBufferと対で呼び出さないといけない diff --git a/source/par2j/lib_opencl.h b/source/par2j/lib_opencl.h index 8748245..ddde655 100644 --- a/source/par2j/lib_opencl.h +++ b/source/par2j/lib_opencl.h @@ -10,7 +10,7 @@ extern "C" { extern int OpenCL_method; -int init_OpenCL(int unit_size, int chunk_size, int *src_max); +int init_OpenCL(int unit_size, int *src_max); int free_OpenCL(void); void info_OpenCL(char *buf, int buf_size); @@ -19,12 +19,11 @@ int gpu_copy_blocks( int unit_size, int src_num); -int gpu_multiply_chunks( +int gpu_multiply_blocks( int src_num, // Number of multiplying source blocks unsigned short *mat, // Matrix of numbers to multiply by unsigned char *buf, // Products go here - int offset, // Offset in each block - int length); // Byte length + int len); // Byte length int gpu_finish(void); diff --git a/source/par2j/list.c b/source/par2j/list.c index e615d85..cf826c7 100644 --- a/source/par2j/list.c +++ b/source/par2j/list.c @@ -1,5 +1,5 @@ // list.c -// Copyright : 2022-10-14 Yutaka Sawada +// Copyright : 2023-10-15 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -348,7 +348,7 @@ if (time_start > 0){ // SSD 上で複数ファイルを同時に検査する // MAX_MULTI_READ の２倍ぐらいにする？ -#define MAX_READ_NUM 10 +#define MAX_READ_NUM 12 int check_file_complete_multi( char *ascii_buf, @@ -370,11 +370,9 @@ unsigned int time_start = GetTickCount(); memset(hSub, 0, sizeof(HANDLE) * MAX_READ_NUM); // Core数に応じてスレッド数を増やす if ((memory_use & 32) != 0){ // NVMe SSD - if (cpu_num >= 8){ // 8 ~ 16 Cores - multi_read = 4; - } else { // 3 Cores + Hyper-threading, or 4 ~ 7 Cores - multi_read = 3; - } + multi_read = (cpu_num + 2) / 3 + 1; // 3=2, 4~6=3, 7~9=4, 10~12=5, 13~=6 + if (multi_read > MAX_READ_NUM / 2) + multi_read = MAX_READ_NUM / 2; } else { // SATA SSD multi_read = 2; } diff --git a/source/par2j/md5_crc.c b/source/par2j/md5_crc.c index 0ebf368..5ded3a3 100644 --- a/source/par2j/md5_crc.c +++ b/source/par2j/md5_crc.c @@ -1,5 +1,5 @@ // md5_crc.c -// Copyright : 2023-08-28 Yutaka Sawada +// Copyright : 2023-10-17 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -671,7 +671,7 @@ time1_start = GetTickCount(); } // バッファー・サイズが大きいのでヒープ領域を使う - for (io_size = IO_SIZE; io_size < 1048576; io_size += IO_SIZE){ // 1 MB までにする + for (io_size = IO_SIZE; io_size <= 1048576; io_size += IO_SIZE){ // 1 MB までにする if ((io_size + IO_SIZE > (cpu_cache & 0xFFFE0000)) || ((__int64)(io_size + IO_SIZE) * 4 > file_left)) break; } @@ -866,7 +866,7 @@ DWORD WINAPI file_hash_crc2(LPVOID lpParameter) // バッファー・サイズが大きいのでヒープ領域を使う prog_tick = 1; - for (io_size = IO_SIZE; io_size < 1048576; io_size += IO_SIZE){ // IO_SIZE の倍数で 1 MB までにする + for (io_size = IO_SIZE; io_size <= 1048576; io_size += IO_SIZE){ // IO_SIZE の倍数で 1 MB までにする if ((io_size + IO_SIZE > (cpu_cache & 0xFFFE0000)) || ((__int64)(io_size + IO_SIZE) * 4 > file_left)) break; prog_tick++; @@ -1303,7 +1303,7 @@ DWORD WINAPI file_hash_background(LPVOID lpParameter) find_next = files[num].b_off; // 先頭ブロックの番号 // バッファー・サイズが大きいのでヒープ領域を使う - for (io_size = IO_SIZE; io_size < 1048576; io_size += IO_SIZE){ // IO_SIZE の倍数で 1 MB までにする + for (io_size = IO_SIZE; io_size <= 1048576; io_size += IO_SIZE){ // IO_SIZE の倍数で 1 MB までにする if ((io_size + IO_SIZE > (cpu_cache & 0xFFFE0000)) || ((__int64)(io_size + IO_SIZE) * 4 > file_size)) break; } diff --git a/source/par2j/par2.c b/source/par2j/par2.c index 6dc44cb..5b0141e 100644 --- a/source/par2j/par2.c +++ b/source/par2j/par2.c @@ -1,5 +1,5 @@ // par2.c -// Copyright : 2023-09-21 Yutaka Sawada +// Copyright : 2023-10-15 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -181,7 +181,7 @@ int par2_create( } } else { // 共通パケットを作成する - if ((memory_use & 16) && (cpu_num >= 4) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する + if ((memory_use & 16) && (cpu_num >= 3) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する common_size = set_common_packet_multi(common_buf, &packet_num, (switch_p & 2) >> 1, files); } else { common_size = set_common_packet(common_buf, &packet_num, (switch_p & 2) >> 1, files); @@ -529,7 +529,7 @@ int par2_verify( // ソース・ファイルが完全かどうかを調べる // ファイルの状態は 完全、消失、追加、破損(完全なブロックの数) の4種類 - if ((memory_use & 16) && (cpu_num >= 4) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する + if ((memory_use & 16) && (cpu_num >= 3) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する err = check_file_complete_multi(ascii_buf, uni_buf, files, s_blk); } else { err = check_file_complete(ascii_buf, uni_buf, files, s_blk); @@ -741,7 +741,7 @@ int par2_repair( // ソース・ファイルが完全かどうかを一覧表示する // ファイルの状態は 完全、消失、追加、破損(完全なブロックの数) の4種類 - if ((memory_use & 16) && (cpu_num >= 4) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する + if ((memory_use & 16) && (cpu_num >= 3) && (entity_num >= 2)){ // SSDなら複数ファイルを同時に処理する err = check_file_complete_multi(ascii_buf, uni_buf, files, s_blk); } else { err = check_file_complete(ascii_buf, uni_buf, files, s_blk); diff --git a/source/par2j/par2_cmd.c b/source/par2j/par2_cmd.c index d3bd36f..df433bc 100644 --- a/source/par2j/par2_cmd.c +++ b/source/par2j/par2_cmd.c @@ -1,5 +1,5 @@ // par2_cmd.c -// Copyright : 2023-09-28 Yutaka Sawada +// Copyright : 2023-10-15 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -87,35 +87,37 @@ static void print_environment(void) printf("CPU thread\t: %d / %d\n", cpu_num & 0xFFFF, cpu_num >> 24); cpu_num &= 0xFFFF; // 利用するコア数だけにしておく printf("CPU cache limit : %d KB, %d KB\n", (cpu_flag & 0xFFFF0000) >> 10, (cpu_cache & 0xFFFE0000) >> 10); -#ifndef _WIN64 // 32-bit 版は MMX, SSE2, SSSE3 のどれかを表示する +#ifndef _WIN64 // 32-bit 版は MMX, SSE2, SSSE3, AVX2 のどれかを表示する printf("CPU extra\t:"); - if (cpu_flag & 1){ + if (((cpu_flag & 16) != 0) && ((cpu_flag & 256) == 0)){ + printf(" AVX2"); + } else if (cpu_flag & 1){ if (cpu_flag & 256){ - printf(" SSSE3(old)"); + printf(" SSSE3(slow)"); } else { printf(" SSSE3"); } - } else if (cpu_flag & 128){ + } else if (((cpu_flag & 128) != 0) && ((cpu_flag & 256) == 0)){ printf(" SSE2"); } else { printf(" MMX"); } -#else // 64-bit 版は SSE2, SSSE3 を表示する +#else // 64-bit 版は SSE2, SSSE3, AVX2 を表示する printf("CPU extra\t: x64"); - if (cpu_flag & 1){ + if (((cpu_flag & 16) != 0) && ((cpu_flag & 256) == 0)){ + printf(" AVX2"); + } else if (cpu_flag & 1){ if (cpu_flag & 256){ - printf(" SSSE3(old)"); + printf(" SSSE3(slow)"); } else { printf(" SSSE3"); } - } else if (cpu_flag & 128){ + } else if (((cpu_flag & 128) != 0) && ((cpu_flag & 256) == 0)){ printf(" SSE2"); } #endif if (cpu_flag & 8) printf(" CLMUL"); - if (cpu_flag & 16) - printf(" AVX2"); printf("\nMemory usage\t: "); if (memory_use & 7){ printf("%d/8", memory_use & 7); @@ -1486,8 +1488,8 @@ ri= switch_set & 0x00040000 } else if (k & 512){ OpenCL_method = -1; // Slower GPU } - if (k & 1024) // CLMUL を使わない、SSSE3 の古いエンコーダーを使う - cpu_flag = (cpu_flag & 0xFFFFFFF7) | 0x100; + if (k & 1024) // CLMUL と ALTMAP を使わない + cpu_flag = (cpu_flag & 0xFFFFFFF7) | 256; if (k & 2048) // JIT(SSE2) を使わない cpu_flag &= 0xFFFFFF7F; if (k & 4096) // SSSE3 を使わない @@ -1506,10 +1508,10 @@ ri= switch_set & 0x00040000 } else if (k == 254){ // 物理コア数より減らす k = ((cpu_num & 0x00FF0000) >> 16) - 1; } else if (k == 255){ // 物理コア数より増やす - k = ((cpu_num & 0x00FF0000) >> 16) + 1; - //k = cpu_num >> 16; - //k = ((k & 0xFF) + (k >> 8)) / 2; // 物理コア数と論理コア数の中間にする？ + k = cpu_num >> 16; + k = ((k & 0xFF) + (k >> 8)) / 2; // 物理コア数と論理コア数の中間にする？ // タスクマネージャーにおける CPU使用率は 100%になるけど、速くはならない・・・ + // k = (k & 0xFF) + ((k >> 8) - (k & 0xFF)) / 4; // 物理コア数の 5/4 にする？ } if (k > MAX_CPU){ k = MAX_CPU; diff --git a/source/par2j/reedsolomon.c b/source/par2j/reedsolomon.c index 6e2ed8e..189ba71 100644 --- a/source/par2j/reedsolomon.c +++ b/source/par2j/reedsolomon.c @@ -1,5 +1,5 @@ // reedsolomon.c -// Copyright : 2023-09-28 Yutaka Sawada +// Copyright : 2023-10-21 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -204,6 +204,48 @@ int read_block_num( return buf_num; } +// 1st encode, decode を何スレッドで実行するか決める +int calc_thread_num1(int max_num) +{ + int i, num; + + // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） + num = 0; + i = 1; + while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 + num++; + i *= 2; + } + if (num > max_num) + num = max_num; + + return num; +} + +// 1st & 2nd encode, decode を何スレッドで実行するか決める +int calc_thread_num2(int max_num, int *cpu_num2) +{ + int i, num1, num2; + + // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） + num1 = 0; + i = 1; + while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 + num1++; + i *= 2; + } + if (num1 > max_num) + num1 = max_num; + + // CPU と GPU で必ず２スレッド使う + num2 = cpu_num; + if (num2 < 2) + num2 = 2; + *cpu_num2 = num2; + + return num1; +} + /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // 戸川 隼人 の「演習と応用FORTRAN77」の逆行列の計算方法を参考にして // Gaussian Elimination を少し修正して行列の数を一つにしてみた @@ -960,8 +1002,7 @@ time_matrix = GetTickCount() - time_matrix; // ブロック数が多いなら、ブロックごとにスレッドを割り当てる (GPU を使う) if (memory_use & 16){ err = -4; // SSD なら Read all 方式でブロックが断片化しても速い - } else - if (read_block_num(block_lost, 0, MEM_UNIT) != 0){ + } else if (read_block_num(block_lost * 2, 0, MEM_UNIT) != 0){ err = -5; // HDD でメモリーが足りてるなら Read some 方式を使う } else { err = -4; // メモリー不足なら Read all 方式でブロックを断片化させる @@ -970,8 +1011,7 @@ time_matrix = GetTickCount() - time_matrix; // ソース・ブロックを全て断片的に読み込むか、いくつかを丸ごと読み込むかを決める if (memory_use & 16){ err = -2; // SSD なら Read all 方式でブロックが断片化しても速い - } else - if (read_block_num(block_lost, 0, sse_unit) != 0){ + } else if (read_block_num(block_lost, 0, sse_unit) != 0){ err = -3; // HDD でメモリーが足りてるなら Read some 方式を使う } else { err = -2; // メモリー不足なら Read all 方式でブロックを断片化させる diff --git a/source/par2j/reedsolomon.h b/source/par2j/reedsolomon.h index 5f62cfb..8fc5ba1 100644 --- a/source/par2j/reedsolomon.h +++ b/source/par2j/reedsolomon.h @@ -17,6 +17,10 @@ extern "C" { #define READ_MIN_RATE 1 // 保持するブロック数の 1/2 = 50% #define READ_MIN_NUM 16 +// CPU cache 最適化のため、同時に処理するブロック数を制限する +#define CACHE_MIN_NUM 8 +#define CACHE_MAX_NUM 128 + /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // Cache Blocking を試みる @@ -35,6 +39,12 @@ int read_block_num( size_t trial_alloc, // 確保できるか確認するのか int alloc_unit); // メモリー単位の境界 (sse_unit か MEM_UNIT) +// 1st encode, decode を何スレッドで実行するか決める +int calc_thread_num1(int max_num); + +// 1st & 2nd encode, decode を何スレッドで実行するか決める +int calc_thread_num2(int max_num, int *cpu_num2); + /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */ // リード・ソロモン符号を使ってエンコードする diff --git a/source/par2j/res_par2j.rc b/source/par2j/res_par2j.rc index c1008f4..3f6a941 100644 --- a/source/par2j/res_par2j.rc +++ b/source/par2j/res_par2j.rc @@ -1,7 +1,7 @@ 1 RT_STRING ".\\source.cl" 1 VERSIONINFO -FILEVERSION 1,3,3,0 +FILEVERSION 1,3,3,1 PRODUCTVERSION 1,3,3,0 FILEOS 0x40004 FILETYPE 0x1 @@ -13,7 +13,7 @@ BLOCK "StringFileInfo" VALUE "FileDescription", "PAR2 client" VALUE "LegalCopyright", "Copyright (C) 2023 Yutaka Sawada" VALUE "ProductName", "par2j" - VALUE "FileVersion", "1.3.3.0" + VALUE "FileVersion", "1.3.3.1" VALUE "ProductVersion", "1.3.3.0" } } diff --git a/source/par2j/rs_decode.c b/source/par2j/rs_decode.c index 29b63b7..5217895 100644 --- a/source/par2j/rs_decode.c +++ b/source/par2j/rs_decode.c @@ -1,5 +1,5 @@ // rs_decode.c -// Copyright : 2023-09-21 Yutaka Sawada +// Copyright : 2023-10-22 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -284,11 +284,11 @@ printf(" 2nd decode %d.%03d sec, %d loop, %d MB/s\n", time_encode2b / 1000, time // GPU 対応のサブ・スレッド (最後のスレッドなので、1st decode では呼ばれない) static DWORD WINAPI thread_decode_gpu(LPVOID lpParameter) { - unsigned char *s_buf, *p_buf; + unsigned char *s_buf, *g_buf; unsigned short *factor; - int i, j, block_lost, max_num, chunk_num; + int i, j, block_lost; int src_num; - unsigned int unit_size, len, off, chunk_size; + unsigned int unit_size; HANDLE hRun, hEnd; RS_TH *th; #ifdef TIMER @@ -296,18 +296,14 @@ unsigned int time_start2, time_encode2 = 0, loop_count2 = 0; #endif th = (RS_TH *)lpParameter; - p_buf = th->buf; + g_buf = th->buf; unit_size = th->size; - chunk_size = th->len; block_lost = th->count; hRun = th->run; hEnd = th->end; //_mm_sfence(); SetEvent(hEnd); // 設定完了を通知する - chunk_num = (unit_size + chunk_size - 1) / chunk_size; - max_num = chunk_num * block_lost; - WaitForSingleObject(hRun, INFINITE); // 計算開始の合図を待つ while (th->now < INT_MAX / 2){ #ifdef TIMER @@ -325,17 +321,10 @@ time_start2 = GetTickCount(); InterlockedExchange(&(th->now), INT_MAX / 3); // サブ・スレッドの計算を中断する } - // スレッドごとに復元する消失ブロックの chunk を変える - len = chunk_size; - while ((j = InterlockedIncrement(&(th->now))) < max_num){ // j = ++th_now - off = j / block_lost; // chunk の番号 - j = j % block_lost; // lost block の番号 - off *= chunk_size; // chunk の位置 - if (off + len > unit_size) - len = unit_size - off; // 最後の chunk だけサイズが異なるかも - - // VRAM上のソース・ブロックごとにパリティを追加していく - i = gpu_multiply_chunks(src_num, factor + source_num * j, p_buf + (size_t)unit_size * j + off, off, len); + // スレッドごとに復元する消失ブロックを変える + while ((j = InterlockedIncrement(&(th->now))) < block_lost){ // j = ++th_now + // 倍率は逆行列から部分的にコピーする + i = gpu_multiply_blocks(src_num, factor + source_num * j, g_buf + (size_t)unit_size * j, unit_size); if (i != 0){ th->len = i; InterlockedExchange(&(th->now), INT_MAX / 3); // サブ・スレッドの計算を中断する @@ -359,7 +348,6 @@ time_encode2 += GetTickCount() - time_start2; WaitForSingleObject(hRun, INFINITE); // 計算開始の合図を待つ } #ifdef TIMER -loop_count2 /= chunk_num; // chunk数で割ってブロック数にする printf("gpu-thread :\n"); if (time_encode2 > 0){ i = (int)((__int64)loop_count2 * unit_size * 125 / ((__int64)time_encode2 * 131072)); @@ -575,16 +563,9 @@ int decode_method2( // ソース・データを全て読み込む場合 len = try_cache_blocking(unit_size); //len = ((len + 2) / 3 + (sse_unit - 1)) & ~(sse_unit - 1); // 1/3の実験用 chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > part_num) - cpu_num1 = part_num; + cpu_num1 = calc_thread_num1(part_num); // 読み込み中はスレッド数を減らす src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num == 1)) + if ((src_max < CACHE_MIN_NUM) || (cpu_num == 1)) src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない #ifdef TIMER printf("\n read all blocks, and keep some recovering blocks\n"); @@ -1020,16 +1001,9 @@ int decode_method3( // 復元するブロックを全て保持できる場合 prog_base = (__int64)(source_num + prog_write) * block_lost + prog_read * source_num; // ブロックの合計掛け算個数 + 読み書き回数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > block_lost) - cpu_num1 = block_lost; + cpu_num1 = calc_thread_num1(block_lost); // 読み込み中はスレッド数を減らす src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num == 1)) + if ((src_max < CACHE_MIN_NUM) || (cpu_num == 1)) src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない #ifdef TIMER printf("\n read some blocks, and keep all recovering blocks\n"); @@ -1364,27 +1338,29 @@ int decode_method4( // 全てのブロックを断片的に保持する場合 (G parity_ctx_r *p_blk, // パリティ・ブロックの情報 unsigned short *mat) { - unsigned char *buf = NULL, *p_buf, *work_buf, *hash; + unsigned char *buf = NULL, *p_buf, *g_buf, *work_buf, *hash; unsigned short *id; int err = 0, i, j, last_file, chunk_num, recv_now; - int cpu_num1, src_off, src_num, src_max, vram_max; + int cpu_num1, src_off, src_num, src_max; + int cpu_num2, vram_max, cpu_end, gpu_end, th_act; unsigned int io_size, unit_size, len, block_off; unsigned int time_last, prog_read, prog_write; __int64 file_off, prog_num = 0, prog_base; HANDLE hFile = NULL; HANDLE hSub[MAX_CPU], hRun[MAX_CPU], hEnd[MAX_CPU]; - RS_TH th[1]; + RS_TH th[1], th2[1]; memset(hSub, 0, sizeof(HANDLE) * MAX_CPU); id = mat + (block_lost * source_num); // 何番目の消失ソース・ブロックがどのパリティで代替されるか // 作業バッファーを確保する // part_num を使わず、全てのブロックを保持する所がdecode_method2と異なることに注意！ - io_size = get_io_size(source_num + block_lost, NULL, 1, MEM_UNIT); + // CPU計算スレッドと GPU計算スレッドで保存先を別けるので、消失ブロック分を２倍確保する + io_size = get_io_size(source_num + block_lost * 2, NULL, 1, MEM_UNIT); //io_size = (((io_size + 1) / 2 + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1)) - HASH_SIZE; // 2分割の実験用 //io_size = (((io_size + 2) / 3 + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1)) - HASH_SIZE; // 3分割の実験用 unit_size = io_size + HASH_SIZE; // チェックサムの分だけ増やす - file_off = (source_num + block_lost) * (size_t)unit_size + HASH_SIZE; + file_off = (source_num + block_lost * 2) * (size_t)unit_size + HASH_SIZE; buf = _aligned_malloc((size_t)file_off, MEM_UNIT); // GPU 用の境界 if (buf == NULL){ printf("malloc, %I64d\n", file_off); @@ -1392,42 +1368,36 @@ int decode_method4( // 全てのブロックを断片的に保持する場合 (G goto error_end; } p_buf = buf + (size_t)unit_size * source_num; // 復元したブロックを記録する領域 - hash = p_buf + (size_t)unit_size * block_lost; + g_buf = p_buf + (size_t)unit_size * block_lost; // GPUスレッド用 + hash = g_buf + (size_t)unit_size * block_lost; prog_base = (block_size + io_size - 1) / io_size; prog_read = (block_lost + 31) / 32; // 読み書きの経過をそれぞれ 3% ぐらいにする prog_write = (source_num + 31) / 32; prog_base *= (__int64)(source_num + prog_write) * block_lost + prog_read * source_num; // 全体の断片の個数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > block_lost) - cpu_num1 = block_lost; + cpu_num1 = calc_thread_num2(block_lost, &cpu_num2); // 使用するスレッド数を調節する src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num <= 2)) - src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない + if ((src_max < CACHE_MIN_NUM) || (src_max > CACHE_MAX_NUM)) + src_max = CACHE_MAX_NUM; // 不明または極端な場合は、規定値にする + //cpu_num1 = 0; // 2nd decode の実験用に 1st decode を停止する #ifdef TIMER printf("\n read all blocks, and keep all recovering blocks (GPU)\n"); printf("buffer size = %I64d MB, io_size = %d, split = %d\n", file_off >> 20, io_size, (block_size + io_size - 1) / io_size); - printf("cache: limit size = %d, chunk_size = %d, split = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); - printf("unit_size = %d, cpu_num1 = %d, src_max = %d\n", unit_size, cpu_num1, src_max); + printf("cache: limit size = %d, chunk_size = %d, chunk_num = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); + printf("unit_size = %d, cpu_num1 = %d, cpu_num2 = %d\n", unit_size, cpu_num1, cpu_num2); #endif // OpenCL の初期化 vram_max = source_num; - i = init_OpenCL(unit_size, len, &vram_max); + i = init_OpenCL(unit_size, &vram_max); if (i != 0){ if (i != 3) // GPU が見つからなかった場合はエラー表示しない printf("init_OpenCL, %d, %d\n", i & 0xFF, i >> 8); i = free_OpenCL(); if (i != 0) printf("free_OpenCL, %d, %d", i & 0xFF, i >> 8); - OpenCL_method = 0; // GPU を使わない設定にする - // GPU を使わずに計算を続行する場合は以下をコメントアウト + OpenCL_method = 0; // GPU を使えなかった印 err = -2; // CPU だけの方式に切り替える goto error_end; } @@ -1437,10 +1407,14 @@ int decode_method4( // 全てのブロックを断片的に保持する場合 (G // マルチ・スレッドの準備をする th->buf = p_buf; + th2->buf = g_buf; th->size = unit_size; + th2->size = unit_size; th->count = block_lost; - th->len = len; // chunk size - for (j = 0; j < cpu_num; j++){ // サブ・スレッドごとに + th2->count = block_lost; + th->len = len ; // chunk size + th2->len = 0; // GPUのエラー通知用にする + for (j = 0; j < cpu_num2; j++){ // サブ・スレッドごとに hRun[j] = CreateEvent(NULL, FALSE, FALSE, NULL); // Auto Reset にする if (hRun[j] == NULL){ print_win32_err(); @@ -1457,12 +1431,13 @@ int decode_method4( // 全てのブロックを断片的に保持する場合 (G goto error_end; } // サブ・スレッドを起動する - th->run = hRun[j]; - th->end = hEnd[j]; - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを起動する - if ((j == cpu_num - 1) && (OpenCL_method != 0)){ // 最後のスレッドを GPU 管理用にする - hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode_gpu, (LPVOID)th, 0, NULL); + if (j == cpu_num2 - 1){ // 最後のスレッドを GPU 管理用にする + th2->run = hRun[j]; + th2->end = hEnd[j]; + hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode_gpu, (LPVOID)th2, 0, NULL); } else { + th->run = hRun[j]; + th->end = hEnd[j]; hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode3, (LPVOID)th, 0, NULL); } if (hSub[j] == NULL){ @@ -1475,7 +1450,6 @@ int decode_method4( // 全てのブロックを断片的に保持する場合 (G } WaitForSingleObject(hEnd[j], INFINITE); // 設定終了の合図を待つ (リセットしない) } - th->len = 0; // GPUのエラー通知用にする // ブロック断片を読み込んで、消失ブロック断片を復元する print_progress_text(0, "Recovering slice"); @@ -1629,6 +1603,7 @@ skip_count++; time_read += GetTickCount() - time_start; #endif + memset(g_buf, 0, (size_t)unit_size * block_lost); // 待機中に GPU用の領域をゼロ埋めしておく WaitForMultipleObjects(cpu_num1, hEnd, TRUE, INFINITE); // サブ・スレッドの計算終了の合図を待つ src_off += 1; // 計算を開始するソース・ブロックの番号 if (src_off > 0){ // 計算不要なソース・ブロックはとばす @@ -1647,74 +1622,150 @@ skip_count++; j = (src_off * 1000) / source_num; printf("partial decode = %d / %d (%d.%d%%), read = %d, skip = %d\n", src_off, source_num, j / 10, j % 10, read_count, skip_count); #endif + recv_now = -1; // 消失ブロックの本来のソース番号 last_file = -1; - - // GPU と CPU のどちらに最適化するかが難しい - src_num = src_max; // 一度に処理するソース・ブロックの数を制限する - if (src_num > vram_max){ // VRAM に収まらない場合は、VRAM のサイズに応じて分割する - src_num = vram_max & ~1; // 減らして偶数にする（元が奇数なら分割数が増えるかも） - i = (source_num - src_off + src_num - 1) / src_num; // 何回に分けて処理するか - src_num = (source_num - src_off + i - 1) / i; // 一度に処理する量を平均化する - src_num = (src_num + 1) & ~1; // 増やして偶数にする - } + th2->size = 0; // 計算前の状態にしておく (th->size は既に 0 になってる) + cpu_end = gpu_end = 0; #ifdef TIMER - printf("remain = %d, src_off = %d, src_num = %d\n", source_num - src_off, src_off, src_num); + printf("remain = %d, src_off = %d, src_max = %d\n", source_num - src_off, src_off, src_max); #endif while (src_off < source_num){ - // ソース・ブロックを何個ずつ処理するか - if (src_off + src_num > source_num){ - src_num = source_num - src_off; -#ifdef TIMER - printf("last1: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } else if (src_off + src_num * 2 - 1 >= source_num){ - src_num = source_num - src_off; - if (src_num > vram_max){ // VRAM のサイズまでにする - src_num = (src_num + 1) / 2; // 半分にする - src_num = (src_num + 1) & ~1; // 偶数にする + // GPUスレッドと CPUスレッドのどちらかが待機中になるまで待つ + do { + th_act = 0; + // CPUスレッドの動作状況を調べる + if (WaitForMultipleObjects(cpu_num2 - 1, hEnd, TRUE, 0) == WAIT_TIMEOUT){ + th_act |= 1; // CPUスレッドが動作中 + } else if (th->size > 0){ // CPUスレッドの計算量を加算する + prog_num += th->size * block_lost; + th->size = 0; } -#ifdef TIMER - printf("last2: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } - - // GPU と CPU がスレッドごとに消失ブロックを計算する - th->buf = buf + (size_t)unit_size * src_off; - th->mat = mat + src_off; - th->size = src_num; - th->now = -1; // 初期値 - 1 - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを再開する - for (j = 0; j < cpu_num; j++){ - ResetEvent(hEnd[j]); // リセットしておく - SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる - } - - // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待つ - while (WaitForMultipleObjects(cpu_num, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ - // th-now が最高値なので、計算が終わってるのは th-now + 1 - cpu_num 個となる - j = th->now + 1 - cpu_num; - if (j < 0) - j = 0; - j /= chunk_num; // chunk数で割ってブロック数にする - // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） - if (print_progress((int)(((prog_num + src_num * j) * 1000) / prog_base))){ - err = 2; - goto error_end; + // GPUスレッドの動作状況を調べる + if (WaitForSingleObject(hEnd[cpu_num2 - 1], 0) == WAIT_TIMEOUT){ + th_act |= 2; // GPUスレッドが動作中 + } else if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * block_lost; + th2->size = 0; } - time_last = GetTickCount(); - } - if (th->len != 0){ // エラー発生 - i = th->len; - printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); - err = 1; - goto error_end; + if (th_act == 3){ // 両方が動作中なら + // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, FALSE, UPDATE_TIME) == WAIT_TIMEOUT){ + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + } + } while (th_act == 3); + + // どちらかのスレッドで消失ブロックを計算する + if ((th_act & 1) == 0){ // CPUスレッドを優先的に開始する + src_num = src_max; // 一度に処理するソース・ブロックの数を制限する + if (src_off + src_num * 2 - 1 >= source_num){ + src_num = source_num - src_off; +#ifdef TIMER + printf("CPU last: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } + cpu_end += src_num; + th->buf = buf + (size_t)unit_size * src_off; + th->mat = mat + src_off; + th->size = src_num; + th->now = -1; // CPUスレッドの初期値 - 1 + //_mm_sfence(); + for (j = 0; j < cpu_num2 - 1; j++){ + ResetEvent(hEnd[j]); // リセットしておく + SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる + } + } else { // CPUスレッドが動作中なら、GPUスレッドを開始する + src_num = (source_num - src_off) * gpu_end / (cpu_end + gpu_end); // 残りブロック数に対する割合 + if (src_num < src_max){ + if (gpu_end / src_max < (cpu_end / src_max) / 2){ // GPU が遅い場合は最低負担量も減らす + if (gpu_end < cpu_end / 4){ + if (src_num < src_max / 4) + src_num = src_max / 4; + } else if (src_num < src_max / 2){ + src_num = src_max / 2; + } + } else { + src_num = src_max; // 最低でも CPUスレッドと同じ量を担当する + } + } + if (src_num > vram_max) + src_num = vram_max; + if (src_off + src_num > source_num){ + src_num = source_num - src_off; +#ifdef TIMER + printf("GPU last 1: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } else if (src_off + src_num + src_max > source_num){ + src_num = source_num - src_off - src_max; +#ifdef TIMER + printf("GPU last 2: src_off = %d, src_num = %d\n", src_off, src_num); + } else { + printf("GPU: remain = %d, src_off = %d, src_num = %d\n", source_num - src_off, src_off, src_num); +#endif + } + gpu_end += src_num; + th2->buf = buf + (size_t)unit_size * src_off; + th2->mat = mat + src_off; + th2->size = src_num; + th2->now = -1; // GPUスレッドの初期値 - 1 + //_mm_sfence(); + ResetEvent(hEnd[cpu_num2 - 1]); // リセットしておく + SetEvent(hRun[cpu_num2 - 1]); // サブ・スレッドに計算を開始させる } // 経過表示 - prog_num += src_num * block_lost; if (GetTickCount() - time_last >= UPDATE_TIME){ - if (print_progress((int)((prog_num * 1000) / prog_base))){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now がGPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ err = 2; goto error_end; } @@ -1724,6 +1775,50 @@ skip_count++; src_off += src_num; } + // 全スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * block_lost; + } + if (th->size > 0) // CPUスレッドの計算量を加算する + prog_num += th->size * block_lost; + #ifdef TIMER time_start = GetTickCount(); #endif @@ -1738,6 +1833,8 @@ time_start = GetTickCount(); } //printf(" lost block[%d] = source block[%d]\n", i, recv_now); + // CPUスレッドと GPUスレッドの計算結果を合わせる + galois_align_xor(g_buf + (size_t)unit_size * i, work_buf, unit_size); // 復元されたソース・ブロックのチェックサムを検証する checksum16_return(work_buf, hash, io_size); if (memcmp(work_buf + io_size, hash, HASH_SIZE) != 0){ @@ -1817,7 +1914,8 @@ if (prog_num != prog_base) error_end: InterlockedExchange(&(th->now), INT_MAX / 2); // サブ・スレッドの計算を中断する - for (j = 0; j < cpu_num; j++){ + InterlockedExchange(&(th2->now), INT_MAX / 2); + for (j = 0; j < cpu_num2; j++){ if (hSub[j]){ // サブ・スレッドを終了させる SetEvent(hRun[j]); WaitForSingleObject(hSub[j], INFINITE); @@ -1843,31 +1941,33 @@ int decode_method5( // 復元するブロックだけ保持する場合 (GPU対 parity_ctx_r *p_blk, // パリティ・ブロックの情報 unsigned short *mat) { - unsigned char *buf = NULL, *p_buf, *work_buf, *hash; + unsigned char *buf = NULL, *p_buf, *g_buf, *work_buf, *hash; unsigned short *id; int err = 0, i, j, last_file, chunk_num, recv_now; int source_off, read_num, parity_now; - int cpu_num1, src_off, src_num, src_max, vram_max; + int cpu_num1, src_off, src_num, src_max; + int cpu_num2, vram_max, cpu_end, gpu_end, th_act; unsigned int unit_size, len; unsigned int time_last, prog_read, prog_write; __int64 file_off, prog_num = 0, prog_base; HANDLE hFile = NULL; HANDLE hSub[MAX_CPU], hRun[MAX_CPU], hEnd[MAX_CPU]; - RS_TH th[1]; + RS_TH th[1], th2[1]; memset(hSub, 0, sizeof(HANDLE) * MAX_CPU); id = mat + (block_lost * source_num); // 何番目の消失ソース・ブロックがどのパリティで代替されるか unit_size = (block_size + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1); // MEM_UNIT の倍数にする // 作業バッファーを確保する - read_num = read_block_num(block_lost, 1, MEM_UNIT); // ソース・ブロックを何個読み込むか + // CPU計算スレッドと GPU計算スレッドで保存先を別けるので、消失ブロック分を２倍確保する + read_num = read_block_num(block_lost * 2, 1, MEM_UNIT); // ソース・ブロックを何個読み込むか if (read_num == 0){ //printf("cannot keep enough blocks, use another method\n"); return -4; // スライスを分割して処理しないと無理 } //read_num = (read_num + 1) / 2 + 1; // 2分割の実験用 //read_num = (read_num + 2) / 3 + 1; // 3分割の実験用 - file_off = (read_num + block_lost) * (size_t)unit_size + HASH_SIZE; + file_off = (read_num + block_lost * 2) * (size_t)unit_size + HASH_SIZE; buf = _aligned_malloc((size_t)file_off, MEM_UNIT); // GPU 用の境界 if (buf == NULL){ printf("malloc, %I64d\n", file_off); @@ -1875,41 +1975,35 @@ int decode_method5( // 復元するブロックだけ保持する場合 (GPU対 goto error_end; } p_buf = buf + (size_t)unit_size * read_num; // パリティ・ブロックを記録する領域 - hash = p_buf + (size_t)unit_size * block_lost; + g_buf = p_buf + (size_t)unit_size * block_lost; // GPUスレッド用 + hash = g_buf + (size_t)unit_size * block_lost; prog_read = (block_lost + 31) / 32; // 読み書きの経過をそれぞれ 3% ぐらいにする prog_write = (source_num + 31) / 32; prog_base = (__int64)(source_num + prog_write) * block_lost + prog_read * source_num; // ブロックの合計掛け算個数 + 書き込み回数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > block_lost) - cpu_num1 = block_lost; + cpu_num1 = calc_thread_num2(block_lost, &cpu_num2); // 使用するスレッド数を調節する src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num <= 2)) - src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない + if ((src_max < CACHE_MIN_NUM) || (src_max > CACHE_MAX_NUM)) + src_max = CACHE_MAX_NUM; // 不明または極端な場合は、規定値にする + //cpu_num1 = 0; // 2nd decode の実験用に 1st decode を停止する #ifdef TIMER printf("\n read some blocks, and keep all recovering blocks (GPU)\n"); printf("buffer size = %I64d MB, read_num = %d, round = %d\n", file_off >> 20, read_num, (source_num + read_num - 1) / read_num); - printf("cache: limit size = %d, chunk_size = %d, split = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); - printf("unit_size = %d, cpu_num1 = %d, src_max = %d\n", unit_size, cpu_num1, src_max); + printf("cache: limit size = %d, chunk_size = %d, chunk_num = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); + printf("unit_size = %d, cpu_num1 = %d, cpu_num2 = %d\n", unit_size, cpu_num1, cpu_num2); #endif // OpenCL の初期化 vram_max = read_num; // 読み込める分だけにする - i = init_OpenCL(unit_size, len, &vram_max); + i = init_OpenCL(unit_size, &vram_max); if (i != 0){ if (i != 3) // GPU が見つからなかった場合はエラー表示しない printf("init_OpenCL, %d, %d\n", i & 0xFF, i >> 8); i = free_OpenCL(); if (i != 0) printf("free_OpenCL, %d, %d", i & 0xFF, i >> 8); - OpenCL_method = 0; // GPU を使わない設定にする - // GPU を使わずに計算を続行する場合は以下をコメントアウト + OpenCL_method = 0; // GPU を使えなかった印 err = -3; // CPU だけの方式に切り替える goto error_end; } @@ -1919,10 +2013,14 @@ int decode_method5( // 復元するブロックだけ保持する場合 (GPU対 // マルチ・スレッドの準備をする th->buf = p_buf; + th2->buf = g_buf; th->size = unit_size; + th2->size = unit_size; th->count = block_lost; - th->len = len; // chunk size - for (j = 0; j < cpu_num; j++){ // サブ・スレッドごとに + th2->count = block_lost; + th->len = len ; // chunk size + th2->len = 0; // GPUのエラー通知用にする + for (j = 0; j < cpu_num2; j++){ // サブ・スレッドごとに hRun[j] = CreateEvent(NULL, FALSE, FALSE, NULL); // Auto Reset にする if (hRun[j] == NULL){ print_win32_err(); @@ -1939,12 +2037,13 @@ int decode_method5( // 復元するブロックだけ保持する場合 (GPU対 goto error_end; } // サブ・スレッドを起動する - th->run = hRun[j]; - th->end = hEnd[j]; - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを起動する - if ((j == cpu_num - 1) && (OpenCL_method != 0)){ // 最後のスレッドを GPU 管理用にする - hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode_gpu, (LPVOID)th, 0, NULL); + if (j == cpu_num2 - 1){ // 最後のスレッドを GPU 管理用にする + th2->run = hRun[j]; + th2->end = hEnd[j]; + hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode_gpu, (LPVOID)th2, 0, NULL); } else { + th->run = hRun[j]; + th->end = hEnd[j]; hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_decode3, (LPVOID)th, 0, NULL); } if (hSub[j] == NULL){ @@ -1957,7 +2056,6 @@ int decode_method5( // 復元するブロックだけ保持する場合 (GPU対 } WaitForSingleObject(hEnd[j], INFINITE); // 設定終了の合図を待つ (リセットしない) } - th->len = 0; // GPUのエラー通知用にする // 何回かに別けてブロックを読み込んで、消失ブロックを少しずつ復元する print_progress_text(0, "Recovering slice"); @@ -2086,6 +2184,8 @@ read_count++; time_read += GetTickCount() - time_start; #endif + if (source_off == 0) + memset(g_buf, 0, (size_t)unit_size * block_lost); // 待機中に GPU用の領域をゼロ埋めしておく WaitForMultipleObjects(cpu_num1, hEnd, TRUE, INFINITE); // サブ・スレッドの計算終了の合図を待つ src_off += 1; // 計算を開始するソース・ブロックの番号 if (src_off == 0) // 1st decode しなかった場合（src_off = 0）は、消失ブロックをゼロ埋めする @@ -2094,75 +2194,151 @@ time_read += GetTickCount() - time_start; j = (src_off - source_off) * 1000 / read_num; printf("partial decode = %d / %d (%d.%d%%), source_off = %d, read = %d\n", src_off - source_off, read_num, j / 10, j % 10, source_off, read_count); #endif + recv_now = -1; // 消失ブロックの本来のソース番号 last_file = -1; - - // GPU と CPU のどちらに最適化するかが難しい + th2->size = 0; // 計算前の状態にしておく (th->size は既に 0 になってる) + cpu_end = gpu_end = 0; src_off -= source_off; // バッファー内でのソース・ブロックの位置にする - src_num = src_max; // 一度に処理するソース・ブロックの数を制限する - if (src_num > vram_max){ // VRAM に収まらない場合は、VRAM のサイズに応じて分割する - src_num = vram_max & ~1; // 減らして偶数にする（元が奇数なら分割数が増えるかも） - i = (read_num - src_off + src_num - 1) / src_num; // 何回に分けて処理するか - src_num = (read_num - src_off + i - 1) / i; // 一度に処理する量を平均化する - src_num = (src_num + 1) & ~1; // 増やして偶数にする - } #ifdef TIMER - printf("remain = %d, src_off = %d, src_num = %d\n", read_num - src_off, src_off, src_num); + printf("remain = %d, src_off = %d, src_max = %d\n", read_num - src_off, src_off, src_max); #endif while (src_off < read_num){ - // ソース・ブロックを何個ずつ処理するか - if (src_off + src_num > read_num){ - src_num = read_num - src_off; -#ifdef TIMER - printf("last1: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } else if (src_off + src_num * 2 - 1 >= read_num){ - src_num = read_num - src_off; - if (src_num > vram_max){ // VRAM のサイズまでにする - src_num = (src_num + 1) / 2; // 半分にする - src_num = (src_num + 1) & ~1; // 偶数にする + // GPUスレッドと CPUスレッドのどちらかが待機中になるまで待つ + do { + th_act = 0; + // CPUスレッドの動作状況を調べる + if (WaitForMultipleObjects(cpu_num2 - 1, hEnd, TRUE, 0) == WAIT_TIMEOUT){ + th_act |= 1; // CPUスレッドが動作中 + } else if (th->size > 0){ // CPUスレッドの計算量を加算する + prog_num += th->size * block_lost; + th->size = 0; } -#ifdef TIMER - printf("last2: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } - - // GPU と CPU がスレッドごとに消失ブロックを計算する - th->buf = buf + (size_t)unit_size * src_off; - th->mat = mat + (source_off + src_off); // ソース・ブロックの番号にする - th->size = src_num; - th->now = -1; // 初期値 - 1 - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを再開する - for (j = 0; j < cpu_num; j++){ - ResetEvent(hEnd[j]); // リセットしておく - SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる - } - - // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待つ - while (WaitForMultipleObjects(cpu_num, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ - // th-now が最高値なので、計算が終わってるのは th-now + 1 - cpu_num 個となる - j = th->now + 1 - cpu_num; - if (j < 0) - j = 0; - j /= chunk_num; // chunk数で割ってブロック数にする - // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） - if (print_progress((int)(((prog_num + src_num * j) * 1000) / prog_base))){ - err = 2; - goto error_end; + // GPUスレッドの動作状況を調べる + if (WaitForSingleObject(hEnd[cpu_num2 - 1], 0) == WAIT_TIMEOUT){ + th_act |= 2; // GPUスレッドが動作中 + } else if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * block_lost; + th2->size = 0; } - time_last = GetTickCount(); - } - if (th->len != 0){ // エラー発生 - i = th->len; - printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); - err = 1; - goto error_end; + if (th_act == 3){ // 両方が動作中なら + // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, FALSE, UPDATE_TIME) == WAIT_TIMEOUT){ + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + } + } while (th_act == 3); + + // どちらかのスレッドで消失ブロックを計算する + if ((th_act & 1) == 0){ // CPUスレッドを優先的に開始する + src_num = src_max; // 一度に処理するソース・ブロックの数を制限する + if (src_off + src_num * 2 - 1 >= read_num){ + src_num = read_num - src_off; +#ifdef TIMER + printf("CPU last: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } + cpu_end += src_num; + th->buf = buf + (size_t)unit_size * src_off; + th->mat = mat + (source_off + src_off); // ソース・ブロックの番号にする + th->size = src_num; + th->now = -1; // CPUスレッドの初期値 - 1 + //_mm_sfence(); + for (j = 0; j < cpu_num2 - 1; j++){ + ResetEvent(hEnd[j]); // リセットしておく + SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる + } + } else { // CPUスレッドが動作中なら、GPUスレッドを開始する + src_num = (read_num - src_off) * gpu_end / (cpu_end + gpu_end); // 残りブロック数に対する割合 + if (src_num < src_max){ + if (gpu_end / src_max < (cpu_end / src_max) / 2){ // GPU が遅い場合は最低負担量も減らす + if (gpu_end < cpu_end / 4){ + if (src_num < src_max / 4) + src_num = src_max / 4; + } else if (src_num < src_max / 2){ + src_num = src_max / 2; + } + } else { + src_num = src_max; // 最低でも CPUスレッドと同じ量を担当する + } + } + if (src_num > vram_max) + src_num = vram_max; + if (src_off + src_num > read_num){ + src_num = read_num - src_off; +#ifdef TIMER + printf("GPU last 1: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } else if (src_off + src_num + src_max > read_num){ + src_num = read_num - src_off - src_max; +#ifdef TIMER + printf("GPU last 2: src_off = %d, src_num = %d\n", src_off, src_num); + } else { + printf("GPU: remain = %d, src_off = %d, src_num = %d\n", read_num - src_off, src_off, src_num); +#endif + } + gpu_end += src_num; + th2->buf = buf + (size_t)unit_size * src_off; + th2->mat = mat + (source_off + src_off); // ソース・ブロックの番号にする + th2->size = src_num; + th2->now = -1; // GPUスレッドの初期値 - 1 + //_mm_sfence(); + ResetEvent(hEnd[cpu_num2 - 1]); // リセットしておく + SetEvent(hRun[cpu_num2 - 1]); // サブ・スレッドに計算を開始させる } // 経過表示 - prog_num += src_num * block_lost; if (GetTickCount() - time_last >= UPDATE_TIME){ - if (print_progress((int)((prog_num * 1000) / prog_base))){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now がGPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ err = 2; goto error_end; } @@ -2172,6 +2348,50 @@ time_read += GetTickCount() - time_start; src_off += src_num; } + // 全スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * block_lost; + } + if (th->size > 0) // CPUスレッドの計算量を加算する + prog_num += th->size * block_lost; + source_off += read_num; } @@ -2189,6 +2409,8 @@ time_start = GetTickCount(); } //printf(" lost block[%d] = source block[%d]\n", i, recv_now); + // CPUスレッドと GPUスレッドの計算結果を合わせる + galois_align_xor(g_buf + (size_t)unit_size * i, work_buf, unit_size); // 復元されたソース・ブロックのチェックサムを検証する checksum16_return(work_buf, hash, unit_size - HASH_SIZE); if (memcmp(work_buf + unit_size - HASH_SIZE, hash, HASH_SIZE) != 0){ @@ -2252,7 +2474,8 @@ if (prog_num != prog_base) error_end: InterlockedExchange(&(th->now), INT_MAX / 2); // サブ・スレッドの計算を中断する - for (j = 0; j < cpu_num; j++){ + InterlockedExchange(&(th2->now), INT_MAX / 2); + for (j = 0; j < cpu_num2; j++){ if (hSub[j]){ // サブ・スレッドを終了させる SetEvent(hRun[j]); WaitForSingleObject(hSub[j], INFINITE); diff --git a/source/par2j/rs_encode.c b/source/par2j/rs_encode.c index b538c89..c83ee10 100644 --- a/source/par2j/rs_encode.c +++ b/source/par2j/rs_encode.c @@ -1,5 +1,5 @@ // rs_encode.c -// Copyright : 2023-09-21 Yutaka Sawada +// Copyright : 2023-10-22 Yutaka Sawada // License : GPL #ifndef _UNICODE @@ -295,11 +295,11 @@ printf(" 2nd encode %d.%03d sec, %d loop, %d MB/s\n", time_encode2b / 1000, time // GPU 対応のサブ・スレッド (最後のスレッドなので、1st encode では呼ばれない) static DWORD WINAPI thread_encode_gpu(LPVOID lpParameter) { - unsigned char *s_buf, *p_buf; + unsigned char *s_buf, *g_buf; unsigned short *constant, *factor; - int i, j, max_num, chunk_num; + int i, j; int src_off, src_num; - unsigned int unit_size, len, off, chunk_size; + unsigned int unit_size; HANDLE hRun, hEnd; RS_TH *th; #ifdef TIMER @@ -308,17 +308,14 @@ unsigned int time_start2, time_encode2 = 0, loop_count2 = 0; th = (RS_TH *)lpParameter; constant = th->mat; - p_buf = th->buf; + g_buf = th->buf; unit_size = th->size; - chunk_size = th->len; hRun = th->run; hEnd = th->end; //_mm_sfence(); SetEvent(hEnd); // 設定完了を通知する factor = constant + source_num; - chunk_num = (unit_size + chunk_size - 1) / chunk_size; - max_num = chunk_num * parity_num; WaitForSingleObject(hRun, INFINITE); // 計算開始の合図を待つ while (th->now < INT_MAX / 2){ @@ -337,27 +334,19 @@ time_start2 = GetTickCount(); InterlockedExchange(&(th->now), INT_MAX / 3); // サブ・スレッドの計算を中断する } - // スレッドごとに作成するパリティ・ブロックの chunk を変える - len = chunk_size; - while ((j = InterlockedIncrement(&(th->now))) < max_num){ // j = ++th_now - off = j / parity_num; // chunk の番号 - j = j % parity_num; // parity の番号 - off *= chunk_size; // chunk の位置 - if (off + len > unit_size) - len = unit_size - off; // 最後の chunk だけサイズが異なるかも - + // 一つの GPUスレッドが全てのパリティ・ブロックを処理する + while ((j = InterlockedIncrement(&(th->now))) < parity_num){ // j = ++th_now // factor は定数行列の乗数になる for (i = 0; i < src_num; i++) factor[i] = galois_power(constant[src_off + i], first_num + j); // VRAM上のソース・ブロックごとにパリティを追加していく - i = gpu_multiply_chunks(src_num, factor, p_buf + (size_t)unit_size * j + off, off, len); + i = gpu_multiply_blocks(src_num, factor, g_buf + (size_t)unit_size * j, unit_size); if (i != 0){ th->len = i; InterlockedExchange(&(th->now), INT_MAX / 3); // サブ・スレッドの計算を中断する break; } - #ifdef TIMER loop_count2 += src_num; #endif @@ -375,7 +364,6 @@ time_encode2 += GetTickCount() - time_start2; WaitForSingleObject(hRun, INFINITE); // 計算開始の合図を待つ } #ifdef TIMER -loop_count2 /= chunk_num; // chunk数で割ってブロック数にする printf("gpu-thread :\n"); if (time_encode2 > 0){ i = (int)((__int64)loop_count2 * unit_size * 125 / ((__int64)time_encode2 * 131072)); @@ -664,16 +652,9 @@ int encode_method2( // ソース・データを全て読み込む場合 len = try_cache_blocking(unit_size); //len = ((len + 2) / 3 + (sse_unit - 1)) & ~(sse_unit - 1); // 1/3の実験用 chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > part_num) - cpu_num1 = part_num; + cpu_num1 = calc_thread_num1(part_num); // 読み込み中はスレッド数を減らす src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num == 1)) + if ((src_max < CACHE_MIN_NUM) || (cpu_num == 1)) src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない #ifdef TIMER printf("\n read all source blocks, and keep some parity blocks\n"); @@ -1145,16 +1126,9 @@ int encode_method3( // パリティ・ブロックを全て保持して、一度 prog_base = (__int64)(source_num + prog_write) * parity_num + prog_read * source_num; // ブロックの合計掛け算個数 + 読み書き回数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > parity_num) - cpu_num1 = parity_num; + cpu_num1 = calc_thread_num1(parity_num); // 読み込み中はスレッド数を減らす src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num == 1)) + if ((src_max < CACHE_MIN_NUM) || (cpu_num == 1)) src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない #ifdef TIMER printf("\n read some source blocks, and keep all parity blocks\n"); @@ -1424,7 +1398,7 @@ time_start = GetTickCount(); memcpy(common_buf + common_size, common_buf, common_size); // 後の半分に前半のをコピーする // 最後にパリティ・ブロックのチェックサムを検証して、リカバリ・ファイルに書き込む err = create_recovery_file_1pass(file_path, recovery_path, packet_limit, block_distri, - packet_num, common_buf, common_size, footer_buf, footer_size, rcv_hFile, p_buf, unit_size); + packet_num, common_buf, common_size, footer_buf, footer_size, rcv_hFile, p_buf, NULL, unit_size); #ifdef TIMER time_write = GetTickCount() - time_start; #endif @@ -1463,26 +1437,28 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G parity_ctx_c *p_blk, // パリティ・ブロックの情報 unsigned short *constant) // 複数ブロック分の領域を確保しておく？ { - unsigned char *buf = NULL, *p_buf, *work_buf, *hash; + unsigned char *buf = NULL, *p_buf, *g_buf, *work_buf, *hash; int err = 0, i, j, last_file, chunk_num; - int cpu_num1, src_off, src_num, src_max, vram_max; + int cpu_num1, src_off, src_num, src_max; + int cpu_num2, vram_max, cpu_end, gpu_end, th_act; unsigned int io_size, unit_size, len, block_off; unsigned int time_last, prog_read, prog_write; __int64 file_off, prog_num = 0, prog_base; HANDLE hFile = NULL; HANDLE hSub[MAX_CPU], hRun[MAX_CPU], hEnd[MAX_CPU]; - RS_TH th[1]; + RS_TH th[1], th2[1]; PHMD5 md_ctx, *md_ptr = NULL; memset(hSub, 0, sizeof(HANDLE) * MAX_CPU); // 作業バッファーを確保する // part_num を使わず、全てのブロックを保持する所がencode_method2と異なることに注意！ - io_size = get_io_size(source_num + parity_num, NULL, 1, MEM_UNIT); + // CPU計算スレッドと GPU計算スレッドで保存先を別けるので、パリティ・ブロック分を２倍確保する + io_size = get_io_size(source_num + parity_num * 2, NULL, 1, MEM_UNIT); //io_size = (((io_size + 1) / 2 + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1)) - HASH_SIZE; // 2分割の実験用 //io_size = (((io_size + 2) / 3 + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1)) - HASH_SIZE; // 3分割の実験用 unit_size = io_size + HASH_SIZE; // チェックサムの分だけ増やす - file_off = (source_num + parity_num) * (size_t)unit_size + HASH_SIZE; + file_off = (source_num + parity_num * 2) * (size_t)unit_size + HASH_SIZE; buf = _aligned_malloc((size_t)file_off, MEM_UNIT); // GPU 用の境界 if (buf == NULL){ printf("malloc, %I64d\n", file_off); @@ -1490,30 +1466,24 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G goto error_end; } p_buf = buf + (size_t)unit_size * source_num; // パリティ・ブロックを記録する領域 - hash = p_buf + (size_t)unit_size * parity_num; + g_buf = p_buf + (size_t)unit_size * parity_num; // GPUスレッド用 + hash = g_buf + (size_t)unit_size * parity_num; prog_base = (block_size + io_size - 1) / io_size; prog_read = (parity_num + 31) / 32; // 読み書きの経過をそれぞれ 3% ぐらいにする prog_write = (source_num + 31) / 32; prog_base *= (__int64)(source_num + prog_write) * parity_num + prog_read * source_num; // 全体の断片の個数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > parity_num) - cpu_num1 = parity_num; - //cpu_num1 = 0; // 2nd encode の実験用に 1st encode を停止する + cpu_num1 = calc_thread_num2(parity_num, &cpu_num2); // 使用するスレッド数を調節する src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num <= 2)) - src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない + if ((src_max < CACHE_MIN_NUM) || (src_max > CACHE_MAX_NUM)) + src_max = CACHE_MAX_NUM; // 不明または極端な場合は、規定値にする + //cpu_num1 = 0; // 2nd encode の実験用に 1st encode を停止する #ifdef TIMER printf("\n read all source blocks, and keep all parity blocks (GPU)\n"); printf("buffer size = %I64d MB, io_size = %d, split = %d\n", file_off >> 20, io_size, (block_size + io_size - 1) / io_size); printf("cache: limit size = %d, chunk_size = %d, chunk_num = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); - printf("unit_size = %d, cpu_num1 = %d, src_max = %d\n", unit_size, cpu_num1, src_max); + printf("unit_size = %d, cpu_num1 = %d, cpu_num2 = %d\n", unit_size, cpu_num1, cpu_num2); #endif if (io_size < block_size){ // スライスが分割される場合だけ、途中までのハッシュ値を保持する @@ -1534,15 +1504,14 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G // OpenCL の初期化 vram_max = source_num; - i = init_OpenCL(unit_size, len, &vram_max); + i = init_OpenCL(unit_size, &vram_max); if (i != 0){ if (i != 3) // GPU が見つからなかった場合はエラー表示しない printf("init_OpenCL, %d, %d\n", i & 0xFF, i >> 8); i = free_OpenCL(); if (i != 0) printf("free_OpenCL, %d, %d", i & 0xFF, i >> 8); - OpenCL_method = 0; // GPU を使わない設定にする - // GPU を使わずに計算を続行する場合は以下をコメントアウト + OpenCL_method = 0; // GPU を使えなかった印 err = -2; // CPU だけの方式に切り替える goto error_end; } @@ -1552,10 +1521,14 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G // マルチ・スレッドの準備をする th->mat = constant; + th2->mat = constant; th->buf = p_buf; + th2->buf = g_buf; th->size = unit_size; + th2->size = unit_size; th->len = len; // chunk size - for (j = 0; j < cpu_num; j++){ // サブ・スレッドごとに + th2->len = 0; // GPUのエラー通知用にする + for (j = 0; j < cpu_num2; j++){ // サブ・スレッドごとに hRun[j] = CreateEvent(NULL, FALSE, FALSE, NULL); // Auto Reset にする if (hRun[j] == NULL){ print_win32_err(); @@ -1572,12 +1545,13 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G goto error_end; } // サブ・スレッドを起動する - th->run = hRun[j]; - th->end = hEnd[j]; - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを起動する - if ((j == cpu_num - 1) && (OpenCL_method != 0)){ // 最後のスレッドを GPU 管理用にする - hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode_gpu, (LPVOID)th, 0, NULL); + if (j == cpu_num2 - 1){ // 最後のスレッドを GPU 管理用にする + th2->run = hRun[j]; + th2->end = hEnd[j]; + hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode_gpu, (LPVOID)th2, 0, NULL); } else { + th->run = hRun[j]; + th->end = hEnd[j]; hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode3, (LPVOID)th, 0, NULL); } if (hSub[j] == NULL){ @@ -1590,7 +1564,6 @@ int encode_method4( // 全てのブロックを断片的に保持する場合 (G } WaitForSingleObject(hEnd[j], INFINITE); // 設定終了の合図を待つ (リセットしない) } - th->len = 0; // GPUのエラー通知用にする // ソース・ブロック断片を読み込んで、パリティ・ブロック断片を作成する time_last = GetTickCount(); @@ -1708,6 +1681,7 @@ skip_count++; time_read += GetTickCount() - time_start; #endif + memset(g_buf, 0, (size_t)unit_size * parity_num); // 待機中に GPU用の領域をゼロ埋めしておく WaitForMultipleObjects(cpu_num1, hEnd, TRUE, INFINITE); // サブ・スレッドの計算終了の合図を待つ src_off += 1; // 計算を開始するソース・ブロックの番号 if (src_off > 0){ @@ -1733,72 +1707,150 @@ skip_count++; len = io_size; } - // GPU と CPU のどちらに最適化するかが難しい - src_num = src_max; // 一度に処理するソース・ブロックの数を制限する - if (src_num > vram_max){ // VRAM に収まらない場合は、VRAM のサイズに応じて分割する - src_num = vram_max & ~1; // 減らして偶数にする（元が奇数なら分割数が増えるかも） - i = (source_num - src_off + src_num - 1) / src_num; // 何回に分けて処理するか - src_num = (source_num - src_off + i - 1) / i; // 一度に処理する量を平均化する - src_num = (src_num + 1) & ~1; // 増やして偶数にする - } + th2->size = 0; // 計算前の状態にしておく (th->size は既に 0 になってる) + cpu_end = gpu_end = 0; #ifdef TIMER - printf("remain = %d, src_off = %d, src_num = %d\n", source_num - src_off, src_off, src_num); + printf("remain = %d, src_off = %d, src_max = %d\n", source_num - src_off, src_off, src_max); #endif while (src_off < source_num){ - // ソース・ブロックを何個ずつ処理するか - if (src_off + src_num > source_num){ - src_num = source_num - src_off; -#ifdef TIMER - printf("last1: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } else if (src_off + src_num * 2 - 1 >= source_num){ - src_num = source_num - src_off; - if (src_num > vram_max){ // VRAM のサイズまでにする - src_num = (src_num + 1) / 2; // 半分にする - src_num = (src_num + 1) & ~1; // 偶数にする + // GPUスレッドと CPUスレッドのどちらかが待機中になるまで待つ + do { + th_act = 0; + // CPUスレッドの動作状況を調べる + if (WaitForMultipleObjects(cpu_num2 - 1, hEnd, TRUE, 0) == WAIT_TIMEOUT){ + th_act |= 1; // CPUスレッドが動作中 + } else if (th->size > 0){ // CPUスレッドの計算量を加算する + prog_num += th->size * parity_num; + th->size = 0; } -#ifdef TIMER - printf("last2: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } - - // GPU と CPU がスレッドごとにパリティ・ブロックを計算する - th->buf = buf + (size_t)unit_size * src_off; - th->off = src_off; // ソース・ブロックの番号にする - th->size = src_num; - th->now = -1; // 初期値 - 1 - //_mm_sfence(); - //for (j = cpu_num - 1; j >= 0; j--){ // GPU から先に計算を開始する？ - for (j = 0; j < cpu_num; j++){ - ResetEvent(hEnd[j]); // リセットしておく - SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる - } - - // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する - while (WaitForMultipleObjects(cpu_num, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ - // th-now が最高値なので、計算が終わってるのは th-now + 1 - cpu_num 個となる - j = th->now + 1 - cpu_num; - if (j < 0) - j = 0; - j /= chunk_num; // chunk数で割ってブロック数にする - // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） - if (print_progress((int)(((prog_num + src_num * j) * 1000) / prog_base))){ - err = 2; - goto error_end; + // GPUスレッドの動作状況を調べる + if (WaitForSingleObject(hEnd[cpu_num2 - 1], 0) == WAIT_TIMEOUT){ + th_act |= 2; // GPUスレッドが動作中 + } else if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * parity_num; + th2->size = 0; } - time_last = GetTickCount(); - } - if (th->len != 0){ // エラー発生 - i = th->len; - printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); - err = 1; - goto error_end; + if (th_act == 3){ // 両方が動作中なら + //if (th_act == 1){ // CPUスレッドだけが動作中か調べる実験 + //if (th_act == 2){ // GPUスレッドだけが動作中か調べる実験 + // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, FALSE, UPDATE_TIME) == WAIT_TIMEOUT){ + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + } + } while (th_act == 3); + //} while (th_act == 1); // CPUスレッドの終了だけを待つ実験 + //} while (th_act == 2); // GPUスレッドの終了だけを待つ実験 + + // どちらかのスレッドでパリティ・ブロックを計算する + if ((th_act & 1) == 0){ // CPUスレッドを優先的に開始する + src_num = src_max; // 一度に処理するソース・ブロックの数を制限する + if (src_off + src_num * 2 - 1 >= source_num){ + src_num = source_num - src_off; +#ifdef TIMER + printf("CPU last: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } + cpu_end += src_num; + th->buf = buf + (size_t)unit_size * src_off; + th->off = src_off; // ソース・ブロックの番号にする + th->size = src_num; + th->now = -1; // CPUスレッドの初期値 - 1 + //_mm_sfence(); + for (j = 0; j < cpu_num2 - 1; j++){ + ResetEvent(hEnd[j]); // リセットしておく + SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる + } + } else { // CPUスレッドが動作中なら、GPUスレッドを開始する + src_num = (source_num - src_off) * gpu_end / (cpu_end + gpu_end); // 残りブロック数に対する割合 + if (src_num < src_max) + if (gpu_end / src_max < (cpu_end / src_max) / 2){ // GPU が遅い場合は最低負担量も減らす + if (gpu_end < cpu_end / 4){ + if (src_num < src_max / 4) + src_num = src_max / 4; + } else if (src_num < src_max / 2){ + src_num = src_max / 2; + } + } else { + src_num = src_max; // 最低でも CPUスレッドと同じ量を担当する + } + if (src_num > vram_max) + src_num = vram_max; + if (src_off + src_num > source_num){ + src_num = source_num - src_off; +#ifdef TIMER + printf("GPU last 1: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } else if (src_off + src_num + src_max > source_num){ + src_num = source_num - src_off - src_max; +#ifdef TIMER + printf("GPU last 2: src_off = %d, src_num = %d\n", src_off, src_num); + } else { + printf("GPU: remain = %d, src_off = %d, src_num = %d\n", source_num - src_off, src_off, src_num); +#endif + } + gpu_end += src_num; + th2->buf = buf + (size_t)unit_size * src_off; + th2->off = src_off; // ソース・ブロックの番号にする + th2->size = src_num; + th2->now = -1; // GPUスレッドの初期値 - 1 + //_mm_sfence(); + ResetEvent(hEnd[cpu_num2 - 1]); // リセットしておく + SetEvent(hRun[cpu_num2 - 1]); // サブ・スレッドに計算を開始させる } // 経過表示 - prog_num += src_num * parity_num; if (GetTickCount() - time_last >= UPDATE_TIME){ - if (print_progress((int)((prog_num * 1000) / prog_base))){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now がGPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ err = 2; goto error_end; } @@ -1808,12 +1860,58 @@ skip_count++; src_off += src_num; } + // 全スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * parity_num; + } + if (th->size > 0) // CPUスレッドの計算量を加算する + prog_num += th->size * parity_num; + #ifdef TIMER time_start = GetTickCount(); #endif // パリティ・ブロックを書き込む work_buf = p_buf; for (i = 0; i < parity_num; i++){ + // CPUスレッドと GPUスレッドの計算結果を合わせる + galois_align_xor(g_buf + (size_t)unit_size * i, work_buf, unit_size); // パリティ・ブロックのチェックサムを検証する checksum16_return(work_buf, hash, io_size); if (memcmp(work_buf + io_size, hash, HASH_SIZE) != 0){ @@ -1931,7 +2029,8 @@ if (prog_num != prog_base) error_end: InterlockedExchange(&(th->now), INT_MAX / 2); // サブ・スレッドの計算を中断する - for (j = 0; j < cpu_num; j++){ + InterlockedExchange(&(th2->now), INT_MAX / 2); + for (j = 0; j < cpu_num2; j++){ if (hSub[j]){ // サブ・スレッドを終了させる SetEvent(hRun[j]); WaitForSingleObject(hSub[j], INFINITE); @@ -1965,24 +2064,26 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ source_ctx_c *s_blk, // ソース・ブロックの情報 unsigned short *constant) { - unsigned char *buf = NULL, *p_buf; + unsigned char *buf = NULL, *p_buf, *g_buf; int err = 0, i, j, last_file, chunk_num; int source_off, read_num, packet_off; - int cpu_num1, src_off, src_num, src_max, vram_max; + int cpu_num1, src_off, src_num, src_max; + int cpu_num2, vram_max, cpu_end, gpu_end, th_act; unsigned int unit_size, len; unsigned int time_last, prog_read, prog_write; __int64 prog_num = 0, prog_base; size_t mem_size; HANDLE hFile = NULL; HANDLE hSub[MAX_CPU], hRun[MAX_CPU], hEnd[MAX_CPU]; - RS_TH th[1]; + RS_TH th[1], th2[1]; PHMD5 file_md_ctx, blk_md_ctx; memset(hSub, 0, sizeof(HANDLE) * MAX_CPU); unit_size = (block_size + HASH_SIZE + (MEM_UNIT - 1)) & ~(MEM_UNIT - 1); // MEM_UNIT の倍数にする // 作業バッファーを確保する - read_num = read_block_num(parity_num, 1, MEM_UNIT); // ソース・ブロックを何個読み込むか + // CPU計算スレッドと GPU計算スレッドで保存先を別けるので、パリティ・ブロック分を２倍確保する + read_num = read_block_num(parity_num * 2, 1, MEM_UNIT); // ソース・ブロックを何個読み込むか if (read_num == 0){ #ifdef TIMER printf("cannot keep enough blocks, use another method\n"); @@ -1991,7 +2092,7 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ } //read_num = (read_num + 1) / 2 + 1; // 2分割の実験用 //read_num = (read_num + 2) / 3 + 1; // 3分割の実験用 - mem_size = (size_t)(read_num + parity_num) * unit_size; + mem_size = (size_t)(read_num + parity_num * 2) * unit_size; buf = _aligned_malloc(mem_size, MEM_UNIT); // GPU 用の境界 if (buf == NULL){ printf("malloc, %Id\n", mem_size); @@ -1999,40 +2100,34 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ goto error_end; } p_buf = buf + (size_t)unit_size * read_num; // パリティ・ブロックを記録する領域 + g_buf = p_buf + (size_t)unit_size * parity_num; // GPUスレッド用 prog_read = (parity_num + 31) / 32; // 読み書きの経過をそれぞれ 3% ぐらいにする prog_write = (source_num + 31) / 32; prog_base = (__int64)(source_num + prog_write) * parity_num + prog_read * source_num; // ブロックの合計掛け算個数 + 書き込み回数 len = try_cache_blocking(unit_size); chunk_num = (unit_size + len - 1) / len; - cpu_num1 = 0; // 読み込み中はスレッド数を減らす（シングル・スレッドの時は 0にする） - i = 1; - while (i * 2 <= cpu_num){ // 1=0, 2~3=1, 4~7=2, 8~15=3, 16~31=4, 32=5 - cpu_num1++; - i *= 2; - } - if (cpu_num1 > parity_num) - cpu_num1 = parity_num; + cpu_num1 = calc_thread_num2(parity_num, &cpu_num2); // 使用するスレッド数を調節する src_max = cpu_cache & 0xFFFE; // CPU cache 最適化のため、同時に処理するブロック数を制限する - if ((src_max < 8) || (cpu_num <= 2)) - src_max = 0x8000; // 不明または少な過ぎる場合は、制限しない + if ((src_max < CACHE_MIN_NUM) || (src_max > CACHE_MAX_NUM)) + src_max = CACHE_MAX_NUM; // 不明または極端な場合は、規定値にする + //cpu_num1 = 0; // 2nd encode の実験用に 1st encode を停止する #ifdef TIMER printf("\n read some source blocks, and keep all parity blocks (GPU)\n"); printf("buffer size = %Id MB, read_num = %d, round = %d\n", mem_size >> 20, read_num, (source_num + read_num - 1) / read_num); printf("cache: limit size = %d, chunk_size = %d, chunk_num = %d\n", cpu_flag & 0x7FFF0000, len, chunk_num); - printf("unit_size = %d, cpu_num1 = %d, src_max = %d\n", unit_size, cpu_num1, src_max); + printf("unit_size = %d, cpu_num1 = %d, cpu_num2 = %d\n", unit_size, cpu_num1, cpu_num2); #endif // OpenCL の初期化 vram_max = read_num; // 読み込める分だけにする - i = init_OpenCL(unit_size, len, &vram_max); + i = init_OpenCL(unit_size, &vram_max); if (i != 0){ if (i != 3) // GPU が見つからなかった場合はエラー表示しない printf("init_OpenCL, %d, %d\n", i & 0xFF, i >> 8); i = free_OpenCL(); if (i != 0) printf("free_OpenCL, %d, %d", i & 0xFF, i >> 8); - OpenCL_method = 0; // GPU を使わない設定にする - // GPU を使わずに計算を続行する場合は以下をコメントアウト + OpenCL_method = 0; // GPU を使えなかった印 err = -3; // CPU だけの方式に切り替える goto error_end; } @@ -2043,10 +2138,14 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ // マルチ・スレッドの準備をする th->mat = constant; + th2->mat = constant; th->buf = p_buf; + th2->buf = g_buf; th->size = unit_size; + th2->size = unit_size; th->len = len; // chunk size - for (j = 0; j < cpu_num; j++){ // サブ・スレッドごとに + th2->len = 0; // GPUのエラー通知用にする + for (j = 0; j < cpu_num2; j++){ // サブ・スレッドごとに hRun[j] = CreateEvent(NULL, FALSE, FALSE, NULL); // Auto Reset にする if (hRun[j] == NULL){ print_win32_err(); @@ -2063,12 +2162,13 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ goto error_end; } // サブ・スレッドを起動する - th->run = hRun[j]; - th->end = hEnd[j]; - //_mm_sfence(); // メモリーへの書き込みを完了してからスレッドを起動する - if ((j == cpu_num - 1) && (OpenCL_method != 0)){ // 最後のスレッドを GPU 管理用にする - hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode_gpu, (LPVOID)th, 0, NULL); + if (j == cpu_num2 - 1){ // 最後のスレッドを GPU 管理用にする + th2->run = hRun[j]; + th2->end = hEnd[j]; + hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode_gpu, (LPVOID)th2, 0, NULL); } else { + th->run = hRun[j]; + th->end = hEnd[j]; hSub[j] = (HANDLE)_beginthreadex(NULL, STACK_SIZE, thread_encode3, (LPVOID)th, 0, NULL); } if (hSub[j] == NULL){ @@ -2081,7 +2181,6 @@ int encode_method5( // ソース・ブロックの一部とパリティ・ブロ } WaitForSingleObject(hEnd[j], INFINITE); // 設定終了の合図を待つ (リセットしない) } - th->len = 0; // GPUのエラー通知用にする // 何回かに別けてソース・ブロックを読み込んで、パリティ・ブロックを少しずつ作成する time_last = GetTickCount(); @@ -2229,6 +2328,8 @@ time_start = GetTickCount(); time_read += GetTickCount() - time_start; #endif + if (source_off == 0) + memset(g_buf, 0, (size_t)unit_size * parity_num); // 待機中に GPU用の領域をゼロ埋めしておく WaitForMultipleObjects(cpu_num1, hEnd, TRUE, INFINITE); // サブ・スレッドの計算終了の合図を待つ src_off += 1; // 計算を開始するソース・ブロックの番号 if (src_off == 0) // 1st encode しなかった場合（src_off = 0）は、生成ブロックをゼロ埋めする @@ -2238,72 +2339,148 @@ time_read += GetTickCount() - time_start; printf("partial encode = %d / %d (%d.%d%%), source_off = %d\n", src_off - source_off, read_num, j / 10, j % 10, source_off); #endif - // GPU と CPU のどちらに最適化するかが難しい + th2->size = 0; // 計算前の状態にしておく (th->size は既に 0 になってる) + cpu_end = gpu_end = 0; src_off -= source_off; // バッファー内でのソース・ブロックの位置にする - src_num = src_max; // 一度に処理するソース・ブロックの数を制限する - if (src_num > vram_max){ // VRAM に収まらない場合は、VRAM のサイズに応じて分割する - src_num = vram_max & ~1; // 減らして偶数にする（元が奇数なら分割数が増えるかも） - i = (read_num - src_off + src_num - 1) / src_num; // 何回に分けて処理するか - src_num = (read_num - src_off + i - 1) / i; // 一度に処理する量を平均化する - src_num = (src_num + 1) & ~1; // 増やして偶数にする - } #ifdef TIMER - printf("remain = %d, src_off = %d, src_num = %d\n", read_num - src_off, src_off, src_num); + printf("remain = %d, src_off = %d, src_max = %d\n", read_num - src_off, src_off, src_max); #endif while (src_off < read_num){ - // ソース・ブロックを何個ずつ処理するか - if (src_off + src_num > read_num){ - src_num = read_num - src_off; -#ifdef TIMER - printf("last1: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } else if (src_off + src_num * 2 - 1 >= read_num){ - src_num = read_num - src_off; - if (src_num > vram_max){ // VRAM のサイズまでにする - src_num = (src_num + 1) / 2; // 半分にする - src_num = (src_num + 1) & ~1; // 偶数にする + // GPUスレッドと CPUスレッドのどちらかが待機中になるまで待つ + do { + th_act = 0; + // CPUスレッドの動作状況を調べる + if (WaitForMultipleObjects(cpu_num2 - 1, hEnd, TRUE, 0) == WAIT_TIMEOUT){ + th_act |= 1; // CPUスレッドが動作中 + } else if (th->size > 0){ // CPUスレッドの計算量を加算する + prog_num += th->size * parity_num; + th->size = 0; } -#ifdef TIMER - printf("last2: src_off = %d, src_num = %d\n", src_off, src_num); -#endif - } - - // GPU と CPU がスレッドごとにパリティ・ブロックを計算する - th->buf = buf + (size_t)unit_size * src_off; - th->off = source_off + src_off; // ソース・ブロックの番号にする - th->size = src_num; - th->now = -1; // 初期値 - 1 - //_mm_sfence(); - for (j = 0; j < cpu_num; j++){ - ResetEvent(hEnd[j]); // リセットしておく - SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる - } - - // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する - while (WaitForMultipleObjects(cpu_num, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ - // th-now が最高値なので、計算が終わってるのは th-now + 1 - cpu_num 個となる - j = th->now + 1 - cpu_num; - if (j < 0) - j = 0; - j /= chunk_num; // chunk数で割ってブロック数にする - // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） - if (print_progress((int)(((prog_num + src_num * j) * 1000) / prog_base))){ - err = 2; - goto error_end; + // GPUスレッドの動作状況を調べる + if (WaitForSingleObject(hEnd[cpu_num2 - 1], 0) == WAIT_TIMEOUT){ + th_act |= 2; // GPUスレッドが動作中 + } else if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * parity_num; + th2->size = 0; } - time_last = GetTickCount(); - } - if (th->len != 0){ // エラー発生 - i = th->len; - printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); - err = 1; - goto error_end; + if (th_act == 3){ // 両方が動作中なら + // サブ・スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, FALSE, UPDATE_TIME) == WAIT_TIMEOUT){ + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + } + } while (th_act == 3); + + // どちらかのスレッドでパリティ・ブロックを計算する + if ((th_act & 1) == 0){ // CPUスレッドを優先的に開始する + src_num = src_max; // 一度に処理するソース・ブロックの数を制限する + if (src_off + src_num * 2 - 1 >= read_num){ + src_num = read_num - src_off; +#ifdef TIMER + printf("lastC: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } + cpu_end += src_num; + th->buf = buf + (size_t)unit_size * src_off; + th->off = source_off + src_off; // ソース・ブロックの番号にする + th->size = src_num; + th->now = -1; // CPUスレッドの初期値 - 1 + //_mm_sfence(); + for (j = 0; j < cpu_num2 - 1; j++){ + ResetEvent(hEnd[j]); // リセットしておく + SetEvent(hRun[j]); // サブ・スレッドに計算を開始させる + } + } else { // CPUスレッドが動作中なら、GPUスレッドを開始する + src_num = (read_num - src_off) * gpu_end / (cpu_end + gpu_end); // 残りブロック数に対する割合 + if (src_num < src_max){ + if (gpu_end / src_max < (cpu_end / src_max) / 2){ // GPU が遅い場合は最低負担量も減らす + if (gpu_end < cpu_end / 4){ + if (src_num < src_max / 4) + src_num = src_max / 4; + } else if (src_num < src_max / 2){ + src_num = src_max / 2; + } + } else { + src_num = src_max; // 最低でも CPUスレッドと同じ量を担当する + } + } + if (src_num > vram_max) + src_num = vram_max; + if (src_off + src_num > read_num){ + src_num = read_num - src_off; +#ifdef TIMER + printf("GPU last 1: src_off = %d, src_num = %d\n", src_off, src_num); +#endif + } else if (src_off + src_num + src_max > read_num){ + src_num = read_num - src_off - src_max; +#ifdef TIMER + printf("GPU last 2: src_off = %d, src_num = %d\n", src_off, src_num); + } else { + printf("GPU: remain = %d, src_off = %d, src_num = %d\n", read_num - src_off, src_off, src_num); +#endif + } + gpu_end += src_num; + th2->buf = buf + (size_t)unit_size * src_off; + th2->off = source_off + src_off; // ソース・ブロックの番号にする + th2->size = src_num; + th2->now = -1; // GPUスレッドの初期値 - 1 + //_mm_sfence(); + ResetEvent(hEnd[cpu_num2 - 1]); // リセットしておく + SetEvent(hRun[cpu_num2 - 1]); // サブ・スレッドに計算を開始させる } // 経過表示 - prog_num += src_num * parity_num; if (GetTickCount() - time_last >= UPDATE_TIME){ - if (print_progress((int)((prog_num * 1000) / prog_base))){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now がGPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ err = 2; goto error_end; } @@ -2313,6 +2490,50 @@ time_read += GetTickCount() - time_start; src_off += src_num; } + // 全スレッドの計算終了の合図を UPDATE_TIME だけ待ちながら、経過表示する + while (WaitForMultipleObjects(cpu_num2, hEnd, TRUE, UPDATE_TIME) == WAIT_TIMEOUT){ + if (th2->size == 0){ + i = 0; + } else { + // th2-now が GPUスレッドの最高値なので、計算が終わってるのは th2-now 個となる + i = th2->now; + if (i < 0){ + i = 0; + } else { + i *= th2->size; + } + } + if (th->size == 0){ + j = 0; + } else { + // th-now が CPUスレッドの最高値なので、計算が終わってるのは th-now + 2 - cpu_num2 個となる + j = th->now + 2 - cpu_num2; + if (j < 0){ + j = 0; + } else { + j /= chunk_num; // chunk数で割ってブロック数にする + j *= th->size; + } + } + // 経過表示（UPDATE_TIME 時間待った場合なので、必ず経過してるはず） + if (print_progress((int)(((prog_num + i + j) * 1000) / prog_base))){ + err = 2; + goto error_end; + } + time_last = GetTickCount(); + } + if (th2->size > 0){ // GPUスレッドの計算量を加算する + if (th2->len != 0){ // エラー発生 + i = th2->len; + printf("error, gpu-thread, %d, %d\n", i & 0xFF, i >> 8); + err = 1; + goto error_end; + } + prog_num += th2->size * parity_num; + } + if (th->size > 0) // CPUスレッドの計算量を加算する + prog_num += th->size * parity_num; + source_off += read_num; } @@ -2322,7 +2543,7 @@ time_start = GetTickCount(); memcpy(common_buf + common_size, common_buf, common_size); // 後の半分に前半のをコピーする // 最後にパリティ・ブロックのチェックサムを検証して、リカバリ・ファイルに書き込む err = create_recovery_file_1pass(file_path, recovery_path, packet_limit, block_distri, - packet_num, common_buf, common_size, footer_buf, footer_size, rcv_hFile, p_buf, unit_size); + packet_num, common_buf, common_size, footer_buf, footer_size, rcv_hFile, p_buf, g_buf, unit_size); #ifdef TIMER time_write = GetTickCount() - time_start; #endif @@ -2337,7 +2558,8 @@ if (prog_num != prog_base - prog_write * parity_num) error_end: InterlockedExchange(&(th->now), INT_MAX / 2); // サブ・スレッドの計算を中断する - for (j = 0; j < cpu_num; j++){ + InterlockedExchange(&(th2->now), INT_MAX / 2); + for (j = 0; j < cpu_num2; j++){ if (hSub[j]){ // サブ・スレッドを終了させる SetEvent(hRun[j]); WaitForSingleObject(hSub[j], INFINITE); diff --git a/source/par2j/source.cl b/source/par2j/source.cl index 0769e85..e5acc46 100644 --- a/source/par2j/source.cl +++ b/source/par2j/source.cl @@ -18,9 +18,7 @@ __kernel void method1( __global uint *src, __global uint *dst, __global ushort *factors, - int blk_num, - int offset, - int length) + int blk_num) { __local uint mtab[512]; int i, blk; @@ -29,15 +27,14 @@ __kernel void method1( const int work_size = get_global_size(0); const int table_id = get_local_id(0); - src += offset; - for (i = work_id; i < length; i += work_size) + for (i = work_id; i < BLK_SIZE; i += work_size) dst[i] = 0; for (blk = 0; blk < blk_num; blk++){ calc_table(mtab, table_id, factors[blk]); barrier(CLK_LOCAL_MEM_FENCE); - for (i = work_id; i < length; i += work_size){ + for (i = work_id; i < BLK_SIZE; i += work_size){ v = src[i]; sum = mtab[(uchar)(v >> 16)] ^ mtab[256 + (v >> 24)]; sum <<= 16; @@ -53,9 +50,7 @@ __kernel void method2( __global uint *src, __global uint *dst, __global ushort *factors, - int blk_num, - int offset, - int length) + int blk_num) { __local uint mtab[512]; int i, blk, pos; @@ -64,8 +59,7 @@ __kernel void method2( const int work_size = get_global_size(0) * 2; const int table_id = get_local_id(0); - src += offset; - for (i = work_id; i < length; i += work_size){ + for (i = work_id; i < BLK_SIZE; i += work_size){ dst[i ] = 0; dst[i + 1] = 0; } @@ -74,7 +68,7 @@ __kernel void method2( calc_table(mtab, table_id, factors[blk]); barrier(CLK_LOCAL_MEM_FENCE); - for (i = work_id; i < length; i += work_size){ + for (i = work_id; i < BLK_SIZE; i += work_size){ pos = (i & ~7) + ((i & 7) >> 1); lo = src[pos ]; hi = src[pos + 4]; @@ -96,9 +90,7 @@ __kernel void method4( __global uint *src, __global uint *dst, __global ushort *factors, - int blk_num, - int offset, - int length) + int blk_num) { __local int table[16]; __local uint cache[256]; @@ -107,8 +99,7 @@ __kernel void method4( const int work_id = get_global_id(0); const int work_size = get_global_size(0); - src += offset; - for (i = work_id; i < length; i += work_size) + for (i = work_id; i < BLK_SIZE; i += work_size) dst[i] = 0; for (blk = 0; blk < blk_num; blk++){ @@ -122,7 +113,7 @@ __kernel void method4( } barrier(CLK_LOCAL_MEM_FENCE); - for (i = work_id; i < length; i += work_size){ + for (i = work_id; i < BLK_SIZE; i += work_size){ pos = i & 255; cache[pos] = src[i]; barrier(CLK_LOCAL_MEM_FENCE); diff --git a/source/par2j/version.h b/source/par2j/version.h index a7db363..5558576 100644 --- a/source/par2j/version.h +++ b/source/par2j/version.h @@ -1,2 +1,2 @@ -#define FILE_VERSION "1.3.3.0" // ファイルのバージョン番号 +#define FILE_VERSION "1.3.3.1" // ファイルのバージョン番号 #define PRODUCT_VERSION "1.3.3" // 製品のバージョン番号