From 730b42bf316ea6f5f0da653bfc3af5758a639988 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=B6nke=20Ludwig?= Date: Wed, 21 Oct 2020 16:57:47 +0200 Subject: [PATCH] Add a test for pipe(). --- tests/vibe.core.stream.pipe.d | 186 ++++++++++++++++++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 tests/vibe.core.stream.pipe.d diff --git a/tests/vibe.core.stream.pipe.d b/tests/vibe.core.stream.pipe.d new file mode 100644 index 0000000..e818605 --- /dev/null +++ b/tests/vibe.core.stream.pipe.d @@ -0,0 +1,186 @@ +/+ dub.sdl: + name "test" + dependency "vibe-core" path=".." ++/ +module test; + +import vibe.core.core; +import vibe.core.stream; +import std.algorithm : min; +import std.array : Appender, appender; +import std.exception; +import std.random; +import core.time : Duration, msecs; + +void main() +{ + auto datau = new uint[](2 * 1024 * 1024); + foreach (ref u; datau) + u = uniform!uint(); + auto data = cast(ubyte[])datau; + + test(data, 0, 0.msecs, 0, 0.msecs); + test(data, 32768, 1.msecs, 32768, 1.msecs); + + test(data, 32768, 0.msecs, 0, 0.msecs); + test(data, 0, 0.msecs, 32768, 0.msecs); + test(data, 32768, 0.msecs, 32768, 0.msecs); + + test(data, 1023*967, 10.msecs, 0, 0.msecs); + test(data, 0, 0.msecs, 1023*967, 20.msecs); + test(data, 1023*967, 10.msecs, 1023*967, 10.msecs); + + test(data, 1023*967, 10.msecs, 32768, 0.msecs); + test(data, 32768, 0.msecs, 1023*967, 10.msecs); + + test(data, 1023*967, 10.msecs, 65535, 0.msecs); + test(data, 65535, 0.msecs, 1023*967, 10.msecs); +} + +void test(ubyte[] data, ulong read_sleep_freq, Duration read_sleep, + ulong write_sleep_freq, Duration write_sleep) +{ + test(data, ulong.max, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, data.length * 2, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, data.length / 2, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 64 * 1024, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 64 * 1024 - 57, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 557, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); +} + +void test(ubyte[] data, ulong chunk_limit, ulong read_sleep_freq, + Duration read_sleep, ulong write_sleep_freq, Duration write_sleep) +{ + test(data, ulong.max, chunk_limit, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 8 * 1024 * 1024, chunk_limit, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 8 * 1024 * 1024 - 37, chunk_limit, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); + test(data, 37, chunk_limit, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); +} + +void test(ubyte[] data, ulong data_limit, ulong chunk_limit, ulong read_sleep_freq, + Duration read_sleep, ulong write_sleep_freq, Duration write_sleep) +{ + import std.traits : EnumMembers; + foreach (m; EnumMembers!PipeMode) + test(data, m, data_limit, chunk_limit, read_sleep_freq, read_sleep, write_sleep_freq, write_sleep); +} + +void test(ubyte[] data, PipeMode mode, ulong data_limit, ulong chunk_limit, + ulong read_sleep_freq, Duration read_sleep, ulong write_sleep_freq, + Duration write_sleep) +{ + import vibe.core.log; + logInfo("test RF=%s RS=%sms WF=%s WS=%sms CL=%s DL=%s M=%s", read_sleep_freq, read_sleep.total!"msecs", write_sleep_freq, write_sleep.total!"msecs", chunk_limit, data_limit, mode); + + auto input = TestInputStream(data, chunk_limit, read_sleep_freq, read_sleep); + auto output = TestOutputStream(write_sleep_freq, write_sleep); + auto datacmp = data[0 .. min(data.length, data_limit)]; + + input.pipe(output, data_limit, mode); + if (output.m_data.data != datacmp) { + logError("MISMATCH: %s b vs. %s b ([%(%s, %) ... %(%s, %)] vs. [%(%s, %) ... %(%s, %)])", + output.m_data.data.length, datacmp.length, + output.m_data.data[0 .. 6], output.m_data.data[$-6 .. $], + datacmp[0 .. 6], datacmp[$-6 .. $]); + assert(false); + } + + // avoid leaking memory due to false pointers + output.freeData(); +} + + +struct TestInputStream { + private { + const(ubyte)[] m_data; + ulong m_chunkLimit = size_t.max; + ulong m_sleepFrequency = 0; + Duration m_sleepAmount; + } + + this(const(ubyte)[] data, ulong chunk_limit, ulong sleep_frequency, Duration sleep_amount) + { + m_data = data; + m_chunkLimit = chunk_limit; + m_sleepFrequency = sleep_frequency; + m_sleepAmount = sleep_amount; + } + + @safe: + + @property bool empty() @blocking { return m_data.length == 0; } + + @property ulong leastSize() @blocking { return min(m_data.length, m_chunkLimit); } + + @property bool dataAvailableForRead() { assert(false); } + + const(ubyte)[] peek() { assert(false); } // currently not used by pipe() + + size_t read(scope ubyte[] dst, IOMode mode) + @blocking { + assert(mode == IOMode.all || mode == IOMode.once); + if (mode == IOMode.once) + dst = dst[0 .. min($, m_chunkLimit)]; + auto oldsleeps = m_sleepFrequency ? m_data.length / m_sleepFrequency : 0; + dst[] = m_data[0 .. dst.length]; + m_data = m_data[dst.length .. $]; + auto newsleeps = m_sleepFrequency ? m_data.length / m_sleepFrequency : 0; + if (oldsleeps != newsleeps) { + if (m_sleepAmount > 0.msecs) + sleep(m_sleepAmount * (oldsleeps - newsleeps)); + else yield(); + } + return dst.length; + } + void read(scope ubyte[] dst) @blocking { auto n = read(dst, IOMode.all); assert(n == dst.length); } +} + +mixin validateInputStream!TestInputStream; + +struct TestOutputStream { + private { + Appender!(ubyte[]) m_data; + ulong m_sleepFrequency = 0; + Duration m_sleepAmount; + } + + this(ulong sleep_frequency, Duration sleep_amount) + { + m_data = appender!(ubyte[]); + m_data.reserve(2*1024*1024); + m_sleepFrequency = sleep_frequency; + m_sleepAmount = sleep_amount; + } + + void freeData() + { + import core.memory : GC; + auto d = m_data.data; + m_data.clear(); + GC.free(d.ptr); + } + + @safe: + + void finalize() @safe @blocking {} + void flush() @safe @blocking {} + + size_t write(in ubyte[] bytes, IOMode mode) + @safe @blocking { + assert(mode == IOMode.all); + + auto oldsleeps = m_sleepFrequency ? m_data.data.length / m_sleepFrequency : 0; + m_data.put(bytes); + auto newsleeps = m_sleepFrequency ? m_data.data.length / m_sleepFrequency : 0; + if (oldsleeps != newsleeps) { + if (m_sleepAmount > 0.msecs) + sleep(m_sleepAmount * (newsleeps - oldsleeps)); + else yield(); + } + return bytes.length; + } + void write(in ubyte[] bytes) @blocking { auto n = write(bytes, IOMode.all); assert(n == bytes.length); } + void write(in char[] bytes) @blocking { write(cast(const(ubyte)[])bytes); } +} + +mixin validateOutputStream!TestOutputStream;