SNF4CGP/SNF4CGP/JobPool.cpp

// SNF4CGP/JobPool.cpp
// Copyright (C) 2009 ARM Research Labs, LLC.
// See www.armresearch.com for more information.

#include "JobPool.hpp"
#include "Command.hpp"
#include "ScannerPool.hpp"
#include "OutputProcessor.hpp"

#include "../SNFMulti/SNFMulti.hpp"

#include "../CodeDweller/threading.hpp"
#include "../CodeDweller/faults.hpp"

#include <iostream>
#include <vector>

using namespace std;

//// Tuning Constants //////////////////////////////////////////////////////////

const int ReadBufferSize = snf_ScanHorizon;
const int StringReserveSize = 2048;

//// Job ///////////////////////////////////////////////////////////////////////

void Job::emitCommentPrefix() {
}

void Job::emitComment(string Comment) {
    ostringstream O;
    O << "* " << Comment << endl;
    OutputBuffer.append(O.str());
}

void Job::emitResponsePrevix() {
}

void Job::emitOK() {
}

void Job::emitINTF() {
}

void Job::emitFAIL() {
}

void Job::emitADDHEADER(string Headers) {
}

void Job::emitERROR() {
}

void Job::emitDISCARD() {
}

void Job::finalize() {
    Output.outputJob(*this);
}

void Job::doWakeUp() {
    emitComment("SNF4CGP Waking Up");
    emitComment(CurrentCommand.Data);
}

void Job::doINTF() {
}

void Job::doFAIL() {
}

void Job::doFILE() {
}

void Job::readTopOfMessage(ifstream& Reader) {
}

void Job::doRead() {
}

void Job::doScan() {
}

void Job::doAction() {
}

void Job::doBypass() {
}

void Job::doAllow() {
}

void Job::doReject() {
}

void Job::doDelete() {
}

void Job::moveMessageToHoldPath(ifstream& Reader) {
}

void Job::doHold() {
}

Job::Job(ScannerPool& S, OutputProcessor& O) :                                  // Construct with important links.
  Scanners(S),
  Output(O),
  ScanResultCode(0),
  ReadLength(0) {                                                               // Minimize heap thrashing.
      OutputBuffer.reserve(StringReserveSize);
      HeadersToInject.reserve(StringReserveSize);
      ReadBuffer.reserve(ReadBufferSize);
}

Job::~Job() {                                                                   // Cleanup when destructing.
    try { clear(); }
    catch(...) {}
}

void Job::clear() {                                                             // Cleanup for the next command.
    CurrentCommand.clear();
    ScanResultCode = 0;
    OutputBuffer.clear();
    HeadersToInject.clear();
    MessageMoveFilePath.clear();
    ReadBuffer.clear();
    ReadLength = 0;
}

LogicFault FaultIfQuitGetsHere("Job::setCommand() Fault(Command::QUIT == C.Type)");

void Job::setCommand(Command& C) {                                              // Assign a command for this job.
    FaultIfQuitGetsHere(Command::QUIT == C.Type);
    CurrentCommand = C;
}

void Job::doIt() {                                                              // Get the job done.
    switch(CurrentCommand.Type) {
        case Command::WAKE: { doWakeUp(); break; }
        case Command::INTF: { doINTF(); break; }
        case Command::FILE: { doFILE(); break; }
        default: { doFAIL(); break; }
    }
    finalize();
}

string& Job::Responses() {                                                      // Read the job's report.
    return OutputBuffer;
}

//// JobPool ///////////////////////////////////////////////////////////////////

JobPool::JobPool() :                                                            // Simple constructor.
  Output_(0),
  Scanners_(new ScannerPool()),
  AllocatedJobs(0),
  Started(false) {
}

JobPool::~JobPool() {                                                           // Clean up on the way out.
    try {
        stop();
        Output_ = 0;
        if(Scanners_) delete Scanners_;
        Scanners_ = 0;
    }
    catch(...) {}
}

RuntimeCheck CheckForValidOutputPool("JobPool::Output() Check(0 != Output_)");

OutputProcessor& JobPool::Output() {
    CheckForValidOutputPool(0 != Output_);
    return (*Output_);
}

RuntimeCheck CheckForValidScannerPool("JobPool::Scanners() Check(0 != Scanners_)");

ScannerPool& JobPool::Scanners() {
    CheckForValidScannerPool(0 != Scanners_);
    return (*Scanners_);
}

// Watch out -- Initializing the JobPool also means starting up SNFMulti and
// the ScannerPool.

void JobPool::init(string Configuration, OutputProcessor& O) {                  // Initialize the JobPool.
    Output_ = &O;
    Scanners().init(Configuration);
    ScopeMutex Busy(AllocationMutex);
    Job* FirstJob = makeJob();
    Jobs.give(FirstJob);
    Started = true;
}

Job* JobPool::makeJob() {                                                       // Create and count a Job.
    Job* NewJob = new Job(Scanners(), Output());
    ++AllocatedJobs;
    return NewJob;
}

LogicCheck CheckGrabJobsOnlyWhenStarted("JobPool::grab() Check(Started)");
RuntimeCheck CheckForValidGrabbedJob("JobPool::grab() Check(0 != GrabbedJob)");

Job& JobPool::grab() {                                                          // Grab a job (prefer from the pool).
    ScopeMutex Busy(AllocationMutex);
    CheckGrabJobsOnlyWhenStarted(Started);
    Job* GrabbedJob = 0;
    if(0 < Jobs.size()) GrabbedJob = Jobs.take();                               // Prefer jobs from the pool.
    else GrabbedJob = makeJob();                                                // Make new ones if needed.
    CheckForValidGrabbedJob(0 != GrabbedJob);
    return (*GrabbedJob);
}

void JobPool::drop(Job& J) {                                                    // Drop a job into the pool.
    Jobs.give(&J);
}

void JobPool::killJobFromPool() {                                               // Kill and count a Job from the pool.
    Job* JobToKill = 0;
    JobToKill = Jobs.take();
    delete JobToKill;
    --AllocatedJobs;
}

// Watch out -- Stopping the JobPool also means shutting down SNFMulti and
// the ScannerPool.

void JobPool::stop() {                                                          // Shut down the JobPool.
    ScopeMutex Busy(AllocationMutex);
    while(0 < AllocatedJobs) killJobFromPool();
    Scanners().stop();
    Started = false;
}