PLearn 0.1
PLearnService.cc
Go to the documentation of this file.
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 :
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Defines