chan.c

早期freebsd实现

/*  $Revision: 1.30 $
**
**  I/O channel (and buffer) processing.
*/
#include "innd.h"


STATIC FDSET	RCHANmask;
STATIC FDSET	SCHANmask;
STATIC FDSET	WCHANmask;
STATIC int	SCHANcount;
STATIC int	CHANlastfd;
STATIC int	CHANlastsleepfd;
STATIC int	CHANccfd;
STATIC int	CHANtablesize;
STATIC CHANNEL	*CHANtable;
STATIC CHANNEL	*CHANcc;
STATIC CHANNEL	CHANnull = { CTfree, CSerror, -1 };


/*
**  Set a buffer's contents, ignoring anything that might have
**  been there.
*/
void
BUFFset(bp, p, length)
    register BUFFER	*bp;
    register char	*p;
    register int	length;
{
    register char	*dest;

    if ((bp->Left = length) != 0) {
	/* Need more space? */
	if (bp->Size < length) {
	    bp->Size = GROW_AMOUNT(length);
	    RENEW(bp->Data, char, bp->Size);
	}

	if (length > MEMCPY_THRESHOLD)
	    (void)memcpy((POINTER)bp->Data, (POINTER)p, (SIZE_T)length);
	else {
	    for (dest = bp->Data, length++; --length > 0; )
		*dest++ = *p++;
	}
    }
    bp->Used = 0;
}


/*
**  Initialize all the I/O channels.
*/
void
CHANsetup(i)
    register int	i;
{
    register CHANNEL	*cp;

    FD_ZERO(&RCHANmask);
    FD_ZERO(&SCHANmask);
    FD_ZERO(&WCHANmask);
    if (CHANtable)
	DISPOSE(CHANtable);
    CHANtablesize = i;
    CHANtable = NEW(CHANNEL, CHANtablesize);
    (void)memset((POINTER)CHANtable, 0,
	    (SIZE_T)(CHANtablesize * sizeof *CHANtable));
    CHANnull.NextLog = CHANNEL_INACTIVE_TIME;
    CHANnull.Address.s_addr = MyAddress.s_addr;
    for (cp = CHANtable; --i >= 0; cp++)
	*cp = CHANnull;
}


/*
**  Create a channel from a descriptor.
*/
CHANNEL *
CHANcreate(fd, Type, State, Reader, WriteDone)
    int			fd;
    CHANNELTYPE		Type;
    CHANNELSTATE	State;
    FUNCPTR		Reader;
    FUNCPTR		WriteDone;
{
    register CHANNEL	*cp;
    BUFFER		in;
    BUFFER		out;

    cp = &CHANtable[fd];

    /* Don't overwrite the buffers with CHANnull. */
    in = cp->In;
    if (in.Size == 0) {
	in.Size = START_BUFF_SIZE;
	in.Data = NEW(char, in.Size);
    }
    in.Used = 0;
    in.Left = in.Size;
    out = cp->Out;
    if (out.Size == 0) {
	out.Size = SMBUF;
	out.Data = NEW(char, out.Size);
    }
    out.Used = 0;
    out.Left = 0;

    /* Set up the channel's info. */
    *cp = CHANnull;
    cp->fd = fd;
    cp->Type = Type;
    cp->State = State;
    cp->Reader = Reader;
    cp->WriteDone = WriteDone;
    cp->Started = cp->LastActive = Now.time;
    cp->In = in;
    cp->Out = out;
    cp->Tracing = Tracing;

    /* Make the descriptor close-on-exec and non-blocking. */
    CloseOnExec(fd, TRUE);
#if	defined(ENOTSOCK)
    if (SetNonBlocking(fd, TRUE) < 0 && errno != ENOTSOCK)
	syslog(L_ERROR, "%s cant nonblock %d %m", LogName, fd);
#else
    if (SetNonBlocking(fd, TRUE) < 0)
	syslog(L_ERROR, "%s cant nonblock %d %m", LogName, fd);
#endif	/* defined(ENOTSOCK) */

    /* Note control channel, for efficiency. */
    if (Type == CTcontrol) {
	CHANcc = cp;
	CHANccfd = fd;
    }
    return cp;
}


/*
**  Start tracing a channel.
*/
void
CHANtracing(cp, Flag)
    register CHANNEL	*cp;
    BOOL		Flag;
{
    char		*p;

    p = CHANname(cp);
    syslog(L_NOTICE, "%s trace %s", p, Flag ? "on" : "off");
    cp->Tracing = Flag;
    if (Flag) {
	syslog(L_NOTICE, "%s trace badwrites %d blockwrites %d badreads %d",
	    p, cp->BadWrites, cp->BadReads, cp->BlockedWrites);
	syslog(L_NOTICE, "%s trace address %s lastactive %ld nextlod %ld",
	    p, inet_ntoa(cp->Address), cp->LastActive, cp->NextLog);
	if (FD_ISSET(cp->fd, &SCHANmask))
	    syslog(L_NOTICE, "%s trace sleeping %ld 0x%x",
		p, (long)cp->Waketime, cp->Waker);
	if (FD_ISSET(cp->fd, &RCHANmask))
	    syslog(L_NOTICE, "%s trace reading %d %s",
		p, cp->In.Used, MaxLength(cp->In.Data, cp->In.Data));
	if (FD_ISSET(cp->fd, &WCHANmask))
	    syslog(L_NOTICE, "%s trace writing %d %s",
		p, cp->Out.Left, MaxLength(cp->Out.Data, cp->Out.Data));
    }
}


/*
**  Close a channel.
*/
void
CHANclose(cp, name)
    register CHANNEL	*cp;
    char		*name;
{
    if (cp->Type == CTfree)
	syslog(L_ERROR, "%s internal closing free channel", name);
    else {
	if (cp->Type == CTnntp)
	    syslog(L_NOTICE,
		"%s closed seconds %ld accepted %ld refused %ld rejected %ld",
		name, (long)(Now.time - cp->Started),
		cp->Received, cp->Refused, cp->Rejected);
	else
	    syslog(L_NOTICE, "%s closed", name);
	WCHANremove(cp);
	RCHANremove(cp);
	SCHANremove(cp);
	if (cp->Argument != NULL)
	    /* Set to NULL below. */
	    DISPOSE(cp->Argument);
	if (cp->fd >= 0 && close(cp->fd) < 0)
	    syslog(L_ERROR, "%s cant close %s %m", LogName, name);
    }

    /* Mark it unused. */
    cp->Type = CTfree;
    cp->State = CSerror;
    cp->fd = -1;
    cp->Argument = NULL;

    /* Free the buffers if they got big. */
    if (cp->In.Size > BIG_BUFFER) {
	cp->In.Size = 0;
	DISPOSE(cp->In.Data);
    }
    if (cp->Out.Size > BIG_BUFFER) {
	cp->Out.Size = 0;
	DISPOSE(cp->Out.Data);
    }
}


/*
**  Return a printable name for the channel.
*/
char *
CHANname(cp)
    register CHANNEL	*cp;
{
    static char		buff[SMBUF];
    register int	i;
    register SITE	*sp;
    STRING		p;
    PID_T		pid;

    switch (cp->Type) {
    default:
	(void)sprintf(buff, "?%d(#%d@%d)?", cp->Type, cp->fd, cp - CHANtable);
	break;
    case CTany:
	(void)sprintf(buff, "any:%d", cp->fd);
	break;
    case CTfree:
	(void)sprintf(buff, "free:%d", cp->fd);
	break;
    case CTremconn:
	(void)sprintf(buff, "remconn:%d", cp->fd);
	break;
    case CTnntp:
	(void)sprintf(buff, "%s:%d",
		cp->Address.s_addr == 0 ? "localhost" : RChostname(cp),
		cp->fd);
	break;
    case CTlocalconn:
	(void)sprintf(buff, "localconn:%d", cp->fd);
	break;
    case CTcontrol:
	(void)sprintf(buff, "control:%d", cp->fd);
	break;
    case CTexploder:
    case CTfile:
    case CTprocess:
	/* Find the site that has this channel. */
	for (p = "?", i = nSites, sp = Sites, pid = 0; --i >= 0; sp++)
	    if (sp->Channel == cp) {
		p = sp->Name;
		if (cp->Type != CTfile)
		    pid = sp->pid;
		break;
	    }
	if (pid == 0)
	    (void)sprintf(buff, "%s:%d:%s",
		    MaxLength(p, p), cp->fd,
		    cp->Type == CTfile ? "file" : "proc");
	else
	    (void)sprintf(buff, "%s:%d:%s:%ld",
		    MaxLength(p, p), cp->fd,
		    cp->Type == CTfile ? "file" : "proc", (long)pid);
	break;
    }
    return buff;
}


/*
**  Return the channel for a specified descriptor.
*/
CHANNEL *
CHANfromdescriptor(fd)
    int		fd;
{
    if (fd <0 || fd > CHANtablesize)
	return NULL;
    return &CHANtable[fd];
}


/*
**  Iterate over all channels of a specified type.
*/
CHANNEL *
CHANiter(ip, Type)
    int			*ip;
    CHANNELTYPE		Type;
{
    register CHANNEL	*cp;
    register int	i;

    if ((i = *ip) >= 0 && i < CHANtablesize) {
	do {
	    cp = &CHANtable[i];
	    if (Type == CTany || cp->Type == Type) {
		*ip = ++i;
		return cp;
	    }
	} while (++i < CHANtablesize);
    }
    return NULL;
}


/*
**  Mark a channel as an active reader.
*/
void
RCHANadd(cp)
    register CHANNEL	*cp;
{
    FD_SET(cp->fd, &RCHANmask);
    if (cp->fd > CHANlastfd)
	CHANlastfd = cp->fd;

    /* Start reading at the beginning of the buffer. */
    cp->In.Used = 0;
}


/*
**  Remove a channel from the set of readers.
*/
void
RCHANremove(cp)
    register CHANNEL	*cp;
{
    if (FD_ISSET(cp->fd, &RCHANmask)) {
	FD_CLR(cp->fd, &RCHANmask);
	if (cp->fd == CHANlastfd) {
	    /* This was the highest descriptor, get a new highest. */
	    while (!FD_ISSET(CHANlastfd, &RCHANmask)
	      && !FD_ISSET(CHANlastfd, &WCHANmask)
	      && CHANlastfd > 1)
		CHANlastfd--;
	}
    }
}


/*
**  Put a channel to sleep, call a function when it wakes.
**  Note that the Argument must be NULL or allocated memory!
*/
void
SCHANadd(cp, Waketime, Event, Waker, Argument)
    register CHANNEL	*cp;
    time_t		Waketime;
    POINTER		Event;
    FUNCPTR		Waker;
    POINTER		Argument;
{
    if (!FD_ISSET(cp->fd, &SCHANmask)) {
	SCHANcount++;
	FD_SET(cp->fd, &SCHANmask);
    }
    if (cp->fd > CHANlastsleepfd)
	CHANlastsleepfd = cp->fd;
    cp->Waketime = Waketime;
    cp->Waker = Waker;
    if (cp->Argument != Argument) {
	DISPOSE(cp->Argument);
	cp->Argument = Argument;
    }
    cp->Event = Event;
}


/*
**  Take a channel off the sleep list.
*/
void
SCHANremove(cp)
    register CHANNEL	*cp;
{
    if (FD_ISSET(cp->fd, &SCHANmask)) {
	FD_CLR(cp->fd, &SCHANmask);
	SCHANcount--;
	cp->Waketime = 0;
	if (cp->fd == CHANlastsleepfd) {
	    /* This was the highest descriptor, get a new highest. */
	    while (!FD_ISSET(CHANlastsleepfd, &WCHANmask)
	      && CHANlastsleepfd > 1)
		CHANlastsleepfd--;
	}
    }
}


/*
**  Is a channel on the sleep list?
*/
BOOL
CHANsleeping(cp)
    CHANNEL	*cp;
{
    return FD_ISSET(cp->fd, &SCHANmask);
}


/*
**  Wake up channels waiting for a specific event.
*/
void
SCHANwakeup(Event)
    register POINTER	Event;
{
    register CHANNEL	*cp;
    register int	i;

    for (cp = CHANtable, i = CHANtablesize; --i >= 0; cp++)
	if (cp->Type != CTfree && cp->Event == Event && CHANsleeping(cp))
	    cp->Waketime = 0;
}


/*