cl_poll.c

linux集群服务器软件代码包

	if (debug) {
		cl_log(LOG_DEBUG
		,	"Signal %d monitors fd %d...", cl_nsig, fd);
	}

	/* Test to see if the file descriptor is good */
	if ((flags = fcntl(fd, F_GETFL)) < 0) {
		cl_perror("cl_poll_assignsig(%d) F_GETFL failure"
		,	fd);
		return -1;
	}

	/* Associate the right signal with the fd */

	if (fcntl(fd, F_SETSIG, cl_nsig) < 0) {
		cl_perror("cl_poll_assignsig(%d) F_SETSIG failure"
		,	fd);
		return -1;
	}

	/* Direct the signals to us */
	if (fcntl(fd, F_SETOWN, getpid()) < 0) {
		cl_perror("cl_poll_assignsig(%d) F_SETOWN failure", fd);
		return -1;
	}

	/* OK... Go ahead and send us signals! */

	if (fcntl(fd, F_SETFL, flags|O_ASYNC) < 0) {
		cl_perror("cl_poll_assignsig(%d) F_SETFL(O_ASYNC) failure"
		,	fd);
		return -1;
	}

	return cl_nsig;
}


/*
 *	This is a function we call as a (fake) signal handler.
 *
 *	It records events to our "monitorinfo" structure.
 *
 *	Except for the cl_log() call, it could be called in a signal
 *	context.
 */

static void
cl_poll_sigaction(int nsig, siginfo_t* info, void* v)
{
	int	fd;

	/* What do you suppose all the various si_code values mean? */

	fd = info->si_fd;
	if (debug) {
		cl_log(LOG_DEBUG
		,	"cl_poll_sigaction(nsig=%d fd=%d"
		", si_code=%d si_band=0x%lx)"
		,	nsig, fd, info->si_code
		,	(unsigned long)info->si_band);
	}

	if (fd <= 0) {
		return;
	}


	if (fd >= max_allocated || !is_monitored[fd]) {
		return;
	}

	/* We should not call logging functions in (real) signal handlers */
	if (nsig != monitorinfo[fd].nsig) {
		cl_log(LOG_ERR, "cl_poll_sigaction called with signal %d/%d\n"
		,	nsig, monitorinfo[fd].nsig);
	}

	/* Record everything as a pending event. */
	RECORDFDEVENT(fd, info->si_band);
}



/*
 *	This is called whenever a file descriptor shouldn't be
 *	monitored any more.
 */
int
cl_poll_ignore(int fd)
{
	short	nsig;
	int	flags;

	if (debug) {
		cl_log(LOG_DEBUG
		,	"cl_poll_ignore(%d)", fd);
	}
	if (fd <  0 || fd >= max_allocated) {
		errno = EINVAL;
		return -1;
	}
	if (!is_monitored[fd]) {
		return 0;
	}
	nsig = monitorinfo[fd].nsig;

	is_monitored[fd] = FALSE;
	memset(monitorinfo+fd, 0, sizeof(monitorinfo[0]));

	if ((flags = fcntl(fd, F_GETFL)) >= 0) {
		flags &= ~O_ASYNC;
		if (fcntl(fd, F_SETFL, flags) < 0) {
			return -1;
		}
	}else{
		return flags;
	}
	return 0;
}


/*
 * cl_poll: fake poll routine based on POSIX realtime signals.
 *
 * We want to emulate poll as exactly as possible, but poll has a couple
 * of problems:  scaleability, and it tends to sleep in the kernel
 * because the first argument is an argument of arbitrary size, and
 * generally requires allocating memory.
 *
 * The challenge is that poll is level-triggered, but the POSIX
 * signals (and sigtimedwait(2)) are edge triggered.  This is
 * one of the reasons why we have the cl_real_poll_fd() function
 * - to get the current "level" before we start.
 * Once we have this level we can compute something like the current
 * level
 */

int
cl_poll(struct pollfd *fds, unsigned int nfds, int timeoutms)
{
	int				nready;
	struct	timespec		ts;
	static const struct timespec	zerotime = {0L, 0L};
	const struct timespec*		itertime = &ts;
	siginfo_t			info;
	int				eventcount = 0;
	unsigned int			j;
	int				savederrno = errno;
	int				stw_errno;
	int				rc;
	longclock_t			starttime;
	longclock_t			endtime;
	const int			msfudge
	=				2* 1000/hz_longclock();
	int				mselapsed = 0;

	/* Do we have any old news to report? */
	if ((nready=cl_init_poll_sig(fds, nfds)) != 0) {
		/* Return error or old news to report */
		if (debug) {
			cl_log(LOG_DEBUG, "cl_poll: early return(%d)", nready);
		}
		return nready;
	}

	/* Nothing to report yet... */

	/* So, we'll do a sigtimedwait(2) to wait for signals 
	 * and see if we can find something to report...
	 *
	 * cl_init_poll() prepared a set of file signals to watch...
	 */

recalcandwaitagain:
	if (timeoutms >= 0) {
		ts.tv_sec = timeoutms / 1000;
		ts.tv_nsec = (((unsigned long)timeoutms) % 1000UL)*1000000UL;
	}else{
		ts.tv_sec = G_MAXLONG;
		ts.tv_nsec = 99999999UL;
	}

	/*
	 * Perform a timed wait for any of our signals...
	 *
	 * We shouldn't sleep for any call but (possibly) the first one.
	 * Subsequent calls should just pick up other events without
	 * sleeping.
	 */

	starttime = time_longclock();
	/*
	 * Wait up to the prescribed time for a signal.
	 * If we get a signal, then loop grabbing all other
	 * pending signals. Note that subsequent iterations will
	 * use &zerotime to get the minimum wait time.
	 */
	if (debug) {
		check_fd_info(fds, nfds);
		dump_fd_info(fds, nfds, timeoutms);
	}
waitagain:
	while (sigtimedwait(&SignalSet, &info, itertime) >= 0) {
		int		nsig = info.si_signo;

		/* Call signal handler to simulate signal reception */

		cl_poll_sigaction(nsig, &info, NULL);
		itertime = &zerotime;
	}
	stw_errno=errno; /* Save errno for later use */
	endtime = time_longclock();
	mselapsed = longclockto_ms(sub_longclock(endtime, starttime));

#ifdef TIME_CALLS
	if (timeoutms >= 0 && mselapsed > timeoutms + msfudge) {
		/* We slept too long... */
		cl_log(LOG_WARNING
		,	"sigtimedwait() sequence for %d ms took %d ms"
		,	timeoutms, mselapsed);
	}
#endif

	if (SigQOverflow) {
		/* OOPS!  Better recover from this! */
		/* This will use poll(2) to correct our current status */
		cl_poll_sigpoll_overflow();
	}

	/* Post observed events and count them... */
	
	for (j=0; j < nfds; ++j) {
		int	fd = fds[j].fd;
		poll_info_t*	moni = monitorinfo+fd;
		fds[j].revents = (moni->pendevents
		&	(fds[j].events|CONSTEVENTS));
		if (fds[j].revents) {
			++eventcount;
			moni->pendevents &= ~(fds[j].revents);
			/* Make POLLHUP persistent */
			if (fds[j].revents & POLLHUP) {
				moni->pendevents |= POLLHUP;
				/* Don't lose input events at EOF */
				if (fds[j].events & POLLIN) {
					cl_real_poll_fd(fds[j].fd);
				}
			}
		}
	}
	if (eventcount == 0 && stw_errno == EAGAIN && timeoutms != 0) {
		/* We probably saw an event the user didn't ask to see. */
		/* Consquently, we may have more waiting to do */
		if (timeoutms < 0) {
			/* Restore our infinite wait time */
			itertime = &ts;
			goto waitagain;
		}else if (timeoutms > 0) {
			if (mselapsed < timeoutms) {
				timeoutms -= mselapsed;
				goto recalcandwaitagain;
			}
		}
	}
	rc = (eventcount > 0 ? eventcount : (stw_errno == EAGAIN ? 0 : -1));

	if (rc >= 0) {
		errno = savederrno;
	}
	return rc;
}
/*
 * Debugging routine for printing current poll arguments, etc.
 */
static void
dump_fd_info(struct pollfd *fds, unsigned int nfds, int timeoutms)
{
	unsigned	j;

	cl_log(LOG_DEBUG, "timeout: %d milliseconds", timeoutms);
	for (j=0; j < nfds; ++j) {
		int	fd = fds[j].fd;
		poll_info_t*	moni = monitorinfo+fd;

		cl_log(LOG_DEBUG, "fd %d flags: 0%o, signal: %d, events: 0x%x"
		", revents: 0x%x, pendevents: 0x%x"
		,	fd, fcntl(fd, F_GETFL), moni->nsig
		,	fds[j].events, fds[j].revents, moni->pendevents);
	}
	for (j=SIGRTMIN; j < (unsigned)SIGRTMAX; ++j) {
		if (!sigismember(&SignalSet, j)) {
			continue;
		}
		cl_log(LOG_DEBUG, "Currently monitoring RT signal %d", j);
	}
}

/*
 * Debugging routine for auditing our file descriptors, etc.
 */
static void
check_fd_info(struct pollfd *fds, unsigned int nfds)
{
	unsigned	j;

	for (j=0; j < nfds; ++j) {
		int	fd = fds[j].fd;
		poll_info_t*	moni = monitorinfo+fd;

		if (!sigismember(&SignalSet, moni->nsig)) {
			cl_log(LOG_ERR, "SIGNAL %d not in monitored SignalSet"
			,	moni->nsig);
		}
	}
	for (j=0; j < 10; ++j) {
		int	sig;
		int	flags;
		int	pid;
		if ((flags = fcntl(j, F_GETFL)) < 0 || (flags & O_ASYNC) ==0){
			continue;
		}
		sig = fcntl(j, F_GETSIG);
		if (sig == 0) {
			cl_log(LOG_ERR, "FD %d will get SIGIO", j);
		}
		if (!sigismember(&SignalSet, sig)) {
			cl_log(LOG_ERR, "FD %d (signal %d) is not in SignalSet"
			,	j, sig);
		}
		if (sig < SIGRTMIN || sig >= SIGRTMAX) {
			cl_log(LOG_ERR, "FD %d (signal %d) is not RealTime"
			,	j, sig);
		}
		pid = fcntl(j, F_GETOWN);
		if (pid != getpid()) {
			cl_log(LOG_ERR, "FD %d (signal %d) owner is pid %d"
			,	j, sig, pid);
		}
	}
}

/* Note that the kernel signalled an event queue overflow */
static void
cl_poll_sigpoll_overflow_sigaction(int nsig, siginfo_t* info, void* v)
{
	SigQOverflow = TRUE;
}

#define	MAXQNAME	"rtsig-max"
/*
 * Called when signal queue overflow is suspected.
 * We then use poll(2) to get the current data.  It's slow, but it
 * should work quite nicely.
 */
static void
cl_poll_sigpoll_overflow(void)
{
	int	fd;
	int	limit;

	if (!SigQOverflow) {
		return;
	}
	cl_log(LOG_WARNING, "System signal queue overflow.");
	limit = cl_poll_get_sigqlimit();
	if (limit > 0) {
		cl_log(LOG_WARNING, "Increase '%s'. Current limit is %d"
		" (see sysctl(8)).", MAXQNAME, limit);
	}

	SigQOverflow = FALSE;

	for (fd = 0; fd < max_allocated; ++fd) {
		if (is_monitored[fd]) {
			cl_real_poll_fd(fd);
		}
	}
}

#define	PSK	"/proc/sys/kernel/"

/* Get current kernel signal queue limit */
/* This only works on Linux - but that's not a big problem... */
static int
cl_poll_get_sigqlimit(void)
{
	int	limit = -1;
	int	pfd;
	char	result[32];

	pfd = open(PSK MAXQNAME, O_RDONLY);
	if (pfd >= 0 && read(pfd, result, sizeof(result)) > 1) {
		limit = atoi(result);
		if (limit < 1) {
			limit = -1;
		}
	}
	if (pfd >= 0) {
		close(pfd);
	}
	return limit;
}
#endif /* HAVE_FCNTL_F_SETSIG */