commit 537f40e69a749d9de483fe659d1454a2ddad15a2 Author: David Sorber Date: Sat Sep 8 19:20:39 2018 -0400 Continued improvements to PCA; handle errors better, make number of processing threads configurable, and reversed the "keep-orig" option to "delete-orig" which makes more sense. diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.cc b/software/photo_compress_archiver/PhotoCompressArchiver.cc index 441e1f5..bf8faf2 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.cc +++ b/software/photo_compress_archiver/PhotoCompressArchiver.cc @@ -13,15 +13,7 @@ #include "PhotoCompressArchiver.hh" -#define OUT(msg) { \ - std::lock_guard lock(m_output_mutex); \ - std::cout << msg << std::endl; \ -} - -#define ERROR(msg) { \ - std::lock_guard lock(m_output_mutex); \ - std::cerr << BOLD << RED << "ERROR: " << ENDC << msg << std::endl; \ -} +#define ERROR std::cerr << BOLD << RED << "ERROR: " << ENDC #define WRKR_OUT_REG(msg) { \ std::lock_guard lock(m_output_mutex); \ @@ -30,19 +22,20 @@ #define WRKR_OUT_ERR(msg) { \ std::lock_guard lock(m_output_mutex); \ - std::cerr << "T[" << std::setw(2) << tid << "] " << msg << std::endl; \ + std::cerr << "T[" << std::setw(2) << tid << "] " << BOLD << RED \ + << "ERROR: " << msg << std::endl; \ } PhotoCompressArchiver::PhotoCompressArchiver( const std::string& path, - uint32_t num_cores, + uint32_t num_threads, bool decompress, - bool keep_orig, + bool delete_orig, boost::regex& filter_regex) : m_path(path), - m_num_cores(num_cores), + m_num_threads(num_threads), m_decompress(decompress), - m_keep_orig(keep_orig), + m_delete_orig(delete_orig), m_files_processed(0), m_num_input_files(0), m_total_uncompressed_size(0), @@ -64,46 +57,45 @@ int PhotoCompressArchiver::execute() auto start = std::chrono::system_clock::now(); // Start the file finder thread - OUT("Starting file finder..." << std::flush); + std::cout << "Starting file finder..." << std::flush; p_finder_thread = new std::thread(&PhotoCompressArchiver::find_files, this, m_path, (m_decompress ? PJG_REGEX : JPEG_REGEX)); - OUT("DONE"); + std::cout << "DONE" << std::endl; // Start worker threads - OUT("Starting worker threads..." << std::flush); - - uint32_t num_threads = m_num_cores; - for (uint32_t idx = 0; idx < num_threads; ++idx) + std::cout << "Starting worker threads..." << std::flush; + for (uint32_t idx = 0; idx < m_num_threads; ++idx) { // Spawn worker thread and add to list for bookkeeping std::thread* worker = new std::thread(&PhotoCompressArchiver::worker_thread_body, this, idx); m_worker_threads.push_back(worker); } - OUT("DONE\n"); + std::cout << "DONE\n" << std::endl; // Join the file finder thread p_finder_thread->join(); - OUT("File finder completed:"); + std::cout << "File finder completed:" << std::endl; // Error out of no input files were found if (m_num_input_files == 0) { - ERROR("No matching input files found. Please check the patch and try " - "again."); + ERROR << "No matching input files found. Please check the patch and " + << "try again." << std::endl; return -1; } - OUT(" Found " << m_num_input_files << " files"); + std::cout << " Found " << m_num_input_files << " files" << std::endl; if (! m_decompress) { - OUT(" Total uncompressed bytes: " << m_total_uncompressed_size << "\n"); + std::cout << " Total uncompressed bytes: " + << m_total_uncompressed_size << "\n" << std::endl; } // Add terminator for each worker thread - for (uint32_t idx = 0; idx < num_threads; ++idx) + for (uint32_t idx = 0; idx < m_num_threads; ++idx) { m_file_list.push_back(nullptr); } @@ -122,23 +114,46 @@ int PhotoCompressArchiver::execute() // Grab start time and calculate duration auto end = std::chrono::system_clock::now(); auto diff = end - start; - double duration = std::chrono::duration(diff).count(); + double duration = std::chrono::duration(diff).count(); + + // Print out message if any errors were found + if (m_errors.size() > 0) + { + std::cout << BOLD << RED << "NOTE: errors were encountered processing " + << "the following " << m_errors.size() << " files:\n" << ENDC + << std::endl; + + WorkUnit* work_unit = nullptr; + for (uint32_t idx = 0; idx < m_errors.size(); ++idx) + { + work_unit = m_errors.pop_front(); + std::cout << " " << work_unit->m_path << std::endl; + delete work_unit; + } + std::cout << std::endl; + } // Print out compression information if compressing if (! m_decompress) { - OUT("\n Total uncompressed bytes: " << m_total_uncompressed_size); - OUT(" Total compressed bytes: " << m_total_compressed_size); + std::cout << "\n Total uncompressed bytes: " + << m_total_uncompressed_size << std::endl; + std::cout << " Total compressed bytes: " << m_total_compressed_size + << std::endl; // Calculate and display ratio double ratio = (double)m_total_compressed_size / m_total_uncompressed_size; - OUT(" Compression ratio: " << std::fixed << std::setprecision(4) - << (ratio * 100) << "%"); - - // Display execution duration - OUT(" Total time: " << duration << " ms\n"); + std::cout << " Compression ratio: " << std::fixed + << std::setprecision(4) << (ratio * 100) << "%" << std::endl; + } + else + { + std::cout << std::endl; } + // Display execution duration + std::cout << " Total time: " << duration << " s\n" << std::endl; + return 0; } @@ -215,43 +230,64 @@ void PhotoCompressArchiver::worker_thread_body(uint32_t tid) std::ios::binary | std::ios::in); if (! input_stream) { - WRKR_OUT_ERR("ERROR: unable to read from \"" << work_unit->m_path + WRKR_OUT_ERR("unable to read from \"" << work_unit->m_path << "\"\n") - // TODO: what now? + m_errors.push_back(work_unit); + continue; } input_stream.read((char*)file_buffer->data(), work_unit->m_file_size); input_stream.close(); - // TODO: revise this - // Determine input, and therefore output, filetype / extension + // Determine input, and therefore output, filetype / extension based + // on mode const std::string* output_file_extension = nullptr; - if ((file_buffer->at(0) == 0xFF) && (file_buffer->at(1) == 0xD8)) - { - output_file_extension = &PJG_EXTENSION; - } - else if ((file_buffer->at(0) == packJPG::pjg_magic[0]) && - (file_buffer->at(1) == packJPG::pjg_magic[1])) + if (m_decompress) { - output_file_extension = &JPG_EXTENSION; + // We are decompressing so the file we found had better be a PJG + if ((file_buffer->at(0) == packJPG::pjg_magic[0]) && + (file_buffer->at(1) == packJPG::pjg_magic[1])) + { + output_file_extension = &JPG_EXTENSION; + } + else + { + WRKR_OUT_ERR("the input file does not appear to be a valid " + "packjpg (.pjg) file even though its filename " + "suggests it is!\n") + m_errors.push_back(work_unit); + continue; + } } else { - WRKR_OUT_ERR("ERROR: Input file does not appear to be valid\n") - // TODO: what now? + // We are compressing so the file we found had better be a JPG + if ((file_buffer->at(0) == 0xFF) && (file_buffer->at(1) == 0xD8)) + { + output_file_extension = &PJG_EXTENSION; + } + else + { + WRKR_OUT_ERR("the input file does not appear to be a valid " + "JPEG (.jpg) file even though its filename " + "suggests it is!\n") + m_errors.push_back(work_unit); + continue; + } } + // Do the thing! instance->pjglib_init_streams(file_buffer->data(), 1, work_unit->m_file_size, nullptr, 1); - // Do the thing! bool rc = instance->pjglib_convert_stream2mem(&out_buffer, &out_size, message); if (!rc) { WRKR_OUT_ERR("An error occurred during the compression" << "/decompression operation: " << message << "\n") - // TODO: what now? + m_errors.push_back(work_unit); + continue; } //~ WRKR_OUT_REG("Status message: " << message) @@ -262,21 +298,28 @@ void PhotoCompressArchiver::worker_thread_body(uint32_t tid) ++m_files_processed; // Create output file path be replacing extension of input file - bfs::path out_path = bfs::path(work_unit->m_path).replace_extension(*output_file_extension); + bfs::path out_path = bfs::path(work_unit->m_path).replace_extension( + *output_file_extension); // Write output file std::ofstream output_stream(out_path.string(), std::ios::binary | std::ios::out); if (! output_stream) { - WRKR_OUT_ERR("ERROR: unable to read from \"" - << work_unit->m_path << "\"") - - // TODO: what now? + WRKR_OUT_ERR("unable to write to \"" << out_path << "\"") + m_errors.push_back(work_unit); + continue; } output_stream.write((const char*)out_buffer, out_size); output_stream.close(); + // Delete original file if so instructed + if (m_delete_orig) + { + bfs::remove(work_unit->m_path); + } + + // Print status if (m_decompress) { WRKR_OUT_REG(" -- (" << std::fixed << std::setprecision(2) @@ -294,12 +337,11 @@ void PhotoCompressArchiver::worker_thread_body(uint32_t tid) << percent << "%") } - delete work_unit; } WRKR_OUT_REG(" -- " << PURPLE << BOLD << "[[exiting]]" << ENDC) - + // Clean up std::free(out_buffer); delete instance; diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.hh b/software/photo_compress_archiver/PhotoCompressArchiver.hh index 8434e57..8690c1a 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.hh +++ b/software/photo_compress_archiver/PhotoCompressArchiver.hh @@ -24,7 +24,7 @@ const std::string RED("\033[31m"); const std::string YELLOW("\033[33m"); const std::string PURPLE("\033[35m"); -const std::string version("v0.2.0"); +const std::string version("v0.3.0"); const boost::regex JPEG_REGEX("^.+\\.(jpg)|(jpeg)$", boost::regex::icase); const boost::regex PJG_REGEX("^.+\\.pjg$", boost::regex::icase); @@ -42,6 +42,7 @@ class WorkUnit : m_path(path), m_file_size(0) { + // Automagically determine the file size m_file_size = bfs::file_size(path); }; }; @@ -52,9 +53,9 @@ class PhotoCompressArchiver PhotoCompressArchiver( const std::string& path, - uint32_t num_cores, + uint32_t num_threads, bool decompress, - bool keep_orig, + bool delete_orig, boost::regex& filter_regex); ~PhotoCompressArchiver(); @@ -85,9 +86,9 @@ class PhotoCompressArchiver private: const bfs::path m_path; // top level path where to look for input files - uint32_t m_num_cores; // number of cores on the machine + uint32_t m_num_threads; // number of processing threads to use bool m_decompress; // decompress files instead of compress them - bool m_keep_orig; // keep; don't delete original files + bool m_delete_orig; // delete original files after de/compression std::mutex m_output_mutex; // output mutex @@ -99,6 +100,9 @@ class PhotoCompressArchiver // input file list (either images or compressed images) ProtectedAndSynchronizedQueue m_file_list; + // list of any errors + ProtectedAndSynchronizedQueue m_errors; + std::thread* p_finder_thread; // file finder thread std::vector m_worker_threads; // worker threads diff --git a/software/photo_compress_archiver/main.cc b/software/photo_compress_archiver/main.cc index b770177..8e348a2 100644 --- a/software/photo_compress_archiver/main.cc +++ b/software/photo_compress_archiver/main.cc @@ -13,7 +13,7 @@ namespace bpo = boost::program_options; void header() { std::cout << BOLD << "Photo Compress Archiver -- " << version << "\n" << ENDC; - std::cout << " Copyright 2018 - David Sorber\n"; + std::cout << " Copyright 2017-2018 -- David Sorber\n"; std::cout << " >>> Utilizing packJPG by Matthias Stirner\n" << std::endl; } @@ -27,21 +27,27 @@ void usage(char** argv) } header(); - std::cout << "USAGE: " << app_name << " [-hkd] [-f ] \n\n"; + std::cout << "USAGE: " << app_name << " [-hdk] [-t ] " + << "[-f ] \n\n"; std::cout << "Positional arguments:\n"; std::cout << " path in which to recursively look " << "for input files\n\n"; std::cout << "General options:\n"; std::cout << " -h [ --help ] display this help message\n"; std::cout << " -d [ --decompress ] find '*.pjg' files and decompress them\n"; - std::cout << " -k [ --keep-orig ] keep original files; don't delete them\n"; - std::cout << " -f [ --filter ] regex with which to filter out input files\n"; + std::cout << " -k [ --delete-orig ] delete original files\n"; + std::cout << " -t [ --num-threads ] number of processing threads to use (max 2x system cores)\n"; + std::cout << " -f [ --filter ] regex with which to filter out input files\n"; std::cout << std::endl; } // NOTE: useful filter: -f ^\\._.+$ int main(int argc, char** argv) { + uint32_t num_cores = std::thread::hardware_concurrency(); + uint32_t num_threads = 0; + std::string search_path(""); + std::string filter_str(""); bpo::variables_map general_opts_vm; try @@ -50,11 +56,22 @@ int main(int argc, char** argv) bpo::options_description general_opts("General options"); general_opts.add_options() ("help,h", "display this help message") + ("decompress,d", "find '*.pjg' files and decompress them") - ("keep-orig,k", "keep orignal files") - ("filter,f", bpo::value(), - "regex with which to filter out input files") - ("file_path", "path in which to recursively look for input files") + + ("delete-orig,k", "delete orignal files") + + ("num-threads,t", + bpo::value(&num_threads)->default_value(num_cores), + "number of processing threads to use") + + ("filter,f", + bpo::value(&filter_str), + "regex with which to filter out input files") + + // Note this has to be here even though it's a positional argument + ("file_path", bpo::value(&search_path)->required(), + "path in which to recursively look for input files") ; // Positional arguments @@ -62,7 +79,8 @@ int main(int argc, char** argv) pos_opts.add("file_path", 1); // Parse general/positional options - bpo::store(bpo::command_line_parser(argc, argv).options(general_opts) + bpo::store(bpo::command_line_parser(argc, argv) + .options(general_opts) .positional(pos_opts).run(), general_opts_vm); @@ -74,43 +92,29 @@ int main(int argc, char** argv) } bpo::notify(general_opts_vm); - } + } catch (std::exception& e) { - std::cerr << BOLD << RED << "ERROR: " << ENDC << e.what() << "\n"; + std::cerr << BOLD << RED << "\nERROR: " << ENDC << e.what() << "\n" + << std::endl; return -1; } - // Make a copy of the file search path for passing into the object below - std::string search_path(general_opts_vm["file_path"].as()); - // Make sure that path specified actually exists before proceeding if (! bfs::exists(search_path)) { - std::cerr << BOLD << RED << "ERROR: " << ENDC << "the path \"" - << search_path << "\" does not exist" << std::endl; + std::cerr << BOLD << RED << "\nERROR: " << ENDC << "the path \"" + << search_path << "\" does not exist\n" << std::endl; return -2; } - // Display header and - header(); - uint32_t num_cores = std::thread::hardware_concurrency(); - std::cout << "[" << num_cores << " cores detected]\n"; - - // Decode/store decompress option - bool decompress = false; - if (general_opts_vm.count("decompress")) - { - decompress = true; - std::cout << " decompress mode => ON\n"; - } - - // Decode/store keep-orig option - bool keep_orig = false; - if (general_opts_vm.count("keep-orig")) + // Validate number of threads + if (num_threads > (2 * std::thread::hardware_concurrency())) { - keep_orig = true; - std::cout << " keep original files => ON\n"; + std::cerr << BOLD << RED << "\nERROR: " << ENDC << "the maximum number " + << "of threads is: " << (2 * num_cores) << "(2 * " << num_cores + << ")\n" << std::endl; + return -2; } // Check for, and if present validate the filter regex @@ -119,25 +123,53 @@ int main(int argc, char** argv) { try { - filter_regex = general_opts_vm["filter"].as().c_str(); + filter_regex = filter_str; } catch (std::exception& e) { std::cerr << BOLD << RED << "ERROR: " << ENDC << "the filter \"" - << general_opts_vm["filter"].as() - << "\" is not a valid regular expression.\n"; + << filter_str << "\" is not a valid regular expression.\n"; std::cerr << " " << e.what() << std::endl; return -1; } - - std::cout << " filter regex => ON (" << filter_regex << ")\n"; } + // Display header and options + header(); + std::cout << " [" << num_cores << " cores detected]\n" << std::endl; + std::cout << BOLD << ">>> Options:" << ENDC << std::endl; + + // Decode/store decompress option + bool decompress = false; + if (general_opts_vm.count("decompress")) + { + decompress = true; + } + std::cout << " decompress mode => " << (decompress ? "ON" : "OFF") + << std::endl; + + // Decode/store delete-orig option + bool delete_orig = false; + if (general_opts_vm.count("delete-orig")) + { + delete_orig = true; + } + std::cout << " delete original files => " << (delete_orig ? "ON" : "OFF") + << std::endl; + + // Number of processing threads option + std::cout << " num processing threads => " << num_threads << std::endl; + + // Filter regex option + if (general_opts_vm.count("filter")) + { + std::cout << " filter regex => ON (" << filter_str << ")\n"; + } std::cout << std::endl; // Create the PCA object and let 'er rip! - auto pca = new PhotoCompressArchiver(search_path, num_cores, decompress, - keep_orig, filter_regex); + auto pca = new PhotoCompressArchiver(search_path, num_threads, decompress, + delete_orig, filter_regex); pca->execute(); delete pca;