/*
 * $Id: select.c,v 1.2 1999/01/26 16:29:11 labovit Exp $
 */

#include <stdarg.h>
#include <mrt.h>
#include <select.h>
/*
#ifdef __linux__
#include <sys/resource.h>
#endif
*/


Select_Struct *SELECT_MASTER;

#ifdef HAVE_LIBPTHREAD
static void 
mrt_select_loop ()
{
    SELECT_MASTER->self = pthread_self ();
    init_mrt_thread_signals ();

    while (1)
	mrt_select ();
}
#endif /* HAVE_LIBPTHREAD */

/* #define MRT_DEBUG 1 */

/* init_select
 */
int 
init_select (trace_t * tr)
{
    int fd[2];
    ll_value_t lv;

    assert (SELECT_MASTER == NULL);

    SELECT_MASTER = New (Select_Struct);
    SELECT_MASTER->ll_descriptors = LL_Create (LL_DestroyFunction, free, 0);
    SELECT_MASTER->trace = tr;
    SELECT_MASTER->ll_close_fds = 
	LL_Create (LL_Intrusive, True, 
		   LL_PointersOffset, LL_Offset (&lv, &lv.ptrs),
		   LL_DestroyFunction, free, 0);

    pthread_mutex_init (&SELECT_MASTER->mutex_lock, NULL);

    pipe (fd);
    SELECT_MASTER->interrupt_fd[1] = fd[1];
    SELECT_MASTER->interrupt_fd[0] = fd[0];
    FD_SET (fd[0], &SELECT_MASTER->fdvar_read);
    /* setting nonblock is safer */
    fcntl (fd[1], F_SETFL, O_NONBLOCK);
    return (1);
}


#ifdef HAVE_LIBPTHREAD
int
start_select ()
{
    if (mrt_thread_create ("Select Thread", NULL,
	 (thread_fn_t) mrt_select_loop, NULL) == NULL) {
	return (-1);
    }
    return (0);
}
#endif /* HAVE_LIBPTHREAD */


#ifdef notdef
void 
__clear_select_interrupt ()
{
    char c;

    printf ("\nhere in interrupt!\n");
    read (SELECT_MASTER->interrupt_fd[0], &c, 1);
    select_enable_fd (SELECT_MASTER->interrupt_fd[0]);

    /*exit (0); */

}
#endif


/* 
 * Add a file descriptor to the list of fds which we are blocking on in a 
 * select call.
 */
int 
select_add_fd_event (char *name, int fd, u_long type_mask, int on,
		     schedule_t *schedule, event_fn_t call_fn, int narg, ...)
{
    Descriptor_Struct *dsp;
    va_list ap;
    event_t *event;
    int i;

    va_start (ap, narg);
    event = New_Event (narg);
    /* I expect that description will not be freed */
    event->description = name;
    for (i = 0;i < narg; i++)
        event->args[i] = va_arg (ap, void *);
    event->call_fn = call_fn;
    va_end (ap);

    pthread_mutex_lock (&SELECT_MASTER->mutex_lock);
    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	if (dsp->fd != fd)
	    continue;
	if (type_mask && (dsp->type_mask & type_mask) == 0)
	    continue;

	if (dsp->type_mask != type_mask) {
	    trace (TR_ERROR, SELECT_MASTER->trace,
		   "SELECT: %s on fd %d mask 0x%x -> 0x%x "
		   "(mask can't be changed)\n", dsp->name,
		   fd, dsp->type_mask, type_mask);
    	    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
	    return (-1);
	}
	if (name)
            dsp->name = name;
    	dsp->event = event;
    	dsp->schedule = schedule;
	trace (TR_WARN, SELECT_MASTER->trace,
	       "SELECT: %s on fd %d mask 0x%x %s (update)\n",
	       name, fd, type_mask, on? "on": "off");
	break;
    }

    if (dsp == NULL) {
        dsp = New (Descriptor_Struct);
        dsp->name = (name)? name: "noname";
        dsp->fd = fd;
        dsp->type_mask = type_mask;
        dsp->event = event;
        dsp->schedule = schedule;
        LL_Add (SELECT_MASTER->ll_descriptors, dsp);
	trace (TR_TRACE, SELECT_MASTER->trace,
	       "SELECT: %s on fd %d mask 0x%x %s (update)\n",
	       dsp->name, fd, type_mask, on? "on": "off");
    }

    if (on) {
        if (type_mask & SELECT_READ)
	    FD_SET (fd, &SELECT_MASTER->fdvar_read);
        if (type_mask & SELECT_WRITE)
	    FD_SET (fd, &SELECT_MASTER->fdvar_write);
        if (type_mask & SELECT_EXCEPTION)
	    FD_SET (fd, &SELECT_MASTER->fdvar_except);
        write (SELECT_MASTER->interrupt_fd[1], "", 1);
    }

    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
    return (1);
}


int 
select_add_fd (int fd, u_long type_mask, void (*call_fn) (), void *arg)
{
    return (select_add_fd_event (NULL, fd, type_mask, 1, NULL, call_fn, 
				 1, arg));
}


int 
select_delete_fd_mask (int fd, u_long mask, int close_at_select)
{
    Descriptor_Struct *dsp, *prev;
    int found = 0;

    pthread_mutex_lock (&SELECT_MASTER->mutex_lock);
    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	if (dsp->fd != fd)
	    continue;
	if (mask && (dsp->type_mask & mask) == 0)
	    continue;

	if (dsp->type_mask & SELECT_READ)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_read);
	if (dsp->type_mask & SELECT_WRITE)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_write);
	if (dsp->type_mask & SELECT_EXCEPTION)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_except);
	found++;
        trace (TR_TRACE, SELECT_MASTER->trace, 
	       "SELECT: %s cleared fd %d mask 0x%x\n", dsp->name, fd, mask);

	if (close_at_select) {
	    ll_value_t *vp;
	    /* close will be done after select resumes */
	    /* if we close here, the same fd may be assigned for open
	       that follows imediately.  Then, select may get confused 
	       to treat an event on the previous fd for the new one */
	    LL_Iterate (SELECT_MASTER->ll_close_fds, vp) {
		if (vp->value == fd)
		    break;
	    }
	    if (vp == NULL) {
		vp = New (ll_value_t);
		vp->value = fd;
		LL_Add (SELECT_MASTER->ll_close_fds, vp);
        	trace (TR_TRACE, SELECT_MASTER->trace, 
	       	       "SELECT: %s close fd %d scheduled\n", dsp->name, fd);
	    }
	}
	prev = LL_GetPrev (SELECT_MASTER->ll_descriptors, dsp);
        Deref_Event (dsp->event);
	LL_Remove (SELECT_MASTER->ll_descriptors, dsp);
	dsp = prev;
    }

    if (found) {
        write (SELECT_MASTER->interrupt_fd[1], "", 1);
        pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
    	return (1);
    }

    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
#ifdef MRT_DEBUG
    /* this could happen when called from bgp */
    trace (TR_ERROR, SELECT_MASTER->trace, "SELECT: no fd %d (delete)\n", fd);
#endif /* MRT_DEBUG */
    return (-1);
}


/*
 * Delete everything associated with this socket descriptor!
 */
int 
select_delete_fd (int fd)
{
    return (select_delete_fd_mask (fd, 0, 1));
}


/* select_delete_fd2
 * without closing fd
 * Delete everything associated with this socket descriptor!
 */
int 
select_delete_fd2 (int fd)
{
    return (select_delete_fd_mask (fd, 0, 0));
}


/* wait on select and dispatch an event associated with the fd */
int 
mrt_select2 (int ms)
{
    fd_set fdvar_read, fdvar_write, fdvar_except;
    Descriptor_Struct *dsp;
    struct timeval *timeout;
    LINKED_LIST *ll = NULL;
    int nfds;

#ifdef HAVE_LIBPTHREAD
    timeout = NULL;
#ifdef MRT_DEBUG
    trace (TR_TRACE, SELECT_MASTER->trace, 
	   "SELECT: blocking in select (thread)\n");
#endif /* MRT_DEBUG */
#else
    struct timeval tv;

    tv.tv_sec = ms / 1000;
    tv.tv_usec = (ms % 1000) * 1000;
    timeout = &tv;
#ifdef MRT_DEBUG
    trace (TR_TRACE, SELECT_MASTER->trace, 
	   "SELECT: blocking in select (timeout %d ms)\n", ms);
#endif /* MRT_DEBUG */
#endif /* HAVE_LIBPTHREAD */
    fdvar_read = SELECT_MASTER->fdvar_read;
    fdvar_write = SELECT_MASTER->fdvar_write;
    fdvar_except = SELECT_MASTER->fdvar_except;


    if ((nfds = select (FD_SETSIZE, &fdvar_read,
		&fdvar_write, &fdvar_except, timeout)) < 0) {
	if (errno == EINTR)
	    return (0);
        trace (TR_ERROR, SELECT_MASTER->trace, "SELECT: error (%s)\n", 
	       strerror (errno));
	return (-1);
    }
    if (nfds <= 0) {
	/* timeout */
	return (0);
    }

    pthread_mutex_lock (&SELECT_MASTER->mutex_lock);

#ifdef MRT_DEBUG
    trace (TR_TRACE, SELECT_MASTER->trace, 
	   "SELECT: processing select (active %d fds)\n", nfds);
#endif /* MRT_DEBUG */

    /* process all pending closes */
    if (LL_GetCount (SELECT_MASTER->ll_close_fds) > 0) {
	ll_value_t *vp;
	LL_Iterate (SELECT_MASTER->ll_close_fds, vp) {
	    close (vp->value);
            trace (TR_TRACE, SELECT_MASTER->trace, 
	       	   "SELECT: closed fd %d\n", vp->value);
	}
	LL_Clear (SELECT_MASTER->ll_close_fds);
    }

    /* okay, read hint to ourselves that a new socket has been added */
    if (FD_ISSET (SELECT_MASTER->interrupt_fd[0], &fdvar_read)) {
	char c;
	read (SELECT_MASTER->interrupt_fd[0], &c, 1);
#ifdef MRT_DEBUG
    trace (TR_TRACE, SELECT_MASTER->trace, 
	   "SELECT: processing select (hint)\n");
#endif /* MRT_DEBUG */
	if (--nfds <= 0) {
    	    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
	    return (0);
	}
    }

    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	if (((dsp->type_mask & SELECT_READ) && 
		FD_ISSET (dsp->fd, &fdvar_read)) ||
	     ((dsp->type_mask & SELECT_WRITE) && 
		FD_ISSET (dsp->fd, &fdvar_write)) ||
	     ((dsp->type_mask & SELECT_EXCEPTION) && 
		FD_ISSET (dsp->fd, &fdvar_except))) {

/* Clearing the bit is required because sometimes this routine is called 
   again before reading data waiting. */
	    if (dsp->type_mask & SELECT_READ)
		FD_CLR (dsp->fd, &SELECT_MASTER->fdvar_read);
	    if (dsp->type_mask & SELECT_WRITE)
		FD_CLR (dsp->fd, &SELECT_MASTER->fdvar_write);
	    if (dsp->type_mask & SELECT_EXCEPTION)
		FD_CLR (dsp->fd, &SELECT_MASTER->fdvar_except);

            /* schedule it right now while locking */
            if (dsp->schedule) {
                schedule_event3 (dsp->schedule, Ref_Event (dsp->event));
            }
	    else {
		/* event is sometimes refered outside of locking
		   so that reference counter is introduced.
		   deletion may happen unexpectedly. */
	        LL_Add2 (ll, Ref_Event (dsp->event));
	    }
	}
    }

    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);

    /* this way should be fare, avoiding to server the same fd
       that frequently becomes ready */
    if (ll) {
	event_t *event;
	LL_Iterate (ll, event) {
	    /* this derefs the event */
	    schedule_event_dispatch (event);
	}
	LL_Destroy (ll);
    }
    else {
        /* timeout or no fds registered (may happen on closing) */
    }
    return (0);
}


int mrt_select () {
  return (mrt_select2 (1000));
}


/* given a mask (SELECT_READ| SELECT_WRITE_ | SELECT_EXCEPTION),
 * disable a socket that matches the fd AND the mask if mask != 0
 */
int 
select_disable_fd_mask (int fd, u_long mask)
{
    Descriptor_Struct *dsp;
    int changed = 0;

    pthread_mutex_lock (&SELECT_MASTER->mutex_lock);

    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	if (dsp->fd != fd)
	    continue;
	if (mask && (dsp->type_mask & mask) == 0)
	    continue;

	if (dsp->type_mask & SELECT_READ)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_read);
	if (dsp->type_mask & SELECT_WRITE)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_write);
	if (dsp->type_mask & SELECT_EXCEPTION)
	    FD_CLR (fd, &SELECT_MASTER->fdvar_except);

	changed++;
       	trace (TR_TRACE, SELECT_MASTER->trace, 
	       "SELECT: %s disabled on %d mask 0x%x\n", dsp->name, dsp->fd, 
	       dsp->type_mask);
    }

    if (changed) {
	write (SELECT_MASTER->interrupt_fd[1], "", 1);
        pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
	return (1);
    }
    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
#ifdef MRT_DEBUG
    trace (FATAL, SELECT_MASTER->trace, "SELECT: no fd %d (disable)\n", fd);
#endif	/* MRT_DEBUG */
    return (-1);
}


/* there is no way to disable for write only */

int 
select_disable_fd (int fd)
{
    return (select_disable_fd_mask (fd, 0));
}


/*
 * the mrt_select master must be notified that a socket is ready for selection 
 * again after the select call routine has fired
 */
int 
select_enable_fd_mask (int fd, u_long mask)
{
    Descriptor_Struct *dsp;
    int changed = 0;

    assert (fd >= 0);

    pthread_mutex_lock (&SELECT_MASTER->mutex_lock);
    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	if (dsp->fd != fd)
	    continue;
	if (mask && (dsp->type_mask & mask) == 0)
	    continue;

	if (dsp->type_mask & SELECT_READ)
	    FD_SET (fd, &SELECT_MASTER->fdvar_read);
	if (dsp->type_mask & SELECT_WRITE)
	    FD_SET (fd, &SELECT_MASTER->fdvar_write);
	if (dsp->type_mask & SELECT_EXCEPTION)
	    FD_SET (fd, &SELECT_MASTER->fdvar_except);

	changed++;
       	trace (TR_TRACE, SELECT_MASTER->trace, 
	       "SELECT: %s enabled on %d mask 0x%x\n", dsp->name, dsp->fd, 
	       dsp->type_mask);
    }

    if (changed) {
	write (SELECT_MASTER->interrupt_fd[1], "", 1);
        pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
	return (1);
    }
    /* this fd does not exist! */
    pthread_mutex_unlock (&SELECT_MASTER->mutex_lock);
#ifdef MRT_DEBUG
    trace (FATAL, SELECT_MASTER->trace, "SELECT: no fd %d (enable)\n", fd);
#endif
    return (-1);
}


int 
select_enable_fd (int fd)
{
    return (select_enable_fd_mask (fd, 0));
}


#ifdef notdef
/* called on exit */
int
select_shutdown ()
{
    Descriptor_Struct *dsp;

    LL_Iterate (SELECT_MASTER->ll_descriptors, dsp) {
	trace (TR_INFO, SELECT_MASTER->trace,  
	       "SELECT: Shutting down %d mask 0x%x\n", dsp->fd, dsp->type_mask);
	close (dsp->fd);
    }

    return (1);
}
#endif


/* leave it disabled */
int 
select_add_fd_off (int fd, u_long type_mask, void (*call_fn) (), void *arg)
{
    return (select_add_fd_event (NULL, fd, type_mask, 0, NULL, call_fn, 
				 1, arg));
}
