Revision f8e600bc
Added by David Sorber over 9 years ago
| software/photo_compress_archiver/PhotoCompressArchiver.cc | ||
|---|---|---|
|
#include <chrono>
|
||
|
#include <iomanip>
|
||
|
#include <iostream>
|
||
|
#include <iterator> // temporary for std::distance
|
||
|
#include <sys/stat.h>
|
||
|
#include <sys/wait.h>
|
||
|
#include <unistd.h>
|
||
| ... | ... | |
|
uint32_t num_cores)
|
||
|
: m_path(path),
|
||
|
m_num_cores(num_cores),
|
||
|
m_files_processed(0),
|
||
|
m_num_input_files(0),
|
||
|
m_total_uncompressed_size(0),
|
||
|
m_total_compressed_size(0),
|
||
| ... | ... | |
|
delete worker;
|
||
|
}
|
||
|
|
||
|
OUT(std::cout << " Total compressed bytes: " << m_total_compressed_size
|
||
|
OUT(std::cout << "\n 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(std::cout << " Compression ratio: " << ratio << "\n" << std::endl;)
|
||
|
|
||
|
OUT(std::cout << "execute terminating" << std::endl;)
|
||
|
OUT(std::cout << " Compression ratio: " << std::fixed
|
||
|
<< std::setprecision(4) << (ratio * 100) << "%\n" << std::endl;)
|
||
|
|
||
|
return 0;
|
||
|
}
|
||
| ... | ... | |
|
uint32_t tid,
|
||
|
std::vector<bfs::path>* file_sublist)
|
||
|
{
|
||
|
OUT(std::cout << "T[" << tid << "] top of the morning to ya" << std::endl;)
|
||
|
uint32_t sublist_file_count = file_sublist->size();
|
||
|
uint32_t files_processed = 0;
|
||
|
|
||
|
OUT(std::cout << "T[" << tid << "] top of the morning to ya: "
|
||
|
<< sublist_file_count << std::endl;)
|
||
|
|
||
|
// Print out the input files given to this worker
|
||
|
#if 0
|
||
| ... | ... | |
|
}
|
||
|
#endif
|
||
|
|
||
|
#ifdef __linux__
|
||
|
#error MAC OS is not supported dumbass!
|
||
|
#endif
|
||
|
|
||
|
|
||
|
uint32_t argv_size = file_sublist->size() + 5;
|
||
| ... | ... | |
|
}
|
||
|
exec_argv[argv_size - 1] = nullptr;
|
||
|
|
||
|
|
||
|
// --- temp idea horribleness
|
||
|
std::vector<std::string*> output_filenames;
|
||
|
for (auto& filepath : *file_sublist)
|
||
|
{
|
||
|
// Create new filename that contains the ".jpg" replaced with ".pjg"
|
||
|
uint32_t end_pos = filepath.string().find(".");
|
||
|
std::string* filename = new std::string(filepath.string().begin(),
|
||
|
filepath.string().begin() + end_pos);
|
||
|
(*filename) += ".pjg";
|
||
|
|
||
|
output_filenames.push_back(filename);
|
||
|
}
|
||
|
//--------------------------
|
||
|
|
||
|
#if 0
|
||
|
for (auto output_filename : output_filenames)
|
||
|
{
|
||
|
OUT(std::cout << *output_filename << std::endl;)
|
||
|
}
|
||
|
#endif
|
||
|
|
||
|
// Create a pipe to hold stdout from child process
|
||
|
int filedes[2];
|
||
|
if (pipe(filedes) == -1)
|
||
| ... | ... | |
|
// error, failed to fork()
|
||
|
OUT(std::cout << "T[" << tid << "] fork failed!" << std::endl;)
|
||
|
}
|
||
|
else if (pid > 0)
|
||
|
{
|
||
|
// Wait for forked child process to terminate
|
||
|
int status;
|
||
|
waitpid(pid, &status, 0);
|
||
|
OUT(std::cout << "T[" << tid << "] RC: " << status << std::endl;)
|
||
|
}
|
||
|
else
|
||
|
else if (pid == 0)
|
||
|
{
|
||
|
// This is the child process...
|
||
|
// Set child process's stdout to the pipe entry
|
||
|
while ((dup2(filedes[1], STDOUT_FILENO) == -1) && (errno == EINTR)) {}
|
||
|
close(filedes[1]);
|
||
| ... | ... | |
|
_exit(EXIT_FAILURE); // exec never returns
|
||
|
}
|
||
|
|
||
|
// Free up the argv array
|
||
|
delete[] exec_argv;
|
||
|
|
||
|
OUT(std::cout << "T[" << tid << "] compression complete" << std::endl;)
|
||
|
// This is the parent process...
|
||
|
|
||
|
// Iterate over the files and find their compressed size
|
||
|
std::string new_file;
|
||
|
for (auto& filepath : *file_sublist)
|
||
|
// 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())
|
||
|
{
|
||
|
// Create new filename that contains the ".jpg" replaced with ".pjg"
|
||
|
uint32_t end_pos = filepath.string().find(".");
|
||
|
new_file.assign(filepath.string().begin(),
|
||
|
filepath.string().begin() + end_pos);
|
||
|
new_file += ".pjg";
|
||
|
|
||
|
// Stat the file to get its size
|
||
|
struct stat statbuf;
|
||
|
int rc = stat(new_file.c_str(), &statbuf);
|
||
|
if (rc)
|
||
|
// Poll, waiting for output file X + 1 to exist
|
||
|
bailout_ctr = 0;
|
||
|
while ((stat(output_filenames[output_idx]->c_str(), &statbuf) != 0) &&
|
||
|
(++bailout_ctr < BAILOUT_MAX))
|
||
|
{
|
||
|
OUT(std::cerr << "T[" << tid << "] ERROR: unable to stat file "
|
||
|
<< new_file << std::endl;)
|
||
|
}
|
||
|
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||
|
}
|
||
|
|
||
|
// Check if timeout has occurred
|
||
|
if (bailout_ctr == BAILOUT_MAX)
|
||
|
{
|
||
|
OUT(std::cerr << "T[" << tid << "] ERROR: timed out while waiting "
|
||
|
<< "for output file: "
|
||
|
<< *output_filenames[output_idx] << "; aborting"
|
||
|
<< std::endl;)
|
||
|
return;
|
||
|
}
|
||
|
|
||
|
// Now stat output file X which should exist
|
||
|
stat_rc = stat(output_filenames[output_idx - 1]->c_str(), &statbuf);
|
||
|
if (stat_rc)
|
||
|
{
|
||
|
OUT(std::cerr << "ERROR while stating: "
|
||
|
<< *output_filenames[output_idx - 1] << std::endl;)
|
||
|
}
|
||
|
|
||
|
// Increment counts
|
||
|
m_total_compressed_size += statbuf.st_size;
|
||
|
++m_files_processed;
|
||
|
++files_processed;
|
||
|
|
||
|
// Calculate the local percent done
|
||
|
complete_percent = (double)files_processed / sublist_file_count;
|
||
|
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)
|
||
|
<< complete_percent
|
||
|
<< "%) file: " << *output_filenames[output_idx - 1]
|
||
|
<< " -- " << statbuf.st_size << std::endl;)
|
||
|
|
||
|
++output_idx;
|
||
|
}
|
||
|
|
||
|
// Wait for forked child process to terminate
|
||
|
int status;
|
||
|
waitpid(pid, &status, 0);
|
||
|
|
||
|
// 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);
|
||
|
if (stat_rc)
|
||
|
{
|
||
|
OUT(std::cerr << "ERROR while stating: "
|
||
|
<< *output_filenames[output_idx - 1] << std::endl;)
|
||
|
}
|
||
|
|
||
|
// Increment counts
|
||
|
m_total_compressed_size += statbuf.st_size;
|
||
|
++m_files_processed;
|
||
|
++files_processed;
|
||
|
|
||
|
// Calculate the local percent done
|
||
|
complete_percent = (double)files_processed / sublist_file_count;
|
||
|
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] << " -- "
|
||
|
<< statbuf.st_size << std::endl;)
|
||
|
|
||
|
OUT(std::cout << "T[" << tid << "] complete ==> RC: " << status << std::endl;)
|
||
|
|
||
|
// Free up the argv array and output filename list
|
||
|
delete[] exec_argv;
|
||
|
for (auto output_filename : output_filenames)
|
||
|
{
|
||
|
delete output_filename;
|
||
|
}
|
||
|
|
||
|
|
||
|
OUT(std::cout << "T[" << tid << "] terminating" << std::endl;)
|
||
|
//~ OUT(std::cout << "T[" << tid << "] terminating" << std::endl;)
|
||
|
}
|
||
|
|
||
|
double PhotoCompressArchiver::get_global_percent_done()
|
||
|
{
|
||
|
return ((double)m_files_processed / m_num_input_files) * 100;
|
||
|
}
|
||
I figured out a somewhat silly, but highly effective way do progress reporting. The basic idea is that we know the output file names and the exact order in which they will appear. Therefore if output X+1 exists on the file system it means that output X is already complete. Now the worker thread polls the file system waiting for the output files to appear and then processses them instead of waiting until the forked subprocess terminates. In basic testing it appears that this mechanism is about the same speed as the previous version that does not include progress reporting. This implementation is cross platform and doesn't suffer from the problems of inotify (Linux only) or kqueues (BSD/Mac OS equivalent of inotify that requires an open file descriptor for each monitored file). I still need to implement input option parsing and a decompress mode so I can do proper testing.