LimService.H

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:57:35 EDT 2005 by ken@xorian.net   
//      modified on Fri Feb 11 17:13:07 PST 2000 by mann   
//      modified on Mon Jun 16 11:07:42 PDT 1997 by heydon 
//      modified on Thu Aug  1 14:42:14 PDT 1996 by levin  

// LimService -- An object for exporting an SRPC interface with limits on
//   the resources it will consume.

#ifndef _LIM_SERVICE_H
#define _LIM_SERVICE_H

#include <Basics.H>
#include "SRPC.H"

/* A "LimService" object is an object for exporting an "SRPC" interface. The
  "LimService" constructor takes as arguments the name of the interface to
  export and a server function to be called in response to a remote procedure
  call. Once the remote call completes, the "SRPC" connection used for that
  call is maintained for a future connection.

  The "LimService" constructor also takes two numeric parameters that allow
  the client to specify limits on the resources consumed by the service. One
  parameter is an upper bound on the number of server threads allowed to
  run simultaneously. If that number of threads are already running, and
  another call comes in, the latter call will block until one of the other
  running threads completes. The second parameter is an integer defining the
  additional number of threads that may be blocked. If a new call comes in,
  and the number of blocked threads is at the client-specified limit, then a
  failure is thrown in response to the new call.

  The type of the callback procedure throws the "SRPC::failure" exception. It
  is imperative that if that exception is thrown during execution of the
  callback procedure as a result of a method call on its "srpc" argument that
  the exception is also thrown by the callback. Hence, if the callback
  procedure catches that exception internally, it must re-throw it. Otherwise,
  the LimService machinery cannot de-allocate the resources associated with
  the failed (and hence, dead) SRPC connection.

  It is important that the "LimService" object not be deleted by the main
  server thread so long as the server is in operation. */

// Callback procedure types ---------------------------------------------------

typedef void (*LimService_Callback)(SRPC *srpc, int procId, void *arg)
  /*throw (SRPC::failure)*/;
/* "LimService_Callback" is the type of the procedure called in response to a
   new remote call. */

typedef void (*LimService_Callback2)(SRPC *srpc, int intfVersion,
                                     int procId, void *arg)
  /*throw (SRPC::failure)*/;
/* "LimService_Callback2" is used in place of LimService_Callback if the
   server wants to accept multiple interface versions. */

typedef void (*LimService_FailureCallback)(
  SRPC *srpc, const SRPC::failure &f, void *arg);
/* "LimService_FailureCallback" is the type of the procedure called in
   response to an SRPC failure that occurs during the execution of a server
   callback. */

// The "LimService" object ----------------------------------------------------

// LimService worker thread object (see below)
class LSWorker;

class LimService {
  public:
    LimService(const Text &intfName, const int intfVersion,
      int maxRunning, int maxBlocked, LimService_Callback callback,
      LimService_FailureCallback failureCallback = NULL,
      void *arg = NULL, long stacksize = -1,
      const Text &hostName = "", int failCode = 1,
      const Text &failMessage = Text("connection refused: server busy"))
      throw ();
    /* Initialize a "LimService" object that, when run, will export the
       service named "intfName" on the host named "hostName". If "stacksize"
       is not defaulted, it specifies the stack size (in bytes) of each
       thread forked by the "LimService" object. Use the "Run" method below
       to run an initialized "LimService" object. */

    LimService(const int intfVersion,
      int maxRunning, int maxBlocked, LimService_Callback callback,
      LimService_FailureCallback failureCallback = NULL,
      void *arg = NULL, long stacksize = -1,
      const Text &hostName = "", int failCode = 1,
      const Text &failMessage = Text("connection refused: server busy"))
      throw ();
    /* Initialize a "LimService" object that, when run, will export an
       anonymous service (i.e., an arbitrarily chosen unused port number)
       on the host named "hostName". If "stacksize" is not defaulted,
       it specifies the stack size (in bytes) of each thread forked by
       the "LimService" object. Use the "Run" method below to run an
       initialized "LimService" object. */

    LimService(const Text &intfName,
      int maxRunning, int maxBlocked, LimService_Callback2 callback,
      LimService_FailureCallback failureCallback = NULL,
      void *arg = NULL, long stacksize = -1,
      const Text &hostName = "", int failCode = 1,
      const Text &failMessage = Text("connection refused: server busy"))
      throw ();
    /* Initialize a "LimService" object that, when run, will export the
       service named "intfName" on the host named "hostName". If "stacksize"
       is not defaulted, it specifies the stack size (in bytes) of each
       thread forked by the "LimService" object. Use the "Run" method below
       to run an initialized "LimService" object.  Calls with any interface
       version will be accepted and passed on to the callback. */
    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 = 1,
      const Text &failMessage = Text("connection refused: server busy"))
      throw ();

    LimService(
      int maxRunning, int maxBlocked, LimService_Callback2 callback,
      LimService_FailureCallback failureCallback = NULL,
      void *arg = NULL, long stacksize = -1,
      const Text &hostName = "", int failCode = 1,
      const Text &failMessage = Text("connection refused: server busy"))
      throw ();
    /* Initialize a "LimService" object that, when run, will export an
       anonymous service (i.e., an arbitrarily chosen unused port number)
       on the host named "hostName". If "stacksize" is not defaulted,
       it specifies the stack size (in bytes) of each thread forked by
       the "LimService" object. Use the "Run" method below to run an
       initialized "LimService" object. Calls with any interface
       version will be accepted and passed on to the callback. */

    ~LimService() throw ()
        { if (this->sock != (TCP_sock *)NULL) delete this->sock; }

    void Run() throw (SRPC::failure);
    /* In the following, references to variable names like "intf" correspond
       to the values of the parameters to the LimService object's constructor.

       The "Run" method exports the service named "intfName" from the host
       named "hostName". It blocks until a fatal failure occurs, in which case
       "SRPC::failure" will be thrown (see below). To run a LimService object
       in a forked thread, use the "ForkedRun" method below.

       If "hostName" is defaulted, then the interface is exported from the
       current host. However, if the local machine has multiple IP addresses,
       "hostName" can be used to differentiate among them. The interface will
       be exported with the version number "intfVersion".

       When the exported interface is called in the normal case, a thread is
       forked that calls the user-supplied "callback" function. This
       function is invoked with three arguments: "srpc" is a 
       pointer to the "SRPC" connection for the call, "procId" is the
       identifier for the procedure to invoke, and "arg" is the "arg" value
       passed to the constructor. The "arg" mechanism provides a way of
       communicating data to each invoked thread. In order that the SRPC
       connection may be reused for future calls, the "callback" function
       is required *not* to delete the "srpc" at the end of the call. 

       In the event that there are already "maxRunning" callback functions
       currently running, the new incoming call blocks until one or more of
       those threads complete.

       In the event that there are already "maxBlocked" threads blocked (in 
       addition to the "maxRunning" running threads), the service immediately
       throws "SRPC::failure" down the connection with the integer code
       "failCode" and the message "failMessage". In this case, the "callback"
       function is not invoked.

       If the exception "SRPC::failure" is thrown in the process of creating
       or waiting on an SRPC connection, the "Run" method exits with that
       exception. Any threads that were already running the "callback"
       function in the server will continue to run until they complete.

       If the "SRPC::failure" exception is thrown by the user-supplied
       "callback" function, the exception is caught internally. If the
       exception indicates a programmer error, an error message is
       written to the standard error output, and the entire process exits.
       Otherwise, the "failureCallback" is invoked with the current SRPC
       connection, the failure obect itself, and the "arg" value passed to
       the "LimService" constructor as arguments. If "failureCallback" is
       "NULL", then the failure is silently ignored. */

    Basics::thread Forked_Run() throw (SRPC::failure);
    /* This function forks a thread to execute a call of Run() and returns
       the thread as its result.  Forked_Run doesn't return until the
       SRPC connection passed to the LimService_Callback invoked from Run
       is ready to accept connection requests.  This is a stronger guarantee
       than if a client simply forked the Run() operation directly. */

    Text IntfName() throw () { return this->intfName; }
      /* returns the interface name used by the LimService object.  If
         the first constructor above was used, this is simply its first
         parameter; otherwise, it is the port number (converted to ASCII)
         chosen for the (anonymous) interface. */

  private:
    friend void* LimService_Worker(void *) throw ();

    void DecNumRunning() throw ();
    /* Decrement the number of running threads and signal the condition
       variable "server_avail". */

    void HandleFailure(const SRPC::failure& f) throw ();
    /* Stash the failure "f" in "this->server_failure", and signal a state
       change to "ForkedRun". */

    friend void *LimService_StartServer(void *) throw ();
    /* Callback function used by "ForkedRun". This invokes the "Run" method to
       run the server. In the event of an "SRPC::failure", it invokes the
       "HandleFailure" method to synchronize with the "Forked_Run" method. */

    void NewConnection(SRPC *srpc, int intfVersion, int procId) throw ();
    /* Callback function invoked when a new SRPC connection is established.
       It checks that the server can run another thread, and then invokes
       the client's "callback" function. Once the client's callback completes,
       this function waits for another connection on the same SRPC object.
       The loop exits if there is an SRPC either while waiting for a new
       connection or in executing the "callback". In either case, the
       client's "failure_callback" is invoked. */

    LSWorker *NewWorker() throw ();
    /* Return an available worker thread. To start the thread running, invoke
       the LSWorker::Start method below. */

    void RegisterIdleWorker(LSWorker *worker) throw ();
    /* Return "worker" to the list of available idle worker threads. This
       method should be called by the worker thread when it is done with
       its job. */

    // linked list of worker threads
    struct WorkerList {
        LSWorker *worker;
        WorkerList *next;
    };

    // readonly after initialization (constructor arguments)
    Text intfName;              // first constructor only
    int intfVersion;
    int maxRunning;
    int maxTotal;               // max total = #(running and blocked threads)
    LimService_Callback callback;   // exactly one of callback, callback2
    LimService_Callback2 callback2; //  is filled in; the other is NULL.
    LimService_FailureCallback failureCallback;
    void *arg;
    Basics::thread_attr worker_attributes;
    Text hostName;
    int failCode;
    Text failMessage;
    TCP_sock *sock;             // used by second constructor only

    // mutable fields
    Basics::mutex mu;              // protects the following fields:
    int numRunning, numBlocked;    // # of threads running and blocked
    Basics::cond server_avail;     // this->numRunning < this->maxRunning
    WorkerList *idleWorkers;       // linked list of idle worker threads
    bool server_state_change;      // true iff "Forked_Run" done
    Basics::cond state_change;     // == server_state_change
    SRPC::failure server_failure;  // saved failure used by "Forked_Run"

    // Invariants:
    // I1. numRunning <= maxRunning
    // I2. numRunning + numBlocked <= maxTotal

    /* prevent copying */

    LimService(const LimService &);
    LimService(LimService &);
    // Copy constructor; should never be accessed.

    void operator=(LimService &);

};

class LSWorker {
  public:
    LSWorker(LimService *ls, const Basics::thread_attr &attr) throw ();
    /* Construct a new LSWorker thread for "ls" that is waiting to run.
       Use the "Start" method below to start the thread running. */

    void Start(SRPC *srpc, int intfVer, int procId) throw ();
    /* Run this worker thread on the given arguments. */

    void Finish() throw ();
    /* This function must be called once the worker callback completes. */

  private:
    friend void *LimService_Worker(void *arg) throw ();
    /* Application function for the forked thread invoked by the
       "LSWorker" constructor. This function never returns. */

    // the thread itself
    Basics::thread th;

    // thread arguments
    LimService *ls;
    SRPC *srpc;
    int intfVersion;
    int procId;

    // synchronization fields
    Basics::mutex mu;      // protects following fields
    bool argsReady;        // does thread have work to do?
    Basics::cond workToDo; // == argsReady
};

#endif // _LIM_SERVICE_H

---