fly/terragear
1
0
Fork 0
Code Activity

Add multicore support for genapt850.

Browse source 
- fix launch

- fix taxiway debug uninitialized variable

- fixed issue when child crashes, not timed out - detect the error
when we get notified that the socket has data, but no line is received
This commit is contained in:
Peter Sadrozinski 2012-05-13 09:49:02 -04:00 • committed by Peter Sadrozinski
parent 18fd0e9546
commit 83f6532af0
8 changed files with 1172 additions and 356 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/Airports/GenAirports850/CMakeLists.txt
View file

@ -17,6 +17,7 @@ add_executable(genapts850
runway.cxx runway.hxx
rwy_simple.cxx
rwy_gen.cxx
scheduler.cxx scheduler.hxx
taxiway.cxx taxiway.hxx
)
@ -24,6 +25,8 @@ target_link_libraries(genapts850
Polygon Geometry
Array Optimize Output poly2tri
TriangleJRS
PocoFoundation
PocoNet
${GDAL_LIBRARY}
${SIMGEAR_CORE_LIBRARIES}
${SIMGEAR_CORE_LIBRARY_DEPENDENCIES}

12
src/Airports/GenAirports850/airport.cxx
View file

@ -91,6 +91,7 @@ Airport::Airport( int c, char* def)
dbg_pvmt_poly = 0;
dbg_feat_poly = 0;
dbg_base_poly = 0;
dbg_taxi_poly = 0;
SG_LOG( SG_GENERAL, SG_DEBUG, "Created airport with icao " << icao << ", control tower " << ct << ", and description " << description );
@ -123,6 +124,11 @@ Airport::~Airport()
delete pavements[i];
}
for (unsigned int i=0; i<taxiways.size(); i++)
{
delete taxiways[i];
}
for (unsigned int i=0; i<lightobjects.size(); i++)
{
delete lightobjects[i];
@ -525,8 +531,6 @@ void Airport::merge_slivers( superpoly_list& polys, poly_list& slivers_list ) {
}
}
void Airport::BuildBtg(const string& root, const string_list& elev_src )
{
ClipPolyType accum;
@ -600,6 +604,8 @@ void Airport::BuildBtg(const string& root, const string_list& elev_src )
}
}
SG_LOG(SG_GENERAL, SG_INFO, "Parse Complete - Runways: " << runways.size() << " Pavements: " << pavements.size() << " Features: " << features.size() << " Taxiways: " << taxiways.size() );
// Starting to clip the polys (for now - only UNIX builds)
build_start.stamp();
@ -1803,5 +1809,5 @@ void Airport::BuildBtg(const string& root, const string_list& elev_src )
tgChopNormalPolygon( holepath, "Hole", divided_base, true );
tgChopNormalPolygon( holepath, "Airport", apt_clearing, false );
SG_LOG( SG_GENERAL, SG_ALERT, SGLOG_GREEN << "\nSUCCESS generating " << icao << SGLOG_NORMAL << "\n" );
//SG_LOG( SG_GENERAL, SG_ALERT, SGLOG_GREEN << "\nSUCCESS generating " << icao << SGLOG_NORMAL << "\n" );
}

1
src/Airports/GenAirports850/airport.hxx
View file

@ -114,6 +114,7 @@ public:
void BuildBtg( const string& root, const string_list& elev_src );
void SetDebugPolys( int rwy, int taxi, int pvmt, int feat, int base );
void DumpStats( void );
private:
int code; // airport, heliport or sea port

97
src/Airports/GenAirports850/main.cxx
View file

@ -16,6 +16,12 @@
#include <string.h>
#include <iostream>
#include <Poco/Environment.h>
#include <Poco/Net/SocketAddress.h>
#include <Poco/Net/Socket.h>
#include <Poco/Net/StreamSocket.h>
#include <Poco/Net/SocketStream.h>
#include <simgear/debug/logstream.hxx>
#include <simgear/misc/sg_path.hxx>
#include <simgear/misc/sgstream.hxx>
@ -28,8 +34,10 @@
#include "closedpoly.hxx"
#include "linearfeature.hxx"
#include "parser.hxx"
#include "scheduler.hxx"
using namespace std;
using namespace Poco;
// TODO : Modularize this function
@ -113,6 +121,11 @@ int nudge = 10;
double slope_max = 0.2;
double gSnap = 0.00000001; // approx 1 mm
// For creating a buffered stream to write to the socket
Net::StreamSocket ss;
Net::SocketStreamBuf ssb( ss );
ostream os(&ssb);
int main(int argc, char **argv)
{
float min_lon = -180;
@ -121,6 +134,7 @@ int main(int argc, char **argv)
float max_lat = 90;
long position = 0;
// Setup elevation directories
string_list elev_src;
elev_src.clear();
setup_default_elevation_sources(elev_src);
@ -128,23 +142,24 @@ int main(int argc, char **argv)
// Set Normal logging
sglog().setLogLevels( SG_GENERAL, SG_INFO );
SG_LOG(SG_GENERAL, SG_INFO, "Run genapt");
// parse arguments
string work_dir = "";
string input_file = "";
string summary_file = "./genapt850.csv";
string start_id = "";
string restart_id = "";
string airport_id = "";
long airport_pos = -1;
string last_apt_file = "./last_apt.txt";
int dump_rwy_poly = -1;
int dump_taxi_poly = -1;
int dump_pvmt_poly = -1;
int dump_feat_poly = -1;
int dump_base_poly = -1;
int num_threads = Poco::Environment::processorCount();
int redirect_port = -1;
int arg_pos;
for (arg_pos = 1; arg_pos < argc; arg_pos++)
{
string arg = argv[arg_pos];
@ -163,7 +178,11 @@ int main(int argc, char **argv)
else if ( arg.find("--start-id=") == 0 )
{
start_id = arg.substr(11);
}
}
else if ( arg.find("--airport-pos=") == 0 )
{
airport_pos = atol( arg.substr(14).c_str() );
}
else if ( arg.find("--restart-id=") == 0 )
{
restart_id = arg.substr(13);
@ -228,6 +247,10 @@ int main(int argc, char **argv)
{
sglog().setLogLevels( SG_GENERAL, SG_BULK );
}
else if ( arg.find("--redirect-port=") == 0 )
{
redirect_port = atoi( arg.substr(16).c_str() );
}
else if ( (arg.find("--max-slope=") == 0) )
{
slope_max = atof( arg.substr(12).c_str() );
@ -264,6 +287,7 @@ int main(int argc, char **argv)
}
}
SG_LOG(SG_GENERAL, SG_INFO, "Run genapt with " << num_threads << " threads" );
SG_LOG(SG_GENERAL, SG_INFO, "Input file = " << input_file);
SG_LOG(SG_GENERAL, SG_INFO, "Terrain sources = ");
for ( unsigned int i = 0; i < elev_src.size(); ++i )
@ -304,6 +328,7 @@ int main(int argc, char **argv)
sgp.append( "dummy" );
sgp.create_dir( 0755 );
string command = argv[0];
string lastaptfile = work_dir+"/last_apt";
// initialize persistant polygon counter
@ -320,63 +345,69 @@ int main(int argc, char **argv)
exit(-1);
}
// Create the parser...
Parser* parser = new Parser(input_file, work_dir, elev_src);
// Create the scheduler
Scheduler* scheduler = new Scheduler(command, input_file, work_dir, elev_src);
// Add any debug
parser->SetDebugPolys( dump_rwy_poly, dump_taxi_poly, dump_pvmt_poly, dump_feat_poly, dump_base_poly );
// TODO : parser->SetDebugPolys( dump_rwy_poly, dump_taxi_poly, dump_pvmt_poly, dump_feat_poly, dump_base_poly );
// check for output redirect
if ( redirect_port >= 0 ) {
// create a stream socket back to the main process
Net::SocketAddress sa( "localhost", redirect_port );
ss.connect(sa);
// then a buffered stream to write to the socket
os.rdbuf(&ssb);
// then hook up SG_LOG to the stream buf
sglog().set_output( os );
} else {
// this is the main program -
SG_LOG(SG_GENERAL, SG_INFO, "Launch command was " << argv[0] );
}
// just one airport
if ( airport_id != "" )
{
// just find and add the one airport
parser->AddAirport( airport_id );
scheduler->AddAirport( airport_id );
SG_LOG(SG_GENERAL, SG_INFO, "Finished Adding airport - now parse");
// and start the parser
parser->Parse( last_apt_file );
// and schedule parsers
scheduler->Schedule( num_threads, summary_file );
}
// We are given an airport position from a main scheduler - parse this
else if ( airport_pos != -1 )
{
// create and start the real parser
Parser parser(input_file, work_dir, elev_src);
parser.Parse( airport_pos );
}
else if ( start_id != "" )
{
SG_LOG(SG_GENERAL, SG_INFO, "move forward to " << start_id );
// scroll forward in datafile
position = parser->FindAirport( start_id );
position = scheduler->FindAirport( start_id );
// add remaining airports within boundary
parser->AddAirports( position, min_lat, min_lon, max_lat, max_lon );
scheduler->AddAirports( position, min_lat, min_lon, max_lat, max_lon );
// parse all the airports that were found
parser->Parse( last_apt_file );
}
else if ( restart_id != "" )
{
SG_LOG(SG_GENERAL, SG_INFO, "move forward airport after " << restart_id );
// scroll forward in datafile
position = parser->FindAirport( restart_id );
// add all remaining airports within boundary
parser->AddAirports( position, min_lat, min_lon, max_lat, max_lon );
// but remove the restart id - it's broken
parser->RemoveAirport( restart_id );
// parse all the airports that were found
parser->Parse( last_apt_file );
scheduler->Schedule( num_threads, summary_file );
}
else
{
// find all airports within given boundary
parser->AddAirports( 0, min_lat, min_lon, max_lat, max_lon );
scheduler->AddAirports( 0, min_lat, min_lon, max_lat, max_lon );
// and parser them
parser->Parse( last_apt_file );
scheduler->Schedule( num_threads, summary_file );
}
delete parser;
SG_LOG(SG_GENERAL, SG_INFO, "Done");
exit(0);

346
src/Airports/GenAirports850/parser.cxx
View file

@ -1,13 +1,26 @@
#include <ctime>
#include <Poco/Mutex.h>
#include <Poco/Pipe.h>
#include <Poco/PipeStream.h>
#include <Poco/Process.h>
#include <Poco/Runnable.h>
#include <Poco/Semaphore.h>
#include <Poco/Thread.h>
#include <Poco/Timespan.h>
#include <Poco/Net/ServerSocket.h>
#include <Poco/Net/SocketAddress.h>
#include <Poco/Net/SocketStream.h>
#include <Poco/Net/Socket.h>
#include <Poco/Net/StreamSocket.h>
#include <simgear/debug/logstream.hxx>
#include <simgear/misc/sgstream.hxx>
#include <simgear/timing/timestamp.hxx>
#include "parser.hxx"
bool Parser::IsAirportDefinition( char* line, string icao )
bool Parser::GetAirportDefinition( char* line, string& icao )
{
char* tok;
int code;
@ -28,38 +41,11 @@ bool Parser::IsAirportDefinition( char* line, string icao )
case SEA_AIRPORT_CODE:
case HELIPORT_CODE:
airport = new Airport( code, line );
if ( airport->GetIcao() == icao )
{
match = true;
}
icao = airport->GetIcao();
match = true;
break;
case LAND_RUNWAY_CODE:
case WATER_RUNWAY_CODE:
case HELIPAD_CODE:
case PAVEMENT_CODE:
case LINEAR_FEATURE_CODE:
case BOUNDRY_CODE:
case NODE_CODE:
case BEZIER_NODE_CODE:
case CLOSE_NODE_CODE:
case CLOSE_BEZIER_NODE_CODE:
case TERM_NODE_CODE:
case TERM_BEZIER_NODE_CODE:
case AIRPORT_VIEWPOINT_CODE:
case AIRPLANE_STARTUP_LOCATION_CODE:
case LIGHT_BEACON_CODE:
case WINDSOCK_CODE:
case TAXIWAY_SIGN:
case LIGHTING_OBJECT:
case COMM_FREQ1_CODE:
case COMM_FREQ2_CODE:
case COMM_FREQ3_CODE:
case COMM_FREQ4_CODE:
case COMM_FREQ5_CODE:
case COMM_FREQ6_CODE:
case COMM_FREQ7_CODE:
case END_OF_FILE :
default:
break;
}
}
@ -67,258 +53,6 @@ bool Parser::IsAirportDefinition( char* line, string icao )
return match;
}
void Parser::AddAirport( string icao )
{
char line[2048];
long cur_pos;
bool found = false;
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
SG_LOG( SG_GENERAL, SG_INFO, "Adding airport " << icao << " to parse list");
while ( !in.eof() && !found )
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
// this is and airport definition - remember it
if ( IsAirportDefinition( line, icao ) )
{
SG_LOG( SG_GENERAL, SG_DEBUG, "Found airport " << icao << " at " << cur_pos );
parse_positions.push_back( cur_pos );
airport_icaos.push_back( icao );
found = true;
}
}
}
long Parser::FindAirport( string icao )
{
char line[2048];
long cur_pos = 0;
bool found = false;
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
SG_LOG( SG_GENERAL, SG_DEBUG, "Finding airport " << icao );
while ( !in.eof() && !found )
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
// this is and airport definition - remember it
if ( IsAirportDefinition( line, icao ) )
{
SG_LOG( SG_GENERAL, SG_DEBUG, "Found airport " << line << " at " << cur_pos );
found = true;
}
}
if (found)
{
return cur_pos;
}
else
{
return 0;
}
}
void Parser::AddAirports( long start_pos, float min_lat, float min_lon, float max_lat, float max_lon )
{
char line[2048];
char* def;
long cur_pos;
long cur_apt_pos;
string cur_apt_name;
char* tok;
int code;
bool match;
bool done;
done = false;
match = false;
// start from current position, and push all airports where a runway start or end
// lies within the given min/max coordinates
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
if (start_pos)
{
in.seekg(start_pos, ios::beg);
}
while (!done)
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
def = &line[0];
// Get the number code
tok = strtok(def, " \t\r\n");
if (tok)
{
def += strlen(tok)+1;
code = atoi(tok);
switch(code)
{
case LAND_AIRPORT_CODE:
case SEA_AIRPORT_CODE:
case HELIPORT_CODE:
{
Airport* airport = new Airport( code, def );
if (match)
{
parse_positions.push_back( cur_apt_pos );
airport_icaos.push_back( cur_apt_name );
}
// remember this new apt pos and name, and clear match
cur_apt_pos = cur_pos;
cur_apt_name = airport->GetIcao();
delete airport;
match = false;
}
break;
case END_OF_FILE:
if (match)
{
parse_positions.push_back( cur_apt_pos );
airport_icaos.push_back( cur_apt_name );
}
done = true;
break;
case LAND_RUNWAY_CODE:
// if the the runway start / end coords are within the rect,
// we have a winner
{
Runway* runway = new Runway(def);
Point3D start = runway->GetStart();
Point3D end = runway->GetEnd();
if ( (start.x() >= min_lon ) &&
(start.y() >= min_lat ) &&
(start.x() <= max_lon ) &&
(start.y() <= max_lat ) ) {
match = true;
}
else if ( (end.x() >= min_lon ) &&
(end.y() >= min_lat ) &&
(end.x() <= max_lon ) &&
(end.y() <= max_lat ) ) {
match = true;
}
delete runway;
}
break;
case WATER_RUNWAY_CODE:
// if the the runway start / end coords are within the rect,
// we have a winner
{
WaterRunway* runway = new WaterRunway(def);
Point3D start = runway->GetStart();
Point3D end = runway->GetEnd();
if ( (start.x() >= min_lon ) &&
(start.y() >= min_lat ) &&
(start.x() <= max_lon ) &&
(start.y() <= max_lat ) ) {
match = true;
}
else if ( (end.x() >= min_lon ) &&
(end.y() >= min_lat ) &&
(end.x() <= max_lon ) &&
(end.y() <= max_lat ) ) {
match = true;
}
delete runway;
}
break;
case HELIPAD_CODE:
// if the heliport coords are within the rect, we have
// a winner
{
Helipad* helipad = new Helipad(def);
Point3D loc = helipad->GetLoc();
if ( (loc.x() >= min_lon ) &&
(loc.y() >= min_lat ) &&
(loc.x() <= max_lon ) &&
(loc.y() <= max_lat ) ) {
match = true;
}
delete helipad;
}
break;
case TAXIWAY_CODE:
case PAVEMENT_CODE:
case LINEAR_FEATURE_CODE:
case BOUNDRY_CODE:
case NODE_CODE:
case BEZIER_NODE_CODE:
case CLOSE_NODE_CODE:
case CLOSE_BEZIER_NODE_CODE:
case TERM_NODE_CODE:
case TERM_BEZIER_NODE_CODE:
case AIRPORT_VIEWPOINT_CODE:
case AIRPLANE_STARTUP_LOCATION_CODE:
case LIGHT_BEACON_CODE:
case WINDSOCK_CODE:
case TAXIWAY_SIGN:
case LIGHTING_OBJECT:
case COMM_FREQ1_CODE:
case COMM_FREQ2_CODE:
case COMM_FREQ3_CODE:
case COMM_FREQ4_CODE:
case COMM_FREQ5_CODE:
case COMM_FREQ6_CODE:
case COMM_FREQ7_CODE:
break;
}
}
}
}
void Parser::RemoveAirport( string icao )
{
for (unsigned int i = 0; i < airport_icaos.size(); i++ ) {
if (airport_icaos[i] == icao) {
parse_positions.erase(parse_positions.begin()+i);
airport_icaos.erase(airport_icaos.begin()+i);
break;
}
}
}
void Parser::SetDebugPolys( int rwy, int taxi, int pvmt, int feat, int base )
{
rwy_poly = rwy;
@ -328,10 +62,11 @@ void Parser::SetDebugPolys( int rwy, int taxi, int pvmt, int feat, int base )
base_poly = base;
}
void Parser::Parse( string last_apt_file )
void Parser::Parse( long pos )
{
char tmp[2048];
char line[2048];
string icao;
SGTimeStamp parse_start;
SGTimeStamp parse_end;
SGTimeStamp parse_time;
@ -346,31 +81,31 @@ void Parser::Parse( string last_apt_file )
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
in.seekg(pos, ios::beg);
// for each position in parse_positions, parse an airport
for ( unsigned int i=0; i < parse_positions.size(); i++)
{
// get a line
in.getline(line, 2048);
// Verify this is and airport definition and get the icao
if( GetAirportDefinition( line, icao ) ) {
SG_LOG( SG_GENERAL, SG_INFO, "Found airport " << icao << " at " << pos );
// Start parse at pos
SetState(STATE_NONE);
in.clear();
parse_start.stamp();
log_time = time(0);
SG_LOG( SG_GENERAL, SG_ALERT, "\n*******************************************************************" );
SG_LOG( SG_GENERAL, SG_ALERT, "Start airport " << airport_icaos[i] << " at " << parse_positions[i] << ": start time " << ctime(&log_time) );
in.seekg(parse_positions[i], ios::beg);
// save the airport we are working on
char command[256];
sprintf( command, "echo %s > %s", airport_icaos[i].c_str(), last_apt_file.c_str() );
system( command );
SG_LOG( SG_GENERAL, SG_ALERT, "Start airport " << icao << " at " << pos << ": start time " << ctime(&log_time) );
in.seekg(pos, ios::beg);
while ( !in.eof() && (cur_state != STATE_DONE ) )
{
in.getline(tmp, 2048);
in.getline(line, 2048);
// Parse the line
ParseLine(tmp);
ParseLine(line);
}
parse_end.stamp();
@ -390,12 +125,11 @@ void Parser::Parse( string last_apt_file )
}
log_time = time(0);
SG_LOG( SG_GENERAL, SG_ALERT, "Finished airport " << airport_icaos[i] << ": end time " << ctime(&log_time) );
SG_LOG( SG_GENERAL, SG_INFO, "Time to parse " << parse_time );
SG_LOG( SG_GENERAL, SG_INFO, "Time to build " << build_time );
SG_LOG( SG_GENERAL, SG_INFO, "Time to clean up " << clean_time );
SG_LOG( SG_GENERAL, SG_INFO, "Time to triangulate " << triangulation_time );
SG_LOG( SG_GENERAL, SG_ALERT, "Finished airport " << icao <<
" : parse " << parse_time << " : build " << build_time <<
" : clean " << clean_time << " : tesselate " << triangulation_time );
} else {
SG_LOG( SG_GENERAL, SG_INFO, "Not an airport at pos " << pos << " line is: " << line );
}
}

17
src/Airports/GenAirports850/parser.hxx
View file

@ -58,10 +58,6 @@
using namespace std;
typedef std::vector <long> ParseList;
typedef std::vector <string> IcaoList;
class Parser
{
public:
@ -94,14 +90,11 @@ public:
}
void SetDebugPolys( int rwy, int taxi, int pvmt, int feat, int base );
long FindAirport( string icao );
void AddAirport( string icao );
void AddAirports( long start_pos, float min_lat, float min_lon, float max_lat, float max_lon );
void RemoveAirport( string icao );
void Parse( string last_apt_file );
void Parse( long pos );
private:
bool IsAirportDefinition( char* line, string icao );
bool GetAirportDefinition( char* line, string& icao );
int SetState( int state );
@ -133,10 +126,6 @@ private:
Beacon* cur_beacon;
Sign* cur_sign;
// List of positions in database file to parse
ParseList parse_positions;
IcaoList airport_icaos;
// debug
int rwy_poly;
int taxi_poly;

852
src/Airports/GenAirports850/scheduler.cxx Normal file
View file

@ -0,0 +1,852 @@
#include <cstring>
#include <simgear/debug/logstream.hxx>
#include <simgear/misc/sgstream.hxx>
#include "airport.hxx"
#include "parser.hxx"
#include "scheduler.hxx"
extern double gSnap;
/*** PROCESS INFO ***/
ProcessInfo::ProcessInfo( AirportInfo* pai, const ProcessHandle ph, Net::StreamSocket s ) : procHandle(ph)
{
pInfo = pai;
sock = s;
pssb = new Net::SocketStreamBuf( s );
pin = new istream( pssb );
state = P_STATE_INIT;
SetTimeout();
}
void ProcessInfo::SetTimeout( void )
{
SGTimeStamp now;
SGTimeStamp to;
now.stamp();
switch( state ) {
case P_STATE_INIT:
to.setTime(P_STATE_INIT_TIME, 0);
break;
case P_STATE_PARSE:
to.setTime(P_STATE_PARSE_TIME, 0);
break;
case P_STATE_BUILD:
to.setTime(P_STATE_BUILD_TIME, 0);
break;
case P_STATE_TRIANGULATE:
to.setTime(P_STATE_TRIANGULATE_TIME, 0);
break;
case P_STATE_OUTPUT:
to.setTime(P_STATE_OUTPUT_TIME, 0);
break;
}
timeout = (now + to);
}
void ProcessInfo::Kill( void )
{
SGTimeStamp now;
now.stamp();
// kill the process
Process::kill( procHandle.id() );
// wait for the zombie
procHandle.wait();
// mark process info - so we can reclaim it
state = P_STATE_KILLED;
}
int ProcessInfo::HandleLine( void )
{
char line[256];
pin->getline( line, 256 );
if ( pin->rdstate() != ifstream::goodbit ) {
SG_LOG( SG_GENERAL, SG_INFO, pInfo->GetIcao() << ": ProcessInfo::HandleLine read from socket error " << pin->rdstate() );
state = P_STATE_KILLED;
} else {
pInfo->SetErrorString( line );
// Print the line
SG_LOG( SG_GENERAL, SG_INFO, pInfo->GetIcao() << ": " << line);
// Update state
if ( strstr( line, "Parse Complete " ) != NULL ) {
// Grab the stats
int rwys, pvmnts, feats, twys;
sscanf(line, "Parse Complete - Runways: %d Pavements: %d Features: %d Taxiways: %d", &rwys, &pvmnts, &feats, &twys);
pInfo->SetRunways( rwys );
pInfo->SetPavements( pvmnts );
pInfo->SetFeats( feats );
pInfo->SetTaxiways( twys );
} else if ( strstr( line, "Finished airport " ) != NULL ) {
// Grab the stats
int parse_sec, parse_nsec;
int build_sec, build_nsec;
int clean_sec, clean_nsec;
int tess_sec, tess_nsec;
SGTimeStamp parse_ts;
SGTimeStamp build_ts;
SGTimeStamp clean_ts;
SGTimeStamp tess_ts;
state = P_STATE_DONE;
sscanf(line, "Finished airport %*s : parse %d.%d : build %d.%d : clean %d.%d : tesselate %d.%d",
&parse_sec, &parse_nsec, &build_sec, &build_nsec, &clean_sec, &clean_nsec, &tess_sec, &tess_nsec );
parse_ts.setTime( parse_sec, parse_nsec );
build_ts.setTime( build_sec, build_nsec );
clean_ts.setTime( clean_sec, clean_nsec );
tess_ts.setTime( tess_sec, tess_nsec );
pInfo->SetParseTime( parse_ts );
pInfo->SetBuildTime( build_ts );
pInfo->SetCleanTime( clean_ts );
pInfo->SetTessTime( tess_ts );
procHandle.wait();
} else if ( strstr( line, "Build Feature Poly " ) != NULL ) {
state = P_STATE_BUILD;
} else if ( strstr( line, "Build Pavement " ) != NULL ) {
state = P_STATE_BUILD;
} else if ( strstr( line, "Build Runway " ) != NULL ) {
state = P_STATE_BUILD;
} else if ( strstr( line, "Tesselating " ) != NULL ) {
state = P_STATE_TRIANGULATE;
} else if ( strstr( line, "Adding runway nodes and normals " ) != NULL ) {
state = P_STATE_OUTPUT;
}
}
SetTimeout();
return state;
}
ostream& operator<< (ostream &out, const AirportInfo &ai)
{
char snap_string[32];
sprintf( snap_string, "%1.8lf", ai.snap );
out << ai.icao;
out << ",";
out << ai.numRunways;
out << ",";
out << ai.numPavements;
out << ",";
out << ai.numFeats;
out << ",";
out << ai.numTaxiways;
out << ",";
out << ai.parseTime;
out << ",";
out << ai.buildTime;
out << ",";
out << ai.cleanTime;
out << ",";
out << ai.tessTime;
out << ",";
out << ai.parseTime+ai.buildTime+ai.cleanTime+ai.tessTime;
out << ",";
out << snap_string,
out << ",";
out << ai.errString;
return out; // MSVC
}
/*** PROCESS LIST CLASS ***/
ProcessList::ProcessList( int n, string& summaryfile, Scheduler* pScheduler ) : available(n), ready(1), state( PL_STATE_WAIT_FOR_LAUNCH )
{
// The process List is responsible for creating new processes (Launch)
// and monitoring the status of the launched parsers (Monitor) These
// functions are called from different threads.
pss = pScheduler->GetServerSocket();
// remember the output file
csvfile.open( summaryfile.c_str(), ios_base::out | ios_base::app );
// remember the scheduler so we can add retries
scheduler = pScheduler;
// remember the number of available helper procs so we know when we're full
threads = n;
}
// When a slot is available, the main thread calls launch to instantiate a
// new pareser process
void ProcessList::Launch( string command, string file, AirportInfo* pai, bool last )
{
Process::Args args;
char arg[64];
Pipe outPipe;
// generate correct command line arguments
args.push_back("--work=work");
sprintf( arg, "--input=%s", file.c_str() );
args.push_back(arg);
sprintf( arg, "--airport-pos=%ld", pai->GetPos() );
args.push_back(arg);
sprintf( arg, "--snap=%1.8lf", pai->GetSnap() );
args.push_back(arg);
sprintf( arg, "--redirect-port=%d", GENAPT_PORT );
args.push_back(arg);
// Launch the child process
ProcessHandle ph = Process::launch(command, args, 0, &outPipe, &outPipe);
// Wait 10 seconds for connection
Timespan timeout( 10, 0 );
int retVal = pss->poll( timeout, Net::Socket::SELECT_READ );
// If we connected - create a new entry
if ( retVal == true ) {
Net::SocketAddress sockaddr;
Net::StreamSocket sock = pss->acceptConnection( sockaddr );
// Make sure the list can't be modified while adding a member
lock.lock();
ProcessInfo pi( pai, ph, sock );
plist.push_back( pi );
lock.unlock();
// If we have all of the airports in our list, we are done
// when the list is empty - set the transition state
if ( last ) {
// The launch list is empty - we're ready to monitor
state = PL_STATE_ALL_LAUNCHED;
ready.set();
} else if ( plist.size() == threads ) {
// The resource list is full - we're ready to monitor
state = PL_STATE_LIST_FULL;
ready.set();
} else {
// resource list has space, and launch list is not empty - hold off monitoring
state = PL_STATE_WAIT_FOR_LAUNCH;
}
}
}
Timespan ProcessList::GetNextTimeout()
{
SGTimeStamp now, min, timeout;
min.setTime( UINT_MAX, 0 );
timeout.setTime( 0, 0 );
for ( unsigned int i=0; i< plist.size(); i++ ) {
if ( plist[i].GetTimeout() < min ) {
min = plist[i].GetTimeout();
}
}
now.stamp();
if ( min > now ) {
timeout = min - now;
}
return Timespan( timeout.get_seconds(), timeout.get_usec() );
}
void ProcessList::HandleReceivedMessages( Net::Socket::SocketList& slr )
{
// for each socket that has data - find the corresponding icao
for (unsigned int i=0; i<slr.size(); i++) {
Net::StreamSocket ss = (Net::StreamSocket)slr[i];
// find the index handling this socket, and let it deal with the line
for ( unsigned int j=0; j < plist.size(); j++ ) {
if ( plist[j].GetSocket() == ss ) {
plist[j].HandleLine( );
break;
}
}
}
}
void ProcessList::HandleFinished( void )
{
AirportInfo* pInfo = NULL;
int num_deleted = 0;
bool done = false;
while (!done) {
done = true;
lock.lock();
for ( unsigned int i=0; i< plist.size(); i++ ) {
switch ( plist[i].GetState() ) {
case P_STATE_DONE:
plist[i].SetErrorString( "success" );
// holding the list lock - only one thread can write to the csvfile at a time
csvfile << plist[i].GetInfo() << "\n";
csvfile.flush();
// remove this airport from the list - it's complete
plist[i].CloseSock();
plist.erase( plist.begin()+i );
// keep track of the number of deleted entries
num_deleted++;
// let's iterate again to look for more timeouts...
done = false;
break;
case P_STATE_KILLED:
// holding the list lock - only one thread can write to the csvfile at a time
csvfile << plist[i].GetInfo() << "\n";
csvfile.flush();
// Schedule a retry
pInfo = plist[i].GetInfoPtr();
pInfo->IncreaseSnap();
scheduler->RetryAirport( pInfo );
// remove the airport from the monitor list - it's complete
plist[i].CloseSock();
plist.erase( plist.begin()+i );
// keep track of the number of deleted entries
num_deleted++;
// let's iterate again to look for more timeouts...
done = false;
break;
default:
break;
}
}
lock.unlock();
}
// Let launcher thread know we have opening(s)
while ( num_deleted-- ) {
// make sure we don't start waiting on output before a new apt is launched...
if (state != PL_STATE_ALL_LAUNCHED) {
state = PL_STATE_WAIT_FOR_LAUNCH;
}
// free each resource that is no longer used
available.set();
}
}
void ProcessList::WaitForSlot( void )
{
available.wait();
}
// list lock is held
void ProcessList::HandleTimeouts()
{
SGTimeStamp now;
now.stamp();
for ( unsigned int i=0; i< plist.size(); i++ ) {
if ( plist[i].GetTimeout() < now ) {
plist[i].Kill();
}
}
}
void ProcessList::Monitor()
{
// Wait until process list has a connection, then continue until we are done
while( state != PL_STATE_DONE ) {
Net::Socket::SocketList slr, slw, sle;
Timespan timeout;
int retVal;
// if we aren't ready to start - wait on ready
if ( state == PL_STATE_WAIT_FOR_LAUNCH ) {
ready.wait();
}
// then lock the list when calculating the timeouts
lock.lock();
// calculate the shortest timeout
timeout = GetNextTimeout();
// Add currently connected sockets
for ( unsigned int i=0; i< plist.size(); i++ ) {
slr.push_back( plist[i].GetSocket() );
}
// unlock before waiting on i/o
lock.unlock();
// this needs to be interrupted when new airports are added to the list
retVal = Net::Socket::select( slr, slw, sle, timeout );
if ( retVal > 0 ) {
HandleReceivedMessages( slr );
} else {
HandleTimeouts();
}
// remove finished or dead processes, and notify launcher
//
HandleFinished();
slr.clear();
slw.clear();
sle.clear();
// if we have launched all airports, we are done
if ( ( state == PL_STATE_ALL_LAUNCHED ) && ( plist.size() == 0 ) ) {
state = PL_STATE_DONE;
}
}
csvfile.close();
}
/*** PROCESS MONITOR ***/
ProcessMonitor::ProcessMonitor(ProcessList* pl) : Runnable()
{
plist = pl;
}
void ProcessMonitor::run()
{
SG_LOG( SG_GENERAL, SG_INFO, "ProcessMonitor Started " );
// Run the monitoring function in this thread
plist->Monitor();
SG_LOG( SG_GENERAL, SG_INFO, "ProcessMonitor Exited " );
}
/*** SCEDULER ***/
bool Scheduler::IsAirportDefinition( char* line, string icao )
{
char* tok;
int code;
Airport* airport = NULL;
bool match = false;
// Get the number code
tok = strtok(line, " \t\r\n");
if (tok)
{
line += strlen(tok)+1;
code = atoi(tok);
switch(code)
{
case LAND_AIRPORT_CODE:
case SEA_AIRPORT_CODE:
case HELIPORT_CODE:
airport = new Airport( code, line );
if ( airport->GetIcao() == icao )
{
match = true;
}
break;
case LAND_RUNWAY_CODE:
case WATER_RUNWAY_CODE:
case HELIPAD_CODE:
case PAVEMENT_CODE:
case LINEAR_FEATURE_CODE:
case BOUNDRY_CODE:
case NODE_CODE:
case BEZIER_NODE_CODE:
case CLOSE_NODE_CODE:
case CLOSE_BEZIER_NODE_CODE:
case TERM_NODE_CODE:
case TERM_BEZIER_NODE_CODE:
case AIRPORT_VIEWPOINT_CODE:
case AIRPLANE_STARTUP_LOCATION_CODE:
case LIGHT_BEACON_CODE:
case WINDSOCK_CODE:
case TAXIWAY_SIGN:
case LIGHTING_OBJECT:
case COMM_FREQ1_CODE:
case COMM_FREQ2_CODE:
case COMM_FREQ3_CODE:
case COMM_FREQ4_CODE:
case COMM_FREQ5_CODE:
case COMM_FREQ6_CODE:
case COMM_FREQ7_CODE:
case END_OF_FILE :
break;
}
}
return match;
}
void Scheduler::AddAirport( string icao )
{
char line[2048];
long cur_pos;
bool found = false;
AirportInfo* pInfo;
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
SG_LOG( SG_GENERAL, SG_INFO, "Adding airport " << icao << " to parse list");
while ( !in.eof() && !found )
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
// this is and airport definition - remember it
if ( IsAirportDefinition( line, icao ) )
{
SG_LOG( SG_GENERAL, SG_DEBUG, "Found airport " << icao << " at " << cur_pos );
pInfo = new AirportInfo( icao, cur_pos, gSnap );
originalList.push_back( *pInfo );
delete pInfo;
found = true;
}
}
}
long Scheduler::FindAirport( string icao )
{
char line[2048];
long cur_pos = 0;
bool found = false;
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
SG_LOG( SG_GENERAL, SG_DEBUG, "Finding airport " << icao );
while ( !in.eof() && !found )
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
// this is and airport definition - remember it
if ( IsAirportDefinition( line, icao ) )
{
SG_LOG( SG_GENERAL, SG_DEBUG, "Found airport " << line << " at " << cur_pos );
found = true;
}
}
if (found)
{
return cur_pos;
}
else
{
return 0;
}
}
void Scheduler::RetryAirport( AirportInfo* pai )
{
retryList.push_back( *pai );
}
void Scheduler::AddAirports( long start_pos, float min_lat, float min_lon, float max_lat, float max_lon )
{
char line[2048];
char* def;
long cur_pos;
long cur_apt_pos = 0;
string cur_apt_name;
char* tok;
int code;
bool match;
bool done;
done = false;
match = false;
// start from current position, and push all airports where a runway start or end
// lies within the given min/max coordinates
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
if (start_pos)
{
in.seekg(start_pos, ios::beg);
}
while (!done)
{
// remember the position of this line
cur_pos = in.tellg();
// get a line
in.getline(line, 2048);
def = &line[0];
// Get the number code
tok = strtok(def, " \t\r\n");
if (tok)
{
def += strlen(tok)+1;
code = atoi(tok);
switch(code)
{
case LAND_AIRPORT_CODE:
case SEA_AIRPORT_CODE:
case HELIPORT_CODE:
{
Airport* airport = new Airport( code, def );
if (match)
{
// Start off with given snap value
AirportInfo* pInfo = new AirportInfo( cur_apt_name, cur_apt_pos, gSnap );
originalList.push_back( *pInfo );
delete pInfo;
}
// remember this new apt pos and name, and clear match
cur_apt_pos = cur_pos;
cur_apt_name = airport->GetIcao();
delete airport;
match = false;
}
break;
case END_OF_FILE:
if (match)
{
// Start off with given snap value
AirportInfo* pInfo = new AirportInfo( cur_apt_name, cur_apt_pos, gSnap );
originalList.push_back( *pInfo );
delete pInfo;
}
done = true;
break;
case LAND_RUNWAY_CODE:
// if the the runway start / end coords are within the rect,
// we have a winner
{
Runway* runway = new Runway(def);
Point3D start = runway->GetStart();
Point3D end = runway->GetEnd();
if ( (start.x() >= min_lon ) &&
(start.y() >= min_lat ) &&
(start.x() <= max_lon ) &&
(start.y() <= max_lat ) ) {
match = true;
}
else if ( (end.x() >= min_lon ) &&
(end.y() >= min_lat ) &&
(end.x() <= max_lon ) &&
(end.y() <= max_lat ) ) {
match = true;
}
delete runway;
}
break;
case WATER_RUNWAY_CODE:
// if the the runway start / end coords are within the rect,
// we have a winner
{
WaterRunway* runway = new WaterRunway(def);
Point3D start = runway->GetStart();
Point3D end = runway->GetEnd();
if ( (start.x() >= min_lon ) &&
(start.y() >= min_lat ) &&
(start.x() <= max_lon ) &&
(start.y() <= max_lat ) ) {
match = true;
}
else if ( (end.x() >= min_lon ) &&
(end.y() >= min_lat ) &&
(end.x() <= max_lon ) &&
(end.y() <= max_lat ) ) {
match = true;
}
delete runway;
}
break;
case HELIPAD_CODE:
// if the heliport coords are within the rect, we have
// a winner
{
Helipad* helipad = new Helipad(def);
Point3D loc = helipad->GetLoc();
if ( (loc.x() >= min_lon ) &&
(loc.y() >= min_lat ) &&
(loc.x() <= max_lon ) &&
(loc.y() <= max_lat ) ) {
match = true;
}
delete helipad;
}
break;
case TAXIWAY_CODE:
case PAVEMENT_CODE:
case LINEAR_FEATURE_CODE:
case BOUNDRY_CODE:
case NODE_CODE:
case BEZIER_NODE_CODE:
case CLOSE_NODE_CODE:
case CLOSE_BEZIER_NODE_CODE:
case TERM_NODE_CODE:
case TERM_BEZIER_NODE_CODE:
case AIRPORT_VIEWPOINT_CODE:
case AIRPLANE_STARTUP_LOCATION_CODE:
case LIGHT_BEACON_CODE:
case WINDSOCK_CODE:
case TAXIWAY_SIGN:
case LIGHTING_OBJECT:
case COMM_FREQ1_CODE:
case COMM_FREQ2_CODE:
case COMM_FREQ3_CODE:
case COMM_FREQ4_CODE:
case COMM_FREQ5_CODE:
case COMM_FREQ6_CODE:
case COMM_FREQ7_CODE:
break;
}
}
}
}
Scheduler::Scheduler(string& cmd, string& datafile, const string& root, const string_list& elev_src)
{
command = cmd;
filename = datafile;
work_dir = root;
elevation = elev_src;
ifstream in( filename.c_str() );
if ( !in.is_open() )
{
SG_LOG( SG_GENERAL, SG_ALERT, "Cannot open file: " << filename );
exit(-1);
}
}
void Scheduler::Schedule( int num_threads, string& summaryfile )
{
ProcessList *procList = NULL;
Thread *monThread = NULL;
ProcessMonitor *procMon = NULL;
bool done = false;
bool last = false;
ofstream csvfile;
// open and truncate the summary file : monitor only appends
csvfile.open( summaryfile.c_str(), ios_base::out | ios_base::trunc );
csvfile.close();
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler: Bind to socket" );
// Bind the parent listener socket for children to connect to
ss.bind(GENAPT_PORT);
ss.listen();
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler: Bound" );
while (!done) {
procList = new ProcessList(num_threads, summaryfile, this);
monThread = new Thread;
procMon = new ProcessMonitor(procList);
// Launch monitor thread
monThread->start(*procMon);
// now try to launch child processes to parse individual airports
for ( unsigned int i=0; i<originalList.size(); i++ ) {
// Wait for an available process slot
procList->WaitForSlot();
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler: originalList has " << originalList.size() << ", i is " << i );
// let the process list know if more airports are coming
if ( i == originalList.size()-1 ) {
last = true;
}
// Launch a new parser
procList->Launch( command, filename, &originalList[i], last );
}
// Sync up before relaunching
monThread->join();
// Delete the old monitor
delete procMon;
delete monThread;
delete procList;
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler: originalList has " << originalList.size() << ", retry list has " << retryList.size() << " entries" );
// delete original, and copy retry to it
if ( retryList.size() ) {
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler: clear original list " );
originalList.clear();
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler - cleared original: originalList has " << originalList.size() << ", retry list has " << retryList.size() << " entries" );
for ( unsigned int i=0; i<retryList.size(); i++ ) {
originalList.push_back( retryList[i] );
}
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler - copied retryList: originalList has " << originalList.size() << ", retry list has " << retryList.size() << " entries" );
retryList.clear();
SG_LOG( SG_GENERAL, SG_INFO, "Scheduler - cleared retry: originalList has " << originalList.size() << ", retry list has " << retryList.size() << " entries" );
} else {
done = true;
}
}
}

200
src/Airports/GenAirports850/scheduler.hxx Normal file
View file

@ -0,0 +1,200 @@
#include <string>
#include <iostream>
#include <fstream>
#include <Poco/Mutex.h>
#include <Poco/Pipe.h>
#include <Poco/PipeStream.h>
#include <Poco/Process.h>
#include <Poco/Runnable.h>
#include <Poco/Semaphore.h>
#include <Poco/Thread.h>
#include <Poco/Timespan.h>
#include <Poco/Net/ServerSocket.h>
#include <Poco/Net/SocketAddress.h>
#include <Poco/Net/SocketStream.h>
#include <Poco/Net/Socket.h>
#include <Poco/Net/StreamSocket.h>
#include <simgear/compiler.h>
#include <simgear/math/sg_types.hxx>
#include <simgear/timing/timestamp.hxx>
#define P_STATE_INIT (0)
#define P_STATE_PARSE (1)
#define P_STATE_BUILD (2)
#define P_STATE_TRIANGULATE (3)
#define P_STATE_OUTPUT (4)
#define P_STATE_DONE (8)
#define P_STATE_KILLED (9)
#define P_STATE_INIT_TIME ( 1*60)
#define P_STATE_PARSE_TIME ( 1*60)
#define P_STATE_BUILD_TIME (30*60)
#define P_STATE_TRIANGULATE_TIME ( 1*60)
#define P_STATE_OUTPUT_TIME (10*60)
#define GENAPT_PORT (12397)
#define PL_STATE_INIT (0)
#define PL_STATE_WAIT_FOR_LAUNCH (1)
#define PL_STATE_LIST_FULL (2)
#define PL_STATE_ALL_LAUNCHED (3)
#define PL_STATE_DONE (10)
using namespace std;
using namespace Poco;
// Forward declaration
class Scheduler;
class AirportInfo
{
public:
AirportInfo( string id, long p, double s )
{
icao = id;
pos = p;
snap = s;
numRunways = -1;
numPavements = -1;
numFeats = -1;
numTaxiways = -1;
}
string GetIcao( void ) { return icao; }
long GetPos( void ) { return pos; }
double GetSnap( void ) { return snap; }
void SetRunways( int r ) { numRunways = r; }
void SetPavements( int p ) { numPavements = p; }
void SetFeats( int f ) { numFeats = f; }
void SetTaxiways( int t ) { numTaxiways = t; }
void SetParseTime( SGTimeStamp t ) { parseTime = t; }
void SetBuildTime( SGTimeStamp t ) { buildTime = t; }
void SetCleanTime( SGTimeStamp t ) { cleanTime = t; }
void SetTessTime( SGTimeStamp t ) { tessTime = t; }
void SetErrorString( char* e ) { errString = e; }
void IncreaseSnap( void ) { snap *= 2.0f; }
friend ostream& operator<<(ostream& output, const AirportInfo& ai);
private:
string icao;
long pos;
int numRunways;
int numPavements;
int numFeats;
int numTaxiways;
SGTimeStamp parseTime;
SGTimeStamp buildTime;
SGTimeStamp cleanTime;
SGTimeStamp tessTime;
double snap;
string errString;
};
typedef std::vector <AirportInfo> parseList;
class ProcessInfo
{
public:
ProcessInfo( AirportInfo* pai, const ProcessHandle ph, Net::StreamSocket s );
void SetTimeout( void );
SGTimeStamp GetTimeout( void ) { return timeout; }
string GetIcao( void ) { return pInfo->GetIcao(); }
Net::StreamSocket GetSocket( void ) { return sock; }
int GetState( void ) { return state; }
AirportInfo GetInfo( void ) { return *pInfo; }
AirportInfo* GetInfoPtr( void ) { return pInfo; }
void SetErrorString( char *e ) { pInfo->SetErrorString( e ); }
int HandleLine( void );
void Kill( void );
void CloseSock( void ) { sock.close(); }
private:
AirportInfo* pInfo;
ProcessHandle procHandle;
Net::StreamSocket sock;
Net::SocketStreamBuf *pssb;
istream *pin;
int state;
SGTimeStamp timeout;
};
typedef std::vector <ProcessInfo> ProcessInfoList;
class ProcessList
{
public:
ProcessList( int n, string& summaryfile, Scheduler* pScheduler );
// The main thread needs to wait until a slot is ready for creating a new
// Parser child process
inline void WaitForSlot(void);
// When a slot is available, the main thread calls launch to instantiate a
// new pareser process
void Launch( string command, string file, AirportInfo* pai, bool last );
Timespan GetNextTimeout();
void HandleReceivedMessages( Net::Socket::SocketList& slr );
void HandleTimeouts();
void HandleFinished( void );
void Monitor();
private:
Semaphore available;
Semaphore ready;
Mutex lock;
ProcessInfoList plist;
Net::ServerSocket* pss;
int state;
int threads;
ofstream csvfile;
Scheduler* scheduler;
};
class ProcessMonitor : public Runnable
{
public:
ProcessMonitor(ProcessList* pl);
virtual void run();
private:
ProcessList* plist;
};
class Scheduler
{
public:
Scheduler(string& cmd, string& datafile, const string& root, const string_list& elev_src);
long FindAirport( string icao );
void AddAirport( string icao );
void AddAirports( long start_pos, float min_lat, float min_lon, float max_lat, float max_lon );
void RetryAirport( AirportInfo* pInfo );
void Schedule( int num_threads, string& summaryfile );
Net::ServerSocket* GetServerSocket( void ) { return &ss; }
private:
bool IsAirportDefinition( char* line, string icao );
string command;
string filename;
string_list elevation;
string work_dir;
Net::ServerSocket ss;
// List of positions in database file to parse
parseList originalList;
parseList retryList;
};