Poll based implementation of the SocketCollection

// -*- Mode: C++; -*-
//                      Package   : omniORB
// SocketCollection.cc	Created on: 23 Jul 2003
//                      Author    : Serguei Kolos
//
//    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:
//	*** PROPRIETORY INTERFACE ***
// 

/*
  $Log: SocketCollection.cc,v $
  Revision 1.3  2004/02/20 17:39:21  kolos
  Apply the omniORB patch: now FD_CLOEXEC flag is set for the TCP sockets, used by the omniORB.

  Revision 1.2  2004/02/19 17:10:21  kolos
  Implement new connection multiplexor, which uses the poll function.

  Revision 1.1.2.15  2003/07/25 16:04:57  dgrisby
  vxWorks patches.

  Revision 1.1.2.14  2003/02/17 10:39:52  dgrisby
  Fix inevitable Windows problem.

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

  Revision 1.1.2.12  2003/01/28 12:17:09  dgrisby
  Bug with Select() ignoring data in buffer indications.

  Revision 1.1.2.11  2002/10/14 15:27:41  dgrisby
  Typo in fcntl error check.

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

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

  Revision 1.1.2.8  2002/03/14 12:21:49  dpg1
  Undo accidental scavenger period change, remove invalid assertion.

  Revision 1.1.2.7  2002/03/13 16:05:38  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.1.2.6  2002/02/26 14:06:45  dpg1
  Recent changes broke Windows.

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

  Revision 1.1.2.4  2001/08/24 15:56:44  sll
  Fixed code which made the wrong assumption about the semantics of
  do { ...; continue; } while(0)

  Revision 1.1.2.3  2001/08/02 13:00:53  sll
  Do not use select(0,0,0,0,&timeout), it doesn't work on win32.

  Revision 1.1.2.2  2001/08/01 15:56:07  sll
  Workaround MSVC++ bug. It generates wrong code with FD_ISSET and FD_SET
  under certain conditions.

  Revision 1.1.2.1  2001/07/31 16:16:26  sll
  New transport interface to support the monitoring of active connections.

*/

#include <omniORB4/CORBA.h>
#include <omniORB4/giopEndpoint.h>
#include <SocketCollection.h>

#if defined(__vxWorks__)
#  include "pipeDrv.h"
#  include "selectLib.h"
#  include "iostream.h"
#endif

#  include "iostream"
#  include "stdio.h"

OMNI_NAMESPACE_BEGIN(omni)

#define GDB_DEBUG

/////////////////////////////////////////////////////////////////////////
void
SocketSetTimeOut(unsigned long abs_sec,
		 unsigned long abs_nsec,struct timeval& t)
{
  unsigned long now_sec, now_nsec;
  omni_thread::get_time(&now_sec,&now_nsec);

  if ((abs_sec <= now_sec) && ((abs_sec < now_sec) || (abs_nsec < now_nsec))) {
    t.tv_sec = t.tv_usec = 0;
  }
  else {
    t.tv_sec = abs_sec - now_sec;
    if (abs_nsec >= now_nsec) {
      t.tv_usec = (abs_nsec - now_nsec) / 1000;
    }
    else {
      t.tv_usec = (1000000000 + abs_nsec - now_nsec) / 1000;
      t.tv_sec -= 1;
    }
  }
}

/////////////////////////////////////////////////////////////////////////
int
SocketSetnonblocking(SocketHandle_t sock) {
# if defined(__vxWorks__)
  int fl = TRUE;
  if (ioctl(sock, FIONBIO, (int)&fl) == ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# elif defined(__WIN32__)
  u_long v = 1;
  if (ioctlsocket(sock,FIONBIO,&v) == RC_SOCKET_ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# else
  int fl = O_NONBLOCK;
  if (fcntl(sock,F_SETFL,fl) == RC_SOCKET_ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# endif
}

/////////////////////////////////////////////////////////////////////////
int
SocketSetblocking(SocketHandle_t sock) {
# if defined(__vxWorks__)
  int fl = FALSE;
  if (ioctl(sock, FIONBIO, (int)&fl) == ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# elif defined(__WIN32__)
  u_long v = 0;
  if (ioctlsocket(sock,FIONBIO,&v) == RC_SOCKET_ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# else
  int fl = 0;
  if (fcntl(sock,F_SETFL,fl) == RC_SOCKET_ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# endif
}

/////////////////////////////////////////////////////////////////////////
// This bit is set in the events field of the pollfd 
// structure to indicate that this fd has not to be 
// taken into account while processing the poll result
#define FDCLRD		0x1000
// This bit is set in the events field of the pollfd 
// structure to indicate that this fd has to be 
// moved to the end of the pollfd array
#define FDMVD		0x2000
// This bit is set in the events field of the pollfd 
// structure to indicate that this fd has data to be read
#define FDDIB		0x4000

/////////////////////////////////////////////////////////////////////////
// Defines the maximum number of connections 
#define MAX_FD_SIZE	4096

#define SWAP( data, index, i1, i2 ) { \
    pollfd tmp = data[i1];	\
    data[i1] = data[i2];	\
    index[data[i1].fd] = i1;	\
    data[i2] = tmp;		\
    index[data[i2].fd] = i2;	\
} \


/////////////////////////////////////////////////////////////////////////
SocketHandleSet::SocketHandleSet() 
  : pd_length_full( 0 ),
    pd_length_now( 0 ),
    pd_used_by_poll_n( 0 ),
    pd_to_be_removed_n( 0 ),
    pd_to_be_moved_n( 0 ),
    pd_dib_n( 0 )
{
    pd_data = new pollfd[MAX_FD_SIZE];
    pd_index = new short[MAX_FD_SIZE];
    memset( pd_data, 0, sizeof( pollfd ) * MAX_FD_SIZE );
    memset( pd_index, -1, sizeof( short ) * MAX_FD_SIZE );
}

SocketHandleSet::~SocketHandleSet()
{
    delete[] pd_data;
    delete[] pd_index;
}

/////////////////////////////////////////////////////////////////////////
void
SocketHandleSet::Add( SocketHandle_t fd, CORBA::Boolean now, CORBA::Boolean data_in_buffer )
{
    short i = pd_index[fd];
	    
    if ( i != -1 ) {
	// fd is already in the set
	
	CORBA::Boolean was_removed = pd_data[i].events & FDCLRD;
	
	// fd was marked to be removed while we were in the poll
	// we will set it again
	if (was_removed) {
	    pd_to_be_removed_n--;
	    pd_data[i].events &= ~FDCLRD;
	}
	
	if (data_in_buffer && !(pd_data[i].events & FDDIB)) {
	    pd_data[i].events |= FDDIB;
	    pd_dib_n++;
	}
	
	if (now || data_in_buffer) {
	// fd has to be in the part of the set, which will 
	// be scheduled for the next poll invocation
	    if (i >= pd_length_now) {
		// it is not there,
		// so we will move it to that part
		SWAP( pd_data, pd_index, pd_length_now, i )
		pd_length_now++;
	    }
	    return;
	}
	
	// fd has to be in the part of the set, which will NOT
	// be scheduled for the next poll invocation
	if ( was_removed && i < pd_length_now) {
	// fd is now in the part of the set, which will 
	// be scheduled for the next poll invocation
	// and it is marked as removed
	    
	    if (i < pd_used_by_poll_n) {
	    // the poll is running 
	    // we mark the fd as to be moved to another part of the fd set
		pd_data[i].events |= FDMVD;
    		pd_to_be_moved_n++;
	    }
	    else {
	    // the poll is NOT running 
	    // we move the fd to another part of the fd set
	   	if (pd_data[i].events & FDMVD) {
		    pd_data[i].events &= ~FDMVD;
		    pd_to_be_moved_n--;
		}
			
		pd_length_now--;
		if (i != pd_length_now) {
		    SWAP( pd_data, pd_index, pd_length_now, i )
		}
	    }
	}
	return;
    }
    
    // fd is NOT in the set
    // lets put it there
    if (now || data_in_buffer) {
    	// put fd to the first part of the array
	// it will be used for the next poll invocation
	if (pd_length_full != pd_length_now) {
	    pd_data[pd_length_full] = pd_data[pd_length_now];
	    pd_index[pd_data[pd_length_full].fd] = pd_length_full;
	}
    	pd_data[pd_length_now].fd = fd;
	pd_data[pd_length_now].events = POLLIN;
    	if (data_in_buffer) {
	    pd_data[pd_length_now].events |= FDDIB;
	    pd_dib_n++;
	}
    	pd_data[pd_length_now].revents = 0;
	pd_index[fd] = pd_length_now;
	pd_length_full++;
	pd_length_now++;
    }
    else {
    	// put fd to the last part of the array
	// it will not be used for the next poll invocation
    	pd_data[pd_length_full].fd = fd;
	pd_data[pd_length_full].events = POLLIN;
    	pd_data[pd_length_full].revents = 0;
	pd_index[fd] = pd_length_full;
	pd_length_full++;
    }	
}

/////////////////////////////////////////////////////////////////////////
void
SocketHandleSet::Remove( SocketHandle_t fd )
{
    short i = pd_index[fd];

    if ( i == -1 )
    	return;
    
    if (pd_data[i].events & FDDIB) {
	pd_data[i].events &= ~FDDIB;
	pd_dib_n--;
    }
    if (pd_data[i].events & FDMVD) {
	pd_data[i].events &= ~FDMVD;
	pd_to_be_moved_n--;
    }
    if (pd_data[i].events & FDCLRD) {
	pd_data[i].events &= ~FDCLRD;
	pd_to_be_removed_n--;
    }
    
    if ( i < pd_used_by_poll_n ) {
	// this fd is used by the poll function - be careful
	// don't modify array, which is now used by poll
	// just mark the fd by setting the FDCLRD bit in the events field
	// it will be removed later in the SocketCollection::Select
	pd_data[i].events |= FDCLRD;
	pd_to_be_removed_n++;
    }
    else {
	// fd is NOT used by the poll function - simply remove fd
	pd_index[fd] = -1;
		
    	pd_length_full--;
	if (i < pd_length_now) {
	   pd_length_now--;
	   if (i != pd_length_now) {
		pd_data[i] = pd_data[pd_length_now];
		pd_index[pd_data[i].fd] = i;
	   }
	   if (pd_length_now != pd_length_full) {
		pd_data[pd_length_now] = pd_data[pd_length_full];
		pd_index[pd_data[pd_length_now].fd] = pd_length_now;
	   }
	}
	else {
	   if (i != pd_length_full) {
		pd_data[i] = pd_data[pd_length_full];
		pd_index[pd_data[i].fd] = i;
	   }
	}
    }
}

/////////////////////////////////////////////////////////////////////////
void
SocketHandleSet::Move( SocketHandle_t fd )
{
    int i = pd_index[fd];
    
    if (!(pd_data[i].events & FDMVD))
    	return; 
    
    // socket has been marked as moved when the poll was running
    // now move it to the end of the array
    pd_data[i].events &= ~FDMVD;
			
    pd_to_be_moved_n--;
    pd_length_now--;
    if (i != pd_length_now) {
	SWAP( pd_data, pd_index, pd_length_now, i )
    }
}

/////////////////////////////////////////////////////////////////////////
CORBA::Boolean
SocketHandleSet::isSet( SocketHandle_t fd, CORBA::Boolean data_in_buffer )
{
    int i = pd_index[fd];
    
    if (      ( i != -1 )
	  &&  ( data_in_buffer == 0 || (pd_data[i].events & FDDIB) ) 
	  && !( pd_data[i].events & FDCLRD )
	  && !( pd_data[i].events & FDMVD ) )
	return 1;
    else
	return 0;
}

/////////////////////////////////////////////////////////////////////////
int
SocketSetCloseOnExec(SocketHandle_t sock) {
# if defined(__vxWorks__)
  // Not supported on vxWorks
  return 0;
# elif defined(__WIN32__)
  SetHandleInformation((HANDLE)sock, HANDLE_FLAG_INHERIT, 0);
  return 0;
# else
  int fl = FD_CLOEXEC;
  if (fcntl(sock,F_SETFD,fl) == RC_SOCKET_ERROR) {
    return RC_INVALID_SOCKET;
  }
  return 0;
# endif
}

/////////////////////////////////////////////////////////////////////////
unsigned long SocketCollection::scan_interval_sec  = 0;
unsigned long SocketCollection::scan_interval_nsec = 50*1000*1000;
CORBA::ULong  SocketCollection::hashsize           = 103;

/////////////////////////////////////////////////////////////////////////
SocketCollection::SocketCollection() :
  pd_poll_cond(&pd_fdset_lock),
  pd_abs_sec(0), pd_abs_nsec(0),
  pd_pipe_read(-1), pd_pipe_write(-1), pd_pipe_full(0),
  pd_refcount(1)
{

#ifdef UnixArchitecture
#  ifdef __vxWorks__
    if (pipeDrv() == OK) {
      if (pipeDevCreate("/pipe/SocketCollection",10,sizeof(int)) == OK) {
	pd_pipe_read = pd_pipe_write = open("/pipe/SocketCollection",
					    O_RDWR,0);
      }
    }
    if (pd_pipe_read <= 0) {
      omniORB::logs(5, "Unable to create pipe for SocketCollection.");
    }
#  else
    int filedes[2];
    int r = pipe(filedes);
    if (r != -1) {
      pd_pipe_read  = filedes[0];
      pd_pipe_write = filedes[1];
    }
    else {
      omniORB::logs(5, "Unable to create pipe for SocketCollection.");
    }
#  endif
#endif

  if (pd_pipe_read > 0) {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    pd_fdset.Add( pd_pipe_read, 1, 0 );
  }

  pd_hash_table = new SocketLink* [hashsize];
  for (CORBA::ULong i=0; i < hashsize; i++)
    pd_hash_table[i] = 0;
}


/////////////////////////////////////////////////////////////////////////
SocketCollection::~SocketCollection()
{
  pd_refcount = -1;
  delete [] pd_hash_table;

#ifdef UnixArchitecture
#  ifdef __vxWorks__
  // *** How do we clean up on vxWorks?
#  else
  close(pd_pipe_read);
  close(pd_pipe_write);
#  endif
#endif
}


/////////////////////////////////////////////////////////////////////////
void
SocketCollection::setSelectable(SocketHandle_t sock, 
				CORBA::Boolean now,
				CORBA::Boolean data_in_buffer,
				CORBA::Boolean hold_lock) {

  ASSERT_OMNI_TRACEDMUTEX_HELD(pd_fdset_lock, hold_lock);

  if (!hold_lock) pd_fdset_lock.lock();

  pd_fdset.Add( sock, now, data_in_buffer );
  
  if (now || data_in_buffer) {
    // Wake up the thread blocked in select() if we can.
    if (pd_pipe_write > 0) {
#ifdef UnixArchitecture
      if (!pd_pipe_full) {
	char data = '\0';
	pd_pipe_full = 1;
	write(pd_pipe_write, &data, 1);
      }
#endif
    }
    else {
      pd_poll_cond.signal();
    }
  }
  if (!hold_lock) pd_fdset_lock.unlock();
}

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

  omni_tracedmutex_lock sync(pd_fdset_lock);
  pd_fdset.Remove( sock );
}

#ifdef GDB_DEBUG

static
int
do_poll(pollfd * r, unsigned int l, int t) {
  return poll(r,l,t);
}

#endif

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

  struct timeval timeout;
  unsigned int	 total;
  pollfd *	 rfds;

 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.enterPoll(total);
    pd_fdset.Reschedule();
  }
  else {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    rfds = pd_fdset.enterPoll(total);
  }

  int nready;

  if (total != 0) {
#ifndef GDB_DEBUG
    nready = poll(rfds,total,timeout.tv_sec*1000+(timeout.tv_usec/1000));
#else
    nready = do_poll(rfds,total,timeout.tv_sec*1000+(timeout.tv_usec/1000));
#endif
  }
  else {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    pd_poll_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) {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    pd_fdset.exitPoll();
    if (ERRNO == RC_EBADF) {
      omniORB::logs(20, "poll() returned EBADF, retrying");
      goto again;
    }
    else if (ERRNO != RC_EINTR) {
      return 0;
    }
    else {
      return 1;
    }
  }

  if (total != 0) {
    omni_tracedmutex_lock sync(pd_fdset_lock);
    pd_fdset.exitPoll();

    while (pd_fdset.needsProcessing() || nready) {
	if (rfds->revents) {
	    nready--;
	}
		
	if (rfds->events & FDCLRD) {
	    pd_fdset.Remove(rfds->fd);
	    continue;	
	}
	
	if (rfds->events & FDMVD) {
	    rfds->revents = 0;
	    pd_fdset.Move(rfds->fd);
	    continue;	
	}
	
	if (rfds->revents || (rfds->events & FDDIB)) {
	    if (rfds->fd == pd_pipe_read) {
#ifdef UnixArchitecture
		char data;
		read(pd_pipe_read, &data, 1);
		pd_pipe_full = 0;
#endif
            }
	    else {
		int fd = rfds->fd;
		pd_fdset.Remove(fd);

		if (!notifyReadable(fd)) return 0;
	      
		continue;
	    }
	}
	rfds++;
    }
  }  
  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 (!pd_fdset.isSet(sock,0))
      return 0;

    // If data in buffer is set, do callback straight away.
    if (pd_fdset.isSet(sock,1)) {
      pd_fdset.Remove(sock);
      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;
  
  pollfd rfds;

  do {
    rfds.fd = sock;
    rfds.events = POLLIN;
#ifndef GDB_DEBUG
    int nready = poll(&rfds,1,timeout.tv_sec*1000+(timeout.tv_usec/1000));
#else
    int nready = do_poll(&rfds,1,timeout.tv_sec*1000+(timeout.tv_usec/1000));
#endif

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

    // Reach here if nready >= 0

    if (rfds.revents) {
      omni_tracedmutex_lock sync(pd_fdset_lock);

      // Are we still interested?
      if (pd_fdset.isSet(sock,0)) {
	pd_fdset.Remove(sock);
	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)

// -*- Mode: C++; -*-
//                            	Package   : omniORB
// SocketCollection.h		Created on: 23 Jul 2003
//				Author    : Serguei Kolos
//
//    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:
//	*** PROPRIETORY INTERFACE ***
//

/*
*/

#ifndef __SOCKETCOLLECTION_H__
#define __SOCKETCOLLECTION_H__

////////////////////////////////////////////////////////////////////////
//  Platform feature selection

#define SOCKNAME_SIZE_T OMNI_SOCKNAME_SIZE_T

#define USE_NONBLOCKING_CONNECT

#if defined(__linux__)
#   define USE_POLL
#endif

#if defined(__sunos__)
#   define USE_POLL
#endif

#if defined(__hpux__)
#   if __OSVERSION__ >= 11
#       define USE_POLL
#   endif
#   define USE_FAKE_INTERRUPTABLE_RECV
#endif

#if defined(__freebsd__)
#  define USE_POLL
#endif

#if defined(__WIN32__)
#   define USE_FAKE_INTERRUPTABLE_RECV
#endif

#if defined(__irix__)
#   define USE_POLL
#endif

////////////////////////////////////////////////////////////////////////


////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
//             win32 API
//
#if defined(__WIN32__)

#  include <sys/types.h>
#  include <libcWrapper.h>

#  define RC_INADDR_NONE     INADDR_NONE
#  define RC_INVALID_SOCKET  INVALID_SOCKET
#  define RC_SOCKET_ERROR    SOCKET_ERROR
#  define INETSOCKET         PF_INET
#  define CLOSESOCKET(sock)  closesocket(sock)
#  define SHUTDOWNSOCKET(sock) ::shutdown(sock,2)
#  define ERRNO              ::WSAGetLastError()
#  define EINPROGRESS        WSAEWOULDBLOCK
#  define RC_EINTR           WSAEINTR
#  define RC_EBADF           WSAENOTSOCK
#  define NEED_SOCKET_SHUTDOWN_FLAG 1

OMNI_NAMESPACE_BEGIN(omni)

typedef SOCKET SocketHandle_t;

OMNI_NAMESPACE_END(omni)

#else

////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
//             unix(ish)
//
#  if defined(__vxWorks__)
#    include <sockLib.h>
#    include <hostLib.h>
#    include <ioLib.h>
#    include <netinet/tcp.h>
#  else
#    include <sys/time.h>
#  endif
#  include <sys/socket.h>
#  include <netinet/in.h>
#  include <netinet/tcp.h>
#  include <arpa/inet.h>
#  include <unistd.h>
#  include <sys/types.h>
#  include <errno.h>
#  include <libcWrapper.h>

#  if defined(USE_POLL)
#    include <poll.h>
#  endif

#  if !defined(__VMS)
#    include <fcntl.h>
#  endif

#  if defined (__uw7__)
#    ifdef shutdown
#      undef shutdown
#    endif
#  endif

#  if defined(__VMS) && defined(USE_tcpSocketVaxRoutines)
#    include "tcpSocketVaxRoutines.h"
#    undef accept
#    undef recv
#    undef send
#    define accept(a,b,c) tcpSocketVaxAccept(a,b,c)
#    define recv(a,b,c,d) tcpSocketVaxRecv(a,b,c,d)
#    define send(a,b,c,d) tcpSocketVaxSend(a,b,c,d)
#  endif

#  ifdef __rtems__
extern "C" int select (int,fd_set*,fd_set*,fd_set*,struct timeval *);
#  endif

#  define RC_INADDR_NONE     ((CORBA::ULong)-1)
#  define RC_INVALID_SOCKET  (-1)
#  define RC_SOCKET_ERROR    (-1)
#  define INETSOCKET         AF_INET
#  define CLOSESOCKET(sock)  close(sock)

#  if defined(__sunos__) && defined(__sparc__) && __OSVERSION__ >= 5
#    define SHUTDOWNSOCKET(sock)  ::shutdown(sock,2)
#  elif defined(__osf1__) && defined(__alpha__)
#    define SHUTDOWNSOCKET(sock)  ::shutdown(sock,2)
#  else
     // XXX none of the above, calling shutdown() may not have the
     // desired effect.
#    define SHUTDOWNSOCKET(sock)  ::shutdown(sock,2)
#  endif

#  define ERRNO              errno
#  define RC_EINTR           EINTR
#  define RC_EBADF           EBADF

OMNI_NAMESPACE_BEGIN(omni)

typedef int    SocketHandle_t;

OMNI_NAMESPACE_END(omni)

#endif

#if defined(NEED_GETHOSTNAME_PROTOTYPE)
extern "C" int gethostname(char *name, int namelen);
#endif

OMNI_NAMESPACE_BEGIN(omni)

// This class is a replacement for the two fd sets used
// by the select-based SocketCollection implementation
// It holds an array of pollfd structures, which consits
// from two parts:
// 1. elements [0..length_now-1] will be used 
//    for the next poll invocation
// 2. elements [length_now..length_full-1] will be passed 
//    to poll only if the Reschedule method has been called
class SocketHandleSet
{
public:
  
  SocketHandleSet();
  ~SocketHandleSet();
	
  // Adds socket to the array of pollfd structures 
  // If now == 1, this socket is set to the first part of the 
  //   array and will be used for the next poll invocation
  // If now == 0 the socket will be stored in the last part and will
  //   be passed to poll only if the Reschedule method is called
  // If data_in_buffer == 1 the FDDTNBFFR bit will be set in the
  //   events field of the corresponding pollfd structure
  void Add( SocketHandle_t fd, CORBA::Boolean now, CORBA::Boolean data_in_buffer );
  
  // Removes socket from the array of pollfd structures 
  void Remove( SocketHandle_t fd );

  // Moves socket from the current position to the end of array of pollfd structures 
  void Move( SocketHandle_t fd );

  // Checks if socket exist in the array of pollfd structures 
  // If data_in_buffer == 1, function returns 1 only if socket has the FDDTNBFFR
  //   bit set in the events field of the corresponding pollfd structure
  // If data_in_buffer == 0, function returns 1 if socket is in 
  //   the array of pollfd structures
  CORBA::Boolean isSet( SocketHandle_t fd, CORBA::Boolean data_in_buffer );
	  
  // Must be called before entering the poll
  inline pollfd * enterPoll( unsigned int & length ) {
    length = pd_length_now;
    pd_used_by_poll_n = pd_length_now;
    return pd_data;
  }

  // Must be called after exiting the poll
  inline void exitPoll( ) {
    pd_used_by_poll_n = 0;
  }
  
  // Schedule all the sockets for the next poll invocation
  inline void Reschedule( ) {
    pd_length_now = pd_length_full;
  }
  
  // Returns 1 if there are sockets in the array, which were 
  // requested to be moved or removed, or if there are 
  // unprocessed sockets marked as having data in buffer
  inline CORBA::Boolean needsProcessing( ) {
    return (    pd_to_be_removed_n
	     || pd_to_be_moved_n
	     || pd_dib_n );
  }
	
private:	
  pollfd *		pd_data;
  short	*		pd_index;	// takes more memory, but allows to find
  					// sockets int the data array very efficiently
  unsigned short	pd_length_full;
  unsigned short	pd_length_now;
  unsigned short	pd_used_by_poll_n;
  unsigned short	pd_to_be_removed_n;
  unsigned short	pd_to_be_moved_n;
  unsigned short	pd_dib_n;
};


class SocketCollection;

extern void SocketSetTimeOut(unsigned long abs_sec,
			     unsigned long abs_nsec,struct timeval& t);

extern int SocketSetnonblocking(SocketHandle_t sock);

extern int SocketSetblocking(SocketHandle_t sock);

extern int SocketSetCloseOnExec(SocketHandle_t sock);

class SocketLink {

public:
  SocketLink(SocketHandle_t s)
    : pd_socket(s),
#ifdef NEED_SOCKET_SHUTDOWN_FLAG
      pd_shutdown(0),
#endif
      pd_next(0) {}

  ~SocketLink() {}

  friend class SocketCollection;

protected:
  SocketHandle_t pd_socket;

#ifdef NEED_SOCKET_SHUTDOWN_FLAG
  // select() on Windows does not return an error after the socket has
  // shutdown, so we have to store an extra flag here.
  CORBA::Boolean pd_shutdown;
#endif

private:
  SocketLink*    pd_next;
};

class SocketCollection {
public:

  SocketCollection();

protected:
  virtual ~SocketCollection();

  virtual CORBA::Boolean notifyReadable(SocketHandle_t) = 0;
  // Callback used by Select(). This method is called while holding
  // pd_fdset_lock.

public:
  void setSelectable(SocketHandle_t sock, CORBA::Boolean now,
		     CORBA::Boolean data_in_buffer,
		     CORBA::Boolean hold_lock=0);
  // Indicates that this socket should be watched by a poll()
  // so that any new data arriving on the connection will be noted.
  // If now == 1, immediately make this socket part of the poll
  // set.
  // If data_in_buffer == 1, treat this socket as if there are
  // data available from the connection already.
  // If hold_lock == 1, pd_fdset_lock is already held.

  void clearSelectable(SocketHandle_t);
  // Indicates that this connection need not be watched any more.

  CORBA::Boolean Select();
  // Returns TRUE(1) if the Select() has successfully done a scan.
  // otherwise returns false(0) to indicate that an error has been
  // detected and this function should not be called again.
  //
  // For each of the sockets that has been marked watchable and indeed
  // has become readable, call notifyReadable() with the socket no.
  // as the argument.

  CORBA::Boolean Peek(SocketHandle_t sock);
  // Do nothing and returns immediately if the socket has not been
  // set to be watched by a previous setSelectable().
  // Otherwise, monitor the socket's status for a short time.
  // Returns TRUE(1) if the socket becomes readable.
  // otherwise returns FALSE(0).

  void incrRefCount();
  void decrRefCount();

  void addSocket(SocketLink* conn);
  // Add this socket to the collection. <conn> is associated with the
  // socket and should be added to the table hashed by the socket number.
  // Increments this collection's refcount.

  SocketLink* removeSocket(SocketHandle_t sock);
  // Remove the socket from this collection. Return the socket which has
  // been removed. Return 0 if the socket is not found.
  // Decrements this collection's refcount if a socket is removed.

  SocketLink* findSocket(SocketHandle_t sock,
			 CORBA::Boolean hold_lock=0);
  // Returns the connection that is associated with this socket.
  // Return 0 if this socket cannot be found in the hash table.
  // if hold_lock == 1, the caller has already got the lock on pd_fdset_lock.
  // (use purely by member functions.)

  static unsigned long scan_interval_sec;
  static unsigned long scan_interval_nsec;

  static CORBA::ULong  hashsize;

private:
  SocketHandleSet      pd_fdset;
  omni_tracedmutex     pd_fdset_lock;
  omni_tracedcondition pd_poll_cond; // timedwait on if nothing to select
  unsigned long        pd_abs_sec;
  unsigned long        pd_abs_nsec;
  int                  pd_pipe_read;
  int                  pd_pipe_write;
  CORBA::Boolean       pd_pipe_full;
  int                  pd_refcount;

protected:
  SocketLink**         pd_hash_table;

};

OMNI_NAMESPACE_END(omni)

#endif // __SOCKETCOLLECTION_H__