LimService.C

Go to the documentation of this file.

// Copyright (C) 2001, Compaq Computer Corporation
// 
// This file is part of Vesta.
// 
// Vesta is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// 
// Vesta is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public
// License along with Vesta; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

// Last modified on Sun May 22 22:47:50 EDT 2005 by ken@xorian.net         
//      modified on Fri Apr 22 15:01:07 EDT 2005 by irina.furman@intel.com 
//      modified on Fri Feb 11 19:52:54 PST 2000 by mann   
//      modified on Mon Jun 16 11:07:43 PDT 1997 by heydon 
//      modified on Thu Aug  1 14:45:35 PDT 1996 by levin  

#include <signal.h>
#include <Basics.H>

#include "SRPC.H"
#include "LimService.H"

using std::cerr;

// fatal programmer error
class FatalError {
  public: FatalError() { /*EMPTY*/ }
};

static void WriteFatalError(const SRPC::failure& f) throw ()
{
    cerr << "Fatal SRPC failure in LimService:\n";
    cerr << "  " << f.msg << " (code = " << f.r << ")\n";
    cerr.flush();
    // It seems that this function is lying about what it can throw in
    // order to force program termination (under the assumption the
    // the C++ runtime will handle unexpected exceptions by aborting
    // execution).  Questionable form at best.
    throw FatalError();
}

LimService::LimService(
  const Text &intfName, const int intfVersion,
  int maxRunning, int maxBlocked, LimService_Callback callback,
  LimService_FailureCallback failureCallback, void *arg, long stacksize,
  const Text &hostName, int failCode, const Text &failMessage)
  throw ()
: intfName(intfName), intfVersion(intfVersion), maxRunning(maxRunning),
  maxTotal(maxRunning + maxBlocked), callback(callback),
  failureCallback(failureCallback), arg(arg),
  hostName(hostName), failCode(failCode), failMessage(failMessage),
  sock(NULL)
{
  if(stacksize > 0)
    worker_attributes.set_stacksize(stacksize);

    this->callback2 = NULL;
    // initialize synchronization data
    this->mu.lock();
    this->numRunning = this->numBlocked = 0;
    this->idleWorkers = (WorkerList *)NULL;
    /* Note: "server_state_change" and "server_failure" are initialized
       by the "Forked_Run" method. */
    this->mu.unlock();
}

LimService::LimService(
  const int intfVersion,
  int maxRunning, int maxBlocked, LimService_Callback callback,
  LimService_FailureCallback failureCallback, void *arg, long stacksize,
  const Text &hostName, int failCode, const Text &failMessage)
  throw ()
: intfVersion(intfVersion), maxRunning(maxRunning),
  maxTotal(maxRunning + maxBlocked), callback(callback),
  failureCallback(failureCallback), arg(arg),
  hostName(hostName), failCode(failCode), failMessage(failMessage)
{
  if(stacksize > 0)
    worker_attributes.set_stacksize(stacksize);

    this->callback2 = NULL;
    // initialize "intfName" and "sock"
    this->sock = NEW(TCP_sock);   // choose a port number
    sockaddr_in sin;
    this->sock->get_local_addr(/*OUT*/ sin);
    char buff[10];
    sprintf(buff, "%d", ntohs(sin.sin_port));
    this->intfName = Text(buff);

    // initialize synchronization data
    this->mu.lock();
    this->numRunning = this->numBlocked = 0;
    this->idleWorkers = (WorkerList *)NULL;
    /* Note: "server_state_change" and "server_failure" are initialized
       by the "Forked_Run" method. */
    this->mu.unlock();
}

LimService::LimService(
  const Text &intfName,
  int maxRunning, int maxBlocked, LimService_Callback2 callback,
  LimService_FailureCallback failureCallback, void *arg, long stacksize,
  const Text &hostName, int failCode, const Text &failMessage)
  throw ()
: intfName(intfName), maxRunning(maxRunning),
  maxTotal(maxRunning + maxBlocked), callback2(callback),
  failureCallback(failureCallback), arg(arg),
  hostName(hostName), failCode(failCode), failMessage(failMessage),
  sock(NULL)
{
  if(stacksize > 0)
    worker_attributes.set_stacksize(stacksize);

    this->intfVersion = SRPC::default_intf_version;
    this->callback = NULL;
    // initialize synchronization data
    this->mu.lock();
    this->numRunning = this->numBlocked = 0;
    this->idleWorkers = (WorkerList *)NULL;
    /* Note: "server_state_change" and "server_failure" are initialized
       by the "Forked_Run" method. */
    this->mu.unlock();
}

LimService::LimService(
  const Text &intfName,
  int maxRunning, int maxBlocked, LimService_Callback2 callback,
  LimService_FailureCallback failureCallback, void *arg,
  const Basics::thread_attr &worker_attr,
  const Text &hostName, int failCode, const Text &failMessage)
  throw ()
: intfName(intfName), maxRunning(maxRunning),
  maxTotal(maxRunning + maxBlocked), callback2(callback),
  failureCallback(failureCallback), arg(arg),
  hostName(hostName), failCode(failCode), failMessage(failMessage),
  sock(NULL), worker_attributes(worker_attr)
{
    this->intfVersion = SRPC::default_intf_version;
    this->callback = NULL;
    // initialize synchronization data
    this->mu.lock();
    this->numRunning = this->numBlocked = 0;
    this->idleWorkers = (WorkerList *)NULL;
    /* Note: "server_state_change" and "server_failure" are initialized
       by the "Forked_Run" method. */
    this->mu.unlock();
}

LimService::LimService(
  int maxRunning, int maxBlocked, LimService_Callback2 callback,
  LimService_FailureCallback failureCallback, void *arg, long stacksize,
  const Text &hostName, int failCode, const Text &failMessage)
  throw ()
: maxRunning(maxRunning),
  maxTotal(maxRunning + maxBlocked), callback2(callback),
  failureCallback(failureCallback), arg(arg),
  hostName(hostName), failCode(failCode), failMessage(failMessage)
{
  if(stacksize > 0)
    worker_attributes.set_stacksize(stacksize);

    this->intfVersion = SRPC::default_intf_version;
    this->callback = NULL;
    // initialize "intfName" and "sock"
    this->sock = NEW(TCP_sock);   // choose a port number
    sockaddr_in sin;
    this->sock->get_local_addr(/*OUT*/ sin);
    char buff[10];
    sprintf(buff, "%d", ntohs(sin.sin_port));
    this->intfName = Text(buff);

    // initialize synchronization data
    this->mu.lock();
    this->numRunning = this->numBlocked = 0;
    this->idleWorkers = (WorkerList *)NULL;
    /* Note: "server_state_change" and "server_failure" are initialized
       by the "Forked_Run" method. */
    this->mu.unlock();
}

void LimService::Run() throw (SRPC::failure)
{
    while (true) {
        // create a new SRPC to listen on (this is non-blocking)
        // This will throw an exception to the caller in the event of failure.
        SRPC *srpc = (sock == NULL)
          ? NEW_CONSTR(SRPC, (SRPC::callee, intfName, hostName))
          : NEW_CONSTR(SRPC, (SRPC::callee, sock));

        // signal "Forked_Run" that listener is active
        mu.lock();
        server_state_change = true;
        state_change.signal();
        mu.unlock();

        // wait for a connection
        try {
            int procId = SRPC::default_proc_id;
            int intfVer = intfVersion;
            srpc->await_call(/*INOUT*/ procId, /*INOUT*/ intfVer);

            // invoke a new worker thread to handle the call
            LSWorker *worker = NewWorker();
            worker->Start(srpc, intfVer, procId);
        }
        catch (SRPC::failure f) {
            // Invoke the user-supplied failure callback
            if (this->failureCallback != NULL) {
                (this->failureCallback)(srpc, f, this->arg);
            }
            // clean up the SRPC connection
            delete srpc;
            // go around the loop, creating a new listener
        }
    }  // while
}

void LimService::HandleFailure(const SRPC::failure& f) throw ()
{
    this->mu.lock();
    this->server_failure = f;
    this->server_state_change = true;
    this->state_change.signal();
    this->mu.unlock();
    // Should probably call client's failure proc, but only
    // if failure occurred after Forked_Run has gone away.
}

void* LimService_StartServer(void *arg) throw ()
// "Forked_Run" thread closure
{
    LimService *ls = (LimService *)arg;
    try {
        ls->Run();
    } catch (SRPC::failure f) {
        ls->HandleFailure(f);
    }
    return (void *)NULL;
}

Basics::thread LimService::Forked_Run() throw (SRPC::failure)
{
    Basics::thread th;
    this->mu.lock();
    this->server_failure = SRPC::failure(0, "");
    this->server_state_change = false;
    this->mu.unlock();
    Basics::thread_attr listener_attr(worker_attributes);
    if(listener_attr.get_stacksize() < 100000)
      listener_attr.set_stacksize(100000);
    th.fork(LimService_StartServer, (void *)this, listener_attr);
    this->mu.lock();
    while (!(this->server_state_change)) {
        this->state_change.wait(this->mu);
    }
    SRPC::failure f = this->server_failure; // save failure with "mu" held
    this->mu.unlock();
    if (f.r != 0) throw(f);
    return th;
}

void LimService::DecNumRunning() throw ()
{
    this->mu.lock();
    this->numRunning--;
    assert(this->numRunning < this->maxRunning);
    this->mu.unlock();
    this->server_avail.signal();
}

const int LSErrorCode = -15;
const Text LSErrorMsg =
  Text("LimService error: callback should have thrown SRPC::failure");

void LimService::NewConnection(SRPC *srpc, int intfVersion, int procId)
     throw ()
{
    try {
        while (true) {
            // check that we have enough resources to run; if not, send failure
            this->mu.lock();
            if (this->numRunning + this->numBlocked >= this->maxTotal) {
                // signal failure and exit loop
                this->mu.unlock();
                srpc->send_failure(this->failCode, this->failMessage);
                break;
            }
            // on entry: if any threads are blocked, get in line
            if (this->numBlocked > 0) {
                (this->numBlocked)++;
                this->server_avail.wait(this->mu);
                (this->numBlocked)--;
            }
            // block on condition variable if at max # of running threads
            while (this->numRunning >= this->maxRunning) {
                (this->numBlocked)++;
                this->server_avail.wait(this->mu);
                (this->numBlocked)--;
            }
            (this->numRunning)++;
            this->mu.unlock();

            // call callback function
            try {
              if (this->callback) {
                (this->callback)(srpc, procId, this->arg);
              } else {
                (this->callback2)(srpc, intfVersion, procId, this->arg);
              }
              DecNumRunning();
            }
            catch (...) {
                DecNumRunning();
                throw;
            }

            // wait for another connection on "srpc"
            SRPC::failure f;
            if (!srpc->previous_failure(&f)) {
                procId = SRPC::default_proc_id;
                intfVersion = this->intfVersion;
                srpc->await_call(/*INOUT*/ procId, /*INOUT*/ intfVersion);
            } else
                throw(SRPC::failure(LSErrorCode,
                                    LSErrorMsg + '(' + f.msg + ')'));
        } // while
    } // try
    catch (SRPC::failure f) {
        // If the client caught the "SRPC::failure" exception and failed to
        // re-raise it, report a programmer error and crash.
        if (f.r == LSErrorCode) {
            // A programmer error occurred -- crash!
            WriteFatalError(f);
        } else {
            // otherwise, invoke the user-supplied failure callback
            if (this->failureCallback != NULL) {
                (this->failureCallback)(srpc, f, this->arg);
            }
        }
    }
}

LSWorker *LimService::NewWorker() throw ()
{
    LSWorker *res;
    this->mu.lock();
    if (this->idleWorkers == NULL) {
        // create new LSWorker object
        res = NEW_CONSTR(LSWorker, (this, this->worker_attributes));
    } else {
        // get LSWorker off avail list
        WorkerList *w = this->idleWorkers;
        res = w->worker;
        this->idleWorkers = w->next;
        delete w;
    }
    this->mu.unlock();
    return res;
}

void LimService::RegisterIdleWorker(LSWorker *worker) throw ()
{
    // clean up thread
    worker->Finish();

    // add it to front of list
    WorkerList *cons = NEW(WorkerList);
    cons->worker = worker;
    this->mu.lock();
    cons->next = this->idleWorkers;
    this->idleWorkers = cons;
    this->mu.unlock();
}

// -------------------------------------------------------------------------

LSWorker::LSWorker(LimService *ls, const Basics::thread_attr &attr) throw ()
  : ls(ls), argsReady(false)
{
    // fork the thread that does the actual work
    this->th.fork_and_detach(LimService_Worker, (void *)this, attr);
}

void* LimService_Worker(void* arg) throw ()
{
    LSWorker *worker = (LSWorker *)arg;
    while (true) {
        // wait for work to do
        worker->mu.lock();
        while (!(worker->argsReady)) {
            worker->workToDo.wait(worker->mu);
        }
        worker->argsReady = false;
        worker->mu.unlock();

        // do the work
        worker->ls->NewConnection(worker->srpc, worker->intfVersion,
                                  worker->procId);

        // restore thread to avail list
        worker->ls->RegisterIdleWorker(worker);
    }
    //return (void *)NULL; // not reached
}

void LSWorker::Start(SRPC *srpc, int intfVer, int procId) throw ()
{
    // save arguments
    this->srpc = srpc;
    this->intfVersion = intfVer;
    this->procId = procId;

    // signal the thread to go
    this->mu.lock();
    this->argsReady = true;
    this->mu.unlock();
    this->workToDo.signal();
}

void LSWorker::Finish() throw ()
{
    // clean up the SRPC connection
    delete this->srpc;
    this->srpc = (SRPC *)NULL;
}

---