Project

General

Profile

« Previous | Next » 

Revision 537f40e6

Added by David Sorber almost 8 years ago

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.

View differences:

software/photo_compress_archiver/PhotoCompressArchiver.cc
#include "PhotoCompressArchiver.hh"
#define OUT(msg) { \
std::lock_guard<std::mutex> lock(m_output_mutex); \
std::cout << msg << std::endl; \
}
#define ERROR(msg) { \
std::lock_guard<std::mutex> 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<std::mutex> lock(m_output_mutex); \
......
#define WRKR_OUT_ERR(msg) { \
std::lock_guard<std::mutex> 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),
......
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);
}
......
// Grab start time and calculate duration
auto end = std::chrono::system_clock::now();
auto diff = end - start;
double duration = std::chrono::duration<double, std::milli>(diff).count();
double duration = std::chrono::duration<double>(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;
}
......
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)
......
++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)
......
<< percent << "%")
}
delete work_unit;
}
WRKR_OUT_REG(" -- " << PURPLE << BOLD << "[[exiting]]" << ENDC)
// Clean up
std::free(out_buffer);
delete instance;
software/photo_compress_archiver/PhotoCompressArchiver.hh
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);
......
: m_path(path),
m_file_size(0)
{
// Automagically determine the file size
m_file_size = bfs::file_size(path);
};
};
......
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();
......
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
......
// input file list (either images or compressed images)
ProtectedAndSynchronizedQueue<WorkUnit*> m_file_list;
// list of any errors
ProtectedAndSynchronizedQueue<WorkUnit*> m_errors;
std::thread* p_finder_thread; // file finder thread
std::vector<std::thread*> m_worker_threads; // worker threads
software/photo_compress_archiver/main.cc
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;
}
......
}
header();
std::cout << "USAGE: " << app_name << " [-hkd] [-f <filter regex>] <file path>\n\n";
std::cout << "USAGE: " << app_name << " [-hdk] [-t <num threads>] "
<< "[-f <filter regex>] <file path>\n\n";
std::cout << "Positional arguments:\n";
std::cout << " <file path> 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
......
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<std::string>(),
"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<uint32_t>(&num_threads)->default_value(num_cores),
"number of processing threads to use")
("filter,f",
bpo::value<std::string>(&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<std::string>(&search_path)->required(),
"path in which to recursively look for input files")
;
// Positional arguments
......
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);
......
}
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<std::string>());
// 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
......
{
try
{
filter_regex = general_opts_vm["filter"].as<std::string>().c_str();
filter_regex = filter_str;
}
catch (std::exception& e)
{
std::cerr << BOLD << RED << "ERROR: " << ENDC << "the filter \""
<< general_opts_vm["filter"].as<std::string>()
<< "\" 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;

Also available in: Unified diff