giopserver.cc

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

      if (conn->pd_has_dedicated_thread) {
	giopWorker* task = new giopWorker(cs->strand, this, 0);
	if (!orbAsyncInvoker->insert(task)) {
	  // Cannot start serving this new connection.
	  if (omniORB::trace(1)) {
	    omniORB::logger log;
	    log << "Cannot create a worker for this endpoint: "
		<< conn->myaddress()
		<< " from "
		<< conn->peeraddress()
		<< "\n";
	  }
	  delete task;
	  {
	    omni_tracedmutex_lock sync(*omniTransportLock);
	    cs->strand->safeDelete();
	  }
	  csRemove(conn);
	  pd_lock.unlock();
	  delete cs;
	  pd_lock.lock();

	  throw outOfResource();
	}
	task->insert(cs->workers);
	conn->pd_n_workers++;
	pd_n_dedicated_workers++;
      }
      else {
	if (!conn->isSelectable()) {
	  if (omniORB::trace(20)) {
	    omniORB::logger log;
	    log << "Connection from " << conn->peeraddress()
		<< " is not selectable. Closing it.\n";
	  }
	  {
	    omni_tracedmutex_lock sync(*omniTransportLock);
	    cs->strand->safeDelete();
	  }
	  csRemove(conn);
	  pd_lock.unlock();
	  delete cs;
	  pd_lock.lock();
	  throw outOfResource();
	}
	pd_lock.unlock();
	conn->setSelectable(1);
	pd_lock.lock();
      }
      break;
    }
  default:
    if (omniORB::trace(25)) {
      omniORB::logger l;
      l << "giopServer terminate connection from "
	<< conn->peeraddress() << ".\n";
    }
    throw Terminate();
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::notifyRzDone(giopRendezvouser* r, CORBA::Boolean exit_on_error)
{
  omni_tracedmutex_lock sync(pd_lock);

  if (!exit_on_error) {
    OMNIORB_ASSERT(pd_state == INFLUX);
    // For the moment, we do not instantiate giopRendezvouser to do
    // single shot.
    // Therefore, this function will *NEVER* be called until:
    //  1. We have called deactivate()
    //  2. giopRendezvouser have encountered a non-recoverable error. In which
    //     case exit_on_error == 1.
  }

  giopEndpoint* ept = r->endpoint();

  if (omniORB::trace(25)) {
    omniORB::logger l;
    l << "giopRendezvouser for endpoint " << ept->address() << " exit.\n";
  }

  r->remove();

  delete r;

  if (exit_on_error) {
    if (omniORB::trace(1)) {
      omniORB::logger log;
      log << "Unrecoverable error for this endpoint: ";
      log << ept->address();
      log << ", it will no longer be serviced.\n";
    }
    ept->Shutdown();
  }
  else {
    pd_endpoints.push_back(ept);
    // A new giopRendezvouser will be instantiated for this endpoint next
    // time activate() is called.
  }

  if (pd_state == INFLUX) {
    if (Link::is_empty(pd_rendezvousers)) {
      pd_cond.broadcast();
    }
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::notifyRzReadable(giopConnection* conn,
			     CORBA::Boolean force_create)
{
  // Theory of operation: read the state diagrams at the end of this file.

  omni_tracedmutex_lock sync(pd_lock);

  switch (pd_state) {
  case ACTIVE:
  case INFLUX:
    {
      if (conn->pd_dying) return;

      if (!force_create &&
	  conn->pd_n_workers >= (int)orbParameters::maxServerThreadPerConnection) {
	conn->pd_has_hit_n_workers_limit = 1;
	return;
      }

      connectionState* cs = csLocate(conn);
      if (!cs) return;

      giopWorker* task = new giopWorker(cs->strand,this,1);
      if (!orbAsyncInvoker->insert(task)) {
	// Cannot start serving this new connection.
	// Should never happen
	OMNIORB_ASSERT(0);
      }
      task->insert(cs->workers);
      conn->pd_n_workers++;
      pd_n_temporary_workers++;
      break;
    }
  default:
    break;
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::notifyCallFullyBuffered(giopConnection* conn)
{
  notifyRzReadable(conn,1);
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::removeConnectionAndWorker(giopWorker* w)
{
  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock, 0);

  connectionState* cs;
  CORBA::Boolean   cs_removed = 0;

  {
    omni_tracedmutex_lock sync(pd_lock);

    giopConnection* conn = w->strand()->connection;

    conn->pd_dying = 1; // From now on, the giopServer will not create
                        // any more workers to serve this connection.

    cs = csLocate(conn);

    // We remove the lock on pd_lock before calling the connection's
    // clearSelectable(). This is necessary so that a simultaneous
    // callback from the Rendezvouser thread will have a chance to
    // look at the connectionState table.
    pd_lock.unlock();

    conn->clearSelectable();

    // Once we reach here, it is certain that the rendezvouser thread
    // would not take any interest in this connection anymore. It
    // is therefore safe to delete this record.
    pd_lock.lock();

    int workers;

    if (w->singleshot())
      workers = --pd_n_temporary_workers;
    else
      workers = --pd_n_dedicated_workers;

    w->remove();
    delete w;

    conn->pd_n_workers--;

    if (Link::is_empty(cs->workers)) {
      csRemove(conn);
      cs_removed = 1;
    }

    if (pd_state == INFLUX) {
      if (workers == 0) {
	pd_cond.broadcast();
      }
    }
  }
  // Must not hold pd_lock when deleting cs, since the deletion may
  // cause a call to SocketCollection::removeSocket(), which needs to
  // lock the fdset lock. The fdset lock comes before pd_lock in the
  // partial order.
  if (cs_removed)
    delete cs;
}

////////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopServer::notifyWkDone(giopWorker* w, CORBA::Boolean exit_on_error)
{
  // Theory of operation: read the state diagrams at the end of this file.

  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock,0);

  if (exit_on_error) {
    removeConnectionAndWorker(w);
    return 0;
  }

  giopConnection* conn = w->strand()->connection;

  if (conn->pd_has_dedicated_thread) {
    // This connection is managed with the thread-per-connection policy
    if (!w->singleshot()) {
      // This is the dedicated thread
      conn->clearSelectable();
      omni_tracedmutex_lock sync(pd_lock);
      conn->pd_dedicated_thread_in_upcall = 0;
      conn->pd_has_hit_n_workers_limit = 0;
      return 1;
    }
    else {
      // This is a temporary worker thread
      omni_tracedmutex_lock sync(pd_lock);
      if (conn->pd_has_hit_n_workers_limit) {
	// Previously we have hit the limit on the no. of worker threads,
	// There is definitely a request pending. We re-cycle this
	// worker to deal with it *UNLESS* the dedicated thread is already
	// doing so.
	conn->pd_has_hit_n_workers_limit = 0;
	if (conn->pd_dedicated_thread_in_upcall ) {
	  return 1;
	}
      }
      if (conn->pd_n_workers == 1 && conn->pd_dying) {
	// Connection is dying. Go round again so this thread spots
	// the condition.
	omniORB::logs(25, "Last worker sees connection is dying.");
	return 1;
      }
      w->remove();
      delete w;
      conn->pd_n_workers--;
      pd_n_temporary_workers--;
      if (pd_state == INFLUX && pd_n_temporary_workers == 0)
	pd_cond.broadcast();
      return 0;
    }
  }
  else {
    // This connection is managed with the thread-pool policy.

    OMNIORB_ASSERT(w->singleshot() == 1); // Never called by a dedicated thread
    CORBA::Boolean select_and_return = 0;
    {
      omni_tracedmutex_lock sync(pd_lock);

      if (conn->pd_has_hit_n_workers_limit) {
	// Previously we have hit the limit on the no. of worker threads,
	// There is definitely a request pending. We re-cycle this
	// worker to deal with it.
	conn->pd_has_hit_n_workers_limit = 0;
	return 1;
      }

      // If there are other workers for this connection, or there are
      // too many temporary workers, let this worker finish.

      if (conn->pd_n_workers > orbParameters::threadPoolWatchConnection ||
	  pd_n_temporary_workers > orbParameters::maxServerThreadPoolSize) {

	select_and_return = 1;
      }
    }

    if (!select_and_return) {
      // Call Peek(). This thread will be used for a short time to
      // monitor the connection. If the connection is available for
      // reading, the callback function peekCallBack is called. We can
      // probably afford to call Peek() here because this thread is
      // otherwise idle.

      CORBA::Boolean readable = 0;
      conn->Peek(peekCallBack,(void*)&readable);
      if (readable) {
	// There is data to be read. Tell the worker to go around again.
	return 1;
      }
    }

    // If connection is selectable, it's selectable now
    if (!conn->pd_dying)
      conn->setSelectable(2);

    // Worker is no longer needed.
    {
      omni_tracedmutex_lock sync(pd_lock);

      w->remove();
      delete w;
      conn->pd_n_workers--;
      pd_n_temporary_workers--;

      if (pd_state == INFLUX) {
	omniORB::logs(25, "Temporary worker finishing.");
	if (pd_n_temporary_workers == 0)
	  pd_cond.broadcast();
      }
    }
    return 0;
  }
#ifndef __DECCXX
  // Never reach here (and the hp/compaq/dec compiler is smart enough
  // to figure that out).
  OMNIORB_ASSERT(0);
  return 0;
#endif
}

////////////////////////////////////////////////////////////////////////////
// Callback function used by giopConnection::Peek(). Called when the
// connection indeed has something to read immediately.
void
giopServer::peekCallBack(void* cookie, giopConnection* conn) {
  CORBA::Boolean* readable = (CORBA::Boolean*)cookie;
  *readable = 1;
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::notifyWkPreUpCall(giopWorker* w, CORBA::Boolean data_in_buffer) {

  // Theory of operation: read the state diagrams at the end of this file.

  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock,0);

  giopConnection* conn = w->strand()->connection;

  if (conn->pd_has_dedicated_thread) {
    // This connection is managed with the thread-per-connection policy
    if (!w->singleshot()) {
      // This is the dedicated thread.
      // setSelectable.
      {
	omni_tracedmutex_lock sync(pd_lock);
	conn->pd_dedicated_thread_in_upcall = 1;
      }
      if (orbParameters::maxServerThreadPerConnection > 1) {
	// If only one thread per connection is allowed, there is no
	// need to setSelectable, since we won't be able to act on any
	// interleaved calls that arrive.
	conn->setSelectable(orbParameters::connectionWatchImmediate,
			    data_in_buffer);
      }
    }
    else {
      // This is a temporary worker thread
      // setSelectable only if the dedicated thread is in upcall.
      CORBA::Boolean n;
      {
	omni_tracedmutex_lock sync(pd_lock);
	n = conn->pd_dedicated_thread_in_upcall;
      }
      if (n) {
	conn->setSelectable(orbParameters::connectionWatchImmediate,
			    data_in_buffer);
      }
    }
  }
  else {
    // This connection is managed with the thread-pool policy
    conn->setSelectable(orbParameters::connectionWatchImmediate,
			data_in_buffer);
  }
}

////////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopServer::notifySwitchToBiDirectional(giopConnection* conn)
{
  return 1;
  // Note: we could override the threading policy to dictate that all
  //       bidirectional connection will be served by a dedicated thread
  //       on the server side. 
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::notifyMrDone(giopMonitor* m, CORBA::Boolean exit_on_error)
{
  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_lock,0);
  omni_tracedmutex_lock sync(pd_lock);

  if (!exit_on_error && !m->collection()->isEmpty()) {
    // We may have seen a race condition in which the Monitor is about
    // to return when another connection has been added to be monitored.
    // We should not remove the monitor in this case.
    if (orbAsyncInvoker->insert(m)) {
      return;
    }
    // Otherwise, we let the following deal with it.
  }
  m->remove();
  delete m;
  if (pd_state == INFLUX) {
    if (Link::is_empty(pd_bidir_monitors)) {
      pd_cond.broadcast();
    }
  }
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::Link::insert(giopServer::Link& head)
{
  next = head.prev->next;
  head.prev->next = this;
  prev = head.prev;
  head.prev = this;
}

////////////////////////////////////////////////////////////////////////////
void
giopServer::Link::remove()
{
  prev->next = next;
  next->prev = prev;
  next = prev = this;
}

////////////////////////////////////////////////////////////////////////////
CORBA::Boolean
giopServer::Link::is_empty(giopServer::Link& head)
{
  return (head.next == &head);
}

/////////////////////////////////////////////////////////////////////////////
giopServer*&
giopServer::singleton() {
  static giopServer* singleton_ = 0;
  if (!singleton_) {
    singleton_ = new giopServer();
  }
  return singleton_;