commit f8e600bcd4375670302d10c2132a8741ca0674bf Author: David Sorber Date: Wed Feb 8 18:51:58 2017 -0500 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. diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.cc b/software/photo_compress_archiver/PhotoCompressArchiver.cc index d7e1cdc..23b7fb7 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.cc +++ b/software/photo_compress_archiver/PhotoCompressArchiver.cc @@ -1,6 +1,6 @@ #include +#include #include -#include // temporary for std::distance #include #include #include @@ -22,6 +22,7 @@ PhotoCompressArchiver::PhotoCompressArchiver( 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), @@ -93,14 +94,13 @@ int PhotoCompressArchiver::execute() 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; } @@ -140,7 +140,11 @@ void PhotoCompressArchiver::fork_worker( uint32_t tid, std::vector* 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 @@ -151,6 +155,9 @@ void PhotoCompressArchiver::fork_worker( } #endif +#ifdef __linux__ +#error MAC OS is not supported dumbass! +#endif uint32_t argv_size = file_sublist->size() + 5; @@ -170,6 +177,28 @@ void PhotoCompressArchiver::fork_worker( } exec_argv[argv_size - 1] = nullptr; + + // --- temp idea horribleness + std::vector 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) @@ -187,15 +216,9 @@ void PhotoCompressArchiver::fork_worker( // 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]); @@ -206,32 +229,108 @@ void PhotoCompressArchiver::fork_worker( _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; } diff --git a/software/photo_compress_archiver/PhotoCompressArchiver.hh b/software/photo_compress_archiver/PhotoCompressArchiver.hh index 290ea83..ccf2949 100644 --- a/software/photo_compress_archiver/PhotoCompressArchiver.hh +++ b/software/photo_compress_archiver/PhotoCompressArchiver.hh @@ -41,21 +41,28 @@ class PhotoCompressArchiver * Fork and wait */ void fork_worker(uint32_t tid, std::vector* file_sublist); + + /** + * Calculate the return the "global" percent done as determined by the + * m_files_processed and m_num_input_files instance variables. + */ + double get_global_percent_done(); private: - const bfs::path m_path; // top level path + 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 std::mutex m_output_mutex; // output mutex - uint32_t m_num_input_files; // total number of input files found - std::atomic m_total_uncompressed_size; // total uncompressed size of all images - std::atomic m_total_compressed_size; // total compressed size of all images + std::atomic m_files_processed; // number of input files processed + uint32_t m_num_input_files; // total number of input files found + std::atomic m_total_uncompressed_size;// total uncompressed size of all images + std::atomic m_total_compressed_size; // total compressed size of all images - std::vector m_file_list; // input file list (either images or compressed images) + std::vector m_file_list; // input file list (either images or compressed images) - std::thread* p_finder_thread; // file finder thread - std::vector m_worker_threads; // worker threads + std::thread* p_finder_thread; // file finder thread + std::vector m_worker_threads; // worker threads }; #endif // PHOTOCOMPRESSARCHIVER_H