Date: Tue, 5 Aug 1997 20:39:47 -0700 From: Frank MacLachlan <fpm@n2.net> To: Matt Eagleson <matt@deepwell.com> Cc: isp@FreeBSD.ORG, fpm@n2.net Subject: Re: INN Problem Message-ID: <19970805203947.52488@jsbach.n2.net> In-Reply-To: <3.0.3.32.19970805171509.02f92a94@deepwell.com>; from Matt Eagleson on Tue, Aug 05, 1997 at 05:15:09PM -0700 References: <3.0.3.32.19970805171509.02f92a94@deepwell.com>
next in thread | previous in thread | raw e-mail | index | archive | help
On Tue, Aug 05, 1997 at 05:15:09PM -0700, Matt Eagleson wrote:
> This entry appears in my news.err
> about every 2 minutes:
>
> Aug 5 17:01:00 news innd: overview!:21:proc:2148 cant write Resource
> temporarily unavailable
> Aug 5 17:01:00 news innd: overview!:21:proc:2148 blocked sleeping 120
>
> I know others have had this problem too.
> Doe anyone know what is causing this?
>
> Any assistance would be greatly appreciated.
>
> Regards,
> Matt
I had the same problem. In my case, the errors began when
expireover was running during an expire. Overchan blocked
waiting on locked overview files and fell behind INN.
Once more than 65536 bytes accumulated for overchan, INN
could *not* write any data to the pipe to overchan w/o blocking
(INN insists upon writing all of the data for a channel in a
single write operation and, under FreeBSD, would get an
EAGAIN error). INN then blocked sleeping for 120 seconds
on the channel and retried the write. During the sleep
interval, more data accumulated and the write operation
failed again. Any attempt to flush the channel also failed.
Eventually INN dumped the buffered data to an OVERVIEW! file.
I modified INN to retry blocked writes in smaller chunks and
reduced the sleep interval from 120 seconds to 15 seconds. I
still saw the error messages, but INN could usually write the
data to overchan within 2 or 3 sleep intervals.
I wasn't happy w/ the above strategy, so I wrote the following
program which dynamically buffers data between INN and overchan.
I haven't seen any 'cant write Resource temporarily unavailable'
errors since I started using it nor have I seen any OVERVIEW!
dump files.
--
Frank MacLachlan (fpm@n2.net)
N2 Networking, San Diego, CA
-------------------CUT HERE------------------------
/*
**
** Name:
** chan_buf
**
**
** Description:
** Provides front-end buffering for inn channel fed programs.
** It can, for example, interpolate a buffer process between inn
** and overchan. I use the following entry in my newsfeeds file
** to invoke this program and overchan:
**
** overview!\
** :*,!junk,!control*\
** :Tc,WO\
** :/news/bin/chan_buf /news/bin/overchan
**
** This results in the following situation:
**
** +------+ +----------+ +----------+
** | innd |--------------->| chan_buf |-------------->| overchan |
** +------+ pipe +----------+ pipe +----------+
**
** Chan_buf reads data coming from innd into a dynamic buffer pool
** and writes data to overchan when the write would not be blocked.
** When overchan falls behind, chan_buf allocates more buffers to
** hold incoming data. As overchan catches up, unused buffers are
** freed. This minimizes the memory size of the chan_buf process.
**
** It has been successfully used with FreeBSD 2.2-STABLE and INN
** 1.5.1. I haven't tried it w/ other operating systems or diff-
** erent versions of INN.
**
** It should linked against INN's libinn.a to include that library's
** SetNonBlocking function.
**
**
** Usage:
**
** chan_buf progname [args]
**
** where progname is the pathname of the program to execute and
** args are the arguments to pass to progname.
**
**
** Written by:
** Frank MacLachlan (fpm@n2.net)
** 20-Jul-97
*/
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <string.h>
#include <stdarg.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/wait.h>
#define MAXLINE 256
#define TRUE 1
#define FALSE 0
#define BUFLEN 8192 /* buffer length (must be power of 2) */
#define BUFMAX (BUFLEN-1) /* max buffer index (0..BUFMAX) */
#define BUFMSK (BUFLEN-1) /* mask used when incrementing buffer ptrs */
/*
** Definition of a buffer:
*/
typedef struct buf {
struct buf *next; /* link to next element */
int curlen; /* current data length (0..BUFLEN) */
int head; /* current buffer head (0..BUFMAX) */
int tail; /* current buffer tail (0..BUFMAX) */
char buf[BUFLEN]; /* data buffer */
} BUF;
/*
** Definition of a buffer list:
*/
typedef struct {
BUF *first; /* pointer to first in list */
BUF *last; /* pointer to last in list */
int nbufs; /* number of buffers in list */
} BUFL;
/*
** Global data:
*/
char *MyName; /* name of this program */
BUFL Bufs; /* active buffers list */
/*
** Function prototypes:
*/
void BufCopy(int RdFD, int WrtFD);
BUF *GetNewBuf(void);
BUF *RmFromBL(BUFL *blp);
void AddToBL(BUFL *blp, BUF *p);
int SetNonBlocking(int fd, int flag);
void ErrRet(const char *fmt, ...);
void ErrSys(const char *fmt, ...);
void ErrQuit(const char *fmt, ...);
void ErrDoit(int errnoflag, const char *fmt, va_list ap);
/*
** Create a pipe, fork w/ child exec'ing program given as parameter.
** Parent writes to pipe, child reads stdin from pipe. Parent becomes
** a buffer process which attempts to promptly read any data present on
** stdin.
*/
int
main(int argc, char *argv[])
{
int fd[2];
pid_t pid;
char *p;
/* Set up MyName for error messages */
if ((p = strrchr(argv[0], '/')) != NULL)
MyName = p + 1;
else
MyName = argv[0];
if (argc < 2)
ErrQuit("Usage: %s pathname [args]", MyName);
if (pipe(fd) < 0)
ErrSys("pipe failed");
if ((pid = fork()) < 0)
ErrSys("fork failed");
if (pid > 0) {
/*
* Parent: copy data from STDIN_FILENO to write end of pipe.
*/
close(fd[0]); /* close read end of pipe */
BufCopy(STDIN_FILENO, fd[1]);
close(fd[1]); /* force EOF for reader */
/* wait for child to terminate */
if (waitpid(pid, NULL, 0) < 0)
ErrSys("waitpid error");
exit(0);
} else {
/*
* Child: exec the program we pipe to.
*/
close(fd[1]); /* close write end of pipe */
/*
* Connect read end of child's pipe to stdin.
*/
if (fd[0] != STDIN_FILENO) {
if (dup2(fd[0], STDIN_FILENO) != STDIN_FILENO)
ErrSys("dup2 failed");
close(fd[0]);
}
if (execv(argv[1], &argv[1]) < 0)
ErrSys("execv failed for %s", argv[1]);
close(STDIN_FILENO);
exit(1);
}
}
/*
** Read any available data from RdFD into buffer, write data to
** WrtFD when possible.
*/
void
BufCopy(int RdFD, int WrtFD)
{
int avail, n;
int ReadEOF = FALSE; /* TRUE if read ret'd EOF */
int MaxFD;
int iovcnt;
BUF *p;
fd_set RdSet, WrtSet;
struct iovec iovecs[2];
/* configure output file for non-blocking writes */
if (SetNonBlocking(WrtFD, TRUE) < 0)
ErrSys("SetNonBlocking TRUE failed");
Bufs.first = NULL;
Bufs.nbufs = 0;
/* Allocate a single buffer at first */
AddToBL(&Bufs, GetNewBuf());
FD_ZERO(&RdSet);
FD_ZERO(&WrtSet);
MaxFD = (WrtFD >= RdFD) ? WrtFD : RdFD;
/*
* Loop until we see EOF from the input and we've drained
* our buffers.
*/
while (! ReadEOF || Bufs.first->curlen > 0) {
/* don't check for read data if we read EOF */
if (! ReadEOF)
FD_SET(RdFD, &RdSet);
/* check for write ready only if we have buffered data */
if (Bufs.first->curlen > 0)
FD_SET(WrtFD, &WrtSet);
else
FD_CLR(WrtFD, &WrtSet);
if (select(MaxFD+1, &RdSet, &WrtSet, NULL, NULL) < 0)
ErrSys("select error");
if (FD_ISSET(RdFD, &RdSet)) {
/*
* Read data available. See how much space we have
* in the buffer. If no space, allocate a new buffer.
*/
if ((avail = BUFLEN - Bufs.last->curlen) <= 0) {
AddToBL(&Bufs, GetNewBuf());
avail = BUFLEN;
}
n = BUFLEN - Bufs.last->tail;
if (avail <= n)
iovcnt = 1;
else {
iovecs[1].iov_len = avail - n;
iovecs[1].iov_base = Bufs.last->buf;
iovcnt = 2;
avail = n;
}
iovecs[0].iov_len = avail;
iovecs[0].iov_base = Bufs.last->buf + Bufs.last->tail;
if ((n = readv(RdFD, iovecs, iovcnt)) <= 0) {
/*
* Error or EOF. Set ReadEOF to TRUE to stop
* further reads from input.
*/
if (n < 0)
ErrRet("read error");
ReadEOF = TRUE;
FD_CLR(RdFD, &RdSet);
} else {
/*
* Got some data. Update buffer pointer, count.
*/
Bufs.last->tail += n;
Bufs.last->tail &= BUFMSK;
Bufs.last->curlen += n;
}
}
if (FD_ISSET(WrtFD, &WrtSet)) {
/*
* We have some data and can write w/o blocking.
*/
avail = Bufs.first->curlen;
n = BUFLEN - Bufs.first->head;
if (avail <= n)
iovcnt = 1;
else {
iovecs[1].iov_len = avail - n;
iovecs[1].iov_base = Bufs.first->buf;
iovcnt = 2;
avail = n;
}
iovecs[0].iov_len = avail;
iovecs[0].iov_base = Bufs.first->buf + Bufs.first->head;
if ((n = writev(WrtFD, iovecs, iovcnt)) <= 0) {
if (n == 0)
/* this should never happen */
ErrQuit("wrote 0 bytes!");
/* ignore error if we were blocked while writing */
if (errno != EAGAIN)
ErrSys("write error");
} else {
/* update buffer pointer, count */
Bufs.first->head += n;
Bufs.first->head &= BUFMSK;
Bufs.first->curlen -= n;
/* free excess empty buffers */
if (Bufs.nbufs > 1 && Bufs.first->curlen == 0)
free(RmFromBL(&Bufs));
}
}
}
}
/*
** Return a new buffer. Print error message and die
** if unsuccessful.
*/
BUF *
GetNewBuf(void)
{
BUF *p;
if ((p = (BUF *)malloc(sizeof(BUF))) == NULL)
ErrSys("can't allocate buffer");
p->curlen = 0;
p->head = p->tail = 0;
return p;
}
/*
** Remove a buffer from a list. Returns NULL if
** the list is empty.
*/
BUF *
RmFromBL(BUFL *blp)
{
BUF *p;
if ((p = blp->first) != NULL) { /* return NULL if list is empty */
blp->first = p->next;
--blp->nbufs;
}
return p;
}
/*
** Add a buffer to a list. No errors are possible here.
*/
void
AddToBL(BUFL *blp, BUF *p)
{
p->next = NULL;
if (blp->first == NULL) /* add to empty list */
blp->first = p;
else
blp->last->next = p;
blp->last = p;
++blp->nbufs;
}
/*
** Non-fatal error related to a system call.
** Print a message and return.
*/
void
ErrRet(const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
ErrDoit(TRUE, fmt, ap);
va_end(ap);
}
/*
** Fatal error related to a system call.
** Print a message and terminate.
*/
void
ErrSys(const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
ErrDoit(TRUE, fmt, ap);
va_end(ap);
exit(1);
}
/*
** Fatal error unrelated to a system call.
** Print a message and terminate.
*/
void
ErrQuit(const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
ErrDoit(FALSE, fmt, ap);
va_end(ap);
exit(1);
}
/*
** Print a message and return to caller.
** Caller specifies "errnoflag".
*/
void
ErrDoit(int errnoflag, const char *fmt, va_list ap)
{
int ErrnoSave;
char buf[MAXLINE];
ErrnoSave = errno;
sprintf(buf, "%s: ", MyName);
vsprintf(buf+strlen(buf), fmt, ap);
if (errnoflag)
sprintf(buf+strlen(buf), " (%s)", strerror(ErrnoSave));
strcat(buf, "\n");
fflush(stdout); /* in case stdout and stderr are the same */
fputs(buf, stderr);
fflush(NULL); /* flushes all stdio output streams */
}
-------------------CUT HERE------------------------
Want to link to this message? Use this URL: <https://mail-archive.FreeBSD.org/cgi/mid.cgi?19970805203947.52488>