A coroutine library for running parallel tasks on a single thread using Iterators. Alternative to using async/await or Task.Run but without needing locks or concurrent collections.
StreamThreads is a coroutine library that allows you to write clean and plain code that can execute processes in parallel, but using only a single thread. It is an alternative to async/await and Task.Run(), but without locks, concurrent collections and synchronization. In some sense, it is more like a game-loop where each object in the scene needs updating every few milliseconds before the screen refreshes. Unfortunately, game-loops often end up with significant amounts of global status variables and case-statements in complex scenes. StreamThreads helps by allowing a game-loop to be written as a "multi-threaded" application, where each function is executing independently.
Starting coroutines
Coroutines can be started in the foreground (normal serial excution) or in the background. Each function can spawn as many background "threads" as it wants, and these will terminate as soon as the function goes out of scope, either by simply exiting or calling yield break.
StreamThreads uses two methods for starting coroutines:
Await() - in the foreground (serial execution)
Background() - in the background (parallel execution)
Below is an example of a function that executes a function (PrintDots) in the background(with the extension .Background()), while executing GetReady() synchroneously (.Await()). In the mean time, if at any point an error happens, whether it is inside the background function or not, the HandleFault (.OnError()) will be called.
Notice how the PrintDots function loops infinitely, and the yield return. This allows the "game-loop" to return and process some of the other running tasks.
StreamThreads is based on Iterators and Extension Methods. As such, yield return is essentially used every time a new function is called - either as a background worker thread, or inline with statements. It is also worth noting that all functions should have a return type of IEnumerable<StreamState>. This allows for it to be interpreted as an iterator by the compiler.
The extension methods also allow for easy ways to control what happens to a function when certain conditions arise. This can be used both with background and synchroneous functions.
RestartOnError causes the function being called to start over. This can be helpful if we know an error can be fixed by re-initializing some local variables at the top of the function. ResumeOnError simply ignores the error and continues the loop. This could cause some problematic overhead if the error is severe.
SwitchOnCondition causes the function to exit entirely and start a new function. The calling function (parent function) will have no knowledge of this. This can be viewed as "switching state". Since background threads are local to each function, they are all cancelled when this happens. This also includes error handlers.
async / await
It is also possible to call async functions, as if they were part of the normal StreamThreads execution. Be aware, that these calls will not be terminated when they go out of scope. They live forever, or at least until they stop running by themselves. However, they do run in a separate thread, and so will be "true" multi-threading. Again, they can be called either as Background() or Await(), but lack the ability to use Until(), While() or any other iterator based chaining.
Alternatively, there is also a Lambda version available.
yield return Background(c => AnAsyncProcess(c), () => cancel);
private void AnAsyncProcess(CancellationToken token)
{
... do some time consuming stuff
if (token.IsCancellationRequested)
{
return;
}
}
Calling the loop
Finally we have the main loop that makes it all happen. From your form, on a timer, or just straight in a loop as below, define a StreamState variable to your function and add Await(). Then put a call to Loop every few milliseconds, or however frequent you want your loop to run. Notice how Loop somewhat counter-intuitively returns true when it should no longer be called.
using StreamThreads;
using static StreamThreads.StreamExtensions;
StreamState state = DoSomething().Await();
while (true)
{
if(state.Loop()) break;
SecondsSinceLast = 0;
Thread.Sleep(10);
}
Timing
Lastly, the SecondsSinceLast is a [ThreadStatic] property that makes it easier during varying loop times to size time-dependent calculations.
using static StreamThreads.StreamExtensions;
double Delta = SecondsSinceLast * speed;
Don't forget to include the using static statement if you want to use SecondsSinceLast, WaitForever, OK, WaitFor and other static properties and methods from the StreamThreads library.
Return values
Return variables and ref parameters are not allowed for iterators, so as a solution StreamThreads contains a small wrapper class IteratorReturnVariable, that can be passed as a parameter and populated by the called function.
IteratorReturnVariable<int> myrefvar = new ();
yield return DoSomething(myrefvar).Await();
Console.WriteLine(myrefvar);
It is also possible for a function to return a value.
StreamThreads includes generic versions of Await() and Background() which allows for easier type casting.
...
yield return ReturnNumbers().Background(out var ret);
...
private IEnumerable<StreamState<int>> ReturnNumbers()
{
var rnd = new Random();
while (true)
{
yield return Return(rnd.Next(0, 100));
}
}
Notice the IEnumerable<StreamState<int>>.
Return variable as status monitor
In addition to the return value itself, the IteratorReturnVariable contains a few fields for monitoring a background tasks status, such as HasEnded, WasTerminated, IsRunning and Faulted.
This screen shows the sample program running a number of threads simultaneously, where some threads print a different character, others change the color, and a few print out computational results.
This image is from the WPF sample of recursive coroutines, showing small boxes slowly spawning on the screen, each doing their own thing.
