Project

General

Profile

Download (7.35 KB) Statistics
| Branch: | Tag: | Revision:

root/software/photo_compress_archiver/PhotoCompressArchiver.cc @ f6c21781

b216a5b0 David Sorber
#include <chrono>
#include <iostream>
f6c21781 David Sorber
#include <iterator> // temporary for std::distance
b216a5b0 David Sorber
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>

#include "PhotoCompressArchiver.hh"

#define OUT(x) {\
std::lock_guard<std::mutex> lock(m_output_mutex);\
x\
}

const char *PROG_NAME = "./packJPG";
const char *PROG_OPT1 = "-np";
const char *PROG_OPT2 = "-o";
const char *PROG_OPT3 = "-p";

PhotoCompressArchiver::PhotoCompressArchiver(
const char* path,
uint32_t num_cores)
: m_path(path),
m_num_cores(num_cores),
m_num_input_files(0),
m_total_uncompressed_size(0),
m_total_compressed_size(0),
p_finder_thread(nullptr)
{}

PhotoCompressArchiver::~PhotoCompressArchiver()
{}

int PhotoCompressArchiver::execute()
{
// Start the file finder thread
OUT(std::cout << "Starting file finder..." << std::flush;);
p_finder_thread = new std::thread(&PhotoCompressArchiver::find_files, this,
m_path, JPEG_REGEX);
OUT(std::cout << "DONE" << std::endl;)
p_finder_thread->join();
OUT(std::cout << "File finder completed:" << std::endl;)
OUT(std::cout << " Found " << m_num_input_files << " images" << std::endl;)
OUT(std::cout << " Total uncompressed bytes: " << m_total_uncompressed_size
<< "\n" << std::endl;)
// Start worker threads
OUT(std::cout << "Starting worker threads..." << std::endl;)
// TODO: need to handle the cause 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)
{
// 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))
{
// 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();
}
else
{
start_iter = m_file_list.begin() + (idx * sublist_len);
end_iter = m_file_list.begin() + ((idx + 1) * sublist_len);
}
auto sublist = new std::vector<bfs::path>(start_iter, end_iter);
// 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);
}
OUT(std::cout << "All worker threads started" << std::endl;)
// Clean up the finder thread
delete p_finder_thread;
p_finder_thread = nullptr;
// Now join the worker threads
for (auto worker : m_worker_threads)
{
worker->join();
delete worker;
}
OUT(std::cout << " Total compressed bytes: " << m_total_compressed_size
f6c21781 David Sorber
<< 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;)
b216a5b0 David Sorber
OUT(std::cout << "execute terminating" << std::endl;)
return 0;
}

void PhotoCompressArchiver::find_files(
const bfs::path& dir_path,
const boost::regex& file_regex)
{
bfs::directory_iterator end_iter;
for (bfs::directory_iterator dir_iter(dir_path); dir_iter != end_iter; ++dir_iter)
{
if (bfs::is_directory(dir_iter->status()))
{
// Found directory recurse
find_files(dir_iter->path(), file_regex);
}
else if (boost::regex_match(dir_iter->path().filename().string(), file_regex))
{
// Found file match
m_file_list.push_back(dir_iter->path());
++m_num_input_files;
// Stat the file to get its size
struct stat statbuf;
int rc = stat(dir_iter->path().string().c_str(), &statbuf);
if (rc)
{
OUT(std::cerr << "ERROR: unable to stat file "
<< dir_iter->path() << std::endl;)
}
m_total_uncompressed_size += statbuf.st_size;
}
}
}

void PhotoCompressArchiver::fork_worker(
uint32_t tid,
std::vector<bfs::path>* file_sublist)
{
OUT(std::cout << "T[" << tid << "] top of the morning to ya" << std::endl;)
// Print out the input files given to this worker
#if 0
for (auto& filepath : *file_sublist)
{
OUT(std::cout << "T[" << tid << "] file: " << filepath.string()
<< std::endl;)
}
#endif
f6c21781 David Sorber

b216a5b0 David Sorber
uint32_t argv_size = file_sublist->size() + 5;
const char **exec_argv = new const char* [argv_size];
// Build the argv array; pass in default program options
exec_argv[0] = &PROG_NAME[2]; // discard leading "./"
exec_argv[1] = PROG_OPT1;
exec_argv[2] = PROG_OPT2;
exec_argv[3] = PROG_OPT3;
// Add the sublist files to the argv array
f6c21781 David Sorber
uint32_t idx = 4;
for (auto& image_path : *file_sublist)
b216a5b0 David Sorber
{
exec_argv[idx++] = image_path.string().c_str();
}
exec_argv[argv_size - 1] = nullptr;
// Create a pipe to hold stdout from child process
int filedes[2];
if (pipe(filedes) == -1)
{
perror("pipe");
exit(1);
}
// Fork off the childprocess
//~ pid_t parent = getpid();
pid_t pid = vfork();

if (pid == -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
{
// Set child process's stdout to the pipe entry
while ((dup2(filedes[1], STDOUT_FILENO) == -1) && (errno == EINTR)) {}
close(filedes[1]);
close(filedes[0]);
// Child after fork
execv(PROG_NAME, (char **)exec_argv);
_exit(EXIT_FAILURE); // exec never returns
}
// Free up the argv array
delete[] exec_argv;
OUT(std::cout << "T[" << tid << "] compression complete" << std::endl;)
// Iterate over the files and find their compressed size
std::string new_file;
for (auto& filepath : *file_sublist)
{
// 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)
{
f6c21781 David Sorber
OUT(std::cerr << "T[" << tid << "] ERROR: unable to stat file "
<< new_file << std::endl;)
b216a5b0 David Sorber
}
m_total_compressed_size += statbuf.st_size;
}

OUT(std::cout << "T[" << tid << "] terminating" << std::endl;)
}