Disposing sequence of resources with Reactive Extensions

Recall my previous post on Disposing sequence of resources where we were solving imperatively the following problems:

• Create single disposable representation for a sequence of disposable resources
• Defer resources allocation to avoid exception propagation until cleanup can be guaranteed and avoid unnecessary (allocated resources that aren’t used) allocations
• Dispose only consumed resources and preserve nested try{}finally{} blocks semantics (resources that are allocated first from sequence order perspective disposed last; any exception thrown from a finally block hides exception being propagated)

Now with Reactive Extensions for .NET (Rx) is out we will do it in more LINQ-ish manner with the help of interactive features of Reactive Extensions:

• EnumerableEx.Publish – publishes the values of source to each use of the bound parameter.
• EnumerableEx.Share - shares cursor of all enumerators to the sequence.
• EnumerableEx.Finally – invokes finallyAction after source sequence terminates normally or by an exception.

A great explanation of how EnumerableEx.Publish works (disregard naming difference) is given in Taming Your Sequence’s Side-Effects Through IEnumerable.Let. The following example illustrates the point.

static Random m_seeder = new Random();

static IEnumerable<int> GenerateRandomSequence(int count)
{
  var rnd = new Random(m_seeder.Next());
  for(int i = 0; i < count; i++)
  {
    yield return rnd.Next(count);
  }
}

...

const int count = 5;
var xs = GenerateRandomSequence(count);
// Each we iterate xs we may get a different sequence 
var equals = xs.SequenceEqual(xs);
// However it can be solved through memoization
xs.Publish(seq =>
             {
               // Every time we iterate seq we get the same 
               // sequence
               equals = seq.SequenceEqual(seq);
               return seq;
             });

EnumerableEx.Share makes sure that any iteration is made with respect to the same cursor.

var xs = Enumerable.Range(0, count);
// Prints every sequence element to console
// Without sharing for each of count iterations it will print 
// first element of a potentially different sequence (recall 
// random sequence example) 
var shared = xs.Share(); 
for(int i = 0; i < count; i++)
{
  shared.Take(1).Run(Console.WriteLine);
}

EnumerableEx.Finally does exactly what its description says (see more details here).

static IEnumerable<int> GenerateThrowingSequence(int count)
{
  for(int i = 0; i < count; i++)
  {
    if (i > 0 && i % 3 == 0)
    {
      throw new Exception();
    }
    yield return i;
  }
}

...

// Prints 0, 1, 2, Finally, Caught
try
{
  GenerateThrowingSequence(count).Finally(() => Console.WriteLine("Finally"))
    .Run(Console.WriteLine);
}
catch (Exception)
{
  Console.WriteLine("Caught");
}

// Prints 0, 1, Finally
GenerateThrowingSequence(count).Finally(() => Console.WriteLine("Finally"))
  .Take(2).Run(Console.WriteLine);

Now putting everything together. Publish will help us to defer resources allocation and avoid unnecessary allocations. Share and Finally will take care of disposal.

static class Disposables
{
  // Disposes projected resources once they are no longer needed
  public static void Using<TSource, TResource>(
    // Source sequence projected to disposable resources
    this IEnumerable<TSource> source,
    // Resource projection
    Func<TSource, TResource> resourceSelector,
    // Resources usage action
    Action<IEnumerable<TResource>> resourcesUsage)
      where TResource : IDisposable
  {
    var rcount = 0;
    source
      // At this point resources are not created but 
      // only projection is set
      .Select(
      s =>
        {
          // As we do not want to unnecessarily create 
          // and then immediately dispose potentially expensive
          // resources we will count created resources
          // and later dispose only used ones
          rcount++;
          return resourceSelector(s);
        })
      .Publish(
      rs =>
        {
          // During sequence iteration resources will be created
          // However not all resources may be iterated through or 
          // an exception may be thrown in the middle and thus 
          // not all resources may be created (therefore not 
          // disposed)
          try
          {
            // Supplied resources sequence can be iterated 
            // multiple times with each of side effects occurs 
            // only once and sequence elements memoized and 
            // reused during each iteration
            resourcesUsage(rs);
            return Enumerable.Empty<TResource>();
          }
          finally
          {
            // We must dispose only those resources we used
            // (counted and memoized above during first 
            // iteration)
            rs = rs.Take(rcount)
              // Disposing resources must be done in the opposite 
              // order to preserve nested try{}finally{} blocks 
              // semantics
              .Reverse().Do(r =>
                              {
                                rcount--;
                                r.Dispose();
                              })
              // Once resource is disposed it must not be 
              // iterated again and this what Share takes 
              // care of
              .Share();

            Action final = null;
            final = () =>
                      {
                        // Stop once every resource was given 
                        // a chance to dispose as Finally is 
                        // called even on empty sequences and 
                        // otherwise it leads to stack overflow
                        if (rcount > 0)
                        {
                          // Dispose only used resources and 
                          // leave untouched the rest
                          rs.Finally(final).Run();
                        }
                      };
            final();
          }
        })
      // Evaluate the sequence (triggers resources usage)
      .Run();
  }
}

Usage example below illustrates situation where during resource disposal an exception is thrown. In this case we must give chance to preceding (from resource sequence order perspective) resource to be disposed. However if an exception is thrown while disposing preceding resources that exception will hide previous one.

// Fake full of side effects resource =)
class Resource : IDisposable
{
  private readonly int m_i;

  public Resource(int i)
  {
    m_i = i;
    Console.WriteLine("Created {0}", m_i);
  }

  public void Use()
  {
    Console.WriteLine("Using {0}", m_i);
  }

  public void Dispose()
  {
    Console.WriteLine("Disposed {0}", m_i);
    // Simulate resource disposal that results in exception
    if (m_i % 2 == 1)
    {
      throw new Exception(m_i.ToString());
    }
  }
}

...

try
{
  Enumerable.Range(0, 5)
    .Using(i => new Resource(i),
           rs =>
             {
               // First resources 0, 1 and 2 are created 
               // and used
               rs.Take(3).Run(r => r.Use());
               // then already created resource 2 is used 
               // and resource 3 is created and used
               rs.Skip(1).Take(3).Run(r => r.Use());
             });
}
catch (Exception ex)
{
  // As resources are disposed in the opposite order
  // the latest exception is propagated
  Console.WriteLine("Exception {0}", ex.Message);
}

This produces the following output:

Created 0 // iterating, if not iterated previously resource is created
Using 0
Created 1
Using 1
Created 2
Using 2
Using 1   // otherwise reused
Using 2   // reused again
Created 3 // wasn’t iterated previously, created
Using 3
Disposed 3 // disposing in the opposite order, throws exception
Disposed 2 // still disposing continues
Disposed 1 // throws exception that hides exception thrown earlier
Disposed 0 // disposing continues
Exception 1 // exception is caught

That’s it! Hopefully you’ve enjoyed.

I hope we’ll meet next year. Happy New Year!

Chaining responsibilities

The idea behind Chain of Responsibility pattern is quite simple and powerful:

Avoid coupling the sender of a request to its receiver by giving more than one object a chance to handle the request. Chain the receiving objects and pass the request along the chain until an object handles it.

You can find lots of object oriented implementations out there so as an exercise we will rather try to do it in a more functional way. For simplicity Func<T, R> will be considered as handler contract. The basic idea looks like this:

Func<T, R> h = t =>
   {
       // Decide whether you can handle request
       bool canHandle = ...;
       // Get successor from somewhere
       Func<T, R> successor = ...;
       if (canHandle)
           // Handle request represented by t
       else
           // Delegate request to successor
           return successor(t);
   };

The first thing to solve is how to get successor. As handler must support composition it cannot simply create closure over successor. On the other hand it can be represented as function that returns actual handler closed over its successor:

Func<Func<T, R>, Func<T, R>> h = successor => t => 
   {
       bool canHandle = ...;
       if (canHandle)
           // Handle request represented by t
       else
           // Delegate request to closed over successor
           return successor(t);
   };

Now we need to compose handlers into a chain.

// Creates chain of responsibility out of handlers
static Func<T, R> Chain<T, R>(IEnumerable<Func<Func<T, R>, Func<T, R>>> handlers)
{
    // By default if none of handlers can handle incoming request an exception is thrown
    Func<T, R> notSupported = t => { throw new NotSupportedException(); };
    return Chain(handlers, notSupported);
}

// Creates chain of responsibility out of regular and default handlers
static Func<T, R> Chain<T, R>(IEnumerable<Func<Func<T, R>, Func<T, R>>> handlers, Func<T, R> def)
{
    // Assuming that order of handlers within the chains must be the same as in handlers sequence 
    return handlers
        // Handlers needs to be reversed first or otherwise they will appear in the opposite order 
        .Reverse()
        // Iteratively close each handler over its successor
        .Aggregate(def, (a, f) => f(a));
}

To make it more clear lets expand chaining of simple two handlers case:

// default handler
Func<int, int> d = x => x;
// two handlers appear in sequence in order of declaration
Func<Func<int, int>, Func<int, int>> h1 = s => t => t < 10 ? t*2 : s(t);
Func<Func<int, int>, Func<int, int>> h2 = s => t => t < 5 ? t + 3 : s(t);

// 1. Reverse handlers
// h2
// h1

// 2. Close h2 over d
// tmp1 = t => t < 10 ? t * 2 : d(t);
Func<int, int> tmp1 = h2(d); 

// 3. Close h1 over tmp1
// tmp2 = t => t < 5 ? t + 3 : tmp1(t);
Func<int, int> tmp2 = h1(tmp1); 

// 4. tmp2 is the chain

Now handlers are dynamically composed into chains to address particular scenario. 

As a chaining exercise let’s create the following application (a team of developers tries to handle a project):

• Project is divided into a number of task of complexity that doesn’t exceed particular threshold.
• In order to handle the project development team is staffed. A developer with skill X can handle task of complexity C when C <= X otherwise he contributes to task completion making task’s complexity smaller by X and delegates the rest of the task to next developer. Completed task is linked to developer who completed it.
• If the team cannot handle particular task they ask for help for an external expert.

Prepare

// Staff development team that will do the project
static IEnumerable<Func<Func<int, int>, Func<int, int>>> Staff(int teamSize, int maxSkill)
{
    var rnd = new Random();
    for (int i = 0; i < teamSize; i++)
    {
        int dev = i;
        // Developers may differ in their skills
        int skill = rnd.Next(maxSkill);
        // If developer can handle incoming task he reports by returning his id that he completed the task
        // If not (not enough skills) he contributes to task and delegates to next developer smaller task
        yield return c => t => t <= skill ? dev : c(t - skill);
    }
}

// Create work break down structure for the project
static IEnumerable<int> Work(int projectSize, int maxTaskComplexity)
{
    var rnd = new Random();
    for (int i = 0; i < projectSize; i++)
    {
        yield return rnd.Next(maxTaskComplexity) + 1;
    }
}

and march to the end.

// Create tasks
var work = Work(projectSize, maxTaskComplexity).ToArray();
// If the team cannot handle particular task they ask for help unknown guru
Func<int, int> askForHelp = t => -1;
// Create chain out of developers to form a team with a backup
var team = Chain(Staff(teamSize, maxTaskComplexity), askForHelp);
// Hand out each task to the team
var project = from w in work
              select new {Task = w, CompletedBy = team(w)};

foreach(var status in project)
{
    Console.WriteLine("Task {0} completed by {1}", status.Task, status.CompletedBy);
}

Have chaining fun!

Making callbacks more explicit

Recall my previous post Events and Callbacks vs. Explicit Calls that outlined pros and cons of both (events and callbacks on one side and explicit calls on the other side) approaches.

Explicit calls imply that callee plays certain role from caller’s perspective thus making the relationship between the two explicit as well. Consider simple undo/redo example:

interface ICommand
{
    // Not every command can be undone. Returns true if it can be undone,
    // false - otherwise.
    bool CanUndo { get; }
    // Executes the command.
    void Do();
    // Reverts changes. NotSupportedException must be thrown in case 
    // command cannot be undone.
    void UnDo();
}

// Provides undo/redo mechanism.
class History
{
    // Makes supplied command part of the history.
    public void Remember(ICommand cmd)
    {
        // implementation is not relevant.
    }

    // other members elided.
}

Caller (History) clearly communicates its expectations by requiring callee (particular command) to conform with command’s contract (indivisible logical unit).

While being explicit makes code more clear it has its price. It requires consumers to create new types that conform to contract. It is not a problem if the type created will be reused by other parts of the application or it has complex logic. But it is also common that it may have trivial logic that is used only in caller (undo/redo mechanism in this case) scenario. In this case creating a new type sounds like an overhead.

I whish C# has capabilities (“Object Expressions”) similar to F# where I can implement interface “in place” like this:

let cmd = 
    { 
        new ICommand with 
            member x.CanUndo 
                with get() = false
            member x.Do() = Console.Write("Done") 
            member x.UnDo() = raise (new NotSupportedException())
    }

Although we can provide default implementation that uses delegates.

class ActionCommand : ICommand
{
    private readonly Action m_do;
    private readonly Action m_undo;

    public ActionCommand(Action doCallback, Action undoCallback)
    {
        if (doCallback == null)
        {
            throw new ArgumentNullException("doCallback");
        }
        m_do = doCallback;
        m_undo = undoCallback;
    }

    public bool CanUndo
    {
        get { return m_undo != null; }
    }

    public void Do()
    {
        m_do();
    }

    public void UnDo()
    {
        if (!CanUndo)
        {
            throw new NotSupportedException();
        }
        m_undo();
    }
}

While conforming to contract ActionCommand eases creation of lightweight scenario dedicated implementations “in place” avoiding types proliferation. But still the type must be discovered first by developers. It is negligible effort but it is still nonzero. In order to level this effort let the original consuming code do the job.

// Provides undo/redo mechanism
class History
{
    // Makes supplied command part of the history
    public void Remember(ICommand cmd)
    {
        // implementation is not relevant
    }

    // Makes command represented by pair of callbacks part of the history
    public void Remember(Action doCallback, Action undoCallback)
    {
        Remember(new ActionCommand(doCallback, undoCallback));
    }

    // other members elided
}

You should not put every possible “shortcut” into overloads but only the most commonly used one.

What benefits does this approach has? It benefits from being close to explicitly defined role making it easier to understand and use callback based API that is useful in case of lightweight single scenario use implementations.

Summary:

• CONSIDER creating callback based overload next to overload that consumes corresponding abstraction