From c5cc6959705c310f4f7a864d77aae54171707c04 Mon Sep 17 00:00:00 2001 From: Ryan Dahl Date: Wed, 7 Nov 2018 13:16:07 -0500 Subject: [PATCH] First pass at bufio. --- buffer.ts | 229 +++++++++++++++++++++++++++++++++++++++++++++++++ buffer_test.ts | 185 +++++++++++++++++++++++++++++++++++++++ bufio.ts | 92 ++++++++++++++++++++ bufio_test.ts | 32 +++++++ file_server.ts | 18 ++++ http.ts | 72 +++++++++++----- util.ts | 6 ++ 7 files changed, 613 insertions(+), 21 deletions(-) create mode 100644 buffer.ts create mode 100644 buffer_test.ts create mode 100644 bufio.ts create mode 100644 bufio_test.ts create mode 100644 file_server.ts create mode 100644 util.ts diff --git a/buffer.ts b/buffer.ts new file mode 100644 index 0000000000..1c4d68e98d --- /dev/null +++ b/buffer.ts @@ -0,0 +1,229 @@ +// This code has been ported almost directly from Go's src/bytes/buffer.go +// Copyright 2009 The Go Authors. All rights reserved. BSD license. +// https://github.com/golang/go/blob/master/LICENSE + +//import * as io from "./io"; +import { Reader, Writer, ReadResult } from "deno"; +import { assert } from "./util.ts"; + +// MIN_READ is the minimum ArrayBuffer size passed to a read call by +// buffer.ReadFrom. As long as the Buffer has at least MIN_READ bytes beyond +// what is required to hold the contents of r, readFrom() will not grow the +// underlying buffer. +const MIN_READ = 512; +const MAX_SIZE = 2 ** 32 - 2; + +// `off` is the offset into `dst` where it will at which to begin writing values +// from `src`. +// Returns the number of bytes copied. +function copyBytes(dst: Uint8Array, src: Uint8Array, off = 0): number { + const r = dst.byteLength - off; + if (src.byteLength > r) { + src = src.subarray(0, r); + } + dst.set(src, off); + return src.byteLength; +} + +/** A Buffer is a variable-sized buffer of bytes with read() and write() + * methods. Based on https://golang.org/pkg/bytes/#Buffer + */ +export class Buffer implements Reader, Writer { + private buf: Uint8Array; // contents are the bytes buf[off : len(buf)] + private off = 0; // read at buf[off], write at buf[buf.byteLength] + + constructor(ab?: ArrayBuffer) { + if (ab == null) { + this.buf = new Uint8Array(0); + } else { + this.buf = new Uint8Array(ab); + } + } + + /** bytes() returns a slice holding the unread portion of the buffer. + * The slice is valid for use only until the next buffer modification (that + * is, only until the next call to a method like read(), write(), reset(), or + * truncate()). The slice aliases the buffer content at least until the next + * buffer modification, so immediate changes to the slice will affect the + * result of future reads. + */ + bytes(): Uint8Array { + return this.buf.subarray(this.off); + } + + /** toString() returns the contents of the unread portion of the buffer + * as a string. Warning - if multibyte characters are present when data is + * flowing through the buffer, this method may result in incorrect strings + * due to a character being split. + */ + toString(): string { + const decoder = new TextDecoder(); + return decoder.decode(this.buf.subarray(this.off)); + } + + /** empty() returns whether the unread portion of the buffer is empty. */ + empty() { + return this.buf.byteLength <= this.off; + } + + /** length is a getter that returns the number of bytes of the unread + * portion of the buffer + */ + get length() { + return this.buf.byteLength - this.off; + } + + /** Returns the capacity of the buffer's underlying byte slice, that is, + * the total space allocated for the buffer's data. + */ + get capacity(): number { + return this.buf.buffer.byteLength; + } + + /** truncate() discards all but the first n unread bytes from the buffer but + * continues to use the same allocated storage. It throws if n is negative or + * greater than the length of the buffer. + */ + truncate(n: number): void { + if (n === 0) { + this.reset(); + return; + } + if (n < 0 || n > this.length) { + throw Error("bytes.Buffer: truncation out of range"); + } + this._reslice(this.off + n); + } + + /** reset() resets the buffer to be empty, but it retains the underlying + * storage for use by future writes. reset() is the same as truncate(0) + */ + reset(): void { + this._reslice(0); + this.off = 0; + } + + /** _tryGrowByReslice() is a version of grow for the fast-case + * where the internal buffer only needs to be resliced. It returns the index + * where bytes should be written and whether it succeeded. + * It returns -1 if a reslice was not needed. + */ + private _tryGrowByReslice(n: number): number { + const l = this.buf.byteLength; + if (n <= this.capacity - l) { + this._reslice(l + n); + return l; + } + return -1; + } + + private _reslice(len: number): void { + assert(len <= this.buf.buffer.byteLength); + this.buf = new Uint8Array(this.buf.buffer, 0, len); + } + + /** read() reads the next len(p) bytes from the buffer or until the buffer + * is drained. The return value n is the number of bytes read. If the + * buffer has no data to return, eof in the response will be true. + */ + async read(p: ArrayBufferView): Promise { + if (!(p instanceof Uint8Array)) { + throw Error("Only Uint8Array supported"); + } + if (this.empty()) { + // Buffer is empty, reset to recover space. + this.reset(); + if (p.byteLength === 0) { + // TODO This edge case should be tested by porting TestReadEmptyAtEOF + // from the Go tests. + return { nread: 0, eof: false }; + } + return { nread: 0, eof: true }; + } + const nread = copyBytes(p, this.buf.subarray(this.off)); + this.off += nread; + return { nread, eof: false }; + } + + async write(p: ArrayBufferView): Promise { + const m = this._grow(p.byteLength); + if (!(p instanceof Uint8Array)) { + throw Error("Only Uint8Array supported"); + } + return copyBytes(this.buf, p, m); + } + + /** _grow() grows the buffer to guarantee space for n more bytes. + * It returns the index where bytes should be written. + * If the buffer can't grow it will throw with ErrTooLarge. + */ + private _grow(n: number): number { + const m = this.length; + // If buffer is empty, reset to recover space. + if (m === 0 && this.off !== 0) { + this.reset(); + } + // Fast: Try to grow by means of a reslice. + const i = this._tryGrowByReslice(n); + if (i >= 0) { + return i; + } + const c = this.capacity; + if (n <= Math.floor(c / 2) - m) { + // We can slide things down instead of allocating a new + // ArrayBuffer. We only need m+n <= c to slide, but + // we instead let capacity get twice as large so we + // don't spend all our time copying. + copyBytes(this.buf, this.buf.subarray(this.off)); + } else if (c > MAX_SIZE - c - n) { + throw Error("ErrTooLarge"); // TODO DenoError(TooLarge) + } else { + // Not enough space anywhere, we need to allocate. + const buf = new Uint8Array(2 * c + n); + copyBytes(buf, this.buf.subarray(this.off)); + this.buf = buf; + } + // Restore this.off and len(this.buf). + this.off = 0; + this._reslice(m + n); + return m; + } + + /** grow() grows the buffer's capacity, if necessary, to guarantee space for + * another n bytes. After grow(n), at least n bytes can be written to the + * buffer without another allocation. If n is negative, grow() will panic. If + * the buffer can't grow it will throw ErrTooLarge. + * Based on https://golang.org/pkg/bytes/#Buffer.Grow + */ + grow(n: number): void { + if (n < 0) { + throw Error("Buffer.grow: negative count"); + } + const m = this._grow(n); + this._reslice(m); + } + + /** readFrom() reads data from r until EOF and appends it to the buffer, + * growing the buffer as needed. It returns the number of bytes read. If the + * buffer becomes too large, readFrom will panic with ErrTooLarge. + * Based on https://golang.org/pkg/bytes/#Buffer.ReadFrom + */ + async readFrom(r: Reader): Promise { + let n = 0; + while (true) { + try { + const i = this._grow(MIN_READ); + this._reslice(i); + const fub = new Uint8Array(this.buf.buffer, i); + const { nread, eof } = await r.read(fub); + this._reslice(i + nread); + n += nread; + if (eof) { + return n; + } + } catch (e) { + return n; + } + } + } +} diff --git a/buffer_test.ts b/buffer_test.ts new file mode 100644 index 0000000000..c614b2e03e --- /dev/null +++ b/buffer_test.ts @@ -0,0 +1,185 @@ +// This code has been ported almost directly from Go's src/bytes/buffer_test.go +// Copyright 2009 The Go Authors. All rights reserved. BSD license. +// https://github.com/golang/go/blob/master/LICENSE +import { test, assert, assertEqual } from "./test_util.ts"; +import { Buffer } from "deno"; + +// N controls how many iterations of certain checks are performed. +const N = 100; +let testBytes: Uint8Array | null; +let testString: string | null; + +function init() { + if (testBytes == null) { + testBytes = new Uint8Array(N); + for (let i = 0; i < N; i++) { + testBytes[i] = "a".charCodeAt(0) + (i % 26); + } + const decoder = new TextDecoder(); + testString = decoder.decode(testBytes); + } +} + +function check(buf: Buffer, s: string) { + const bytes = buf.bytes(); + assertEqual(buf.length, bytes.byteLength); + const decoder = new TextDecoder(); + const bytesStr = decoder.decode(bytes); + assertEqual(bytesStr, s); + assertEqual(buf.length, buf.toString().length); + assertEqual(buf.length, s.length); +} + +// Fill buf through n writes of byte slice fub. +// The initial contents of buf corresponds to the string s; +// the result is the final contents of buf returned as a string. +async function fillBytes( + buf: Buffer, + s: string, + n: number, + fub: Uint8Array +): Promise { + check(buf, s); + for (; n > 0; n--) { + let m = await buf.write(fub); + assertEqual(m, fub.byteLength); + const decoder = new TextDecoder(); + s += decoder.decode(fub); + check(buf, s); + } + return s; +} + +// Empty buf through repeated reads into fub. +// The initial contents of buf corresponds to the string s. +async function empty(buf: Buffer, s: string, fub: Uint8Array): Promise { + check(buf, s); + while (true) { + const r = await buf.read(fub); + if (r.nread == 0) { + break; + } + s = s.slice(r.nread); + check(buf, s); + } + check(buf, ""); +} + +test(function bufferNewBuffer() { + init(); + const buf = new Buffer(testBytes.buffer as ArrayBuffer); + check(buf, testString); +}); + +test(async function bufferBasicOperations() { + init(); + let buf = new Buffer(); + for (let i = 0; i < 5; i++) { + check(buf, ""); + + buf.reset(); + check(buf, ""); + + buf.truncate(0); + check(buf, ""); + + let n = await buf.write(testBytes.subarray(0, 1)); + assertEqual(n, 1); + check(buf, "a"); + + n = await buf.write(testBytes.subarray(1, 2)); + assertEqual(n, 1); + check(buf, "ab"); + + n = await buf.write(testBytes.subarray(2, 26)); + assertEqual(n, 24); + check(buf, testString.slice(0, 26)); + + buf.truncate(26); + check(buf, testString.slice(0, 26)); + + buf.truncate(20); + check(buf, testString.slice(0, 20)); + + await empty(buf, testString.slice(0, 20), new Uint8Array(5)); + await empty(buf, "", new Uint8Array(100)); + + // TODO buf.writeByte() + // TODO buf.readByte() + } +}); + +test(async function bufferLargeByteWrites() { + init(); + const buf = new Buffer(); + const limit = 9; + for (let i = 3; i < limit; i += 3) { + const s = await fillBytes(buf, "", 5, testBytes); + await empty(buf, s, new Uint8Array(Math.floor(testString.length / i))); + } + check(buf, ""); +}); + +test(async function bufferLargeByteReads() { + init(); + const buf = new Buffer(); + for (let i = 3; i < 30; i += 3) { + const n = Math.floor(testBytes.byteLength / i); + const s = await fillBytes(buf, "", 5, testBytes.subarray(0, n)); + await empty(buf, s, new Uint8Array(testString.length)); + } + check(buf, ""); +}); + +test(function bufferCapWithPreallocatedSlice() { + const buf = new Buffer(new ArrayBuffer(10)); + assertEqual(buf.capacity, 10); +}); + +test(async function bufferReadFrom() { + init(); + const buf = new Buffer(); + for (let i = 3; i < 30; i += 3) { + const s = await fillBytes( + buf, + "", + 5, + testBytes.subarray(0, Math.floor(testBytes.byteLength / i)) + ); + const b = new Buffer(); + await b.readFrom(buf); + const fub = new Uint8Array(testString.length); + await empty(b, s, fub); + } +}); + +function repeat(c: string, bytes: number): Uint8Array { + assertEqual(c.length, 1); + const ui8 = new Uint8Array(bytes); + ui8.fill(c.charCodeAt(0)); + return ui8; +} + +test(async function bufferTestGrow() { + const tmp = new Uint8Array(72); + for (let startLen of [0, 100, 1000, 10000, 100000]) { + const xBytes = repeat("x", startLen); + for (let growLen of [0, 100, 1000, 10000, 100000]) { + const buf = new Buffer(xBytes.buffer as ArrayBuffer); + // If we read, this affects buf.off, which is good to test. + const { nread, eof } = await buf.read(tmp); + buf.grow(growLen); + const yBytes = repeat("y", growLen); + await buf.write(yBytes); + // Check that buffer has correct data. + assertEqual( + buf.bytes().subarray(0, startLen - nread), + xBytes.subarray(nread) + ); + assertEqual( + buf.bytes().subarray(startLen - nread, startLen - nread + growLen), + yBytes + ); + } + } +}); diff --git a/bufio.ts b/bufio.ts new file mode 100644 index 0000000000..dc1a2095d9 --- /dev/null +++ b/bufio.ts @@ -0,0 +1,92 @@ +import * as deno from "deno"; + +const DEFAULT_BUF_SIZE = 4096; +const MIN_BUF_SIZE = 16; +const MAX_CONSECUTIVE_EMPTY_READS = 100; + +export class Reader implements deno.Reader { + private buf: Uint8Array; + private rd: deno.Reader; // Reader provided by caller. + private r = 0; // buf read position. + private w = 0; // buf write position. + private lastByte: number; + private lastCharSize: number; + + constructor(rd: deno.Reader, size = DEFAULT_BUF_SIZE) { + if (size < MIN_BUF_SIZE) { + size = MIN_BUF_SIZE; + } + this._reset(new Uint8Array(size), rd) + } + + /** Returns the size of the underlying buffer in bytes. */ + get byteLength(): number { + return this.buf.byteLength; + } + + // Reads a new chunk into the buffer. + // Returns true if EOF, false on successful read. + async _fill(): Promise { + // Slide existing data to beginning. + if (this.r > 0) { + this.buf.copyWithin(0, this.r, this.w); + this.w -= this.r; + this.r = 0; + } + + if (this.w >= this.buf.byteLength) { + throw Error("bufio: tried to fill full buffer"); + } + + // Read new data: try a limited number of times. + for (let i = MAX_CONSECUTIVE_EMPTY_READS; i > 0; i--) { + const { nread, eof } = await this.rd.read(this.buf.subarray(this.w)); + if (nread < 0) { + throw Error("negative read"); + } + this.w += nread; + if (eof) { + return true; + } + if (nread > 0) { + return false; + } + } + throw Error("No Progress"); + } + + /** Discards any buffered data, resets all state, and switches + * the buffered reader to read from r. + */ + reset(r: deno.Reader): void { + this._reset(this.buf, r); + } + + private _reset(buf: Uint8Array, rd: deno.Reader): void { + this.buf = buf; + this.rd = rd; + this.lastByte = -1; + this.lastCharSize = -1; + } + + async read(p: ArrayBufferView): Promise { + throw Error("not implemented"); + return { nread: 0, eof: false }; + } + + /** Returns the next byte [0, 255] or -1 if EOF. */ + async readByte(): Promise { + while (this.r === this.w) { + const eof = await this._fill(); // buffer is empty. + if (eof) { + return -1; + } + } + const c = this.buf[this.r]; + this.r++; + this.lastByte = c; + return c; + } +} + + diff --git a/bufio_test.ts b/bufio_test.ts new file mode 100644 index 0000000000..a66052ba16 --- /dev/null +++ b/bufio_test.ts @@ -0,0 +1,32 @@ +import * as deno from "deno"; +import { test, assertEqual } from "http://deno.land/x/testing/testing.ts"; +import * as bufio from "./bufio.ts"; +import { Buffer } from "./buffer.ts"; + +async function readBytes(buf: bufio.Reader): Promise { + const b = new Uint8Array(1000); + let nb = 0; + while (true) { + let c = await buf.readByte(); + if (c < 0) { + break; // EOF + } + b[nb] = c; + nb++; + } + const decoder = new TextDecoder(); + return decoder.decode(b.subarray(0, nb)); +} + +function stringsReader(s: string): deno.Reader { + const encoder = new TextEncoder(); + const ui8 = encoder.encode(s); + return new Buffer(ui8.buffer as ArrayBuffer); +} + +test(async function bufioReaderSimple() { + const data = "hello world"; + const b = new bufio.Reader(stringsReader(data)); + const s = await readBytes(b); + assertEqual(s, data); +}); diff --git a/file_server.ts b/file_server.ts new file mode 100644 index 0000000000..9d3d5366ea --- /dev/null +++ b/file_server.ts @@ -0,0 +1,18 @@ +import { listenAndServe } from "./http.ts"; +import { open, cwd } from "deno"; + +const addr = "0.0.0.0:4500"; +const d = cwd(); + +listenAndServe(addr, async (req) => { + const filename = d + "/" + req.url; + let res; + try { + res = { status: 200, body: open(filename) }; + } catch(e) { + res = { status: 500, body: "bad" }; + } + req.respond(res); +}); + +console.log(`HTTP server listening on http://${addr}/`); diff --git a/http.ts b/http.ts index d4c496e152..c8b7d8d904 100644 --- a/http.ts +++ b/http.ts @@ -1,42 +1,72 @@ import * as deno from "deno"; +import * as bufio from "./bufio.ts"; +import { TextProtoReader } from "./textproto.ts"; + +type Handler = (req: ServerRequest) => Promise; class Server { _closing = false; constructor(readonly listener: deno.Listener) {} - async serveConn(conn: deno.Conn) { - const buffer = new Uint8Array(1024); - try { - while (true) { - const r = await conn.read(buffer); - if (r.eof) { - break; - } - const response = new TextEncoder().encode( - "HTTP/1.1 200 OK\r\nContent-Length: 12\r\n\r\nHello World\n" - ); - await conn.write(response); - } - } finally { - conn.close(); - } - } - - async serve() { + async serve(handler: Handler) { while (!this._closing) { - const conn = await this.listener.accept(); - this.serveConn(conn); + const c = await this.listener.accept(); + const sc = new ServerConn(c); + sc.serve(handler); } } close() { this._closing = true; + this.listener.close(); } } +class ServerConn { + constructor(readonly c: deno.Conn) { + // TODO Use memory pool to obtain bufr and bufw. + this.bufr = new bufio.Reader(c); + this.bufw = new bufio.Writer(c); + } + + async serve(handler: Handler): Promise { + const buffer = new Uint8Array(1024); + try { + while (true) { + const req = readRequest(this.bufr); + + /* + const response = new TextEncoder().encode( + "HTTP/1.1 200 OK\r\nContent-Length: 12\r\n\r\nHello World\n" + ); + await this.c.write(response); + */ + } + } finally { + this.c.close(); + } + } +} + +function readRequest(b: bufio.Reader): ServerRequest { + const tp = new TextProtoReader(b); + const req = new ServerRequest(); + + // First line: GET /index.html HTTP/1.0 + const s = await tp.readLine(); + const [ method, url, proto ] = parseRequestLine(s); + console.log("readRequest", method, url); +} + +// Returns [method, url, proto] +function parseRequestLine(line: string): [ string, string, string ] { + return line.split(" ", 3); +} + export function listen(addr: string): Server { const listener = deno.listen("tcp", addr); const s = new Server(listener); return s; } + diff --git a/util.ts b/util.ts new file mode 100644 index 0000000000..decf4d043f --- /dev/null +++ b/util.ts @@ -0,0 +1,6 @@ + +export function assert(cond: boolean, msg = "assert") { + if (!cond) { + throw Error(msg); + } +}