socketcollection.cc

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

  }
}

/////////////////////////////////////////////////////////////////////////
void
SocketCollection::clearSelectable(SocketHandle_t sock)
{

#ifdef SELECTABLE_FD_LIMIT
  if (sock >= SELECTABLE_FD_LIMIT)
    return;
#endif

  omni_tracedmutex_lock sync(pd_fdset_lock);

  if (FD_ISSET(sock,&pd_fdset_1)) {
    pd_n_fdset_1--;
    FD_CLR(sock,&pd_fdset_1);
  }
  if (FD_ISSET(sock,&pd_fdset_2)) {
    pd_n_fdset_2--;
    FD_CLR(sock,&pd_fdset_2);
  }
  if (FD_ISSET(sock,&pd_fdset_dib)) {
    pd_n_fdset_dib--;
    FD_CLR(sock,&pd_fdset_dib);
  }
}

/////////////////////////////////////////////////////////////////////////
CORBA::Boolean
SocketCollection::isSelectable(SocketHandle_t sock)
{
#ifdef SELECTABLE_FD_LIMIT
  return sock < SELECTABLE_FD_LIMIT;
#else
  return 1;
#endif
}


#ifdef GDB_DEBUG

static
int
do_select(int maxfd,fd_set* r, fd_set* w, fd_set* e, struct timeval* t) {
  return select(maxfd,r,w,e,t);
}

#endif

/////////////////////////////////////////////////////////////////////////
CORBA::Boolean
SocketCollection::Select() {

  struct timeval timeout;
  fd_set         rfds;
  int            total;

 again:

  // (pd_abs_sec,pd_abs_nsec) define the absolute time when we switch fdset
  SocketSetTimeOut(pd_abs_sec,pd_abs_nsec,timeout);

  if (timeout.tv_sec == 0 && timeout.tv_usec == 0) {

    omni_thread::get_time(&pd_abs_sec,&pd_abs_nsec,
			  scan_interval_sec,scan_interval_nsec);
    timeout.tv_sec  = scan_interval_sec;
    timeout.tv_usec = scan_interval_nsec / 1000;

    omni_tracedmutex_lock sync(pd_fdset_lock);
    rfds  = pd_fdset_2;
    total = pd_n_fdset_2;
    pd_fdset_2 = pd_fdset_1;
    pd_n_fdset_2 = pd_n_fdset_1;
  }
  else {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    rfds  = pd_fdset_2;
    total = pd_n_fdset_2;
  }

  int maxfd = 0;
  int fd = 0;

#ifndef __WIN32__
  // Win32 ignores the first argument to select()
  while (total) {
    if (FD_ISSET(fd,&rfds)) {
      maxfd = fd;
      total--;
    }
    fd++;
  }
#else
  fd = total;
#endif

  int nready;

  if (fd != 0) {
#ifndef GDB_DEBUG
    nready = select(maxfd+1,&rfds,0,0,&timeout);
#else
    nready = do_select(maxfd+1,&rfds,0,0,&timeout);
#endif
  }
  else {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    pd_select_cond.timedwait(pd_abs_sec,pd_abs_nsec);
    // The condition variable should be poked so we are woken up
    // immediately when there is something to monitor.  We cannot use
    // select(0,0,0,0,&timeout) because win32 doesn't like it.
    nready = 0; // simulate a timeout
  }

  if (nready == RC_SOCKET_ERROR) {
    if (ERRNO == RC_EBADF) {
      omniORB::logs(20, "select() returned EBADF, retrying");
      goto again;
    }
    else if (ERRNO != RC_EINTR) {
      return 0;
    }
    else {
      return 1;
    }
  }

  {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    SocketHandle_t fd = 0;

    if (nready > 0) {
      // Process the result from the select.
      while (nready) {
	if (FD_ISSET(fd,&rfds)) {
	  nready--;

          if (fd == pd_pipe_read) {
#ifdef UnixArchitecture
            char data;
            read(pd_pipe_read, &data, 1);
            pd_pipe_full = 0;
#endif
          }
	  else {
	    if (FD_ISSET(fd,&pd_fdset_2)) {
	      pd_n_fdset_2--;
	      FD_CLR(fd,&pd_fdset_2);
	      if (FD_ISSET(fd,&pd_fdset_1)) {
		pd_n_fdset_1--;
		FD_CLR(fd,&pd_fdset_1);
	      }
	      if (FD_ISSET(fd,&pd_fdset_dib)) {
		pd_n_fdset_dib--;
		FD_CLR(fd,&pd_fdset_dib);
	      }
	      if (!notifyReadable(fd)) return 0;
	    }
	  }
	}
	fd++;
      }
    }

    // Process pd_fdset_dib. Those sockets with their bit set have
    // already got data in buffer. We do a call back for these sockets if
    // their entries in pd_fdset_2 is also set.
    fd = 0;
    nready = pd_n_fdset_dib;
    while (nready) {
      if (FD_ISSET(fd,&pd_fdset_dib)) {
	if (FD_ISSET(fd,&pd_fdset_2)) {
	  pd_n_fdset_2--;
	  FD_CLR(fd,&pd_fdset_2);
	  pd_n_fdset_dib--;
	  FD_CLR(fd,&pd_fdset_dib);
	  if (FD_ISSET(fd,&pd_fdset_1)) {
	    pd_n_fdset_1--;
	    FD_CLR(fd,&pd_fdset_1);
	  }
	  if (!notifyReadable(fd)) return 0;
	}
	nready--;
      }
      fd++;
    }
  }
  return 1;

}

/////////////////////////////////////////////////////////////////////////
CORBA::Boolean
SocketCollection::Peek(SocketHandle_t sock) {

  {
    omni_tracedmutex_lock sync(pd_fdset_lock);
   
    // Do nothing if this socket is not set to be monitored.
    if (!FD_ISSET(sock,&pd_fdset_1))
      return 0;

    // If data in buffer is set, do callback straight away.
    if (FD_ISSET(sock,&pd_fdset_dib)) {
      if (FD_ISSET(sock,&pd_fdset_1)) {
	pd_n_fdset_1--;
	FD_CLR(sock,&pd_fdset_1);
      }
      if (FD_ISSET(sock,&pd_fdset_2)) {
	pd_n_fdset_2--;
	FD_CLR(sock,&pd_fdset_2);
      }
      pd_n_fdset_dib--;
      FD_CLR(sock,&pd_fdset_dib);
      return 1;
    }
  }

  struct timeval timeout;
  // select on the socket for half the time of scan_interval, if no request
  // arrives in this interval, we just let AcceptAndMonitor take care
  // of it.
  timeout.tv_sec  = scan_interval_sec / 2;
  timeout.tv_usec = scan_interval_nsec / 1000 / 2;
  if (scan_interval_sec % 2) timeout.tv_usec += 500000;
  fd_set rfds;

  do {
    FD_ZERO(&rfds);
    FD_SET(sock,&rfds);
#ifndef GDB_DEBUG
    int nready = select(sock+1,&rfds,0,0,&timeout);
#else
    int nready = do_select(sock+1,&rfds,0,0,&timeout);
#endif

    if (nready == RC_SOCKET_ERROR) {
      if (ERRNO != RC_EINTR) {
	break;
      }
      else {
	continue;
      }
    }

    // Reach here if nready >= 0

    if (FD_ISSET(sock,&rfds)) {
      omni_tracedmutex_lock sync(pd_fdset_lock);

      // Are we still interested?
      if (FD_ISSET(sock,&pd_fdset_1)) {
	if (FD_ISSET(sock,&pd_fdset_2)) {
	  pd_n_fdset_2--;
	  FD_CLR(sock,&pd_fdset_2);
	}
	pd_n_fdset_1--;
	FD_CLR(sock,&pd_fdset_1);
	if (FD_ISSET(sock,&pd_fdset_dib)) {
	  pd_n_fdset_dib--;
	  FD_CLR(sock,&pd_fdset_dib);
	}
	return 1;
      }
    }
    break;

  } while(1);

  return 0;
}


/////////////////////////////////////////////////////////////////////////
void
SocketCollection::incrRefCount()
{
  omni_tracedmutex_lock sync(pd_fdset_lock);
  OMNIORB_ASSERT(pd_refcount > 0);
  pd_refcount++;
}

/////////////////////////////////////////////////////////////////////////
void
SocketCollection::decrRefCount()
{
  int refcount;
  {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    OMNIORB_ASSERT(pd_refcount > 0);
    refcount = --pd_refcount;
  }
  if (refcount == 0) delete this;
}

/////////////////////////////////////////////////////////////////////////
void
SocketCollection::addSocket(SocketLink* conn)
{
  omni_tracedmutex_lock sync(pd_fdset_lock);
  SocketLink** head = &(pd_hash_table[conn->pd_socket % hashsize]);
  conn->pd_next = *head;
  *head = conn;
  OMNIORB_ASSERT(pd_refcount > 0);
  pd_refcount++;
}

/////////////////////////////////////////////////////////////////////////
SocketLink*
SocketCollection::removeSocket(SocketHandle_t sock)
{
  int refcount  = 0; // Initialise to stop over-enthusiastic compiler warnings
  SocketLink* l = 0;
  {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    SocketLink** head = &(pd_hash_table[sock % hashsize]);
    while (*head) {
      if ((*head)->pd_socket == sock) {
	l = *head;
	*head = (*head)->pd_next;
	OMNIORB_ASSERT(pd_refcount > 0);
	refcount = --pd_refcount;
	break;
      }
      head = &((*head)->pd_next);
    }
  }
  if (l && refcount == 0) delete this;
  return l;
}

/////////////////////////////////////////////////////////////////////////
SocketLink*
SocketCollection::findSocket(SocketHandle_t sock,
			     CORBA::Boolean hold_lock) {

  if (!hold_lock) pd_fdset_lock.lock();

  SocketLink* l = 0;
  SocketLink** head = &(pd_hash_table[sock % hashsize]);
  while (*head) {
    if ((*head)->pd_socket == sock) {
      l = *head;
      break;
    }
    head = &((*head)->pd_next);
  }

  if (!hold_lock) pd_fdset_lock.unlock();

  return l;
}

OMNI_NAMESPACE_END(omni)