|
PLearn 0.1
|
#include <PLearnService.h>


Public Types | |
| typedef void(* | log_callback_t )(PP< RemotePLearnServer > server, const string &module_name, int vlevel, const string &msg) |
| typedef void(* | progress_callback_t )(PP< RemotePLearnServer > server, unsigned int pbar, char action, unsigned int pos, const string &title) |
Public Member Functions | |
| void | connectToServers (TVec< pair< string, int > > server_names_and_port) |
| Will establish a TCP connection to all hostname,port in given list. | |
| void | connectToServers (PPath serversfile) |
| Will get the list of hostname, pid, port from the serversfile and establish a TCP connection to those. | |
| void | disconnectFromServers () |
| void | disconnectFromServer (PP< RemotePLearnServer > server) |
| int | availableServers () const |
| returns the number of available processing ressources | |
| PP< RemotePLearnServer > | reserveServer () |
| Reserves a remote processing ressource from the pool of servers. | |
| TVec< PP< RemotePLearnServer > > | reserveServers (int nservers) |
| Reserves up to nservers servers. | |
| void | freeServer (PP< RemotePLearnServer > server) |
| Frees a previously reserved servers. | |
| void | freeServers (TVec< PP< RemotePLearnServer > > servers) |
| Frees all the servers in the list. | |
| int | watchServers (TVec< PP< RemotePLearnServer > > servers, int timeout=0) |
| int | watchServers (TVec< PP< RemotePLearnServer > > servers, log_callback_t the_log_callback, progress_callback_t the_progress_callback) |
| PP< RemotePLearnServer > | waitForResult (TVec< PP< RemotePLearnServer > > servers=TVec< PP< RemotePLearnServer > >(), log_callback_t the_log_callback=log_callback, progress_callback_t the_progress_callback=progress_callback) |
| wait for one result | |
| void | waitForResultFrom (PP< RemotePLearnServer > from, log_callback_t the_log_callback=log_callback, progress_callback_t the_progress_callback=progress_callback) |
| ~PLearnService () | |
Static Public Member Functions | |
| static PLearnService & | instance () |
| static void | log_callback (PP< RemotePLearnServer > server, const string &module_name, int vlevel, const string &msg) |
| static void | progress_callback (PP< RemotePLearnServer > server, unsigned int pbar, char action, unsigned int pos=0, const string &title="") |
| static pair< string, int > | getId (RemotePLearnServer *server) |
Private Types | |
| typedef map < RemotePLearnServer *, map < unsigned int, PP < ProgressBar > > > | progress_bars_t |
Private Member Functions | |
| PLearnService () | |
Private Attributes | |
| TVec< PP< RemotePLearnServer > > | available_servers |
| std::set< PP < RemotePLearnServer > > | reserved_servers |
Static Private Attributes | |
| static progress_bars_t | progress_bars |
| static map< RemotePLearnServer *, pair< string, int > > | servers_ids |
Friends | |
| class | RemotePLearnServer |
Definition at line 61 of file PLearnService.h.
| typedef void(* PLearn::PLearnService::log_callback_t)(PP< RemotePLearnServer > server, const string &module_name, int vlevel, const string &msg) |
Definition at line 116 of file PLearnService.h.
typedef map<RemotePLearnServer*, map<unsigned int, PP<ProgressBar> > > PLearn::PLearnService::progress_bars_t [private] |
Definition at line 73 of file PLearnService.h.
| typedef void(* PLearn::PLearnService::progress_callback_t)(PP< RemotePLearnServer > server, unsigned int pbar, char action, unsigned int pos, const string &title) |
Definition at line 117 of file PLearnService.h.
| PLearn::PLearnService::PLearnService | ( | ) | [private] |
Definition at line 76 of file PLearnService.cc.
{}
| PLearn::PLearnService::~PLearnService | ( | ) |
Definition at line 377 of file PLearnService.cc.
{
if(reserved_servers.size() != 0)
PLWARNING("PLearnService::~PLearnService : some servers are still reserved; free them first.");
TVec<PP<RemotePLearnServer> > servers(available_servers.length());
servers << available_servers;
disconnectFromServers();
//now, get what's remaining on the servers streams
for(int i= 0; i < servers.length(); ++i)
{
try
{
for(;;) watchServers(servers, log_callback, progress_callback);
}
catch(const PLearnError&)
{
// do nothing...
}
}
}
| int PLearn::PLearnService::availableServers | ( | ) | const |
returns the number of available processing ressources
Definition at line 162 of file PLearnService.cc.
Referenced by PLearn::TestClientCommand::run(), PLearn::PLearner::test(), and PLearn::BaggingLearner::train().
{
return available_servers.size();
}

Will establish a TCP connection to all hostname,port in given list.
Definition at line 104 of file PLearnService.cc.
References DBG_LOG, PLearn::endl(), PLearn::PL_Log::instance(), PLearn::openSocket(), PLearn::PStream::plearn_ascii, PLearn::PStream::plearn_binary, and PLearn::TVec< T >::push_back().
Referenced by PLearn::global_options(), and PLearn::globalConnectToServers().
{
DBG_LOG << "PLearnService::connectToServers(" << hostname_and_port << ')' << endl;
if(available_servers.size()>0)
disconnectFromServers();
for(int k=0; k<hostname_and_port.length(); k++)
{
pair<string, int> host_port = hostname_and_port[k];
PStream servio = openSocket(host_port.first, host_port.second, PStream::plearn_ascii);
PP<RemotePLearnServer> serv = new RemotePLearnServer(servio);
serv->callFunction("binary");
TVec<PP<RemotePLearnServer> > ss;
ss.push_back(serv);
servers_ids[serv]= host_port;
watchServers(ss, log_callback, progress_callback);
servio << PStream::plearn_binary;
reserved_servers.insert(serv);
serv->getResults();
serv->callFunction("loggingControl",
PL_Log::instance().verbosity(),
PL_Log::instance().namedLogging());
serv->getResults();
reserved_servers.erase(serv);
available_servers.push(serv);
}
}


| void PLearn::PLearnService::connectToServers | ( | PPath | serversfile | ) |
Will get the list of hostname, pid, port from the serversfile and establish a TCP connection to those.
Definition at line 80 of file PLearnService.cc.
References PLearn::TVec< T >::append(), DBG_LOG, PLearn::endl(), PLearn::PStream::good(), PLearn::hostname(), in, PLearn::openFile(), PLearn::PStream::raw_ascii, and PLearn::PStream::skipBlanksAndComments().
{
PStream in = openFile(serversfile, PStream::raw_ascii, "r");
DBG_LOG << "PLearnService::connectToServers(" << serversfile << ')' << endl;
string hostname;
string pid;
int tcpport = -1;
TVec< pair<string,int> > hostname_and_port;
while(in.good())
{
in.skipBlanksAndComments();
if(!in)
break;
in >> hostname >> tcpport >> pid;
if(in.good())
hostname_and_port.append(pair<string,int>(hostname,tcpport));
}
connectToServers(hostname_and_port);
}

| void PLearn::PLearnService::disconnectFromServer | ( | PP< RemotePLearnServer > | server | ) |
Definition at line 146 of file PLearnService.cc.
References PLearn::endl(), i, and PLERROR.
{
for(int i= 0; i < available_servers.length(); ++i)
if(available_servers[i] == server)
{
available_servers.remove(i);
server->io.write("!Q ");
server->io << endl;
if(progress_bars.find(server) != progress_bars.end())
progress_bars.erase(server);
return;
}
PLERROR("PLearnService::disconnectFromServer : trying to disconnect from a server which is not available"
" (not connected to or reserved)");
}

| void PLearn::PLearnService::disconnectFromServers | ( | ) |
Definition at line 139 of file PLearnService.cc.
{
while(available_servers.length() > 0)
disconnectFromServer(available_servers[0]);
}
| void PLearn::PLearnService::freeServer | ( | PP< RemotePLearnServer > | server | ) |
Frees a previously reserved servers.
Putting it back into the list of available_servers
Definition at line 203 of file PLearnService.cc.
References DBG_LOG, PLearn::endl(), and PLERROR.
Referenced by PLearn::PLearner::test(), and PLearn::BaggingLearner::train().
{
DBG_LOG << "PLearnService::freeServer(...)" << endl;
server->deleteAllObjects();
server->clearMaps();
if(reserved_servers.erase(server)!=1)
PLERROR("Problem in PLearnService::freeServer are you sure this server had been properly reserved?");
available_servers.push(server);
}


| void PLearn::PLearnService::freeServers | ( | TVec< PP< RemotePLearnServer > > | servers | ) |
Frees all the servers in the list.
Definition at line 213 of file PLearnService.cc.
{
for(int k=0; k<servers.length(); k++)
freeServer(servers[k]);
}
| static pair<string, int> PLearn::PLearnService::getId | ( | RemotePLearnServer * | server | ) | [inline, static] |
Definition at line 136 of file PLearnService.h.
References servers_ids.
Referenced by PLearn::RemotePLearnServer::expectResults().
{ return servers_ids[server]; }

| PLearnService & PLearn::PLearnService::instance | ( | ) | [static] |
Definition at line 68 of file PLearnService.cc.
References PLearn::PP< T >::isNull().
Referenced by PLearn::RemotePLearnServer::expectResults(), PLearn::global_options(), PLearn::globalConnectToServers(), PLearn::TestClientCommand::run(), PLearn::PLearner::test(), PLearn::BaggingLearner::train(), and PLearn::PLearner::use().
{
static PP<PLearnService> inst;
if(inst.isNull())
inst = new PLearnService();
return *inst;
}


| void PLearn::PLearnService::log_callback | ( | PP< RemotePLearnServer > | server, |
| const string & | module_name, | ||
| int | vlevel, | ||
| const string & | msg | ||
| ) | [static] |
Definition at line 403 of file PLearnService.cc.
References PLearn::flush(), and PL_LOG.
Referenced by PLearn::TestClientCommand::run().
{
PL_LOG(vlevel) << "<From server " << servers_ids[server] << "> [" << module_name << "] " << msg << flush;
}


| void PLearn::PLearnService::progress_callback | ( | PP< RemotePLearnServer > | server, |
| unsigned int | pbar, | ||
| char | action, | ||
| unsigned int | pos = 0, |
||
| const string & | title = "" |
||
| ) | [static] |
Definition at line 410 of file PLearnService.cc.
References PLearn::ProgressBar::getCurrentPlugin(), PLERROR, PLearn::ProgressBar::setPlugin(), and PLearn::tostring().
Referenced by PLearn::TestClientCommand::run().
{
static bool need_to_set_pb_plugin= true;
if(need_to_set_pb_plugin)
{
PP<ProgressBarPlugin> orig_pb_plugin= ProgressBar::getCurrentPlugin();
if(dynamic_cast<NullProgressBarPlugin*>(static_cast<ProgressBarPlugin*>(orig_pb_plugin)) == 0)
ProgressBar::setPlugin(new LineOutputProgressBarPlugin(cerr));
need_to_set_pb_plugin= false;
}
switch(action)
{
case 'A': // add new progress bar
if(progress_bars.find(server) == progress_bars.end())
progress_bars[server]= map<unsigned int, PP<ProgressBar> >();
{//local environment for 'fulltitle'... silly c++ switch/case...
string fulltitle= string("<server#") + tostring(servers_ids[server])
+ ":pb#" + tostring(pbar) + "> " + title;//adjust title w/server info
progress_bars[server][pbar]= new ProgressBar(fulltitle, pos);
}
break;
case 'U': // update progress bar
progress_bars[server][pbar]->update(pos);
break;
case 'K': // kill progress bar
progress_bars[server].erase(pbar);
break;
default:
PLERROR("in PLearnService::progress_callback: unknown action %c", action);
break;
}
}


| PP< RemotePLearnServer > PLearn::PLearnService::reserveServer | ( | ) |
Reserves a remote processing ressource from the pool of servers.
If sucessful returns a pointer to a new RemotePLearnServer If no server is available, returns 0.
Definition at line 181 of file PLearnService.cc.
{
if(available_servers.size()==0)
return 0;
PP<RemotePLearnServer> serv = available_servers.pop();
reserved_servers.insert(serv);
return serv;
}
| TVec< PP< RemotePLearnServer > > PLearn::PLearnService::reserveServers | ( | int | nservers | ) |
Reserves up to nservers servers.
Definition at line 190 of file PLearnService.cc.
References PLearn::TVec< T >::append(), PLearn::PP< T >::isNull(), and PLearn::TVec< T >::length().
Referenced by PLearn::TestClientCommand::run(), PLearn::PLearner::test(), PLearn::BaggingLearner::train(), and PLearn::PLearner::use().
{
TVec< PP<RemotePLearnServer> > servers;
while(servers.length()<nservers)
{
PP<RemotePLearnServer> serv = reserveServer();
if(serv.isNull())
break;
servers.append(serv);
}
return servers;
}


| PP< RemotePLearnServer > PLearn::PLearnService::waitForResult | ( | TVec< PP< RemotePLearnServer > > | servers = TVec< PP<RemotePLearnServer> >(), |
| log_callback_t | the_log_callback = log_callback, |
||
| progress_callback_t | the_progress_callback = progress_callback |
||
| ) |
wait for one result
Definition at line 322 of file PLearnService.cc.
References PLERROR.
Referenced by PLearn::TestClientCommand::run(), PLearn::PLearner::test(), and PLearn::BaggingLearner::train().
{
int min_server= 0;
if(servers.isEmpty())
{
servers= available_servers;
min_server= available_servers.length();
for(std::set<PP<RemotePLearnServer> >::iterator it= reserved_servers.begin();
it != reserved_servers.end(); ++it)
servers.push_back(*it);
}
if(servers.isEmpty())
PLERROR("in PLearnService::waitForResult : cannot wait for a result"
" when you are not connected to any server.");
int server= servers.length();
//send results from reserved servers only even if polling all servers
while((server >= 0 && server < min_server) || server == servers.length())
server= watchServers(servers, the_log_callback, the_progress_callback);
if(server < 0)
PLERROR("in PLearnService::waitForResult : no server returned anything.");
return servers[server];
}

| void PLearn::PLearnService::waitForResultFrom | ( | PP< RemotePLearnServer > | from, |
| log_callback_t | the_log_callback = log_callback, |
||
| progress_callback_t | the_progress_callback = progress_callback |
||
| ) |
Definition at line 351 of file PLearnService.cc.
Referenced by PLearn::RemotePLearnServer::expectResults().
{
PP<RemotePLearnServer> server = waitForResult();
while(server != from)
server = waitForResult();
}

| int PLearn::PLearnService::watchServers | ( | TVec< PP< RemotePLearnServer > > | servers, |
| log_callback_t | the_log_callback, | ||
| progress_callback_t | the_progress_callback | ||
| ) |
Definition at line 241 of file PLearnService.cc.
References c, PLearn::PStream::get(), PLearn::Poll::getNextPendingEvent(), n, PLearn::PStream::peek(), PLERROR, PLearn::Poll::setStreamsToWatch(), and PLearn::Poll::waitForEvents().
{
Poll p;
int n = servers.size();
vector<PStream> streams(n);
for(int k=0; k<n; k++)
streams[k] = servers[k]->io;
p.setStreamsToWatch(streams);
for(;;)
{
PRInt32 npending = p.waitForEvents(PR_INTERVAL_NO_TIMEOUT, true);
if(npending<=0)
return -1;
int the_k= -1;
PStream io = p.getNextPendingEvent();
for(int k=0; k<n; k++)
if(streams[k] == io)
the_k= k;
if(the_k >= 0)
{
int c= io.peek();
// skip blanks one at a time and poll again
if(static_cast<char>(c) == ' ' || static_cast<char>(c) == '\t'
|| static_cast<char>(c) == '\n' || static_cast<char>(c) == '\r')
{
io.get();
continue;
}
else if(static_cast<char>(c) == '*') //async. message (log or progress)
{
io.get(); // get '*'
c= io.get();// get msg type ('L'og or 'P'rogress)
int vlevel, c0;
unsigned int pos, ptr;
string mesg;
string module("");
switch(static_cast<char>(c))
{
case 'L' : // log message
io >> module >> vlevel >> mesg;
the_log_callback(servers[the_k], module, vlevel, mesg);
break;
case 'P' : // progress message
c0= io.get(); // action: 'A'dd, 'U'pdate or 'K'ill
io >> ptr;// pbar id.
if(static_cast<char>(c0) != 'K')
io >> pos;// Add: maxpos; Update: curpos
if(static_cast<char>(c0) == 'A')
io >> mesg;// pbar title
the_progress_callback(servers[the_k], ptr, static_cast<char>(c0), pos, mesg);
break;
default:
PLERROR("PLearnService::watchServers : Expected *L or *P, received *%c", c);
break;
}
}
else if(c == EOF)
PLERROR("Got EOF while reading a RemoteServer's io stream (connection reset by peer?).");
else //synchronous message, return server's id
return the_k;
}
else
PLERROR("stream returned by NextPendingEvent is none of the servers' io field. This should not happen!");
}
#ifndef __INTEL_COMPILER
// To make the compiler happy (never reached). This is unnecessary with
// Intel Compiler, which detects automatically that this statement is never
// reached, and actually gives a warning.
return -1;
#endif
}

| int PLearn::PLearnService::watchServers | ( | TVec< PP< RemotePLearnServer > > | servers, |
| int | timeout = 0 |
||
| ) |
Definition at line 219 of file PLearnService.cc.
References PLearn::Poll::getNextPendingEvent(), n, PLERROR, PLearn::Poll::setStreamsToWatch(), and PLearn::Poll::waitForEvents().
Referenced by PLearn::TestClientCommand::run().
{
Poll p;
int n = servers.size();
vector<PStream> streams(n);
for(int k=0; k<n; k++)
streams[k] = servers[k]->io;
p.setStreamsToWatch(streams);
PRInt32 npending = p.waitForEvents(timeout);
if(npending<=0)
return -1;
PStream io = p.getNextPendingEvent();
for(int k=0; k<n; k++)
if(streams[k] == io)
return k;
PLERROR("stream returned by NextPendingEvent is none of the servers' io field. This should not happen!");
return -1; // To make the compiler happy (never reached).
}


friend class RemotePLearnServer [friend] |
Definition at line 78 of file PLearnService.h.
TVec< PP<RemotePLearnServer> > PLearn::PLearnService::available_servers [private] |
Definition at line 70 of file PLearnService.h.
PLearnService::progress_bars_t PLearn::PLearnService::progress_bars [static, private] |
Definition at line 74 of file PLearnService.h.
std::set< PP<RemotePLearnServer> > PLearn::PLearnService::reserved_servers [private] |
Definition at line 71 of file PLearnService.h.
map< RemotePLearnServer *, pair< string, int > > PLearn::PLearnService::servers_ids [static, private] |
Definition at line 75 of file PLearnService.h.
Referenced by getId().
1.7.4