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().