|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802 |
- // snfXCImgr.cpp
- // Copyright (C) 2007 - 2020 ARM Research Labs, LLC
- // See www.armresearch.com for the copyright terms.
- //
- // See snfXCImgr.hpp for details.
-
- #include "SNFMulti.hpp"
- #include "snfXCImgr.hpp"
-
- namespace cd = codedweller;
-
- // snfXCIServerCommandHandler Virtual Base Class Default Processor.
-
- const std::string XCIServerCommandDefaultResponse =
- "<snf><xci><server><response message=\'Not Implemented\' code=\'-1\'/></server></xci></snf>\n";
-
- std::string snfXCIServerCommandHandler::processXCIRequest(snf_xci& X) { // A Server using SNFMulti
- return XCIServerCommandDefaultResponse; // can provide a useful processor.
- }
-
- // snfXCIJob encapsulates a single XCI transaction.
-
- void snfXCIJob::clear() { // Clear the buffers.
- Request.clear(); // Clear the request and
- Response.clear(); // response buffers.
- SetupTime = 0; // No setup time yet.
- }
-
- // snfXCIJobProcessor encapsulates the logic to respond to an XCI request.
-
- snfXCIJobProcessor::snfXCIJobProcessor(snf_RulebaseHandler* H) : // Setup scanner.
- myHome(H) { // Establish myHome from H.
- myEngine = new snf_EngineHandler(); // Create an engine handler and
- myEngine->open(H); // tie it in to our home rulebase.
- }
-
- snfXCIJobProcessor::~snfXCIJobProcessor() { // Tear down scanner.
- if(myEngine) { // Checking first that we have one,
- myEngine->close(); // close the engine and then
- delete myEngine; // delete it. Set the pointer to
- myEngine = 0; // NULL to enforce the point.
- }
- myHome = 0; // NULL out our home too.
- }
-
- //// This collection of functions handle the processing of all XCI requests.
-
- bool snfXCIJobProcessor::isScanJob() { // True if myXCI is a scan job.
- if(0 < myXCI.scanner_scan_file.length()) return true; // If we have a scan file: true!
- return false; // otherwise false.
- }
-
- bool snfXCIJobProcessor::isGBUdbJob() { // True if myXCI is a GBUdb job.
- if( // GBUdb jobs have either
- 0 < myXCI.gbudb_test_ip.length() || // an IP to test or
- 0 < myXCI.gbudb_set_ip.length() || // an IP to setup or
- 0 < myXCI.gbudb_bad_ip.length() || // a bad IP to flag or
- 0 < myXCI.gbudb_good_ip.length() || // a good IP to flag or
- 0 < myXCI.gbudb_drop_ip.length()
- ) return true; // If we have one of these: true!
- return false; // otherwise false.
- }
-
- bool snfXCIJobProcessor::isReportJob() { // True if myXCI is a Report job.
- if(0 < myXCI.report_request_status_class.length()) return true; // If we have a report status class
- return false; // it's a report otherwise it's not.
- }
-
- bool snfXCIJobProcessor::isCommandJob() { // True if myXCI is a Command job.
- if(0 < myXCI.xci_server_command.length()) return true; // If we have a command string: true!
- return false; // otherwise false.
- }
-
- void snfXCIJobProcessor::processScan(snfXCIJob& J) { // Process a scan request.
- try { // Safely perform our scan.
-
- // Check for forced IP.
-
- cd::IP4Address ForcedIP = 0UL; // Don't expect a forced IP.
- if(0 < myXCI.scanner_scan_ip.length()) { // If we have one then
- ForcedIP = myXCI.scanner_scan_ip; // convert it from the string.
- }
-
- // Scan the message file.
-
- int ScanResult = // Scan the file using our
- myEngine->scanMessageFile( // engine. Use the file
- myXCI.scanner_scan_file.c_str(), // path in the XCI request, and
- J.SetupTime, // the recorded setup time. Use the
- ForcedIP // forced IP if provided.
- );
-
- // Create a proper xci resposne.
-
- std::ostringstream ResultString; // Use a stringstream to make it easier.
- ResultString
- << "<snf><xci><scanner><result code=\'" // Emit the preamble.
- << ScanResult << "\'"; // Emit the scan result.
-
- if( // Check for optional data requests.
- false == myXCI.scanner_scan_xhdr &&
- false == myXCI.scanner_scan_log
- ) { // If no optional data was requested
- ResultString // then close the <request/> and
- << "/></scanner></xci></snf>\n" // emit the closing elements.
- << std::endl; // End of the line.
-
- } else { // If optional data is requested:
- ResultString << ">" << std::endl; // Complete the <result> open tag.
-
- if(true == myXCI.scanner_scan_xhdr) { // Optionally include XHDR data.
- ResultString // If xheaders are requested...
- << "<xhdr>" << myEngine->getXHDRs() // Emit the xhdr element & contents.
- << "</xhdr>" << std::endl; // End the xhdr and end of line.
- }
-
- if(true == myXCI.scanner_scan_log) { // Optionally include XMLLog data.
- ResultString // If the log data is requested...
- << "<log>" << myEngine->getXMLLog() // Emit the log element & data.
- << "</log>" << std::endl; // End the log data and end of line.
- }
-
- ResultString << "</result></scanner></xci></snf>\n"; // Emit the closing elements.
- }
-
- J.Response = ResultString.str(); // Capture the formatted response.
- }
-
- // Decode the known exceptions
-
- catch(const snf_EngineHandler::AllocationError& e) {
- J.Response = "<snf><xci><error message=\'AllocationError ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::BadMatrix& e) {
- J.Response = "<snf><xci><error message=\'BadMatrix ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::Busy& e) {
- J.Response = "<snf><xci><error message=\'Busy ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::FileError& e) {
- J.Response = "<snf><xci><error message=\'FileError ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::MaxEvals& e) {
- J.Response = "<snf><xci><error message=\'MaxEvals ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::Panic& e) {
- J.Response = "<snf><xci><error message=\'Panic ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(snf_EngineHandler::XHDRError& e) {
- J.Response = "<snf><xci><error message=\'XHDRError ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- // Decode the unknown exceptions
-
- catch(const std::exception& e) {
- J.Response = "<snf><xci><error message=\'Exception! ";
- J.Response.append(e.what());
- J.Response.append("\'/></xci></snf>\n");
- }
-
- catch(...) {
- J.Response = "<snf><xci><error message=\'... Thrown!\'/></xci></snf>\n";
- }
- }
-
- std::string snfXCIJobProcessor::processGBUdb() { // Process a GBUdb request.
- GBUdb& myGBUdb = myHome->MyGBUdb; // Make a convenient GBUdb handle.
- cd::IP4Address IP; // We will work with an IP.
- GBUdbRecord R; // We will get a record to return.
-
- // Test an IP - return it's current data.
-
- if(0 < myXCI.gbudb_test_ip.length()) { // IF: Test an IP
- IP = myXCI.gbudb_test_ip; // Convert the IP.
- } else
-
- // Set or update an IP's data.
-
- if(0 < myXCI.gbudb_set_ip.length()) { // IF: Set an IP's data.
- IP = myXCI.gbudb_set_ip; // Convert the IP.
- if( // Check for a compound update:
- 0 <= myXCI.gbudb_set_bad_count || // If we are changing the bad
- 0 <= myXCI.gbudb_set_good_count // or good count then this is
- ) { // a compound update (read then write).
- R = myGBUdb.getRecord(IP); // Get the record (or a safe blank).
- if(0 <= myXCI.gbudb_set_bad_count) // If we have a bad count to set
- R.Bad(myXCI.gbudb_set_bad_count); // then set the bad count.
- if(0 <= myXCI.gbudb_set_good_count) // If we have a good count to set
- R.Good(myXCI.gbudb_set_good_count); // then set the good count.
- if(0 < myXCI.gbudb_set_type.length()) { // If type, set type...
- switch(myXCI.gbudb_set_type.at(0)) { // Determine the type based on the
- case 'g': case 'G': { R.Flag(Good); break; } // first character of the name and
- case 'b': case 'B': { R.Flag(Bad); break; } // set the appropriate flag.
- case 'u': case 'U': { R.Flag(Ugly); break; }
- case 'i': case 'I': { R.Flag(Ignore); break; }
- }
- }
- myGBUdb.setRecord(IP, R); // Save the data.
-
- } else // This might be a simple flag change.
- if(0 < myXCI.gbudb_set_type.length()) { // If type, set type...
- switch(myXCI.gbudb_set_type.at(0)) { // Determine the type based on the
- case 'g': case 'G': { R = myGBUdb.setGood(IP); break; } // first character of the name and
- case 'b': case 'B': { R = myGBUdb.setBad(IP); break; } // set the appropriate flag. Simple
- case 'u': case 'U': { R = myGBUdb.setUgly(IP); break; } // flag changes are atomic so there is
- case 'i': case 'I': { R = myGBUdb.setIgnore(IP); break; } // no need to "save" later.
- }
- } else { // Empty set command?
- return XCIBadSetResponse; // That's bad. Use test!
- }
- } else
-
- // Add a bad event to an IPs data.
-
- if(0 < myXCI.gbudb_bad_ip.length()) { // IF: Add a bad mark for this IP
- IP = myXCI.gbudb_bad_ip; // Convert the IP.
- R = myGBUdb.addBad(IP); // Add a bad mark.
- } else
-
- // Add a good event to an IPs data.
-
- if(0 < myXCI.gbudb_good_ip.length()) { // IF: Add a good mark for this IP
- IP = myXCI.gbudb_good_ip; // Convert the IP.
- R = myGBUdb.addGood(IP); // Add a bad mark.
- } else
-
- // Drop an IP from the database.
-
- if(0 < myXCI.gbudb_drop_ip.length()) { // IF: Drop an IP's data.
- IP = myXCI.gbudb_drop_ip; // Convert the IP.
- myGBUdb.dropRecord(IP); // Forget about it.
- }
-
- // Return the final state of the IP's data.
-
- IPTestRecord IPState(IP);
- myHome->performIPTest(IPState);
-
- std::ostringstream Response; // Use a stringstream for our output.
- Response
- << "<snf><xci><gbudb><result " // Get the response started.
- << "ip=\'" << (std::string) IP // Emit the ip.
- << "\' type=\'" // Emit the type.
- << ((Good == IPState.G.Flag()) ? "good" :
- ((Bad == IPState.G.Flag()) ? "bad" :
- ((Ugly == IPState.G.Flag()) ? "ugly" :
- ((Ignore == IPState.G.Flag()) ? "ignore" : "error"))))
- << "\' p=\'" << IPState.G.Probability() // Emit the probability.
- << "\' c=\'" << IPState.G.Confidence() // Emit the confidence.
- << "\' b=\'" << IPState.G.Bad() // Emit the bad count.
- << "\' g=\'" << IPState.G.Good() // Emit the good count.
- << "\' range=\'"
- << ((snfIPRange::Unknown == IPState.R) ? "unknown" :
- ((snfIPRange::White == IPState.R) ? "white" :
- ((snfIPRange::Normal == IPState.R) ? "normal" :
- ((snfIPRange::New == IPState.R) ? "new" :
- ((snfIPRange::Caution == IPState.R) ? "caution" :
- ((snfIPRange::Black == IPState.R) ? "black" :
- ((snfIPRange::Truncate == IPState.R) ? "truncate" : "error")))))))
- << "\' code=\'" << IPState.Code
- << "\'"
- << "/></gbudb></xci></snf>" // Finish it up.
- << std::endl;
-
- return Response.str(); // Return the formatted response.
- }
-
- std::string snfXCIJobProcessor::processStatusReport() { // Process a report request.
- std::string ReportToSend; // Keep this in scope.
-
- if(0 == myXCI.report_request_status_class.find("hour")) { // Please send the hour report.
- ReportToSend = myHome->MyLOGmgr.getStatusHourReport();
- } else
-
- if(0 == myXCI.report_request_status_class.find("minute")) { // Please send the minute report.
- ReportToSend = myHome->MyLOGmgr.getStatusMinuteReport();
- } else { // Please send the second report.
- ReportToSend = myHome->MyLOGmgr.getStatusSecondReport();
- }
-
- std::string Response = "<snf><xci><report><response>"; // Construct the response using the
- Response.append(ReportToSend); // snf/xci template and the selected
- Response.append("</response></report></xci></snf>"); // status report text.
-
- return Response; // Return the response.
- }
-
- void snfXCIJobProcessor::process(snfXCIJob& J) { // Process a Job.
-
- // Parse the XCI request and check for an error.
-
- myXCI.read(J.Request); // Parse the request.
- if(myXCI.bad()) { // If it's bad then
- J.Response = XCIErrorResponse; // respond with an error.
- myHome->logThisError("XCI",-1,"Bad Request"); // Log the error.
- return; // Done.
- } else
-
- // Process scan requests.
-
- if(isScanJob()) { // If this is a Scan request
- processScan(J); // respond with the result.
- return; // Done.
- } else
-
- // Process gbudb requests.
-
- if(isGBUdbJob()) { // If this is a GBUdb request
- J.Response = processGBUdb(); // respond with the result.
- return; // Done.
- } else
-
- // Process report requests.
-
- if(isReportJob()) { // If this is a Status report request
- J.Response = processStatusReport(); // respond with the desired report.
- return; // Done.
- } else
-
- // Process server commands.
-
- if(isCommandJob()) { // If this is a server command
- J.Response = myHome->processXCIServerCommandRequest(myXCI); // pass it up and return the
- return; // result. Done.
- } else
-
- // If we get to this point we don't understand the well formed request.
-
- J.Response = XCIErrorResponse; // Don't understand?
- myHome->logThisError("XCI",-2,"Unrecognized Request"); // Log the error. Respond with
- return; // the standard error response.
- }
-
- // ChannelJob encapsulates a Client Job while in the queue and how long it has
- // been in the system (since created).
-
- ChannelJob::ChannelJob() : myClient(0) {} // Empty is the null client.
-
- ChannelJob::ChannelJob(cd::TCPClient* C) : // We are created like this.
- myClient(C) { // We capture the client and
- } // our timer starts automaticially.
-
- cd::msclock ChannelJob::Age() { // How old is this job?
- return Lifetime.getElapsedTime(); // Return the elapsed time in ms.
- }
-
- cd::TCPClient* ChannelJob::Client() { // What client does it hold?
- return myClient; // Return the Client pointer.
- }
-
- // snfXCITCPChannel encapsulates the logic to queue and handle TCPClients for
- // the XCI interface. The queued TCPClients each represent a single request.
- // Each request is handled in turn by reading the request into an snfXCIJob,
- // handing that snfXCIJob to an snfXCIJobProcessor, transmitting the result
- // back to the TCPClient, closing the connection, and recycling the snfXCIJob
- // object for the next round.
-
- // snfXCITCPChannel shuts down when given a NULL TCPClient; This allows any
- // jobs in queue to be handled before the thread stops. To shut down a channel
- // { C->submit(NULL); C->join(); delete C; C = NULL;}
-
- void snfXCITCPChannel::give(ChannelJob& J) { // Give a job to the queue.
- cd::ScopeMutex OneAtATimePlease(QueueMutex); // Protected with a mutex...
- JobQueue.push(J); // Push the job in.
- LatestSize = JobQueue.size(); // Set the blinking light.
- QueueGateway.produce(); // Add the item to our gateway.
- }
-
- ChannelJob snfXCITCPChannel::take() { // Take a job from the queue.
- QueueGateway.consume(); // Hold on until there is work.
- cd::ScopeMutex OneAtATimePlease(QueueMutex); // Queue Data Protected with a mutex.
- ChannelJob J = JobQueue.front(); // Grab the next job in the queue.
- JobQueue.pop(); // Pop that job out of the queue.
- LatestSize = JobQueue.size(); // Set the blinking light.
- return J; // Return the Job.
- }
-
- const int RWTimeLimit = 30000; // RWTimeLimit in ms. 30 seconds.
- const std::string endSNF = "</snf>"; // snf_xci snf element terminator.
- const int RWPollMin = 15; // Minimum time between polls.
- const int RWPollMax = 75; // Maximum time between polls.
- const int MaxQueueLength = 32; // Most waiting in any queue.
- const int MaxTCPQueueLength = 4 * MaxQueueLength; // Most connections waiting.
-
- void snfXCITCPChannel::readRequest(cd::TCPClient* Client) { // Read Job.Request from Client.
- cd::Timeout ReadTimeLimit(RWTimeLimit); // We have time limits.
- cd::PollTimer ReadThrottle(RWPollMin, RWPollMax); // Throttle with a spiral delay.
- while(
- false == ReadTimeLimit.isExpired() && // Read stuff until we're out of time
- std::string::npos == Job.Request.find(endSNF,0) // or we have a complete request.
- ) {
- memset(LineBuffer, 0, sizeof(LineBuffer)); // Clear the buffer.
- int bytes = Client->delimited_receive( // Read up to all but one byte
- LineBuffer, sizeof(LineBuffer)-1, '\n'); // of the buffer up to the first \n.
- if(0 < bytes) { // If we got some bytes
- Job.Request.append(LineBuffer); // Append the data we got and
- ReadThrottle.reset(); // reset the throttle.
- } else { // If we didn't get any bytes then
- ReadThrottle.pause(); // wait a little bit more each round.
- }
- } // When we're done we will return.
- }
-
- void snfXCITCPChannel::writeResponse(cd::TCPClient* Client) { // Write Job.Request from Client.
- cd::Timeout WriteTimeLimit(RWTimeLimit); // We have a time limit.
- cd::PollTimer WriteThrottle(RWPollMin, RWPollMax); // Throttle with a spiral delay.
- for( // For all the bytes in the response:
- int Length = Job.Response.length(), BytesThisTime = 0, Bytes = 0; // Bytes to send, this time and sent.
- Bytes < Length && // Keep going if we've got more to
- false == WriteTimeLimit.isExpired(); // send and we still have time.
- ) {
- BytesThisTime = Client->transmit( // Transmit some bytes.
- &Job.Response[Bytes], Job.Response.length()-Bytes); // from where we are, what is left.
- if(0 < BytesThisTime) { // If we sent bytes
- Bytes += BytesThisTime; // then keep track of how many
- WriteThrottle.reset(); // and reset our throttle to min.
- } else { // If we didn't then pause a bit
- WriteThrottle.pause(); // and let our delay grow.
- }
- }
- }
-
- const int XCI_Reading = 0; // XCI Mode Flags.
- const int XCI_Processing = 1;
- const int XCI_Writing = 2;
-
- void snfXCITCPChannel::myTask() { // Thread's main loop.
- bool WeAreAlive = true; // It's not over 'til it's over.
- while(WeAreAlive) { // While we are alive:
- CurrentThreadState(XCI_Wait); // Mark our state.
- ChannelJob J = take(); // Pull a Client Job from the queue.
- if(0 == J.Client()) { // If the job is empty we're done.
- CurrentThreadState(XCI_Shutdown); // Mark our state.
- WeAreAlive = false; // Turn off the alive flag and
- break; // break out of the loop.
-
- } else { // When we have a job to do:
- int XCIMode = XCI_Reading;
- try {
- CurrentThreadState(XCI_Read);
- XCIMode = XCI_Reading; // Now we are reading.
- readRequest(J.Client()); // Read the client job.
-
- CurrentThreadState(XCI_Process);
- XCIMode = XCI_Processing; // Now we are processing.
- Job.SetupTime = J.Age(); // Capture the read and queue time.
- Processor.process(Job); // Pass the XCIJob to our processor.
-
- CurrentThreadState(XCI_Write);
- XCIMode = XCI_Writing; // Now we are writing.
- writeResponse(J.Client()); // Write the response.
- }
-
- // Log any exceptions that were thrown.
-
- catch(...) {
- switch(XCIMode) {
- case XCI_Reading: {
- myHome->logThisError("XCI",-5,"SocketReadError");
- break;
- }
- case XCI_Processing: {
- myHome->logThisError("XCI",-6,"ProcessError");
- break;
- }
- case XCI_Writing: {
- myHome->logThisError("XCI",-7,"SocketWriteError");
- break;
- }
- }
- }
- }
-
- // At the end of every job we clean up no matter what.
-
- if(0 != J.Client()) { // If we have a client
- CurrentThreadState(XCI_Close);
- J.Client()->close(); // Close the client.
- delete J.Client(); // Delete the client.
- }
-
- CurrentThreadState(XCI_Clear);
- Job.clear(); // Clear the job buffer.
- } // Go again.
- }
-
- const cd::ThreadType snfXCITCPChannel::Type("snfXCITCPChannel"); // The thread's type.
-
- //// XCI Thread States
-
- const cd::ThreadState snfXCITCPChannel::XCI_Wait("Waiting For Take()");
- const cd::ThreadState snfXCITCPChannel::XCI_Read("Reading Request");
- const cd::ThreadState snfXCITCPChannel::XCI_Process("Processing Job");
- const cd::ThreadState snfXCITCPChannel::XCI_Write("Writing Results");
- const cd::ThreadState snfXCITCPChannel::XCI_Close("Closing Connection");
- const cd::ThreadState snfXCITCPChannel::XCI_Clear("Clearing Workspace");
- const cd::ThreadState snfXCITCPChannel::XCI_Shutdown("Shutting Down");
-
- snfXCITCPChannel::snfXCITCPChannel(snf_RulebaseHandler* H, std::string N) : // Create these with a home rulebase.
- Thread(snfXCITCPChannel::Type, N), // XCI TCP Channel Type & name.
- myHome(H), // We know our home.
- Processor(H), // Our processor has a rulebase.
- LatestSize(0) { // Our job queue size is zero.
- run(); // We start our thread.
- }
-
- snfXCITCPChannel::~snfXCITCPChannel() { // Destroy them very carefully.
- ChannelJob EndJob; // On the way down feed ourselves
- give(EndJob); // an empty job - that will end our
- join(); // thread once other jobs are done.
- myHome = 0; // Once joined our home is gone.
- } // We're done.
-
- int snfXCITCPChannel::Size() { // Keep track of how full they are.
- return LatestSize; // Flash the blinking light.
- }
-
- void snfXCITCPChannel::submit(cd::TCPClient* C) { // This is how we submit jobs.
- ChannelJob J(C); // Create a Job for this client.
- give(J); // Give it (copy) to the queue.
- }
-
- // snfXCImgr encapsulates a service engine that takes XCI requests via TCP,
- // performs the required actions, and returns an XCI response. It also checks
- // to see if the configuration for the XCI interface has changed.
-
- void snfXCImgr::checkCFG() { // Checks the configuration.
- CurrentThreadState(XCI_CheckConfig); // Update our status.
- int NEW_XCI_Port; // Prepare for a change in port.
-
- // Quickly as we can, grab a config packet, capture the XCI parts, and
- // then let it go.
-
- if(myHome->isReady()) { // If we know our home then
- snfCFGPacket MyCFGPacket(myHome); // Grab a configuration packet.
- if(MyCFGPacket.bad()) { // If it's not valid then
- return; // wait (skip this) till next time.
- } else { // If we've got a good config then
- CFG_XCI_ON = MyCFGPacket.Config()->XCI_OnOff; // Is XCI turned on?
- NEW_XCI_Port = MyCFGPacket.Config()->XCI_Port; // What port we listen to?
- } // If our rulebase manager was
- } else return; // not ready (skip this) for now.
-
- if(CFG_XCI_ON) { // If the XCI is configured up:
-
- if(NEW_XCI_Port != CFG_XCI_PORT) { // Check for a port change. If the
- CFG_XCI_PORT = NEW_XCI_Port; // port changed then check for a live
- if(Listener) { // listener. For a live port change
- shutdown_Listener(); // shut down the current listener and
- myHome->logThisInfo("XCI", 0, "ListenerDown:PortChanged"); // log the activity.
- startup_Listener(); // Restart the listener with the new
- myHome->logThisInfo("XCI", 0, "ListenerUp:PortChanged"); // port and log the event.
- }
- }
-
- startup_XCI(); // Make sure the XCI is up.
-
- } else { // If the XCI is configured down
- shutdown_XCI(); // then make sure it is down.
- }
- }
-
- snfXCITCPChannel* LowestQueue(snfXCITCPChannel* A, snfXCITCPChannel* B) { // Pick the lowest queue of two.
- return ((A->Size() < B->Size()) ? A : B); // Pick one and return it.
- }
-
- snfXCITCPChannel* snfXCImgr::BestAvailableChannel() { // Selects XCI channel w/ lowest queue.
- return LowestQueue( // Pick the lowest of the lowest.
- LowestQueue(
- LowestQueue(C0, C1),
- LowestQueue(C2, C3)
- ),
- LowestQueue(
- LowestQueue(C4, C5),
- LowestQueue(C6, C7)
- )
- );
- }
-
- void snfXCImgr::startup_Listener() { // Listener startup function.
- if(0 == Listener) { // If we need a new listener:
- Listener = new cd::TCPListener(CFG_XCI_PORT); // Create a new listener.
- Listener->MaxPending = MaxTCPQueueLength; // We may get a lot of hits ;-)
- Listener->open(); // Open it for business.
- Listener->makeNonBlocking(); // Make it non-blocking.
- }
- }
-
- void snfXCImgr::shutdown_Listener() { // Listener shutdown function.
- if(Listener) { // Only act if there is a listener:
- Listener->close(); // The listener gets closed,
- delete Listener; // then deleted, then the
- Listener = 0; // Listener pointer is zeroed.
- }
- }
-
- void snfXCImgr::startup_XCI() { // XCI startup function.
- if(true == XCI_UP) return; // If we're already up we're done.
- cd::ScopeMutex IGotIt(ChannelMutex); // Serialize state control for safety.
- if(myHome) { // We need to know our home.
- if(CFG_XCI_ON) { // If XCI is configured on, startup!
- C0 = new snfXCITCPChannel(myHome, "C0"); // Launch our 8 processing channels.
- C1 = new snfXCITCPChannel(myHome, "C1");
- C2 = new snfXCITCPChannel(myHome, "C2");
- C3 = new snfXCITCPChannel(myHome, "C3");
- C4 = new snfXCITCPChannel(myHome, "C4");
- C5 = new snfXCITCPChannel(myHome, "C5");
- C6 = new snfXCITCPChannel(myHome, "C6");
- C7 = new snfXCITCPChannel(myHome, "C7");
- startup_Listener(); // Start up our listener.
- myHome->logThisInfo("XCI", 0, "Startup"); // Log the startup.
- XCI_UP = true; // Set the flag. We're up!
- }
- }
- }
-
- void snfXCImgr::shutdown_XCI() { // XCI shutdown function.
- if(false == XCI_UP) return; // If we're already down we're done.
- cd::ScopeMutex IGotIt(ChannelMutex); // Serialize state control for safety.
- shutdown_Listener(); // If up, take down & 0 the Listener.
- if(C0) { delete C0; C0 = 0; } // If up, take C0 down and NULL it.
- if(C1) { delete C1; C1 = 0; } // If up, take C1 down and NULL it.
- if(C2) { delete C2; C2 = 0; } // If up, take C2 down and NULL it.
- if(C3) { delete C3; C3 = 0; } // If up, take C3 down and NULL it.
- if(C4) { delete C4; C4 = 0; } // If up, take C4 down and NULL it.
- if(C5) { delete C5; C5 = 0; } // If up, take C5 down and NULL it.
- if(C6) { delete C6; C6 = 0; } // If up, take C6 down and NULL it.
- if(C7) { delete C7; C7 = 0; } // If up, take C7 down and NULL it.
-
- myHome->logThisInfo("XCI", 0, "Shutdown"); // Log the shutdown.
- XCI_UP = false; // Set the flag. We're down!
- }
-
- int snfXCImgr::pollLoopCount() { // Retrieve & reset Loop Count.
- int x = diagLoopCount;
- diagLoopCount = 0;
- return x;
- }
-
- int snfXCImgr::pollClientCount() { // Retrieve & reset Client Count.
- int x = diagClientCount;
- diagClientCount = 0;
- return x;
- }
-
- const cd::ThreadState snfXCImgr::XCI_InitialConfig("Initial Config"); // Getting initial configuration.
- const cd::ThreadState snfXCImgr::XCI_InitialStartup("Initial Startup"); // Performing first startup.
- const cd::ThreadState snfXCImgr::XCI_CheckConfig("Checking Config"); // Checking configuration.
- const cd::ThreadState snfXCImgr::XCI_PollingListener("Polling Listener"); // Polling Listener for jobs.
- const cd::ThreadState snfXCImgr::XCI_SubmittingJob("Submitting Job"); // Submitting a new job.
- const cd::ThreadState snfXCImgr::XCI_ListenerDown("Listener Down!"); // Listener is down.
- const cd::ThreadState snfXCImgr::XCI_Stopping("Exited Polling Loop"); // XCImgr Exiting Big Loop
-
- void snfXCImgr::myTask() { // Main thread task.
- cd::PollTimer PollingThrottle(RWPollMin, RWPollMax); // Set up a dynamic delay.
- cd::Timeout WaitForCFG(1000); // CFG Check every second or so.
-
- // Wait for our initial configuration.
- CurrentThreadState(XCI_InitialConfig); // Update our status.
-
- cd::Sleeper WaitATic(1000); // One second sleeper.
- while(false == CFG_XCI_ON) { // Before we've been turned on
- if(TimeToStop) return; // loop unless it's time to stop.
- checkCFG(); WaitForCFG.restart(); // Check our configuration
- WaitATic(); // every second or so.
- }
-
- // Once our configuration is good and we are turned on we get here.
-
- try { // Safely accept/process requests.
-
- CurrentThreadState(XCI_InitialStartup); // Update our status.
-
- startup_XCI(); // We're on, so turn on!
-
- while(false == TimeToStop) { // While it is not time to stop:
-
- // Occasionally we check to see what our configuration says. If
- // the XCI is configured up, or down, or if the port changes then
- // the checkCFG() function handles the changes. After that all we
- // need to do here is check for a listener -- if we're up we will
- // have one and if not then we won't. Without a listener we will
- // slow down and keep checking for a configuration change.
-
- if(WaitForCFG.isExpired()) { checkCFG(); WaitForCFG.restart(); } // Check the CFG periodically.
-
- // Get a new client if we have room in the queue
- // and the listener is live.
-
- int JobsThisRound = 0; // Keep track of each batch.
- if(Listener) { // Check for a good listener.
- CurrentThreadState(XCI_PollingListener); // Update our status.
- cd::TCPClient* NewClient; // This will be our client.
- do { // Fast as we can - grab the work:
- ++diagLoopCount; // Count Polling Loops.
- NewClient = 0; // Clear our client pointer.
- snfXCITCPChannel* Channel = BestAvailableChannel(); // Pick a channel to use then
- if(MaxQueueLength > Channel->Size()) { // If we have room in the queue
- NewClient = Listener->acceptClient(); // get a new client.
- if(NewClient) { // If we got one:
- CurrentThreadState(XCI_SubmittingJob); // Update our status.
- ++diagClientCount; // Count Clients.
- NewClient->makeNonBlocking(); // Make the client non-blocking.
- Channel->submit(NewClient); // Submit the new client.
- }
- }
- } while( // Keep getting work in this tight
- (0 != NewClient)&& // loop until we run out of work
- (MaxTCPQueueLength > diagClientCount) // or we've pulled a full queue.
- );
- } else {
- CurrentThreadState(XCI_ListenerDown); // Update our status.
- } // Throttle our loop to keep it real:
- if(0 == JobsThisRound) PollingThrottle.pause(); // If we got nothing then slow down.
- else PollingThrottle.reset(); // If we got some, keep getting it!
- } // When we're done with the big loop:
- CurrentThreadState(XCI_Stopping); // Update our status.
- shutdown_XCI(); // Shutdown if we're not already.
- } // End of the active section.
-
- catch(const std::exception& e) { // If we get a knowable exception
- myHome->logThisError("XCI", -9, e.what()); // then we report it in detail,
- try { shutdown_XCI(); } catch(...) {} // shutdown if we're not already,
- WaitATic(); // wait a tic and try again.
- }
-
- catch(...) { // If we have an unhandled exception
- myHome->logThisError("XCI", -10, "Panic!"); // Panic and reset. Notify the log.
- try { shutdown_XCI(); } catch(...) {} // Shutdown if we're not already.
- WaitATic(); // Pause to let things settle.
- } // Let's try this again.
- }
-
-
- const cd::ThreadType snfXCImgr::Type("snfXCIManager"); // The thread's type.
- const int XCI_Default_Port = 9001; // Listener Default port = 9001.
-
- snfXCImgr::snfXCImgr() : // Construct with no home.
- cd::Thread(snfXCImgr::Type, "XCI Manager"), // XCI Manager type and Name.
- CFG_XCI_ON(false), // Everything starts off,
- CFG_XCI_PORT(XCI_Default_Port), // default, and
- myHome(0), // nulled.
- C0(0), C1(0), C2(0), C3(0),
- Listener(0),
- XCI_UP(false),
- diagLoopCount(0), diagClientCount(0),
- TimeToStop(true) { // We don't run until linkHome().
- }
-
- snfXCImgr::~snfXCImgr() { // Stop when we are destroyed.
- stop(); // Like I said, stop().
- }
-
- void snfXCImgr::linkHome(snf_RulebaseHandler* Home) { // Link to Home and set up shop.
- if(0 != Home && 0 == myHome) { // If we are getting our home
- myHome = Home; // then capture it,
- myHome->use(); // Update it's use count.
- TimeToStop = false; // clear the time to stop bit,
- run(); // run our thread.
- }
- }
-
- int snfXCImgr::TotalQueue() { // Return the total work queue size.
- cd::ScopeMutex IGotIt(ChannelMutex); // Serialize state control for safety.
- return (
- ((0 == C0) ? 0 : C0->Size()) +
- ((0 == C1) ? 0 : C1->Size()) +
- ((0 == C2) ? 0 : C2->Size()) +
- ((0 == C3) ? 0 : C3->Size())
- );
- }
-
- void snfXCImgr::stop() { // Called to shut down.
- if(false == TimeToStop) { // If we are not stopped then
- TimeToStop = true; // it is time to stop.
- join(); // Wait for our main thread first,
- shutdown_XCI(); // then shut down the XCI.
- myHome->unuse(); // Let go of the rulebase manager.
- myHome = 0; // Null it out for safety.
- }
- }
-
|