2
\$\begingroup\$

So, I've been reading about monads and I wanted to see if I could implement a system for asynchronous computation in a monadic way.
I came up with two solutions:

The first one spawns a thread for each function bound, and this thread waits for the previous thread to finish. Should this bother me? These threads aren't doing anything until they have something to do so I'm not putting big loads on the computer.

using System;
using System.Threading;

namespace Philistine
{
    public static class Task
    {
        public static Task<T> Unit<T>(T value) => new Task<T>(value);
        public static Task<T> Do<T>(Func<T> computation) => new Task<T>(computation);
    }

    public class Task<T>
    {
        private T result;
        private readonly Thread thread;
        public bool IsDone => thread == null || !thread.IsAlive;

        public Task(T value)
        {
            result = value;
        }

        public Task(Func<T> computation)
        {
            thread = new Thread(() => result = computation());
            thread.Start();
        }

        private void SetResult(T t)
        {
            result = t;
        }

        private T Wait()
        {
            if (!IsDone)
                thread.Join();
            return result;
        }

        public Task<TR> Bind<TR>(Func<T, Task<TR>> function)
        {
            if (IsDone)
            {
                return function(result);
            }
            else
            {
                return Task.Do(
                    () =>
                    {
                        var other = function(Wait());
                        return other.Wait();
                    });
            }
        }

        public Task<TR> Bind<TR>(Func<T, TR> function)
        {
            return Task.Do(() => function(Wait()));
        }

        public void Bind(Action<T> action)
        {
            new Thread(() => action(Wait())).Start();
        }
    }
}

The alternative is to try and have linear task chains be executed on a single thread, and try to make use of a thread pool. What bothers me here is that the implementation is quite a bit more complex. The Wait method is implemented like it is because since I'm using ThreadPool I don't have access to Thread.Join and I felt that adding EventWaitHandles would add more overhead than speedup.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace Philistine
{
    public static class Task
    {
        public static ITask<T> Unit<T>(T value) => new UnitTask<T>(value);
        public static ITask<T> Do<T>(Func<T> computation) => new RootTask<T>(computation);
    }

    public interface ITaskIn<in TIn>
    {
        void Run(TIn parameter);
        void Invoke(TIn parameter);
    }

    public interface ITask<out TOut>
    {
        bool IsDone { get; }
        ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, ITask<TOtherOut>> function);
        ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, TOtherOut> function);
        void Bind(Action<TOut> action);
        TOut Wait();
    }

    public class UnitTask<TOut> : ITask<TOut>
    {
        private readonly TOut result;
        public bool IsDone => true;

        public UnitTask(TOut value)
        {
            result = value;
        }

        public ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, ITask<TOtherOut>> function)
        {
            return function(result);
        }

        public ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, TOtherOut> function)
        {
            return Task.Do(() => function(result));
        }

        public void Bind(Action<TOut> action)
        {
            ThreadPool.QueueUserWorkItem(_ => action(result));
        }

        public TOut Wait() => result;
    }

    public class RootTask<TOut> : ITask<TOut>
    {
        protected TOut result;
        private List<ITaskIn<TOut>> nextTasks;
        public bool IsDone { get; private set; }

        protected RootTask() { }

        public RootTask(Func<TOut> computation)
        {
            ThreadPool.QueueUserWorkItem(
                _ =>
                {
                    result = computation();
                    Continue();
                });
        }

        protected void Continue()
        {
            lock (this)
                IsDone = true;
            if (nextTasks != null)
            {
                foreach (var taskIn in nextTasks.Skip(1))
                    taskIn.Run(result);
                nextTasks.First().Invoke(result);
            }
        }

        public ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, ITask<TOtherOut>> function)
        {
            Monitor.Enter(this);
            if (IsDone)
            {
                Monitor.Exit(this);
                return function(result);
            }
            else
            {
                var continuation = ContinueWith(function);
                Monitor.Exit(this);
                return continuation;
            }
        }

        public ITask<TOtherOut> Bind<TOtherOut>(Func<TOut, TOtherOut> function)
        {
            Monitor.Enter(this);
            if (IsDone)
            {
                Monitor.Exit(this);
                return Task.Do(() => function(result));
            }
            else
            {
                var continuation = ContinueWith(function);
                Monitor.Exit(this);
                return continuation;
            }
        }

        public void Bind(Action<TOut> action)
        {
            Monitor.Enter(this);
            if (IsDone)
            {
                Monitor.Exit(this);
                ThreadPool.QueueUserWorkItem(_ => action(result));
            }
            else
            {
                ContinueWith(action);
                Monitor.Exit(this);
            }
        }

        private ITask<TOtherOut> ContinueWith<TOtherOut>(Func<TOut, ITask<TOtherOut>> function)
        {
            TOtherOut Unwrapped(TOut tIn) => function(tIn).Wait();
            var task = new Task<TOut, TOtherOut>(Unwrapped);
            ContinueWithTask(task);
            return task;
        }

        private ITask<TOtherOut> ContinueWith<TOtherOut>(Func<TOut, TOtherOut> function)
        {
            var task = new Task<TOut, TOtherOut>(function);
            ContinueWithTask(task);
            return task;
        }

        private void ContinueWith(Action<TOut> action)
        {
            var task = new EndTask<TOut>(action);
            if (nextTasks == null)
                nextTasks = new List<ITaskIn<TOut>>();
            nextTasks.Add(task);
        }

        private void ContinueWithTask(ITaskIn<TOut> task)
        {
            if (nextTasks == null)
                nextTasks = new List<ITaskIn<TOut>>();
            nextTasks.Add(task);
        }

        public TOut Wait()
        {
            while (!IsDone)
                Thread.Sleep(10);
            return result;
        }
    }

    public class EndTask<TIn> : ITaskIn<TIn>
    {
        private readonly Action<TIn> action;

        public EndTask(Action<TIn> action)
        {
            this.action = action;
        }

        public void Run(TIn parameter)
        {
            ThreadPool.QueueUserWorkItem(_ => action(parameter));
        }

        public void Invoke(TIn parameter)
        {
            action(parameter);
        }
    }

    public class Task<TIn, TOut> : RootTask<TOut>, ITaskIn<TIn>
    {
        private readonly Func<TIn, TOut> function;

        public Task(Func<TIn, TOut> function)
        {
            this.function = function;
        }

        public void Run(TIn parameter)
        {
            ThreadPool.QueueUserWorkItem(
                _ =>
                {
                    result = function(parameter);
                    Continue();
                });
        }

        public void Invoke(TIn parameter)
        {
            result = function(parameter);
            Continue();
        }
    }
}

Both implementations are used like this:

var five = Task.Do(
    () =>
    {
        Thread.Sleep(2000);
        return 5;
    });
var seven = five.Bind(i => i + 2);
var fourteen = seven.Bind(
    i =>
    {
        Thread.Sleep(2000);
        return i * 2;
    });
var treeFiddy = seven.Bind(
    i =>
    {
        Thread.Sleep(2000);
        return Task.Unit(i / 2.0);
    });

fourteen.Bind(Console.Out.WriteLine);
treeFiddy.Bind(Console.Out.WriteLine);

Actual question:
Since I don't have any realworld experience with heavily multi-threaded programs, and I don't have solid understanding of the inner workings of threads, I'd like some help in assessing the costs and benefits of both approaches (performance-wise). Would the ideal solution be some combination of both? The first solution, but with a pool? The second solution but without a pool? Better implementation of Wait?

P. S. Please let me know if some parts of the code could use documentation. I've been staring at it for a while now, so in my eyes it all makes sense and I don't like to add unnecessary comments (I think too much verbosity makes it harder to read).

\$\endgroup\$
  • 1
    \$\begingroup\$ Could you elaborate one what this code allows which the TPL and System.Threading.Task<T>.ContinueWith (something like this) does not? I feel I'm missing something. \$\endgroup\$ – VisualMelon Nov 2 '18 at 18:41
  • \$\begingroup\$ Regarding comments, we can all agree that too many comments is a bad thing, but inline documentation (///) on (public) methods is generally a good thing (so long as they are correct and you don't change the interface). Describing what a method is meant to do is a great why to make sure that you understand it, make it accessible to those consuming the API, and indeed helps to prevent inadvertently changing its behaviour in future (because its documented inline, so the programmer can actually see it). \$\endgroup\$ – VisualMelon Nov 2 '18 at 18:48
  • 1
    \$\begingroup\$ @VisualMelon your link leads me to a Hello World program. But yes, C#'s async/await and Tasks are monadic in nature. I'm trying to explore and learn. That's why I did this. You are correct when it comes to comments as well. Do you want me to add comments here? Would it help you understand things better? Something irks me to keep the line count as low as possible when posting code inline. \$\endgroup\$ – Markonius Nov 2 '18 at 21:12
  • \$\begingroup\$ Sorry, should have checked the link! It was meant to be this. If it's just for the sake of experiment then your call. \$\endgroup\$ – VisualMelon Nov 2 '18 at 21:17
  • \$\begingroup\$ Yeah, that link does pretty much what I wanted to achieve :D Anyway, my primary question still stands. Perhaps it's better suited for stack overflow. \$\endgroup\$ – Markonius Nov 3 '18 at 18:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.