Project

General

Profile

« Previous | Next » 

Revision 3eeb1485

Added by David Sorber over 9 years ago

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!

View differences:

software/photo_compress_archiver/PhotoCompressArchiver.cc
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";
......
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;)
......
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)
{
......
<< 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<std::mutex> 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<bfs::path>(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<std::mutex> 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<std::mutex> 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;
......
}
void PhotoCompressArchiver::fork_worker(
uint32_t tid,
std::vector<bfs::path>* 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<const char*> 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<std::string*> 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<std::mutex> 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<const char*>(exec_argv[idx++]) << std::endl;
<< static_cast<const char*>(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)
......
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;
......
}
// 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));
}
......
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);
......
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
......
// 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;
......
// 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
......
// 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: "
software/photo_compress_archiver/PhotoCompressArchiver.hh
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <mutex>
#include <string>
#include <thread>
......
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<const char*> exec_argv;
std::vector<std::string*> output_filenames;
};
typedef struct work_unit_t WorkUnit;
class PhotoCompressArchiver
{
public:
......
/**
* Fork and wait
*/
void fork_worker(uint32_t tid, std::vector<bfs::path>* file_sublist);
void fork_worker(uint32_t tid);
/**
* Calculate the return the "global" percent done as determined by the
......
std::thread* p_finder_thread; // file finder thread
std::vector<std::thread*> m_worker_threads; // worker threads
std::deque<WorkUnit*> 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
};

Also available in: Unified diff