gioprope.cc

来自「编译工具」· CC 代码 · 共 883 行 · 第 1/2 页

// -*- Mode: C++; -*-
//                            Package   : omniORB
// giopRope.cc                Created on: 16/01/2001
//                            Author    : Sai Lai Lo (sll)
//
//    Copyright (C) 2001 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:
//

/*
  $Log: giopRope.cc,v $
  Revision 1.1.4.28  2005/10/17 15:50:55  dgrisby
  Permit clients to scavenge bidir connections if no object references
  have been transmitted.

  Revision 1.1.4.27  2005/08/23 11:46:05  dgrisby
  Race condition in strand selection could lead to an assertion failure.

  Revision 1.1.4.26  2004/12/20 20:13:21  dgrisby
  Log when switching addresses in a rope.

  Revision 1.1.4.25  2003/09/25 13:12:20  dgrisby
  Correctly complain if some parameters are set to zero.

  Revision 1.1.4.24  2003/06/02 09:28:22  dgrisby
  Segfault in log message when throwing an exception in an interceptor.

  Revision 1.1.4.23  2003/02/17 01:20:00  dgrisby
  Avoid deadlock with bidir connection shutdown.

  Revision 1.1.4.22  2002/09/08 21:12:38  dgrisby
  Properly handle IORs with no usable profiles.

  Revision 1.1.4.21  2002/08/23 14:15:02  dgrisby
  Avoid exception with bidir when no POA.

  Revision 1.1.4.20  2002/03/18 15:13:08  dpg1
  Fix bug with old-style ORBInitRef in config file; look for
  -ORBtraceLevel arg before anything else; update Windows registry
  key. Correct error message.

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

  Revision 1.1.4.18  2001/09/10 17:47:17  sll
  startIdleCounter when the strand is definitely idle.

  Revision 1.1.4.17  2001/09/04 14:40:30  sll
  Added the boolean argument to notifyCommFailure to indicate if
  omniTransportLock is held by the caller.

  Revision 1.1.4.16  2001/09/03 17:31:52  sll
  Make sure that acquireClient honours the deadline set in the calldescriptor.

  Revision 1.1.4.15  2001/09/03 13:31:45  sll
  Removed debug trace.

  Revision 1.1.4.14  2001/09/03 13:26:35  sll
  In filterAndSortAddressList, change to use the lowest value to represent
  the highest priority.

  Revision 1.1.4.13  2001/08/31 11:57:16  sll
  Client side transport selection is now determined by the clientTransportRules.

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

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

  Revision 1.1.4.10  2001/08/06 15:50:27  sll
  In filterAndSortAddressList, make unix transport as the first choice if
  available. This is just a temporary solution until we have table driven
  transport selection.

  Revision 1.1.4.9  2001/08/03 17:41:21  sll
  System exception minor code overhaul. When a system exeception is raised,
  a meaning minor code is provided.

  Revision 1.1.4.8  2001/08/01 18:12:54  sll
  In filterAndSortAddressList, use_bidir could be left uninitialised.

  Revision 1.1.4.7  2001/07/31 16:24:23  sll
  Moved filtering and sorting of available addresses into a separate
  function. Make acquireClient, decrRefCount and notifyCommFailure virtual.

  Revision 1.1.4.6  2001/07/13 15:26:58  sll
  Use safeDelete to remove a strand.

  Revision 1.1.4.5  2001/06/13 20:13:15  sll
  Minor updates to make the ORB compiles with MSVC++.

  Revision 1.1.4.4  2001/06/11 18:01:58  sll
  Temporarily hardwared to choose ssl over tcp transport if the IOR has both.

  Revision 1.1.4.3  2001/05/31 16:18:13  dpg1
  inline string matching functions, re-ordered string matching in
  _ptrToInterface/_ptrToObjRef

  Revision 1.1.4.2  2001/05/08 17:06:53  sll
  Client side now closes the connection if it encounters any error in
  processing a call.

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

  */


#include <omniORB4/CORBA.h>
#include <omniORB4/IOP_C.h>
#include <giopRope.h>
#include <giopStream.h>
#include <giopStrand.h>
#include <giopStreamImpl.h>
#include <giopBiDir.h>
#include <GIOP_C.h>
#include <objectAdapter.h>
#include <exceptiondefs.h>
#include <omniORB4/minorCode.h>
#include <initialiser.h>
#include <orbOptions.h>
#include <orbParameters.h>
#include <transportRules.h>
#include <omniORB4/callDescriptor.h>

#include <stdlib.h>

OMNI_NAMESPACE_BEGIN(omni)

////////////////////////////////////////////////////////////////////////////
//             Configuration options                                      //
////////////////////////////////////////////////////////////////////////////
CORBA::Boolean orbParameters::oneCallPerConnection = 1;
//  1 means only one call can be in progress at any time per connection.
//
//  Valid values = 0 or 1

CORBA::ULong orbParameters::maxGIOPConnectionPerServer = 5;
//  The ORB could open more than one connections to a server
//  depending on the number of concurrent invocations to the same
//  server. This variable decide what is the maximum number of
//  connections to use per server. This variable is read only once
//  at ORB_init. If the number of concurrent invocations exceed this
//  number, the extra invocations would be blocked until the
//  the outstanding ones return.
//
//  Valid values = (n >= 1) 


///////////////////////////////////////////////////////////////////////
RopeLink giopRope::ropes;

////////////////////////////////////////////////////////////////////////
giopRope::giopRope(const giopAddressList& addrlist,
		   const omnivector<CORBA::ULong>& preferred) :
  pd_refcount(0),
  pd_address_in_use(0),
  pd_maxStrands(orbParameters::maxGIOPConnectionPerServer),
  pd_oneCallPerConnection(orbParameters::oneCallPerConnection),
  pd_nwaiting(0),
  pd_cond(omniTransportLock)
{
  {
    giopAddressList::const_iterator i, last;
    i    = addrlist.begin();
    last = addrlist.end();
    for (; i != last; i++) {
      giopAddress* a = (*i)->duplicate();
      pd_addresses.push_back(a);
    }
  }

  {
    omnivector<CORBA::ULong>::const_iterator i, last;
    i    = preferred.begin();
    last = preferred.end();
    for (; i != last; i++) {
      pd_addresses_order.push_back(*i);
    }
  }
}


////////////////////////////////////////////////////////////////////////
giopRope::giopRope(giopAddress* addr,int initialRefCount) :
  pd_refcount(initialRefCount),
  pd_address_in_use(0),
  pd_maxStrands(orbParameters::maxGIOPConnectionPerServer),
  pd_oneCallPerConnection(orbParameters::oneCallPerConnection),
  pd_nwaiting(0),
  pd_cond(omniTransportLock)
{
  pd_addresses.push_back(addr);
  pd_addresses_order.push_back(0);
}

////////////////////////////////////////////////////////////////////////
giopRope::~giopRope() {
  OMNIORB_ASSERT(pd_nwaiting == 0);
  giopAddressList::iterator i, last;
  i    = pd_addresses.begin();
  last = pd_addresses.end();
  for (; i != last; i++) {
    delete (*i);
  }
}

////////////////////////////////////////////////////////////////////////
IOP_C*
giopRope::acquireClient(const omniIOR* ior,
			const CORBA::Octet* key,
			CORBA::ULong keysize,
			omniCallDescriptor* calldesc) {

  GIOP::Version v = ior->getIORInfo()->version();
  giopStreamImpl* impl = giopStreamImpl::matchVersion(v);
  if (!impl) {
    impl = giopStreamImpl::maxVersion();
    v = impl->version();
  }

  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);

  omni_tracedmutex_lock sync(*omniTransportLock);

 again:

  unsigned int nbusy = 0;
  unsigned int ndying = 0;
  unsigned int nwrongver = 0;
  CORBA::ULong max = pd_maxStrands;  // snap the value now as it may
                                     // change by the application anytime.
  RopeLink* p = pd_strands.next;
  for (; p != &pd_strands; p = p->next) {
    giopStrand* s = (giopStrand*)p;
    switch (s->state()) {
    case giopStrand::DYING:
      {
	// Bidirectional strands do not count towards the total of
	// dying strands. This is because with a bidirectional rope,
	// the max number of strands is one. Below, if the number of
	// dying strands is > the max, we wait for the strands to die.
	// However, it is possible that we are the client keeping the
	// strand alive, leading to a deadlock. To avoid the
	// situation, we do not count dying bidir strands, allowing us
	// to create a new one, and release the one that is dying.
	if (!s->biDir)
	  ndying++;
	break;
      }
    case giopStrand::TIMEDOUT:
      {
	s->StrandList::remove();
	s->state(giopStrand::ACTIVE);
	s->StrandList::insert(giopStrand::active);
	// falls through
      }
    case giopStrand::ACTIVE:
      {
	if (s->version.major != v.major || s->version.minor != v.minor) {
	  // Wrong GIOP version. Each strand can only be used
	  // for one GIOP version.
	  // If ever we allow more than one GIOP version
	  // to use one strand, make sure the client side interceptor
	  // for codeset is updated to reflect this.
	  nwrongver++;
	}
	else {
	  GIOP_C* g;
	  if (!giopStreamList::is_empty(s->clients)) {
	    giopStreamList* gp = s->clients.next;
	    for (; gp != &s->clients; gp = gp->next) {
	      g = (GIOP_C*)gp;
	      if (g->state() == IOP_C::UnUsed) {
		g->initialise(ior,key,keysize,calldesc);
		return g;
	      }
	    }
	    nbusy++;
	  }
	  else {
	    g = new GIOP_C(this,s);
	    g->impl(s->giopImpl);
	    g->initialise(ior,key,keysize,calldesc);
	    g->giopStreamList::insert(s->clients);
	    return g;
	  }
	}
      }
    }
  }

  // Reach here if we haven't got a strand to grab a GIOP_C.
  if ((nbusy + ndying) < max) {
    // Create a new strand.
    // Notice that we can have up to
    //  pd_maxStrands * <no. of supported GIOP versions> strands created.
    //
    
    // Do a sanity check here. It could be the case that this rope has
    // no valid address to use. This can be the case if we have
    // unmarshalled an IOR which has no profiles we can support. In
    // theory it shouldn't happen because all IORs must support IIOP.
    // However, this could be a special ORB version in which the IIOP
    // transport is taken out. Notice that we do not raise an
    // exception at the time when the IOR was unmarshalled because we
    // would like to be able to receive and pass along object
    // references that we ourselves cannot talk to.
    if (pd_addresses_order.empty()) {
      OMNIORB_THROW(TRANSIENT,TRANSIENT_NoUsableProfile,CORBA::COMPLETED_NO);
    }

    giopStrand* s = new giopStrand(pd_addresses[pd_addresses_order[pd_address_in_use]]);
    s->state(giopStrand::ACTIVE);
    s->RopeLink::insert(pd_strands);
    s->StrandList::insert(giopStrand::active);
    s->version = v;
    s->giopImpl = impl;
  }
  else if (pd_oneCallPerConnection || ndying >= max) {
    // Wait for a strand to be unused.
    pd_nwaiting++;
    unsigned long deadline_secs,deadline_nanosecs;
    calldesc->getDeadline(deadline_secs,deadline_nanosecs);
    if (deadline_secs || deadline_nanosecs) {
      if (pd_cond.timedwait(deadline_secs,deadline_nanosecs) == 0) {
	pd_nwaiting--;
	OMNIORB_THROW(TRANSIENT,TRANSIENT_CallTimedout,CORBA::COMPLETED_NO);
      }
    }
    else {
      pd_cond.wait();
    }
    pd_nwaiting--;
  }
  else {
    // Pick a random non-dying strand.
    OMNIORB_ASSERT(nbusy);  // There must be a non-dying strand that can
                            // serve this GIOP version
    int n = rand() % nbusy;
    // Pick a random and non-dying strand
    RopeLink* p = pd_strands.next;
    giopStrand* q = 0;
    giopStrand* s = 0;
    while (n >=0 && p != &pd_strands) {
      s = (giopStrand*)p;
      if (s->state() == giopStrand::ACTIVE &&
	  s->version.major == v.major &&
	  s->version.minor == v.minor)
	{
	  n--;
	  if (!q) q = s;
	}
      else {
	s = 0;
      }
      p = p->next;
    }
    s = (s) ? s : q;
    // By the time we look for busy strands, it's possible that they
    // are all dying, in which case we have to start again.
    if (s) {
      GIOP_C* g = new GIOP_C(this,s);
      g->impl(s->giopImpl);
      g->initialise(ior,key,keysize,calldesc);
      g->giopStreamList::insert(s->clients);
      return g;
    }
  }
  goto again;
}

////////////////////////////////////////////////////////////////////////
void
giopRope::releaseClient(IOP_C* iop_c) {

  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);

  omni_tracedmutex_lock sync(*omniTransportLock);

  GIOP_C* giop_c = (GIOP_C*) iop_c;

  giop_c->rdUnLock();
  giop_c->wrUnLock();

  // We decide in the following what to do with this GIOP_C and the strand
  // it is attached:
  //
  // 1. If the strand is used simultaneously for multiple calls, it will have
  //    multiple GIOP_Cs attached. We only want to keep at most 1 idle GIOP_C.
  //    In other words, if this is not the last GIOP_C attached to the strand
  //    we delete it. (Actually it does no harm to delete all GIOP_C
  //    irrespectively. It will be slower to do an invocation because a
  //    new GIOP_C has to be instantiated in every call.
  //
  // 2. If the strand is in the DYING state, we obviously should delete the
  //    GIOP_C. If this is also the last GIOP_C, we delete the strand as
  //    well. If the strand is used to support bidirectional GIOP, we
  //    also check to ensure that the GIOP_S list is empty.
  //

  giopStrand* s = &((giopStrand&)(*(giopStream*)giop_c));
  giop_c->giopStreamList::remove();

  CORBA::Boolean remove = 0;
  CORBA::Boolean avail = 1;

  if (giop_c->state() != IOP_C::Idle && s->state() != giopStrand::DYING ) {
    s->state(giopStrand::DYING);
    if (omniORB::trace(30)) {
      omniORB::logger l;

      if (s->connection) {
	l << "Unexpected error encountered in talking to the server "
	  << s->connection->peeraddress()
	  << " . The connection is closed immediately.\n";
      }
      else {
	OMNIORB_ASSERT(s->address);
	l << "Unexpected error encountered before talking to the server "
	  << s->address->address()