diff --git a/encoder/basisu_astc_hdr_6x6_enc.cpp b/encoder/basisu_astc_hdr_6x6_enc.cpp index 684982bb..7372af2c 100644 --- a/encoder/basisu_astc_hdr_6x6_enc.cpp +++ b/encoder/basisu_astc_hdr_6x6_enc.cpp @@ -6717,6 +6717,8 @@ bool compress_photo(const basisu::imagef &orig_src_img, const astc_hdr_6x6_globa interval_timer tm; tm.start(); + job_pool::token token{0}; + std::atomic_bool any_failed_flag; any_failed_flag.store(false); @@ -6745,14 +6747,14 @@ bool compress_photo(const basisu::imagef &orig_src_img, const astc_hdr_6x6_globa any_failed_flag.store(true, std::memory_order_relaxed); } } - } ); + }, &token); if (any_failed_flag) break; } // strip_index - pJob_pool->wait_for_all(); + pJob_pool->wait_for_all(&token); if (any_failed_flag) { diff --git a/encoder/basisu_astc_ldr_encode.cpp b/encoder/basisu_astc_ldr_encode.cpp index 220f432a..e0cf018a 100644 --- a/encoder/basisu_astc_ldr_encode.cpp +++ b/encoder/basisu_astc_ldr_encode.cpp @@ -5603,6 +5603,8 @@ bool ldr_astc_block_encode_image( if (enc_cfg.m_debug_output) fmt_debug_printf("ASTC packing superpass: {}\n", 1 + superpass_index); + job_pool::token token{0}; + uint32_t total_blocks_done = 0; float last_printed_progress_val = -100.0f; @@ -6650,7 +6652,7 @@ bool ldr_astc_block_encode_image( } // if (superpass_index == ...) - }); + }, &token); if (encoder_failed_flag) break; @@ -6662,13 +6664,7 @@ bool ldr_astc_block_encode_image( } // by - if (encoder_failed_flag) - { - fmt_error_printf("Main compressor block loop failed!\n"); - return false; - } - - job_pool.wait_for_all(); + job_pool.wait_for_all(&token); if (encoder_failed_flag) { diff --git a/encoder/basisu_comp.cpp b/encoder/basisu_comp.cpp index acbedc31..e2d096c3 100644 --- a/encoder/basisu_comp.cpp +++ b/encoder/basisu_comp.cpp @@ -1083,6 +1083,8 @@ namespace basisu const uint32_t num_blocks_y = tex.get_blocks_y(); const uint32_t total_blocks = tex.get_total_blocks(); const imagef& source_image = m_slice_images_hdr[slice_index]; + + job_pool::token token{0}; const uint32_t N = 256; for (uint32_t block_index_iter = 0; block_index_iter < total_blocks; block_index_iter += N) @@ -1232,11 +1234,11 @@ namespace basisu debug_printf("basis_compressor::encode_slices_to_uastc_4x4_hdr: %3.1f%% done\n", percent_done); } - }); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); if (any_failures) return cECFailedEncodeUASTC; @@ -1500,6 +1502,8 @@ namespace basisu uint32_t total_blocks_processed = 0; float last_percentage_printed = 0; + job_pool::token token{0}; + const uint32_t N = 256; for (uint32_t block_index_iter = 0; block_index_iter < total_blocks; block_index_iter += N) { @@ -1555,11 +1559,11 @@ namespace basisu debug_printf("basis_compressor::encode_slices_to_uastc_4x4_ldr: %3.1f%% done\n", percent_done); } - }); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); if (m_params.m_rdo_uastc_ldr_4x4) { @@ -4845,7 +4849,7 @@ namespace basisu for (uint32_t pindex = 0; pindex < params_vec.size(); pindex++) { - jpool.add_job([pindex, ¶ms_vec, &results_vec, &result, &opencl_failed] { + jpool.add_job([pindex, ¶ms_vec, &results_vec, &result, &opencl_failed, &jpool] { basis_compressor_params params = params_vec[pindex]; parallel_results& results = results_vec[pindex]; @@ -4855,9 +4859,7 @@ namespace basisu basis_compressor c; - // Dummy job pool - job_pool task_jpool(1); - params.m_pJob_pool = &task_jpool; + params.m_pJob_pool = &jpool; // TODO: Remove this flag entirely params.m_multithreading = true; diff --git a/encoder/basisu_enc.cpp b/encoder/basisu_enc.cpp index e92b4b8e..ff98637f 100644 --- a/encoder/basisu_enc.cpp +++ b/encoder/basisu_enc.cpp @@ -2238,7 +2238,7 @@ namespace basisu } job_pool::job_pool(uint32_t num_threads) : - m_num_active_jobs(0) + m_num_pending_jobs(0) { m_kill_flag.store(false); m_num_active_workers.store(0); @@ -2285,66 +2285,83 @@ namespace basisu m_threads[i].join(); } - void job_pool::add_job(const std::function& job) + void job_pool::add_job(std::function job, token* tok) { - std::unique_lock lock(m_mutex); + { + std::unique_lock lock(m_mutex); - m_queue.emplace_back(job); + m_queue.push_back(item{ std::move(job), tok }); - const size_t queue_size = m_queue.size(); + if (tok) + (*tok)++; - lock.unlock(); + m_num_pending_jobs++; + } - if (queue_size > 1) - m_has_work.notify_one(); + m_has_work.notify_one(); } - void job_pool::add_job(std::function&& job) + void job_pool::wait_for_all(token* tok) { + token* wait_token = tok ? tok : &m_num_pending_jobs; + std::unique_lock lock(m_mutex); - m_queue.emplace_back(std::move(job)); - - const size_t queue_size = m_queue.size(); + while (true) + { + if (*wait_token == 0) + return; - lock.unlock(); + item job; + if (job_steal(job, tok, lock)) + { + job_run(job, lock); + continue; + } + + // Nothing stealable right now; wait until a job completes, then retry to see if we can steal freshly enqueued work +#ifndef __EMSCRIPTEN__ + m_job_done.wait(lock); +#else + // Avoid infinite blocking + m_job_done.wait_for(lock, std::chrono::milliseconds(50)); +#endif + } + } - if (queue_size > 1) + bool job_pool::job_steal(item& job, token* tok, std::unique_lock&) + { + for (size_t i = m_queue.size(); i > 0; --i) { - m_has_work.notify_one(); + item& victim = m_queue[i - 1]; + + if (tok == nullptr || victim.tok == tok) + { + job = std::move(victim); + victim = std::move(m_queue.back()); + m_queue.pop_back(); + + return true; + } } + + return false; } - void job_pool::wait_for_all() + void job_pool::job_run(item& job, std::unique_lock& lock) { - std::unique_lock lock(m_mutex); + lock.unlock(); - // Drain the job queue on the calling thread. - while (!m_queue.empty()) - { - std::function job(m_queue.back()); - m_queue.pop_back(); + job.fn(); - lock.unlock(); + lock.lock(); - job(); + if (job.tok) + (*job.tok)--; - lock.lock(); - } + m_num_pending_jobs--; - // The queue is empty, now wait for all active jobs to finish up. -#ifndef __EMSCRIPTEN__ - m_no_more_jobs.wait(lock, [this]{ return !m_num_active_jobs; } ); -#else - // Avoid infinite blocking - for (; ; ) - { - if (m_no_more_jobs.wait_for(lock, std::chrono::milliseconds(50), [this] { return !m_num_active_jobs; })) - { - break; - } - } -#endif + m_job_done.notify_all(); } void job_pool::job_thread(uint32_t index) @@ -2376,26 +2393,10 @@ namespace basisu continue; // Get the job and execute it. - std::function job(m_queue.back()); + item job = std::move(m_queue.back()); m_queue.pop_back(); - ++m_num_active_jobs; - - lock.unlock(); - - job(); - - lock.lock(); - - --m_num_active_jobs; - - // Now check if there are no more jobs remaining. - const bool all_done = m_queue.empty() && !m_num_active_jobs; - - lock.unlock(); - - if (all_done) - m_no_more_jobs.notify_all(); + job_run(job, lock); } m_num_active_workers.fetch_add(-1); diff --git a/encoder/basisu_enc.h b/encoder/basisu_enc.h index ebf54737..378cc302 100644 --- a/encoder/basisu_enc.h +++ b/encoder/basisu_enc.h @@ -804,31 +804,39 @@ namespace basisu BASISU_NO_EQUALS_OR_COPY_CONSTRUCT(job_pool); public: + using token = uint32_t; + // num_threads is the TOTAL number of job pool threads, including the calling thread! So 2=1 new thread, 3=2 new threads, etc. job_pool(uint32_t num_threads); ~job_pool(); - void add_job(const std::function& job); - void add_job(std::function&& job); - - void wait_for_all(); + void add_job(std::function job, token* tok = nullptr); + void wait_for_all(token* tok = nullptr); size_t get_total_threads() const { return 1 + m_threads.size(); } private: + struct item + { + std::function fn; + token* tok; + }; + std::vector m_threads; - std::vector > m_queue; + std::vector m_queue; std::mutex m_mutex; std::condition_variable m_has_work; - std::condition_variable m_no_more_jobs; - - uint32_t m_num_active_jobs; - - std::atomic m_kill_flag; + std::condition_variable m_job_done; + + uint32_t m_num_pending_jobs; + std::atomic m_kill_flag; std::atomic m_num_active_workers; + bool job_steal(item& job, token* tok, std::unique_lock& lock); + void job_run(item& job, std::unique_lock& lock); + void job_thread(uint32_t index); }; @@ -2076,6 +2084,8 @@ namespace basisu basisu::vector local_clusters[cMaxThreads]; basisu::vector local_parent_clusters[cMaxThreads]; + job_pool::token token{0}; + for (uint32_t thread_iter = 0; thread_iter < max_threads; thread_iter++) { pJob_pool->add_job( [thread_iter, &local_clusters, &local_parent_clusters, &success_flags, &quantizers, &initial_codebook, &q, &limit_clusterizers, &max_codebook_size, &max_threads, &max_parent_codebook_size] { @@ -2119,11 +2129,11 @@ namespace basisu } } - } ); + }, &token); } // thread_iter - pJob_pool->wait_for_all(); + pJob_pool->wait_for_all(&token); uint32_t total_clusters = 0, total_parent_clusters = 0; diff --git a/encoder/basisu_frontend.cpp b/encoder/basisu_frontend.cpp index d721b37d..d39c3601 100644 --- a/encoder/basisu_frontend.cpp +++ b/encoder/basisu_frontend.cpp @@ -348,6 +348,8 @@ namespace basisu m_block_selector_cluster_index.resize(m_total_blocks); #if 0 + job_pool::token token{0}; + for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { const uint32_t first_index = block_index_iter; @@ -386,11 +388,11 @@ namespace basisu m_block_selector_cluster_index[block_index] = best_index; } - }); + }, &token); } - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); m_encoded_blocks.resize(m_total_blocks); for (uint32_t block_index = 0; block_index < m_total_blocks; block_index++) @@ -413,6 +415,8 @@ namespace basisu { debug_printf("init_global_codebooks: pass %u\n", pass); + job_pool::token token{0}; + const uint32_t N = 128; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -467,11 +471,11 @@ namespace basisu } // block_index - }); + }, &token); } - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); m_endpoint_clusters.resize(0); m_endpoint_clusters.resize(endpoints.size()); @@ -484,6 +488,8 @@ namespace basisu m_block_selector_cluster_index.resize(m_total_blocks); + token = 0; + for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { const uint32_t first_index = block_index_iter; @@ -522,11 +528,11 @@ namespace basisu m_block_selector_cluster_index[block_index] = best_index; } - }); + }, &token); } - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); m_encoded_blocks.resize(m_total_blocks); for (uint32_t block_index = 0; block_index < m_total_blocks; block_index++) @@ -764,6 +770,8 @@ namespace basisu if (use_cpu) { + job_pool::token token{0}; + const uint32_t N = 4096; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -811,11 +819,11 @@ namespace basisu blk.set_selector(x, y, selectors[x + y * 4]); } - }); + }, &token); } - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); } // use_cpu @@ -830,6 +838,8 @@ namespace basisu training_vecs.resize(m_total_blocks * 2); + job_pool::token token{0}; + const uint32_t N = 16384; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -858,11 +868,11 @@ namespace basisu } // block_index; - } ); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); } void basisu_frontend::generate_endpoint_clusters() @@ -1007,6 +1017,8 @@ namespace basisu { m_subblock_endpoint_quant_err_vec.resize(0); + job_pool::token token{0}; + const uint32_t N = 512; for (uint32_t cluster_index_iter = 0; cluster_index_iter < m_endpoint_clusters.size(); cluster_index_iter += N) { @@ -1081,11 +1093,11 @@ namespace basisu } } // cluster_index - } ); + }, &token); } // cluster_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); vector_sort(m_subblock_endpoint_quant_err_vec); } @@ -1481,6 +1493,8 @@ namespace basisu if (use_cpu) { + job_pool::token token{0}; + const uint32_t N = 128; for (uint32_t cluster_index_iter = 0; cluster_index_iter < m_endpoint_clusters.size(); cluster_index_iter += N) { @@ -1606,11 +1620,11 @@ namespace basisu } // cluster_index - }); + }, &token); } // cluster_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); } debug_printf("Elapsed time: %3.3f secs\n", tm.get_elapsed_secs()); @@ -1771,6 +1785,8 @@ namespace basisu if (use_cpu) { + job_pool::token token{0}; + const uint32_t N = 1024; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -1908,11 +1924,11 @@ namespace basisu } // block_index - }); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); } // use_cpu @@ -2051,6 +2067,8 @@ namespace basisu if (use_cpu) { + job_pool::token token{0}; + const uint32_t N = 4096; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -2082,11 +2100,11 @@ namespace basisu } // block_index - }); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); } // use_cpu @@ -2145,6 +2163,8 @@ namespace basisu vec16F_clusterizer::array_of_weighted_training_vecs training_vecs(m_total_blocks); + job_pool::token token{0}; + const uint32_t N = 4096; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -2178,11 +2198,11 @@ namespace basisu } // block_index - } ); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); vec16F_clusterizer selector_clusterizer; for (uint32_t i = 0; i < m_total_blocks; i++) @@ -2270,6 +2290,8 @@ namespace basisu m_optimized_cluster_selectors.resize(total_selector_clusters); // For each selector codebook entry, and for each of the 4x4 selectors, determine which selector minimizes the error across all the blocks that use that quantized selector. + job_pool::token token{0}; + const uint32_t N = 256; for (uint32_t cluster_index_iter = 0; cluster_index_iter < total_selector_clusters; cluster_index_iter += N) { @@ -2345,11 +2367,11 @@ namespace basisu } // cluster_index - }); + }, &token); } // cluster_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); debug_printf("Elapsed time: %3.3f secs\n", tm.get_elapsed_secs()); @@ -2544,6 +2566,8 @@ namespace basisu } } + job_pool::token token{0}; + const uint32_t N = 2048; for (uint32_t block_index_iter = 0; block_index_iter < m_total_blocks; block_index_iter += N) { @@ -2689,11 +2713,11 @@ namespace basisu } // block_index - } ); + }, &token); } // block_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); for (uint32_t i = 0; i < m_selector_cluster_block_indices.size(); i++) { @@ -3004,6 +3028,8 @@ namespace basisu basisu::vector cluster_valid(new_endpoint_cluster_block_indices.size()); basisu::vector cluster_improved(new_endpoint_cluster_block_indices.size()); + job_pool::token token{0}; + const uint32_t N = 256; for (uint32_t cluster_index_iter = 0; cluster_index_iter < new_endpoint_cluster_block_indices.size(); cluster_index_iter += N) { @@ -3100,11 +3126,11 @@ namespace basisu } // cluster_index - } ); + }, &token); } // cluster_index_iter - m_params.m_pJob_pool->wait_for_all(); + m_params.m_pJob_pool->wait_for_all(&token); uint32_t total_unused_clusters = 0; uint32_t total_improved_clusters = 0; diff --git a/encoder/basisu_uastc_enc.cpp b/encoder/basisu_uastc_enc.cpp index 183b646e..97854c07 100644 --- a/encoder/basisu_uastc_enc.cpp +++ b/encoder/basisu_uastc_enc.cpp @@ -4125,6 +4125,8 @@ namespace basisu } else { + job_pool::token token{0}; + bool all_succeeded = true; for (uint32_t block_index_iter = 0; block_index_iter < num_blocks; block_index_iter += blocks_per_job) @@ -4148,12 +4150,11 @@ namespace basisu total_smooth += job_smooth; } - } - ); + }, &token); } // block_index_iter - pJob_pool->wait_for_all(); + pJob_pool->wait_for_all(&token); status = all_succeeded; }