stream.mx

这个是内存数据库中的一个管理工具

	bs *s = (bs *) ss->stream_data.p;
	size_t todo = cnt * elmsize;
	size_t n;

	assert(ss->access == ST_READ);
	assert(s->nr <= 1);

	if (s->itotal == 0) {
		short blksize = 0;

		if (s->nr) {
			/* We read the closing block but hadn't returned that yet.
			   Return it now, and note that we did by setting s->nr to
			   0.
			*/
			assert(s->nr == 1);
			s->nr = 0;
			return 0;
		}

		assert(s->nr == 0);

		/* There is nothing more to read in the current block,
		   so read the count for the next block */
		if (!stream_readSht(s->s, &blksize) || blksize < 0) {
			ss->errnr = READ_ERROR;
			return -1;
		}
#ifdef BSTREAM_DEBUG
		printf("RC size: %d, final: %s\n", blksize >> 1, blksize & 1 ? "true" : "false");
		printf("RC %s %d\n", ss->name, blksize);
#endif
		s->itotal = (unsigned) (blksize >> 1); /* amount readable */
		/* store whether this was the last block or not */
		s->nr = blksize & 1;
	}

	/* Fill the caller's buffer. */
	cnt = 0;		/* count how much we put into the buffer */
	while (todo > 0) {
		/* there is more data waiting in the current block,
		   so read it */
		n = todo < s->itotal ? todo : s->itotal;
		while (n > 0) {
			ssize_t m = s->s->read(s->s, buf, 1, n);

			if (m <= 0) {
				ss->errnr = READ_ERROR;
				return -1;
			}
#ifdef BSTREAM_DEBUG
			{
				ssize_t i;

				printf("RD %s %zd \"", ss->name, m);
				for (i = 0; i < m; i++)
					if (' ' <= ((char *) buf)[i] && ((char *) buf)[i] < 127)
						putchar(((char *) buf)[i]);
					else
						printf("\\%03o", ((char *) buf)[i]);
				printf("\"\n");
			}
#endif
			if (m == 0 && !isa_block_stream(s->s))
				break;
			buf = (void *) ((char *) buf + m);
			cnt += m;
			n -= m;
			s->itotal -= (int) m;
			todo -= m;
		}

		if (s->itotal == 0) {
			short blksize = 0;

			/* The current block has been completely read,
			   so read the count for the next block, only if the
			   previous was not the last one */
			if (s->nr) {
				break;
			} else if (!stream_readSht(s->s, &blksize) || blksize < 0) {
				ss->errnr = READ_ERROR;
				return -1;
			}
#ifdef BSTREAM_DEBUG
			printf("RC size: %d, final: %s\n", blksize >> 1, blksize & 1 ? "true" : "false");
			printf("RC %s %d\n", ss->name, s->nr);
			printf("RC %s %d\n", ss->name, blksize);
#endif
			s->itotal = (unsigned) (blksize >> 1); /* amount readable */
			/* store whether this was the last block or not */
			s->nr = blksize & 1;
		}
	}
	/* if we got an empty block with the end-of-sequence marker
	   set (low-order bit) we must only return an empty read once, so
	   we must squash the flag that we still have to return an empty
	   read */
	if (todo > 0 && cnt == 0)
		s->nr = 0;
	return cnt / elmsize;
}

/* Read the next bit of a block.  If this was the last bit of the
   current block, set the value pointed to by last to 1, otherwise set
   it to 0. */
ssize_t
bs_read_next(stream *ss, void *buf, size_t nbytes, int *last)
{
	ssize_t n;
	bs *s = (bs *) ss->stream_data.p;

	n = bs_read(ss, buf, 1, nbytes);
	if (n < 0) {
		if (last)
			*last = 1;
		return -1;
	}
	if (last)
		*last = s->itotal == 0;
	if (s->itotal == 0) {
		/* we don't want to get an empty buffer at the next read */
		s->nr = 0;
	}
	return n;
}

static void
bs_close(stream *ss)
{
	bs *s = (bs *) ss->stream_data.p;

	assert(s);
	assert(s->s);
	s->s->close(s->s);
}

static void
bs_destroy(stream *ss)
{
	bs *s = (bs *) ss->stream_data.p;

	assert(s);
	assert(s->s);
	s->s->destroy(s->s);
	free(s);
	destroy(ss);
}

stream *
block_stream(stream *s)
{
	stream *ns;
	bs *b;

#ifdef STREAM_DEBUG
	printf("block_stream %s\n", s->name ? s->name : "<unnamed>");
#endif
	if ((ns = create_stream(s->name)) == NULL)
		return NULL;
	if ((b = bs_create(s)) == NULL)
		ns->errnr = OPEN_ERROR;
	/* blocksizes have a fixed little endian byteorder */ 
#ifdef WORDS_BIGENDIAN
	s->byteorder = 3412; /* simply != 1234 */
#endif
	ns->type = s->type;
	ns->access = s->access;
	ns->read = bs_read;
	ns->write = bs_write;
	ns->close = bs_close;
	ns->flush = bs_flush;
	ns->destroy = bs_destroy;
	ns->stream_data.p = (void *) b;

	return ns;
}

int
isa_block_stream(stream *s)
{
	return s->read == bs_read || s->write == bs_write;
}

ssize_t 
stream_read_block(stream *s, void *buf, size_t elmsize, size_t cnt)
{
	int len = 0;
	char x = 0;
	assert( s->read == bs_read || s->write == bs_write );
	len = stream_read(s, buf, elmsize, cnt);
	stream_read(s, &x, 0, 0); /* read prompt */
	if (x > 0)
		return -1;
	return len;
}

@= stream_readVal
int
stream_read@1(stream *s, @2 *val)
{
	switch (s->read(s, (void *) val, sizeof(*val), 1)) {
	case 1:
		if (s->byteorder != 1234)
			*val = @3_SWAP(*val);
		return 1;
	case 0:
		/* consider EOF an error */
		s->errnr = READ_ERROR;
		/* fall through */
	default:
		/* read failed */
		return 0;
	}
}
@c

#define no_SWAP(x) x

@:stream_readVal(Bte,signed char,no)@

int
stream_writeBte(stream *s, signed char val)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) &val, sizeof(val), 1) == 1;
}

@:stream_readVal(Sht,short,short_int)@

int
stream_writeSht(stream *s, short val)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) &val, sizeof(val), 1) == 1;
}

@:stream_readVal(Int,int,normal_int)@

int
stream_writeInt(stream *s, int val)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) &val, sizeof(val), (size_t) 1) == 1;
}

@:stream_readVal(Lng,lng,long_long)@

int
stream_writeLng(stream *s, lng val)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) &val, sizeof(val), (size_t) 1) == 1;
}

@= stream_readArray
int
stream_read@1Array(stream *s, @2 *val, size_t cnt)
{
	if (s->read(s, (void *) val, sizeof(*val), cnt) < (ssize_t) cnt) {
		s->errnr = READ_ERROR;
		return 0;
	}

	if (s->byteorder != 1234) {
		size_t i;
		for (i = 0; i < cnt; i++, val++)
			*val = @3_SWAP(*val);	
	}
	return 1;
}
@c

@:stream_readArray(Bte,signed char,no)@

int
stream_writeBteArray(stream *s, const signed char *val, size_t cnt)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) val, sizeof(*val), cnt) == (ssize_t) cnt;
}

@:stream_readArray(Sht,short,short_int)@

int
stream_writeShtArray(stream *s, const short *val, size_t cnt)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) val, sizeof(*val), cnt) == (ssize_t) cnt;
}

@:stream_readArray(Int,int,normal_int)@

int
stream_writeIntArray(stream *s, const int *val, size_t cnt)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) val, sizeof(*val), cnt) == (ssize_t) cnt;
}

@:stream_readArray(Lng,lng,long_long)@

int
stream_writeLngArray(stream *s, const lng *val, size_t cnt)
{
	if (!s || s->errnr) return(0);
	return s->write(s, (void *) val, sizeof(*val), cnt) == (ssize_t) cnt;
}

int
stream_printf(stream *s, const char *format, ...)
{
	char buf[BUFSIZ], *bf = buf;
	int i = 0;
	size_t bfsz = BUFSIZ;
	va_list ap;
	if (!s || s->errnr) return(-1);

@= stream_printf_va_vsnprintf
	va_start(ap, format);
	i = vsnprintf(bf, bfsz, format, ap);
	va_end (ap);
@c
	@:stream_printf_va_vsnprintf@
	while (i < 0 || (size_t) i >= bfsz) {
		if (i >= 0)	/* glibc 2.1 */
			bfsz = (size_t) i + 1;	/* precisely what is needed */
		else		/* glibc 2.0 */
			bfsz *= 2;	/* twice the old size */
		if (bf != buf)
			free(bf);
		bf = malloc(bfsz);
		assert(bf != NULL);
		if (bf == NULL) {
			s->errnr = WRITE_ERROR;
			return -1;
		}
		@:stream_printf_va_vsnprintf@
	}
	s->write(s, (void *) bf, (size_t) i, (size_t) 1);
	if (bf != buf)
		free(bf);
	return s->errnr ? -1 : i;
}


bstream *
bstream_create(stream *s, size_t size)
{
	bstream *b;

	if ((b = malloc(sizeof(*b))) == NULL)
		return NULL;
	b->mode = size;		/* 64bit: should check that size isn't too large and fits in an int */
	if (size == 0)
		size = BUFSIZ;
	b->s = s;
	b->buf = malloc(size + 1);
	b->size = b->buf ? size : 0;
	b->pos = 0;
	b->len = 0;
	b->eof = 0;
	return b;
}

ssize_t
bstream_read(bstream *s, size_t size)
{
	ssize_t rd;

	if (s->eof)
		return 0;

	if (s->pos > 0) {
		if (s->pos < s->len)
			/* move all data and end of string marker */
			memmove(s->buf, s->buf + s->pos, s->len - s->pos + 1);
		s->len -= s->pos;
		s->pos = 0;
	}

	if (s->len == s->size && (s->buf = realloc(s->buf, (s->size <<= 1) + 1)) == NULL) {
		s->size = 0;
		s->len = 0;
		s->pos = 0;
		return -1;
	}

	if (size > s->size - s->len)
		size = s->size - s->len;

	rd = s->s->read(s->s, s->buf + s->len, 1, size);

	if (rd < 0)
		return rd;

	if (rd == 0) {
		s->eof = 1;
		return 0;
	}
	s->len += rd;
	s->buf[s->len] = 0;	/* fill in the spare with EOS */
	return rd;
}

#ifdef _POSIX2_LINE_MAX
#define STREAM_LINE_MAX _POSIX2_LINE_MAX
#else
#define STREAM_LINE_MAX 2048
#endif

static ssize_t
bstream_readline(bstream *s)
{
	size_t size = STREAM_LINE_MAX;
	ssize_t rd;

	if (s->eof)
		return 0;

	if (s->pos > 0 && s->len + size >= s->size) {
		if (s->pos < s->len)
			/* move all data and end of string marker */
			memmove(s->buf, s->buf + s->pos, s->len - s->pos + 1);
		s->len -= s->pos;
		s->pos = 0;
	}

	if (s->len == s->size && (s->buf = realloc(s->buf, (s->size <<= 1) + 1)) == NULL) {
		s->size = 0;
		s->len = 0;
		s->pos = 0;
		return -1;
	}

	if (size > s->size - s->len)
		size = s->size - s->len;

	if (fgets(s->buf + s->len, (int) size, s->s->stream_data.p) == NULL)
		return -1;

	rd = strlen(s->buf + s->len);

	if (rd == 0) {
		s->eof = 1;
		return 0;
	}
	s->len += rd;
	s->buf[s->len] = 0;	/* fill in the spare with EOS */
	return rd;
}


ssize_t
bstream_next(bstream *s)
{
	if (s->mode) {
		return bstream_read(s, s->mode);
	} else if (s->s->read == file_read) {
		return bstream_readline(s);
	} else {
		ssize_t sz = 0, rd;

		while ((rd = bstream_read(s, 1)) == 1 && s->buf[s->pos + sz] != '\n') {
			sz += rd;
		}
		if (rd < 0)
			return rd;
		return sz;
	}
}

void
bstream_destroy(bstream *s)
{
	free(s->buf);
	free(s);
}

#ifdef HAVE_PIPE
static ssize_t
pipe_write(stream *s, const void *buf, size_t elmsize, size_t cnt)
{
	size_t size = elmsize * cnt;
	ssize_t sz;

	if (s->errnr)
		return -1;

	if (size == 0 || elmsize == 0)
		return cnt;
	sz = write(s->stream_data.i, buf, size);
	if (sz > 0)
		return sz / elmsize;
	if (sz < 0)
		s->errnr = WRITE_ERROR;
	return sz;
}

static ssize_t
pipe_read(stream *s, void *buf, size_t elmsize, size_t cnt)
{
	ssize_t nr = 0;
	size_t res = 0, size = elmsize * cnt;

	if (s->errnr)
		return -1;

	while (res < size && (nr = read(s->stream_data.i, (void *) ((char *) buf + res), size - res)) > 0) {
		res += nr;
	}
	if (nr < 0) {
		s->errnr = READ_ERROR;
		return nr;
	}
	if (res)
		return res / elmsize;
	return 0;
}

static void
pipe_close(stream *s)
{
	if (s->stream_data.i >= 0)
		close(s->stream_data.i);
	s->stream_data.i = -1;
}

int
rendezvous_streams(stream **in, stream **out, const char *name)
{
	stream *sin, *sout;
	int pipes[2];

	if ((sin = create_stream(name)) == NULL)
		return 0;
	if ((sout = create_stream(name)) == NULL) {
		destroy(sin);
		return 0;
	}
	if (pipe(pipes) != 0) {
		destroy(sin);
		destroy(sout);
		return 0;
	}
	sin->access = ST_READ;
	sin->close = pipe_close;
	sin->read = pipe_read;
	sin->stream_data.i = pipes[0];
	sin->type = ST_BIN;
	sout->access = ST_WRITE;
	sout->close = pipe_close;
	sout->stream_data.i = pipes[1];
	sout->type = ST_BIN;
	sout->write = pipe_write;
	*in = sin;
	*out = sout;
	return 1;
}
#else
int
rendezvous_streams(stream **in, stream **out, const char *name)
{
	(void) in;
	(void) out;
	(void) name;
	return 0;
}
#endif /* HAVE_PIPE */