giopserver.cc

来自「编译工具」· CC 代码 · 共 1,880 行 · 第 1/4 页

// -*- Mode: C++; -*-
//                            Package   : omniORB
// giopServer.cc              Created on: 20 Dec 2000
//                            Author    : Sai Lai Lo (sll)
//
//    Copyright (C) 2000 AT&T Laboratories Cambridge
//
//    This file is part of the omniORB library
//
//    The omniORB library is free software; you can redistribute it and/or
//    modify it under the terms of the GNU Library General Public
//    License as published by the Free Software Foundation; either
//    version 2 of the License, or (at your option) any later version.
//
//    This library 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
//    Library General Public License for more details.
//
//    You should have received a copy of the GNU Library General Public
//    License along with this library; if not, write to the Free
//    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
//    02111-1307, USA
//
//
// Description:
//	*** PROPRIETORY INTERFACE ***
//

/*
  $Log: giopServer.cc,v $
  Revision 1.22.2.46  2005/11/20 15:33:03  dgrisby
  New connection cleanup meant some connection state was not completely
  deleted.

  Revision 1.22.2.45  2005/11/17 11:35:52  dgrisby
  In thread pool mode, bidir workers become wedged waiting for incoming
  calls, so they must be closed in giopServer::deactivate.

  Revision 1.22.2.44  2005/11/16 17:35:25  dgrisby
  New connectionWatchPeriod and connectionWatchImmediate parameters.

  Revision 1.22.2.43  2005/11/15 11:07:56  dgrisby
  More shutdown cleanup.

  Revision 1.22.2.42  2005/11/14 15:20:08  dgrisby
  Race between giopServer destruction and closure of connections in
  thread pool mode.

  Revision 1.22.2.41  2005/11/14 10:58:23  dgrisby
  Better connection / thread shutdown behaviour.

  Revision 1.22.2.40  2005/11/04 14:26:02  dgrisby
  Open connections would prevent shutdown on Windows.

  Revision 1.22.2.39  2005/10/13 11:38:16  dgrisby
  Dump CloseConnection messages.

  Revision 1.22.2.38  2005/05/03 10:10:48  dgrisby
  Avoid deadlock caused by trying to deactivation SSL rendezvouser more
  than once.

  Revision 1.22.2.37  2005/04/10 22:17:19  dgrisby
  Fixes to connection management. Thanks Jon Biggar.

  Revision 1.22.2.36  2005/03/10 11:28:29  dgrisby
  Race condition between setSelectable / clearSelectable.

  Revision 1.22.2.35  2005/02/23 12:27:31  dgrisby
  Another race in setSelectable with connection shutdown. Thanks Peter
  Klotz.

  Revision 1.22.2.34  2005/02/13 20:52:51  dgrisby
  Change threadPoolWatchConnection parameter to be an integer rather
  than a boolean. Value is the count of threads that can be handling a
  connection when one decides to watch it.

  Revision 1.22.2.33  2004/12/20 18:17:55  dgrisby
  Avoid race condition on connection disconnection.

  Revision 1.22.2.32  2004/10/17 20:14:32  dgrisby
  Updated support for OpenVMS. Many thanks to Bruce Visscher.

  Revision 1.22.2.31  2004/09/13 10:03:13  dgrisby
  Fix for giopServer shutdown issue. Thanks Serguei Kolos.

  Revision 1.22.2.30  2004/04/08 10:02:20  dgrisby
  In thread pool mode, close connections that will not be selectable.

  Revision 1.22.2.29  2004/03/02 15:32:07  dgrisby
  Fix locking in connection shutdown.

  Revision 1.22.2.28  2004/02/27 12:40:09  dgrisby
  Fix race conditions in connection shutdown. Thanks Serguei Kolos.

  Revision 1.22.2.27  2003/10/20 16:11:12  dgrisby
  Race condition in connection closure.

  Revision 1.22.2.26  2003/07/16 14:22:38  dgrisby
  Speed up oneway handling a little. More tracing for split messages.

  Revision 1.22.2.25  2003/05/22 13:47:40  dgrisby
  Failed to setSelectable in some cases.

  Revision 1.22.2.24  2003/02/17 01:46:23  dgrisby
  Pipe to kick select thread (on Unix).

  Revision 1.22.2.23  2002/09/09 22:11:50  dgrisby
  SSL transport cleanup even if certificates are wrong.

  Revision 1.22.2.22  2002/09/04 23:29:30  dgrisby
  Avoid memory corruption with multiple list removals.

  Revision 1.22.2.21  2002/08/21 06:23:15  dgrisby
  Properly clean up bidir connections and ropes. Other small tweaks.

  Revision 1.22.2.20  2002/03/18 16:50:18  dpg1
  New threadPoolWatchConnection parameter.

  Revision 1.22.2.19  2002/03/13 16:05:39  dpg1
  Transport shutdown fixes. Reference count SocketCollections to avoid
  connections using them after they are deleted. Properly close
  connections when in thread pool mode.

  Revision 1.22.2.18  2002/02/13 16:02:39  dpg1
  Stability fixes thanks to Bastiaan Bakker, plus threading
  optimisations inspired by investigating Bastiaan's bug reports.

  Revision 1.22.2.17  2002/02/01 11:21:19  dpg1
  Add a ^L to comments.

  Revision 1.22.2.16  2001/09/20 13:26:14  dpg1
  Allow ORB_init() after orb->destroy().

  Revision 1.22.2.15  2001/09/19 17:26:49  dpg1
  Full clean-up after orb->destroy().

  Revision 1.22.2.14  2001/09/10 17:47:57  sll
  startIdleCounter in csInsert.

  Revision 1.22.2.13  2001/08/21 11:02:15  sll
  orbOptions handlers are now told where an option comes from. This
  is necessary to process DefaultInitRef and InitRef correctly.

  Revision 1.22.2.12  2001/08/17 17:12:37  sll
  Modularise ORB configuration parameters.

  Revision 1.22.2.11  2001/07/31 16:28:00  sll
  Added GIOP BiDir support.

  Revision 1.22.2.10  2001/07/13 15:27:42  sll
  Support for the thread-pool as well as the thread-per-connection policy
  Support for serving multiple incoming calls on the same connection
  simultaneously.

  Revision 1.22.2.9  2001/06/20 18:35:17  sll
  Upper case send,recv,connect,shutdown to avoid silly substutition by
  macros defined in socket.h to rename these socket functions
  to something else.

  Revision 1.22.2.8  2001/06/11 18:00:52  sll
  Fixed silly mistake in shutdown multiple endpoints.

  Revision 1.22.2.7  2001/04/18 18:10:49  sll
  Big checkin with the brand new internal APIs.


*/

#include <omniORB4/CORBA.h>
#include <invoker.h>
#include <giopServer.h>
#include <giopWorker.h>
#include <giopRendezvouser.h>
#include <giopMonitor.h>
#include <giopStrand.h>
#include <giopStream.h>
#include <giopStreamImpl.h>
#include <initialiser.h>
#include <omniORB4/omniInterceptors.h>
#include <orbOptions.h>
#include <orbParameters.h>

OMNI_NAMESPACE_BEGIN(omni)

////////////////////////////////////////////////////////////////////////////
//             Configuration options                                      //
////////////////////////////////////////////////////////////////////////////
CORBA::Boolean orbParameters::threadPerConnectionPolicy      = 1;
//   1 means the ORB should dedicate one thread per connection on the 
//   server side. 0 means the ORB should dispatch a thread from a pool
//   to a connection only when a request has arrived.
//
//  Valid values = 0 or 1

CORBA::ULong   orbParameters::threadPerConnectionUpperLimit  = 10000;
//   If the one thread per connection is in effect, this number is
//   the max. no. of connections the server will allow before it
//   switch off the one thread per connection policy and move to
//   the thread pool policy.
//
//   Valid values = (n >= 1) 

CORBA::ULong   orbParameters::threadPerConnectionLowerLimit  = 9000;
//   If the one thread per connection was in effect and was switched
//   off because threadPerConnectionUpperLimit has been exceeded
//   previously, this number tells when the policy should be restored
//   when the number of connections drop.
//
//   Valid values = (n >= 1 && n < threadPerConnectionUpperLimit) 

CORBA::ULong   orbParameters::maxServerThreadPerConnection   = 100;
//   The max. no. of threads the server will dispatch to server the
//   requests coming from one connection.
//
//   Valid values = (n >= 1) 

CORBA::ULong   orbParameters::maxServerThreadPoolSize        = 100;
//   The max. no. of threads the server will allocate to do various
//   ORB tasks. This number does not include the dedicated thread
//   per connection when the threadPerConnectionPolicy is in effect
//
//   Valid values = (n >= 1) 

CORBA::ULong   orbParameters::threadPoolWatchConnection      = 1;
//   After dispatching an upcall in thread pool mode, the thread that
//   has just performed the call can watch the connection for a short
//   time before returning to the pool. This leads to less thread
//   switching for a series of calls from a single client, but is less
//   fair if there are concurrent clients. The connection is watched
//   if the number of threads concurrently handling the connection is
//   <= the value of this parameter. i.e. if the parameter is zero,
//   the connection is never watched; if it is 1, the last thread
//   managing a connection watches it; if 2, the connection is still
//   watched if there is one other thread still in an upcall for the
//   connection, and so on.
//
//  Valid values = (n >= 0)

CORBA::Boolean orbParameters::connectionWatchImmediate      = 0;
//   When a thread handles an incoming call, it unmarshals the
//   arguments then marks the connection as watchable by the connection
//   watching thread, in case the client sends a concurrent call on the
//   same connection. If this parameter is set to the default false,
//   the connection is not actually watched until the next connection
//   watch period (determined by the connectionWatchPeriod parameter).
//   If connectionWatchImmediate is set true, the connection watching
//   thread is immediately signalled to watch the connection. That
//   leads to faster interactive response to clients that multiplex
//   calls, but adds significant overhead along the call chain.
//
//   Note that this setting has no effect on Windows, since it has no
//   mechanism for signalling the connection watching thread.


////////////////////////////////////////////////////////////////////////////
static const char* plural(CORBA::ULong val)
{
  return val == 1 ? "" : "s";
}


////////////////////////////////////////////////////////////////////////////
giopServer::giopServer() : pd_state(IDLE), pd_nconnections(0),
			   pd_cond(&pd_lock),
			   pd_n_temporary_workers(0),
			   pd_n_dedicated_workers(0)
{
  pd_thread_per_connection = orbParameters::threadPerConnectionPolicy;
  pd_connectionState = new connectionState*[connectionState::hashsize];
  for (CORBA::ULong i=0; i < connectionState::hashsize; i++) {
    pd_connectionState[i] = 0;
  }
}

////////////////////////////////////////////////////////////////////////////
giopServer::~giopServer()
{
  singleton() = 0;
  delete [] pd_connectionState;
}


////////////////////////////////////////////////////////////////////////////
const char*
giopServer::instantiate(const char* uri,
			CORBA::Boolean no_publish,
			CORBA::Boolean no_listen)
{
  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock,0);
  omni_tracedmutex_lock sync(pd_lock);

  ensureNotInFlux();

  if (!no_listen) {

    giopEndpoint* ept = giopEndpoint::str2Endpoint(uri);
    if (!ept) return 0;

    OMNIORB_ASSERT(pd_state != ZOMBIE);

    if (ept->Bind()) {
      pd_endpoints.push_back(ept);
      if (pd_state == ACTIVE) activate();
      uri =  ept->address();
    }
    else {
      delete ept;
      return 0;
    }
  }
  else if ( !giopEndpoint::strIsValidEndpoint(uri) ) {
    return 0;
  }

  if (!no_publish) {
    (void)giopEndpoint::addToIOR(uri);
  }
  return uri;
}

////////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopServer::addBiDirStrand(giopStrand* s,giopActiveCollection* watcher) {

  OMNIORB_ASSERT(s->isClient() && s->biDir && s->connection);
  {
    ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);
    omni_tracedmutex_lock sync(*omniTransportLock);
    s->connection->incrRefCount();
  }

  CORBA::Boolean status = 0;

  {
    ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock,0);
    omni_tracedmutex_lock sync(pd_lock);

    ensureNotInFlux();

    if (pd_state == ACTIVE) {

      pd_bidir_strands.push_back(s);

      CORBA::Boolean matched = 0;
      {
	omnivector<giopActiveCollection*>::iterator i = pd_bidir_collections.begin();
	omnivector<giopActiveCollection*>::iterator last = pd_bidir_collections.end();
	while (i != last) {
	  if ((*i) == watcher) {
	    matched = 1;
	    break;
	  }
	  ++i;
	}
      }
      if (!matched) {
	Link* p = pd_bidir_monitors.next;
	for (; p != &pd_bidir_monitors; p = p->next) {
	  if (((giopMonitor*)p)->collection() == watcher) {
	    matched = 1;
	    break;
	  }
	}
      }
      if (!matched) {
	pd_bidir_collections.push_back(watcher);
      }
      activate();

      status = 1;
    }
  }

  if (!status) {
    ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);
    omni_tracedmutex_lock sync(*omniTransportLock);
    s->connection->decrRefCount();
  }
  return status;
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::start()
{
  omni_tracedmutex_lock sync(pd_lock);

  ensureNotInFlux();

  switch (pd_state) {
  case IDLE:
    {
      pd_state = ACTIVE;
      activate();
    }
    break;
  default:
    break;
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::stop()
{
  omni_tracedmutex_lock sync(pd_lock);

  ensureNotInFlux();

  switch (pd_state) {
  case ACTIVE:
    {
      deactivate();
    }
    break;
  default:
    break;
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::remove()
{
  omni_tracedmutex_lock sync(pd_lock);

  ensureNotInFlux();

  switch (pd_state) {
  case ACTIVE:
    {
      deactivate();
    }
  case IDLE:
    {
      pd_state = ZOMBIE;
    }
    break;
  default:
    return;
  }

  giopEndpointList::iterator i;
  i = pd_endpoints.begin();

  while (i != pd_endpoints.end()) {
    (*i)->Shutdown();
    pd_endpoints.erase(i);
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::activate()
{
  // Caller is holding pd_lock
  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock, 1);

  giopEndpointList::iterator i;
  i    = pd_endpoints.begin();

  while (i != pd_endpoints.end()) {
    giopRendezvouser* task = new giopRendezvouser(*i,this);

    if (!orbAsyncInvoker->insert(task)) {
      // Cannot start serving this endpoint.
      // Leave it in pd_endpoints.
      // Do not raise an exception. Instead, just moan about it.
      if (omniORB::trace(1)) {