Added guts to JobPool

Added safety to ScannerPool and WorkerPool


git-svn-id: https://svn.microneil.com/svn/SNF4CGP/trunk@22 59e8e3e7-56fa-483b-b4b4-fa6ab0af3dfc

SNF4CGP/JobPool.cpp

@@ -50,7 +50,7 @@ void Job::emitERROR() {
 void Job::emitDISCARD() {
 }
 
-void job::finalize() {
+void Job::finalize() {
 }
 
 void Job::doWakeUp() {
@@ -101,7 +101,7 @@ Job::Job(ScannerPool& S, OutputProcessor& O) :
   ScanResultCode(0),
   ReadLength(0) {                                                               // Minimize heap thrashing.
       OutputBuffer.reserve(StringReserveSize);
-      HeadersToInject.researve(StringReserveSize);
+      HeadersToInject.reserve(StringReserveSize);
       ReadBuffer.reserve(ReadBufferSize);
 }
 
@@ -120,7 +120,10 @@ void Job::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;
 }
 
@@ -142,39 +145,84 @@ string& Job::Responses() {
 
 JobPool::JobPool() :                                                            // Simple constructor.
   Output_(0),
-  Scanners_(0),
+  Scanners_(new ScannerPool()),
   AllocatedJobs(0),
   Started(false) {
 }
 
 JobPool::~JobPool() {                                                           // Clean up on the way out.
-    try { stop(); }
+    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.
-    // Kill off all of the allocated Jobs.
-    // Shut down and destroy the ScannerPool.
+    ScopeMutex Busy(AllocationMutex);
+    while(0 < AllocatedJobs) killJobFromPool();
+    Scanners().stop();
+    Started = false;
 }
 

SNF4CGP/ScannerPool.cpp

@@ -66,7 +66,8 @@ snf_RulebaseHandler& ScannerPool::Rulebase() {
 
 void ScannerPool::init(const string Configuration) {                            // Get ready to provide scanners.
     Rulebase().open(Configuration.c_str(), "", "");                             // Light up the rulebase manager.
-    Scanner* FirstScanner = new Scanner(Rulebase());                            // Create our first scanner.
+    ScopeMutex Busy(AllocationControl);
+    Scanner* FirstScanner = makeScanner();                                      // Create our first scanner.
     PooledScanners.give(FirstScanner);                                          // Drop it into the pool.
     Started = true;
 }

SNF4CGP/WorkerPool.cpp

@@ -98,8 +98,9 @@ WorkerPool::~WorkerPool() {
 }
 
 void WorkerPool::init() {                                                       // Initialize the worker pool.
+    ScopeMutex Busy(AllocationMutex);
     Worker* FirstWorker = makeWorker();                                         // Make the first worker and
-    drop(*FirstWorker);                                                         // drop it into the pool.
+    RecycledWorkers.give(FirstWorker);                                          // drop it into the pool.
     Started = true;                                                             // We are now initialized.
 }