commit 3eeb1485cfffd34bf2bf026862951e9a56d209d2 Author: David Sorber Date: Tue Feb 28 21:30:46 2017 -0500 Moved things around so that the main thread breaks off pieces of work and hands them to the threads. This version has less than half the runtime of the previous version for the Photos library test set! diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.cc b/software/photo_compress_archiver/PhotoCompressArchiver.cc index a08206b..e4e0f06 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.cc +++ b/software/photo_compress_archiver/PhotoCompressArchiver.cc @@ -14,6 +14,10 @@ x\ } +const uint32_t PTR_LEN = sizeof(char*); +const uint32_t BAILOUT_MAX = 600; // 30s in 50ms increments +const uint32_t FILES_PER_WORK_UNIT = 40; + const char *PROG_NAME = "./packjpg"; const char *PROG_OPT1 = "-np"; const char *PROG_OPT2 = "-o"; @@ -59,6 +63,21 @@ int PhotoCompressArchiver::execute() m_path, (m_decompress ? PJG_REGEX : JPEG_REGEX)); OUT(std::cout << "DONE" << std::endl;) + + // Start worker threads + OUT(std::cout << "Starting worker threads..." << std::flush;) + + uint32_t num_threads = m_num_cores; + for (uint32_t idx = 0; idx < num_threads; ++idx) + { + // Spawn worker thread and add to list for bookkeeping + std::thread* worker = new std::thread(&PhotoCompressArchiver::fork_worker, + this, idx); + m_worker_threads.push_back(worker); + } + OUT(std::cout << "DONE\n" << std::endl;) + + // Join the file finder thread p_finder_thread->join(); OUT(std::cout << "File finder completed:" << std::endl;) @@ -71,7 +90,7 @@ int PhotoCompressArchiver::execute() return -1; } - OUT(std::cout << " Found " << m_num_input_files << " images" << std::endl;) + OUT(std::cout << " Found " << m_num_input_files << " files" << std::endl;) if (! m_decompress) { @@ -79,42 +98,90 @@ int PhotoCompressArchiver::execute() << m_total_uncompressed_size << "\n" << std::endl;) } - // Start worker threads - OUT(std::cout << "Starting worker threads..." << std::flush;) - // TODO: need to handle the case were number of input files is less than - // the number of threads - uint32_t num_threads = m_num_cores; - uint32_t sublist_len = m_num_input_files / num_threads; - - for (uint32_t idx = 0; idx < num_threads; ++idx) + // Build work unit queue + uint32_t input_list_idx = 0; + uint32_t queue_size = 0; + while (input_list_idx < m_file_list.size()) { - // Subdivide the input file list (using the vector copy constructor - // and start/end iterators) - decltype(m_file_list.begin()) start_iter; - decltype(m_file_list.begin()) end_iter; - if (idx == (num_threads - 1)) + // Safely read the size of the queue { - // Make sure the last sublist goes all the way to the end of the - // input file list (if number of images is not an even multiple of - // number of threads) - start_iter = m_file_list.begin() + (idx * sublist_len); - end_iter = m_file_list.end(); + std::lock_guard lock(m_work_unit_mtx); + queue_size = m_work_unit_queue.size(); } - else + + // Wait until the queue has decreased in the size before proceeding, + // otherwise we could chew up a bunch of memory + if (queue_size > num_threads) { - start_iter = m_file_list.begin() + (idx * sublist_len); - end_iter = m_file_list.begin() + ((idx + 1) * sublist_len); + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + continue; } - auto sublist = new std::vector(start_iter, end_iter); + WorkUnit* work_unit = new WorkUnit(); + + work_unit->exec_argv.push_back(&PROG_NAME[2]); + work_unit->exec_argv.push_back(PROG_OPT1); + work_unit->exec_argv.push_back(PROG_OPT2); + work_unit->exec_argv.push_back(PROG_OPT3); + + uint32_t argv_size = sizeof(PROG_NAME) + sizeof(PROG_OPT1) + + sizeof(PROG_OPT2) + sizeof(PROG_OPT3) + ARGV_EXTRA; - // Spawn worker thread and add to list for bookkeeping - std::thread* worker = new std::thread(&PhotoCompressArchiver::fork_worker, - this, idx, sublist); - m_worker_threads.push_back(worker); + uint32_t local_count = 0; + while ((input_list_idx < m_file_list.size()) && + (local_count < FILES_PER_WORK_UNIT)) + { + bfs::path& file_path = m_file_list[input_list_idx]; + uint32_t path_len = file_path.string().size(); + + // Check if adding this next file would make the argv too big; if + // so bail out + if (argv_size + path_len + PTR_LEN > ARGV_MAX_SIZE) + { + break; + } + + // Add the file to the argv array + work_unit->exec_argv.push_back(file_path.string().c_str()); + argv_size += path_len + PTR_LEN; + + // Create new filename for expected + uint32_t end_pos = file_path.string().rfind("."); + std::string* filename = new std::string(file_path.string().begin(), + file_path.string().begin() + + end_pos); + + // Add extension depending on if decompress mode is enabled or not + (*filename) += (m_decompress ? ".jpg" : ".pjg"); + work_unit->output_filenames.push_back(filename); + + // Increment our index to the next file + ++input_list_idx; + ++local_count; + } + work_unit->exec_argv.push_back(nullptr); + + + // Add the create work unit to the queue; and notify the worker threads + // that it has been handed off + { + std::lock_guard lock(m_work_unit_mtx); + m_work_unit_queue.push_back(work_unit); + } + m_work_unit_cv.notify_all(); } - OUT(std::cout << "DONE\n" << std::endl;) + + // Add null terminate indicator for each thread + { + std::lock_guard lock(m_work_unit_mtx); + for (uint32_t idx = 0; idx < num_threads; ++idx) + { + + m_work_unit_queue.push_back(nullptr); + } + } + m_work_unit_cv.notify_all(); // Clean up the finder thread delete p_finder_thread; @@ -189,102 +256,55 @@ void PhotoCompressArchiver::find_files( } void PhotoCompressArchiver::fork_worker( - uint32_t tid, - std::vector* file_sublist) + uint32_t tid) { - uint32_t sublist_file_count = file_sublist->size(); - uint32_t files_processed = 0; - -#if 0 - // Print out the input files given to this worker - OUT(std::cout << "T[" << tid << "] top of the morning to ya: " - << sublist_file_count << std::endl;) - - for (auto& filepath : *file_sublist) - { - OUT(std::cout << "T[" << tid << "] file: " << filepath.string() - << std::endl;) - } - - return; // temporary for debugging -#endif - - uint32_t sublist_idx = 0; - const uint32_t ptr_len = sizeof(char*); + // Loop vars int status; + uint32_t work_unit_items = 0; + uint32_t bailout_ctr = 0; + uint32_t output_idx = 1; + int stat_rc = 0; + struct stat statbuf; + uint32_t files_processed = 0; + double complete_percent = 0.0; - // Vector of pointers that will become the char** argv passed to packjpg - std::vector exec_argv; - - // Build a list of the output files names for use in output file monitoring - // below. Output file names have ".jpg" or ".jpeg" replaced with ".pjg" - // in compression mode and vice versa in decompression mode. - std::vector output_filenames; - - while (sublist_idx < sublist_file_count) + while (true) { - // Pass in default program options for packjpg - exec_argv.clear(); - exec_argv.push_back(&PROG_NAME[2]); - exec_argv.push_back(PROG_OPT1); - exec_argv.push_back(PROG_OPT2); - exec_argv.push_back(PROG_OPT3); - - uint32_t argv_size = sizeof(PROG_NAME) + sizeof(PROG_OPT1) + - sizeof(PROG_OPT2) + sizeof(PROG_OPT3) + ARGV_EXTRA; - output_filenames.clear(); + std::unique_lock mux_lock(m_work_unit_mtx); + while (m_work_unit_queue.empty()) + { + m_work_unit_cv.wait_for(mux_lock, std::chrono::milliseconds(20)); + } + + // While holding the mutex grab the next chunk off of the add chunk + // queue + WorkUnit* work_unit = m_work_unit_queue.front(); + m_work_unit_queue.pop_front(); + mux_lock.unlock(); - uint32_t local_count = 0; - while ((sublist_idx < sublist_file_count) && (local_count < 40)) + if (work_unit == nullptr) { - bfs::path& file_path = file_sublist->at(sublist_idx); - uint32_t path_len = file_path.string().size(); - - // Check if adding this next file would make the argv too big; if - // so bail out - if (argv_size + path_len + ptr_len > ARGV_MAX_SIZE) - { - break; - } - - // Add the file to the argv array - exec_argv.push_back(file_path.string().c_str()); - argv_size += path_len + ptr_len; - - // Create new filename for expected - uint32_t end_pos = file_path.string().rfind("."); - std::string* filename = new std::string(file_path.string().begin(), - file_path.string().begin() - + end_pos); - - // Add extension depending on if decompress mode is enabled or not - (*filename) += (m_decompress ? ".jpg" : ".pjg"); - output_filenames.push_back(filename); - - - // Increment our index to the next file - ++sublist_idx; - ++local_count; + break; } + work_unit_items = work_unit->output_filenames.size(); + #if 0 // DEBUGGING uint32_t idx = 4; OUT( - for (auto outfile : output_filenames) + for (auto outfile : work_unit->output_filenames) { std::cout << "T[" << tid << "] output file: " << *outfile << "\n input file: " - << static_cast(exec_argv[idx++]) << std::endl; + << static_cast(work_unit->exec_argv[idx++]) + << std::endl; } ) - //~ continue; + continue; #endif - // Add a nullptr to the end of the argv array as a terminator - exec_argv.push_back(nullptr); - // Create a pipe to hold stdout from child process int filedes[2]; if (pipe(filedes) == -1) @@ -310,7 +330,7 @@ void PhotoCompressArchiver::fork_worker( close(filedes[1]); // Child after fork - int rc = execv(PROG_NAME, (char **)exec_argv.data()); + int rc = execv(PROG_NAME, (char **)work_unit->exec_argv.data()); if (rc) { std::cerr << "execv failed: " << errno << std::endl; @@ -319,22 +339,20 @@ void PhotoCompressArchiver::fork_worker( } // This is the parent process... - // This is a little silly... but it works. We know the order in which // the output files will be created so we wait until output files X + 1 // exists (i.e. we can stat() it) and then process file X. - const uint32_t BAILOUT_MAX = 600; // 30s in 50ms increments - uint32_t bailout_ctr = 0; - uint32_t output_idx = 1; - int stat_rc = 0; - struct stat statbuf; - double complete_percent = 0.0; - while (output_idx < (output_filenames.size())) + bailout_ctr = 0; + output_idx = 1; + stat_rc = 0; + files_processed = 0; + complete_percent = 0.0; + while (output_idx < work_unit_items) { - // Poll, waiting for output file X + 1 to exist + // Poll, waiting for output file X + 1 (aka "z") to exist bailout_ctr = 0; - while ((stat(output_filenames[output_idx]->c_str(), &statbuf) != 0) && - (++bailout_ctr < BAILOUT_MAX)) + const char* filename_z = work_unit->output_filenames[output_idx]->c_str(); + while ((stat(filename_z, &statbuf) != 0) && (++bailout_ctr < BAILOUT_MAX)) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); } @@ -345,8 +363,8 @@ void PhotoCompressArchiver::fork_worker( OUT(std::cerr << "T[" << tid << "] " << BOLD << RED << "ERROR: " << ENDC << "timed out while waiting " << "for output file: " - << *output_filenames[output_idx] << "; aborting" - << std::endl;) + << *work_unit->output_filenames[output_idx] + << "; aborting" << std::endl;) // The subprocess timed out... get its status waitpid(pid, &status, 0); @@ -357,13 +375,14 @@ void PhotoCompressArchiver::fork_worker( return; } - // Now stat output file X which should exist - stat_rc = stat(output_filenames[output_idx - 1]->c_str(), &statbuf); + // Now stat output file X (aka "y") which should exist + const char* filename_y = work_unit->output_filenames[output_idx - 1]->c_str(); + stat_rc = stat(filename_y, &statbuf); if (stat_rc) { OUT(std::cerr << "T[" << tid << "] " << BOLD << RED << "ERROR " - << ENDC << "while stating: " - << *output_filenames[output_idx - 1] << std::endl;) + << ENDC << "while stating: " << filename_y + << std::endl;) } // Increment counts @@ -374,20 +393,20 @@ void PhotoCompressArchiver::fork_worker( // Unless keeping original files, remove the source file if (! m_keep_orig) { - std::remove(file_sublist->at(output_idx - 1).string().c_str()); + std::remove(work_unit->exec_argv[4 + output_idx - 1]); } // Calculate the local percent done - complete_percent = (double)files_processed / sublist_file_count; + complete_percent = (double)files_processed / work_unit_items; complete_percent *= 100; OUT(std::cout << "T[" << tid << "] " << std::setw(6) << std::fixed << std::setprecision(2) << get_global_percent_done() - << "% -- (" << files_processed << "/" - << sublist_file_count << " -- " << std::setw(6) + << "% -- (" << std::setw(2) << files_processed << "/" + << work_unit_items << " -- " << std::setw(6) << complete_percent - << "%) file: " << *output_filenames[output_idx - 1] + << "%) file: " << filename_y << " -- " << statbuf.st_size << std::endl;) ++output_idx; @@ -400,12 +419,12 @@ void PhotoCompressArchiver::fork_worker( // Now that the forked process has completed, we can handle the last // output file - stat_rc = stat(output_filenames[output_idx - 1]->c_str(), &statbuf); + const char* filename_y = work_unit->output_filenames[output_idx - 1]->c_str(); + stat_rc = stat(filename_y, &statbuf); if (stat_rc) { OUT(std::cerr << "T[" << tid << "] " << BOLD << RED << "ERROR " - << ENDC << "while stating: " - << *output_filenames[output_idx - 1] << std::endl;) + << ENDC << "while stating: " << filename_y << std::endl;) } // Increment counts @@ -416,25 +435,26 @@ void PhotoCompressArchiver::fork_worker( // Unless keeping original files, remove the source file if (! m_keep_orig) { - std::remove(file_sublist->at(output_idx - 1).string().c_str()); + std::remove(work_unit->exec_argv[4 + output_idx - 1]); } // Calculate the local percent done - complete_percent = (double)files_processed / sublist_file_count; + complete_percent = (double)files_processed / work_unit_items; complete_percent *= 100; OUT(std::cout << "T[" << tid << "] " << std::setw(6) << std::fixed << std::setprecision(2) << get_global_percent_done() - << "% -- (" << files_processed << "/" << sublist_file_count - << " -- " << std::setw(5) << complete_percent - << "%) file: " << *output_filenames[output_idx - 1] + << "% -- (" << std::setw(2) << files_processed + << "/" << work_unit_items << " -- " << std::setw(5) + << complete_percent << "%) file: " << filename_y << " -- " << statbuf.st_size << std::endl;) // Free up the output filename list - for (auto output_filename : output_filenames) + for (auto output_filename : work_unit->output_filenames) { delete output_filename; } + delete work_unit; } OUT(std::cout << "T[" << tid << "] " << PURPLE << "complete ==> RC: " diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.hh b/software/photo_compress_archiver/PhotoCompressArchiver.hh index 7c4361f..3723d63 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.hh +++ b/software/photo_compress_archiver/PhotoCompressArchiver.hh @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -26,6 +27,12 @@ const std::string version("v0.1.0"); const boost::regex JPEG_REGEX("^.+\\.(jpg)|(jpeg)$", boost::regex::icase); const boost::regex PJG_REGEX("^.+\\.pjg$", boost::regex::icase); +struct work_unit_t { + std::vector exec_argv; + std::vector output_filenames; +}; +typedef struct work_unit_t WorkUnit; + class PhotoCompressArchiver { public: @@ -54,7 +61,7 @@ class PhotoCompressArchiver /** * Fork and wait */ - void fork_worker(uint32_t tid, std::vector* file_sublist); + void fork_worker(uint32_t tid); /** * Calculate the return the "global" percent done as determined by the @@ -81,6 +88,10 @@ class PhotoCompressArchiver std::thread* p_finder_thread; // file finder thread std::vector m_worker_threads; // worker threads + std::deque m_work_unit_queue; + std::mutex m_work_unit_mtx; + std::condition_variable m_work_unit_cv; + bool m_filter_empty; // is the filter regex empty boost::regex& m_filter_regex; // filter regular expression };