archiver.cc

Go to the documentation of this file.

#include "config.h"

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif

#include <ctype.h>

#include <iostream>
#include <string>
#include <vector>

#include "asserts.h"
#include "error.h"
#include "estring.h"
#include "fs.h"
#include "rconfig.h"
#include "timer.h"
#include "logger.h"

#include "archiver.h"

//-----------------------------------------------------------------------------

/** C'tor */
rstat::rstat()
{
        TRY_nomem(m_exit_str[0] = "Success");
        TRY_nomem(m_exit_str[1] = "Syntax or usage error");
        TRY_nomem(m_exit_str[2] = "Protocol incompatability error");
        TRY_nomem(m_exit_str[3] = "Errors selecting I/O files or directories");
        TRY_nomem(m_exit_str[4] = "Requested action not supported");
        TRY_nomem(m_exit_str[10] = "Error in socket I/O");
        TRY_nomem(m_exit_str[11] = "Error in file I/O");
        TRY_nomem(m_exit_str[12] = "Error in rsync protocol data stream");
        TRY_nomem(m_exit_str[13] = "Errors with program diagnostics");
        TRY_nomem(m_exit_str[14] = "Error in IPC code");
        TRY_nomem(m_exit_str[20] = "Received SIGUSR1 or SIGINT");
        TRY_nomem(m_exit_str[21] = "Some error returned by waitpid()");
        TRY_nomem(m_exit_str[22] = "Error allocating core memory buffers");
        TRY_nomem(m_exit_str[23] = "Partial transfer");
        TRY_nomem(m_exit_str[30] = "Timeout in data send/receive");
        TRY_nomem(m_exit_str[124] = "The command executed by SSH exited with status 255");
        TRY_nomem(m_exit_str[125] = "The command executed by SSH was killed by a signal");
        TRY_nomem(m_exit_str[126] = "The command given to SSH cannot be run");
        TRY_nomem(m_exit_str[127] = "The command given to SSH cannot be found");

        TRY_nomem(m_signal_str[1] = "[HUP]: Hangup");
        TRY_nomem(m_signal_str[2] = "[INT]: Interrupt ");
        TRY_nomem(m_signal_str[3] = "[QUIT]: Quit");
        TRY_nomem(m_signal_str[4] = "[ILL]: Illegal instruction");
        TRY_nomem(m_signal_str[5] = "[TRAP]: Trace trap");
        TRY_nomem(m_signal_str[6] = "[IOT]: IOT instruction or hardware fault");
        TRY_nomem(m_signal_str[7] = "[ABRT]: Abnormal termination");
        TRY_nomem(m_signal_str[8] = "[EMT]: EMT instruction or hardware fault");
        TRY_nomem(m_signal_str[9] = "[FPE]: Floating point exception");
        TRY_nomem(m_signal_str[10] = "[KILL]: Killed");
        TRY_nomem(m_signal_str[11] = "[BUS]: Bus error");
        TRY_nomem(m_signal_str[12] = "[SEGV]: Segmentation fault");
        TRY_nomem(m_signal_str[13] = "[SYS]: Invalid system call or invalid argument to system call");
        TRY_nomem(m_signal_str[14] = "[PIPE]: Write to pipe with no readers");
        TRY_nomem(m_signal_str[15] = "[ALRM]: Alarm");
        TRY_nomem(m_signal_str[16] = "[TERM]: Software termination");
        TRY_nomem(m_signal_str[17] = "[USR1]: User-defined signal 1");
        TRY_nomem(m_signal_str[18] = "[USR2]: User-defined signal 2");
        TRY_nomem(m_signal_str[19] = "[CLD]: Child status change");
        TRY_nomem(m_signal_str[20] = "[PWR]: Power fail/restart");
        TRY_nomem(m_signal_str[21] = "[WINCH]: Terminal window size change");
        TRY_nomem(m_signal_str[22] = "[URG]: Urgent condition");
        TRY_nomem(m_signal_str[23] = "[POLL]: Pollable event or socket I/O");
        TRY_nomem(m_signal_str[24] = "[STOP]: Stop");
        TRY_nomem(m_signal_str[25] = "[TSTP]: Terminal stop character");
        TRY_nomem(m_signal_str[26] = "[CONT]: Continue stopped process");
        TRY_nomem(m_signal_str[27] = "[TTIN]: Background tty read");
        TRY_nomem(m_signal_str[28] = "[TTOU]: Background tty write");
        TRY_nomem(m_signal_str[29] = "[VTALRM]: Virtual timer expired");
        TRY_nomem(m_signal_str[30] = "[PROF]: Profiling timer expired");
        TRY_nomem(m_signal_str[31] = "[XCPU]: Exceeded CPU limit");
        TRY_nomem(m_signal_str[32] = "[XFSZ]: Exceeded file size limit");
        TRY_nomem(m_signal_str[33] = "[WAITING]: Process' LWPs are blocked");
        TRY_nomem(m_signal_str[34] = "[LWP]: Special thread library signal");
        TRY_nomem(m_signal_str[35] = "[FREEZE]: Special signal used by CPR");
        TRY_nomem(m_signal_str[36] = "[THAW]: Special signal used by CPR");
        TRY_nomem(m_signal_str[37] = "[CANCEL]: Thread cancellation");
        TRY_nomem(m_signal_str[38] = "[LOST]: Resource lost");
        TRY_nomem(m_signal_str[39] = "[RTMIN]: Highest priority real-time signal");
        TRY_nomem(m_signal_str[46] = "[RTMAX]: Lowest priority real-time signal");

        TRY_nomem(m_unknown_exit = "(Unknown exit code)");
        TRY_nomem(m_unknown_signal = "(Unknown signal)");
}

/** Get a verbose string for an exit code */
const std::string& rstat::exit(const int a_int) const
{
        if (m_exit_str.find(a_int) != m_exit_str.end()) {
                return(m_exit_str.find(a_int)->second);
        }
        return(m_unknown_exit);
}

/** Get a verbose string for a signal number */
const std::string& rstat::signal(const int a_int) const
{
        if (m_signal_str.find(a_int) != m_signal_str.end()) {
                return(m_signal_str.find(a_int)->second);
        }
        return(m_unknown_signal);
}

class rstat rsync_estat_str;

//-----------------------------------------------------------------------------

/** C'tor

        Set a job to be assiciated with this job archiver and initialize it's
        processing status to "pending".

 */
job_archiver::job_archiver(const job * a_job)
{
        clear();
        m_job = a_job;
        m_status = status_pending;
}

/** Generate a job prefix string

        Create a string to uniquely identify this job to be used in the log file to
        uniquely identify this job

 */
const std::string job_archiver::prefix(void)
{
        estring lstr;

        lstr = "[Job:";
        lstr += estring((void*)m_job);
        lstr += "] ";

        return(lstr);
}

/** Generate a job id string */
const std::string job_archiver::id(void)
{
        estring lstr;

        lstr = prefix();
        lstr += " ";
        lstr += m_job->generate_job_id();

        return(lstr);
}

/** Clear the job archiver and return it to it's initial state
        
        End any processes handling this job and return the job archiver to it's
        "pending" state.

 */
void job_archiver::clear(void)
{
        end();
        m_child_pid = m_exec.my_pid();
        m_status = status_pending;
        m_success = true;
        m_jr.clear();
        m_jpr.clear();
        m_error_msg.erase();
}

/** End any processes handling this job

        If any child processes are handling this job, terminate them.  Erase any
        pending I/O for the now defunct child.  Set our processing status to "done".

*/
void job_archiver::end(void)
{
        estring lstr;

        m_timer.stop();
        if (m_exec.child_running()) {
                lstr = prefix();
                lstr += "Terminating child process!\n";
                logger.write(lstr,0,configuration_manager::logging_manager,m_child_pid);
                m_exec.kill_child();
        }
        m_exec.clear();
        m_io_out.erase();
        m_io_err.erase();
        m_status = status_done;
}

/** Return the processing status of this job archiver */
const job_archiver::archiving_status job_archiver::status(void)
{
        return(m_status);
}

/** Begin processing

        Attempt to fork a child process to handle this job.  If unsuccessful then
        retry again later.  The child then calls mf_do_chores() to handle the actual
        processing, while the parent updates the job archiver's status from
        "pending" to "processing" and begins a timer to measure the duration of the
        job process.
        
*/
void job_archiver::start(void)
{
        estring lstr;

        m_jr.id(m_job->generate_job_id());

        try {
                m_exec.fork();
        }
        catch(error e) {
                lstr = prefix();
                lstr += "Could not fork:\n";
                logger.write(lstr);

                lstr = e.str(prefix());
                logger.write(lstr);

                lstr = prefix();
                lstr += "Will retry job later\n";
                logger.write(lstr);

                m_status = status_reschedule;
        }
        catch(...) {
                error e = err_unknown;

                lstr = prefix();
                lstr += "Could not fork:\n";
                logger.write(lstr);

                lstr = e.str(prefix());
                logger.write(lstr);

                lstr = prefix();
                lstr += "Will retry job later\n";
                logger.write(lstr);

                m_status = status_reschedule;
        }

        if (m_exec.is_child()) {
                // wait_for_debugger = true;

                // while (wait_for_debugger);

                m_exec.reroute_stdio();
                try {
                        mf_do_chores();
                }
                catch(error e) {
                        std::cerr << e;
                        m_success = false;
                }
                catch(...) {
                        std::cerr << err_unknown;
                        m_success = false;
                }
                if (m_success)
                        m_exec.exit(0);
                else
                        m_exec.exit(1);
        }

        m_child_pid = m_exec.child_pid();

        lstr = prefix();
        lstr += "Spawning child process: PID ";
        lstr += estring(static_cast<unsigned long>(m_exec.child_pid()));
        lstr += "\n";
        logger.write(lstr);

        m_status = status_processing;
        m_timer.start();
}

/** Parent processor for a job

        Check for I/O from the child.  Check the child's status to see if it's still
        running, has exited with an exit code, or has exited from a signal.  If the
        child sis not exit normally (i.e. exit from a signal or exit with a non-zero
        exit code) then check the vault for overflow.  If the vault has exceeded
        it's overflow threshold then that could be the cause for the child's
        failure, in which case we reschedule the child to be processed again later.

        If the job is finished (whether successful or not), update the job
        archiver's status to "completed".
        
 */
void job_archiver::process(void)
{
        estring lstr;
        
        if (m_exec.child_running()) {
                // Process child I/O
                mf_process_child_io(false);
        }
        else {
                // Process remaining child I/O
                mf_process_child_io(true);

                // If child exited with an error, check vault overflow.  If the vault is
                // filling up, then reschedule the job for later retry.
                lstr = prefix();
                if (m_exec.child_signaled()) {
                        lstr += "Child exited from signal: ";
                        lstr += estring(m_exec.child_signal_no());
                        lstr += "\n";
                }
                else if (m_exec.child_exit_code() != 0) {
                        lstr += "Child exited abnormally with code: ";
                        lstr += estring(m_exec.child_exit_code());
                        lstr += "\n";
                }
                else {
                        lstr += "Child exited successfully\n";
                        m_status = status_completed;
                }
                logger.write(lstr);

                if (m_exec.child_signaled() || !m_exec.child_exited_normally()) {
                        /*
                        if (vaulter.overflow()) {
                                lstr = prefix();
                                lstr += "Vault overflow detected, will retry later\n";
                                logger.write(lstr);
                                m_status = status_reschedule;
                        }
                        else {
                                m_status = status_error;
                        }
                        */
                        m_status = status_error;
                }
                else {
                        m_status = status_completed;
                }

                m_timer.stop();
                lstr = prefix();
                lstr += "Finished, duration: ";
                lstr += m_timer.duration();
                lstr += "\n";
                logger.write(lstr);
                // m_status = status_completed;
        }
}

/** Return the job report for this job */
single_job_report job_archiver::report(void) const
{
        return(m_jr);
}

/** Child processor for a job

        For each path in this job:
        - Create the directory heiararchy for this job in the archive
        - Until done or until out of retrys:
                - Choose a hardlink source, if applicable and available
                - Construct the command line to pass to rsync
                - Spawn rsync
                - Process I/O sent back from rsync
                - Process exit code or signal number returned from rsync
        - Generate and submit a report to the report manager
        
 */
void job_archiver::mf_do_chores(void)
{
        /*
        {
                bool wait_for_debugger = true;

                std::cerr << "Waiting for debugger to attach..." << std::endl;
                while (wait_for_debugger);
                std::cerr << "Debugger attached." << std::endl;
        }
        */

        job::paths_type::const_iterator pi;

        for (pi = m_job->paths.begin(); pi != m_job->paths.end(); ++pi) {
                estring archive_dir;
                estring path;
                // estring command_line;
                std::string binary;
                std::vector<std::string> argv;
                estring opt;
                bool hardlink = false;
                bool multi_hardlink = false;
                int num_retries = 0;
                bool done = false;
                int exit_code = 0;
                int signal_num = 0;
                timer t;
                uint64 files_total = 0;
                uint64 files_xferd = 0;
                uint64 size_total = 0;
                uint64 size_xferd = 0;
                bool overflow_detected = 0;
                estring error_msg;

                archive_dir = m_job->generate_archive_path(*pi);
                // If archive_dir does not end with a '/', strip off characters until it
                // does.
                /*
                while ((archive_dir.size() > 0) 
                        && (archive_dir[archive_dir.size()-1] != '/'))
                {
                        archive_dir.erase(archive_dir.size()-1);
                }
                */

                path = archiver.working_archive_path();
                path += "/";
                path += archive_dir;
                path = reform_path(path);
                if (!exists(path)) {
                        std::cout << "Creating job archive path: " << archive_dir << std::endl;
                        mk_dirhier(path);
                }
                else
                        std::cout << "Archiving to existing path: " << archive_dir << std::endl;

                if (!writable(path)) {
                        std::string es;

                        TRY_nomem(es = "Cannot write to archive directory: \"");
                        TRY_nomem(es += archive_dir);
                        TRY_nomem(es += "\"");

                        throw(ERROR(EACCES,es));
                }

                logger.set_error_logging(false);
                hardlink = m_job->rsync_hardlink;
                multi_hardlink = m_job->rsync_multi_hardlink;
                while ((num_retries <= m_job->rsync_retry_count) && !done) {
                        execute exec;
                        job::excludes_type::const_iterator ei;
                        job::includes_type::const_iterator ii;
                        std::vector<std::string>::const_iterator oi;

                        exit_code = 0;
                        signal_num = 0;

                        /*
                        command_line = config.rsync_local_path();
                        command_line += " ";
                        command_line += m_job->rsync_options;
                        */
                        /**/
                        binary = config.rsync_local_path();
                        argv = m_job->generate_rsync_options_vector();
                        /**/

                        if (m_job->rsync_connection != job::connection_local) {
                                /*
                                command_line += " ";
                                command_line += " --rsync-path=";
                                if (m_job->rsync_remote_path.size()) {
                                        command_line += m_job->rsync_remote_path;
                                }
                                else {
                                        command_line += config.rsync_local_path();
                                }
                                */
                                /**/
                                opt = "--rsync-path=";
                                if (m_job->rsync_remote_path.size() != 0)
                                        opt += m_job->rsync_remote_path;
                                else
                                        opt += config.rsync_local_path();
                                argv.push_back(opt);
                                /**/
                        }

                        if (hardlink) {
                                subdirectory subdir;
                                std::string youngest;
                                std::string relative_path;
                                bool hardlink_opt = false;
                                bool first_hardlink = false;
                                int linkdest_count = 0;

                                subdir.path(vaulter.vault());
                                if (subdir.size() > 0) {
                                        subdirectory::const_iterator si;

                                        sort(subdir.begin(), subdir.end());
                                        reverse(subdir.begin(), subdir.end());
                                        for (si = subdir.begin(); si != subdir.end(); ++si) {
                                                estring ypath;

                                                if (first_hardlink && !multi_hardlink) {
                                                        continue;
                                                }
                                                if (linkdest_count >= m_job->rsync_multi_hardlink_max) {
                                                        continue;
                                                }
                                                if (!is_timestamp(*si))
                                                        continue;
                                                if (*si == config.timestamp().str())
                                                        continue;
                                                std::cout 
                                                        << "Considering potential hardlink source: "
                                                        << *si
                                                        << std::endl;
                                                ypath = vaulter.vault();
                                                ypath += "/";
                                                ypath += *si;
                                                ypath += "/";
                                                ypath += archive_dir;
                                                if (exists(ypath)) {
                                                        std::cout 
                                                                << "Using archive as hardlink source: " 
                                                                << *si
                                                                << std::endl;
                                                        // youngest = ypath;
                                                        if (!hardlink_opt) {
                                                                /**/
                                                                argv.push_back(std::string("--hard-links"));
                                                                /**/
                                                                /*
                                                                command_line += " --hard-links";
                                                                */
                                                                hardlink_opt = true;
                                                        }

                                                        relative_path = mk_relative_path(ypath,path);
                                                        /*
                                                        command_line += " --link-dest=";
                                                        command_line += relative_path;
                                                        */
                                                        /**/
                                                        opt="--link-dest=";
                                                        opt += relative_path;
                                                        argv.push_back(opt);
                                                        /**/
                                                        first_hardlink = true;
                                                        linkdest_count++; // At most 20 link-dest options can be used w/ rsync
                                                }
                                                else {
                                                        std::cout
                                                                << "- No such path: " 
                                                                << ypath
                                                                << std::endl;
                                                }
                                        }
                                }
                                /*
                                if (youngest.size() > 0) {
                                        relative_path = mk_relative_path(youngest,path);
                                        command_line += " --hard-links --link-dest=";
                                        command_line += relative_path;
                                }
                                */
                        }

                        for (
                                ei = m_job->excludes.begin();
                                ei != m_job->excludes.end();
                                ++ei
                                )
                        {
                                /*
                                command_line += " --exclude-from=";
                                command_line += *ei;
                                */
                                /**/
                                opt = "--exclude-from=";
                                opt += *ei;
                                argv.push_back(opt);
                                /**/
                        }

                        for (
                                ii = m_job->includes.begin();
                                ii != m_job->includes.end();
                                ++ii
                                )
                        {
                                /*
                                command_line += " --include-from=";
                                command_line += *ii;
                                */
                                /**/
                                opt = "--include-from=";
                                opt += *ei;
                                argv.push_back(opt);
                                /**/
                        }

                        /*
                        command_line += " ";
                        command_line += m_job->generate_source_path(*pi);
                        */
                        /**/
                        argv.push_back(m_job->generate_source_path(*pi));
                        /**/
                        
                        /*
                        command_line += " ";
                        command_line += path;
                        */
                        /**/
                        argv.push_back(path);
                        /**/
        
                        /*
                        std::cout << "Command Line: " << command_line << std::endl;
                        */
                        /**/
                        std::cout << "Command Line:" << std::endl;
                        {
                                uint16 c;
                                std::cout << "  Binary: " << binary << std::endl;
                                for (c = 0; c < argv.size(); c++) {
                                        std::cout << "    Argv[" << c << "] = " << argv[c] << std::endl;
                                }
                        }
                        /**/

                        m_error_msg.erase();

                        t.start();
                        /*
                        exec.exec(command_line);
                        */
                        /**/
                        exec.exec(binary, argv);
                        /**/
                        mf_process_rsync_io(
                                exec, 
                                m_job->rsync_timeout,
                                files_total,
                                files_xferd,
                                size_total,
                                size_xferd,
                                overflow_detected
                                );
                        t.stop();

                        signal_num = 0;
                        if (exec.child_signaled()) {
                                std::cout 
                                        << "Rsync caught signal: [" 
                                        << exec.child_signal_no()
                                        << "] "
                                        << rsync_estat_str.signal(exec.child_signal_no())
                                        << std::endl;
                                signal_num = exec.child_signal_no();
                        }
                        std::cout
                                << "Rsync exit code: ["
                                << exec.child_exit_code()
                                << "] "
                                << rsync_estat_str.exit(exec.child_exit_code())
                                << std::endl;

                        exit_code = exec.child_exit_code();
                        if (exec.child_exited_normally() && (exit_code == 0))
                                done = true;
                        else if (overflow_detected) {
                                std::cout 
                                        << "Vault overflow detected"
                                        << std::endl;
                                break;
                        }
                        else {
                                ++num_retries;
                                logger.set_error_logging(true);
                        }

                        if (m_job->rsync_behavior[exit_code] == rsync_behavior::fail)
                        {
                                std::cout << "Failing, will not attempt to retry" << std::endl;
                                break;
                        }
                        if (m_job->rsync_behavior[exit_code] 
                                == rsync_behavior::retry_without_hardlinks)
                        {
                                std::cout << "Retrying without hardlinks..." << std::endl;
                                hardlink = false;
                        }

                        if ((!done) && (m_job->rsync_retry_delay > 0)) {
                                std::cout << "Waiting " << m_job->rsync_retry_delay 
                                        << " minutes before retrying..." << std::endl;
                                sleep( (m_job->rsync_retry_delay) * 60);
                        }
                }
                if (!done) {
                        if (num_retries >= m_job->rsync_retry_count) {
                                std::cout << "Retry count exceeded" << std::endl;
                        }
                        if (m_job->rsync_behavior[exit_code] == rsync_behavior::ok)
                                std::cout << "Ignoring rsync failure" << std::endl;
                        else {
                                std::cout << "Giving up on this path" << std::endl;
                                m_success = false;
                        }
                }
                reportio().write_report(
                        m_job->generate_source_path(*pi),
                        t,
                        exit_code,
                        signal_num,
                        m_error_msg
                        );
        }
}

void job_archiver::mf_process_report(const std::string& a_str)
{
        if (reportio().is_report(a_str)) {
                m_jpr = reportio().parse(a_str);
                m_jr.add_report(m_jpr);
        }
}

/** Process I/O from the child

        While there is I/O to be read, read and parse it.  When the end of a line is
        reached write that line to the log file.  If a_finalize is true, the flush
        the child I/O buffer string.
        
 */
void job_archiver::mf_process_child_io(bool a_finalize)
{
        estring lstr;
        bool io_flag = false;

        while (!m_exec.out_eof() && (a_finalize || m_exec.out_ready())) {
                int r;
                const int buffer_size = 128;
                char buffer[buffer_size] = { 0 };

                r = m_exec.out_read(buffer, buffer_size);
                if (r > 0) {
                        int c;

                        io_flag = true;
                        for (c = 0; c < r; ++c) {
                                if ((buffer[c] == '\r') || (buffer[c] == '\n')) {
                                        lstr = prefix();
                                        lstr += m_io_out;
                                        lstr += "\n";
                                        logger.write(lstr,0,configuration_manager::logging_rsync,m_child_pid);
                                        mf_process_report(lstr);
                                        m_io_out.erase();
                                }
                                else {
                                        m_io_out += buffer[c];
                                }
                        }
                }
        }
        if (a_finalize && (m_io_out.size() > 0)) {
                lstr = prefix();
                lstr += m_io_out;
                lstr += "\n";
                logger.write(lstr,0,configuration_manager::logging_rsync,m_child_pid);
                mf_process_report(lstr);
                m_io_out.erase();
        }

        while (!m_exec.err_eof() && (a_finalize || m_exec.err_ready())) {
                int r;
                const int buffer_size = 128;
                char buffer[buffer_size] = { 0 };

                r = m_exec.err_read(buffer, buffer_size);
                if (r > 0) {
                        int c;

                        io_flag = true;
                        for (c = 0; c < r; ++c) {
                                if ((buffer[c] == '\r') || (buffer[c] == '\n')) {
                                        lstr = prefix();
                                        lstr += m_io_err;
                                        lstr += "\n";
                                        logger.write(lstr,0,configuration_manager::logging_rsync,m_child_pid);
                                        mf_process_report(lstr);
                                        m_io_err.erase();
                                }
                                else {
                                        m_io_err += buffer[c];
                                }
                        }
                }
        }
        if (a_finalize && (m_io_err.size() > 0)) {
                lstr = prefix();
                lstr += m_io_err;
                lstr += "\n";
                logger.write(lstr,0,configuration_manager::logging_rsync,m_child_pid);
                mf_process_report(lstr);
                m_io_err.erase();
        }
        if (!io_flag)
                sleep(config.io_poll_interval());
}

/** Trim off all non-digit leading and trailing characters from a string */
void job_archiver::mf_trim_string(std::string& a_str)
{
        while ((a_str.size() > 0) && (!isdigit(a_str[0])))
                a_str.erase(0,1);
        while ((a_str.size() > 0) && (!isdigit(a_str[a_str.size()-1])))
                a_str.erase(a_str.size()-1,1);
}

/** Parse I/O from rsync

        Search for special output from rsync to tell us something about the number
        and size of files and files transfered.

 */
void job_archiver::mf_parse_rsync_io(
        const std::string a_str,
        uint64& a_files_total,
        uint64& a_files_xferd,
        uint64& a_size_total,
        uint64& a_size_xferd
        )
{
        estring estr;

        if (a_str.find("Number of files: ") == 0) {
                estr = a_str;
                mf_trim_string(estr);
                try {
                        a_files_total = estr;
                }
                catch(error e) {
                        estring es;

                        es = "Could not parse total number of files processed by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
                catch(...) {
                        error e = err_unknown;
                        estring es;

                        es = "Could not parse total number of files processed by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
        }
        else if (a_str.find("Number of files transferred: ") == 0) {
                estr = a_str;
                mf_trim_string(estr);
                try {
                        a_files_xferd = estr;
                }
                catch(error e) {
                        estring es;

                        es = "Could not parse total number of files transferred by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
                catch(...) {
                        error e = err_unknown;
                        estring es;

                        es = "Could not parse total number of files transferred by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
        }
        else if (a_str.find("Total file size: ") == 0) {
                estr = a_str;
                mf_trim_string(estr);
                try {
                        a_size_total = estr;
                }
                catch(error e) {
                        estring es;

                        es = "Could not parse total size of files processed by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
                catch(...) {
                        error e = err_unknown;
                        estring es;

                        es = "Could not parse total size of files processed by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
        }
        else if (a_str.find("Total transferred file size: ") == 0) {
                estr = a_str;
                mf_trim_string(estr);
                try {
                        a_size_xferd = estr;
                }
                catch(error e) {
                        estring es;

                        es = "Could not parse total size of files transferred by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
                catch(...) {
                        error e = err_unknown;
                        estring es;

                        es = "Could not parse total size of files transferred by rsync";
                        e.push_back(ERROR_INSTANCE(es));

                        // Not sure this is the best way to handle this...
                        std::cerr << e;
                }
        }
}

/** Process I/O from rsync

        If there is I/O from rsync to be read, read it and then send it through the
        parser.

 */
void job_archiver::mf_process_rsync_io(
        execute& a_exec, 
        uint16 a_timeout,
        uint64& a_files_total,
        uint64& a_files_xferd,
        uint64& a_size_total,
        uint64& a_size_xferd,
        bool& a_overflow_detected
        )
{
        size_t ro;
        size_t re;
        estring out;
        estring err;
        timer t;
        bool io_flag;
        char buffer[1024] = { 0 };
        char *ptr;

        ro = 1;
        re = 1;
        t.start();
        while ((ro != 0) || (re != 0) || a_exec.child_running()) {
                io_flag = false;
                ro = 0;
                re = 0;

                if (!a_overflow_detected) {
                        a_overflow_detected = vaulter.overflow();
                }

                m_error_msg.erase();

                if (a_exec.out_ready()) {
                        ro = read(a_exec.out_fd(), buffer, 1024);
                        if (ro > 0) {
                                io_flag = true;
                                t.start();
                                for (ptr = buffer; ptr != buffer+ro; ++ptr) {
                                        if ((*ptr != '\r') && (*ptr != '\n')) {
                                                out += *ptr;
                                        }
                                        else {
                                                reportio().write_rsync_out(out);
                                                out.erase();
                                        }
                                }
                        }
                }

                if (a_exec.err_ready()) {
                        re = read(a_exec.err_fd(), buffer, 1024);
                        if (re > 0) {
                                io_flag = true;
                                t.start();
                                for (ptr = buffer; ptr != buffer+re; ++ptr) {
                                        if ((*ptr != '\r') && (*ptr != '\n')) {
                                                err += *ptr;
                                        }
                                        else {
                                                reportio().write_rsync_err(err);
                                                err.erase();
                                        }
                                }
                        }
                }

                t.stop();
                if (t.duration_secs() > a_timeout) {
                        std::cerr << "*** Rsync program inactivity timeout" << std::endl;
                        a_exec.kill_child();
                        TRY_nomem(m_error_msg = "Rsync inactivity timeout");
                }

                if (!io_flag)
                        sleep(config.io_poll_interval());

        }
        if (out.size() > 0) {
                std::cerr << out << std::endl;
                mf_parse_rsync_io(
                        out, 
                        a_files_total, 
                        a_files_xferd, 
                        a_size_total,
                        a_size_xferd
                );
                out.erase();
        }
        if (err.size() > 0) {
                std::cerr << err << std::endl;
                mf_parse_rsync_io(
                        out, 
                        a_files_total, 
                        a_files_xferd, 
                        a_size_total,
                        a_size_xferd
                );
                err.erase();
        }
}

//-----------------------------------------------------------------------------

/** C'tor */
archive_manager::archive_manager()
{
        if (this != &archiver)
                throw(INTERNAL_ERROR(0,"Attempt to alocate multiple archive managers"));

        clear();
}

/** Clear the archive manager and clear the job list */
void archive_manager::clear(void)
{
        m_jobs.clear();
        m_initialized = false;
}

/** Initialize the archive manager

        Log the archive timestamp, select and prepare a vault.

 */
void archive_manager::init(void)
{
        timer t;
        estring lstr;

        lstr = "Archive Manager - Beginning initialization\n";
        logger.write(lstr);
        t.start();

        lstr = "Timestamp: ";
        lstr += config.timestamp().str();
        lstr += "\n";
        logger.write(lstr);

        // Select a vault?
        vaulter.select();
        lstr = "Vault selected: ";
        lstr += vaulter.vault();
        lstr += "\n";
        logger.write(lstr);

        reporter.vault().add_report(
                vault_stats_report(estring("Initial Stats:"),filesystem(vaulter.vault()))
                );

        // Prepare the vault?
        vaulter.prepare();

        t.stop();
        lstr = "Archive Manager - Finished initialization";
        lstr += ", duration: ";
        lstr += t.duration();
        lstr += "\n";
        logger.write(lstr);

        m_initialized = true;
}

/** Return the initialized status of the archive manager */
const bool archive_manager::initialized(void) const
{
        return(m_initialized);
}

/** Give a status report

        After so many minutes of inactivity write a report to the log file of our
        current status of affairs.

 */
void archive_manager::mf_log_status(void)
{
        static timer t;
        const timer::value_type timeout = 30;
        estring lstr;
        std::vector<job_archiver*>::const_iterator ji;
        uint64 jobs_pending = 0;
        uint64 jobs_processing = 0;
        uint64 jobs_completed =0;
        uint64 jobs_remaining = 0;

        if (!t.is_started())
                t.start();
        
        t.stop();
        if (t.duration_mins() < timeout)
                return;
        
        lstr  = "\n";
        lstr += "STATUS REPORT:\n";
        lstr += "================================================================\n";
        logger.write(lstr);
        for (ji = m_jobs.begin(); ji != m_jobs.end(); ++ji) {
                lstr = "[";
                switch ((*ji)->status()) {
                        case job_archiver::status_pending:
                                lstr += "Pending    ";
                                ++jobs_pending;
                                break;
                        case job_archiver::status_processing:
                                lstr += "Processing ";
                                ++jobs_processing;
                                break;
                        case job_archiver::status_reschedule:
                                lstr += "Reschedule ";
                                ++jobs_pending;
                                break;
                        case job_archiver::status_fatal_error:
                                lstr += "Fatal Error";
                                ++jobs_completed;
                                break;
                        case job_archiver::status_error:
                                lstr += "Error      ";
                                ++jobs_completed;
                                break;
                        case job_archiver::status_completed:
                                lstr += "Completed  ";
                                ++jobs_completed;
                                break;
                        case job_archiver::status_done:
                                lstr += "Done       ";
                                ++jobs_completed;
                                break;
                        default:
                                lstr += "<unknown>  ";
                                break;
                }
                lstr += "]: ";
                lstr += (*ji)->id();
                lstr += "\n";
                logger.write(lstr);
        }

        lstr  = "---------------------------------------------------------------\n";
        lstr += "     Jobs Pending: ";
        lstr += estring(jobs_pending);
        lstr += "\n";

        lstr += "  Jobs Processing: ";
        lstr += estring(jobs_processing);
        lstr += "\n";

        lstr += "   Jobs Completed: ";
        lstr += estring(jobs_completed);
        lstr += "\n";

        lstr += "            Total: ";
        lstr += estring(jobs_pending+jobs_processing+jobs_completed+jobs_remaining);
        lstr += "\n";

        lstr += "Overflow Detected: ";
        if (vaulter.overflow()) {
                lstr += "TRUE";
        }
        else {
                lstr += "false";
        }
        lstr += "\n";

        logger.write(lstr);
        logger.write("\n");
        t.start();
}

/** Archive jobs

        Create an archive directory.  Generate a to-do list of job archiver objects.
        Process the job archiver objects:
        - While there are less than rsync-parallel job archivers processing, start
                the first available job archiver.
        - Check the status of each job and process I/O from jobs underway.
        - Remove jobs that failed to start.
        - Possibly reschedule failed jobs.
        - Remove completed jobs from active processing.
        - Call mf_log_status().

 */
void archive_manager::archive(void)
{
        timer t;
        estring lstr;
        configuration_manager::jobs_type::const_iterator cji;
        int num_processing = 0;
        std::vector<job_archiver*>::iterator ji;
        uint64 num_completed = 0;
        bool overflow_detected = false;
        estring debug_estr;

        if (!initialized())
                throw(INTERNAL_ERROR(0,"Archive manager is not initialized"));

        lstr = "Archive Manager - Begin archiving\n";
        logger.write(lstr);
        t.start();

        // Create archive directory
        try {
                if (exists(archive_path())) {
                        lstr = "Found existing archive directory: ";
                        lstr += archive_path();
                        lstr += "\n";
                        logger.write(lstr);
                        rename_file(archive_path(), working_archive_path());
                }
                else if (exists(working_archive_path())) {
                        lstr = "Found existing archive directory: ";
                        lstr += working_archive_path();
                        lstr += "\n";
                        logger.write(lstr);
                }
                else {
                        lstr = "Creating archive directory: ";
                        lstr += working_archive_path();
                        lstr += "\n";
                        logger.write(lstr);
                        mk_dir(working_archive_path());
                }
        }
        catch(error e) {
                logger.write("An error has occured: ");
                logger.write(e[0].what());
                logger.write("\n");
                throw(e);
        }
        catch(...) {
                error e = err_unknown;

                logger.write("An error has occured: ");
                logger.write(e[0].what());
                logger.write("\n");
                throw(e);
        }

        // Create a todo list
        logger.write("Creating to-do list\n");
        for (
                cji = config.jobs().begin();
                cji != config.jobs().end();
                ++cji
                )
        {
                job_archiver* ptr;

                ptr = new job_archiver(&(*cji));
                if (ptr == 0)
                        throw(err_nomem);
                TRY_nomem(m_jobs.push_back(ptr));
        }

        // Backup clients
        logger.write("Processing jobs...\n");
        while (m_jobs.size() > num_completed) {

                /*
                logger.write("[DEBUG]: ---[ TOP OF LOOP ]---\n");

                debug_estr = "[DEBUG]: overflow_detected = ";
                debug_estr += estring(overflow_detected);
                debug_estr += "\n";
                logger.write(debug_estr);

                debug_estr = "[DEBUG]: num_processing = ";
                debug_estr += estring(num_processing);
                debug_estr += "\n";
                logger.write(debug_estr);
                */

                if (!overflow_detected) {
                        overflow_detected = vaulter.overflow(true);
                        /*
                        if (overflow_detected) {
                                logger.write("[DEBUG]: Variable Change :: ");
                                logger.write("overflow_detected = true");
                                logger.write("\n");
                        }
                        */
                }

                // If the vault has exceeded it's highwater mark, wait for the
                // currently-processing jobs to terminate, and then attempt to prepare the
                // vault.
                if (overflow_detected && (num_processing == 0)) {
                        TRY(vaulter.prepare(true),"Cannot complete archive");
                        overflow_detected = vaulter.overflow();
                        /*
                        if (!overflow_detected) {
                                logger.write("[DEBUG]: Variable Change :: ");
                                logger.write("overflow_detected = false");
                                logger.write("\n");
                        }
                        */
                }

                // For each job in the list...
                for (ji = m_jobs.begin(); ji != m_jobs.end(); ++ji)
                {
                        // While we're running less than rsync-parallel jobs, start new ones
                        if (
                                !overflow_detected
                                && (num_processing < config.rsync_parallel())
                                && ((*ji)->status() == job_archiver::status_pending)
                                )
                        {
                                try {
                                        (*ji)->start();
                                }
                                catch(error e) {
                                        if (e.num() == 12) {
                                                lstr = "Error starting job: Out of memory, will retry later\n";
                                                (*ji)->clear();
                                        }
                                        else {
                                                (*ji)->end();
                                                lstr = "Error starting job, aborting\n";
                                                // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                                num_processing--;
                                                reporter.jobs().add_report((*ji)->report());
                                        }
                                        logger.write(lstr);
                                }
                                catch(...) {
                                        (*ji)->end();
                                        lstr = "*** AN UNKNOWN ERROR HAS OCCURED WHILE STARTING JOB ";
                                        lstr += "-- JOB TERMINATED\n";
                                        logger.write(lstr);
                                        // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                        num_processing--;
                                        reporter.jobs().add_report((*ji)->report());
                                }
                                // logger.write("[DEBUG]: Variable Change :: num_processing++\n");
                                num_processing++;
                        }
        
                        // Process jobs
                        if ((*ji)->status() == job_archiver::status_processing) {
                                try {
                                        (*ji)->process();
                                }
                                catch(error e) {
                                        // TODO: Change 12 to ENOMEM?
                                        if (e.num() == 12) {
                                                lstr  = "Error starting job: Out of memory, will retry later\n";
                                                (*ji)->clear();
                                        }
                                        else {
                                                (*ji)->end();
                                                lstr = "Error starting job, aborting\n";
                                                // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                                num_processing--;
                                                reporter.jobs().add_report((*ji)->report());
                                        }
                                        logger.write(lstr);
                                }
                                catch(...) {
                                        (*ji)->end();
                                        lstr = "*** AN UNKNOWN ERROR HAS OCCURED WHILE PROCESSING JOB ";
                                        lstr += "-- JOB TERMINATED\n";
                                        logger.write(lstr);
                                        // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                        num_processing--;
                                        reporter.jobs().add_report((*ji)->report());
                                }
                        }

                        // Remove jobs that could not start from active duty
                        if ((*ji)->status() == job_archiver::status_reschedule) {
                                (*ji)->clear();
                                // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                num_processing--;
                        }

                        // If a job exited with an error, and the vault is full, then reschedule
                        // the job for later
                        if (
                                        ((*ji)->status() == job_archiver::status_error)
                                &&
                                        overflow_detected
                                )
                        {
                                lstr = "Vault overflow detected, will retry job later\n";
                                logger.write(lstr);
                                (*ji)->clear();
                                num_processing--;
                        }

                        // Remove completed jobs from the processing list
                        if (
                                ((*ji)->status() == job_archiver::status_completed)
                                || ((*ji)->status() == job_archiver::status_fatal_error)
                                || ((*ji)->status() == job_archiver::status_error)
                                ) {
                                (*ji)->end();
                                // logger.write("[DEBUG]: Variable Change :: num_processing--\n");
                                num_processing--;
                                num_completed++;

                                // logger.write("Adding job report to report manager\n");
                                reporter.jobs().add_report((*ji)->report());
                        }
                }

                mf_log_status();
                sleep(1);

                // logger.write("[DEBUG]: ---[ BOTTOM OF LOOP ]---\n");
        }

        t.stop();
        lstr = "Archive Manager - Finished archiving, duration: ";
        lstr += t.duration();
        lstr += "\n";
        logger.write(lstr);

        lstr = "Archive Manager - Finalizing archive path\n";
        logger.write(lstr);
        TRY(
                rename_file(working_archive_path(), archive_path()),
                "Cannot finalize archive"
                );

        reporter.vault().add_report(
                vault_stats_report(estring("Final Stats:"),filesystem(vaulter.vault()))
                );
}

/** Return an absolute path to the finished archive directory */
const std::string archive_manager::archive_path(void) const
{
        estring path;

        if (!initialized())
                throw(INTERNAL_ERROR(0,"Archive manager is not initialized"));

        path = vaulter.vault();
        path += "/";
        path += config.timestamp().str();

        return(path);
}

/** Return the absolute path to the unfinished working archive directory */
const std::string archive_manager::working_archive_path(void) const
{
        estring path;

        if (!initialized())
                throw(INTERNAL_ERROR(0,"Archive manager is not initialized"));
        
        path = archive_path();
        path += ".incomplete";

        return(path);
}

//-----------------------------------------------------------------------------

/** The global archive manager */
archive_manager archiver;

---