PLearn 0.1
|
00001 // -*- C++ -*- 00002 00003 // PLearnService.cc 00004 // 00005 // Copyright (C) 2005 Pascal Vincent 00006 // Copyright (C) 2007 Xavier Saint-Mleux, ApSTAT Technologies inc. 00007 // 00008 // Redistribution and use in source and binary forms, with or without 00009 // modification, are permitted provided that the following conditions are met: 00010 // 00011 // 1. Redistributions of source code must retain the above copyright 00012 // notice, this list of conditions and the following disclaimer. 00013 // 00014 // 2. Redistributions in binary form must reproduce the above copyright 00015 // notice, this list of conditions and the following disclaimer in the 00016 // documentation and/or other materials provided with the distribution. 00017 // 00018 // 3. The name of the authors may not be used to endorse or promote 00019 // products derived from this software without specific prior written 00020 // permission. 00021 // 00022 // THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND ANY EXPRESS OR 00023 // IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES 00024 // OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN 00025 // NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 00026 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 00027 // TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 00028 // PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 00029 // LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 00030 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 00031 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 00032 // 00033 // This file is part of the PLearn library. For more information on the PLearn 00034 // library, go to the PLearn Web site at www.plearn.org 00035 00036 /* ******************************************************* 00037 * $Id: PLearnService.cc 9759 2008-12-09 16:14:33Z nouiz $ 00038 ******************************************************* */ 00039 00040 // Authors: Pascal Vincent 00041 00045 #include "PLearnService.h" 00046 #include "RemotePLearnServer.h" 00047 #include <plearn/io/openFile.h> 00048 #include <plearn/io/openSocket.h> 00049 #include <plearn/io/pl_log.h> 00050 #include <plearn/io/Poll.h> 00051 00052 namespace PLearn { 00053 using namespace std; 00054 00055 00056 /* 00057 void PLearnService::remoteLaunchServers(int nservers, int tcpport, const string& launch_command) 00058 { 00059 string full_command = launch_command+'--tcp '+tostring(tcpport); 00060 for(int k=0; k<nservers; k++) 00061 { 00062 PP<Popen> p = new Popen(full_command); 00063 } 00064 reserved_servers.add() 00065 } 00066 */ 00067 00068 PLearnService& PLearnService::instance() 00069 { 00070 static PP<PLearnService> inst; 00071 if(inst.isNull()) 00072 inst = new PLearnService(); 00073 return *inst; 00074 } 00075 00076 PLearnService::PLearnService() 00077 {} 00078 00079 map<RemotePLearnServer*, pair<string,int> > PLearnService::servers_ids;//init. 00080 void PLearnService::connectToServers(PPath serversfile) 00081 { 00082 PStream in = openFile(serversfile, PStream::raw_ascii, "r"); 00083 00084 DBG_LOG << "PLearnService::connectToServers(" << serversfile << ')' << endl; 00085 00086 string hostname; 00087 string pid; 00088 int tcpport = -1; 00089 00090 TVec< pair<string,int> > hostname_and_port; 00091 00092 while(in.good()) 00093 { 00094 in.skipBlanksAndComments(); 00095 if(!in) 00096 break; 00097 in >> hostname >> tcpport >> pid; 00098 if(in.good()) 00099 hostname_and_port.append(pair<string,int>(hostname,tcpport)); 00100 } 00101 connectToServers(hostname_and_port); 00102 } 00103 00104 void PLearnService::connectToServers(TVec< pair<string,int> > hostname_and_port) 00105 00106 { 00107 00108 DBG_LOG << "PLearnService::connectToServers(" << hostname_and_port << ')' << endl; 00109 00110 if(available_servers.size()>0) 00111 disconnectFromServers(); 00112 for(int k=0; k<hostname_and_port.length(); k++) 00113 { 00114 pair<string, int> host_port = hostname_and_port[k]; 00115 PStream servio = openSocket(host_port.first, host_port.second, PStream::plearn_ascii); 00116 PP<RemotePLearnServer> serv = new RemotePLearnServer(servio); 00117 serv->callFunction("binary"); 00118 00119 TVec<PP<RemotePLearnServer> > ss; 00120 ss.push_back(serv); 00121 00122 servers_ids[serv]= host_port; 00123 00124 watchServers(ss, log_callback, progress_callback); 00125 00126 servio << PStream::plearn_binary; 00127 00128 reserved_servers.insert(serv); 00129 serv->getResults(); 00130 serv->callFunction("loggingControl", 00131 PL_Log::instance().verbosity(), 00132 PL_Log::instance().namedLogging()); 00133 serv->getResults(); 00134 reserved_servers.erase(serv); 00135 available_servers.push(serv); 00136 } 00137 } 00138 00139 void PLearnService::disconnectFromServers() 00140 { 00141 while(available_servers.length() > 0) 00142 disconnectFromServer(available_servers[0]); 00143 } 00144 00145 00146 void PLearnService::disconnectFromServer(PP<RemotePLearnServer> server) 00147 { 00148 for(int i= 0; i < available_servers.length(); ++i) 00149 if(available_servers[i] == server) 00150 { 00151 available_servers.remove(i); 00152 server->io.write("!Q "); 00153 server->io << endl; 00154 if(progress_bars.find(server) != progress_bars.end()) 00155 progress_bars.erase(server); 00156 return; 00157 } 00158 PLERROR("PLearnService::disconnectFromServer : trying to disconnect from a server which is not available" 00159 " (not connected to or reserved)"); 00160 } 00161 00162 int PLearnService::availableServers() const 00163 { 00164 return available_servers.size(); 00165 } 00166 00167 /* 00168 RemotePLearnServer* PLearnService::OLD_reserveServer() 00169 { 00170 RemotePLearnServer* serv = 0; 00171 if(available_servers.size()>0) 00172 { 00173 int servnum = available_servers.pop(); 00174 serv = new RemotePLearnServer(serversio[servnum]); 00175 reserved_servers[serv] = servnum; 00176 } 00177 return serv; 00178 } 00179 */ 00180 00181 PP<RemotePLearnServer> PLearnService::reserveServer() 00182 { 00183 if(available_servers.size()==0) 00184 return 0; 00185 PP<RemotePLearnServer> serv = available_servers.pop(); 00186 reserved_servers.insert(serv); 00187 return serv; 00188 } 00189 00190 TVec< PP<RemotePLearnServer> > PLearnService::reserveServers(int nservers) 00191 { 00192 TVec< PP<RemotePLearnServer> > servers; 00193 while(servers.length()<nservers) 00194 { 00195 PP<RemotePLearnServer> serv = reserveServer(); 00196 if(serv.isNull()) 00197 break; 00198 servers.append(serv); 00199 } 00200 return servers; 00201 } 00202 00203 void PLearnService::freeServer(PP<RemotePLearnServer> server) 00204 { 00205 DBG_LOG << "PLearnService::freeServer(...)" << endl; 00206 server->deleteAllObjects(); 00207 server->clearMaps(); 00208 if(reserved_servers.erase(server)!=1) 00209 PLERROR("Problem in PLearnService::freeServer are you sure this server had been properly reserved?"); 00210 available_servers.push(server); 00211 } 00212 00213 void PLearnService::freeServers(TVec< PP<RemotePLearnServer> > servers) 00214 { 00215 for(int k=0; k<servers.length(); k++) 00216 freeServer(servers[k]); 00217 } 00218 00219 int PLearnService::watchServers(TVec< PP<RemotePLearnServer> > servers, int timeout) 00220 { 00221 Poll p; 00222 int n = servers.size(); 00223 vector<PStream> streams(n); 00224 for(int k=0; k<n; k++) 00225 streams[k] = servers[k]->io; 00226 p.setStreamsToWatch(streams); 00227 00228 PRInt32 npending = p.waitForEvents(timeout); 00229 if(npending<=0) 00230 return -1; 00231 00232 PStream io = p.getNextPendingEvent(); 00233 for(int k=0; k<n; k++) 00234 if(streams[k] == io) 00235 return k; 00236 PLERROR("stream returned by NextPendingEvent is none of the servers' io field. This should not happen!"); 00237 return -1; // To make the compiler happy (never reached). 00238 } 00239 00240 00241 int PLearnService::watchServers(TVec< PP<RemotePLearnServer> > servers, 00242 log_callback_t the_log_callback, 00243 progress_callback_t the_progress_callback) 00244 { 00245 Poll p; 00246 int n = servers.size(); 00247 vector<PStream> streams(n); 00248 for(int k=0; k<n; k++) 00249 streams[k] = servers[k]->io; 00250 p.setStreamsToWatch(streams); 00251 00252 for(;;) 00253 { 00254 PRInt32 npending = p.waitForEvents(PR_INTERVAL_NO_TIMEOUT, true); 00255 if(npending<=0) 00256 return -1; 00257 00258 int the_k= -1; 00259 PStream io = p.getNextPendingEvent(); 00260 00261 for(int k=0; k<n; k++) 00262 if(streams[k] == io) 00263 the_k= k; 00264 00265 if(the_k >= 0) 00266 { 00267 int c= io.peek(); 00268 // skip blanks one at a time and poll again 00269 if(static_cast<char>(c) == ' ' || static_cast<char>(c) == '\t' 00270 || static_cast<char>(c) == '\n' || static_cast<char>(c) == '\r') 00271 { 00272 io.get(); 00273 continue; 00274 } 00275 else if(static_cast<char>(c) == '*') //async. message (log or progress) 00276 { 00277 io.get(); // get '*' 00278 c= io.get();// get msg type ('L'og or 'P'rogress) 00279 00280 int vlevel, c0; 00281 unsigned int pos, ptr; 00282 string mesg; 00283 string module(""); 00284 switch(static_cast<char>(c)) 00285 { 00286 case 'L' : // log message 00287 io >> module >> vlevel >> mesg; 00288 the_log_callback(servers[the_k], module, vlevel, mesg); 00289 break; 00290 case 'P' : // progress message 00291 c0= io.get(); // action: 'A'dd, 'U'pdate or 'K'ill 00292 io >> ptr;// pbar id. 00293 if(static_cast<char>(c0) != 'K') 00294 io >> pos;// Add: maxpos; Update: curpos 00295 if(static_cast<char>(c0) == 'A') 00296 io >> mesg;// pbar title 00297 the_progress_callback(servers[the_k], ptr, static_cast<char>(c0), pos, mesg); 00298 break; 00299 default: 00300 PLERROR("PLearnService::watchServers : Expected *L or *P, received *%c", c); 00301 break; 00302 } 00303 } 00304 else if(c == EOF) 00305 PLERROR("Got EOF while reading a RemoteServer's io stream (connection reset by peer?)."); 00306 else //synchronous message, return server's id 00307 return the_k; 00308 } 00309 else 00310 PLERROR("stream returned by NextPendingEvent is none of the servers' io field. This should not happen!"); 00311 } 00312 #ifndef __INTEL_COMPILER 00313 // To make the compiler happy (never reached). This is unnecessary with 00314 // Intel Compiler, which detects automatically that this statement is never 00315 // reached, and actually gives a warning. 00316 return -1; 00317 #endif 00318 } 00319 00320 00322 PP<RemotePLearnServer> PLearnService::waitForResult(TVec< PP<RemotePLearnServer> > servers, 00323 log_callback_t the_log_callback, 00324 progress_callback_t the_progress_callback) 00325 { 00326 int min_server= 0; 00327 if(servers.isEmpty()) 00328 { 00329 servers= available_servers; 00330 min_server= available_servers.length(); 00331 for(std::set<PP<RemotePLearnServer> >::iterator it= reserved_servers.begin(); 00332 it != reserved_servers.end(); ++it) 00333 servers.push_back(*it); 00334 } 00335 00336 if(servers.isEmpty()) 00337 PLERROR("in PLearnService::waitForResult : cannot wait for a result" 00338 " when you are not connected to any server."); 00339 int server= servers.length(); 00340 00341 //send results from reserved servers only even if polling all servers 00342 while((server >= 0 && server < min_server) || server == servers.length()) 00343 server= watchServers(servers, the_log_callback, the_progress_callback); 00344 00345 if(server < 0) 00346 PLERROR("in PLearnService::waitForResult : no server returned anything."); 00347 return servers[server]; 00348 } 00349 00350 00351 void PLearnService::waitForResultFrom(PP<RemotePLearnServer> from, 00352 log_callback_t the_log_callback, 00353 progress_callback_t the_progress_callback) 00354 { 00355 PP<RemotePLearnServer> server = waitForResult(); 00356 while(server != from) 00357 server = waitForResult(); 00358 } 00359 00360 00361 /* 00362 void PLearnService::freeServer(RemotePLearnServer* remoteserv) 00363 { 00364 DBG_LOG << "PLearnService::freeServer(" << (unsigned long) remoteserv << ")" << endl; 00365 remoteserv->clearMaps(); 00366 remoteserv->deleteAllObjects(); 00367 std::map<RemotePLearnServer*,int>::iterator it = reserved_servers.find(remoteserv); 00368 if(it==reserved_servers.end()) 00369 PLERROR("Strange bug in PLearnService::freeServer Nothing known about this remoteserv. This should never happen!"); 00370 // put servernum back in available servers 00371 available_servers.push(it->second); 00372 // erase servernum from reserved_servers 00373 reserved_servers.erase(it); 00374 } 00375 */ 00376 00377 PLearnService::~PLearnService() 00378 { 00379 if(reserved_servers.size() != 0) 00380 PLWARNING("PLearnService::~PLearnService : some servers are still reserved; free them first."); 00381 00382 TVec<PP<RemotePLearnServer> > servers(available_servers.length()); 00383 servers << available_servers; 00384 00385 disconnectFromServers(); 00386 00387 //now, get what's remaining on the servers streams 00388 for(int i= 0; i < servers.length(); ++i) 00389 { 00390 try 00391 { 00392 for(;;) watchServers(servers, log_callback, progress_callback); 00393 } 00394 catch(const PLearnError&) 00395 { 00396 // do nothing... 00397 } 00398 00399 } 00400 } 00401 00402 00403 void PLearnService::log_callback(PP<RemotePLearnServer> server, const string& module_name, int vlevel, const string& msg) 00404 { 00405 PL_LOG(vlevel) << "<From server " << servers_ids[server] << "> [" << module_name << "] " << msg << flush; 00406 } 00407 00408 PLearnService::progress_bars_t PLearnService::progress_bars; // init 00409 00410 void PLearnService::progress_callback(PP<RemotePLearnServer> server, unsigned int pbar, char action, 00411 unsigned int pos, const string& title) 00412 { 00413 static bool need_to_set_pb_plugin= true; 00414 if(need_to_set_pb_plugin) 00415 { 00416 PP<ProgressBarPlugin> orig_pb_plugin= ProgressBar::getCurrentPlugin(); 00417 if(dynamic_cast<NullProgressBarPlugin*>(static_cast<ProgressBarPlugin*>(orig_pb_plugin)) == 0) 00418 ProgressBar::setPlugin(new LineOutputProgressBarPlugin(cerr)); 00419 need_to_set_pb_plugin= false; 00420 } 00421 00422 switch(action) 00423 { 00424 case 'A': // add new progress bar 00425 if(progress_bars.find(server) == progress_bars.end()) 00426 progress_bars[server]= map<unsigned int, PP<ProgressBar> >(); 00427 {//local environment for 'fulltitle'... silly c++ switch/case... 00428 string fulltitle= string("<server#") + tostring(servers_ids[server]) 00429 + ":pb#" + tostring(pbar) + "> " + title;//adjust title w/server info 00430 progress_bars[server][pbar]= new ProgressBar(fulltitle, pos); 00431 } 00432 break; 00433 case 'U': // update progress bar 00434 progress_bars[server][pbar]->update(pos); 00435 break; 00436 case 'K': // kill progress bar 00437 progress_bars[server].erase(pbar); 00438 break; 00439 default: 00440 PLERROR("in PLearnService::progress_callback: unknown action %c", action); 00441 break; 00442 } 00443 } 00444 00445 // Connect the global PLearnService instance to the listed servers. 00446 void globalConnectToServers(TVec< pair<string,int> > hostname_and_port) 00447 { 00448 PLearnService::instance().connectToServers(hostname_and_port); 00449 } 00450 00451 BEGIN_DECLARE_REMOTE_FUNCTIONS 00452 declareFunction("globalConnectToServers", &globalConnectToServers, 00453 (BodyDoc("Connect the PLearnService instance to the listed servers."), 00454 ArgDoc ("hostname_and_port", "List of ('hostname', port#) pairs"))); 00455 END_DECLARE_REMOTE_FUNCTIONS 00456 00457 00458 } // end of namespace PLearn 00459 00460 00461 /* 00462 Local Variables: 00463 mode:c++ 00464 c-basic-offset:4 00465 c-file-style:"stroustrup" 00466 c-file-offsets:((innamespace . 0)(inline-open . 0)) 00467 indent-tabs-mode:nil 00468 fill-column:79 00469 End: 00470 */ 00471 // vim: filetype=cpp:expandtab:shiftwidth=4:tabstop=8:softtabstop=4:encoding=utf-8:textwidth=79 :