PLearn 0.1
Public Types | Public Member Functions | Static Public Member Functions | Private Types | Private Member Functions | Private Attributes | Static Private Attributes | Friends
PLearn::PLearnService Class Reference

#include <PLearnService.h>

Inheritance diagram for PLearn::PLearnService:
Inheritance graph
[legend]
Collaboration diagram for PLearn::PLearnService:
Collaboration graph
[legend]

List of all members.

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< RemotePLearnServerreserveServer ()
 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< RemotePLearnServerwaitForResult (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 PLearnServiceinstance ()
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, intgetId (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

Detailed Description

Definition at line 61 of file PLearnService.h.


Member Typedef Documentation

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.


Constructor & Destructor Documentation

PLearn::PLearnService::PLearnService ( ) [private]

Definition at line 76 of file PLearnService.cc.

{}
PLearn::PLearnService::~PLearnService ( )

Definition at line 377 of file PLearnService.cc.

References i, and PLWARNING.

{
    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...
        }

    }
}

Member Function Documentation

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();
}

Here is the caller graph for this function:

void PLearn::PLearnService::connectToServers ( TVec< pair< string, int > >  server_names_and_port)

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);
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

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);
}

Here is the call graph for this function:

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)");
}

Here is the call graph for this function:

void PLearn::PLearnService::disconnectFromServers ( )

Definition at line 139 of file PLearnService.cc.

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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

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]; }

Here is the caller graph for this function:

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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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; 
}

Here is the call graph for this function:

Here is the caller graph for this function:

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;
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

PP< RemotePLearnServer > PLearn::PLearnService::waitForResult ( TVec< PP< RemotePLearnServer > >  servers = TVecPP<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];
}

Here is the caller graph for this function:

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();
}

Here is the caller graph for this function:

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
}

Here is the call graph for this function:

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

Here is the call graph for this function:

Here is the caller graph for this function:


Friends And Related Function Documentation

friend class RemotePLearnServer [friend]

Definition at line 78 of file PLearnService.h.


Member Data Documentation

Definition at line 70 of file PLearnService.h.

Definition at line 74 of file PLearnService.h.

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


The documentation for this class was generated from the following files:
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Defines