using System.Linq; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using Wabbajack.Common.CSP; namespace Wabbajack.Test.CSP { [TestClass] public class CSPTests { public TestContext TestContext { get; set; } public void Log(string msg) { TestContext.WriteLine(msg); } [TestInitialize] public void Startup() { } /// /// Test that we can put a value onto a channel without a buffer, and that the put is released once the /// take finalizes /// /// [TestMethod] public async Task TestTakePutBlocking() { var channel = Channel.Create(); var ptask = channel.Put(1); var (open, val) = await channel.Take(); Assert.AreEqual(1, val); Assert.IsTrue(open); Assert.IsTrue(await ptask); } /// /// If we create a channel with a fixed buffer size, we can enqueue that number of items without blocking /// We can then take those items later on. /// /// [TestMethod] public async Task TestTakePutBuffered() { var channel = Channel.Create(10); foreach (var itm in Enumerable.Range(0, 10)) await channel.Put(itm); foreach (var itm in Enumerable.Range(0, 10)) { var (is_open, val) = await channel.Take(); Assert.AreEqual(itm, val); Assert.IsTrue(is_open); } } /// /// We can convert a IEnumerable into a channel by inlining the enumerable into the channel's buffer. /// /// [TestMethod] public async Task TestToChannel() { var channel = Enumerable.Range(0, 10).ToChannel(); foreach (var itm in Enumerable.Range(0, 10)) { var (is_open, val) = await channel.Take(); Assert.AreEqual(itm, val); Assert.IsTrue(is_open); } } /// /// TakeAll will continue to take from a channel as long as the channel is open. Once the channel closes /// TakeAll returns a list of the items taken. /// /// [TestMethod] public async Task TestTakeAll() { var results = await Enumerable.Range(0, 10).ToChannel().TakeAll(); CollectionAssert.AreEqual(Enumerable.Range(0, 10).ToList(), results); } /// /// We can add Rx transforms as transforms inside a channel. This allows for cheap conversion and calcuation /// to be performed in a channel without incuring the dispatch overhead of swapping values between threads. /// These calculations happen inside the channel's lock, however, so be sure to keep these operations relatively /// cheap. /// /// [TestMethod] public async Task RxTransformInChannel() { var chan = Channel.Create(1, o => o.Select(v => v + 1)); var finished = Enumerable.Range(0, 10).OntoChannel(chan); foreach (var itm in Enumerable.Range(0, 10)) { var (is_open, val) = await chan.Take(); Assert.AreEqual(itm + 1, val); Assert.IsTrue(is_open); } await finished; } [TestMethod] public async Task UnorderedPipeline() { var o = Channel.Create(3); var finished = Enumerable.Range(0, 3) .ToChannel() .UnorderedPipeline(1, o, obs => obs.Select(itm => itm.ToString())); var results = (await o.TakeAll()).OrderBy(e => e).ToList(); var expected = Enumerable.Range(0, 3).Select(i => i.ToString()).OrderBy(e => e).ToList(); CollectionAssert.AreEqual(expected, results); } [TestMethod] public async Task UnorderedPipelineWithParallelism() { // Do it a hundred times to try and catch rare deadlocks var o = Channel.Create(3); var finished = Enumerable.Range(0, 1024) .ToChannel() .UnorderedPipeline(4, o, obs => obs.Select(itm => itm.ToString())); var results = (await o.TakeAll()).OrderBy(e => e).ToList(); var expected = Enumerable.Range(0, 1024).Select(i => i.ToString()).OrderBy(e => e).ToList(); CollectionAssert.AreEqual(expected, results); await finished; } [TestMethod] public async Task UnorderedTaskPipeline() { // Do it a hundred times to try and catch rare deadlocks var o = Channel.Create(3); var finished = Enumerable.Range(0, 1024) .ToChannel() .UnorderedPipeline(4, o, async v => { await Task.Delay(1); return v; }); var results = (await o.TakeAll()).OrderBy(e => e).ToList(); var expected = Enumerable.Range(0, 1024).ToList(); CollectionAssert.AreEqual(expected, results); await finished; } [TestMethod] public async Task UnorderedThreadPipeline() { // Do it a hundred times to try and catch rare deadlocks var o = Channel.Create(3); var finished = Enumerable.Range(0, 1024) .ToChannel() .UnorderedThreadedPipeline(4, o, v => { Thread.Sleep(1); return v; }); var results = (await o.TakeAll()).OrderBy(e => e).ToList(); var expected = Enumerable.Range(0, 1024).ToList(); CollectionAssert.AreEqual(expected, results); await finished; } [TestMethod] public async Task ChannelStressTest() { var chan = Channel.Create(); var putter = Task.Run(async () => { for (var i = 0; i < 1000; i++) { var result = await chan.Put(i); } }); var taker = Task.Run(async () => { try { for (var i = 0; i < 1000; i++) { var (is_open, val) = await chan.Take(); Assert.AreEqual(i, val); } } finally { chan.Close(); } }); await putter; await taker; } [TestMethod] public async Task ChannelStressWithBuffer() { var chan = Channel.Create(1); var putter = Task.Run(async () => { for (var i = 0; i < 1000; i++) { await chan.Put(i); } }); var taker = Task.Run(async () => { try { for (var i = 0; i < 1000; i++) { var (is_open, val) = await chan.Take(); Assert.AreEqual(i, val); } } finally { chan.Close(); } }); await putter; await taker; } } }