giopstrand.cc

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

                                // when the giopServer call this function.
      }
    }
  }
  if (omniORB::trace(20) && connection) {
    omniORB::logger log;
    log << (isClient() ? "Client" : "Server")
	<< " close connection" 
	<< (isClient() ? " to " : " from ")
	<< (const char*)peeraddr << "\n";
  }
  pd_state = DYING;     // satisfy the invariant in the dtor.
  delete this;
}

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopStrand::deletePending()
{
  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,1);
  return pd_safelyDeleted;
}


////////////////////////////////////////////////////////////////////////
GIOP_S*
giopStrand::acquireServer(giopWorker* w)
{
  // Theory of operation:
  //
  // One or more threads may serve the same strand. However, only
  // one thread works on the same GIOP_S instance. This invariant
  // is enforced by this function.
  //
  // There may be one or more GIOP_S instances linked to this
  // strand. They can be in one of the following states:
  //
  //  1. UnUsed*
  //        - the instance has not been claimed by any thread
  //  2. InputFullyBuffered*
  //        - the instance has not been claimed by
  //          any thread and a complete message has
  //          been received.
  //  3. InputPartiallyBuffered**
  //        - a message has been partially received. The thread
  //          that is currently holding the read lock on this
  //          strand may append additional data to this instance.
  //          If no thread is holding the read lock, this instance
  //          can be claimed by the calling thread.
  //  4. Other states***
  //        - the instance is currently in used by another thread.
  //
  //  Note:
  //    *   the instance can be claimed by the calling thread
  //    **  the instance cannot be claimed by the calling thread unless
  //        no other thread is holding the read lock.
  //    *** the instance cannot be claimed by the calling thread
  //
  // This function also acquires the Read lock on the strand if the returned
  // giopStream instance is not InputFullyBuffered. (If it is
  // InputFullyBuffered, there is no need to read from the strand, hence
  // no need for a Read lock.


  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,0);

  omni_tracedmutex_lock sync(*omniTransportLock);

  if (deletePending()) {
    // Check if safeDelete() has been called on the strand and has returned
    // true. If this is is the case, we should not proceed or else the
    // invarant
    //
    //    giopStreamList::is_empty(clients) &&
    //	  giopStreamList::is_empty(servers) &&
    //	  giopStream::noLockWaiting(this)
    //     
    // would be violated.
    //
    // Notice that giopServer (worker->server()) may schedule any number of
    // threads to serve this strand and each of these threads may call this
    // function at any time before or after safeDelete() is called and
    // returned true.  We use this check to stop all these threads right
    // here or else the invarant will be violated.
    return 0;
  }

 again:
  // Scan the list to identify the 1st occurrance of an instance in
  // one of these states: UnUsed, InputFullyBuffered, InputPartiallyBuffered.

  GIOP_S* up = 0; // 1st GIOP_S in unused state
  GIOP_S* fp = 0; // 1st GIOP_S in InputFullyBuffered state
  GIOP_S* pp = 0; // 1st GIOP_S in InputPartiallyBuffered state;

  GIOP_S* sp;

  giopStreamList* p = servers.next;
  while (p != &servers) {
    sp = (GIOP_S*)p;
    switch (sp->state()) {
    case IOP_S::UnUsed:
      if (!up) up = sp;
      break;
    case IOP_S::InputFullyBuffered:
      if (!fp) fp = sp;
      break;
    case IOP_S::InputPartiallyBuffered:
      if (!pp) pp = sp;
      break;
    default:
      break;
    }
    p = p->next;
  }
  if (fp) {
    // This is good, no need to acquire a Read Lock on the strand because
    // the whole request message is already in buffer.
    sp = fp;
  }
  else if (giopStream::rdLockNonBlocking(this)) {
    // Now we have got the Read Lock, this is *the* thread to read from
    // the strand until we release the lock.
    // Choose a GIOP_S instance in the following order:
    // 1. InputPartiallyBuffered
    // 2. UnUsed
    // 3. None of the above
    if (pp) {
      sp = pp;
    }
    else if (up) {
      sp = up;
      sp->impl(0);
    }
    else {
      sp = new GIOP_S(this);
      sp->giopStreamList::insert(servers);
    }
    sp->markRdLock();
  }
  else {
    // Another thread is already reading from the strand, we let it does
    // the work and let it wake us up if there is some work to do.
    giopStream::sleepOnRdLock(this);
    goto again;
  }

  sp->state(IOP_S::Idle);
  if (!sp->impl()) {
    giopStreamImpl *impl = giopStreamImpl::maxVersion();
    sp->impl(impl);
    if (version.major == 0)
	version = impl->version();
  }
  // the codeset convertors are filled in by the codeset interceptor
  // before a request is unmarshalled.
  sp->TCS_C(0);
  sp->TCS_W(0);
  sp->worker(w);
  sp->setDeadline(0,0);
  return sp;
}

////////////////////////////////////////////////////////////////////////
void
giopStrand::releaseServer(IOP_S* iop_s)
{
  omni_tracedmutex_lock sync(*omniTransportLock);

  GIOP_S* giop_s = (GIOP_S*) iop_s;

  giop_s->rdUnLock();
  giop_s->wrUnLock();
  giop_s->giopStreamList::remove();

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

  if ( state() == giopStrand::DYING ) {
    remove = 1;
    restart_idle = 0;

    // We have to go through the GIOP_S list and delete any ones
    // that are in the Unused and InputFullyBuffered states.
    // We can also delete the ones in InputPartiallyBuffered state
    // if no one is holding the Read lock. See the "theory of operation"
    // in acquireServer() for more info.
    CORBA::Boolean remove_partial = !giopStream::RdLockIsHeld(this);
    giopStreamList* p = servers.next;
    while (p != &servers) {
      GIOP_S* sp = (GIOP_S*)p;
      switch (sp->state()) {
      case IOP_S::InputPartiallyBuffered:
	if (! remove_partial) break;
	// else falls through
      case IOP_S::UnUsed:
      case IOP_S::InputFullyBuffered:
	p = p->next;
	sp->giopStreamList::remove();
	delete sp;
	continue;
      default:
	break;
      }
      p = p->next;
    }

    safeDelete();
  }
  else if ( !giopStreamList::is_empty(servers) ) {
    remove = 1;
    giopStreamList* p = servers.next;
    while (p != &servers) {
      GIOP_S* sp = (GIOP_S*)p;
      if (sp->state() > IOP_S::WaitForRequestHeader) {
	restart_idle = 0;
      }
      p = p->next;
    }
  }
  else {
    giop_s->state(IOP_S::UnUsed);
    giop_s->giopStreamList::insert(servers);
  }

  if (remove) {
    if (giop_s->state() != IOP_S::WaitingForReply)
      delete giop_s;
    else
      restart_idle = 0;
  }

  if (restart_idle && !biDir) {
    CORBA::Boolean success = startIdleCounter();
    OMNIORB_ASSERT(success);
  }
}

////////////////////////////////////////////////////////////////////////
CORBA::ULong
giopStrand::newSeqNumber() {

  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,1);
  seqNumber += 2;
  return seqNumber;
}

////////////////////////////////////////////////////////////////////////
StrandList giopStrand::active;
StrandList giopStrand::active_timedout;
StrandList giopStrand::passive;
// Throughout the lifetime of a strand, it is a member of one and only one
// of the lists:
//   active           - the ORB uses this connection in the role of a client
//                      it is 'active' in the sense that the connection was
//                      initiated by this ORB
//   active_timedout  - the connection was previously active and has been
//                      idled for some time. It will be deleted soon.
//   passive          - the ORB uses this connection in the role of a server
//                      it is 'passive' because the connection was initiated
//                      by the remote party.
//

CORBA::ULong giopStrand::idleIncomingBeats = 36;
CORBA::ULong giopStrand::idleOutgoingBeats = 24;

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopStrand::startIdleCounter() {
  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,1);

  if (idlebeats >= 0) {
    // The idle counter is already active or has already expired.
    return 0;
  }
  if (isClient()) {
    idlebeats = (idleOutgoingBeats) ? (CORBA::Long)idleOutgoingBeats : -1;
  }
  else {
    idlebeats = (idleIncomingBeats) ? (CORBA::Long)idleIncomingBeats : -1;
  }
  return 1;
}

////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopStrand::stopIdleCounter() {
  ASSERT_OMNI_TRACEDMUTEX_HELD(*omniTransportLock,1);

  if (idlebeats == 0) {
    // The idle counter has already expired.
    return 0;
  }
  idlebeats = -1;
  return 1;
}


////////////////////////////////////////////////////////////////////////
void
Scavenger::removeIdle(StrandList& src,StrandList& dest,
		      CORBA::Boolean skip_bidir)
{
  StrandList* p = src.next;
  while (p != &src) {
    giopStrand* s = (giopStrand*)p;

    if ( s->idlebeats >= 0 ) {

      if (omniORB::trace(30)) {
	omniORB::logger log;
	log << "Scavenger reduce idle count for strand "
	    << (void*)s << " to " << (s->idlebeats - 1) << "\n";
      }
      if (--(s->idlebeats) <= 0) {
	p = p->next;
	s->StrandList::remove();
	s->RopeLink::remove();
	s->StrandList::insert(dest);
	continue;
      }
    }

    p = p->next;
  }
}

////////////////////////////////////////////////////////////////////////
void
Scavenger::execute()
{
  omniORB::logs(25, "Scavenger task execute.");

  unsigned long abs_sec,abs_nsec;
  omni_thread::get_time(&abs_sec,&abs_nsec);

  while (1) {
    {
      omni_tracedmutex_lock sync(*mutex);
      if ( shutdown || !orbParameters::scanGranularity ) {
	goto died;
      }
      omni_thread::get_time(&abs_sec,&abs_nsec,orbParameters::scanGranularity);
      cond->timedwait(abs_sec,abs_nsec);
    }

    if (omniORB::trace(30)) {
      omniORB::logger log;
      log << "Scan for idle connections ("
	  << abs_sec << "," << abs_nsec << ")\n";
    }

    StrandList client_shutdown_list;
    StrandList server_shutdown_list;
    {
      // Collect all the strands that should be removed
      // We want to minimise the time we hold the omniTransportLock.