MCS UK Solution Development Team

.Net Implementation of a Priority Queue (aka Heap)

MCS UK Solution Development — Thu, 10 May 2012 16:16:04 GMT

I thought I would take a break for a while from Hadoop and put together an F# .Net implementation of a Priority Queue; implemented using a heap data structure. Conceptually we can think of a heap as a balanced binary tree. The tree will have a root, and each node can have up to two children; a left and a right child. The keys in such a binary tree are said to be in heap order, such that any element is at least as large as its parent element.

For every element v, at a node i, the element w at i’s parent satisfies key(w)≤key(v)

The heap is maintained as an array, indexed by i=1…N, such that for any node i, the nodes at positions leftChild(i)=2i and rightChild(i)=2i+1, and the parent of the node is at position parent(i)=(1/2).

The idea behind a Priority Queue is that finding the element with the lowest key is easy, O(1) time, as it is always the root. When adding a new element it is always added to the end of the list. The heap is then fixed by recursively checking the new element with its parent, and swapping their positions in the list until the heap condition is satisfied. This operation takes O(log n) time.

Within .Net we implemented a Priority Queue using a List<KeyValuePair<'TKey, 'TValue>>.

The code, as always can be downloaded from:

http://code.msdn.microsoft.com/Net-Implementation-of-a-d3ac7b9d

The basic operations supported on the Priority Queue will be:

Enqueue	Adds an item to the Queue
Dequeue	Removes the root element; that with the lowest key
Peek	Queries the root element
IndexOf	Returns the index of a given key value pair
RemoveAt	Removes the element at the specified index value
ContainsKey	Determines whether the Queue contains a specific key value
ChangeKey	Changes a specific key value
RemoveKey	Removes the element with the specific key value
Item	An indexer based on the key value

Before showing the code a quick word is warranted about the implementation. The Key operations shown above rely on a second data structure; a Dictionary<'TKey, int>. This secondary data structure maintains a reference of the index within the list for each key. This positional structure does restrict the Priority Queue to unique key values, but does provide greater flexibility for the heap operations; all in O(1) time.

So what does the F# code look like:

namespace MSDN.FSharp
open System
open System.Collections
open System.Collections.Generic
type PriorityQueue<'TKey, 'TValue when 'TKey : comparison> private (data:IEnumerable<KeyValuePair<'TKey, 'TValue>> option, capacity:int, comparer:IComparer<'TKey>) =
let mutable heapList:List<KeyValuePair<'TKey, 'TValue>> = null
let mutable positionDict:Dictionary<'TKey, int> = null
// Determines if a value is in the heap
let inRange index =
if index >= 0 && index < heapList.Count then Some(index) else None
let checkIndex index =
if ((inRange index).IsNone) then raise (ArgumentException(sprintf "Index specified is not within range - %i" index))
// Gets the children of a node
let getChildren (index:int) =
// children left[2*pos] and right[2*pos + 1] where pos = index + 1
let left = (2 * index) + 1
let right = (2 * index) + 2
(inRange left, inRange right)
// Gets the parent of an index
let getParent (index:int) =
// parent index [pos/2] where index = pos - 1
if (index = 0) then None
else Some((index-1) / 2)
// Tests to see if the first value is greater than the first
let isGreater parent child =
if (comparer.Compare(heapList.[parent].Key, heapList.[child].Key) > 0) then true
else false
// Swaps two elements of the heap list
let swapElements idx1 idx2 =
let element1 = heapList.[idx1]
let element2 = heapList.[idx2]
heapList.[idx1] <- heapList.[idx2]
heapList.[idx2] <- element1
positionDict.[element1.Key] <- idx2
positionDict.[element2.Key] <- idx1
// Heapifys toward the parent
let rec heapifyUp (index:int) =
if (index > 0) then
let parent = getParent index
if (isGreater parent.Value index) then
swapElements parent.Value index
heapifyUp parent.Value
// Heapifys down to the children
let rec heapifyDown (index:int) =
let (left, right) = getChildren index
if (left.IsSome) then
let childindex =
if (right.IsSome && (isGreater left.Value right.Value)) then right.Value
else left.Value
if (isGreater index childindex) then
swapElements index childindex
heapifyDown childindex
// Heapifys down to the children
let heapifyUpDown (index:int) =
let parent = getParent index
if (parent.IsSome && (isGreater parent.Value index)) then
heapifyUp index
else
heapifyDown index
// Adds an items and heapifys
let insertItem (key:'TKey) (value:'TValue) =
let insertindex = heapList.Count
positionDict.Add(key, insertindex)
heapList.Add(new KeyValuePair<'TKey, 'TValue>(key, value))
heapifyUp(insertindex)
// Delete the root node and heapifys
let deleteItem index =
if (heapList.Count <= 1) then
heapList.Clear()
positionDict.Clear()
else
let lastindex = heapList.Count - 1
let indexKey = heapList.[index].Key
let lastKey = heapList.[lastindex].Key
heapList.[index] <- heapList.[lastindex]
positionDict.[lastKey] <- index
heapList.RemoveAt(lastindex)
positionDict.Remove(indexKey) |> ignore
heapifyDown index
// Default do bindings
do
if (comparer = null) then
raise (ArgumentException("Comparer cannot be null"))
let equalityComparer =
{ new IEqualityComparer<'TKey> with
override x.Equals(c1, c2) =
if ((comparer.Compare(c1, c2)) = 0) then true
else false
override x.GetHashCode(value) =
value.GetHashCode()
}
heapList <- new List<KeyValuePair<'TKey, 'TValue>>(capacity)
positionDict <- new Dictionary<'TKey, int>(capacity, equalityComparer)
if data.IsSome then
data.Value |> Seq.iter (fun item -> insertItem item.Key item.Value)
// Set of constructors
new() = PriorityQueue(None, 0, ComparisonIdentity.Structural<'TKey>)
new(capacity:int) = PriorityQueue(None, capacity, ComparisonIdentity.Structural<'TKey>)
new(data:IEnumerable<KeyValuePair<'TKey, 'TValue>>) = PriorityQueue(Some(data), 0, ComparisonIdentity.Structural<'TKey>)
new(comparer:IComparer<'TKey>) = PriorityQueue(None, 0, comparer)
new(capacity:int, comparer:IComparer<'TKey>) = PriorityQueue(None, capacity, comparer)
new(data:IEnumerable<KeyValuePair<'TKey, 'TValue>>, comparer:IComparer<'TKey>) = PriorityQueue(Some(data), 0, comparer)
// Checks to see if the heap is empty
member this.IsEmpty
with get() = (heapList.Count = 0)
// Enqueues a new entry into the heap
member this.Enqueue (key:'TKey) (value:'TValue) =
insertItem key value
()
// Peeks at the head of the heap
member this.Peek() =
if not(this.IsEmpty) then
heapList.[0]
else
raise (InvalidOperationException("Priority Queue is empty"))
// Dequeues the head entry into the heap
member this.Dequeue() =
let value = this.Peek()
deleteItem 0
value
// Determines whether an item is in the queue
member this.Contains (key:'TKey) (value:'TValue) =
heapList.Contains(new KeyValuePair<'TKey, 'TValue>(key, value))
// Returns the index of a specified item
member this.IndexOf (key:'TKey) (value:'TValue) =
heapList.IndexOf(new KeyValuePair<'TKey, 'TValue>(key, value))
// Removes an item from the queue at the specified index
member this.RemoveAt (index:int) =
checkIndex index
deleteItem index
// Determines whether an item is in the queue
member this.ContainsKey (key:'TKey) =
positionDict.ContainsKey key
// Determines whether an item is in the queue
member this.ChangeKey (key:'TKey) (value:'TKey)=
let index = positionDict.[key]
let item = heapList.[index]
heapList.[index] <- new KeyValuePair<'TKey, 'TValue>(value, item.Value)
positionDict.Remove(key) |> ignore
positionDict.Add(value, index)
heapifyUpDown index
// Removes an item from the queue for the specified key
member this.RemoveKey (key:'TKey) =
let index = positionDict.[key]
this.RemoveAt index
// Modifies elements based on index values
member this.Item
with get(key) =
let index = positionDict.[key]
heapList.[index]
and set(key) (value) =
let index = positionDict.[key]
heapList.[index] <- new KeyValuePair<'TKey, 'TValue>(key, value)
heapifyUpDown index
// Returns the count of the queue
member this.Count
with get() = heapList.Count
// Resets the capacity of the Queue
member this.TrimExcess() =
heapList.TrimExcess()
// Returns the capacity of the queue
member this.Capacity
with get() = heapList.Capacity
// Clears the queue
member this.Clear() =
heapList.Clear()
// Standard IList members
interface ICollection<KeyValuePair<'TKey, 'TValue>> with
member this.Add(item:KeyValuePair<'TKey, 'TValue>) =
this.Enqueue item.Key item.Value
member this.Clear() =
heapList.Clear()
member this.Contains(item:KeyValuePair<'TKey, 'TValue>) =
heapList.Contains(item)
member this.Count
with get() = heapList.Count
member this.CopyTo(toArray:KeyValuePair<'TKey, 'TValue>[], arrayIndex:int) =
heapList.CopyTo(toArray, arrayIndex)
member this.IsReadOnly
with get() = false
member this.Remove(item:KeyValuePair<'TKey, 'TValue>) =
let index = heapList.IndexOf(item)
if (inRange index).IsSome then
deleteItem index
true
else
false
member this.GetEnumerator() =
upcast heapList.GetEnumerator()
// IEnumerable GetEnumerator implementation
interface IEnumerable with
member this.GetEnumerator() =
upcast heapList.GetEnumerator()

This code implementation has the PriorityQueue derive from ICollection. The implementation is analogous to the .Net Queue implementation, but with Key and Value pairs; but more importantly where Dequeue is guaranteed to return the element with the lowest value.

The correctness of the heap is managed through the heapify operations. It is these recursive operations that compare parent and child nodes and adjust the positions accordingly.

One has to remember though that the enumerator will effectively return elements in a random order.

Written by Carl Nolan

ADFS 2.0: Single sign-on when a website references remote images

MCS UK Solution Development — Wed, 02 May 2012 16:25:00 GMT

I recently had an issue where a website, secured with Active Directory Federation Services 2.0 (ADFS), was referencing images stored in another website, secured by the same instance of ADFS. This meant that if a user logged into the main website without going first to the website hosting the images, they would not see the images. The error scenario is as follows:

The user logs into the main website and is redirected to ADFS to login. Once successfully logged in, the browser then holds a session cookie for that site

A page is then displayed on the main website that references images on a remote site via simple html

<img src="https://imageswebsite.dev/image1.jpg" alt="imagecontent" />

The image is not displayed on the page. This is because referencing an image does not cause the browser to be redirected to ADFS for authentication, and so no session cookie exists for the image site (as the host header differs)

N.B. - If the image site was secured using windows authentication, you would simply see the standard browser login prompt.

As both sites are secured using the same ADFS 2.0 instance, we want the user to simply log into the main site and achieve Single Sign On (SSO) across both sites. To this end we need some mechanism to bounce the browser against the images site, thus obtaining a session cookie.

This is a problem that must be resolved in the client browser and there are a few mechanisms to achieve this. e.g. - using a HTTP module to redirect the user to a page on the images site. The option I ended up choosing was iframe based, as this is capable of handling redirects (plus most browsers would cache the image once obtained, so performance impact would be minimal).

The methodology is as follows:

Set the image src to use a placeholder image whilst it is loading (so the user doesn’t see a image missing cross)

Add an iframe that has its src set to the actual image src

Once the iframe is loaded (and hence the browser has obtained a session cookie for the image site – this is invisible to the end user), set the src of the image to the correct path

Example html/javascript is as follows:

<html>
  <head>
    <title>Accenture Reach - Example iframe preloader</title>
    <script type="text/javascript" language="javascript">     
            function LoadImageWhenFrameHasLoaded() 
            {
                currentFrame = document.getElementById("frameid");
                if (currentFrame.readyState != "complete") 
                {
                    setTimeout("LoadImageWhenFrameHasLoaded();", 100);
                }
                else {
                   document.getElementById("imageContent").src = "https://imageswebsite.dev/image1.gif";
               }
           }
       
</script>     
      </head>
      <body>        
        <iframe src="https://imageswebsite.dev/image1.jpg" id="frameid" style="display: none"></iframe> 
        <script type="text/javascript" language="javascript">LoadImageWhenFrameHasLoaded();
</script>
    <img src="placeholder.gif" id="imageContent" alt="imagecontent" /> 
  </body>
</html>

Note that when implementing this in a production scenario, I did have a problem where the iframe says it has completed loading before a session cookie is obtained; this is because of the redirects. I ended up just waiting a period of time to overcome this, but please let me know if anyone can suggest an alternative.

Written by Rob Nowik

Generic based Framework for .Net Hadoop MapReduce Job Submission

MCS UK Solution Development — Wed, 02 May 2012 14:54:37 GMT

Over the past month I have been working on a framework to allow composition and submission of MapReduce jobs using .Net. I have put together two previous blog posts on this, so rather than put together a third on the latest change I thought I would create a final composite post. To understand why lets run through a quick version history of the code:

Initial release where the values are treated as strings, and serialization was handled through Object.ToString()
Made minor modifications to the submission APIs
Modified the Reducer and Combiner types to allow In-Reducer optimizations through the ability to yield a Tuple of the key and value in the Cleanup
Modified the Combiner and Reducer base classes such that data out of the mapper, in and out of the combiner, and in to the reducer uses a binary formatter; thus changing the base classes from strings to objects; meaning the classes can now cast to the expected type rather than performing string parsing
Added support for multiple mapper keys; with supporting utilities

The latest change takes advantage of the fact the objects are serialized in Binary format. This change has allowed for the base abstract classes to move away from object based APIs to one based on Generics. This change hopefully greatly simplifies the creation of .Net MapReduce jobs.

As always to submit MapReduce jobs one can use the following command line syntax:

MSDN.Hadoop.Submission.Console.exe -input "mobile/data/debug/sampledata.txt" -output "mobile/querytimes/debug"
-mapper "MSDN.Hadoop.MapReduceFSharp.MobilePhoneQueryMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.MobilePhoneQueryReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduce\Release\MSDN.Hadoop.MapReduceFSharp.dll"

The mapper and reducer parameters are .Net types that derive from a base Map and Reduce abstract classes shown below. The input, output, and files options are analogous to the standard Hadoop streaming submissions. The mapper and reducer options (more on a combiner option later) allow one to define a .Net type derived from the appropriate abstract base classes. Under the covers standard Hadoop Streaming is being used, where controlling executables are used to handle the StdIn and StdOut operations and activating the required .Net types. The “file” parameter is required to specify the DLL for the .Net type to be loaded at runtime, in addition to any other required files.

As always the source can be downloaded from:

http://code.msdn.microsoft.com/Framework-for-Composing-af656ef7

Mapper and Reducer Base Classes

The following definitions outline the abstract base classes from which one needs to derive. Lets start with the C# definitions:

C# Abstract Classes

namespace MSDN.Hadoop.MapReduceBase
{
[Serializable]
[AbstractClass]
public abstract class MapReduceBase<V2>
{
protected MapReduceBase();
public override IEnumerable<Tuple<string, V2>> Cleanup();
public override void Setup();
}
[Serializable]
[AbstractClass]
public abstract class MapperBaseText<V2> : MapReduceBase<V2>
{
protected MapperBaseText();
public abstract override IEnumerable<Tuple<string, V2>> Map(string value);
}
[Serializable]
[AbstractClass]
public abstract class MapperBaseXml<V2> : MapReduceBase<V2>
{
protected MapperBaseXml();
public abstract override IEnumerable<Tuple<string, V2>> Map(XElement element);
}
[Serializable]
[AbstractClass]
public abstract class MapperBaseBinary<V2> : MapReduceBase<V2>
{
protected MapperBaseBinary();
public abstract override IEnumerable<Tuple<string, V2>> Map(string filename, Stream document);
}
[Serializable]
[AbstractClass]
public abstract class CombinerBase<V2> : MapReduceBase<V2>
{
protected CombinerBase();
public abstract override IEnumerable<Tuple<string, V2>> Combine(string key, IEnumerable<V2> values);
}
[Serializable]
[AbstractClass]
public abstract class ReducerBase<V2, V3> : MapReduceBase<V2>
{
protected ReducerBase();
public abstract override IEnumerable<Tuple<string, V3>> Reduce(string key, IEnumerable<V2> values);
}
}

The equivalent F# definitions are:

F# Abstract Classes

namespace MSDN.Hadoop.MapReduceBase
[<AbstractClass>]
type MapReduceBase<'V2>() =
abstract member Setup: unit -> unit
default this.Setup() = ()
abstract member Cleanup: unit -> IEnumerable<string * 'V2>
default this.Cleanup() = Seq.empty
[<AbstractClass>]
type MapperBaseText<'V2>() =
inherit MapReduceBase<'V2>()
abstract member Map: value:string -> IEnumerable<string * 'V2>
[<AbstractClass>]
type MapperBaseXml<'V2>() =
inherit MapReduceBase<'V2>()
abstract member Map: element:XElement -> IEnumerable<string * 'V2>
[<AbstractClass>]
type MapperBaseBinary<'V2>() =
inherit MapReduceBase<'V2>()
abstract member Map: filename:string -> document:Stream -> IEnumerable<string * 'V2>
[<AbstractClass>]
type CombinerBase<'V2>() =
inherit MapReduceBase<'V2>()
abstract member Combine: key:string -> values:IEnumerable<'V2> -> IEnumerable<string * 'V2>
[<AbstractClass>]
type ReducerBase<'V2, 'V3>() =
inherit MapReduceBase<'V2>()
abstract member Reduce: key:string -> values:IEnumerable<'V2> -> IEnumerable<string * 'V3>

The objective in defining these base classes was to not only support creating .Net Mapper and Reducers but also to provide a means for Setup and Cleanup operations to support In-Place Mapper/Combiner/Reducer optimizations, utilize IEnumerable and sequences for publishing data from all classes, and finally provide a simple submission mechanism analogous to submitting Java based jobs.

The usage of the Generic types V2 and V3 equate to the names used in the Java definitions. The current type of the input into the Mapper is a string (this normally being V1). This is needed as the mapper, in Streaming jobs, performs the projection from the textual input.

For each class a Setup function is provided to allow one to perform tasks related to the instantiation of the class. The Mapper’s Map and Cleanup functions return an IEnumerable consisting of tuples with a Key/Value pair. It is these tuples that represent the mappers output. The returned types are written to file using binary serialization.

The Combiner and Reducer takes in an IEnumerable, for each key, and reduces this into a key/value enumerable. Once again the Cleanup allows for return values, to allow for In-Reducer optimizations.

Binary and XML Processing and Multiple Keys

As one can see from the abstract class definitions the framework also provides support for submitting jobs that support Binary and XML based Mappers. To support using Mappers derived from these types a “format” submission parameter is required. Supported values being Text, Binary, and XML; the default value being “Text”.

To submit a binary streaming job one just has to use a Mapper derived from the MapperBaseBinary abstract class and use the binary format specification:

-format Binary

In this case the input into the Mapper will be a Stream object that represents a complete binary document instance.

To submit an XML streaming job one just has to use a Mapper derived from the MapperBaseXml abstract class and use the XML format specification, along with a node to be processed within the XML documents:

-format XML –nodename Node

In this case the input into the Mapper will be an XElement node derived from the XML document based on the nodename parameter.

Using multiple keys from the Mapper is a two-step process. Firstly the Mapper needs to be modified to output a string based key in the correct format. This is done by passing the set of string key values into the Utilities.FormatKeys() function. This concatenates the keys using the necessary tab character. Secondly, the job has to be submitted specifying the expected number of keys:

MSDN.Hadoop.Submission.Console.exe -input "stores/demographics" -output "stores/banking"
-mapper "MSDN.Hadoop.MapReduceFSharp.StoreXmlElementMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.StoreXmlElementReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%HOMEPATH%\Projects\MSDN.Hadoop.MapReduce\Release\MSDN.Hadoop.MapReduceFSharp.dll"
-nodename Store -format Xml -numberKeys 2

This parameter equates to the necessary Hadoop job configuration parameter.

Samples

To demonstrate the submission framework, here are some sample Mappers and Reducers with the corresponding command line submissions:

C# Mobile Phone Range (with In-Mapper optimization)

Calculates the mobile phone query time range for a device with an In-Mapper optimization yielding just the Min and Max values:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MSDN.Hadoop.MapReduceBase;
namespace MSDN.Hadoop.MapReduceCSharp
{
public class MobilePhoneRangeMapper : MapperBaseText<TimeSpan>
{
private Dictionary<string, Tuple<TimeSpan, TimeSpan>> ranges;
private Tuple<string, TimeSpan> GetLineValue(string value)
{
try
{
string[] splits = value.Split('\t');
string devicePlatform = splits[3];
TimeSpan queryTime = TimeSpan.Parse(splits[1]);
return new Tuple<string, TimeSpan>(devicePlatform, queryTime);
}
catch (Exception)
{
return null;
}
}
public override void Setup()
{
this.ranges = new Dictionary<string, Tuple<TimeSpan, TimeSpan>>();
}
public override IEnumerable<Tuple<string, TimeSpan>> Map(string value)
{
var range = GetLineValue(value);
if (range != null)
{
if (ranges.ContainsKey(range.Item1))
{
var original = ranges[range.Item1];
if (range.Item2 < original.Item1)
{
// Update Min amount
ranges[range.Item1] = new Tuple<TimeSpan, TimeSpan>(range.Item2, original.Item2);
}
if (range.Item2 > original.Item2)
{
//Update Max amount
ranges[range.Item1] = new Tuple<TimeSpan, TimeSpan>(original.Item1, range.Item2);
}
}
else
{
ranges.Add(range.Item1, new Tuple<TimeSpan, TimeSpan>(range.Item2, range.Item2));
}
}
return Enumerable.Empty<Tuple<string, TimeSpan>>();
}
public override IEnumerable<Tuple<string, TimeSpan>> Cleanup()
{
foreach (var range in ranges)
{
yield return new Tuple<string, TimeSpan>(range.Key, range.Value.Item1);
yield return new Tuple<string, TimeSpan>(range.Key, range.Value.Item2);
}
}
}
public class MobilePhoneRangeReducer : ReducerBase<TimeSpan, Tuple<TimeSpan, TimeSpan>>
{
public override IEnumerable<Tuple<string, Tuple<TimeSpan, TimeSpan>>> Reduce(string key, IEnumerable<TimeSpan> value)
{
var baseRange = new Tuple<TimeSpan, TimeSpan>(TimeSpan.MaxValue, TimeSpan.MinValue);
var rangeValue = value.Aggregate(baseRange, (accSpan, timespan) =>
new Tuple<TimeSpan, TimeSpan>((timespan < accSpan.Item1) ? timespan : accSpan.Item1, (timespan > accSpan.Item2) ? timespan : accSpan.Item2));
yield return new Tuple<string, Tuple<TimeSpan, TimeSpan>>(key, rangeValue);
}
}
}

MSDN.Hadoop.Submission.Console.exe -input "mobilecsharp/data" -output "mobilecsharp/querytimes"
-mapper "MSDN.Hadoop.MapReduceCSharp.MobilePhoneRangeMapper, MSDN.Hadoop.MapReduceCSharp"
-reducer "MSDN.Hadoop.MapReduceCSharp.MobilePhoneRangeReducer, MSDN.Hadoop.MapReduceCSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduceCSharp\Release\MSDN.Hadoop.MapReduceCSharp.dll"

C# Mobile Min (with Mapper, Combiner, Reducer)

Calculates the mobile phone minimum time for a device with a combiner yielding just the Min value:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MSDN.Hadoop.MapReduceBase;
namespace MSDN.Hadoop.MapReduceCSharp
{
public class MobilePhoneMinMapper : MapperBaseText<TimeSpan>
{
private Tuple<string, TimeSpan> GetLineValue(string value)
{
try
{
string[] splits = value.Split('\t');
string devicePlatform = splits[3];
TimeSpan queryTime = TimeSpan.Parse(splits[1]);
return new Tuple<string, TimeSpan>(devicePlatform, queryTime);
}
catch (Exception)
{
return null;
}
}
public override IEnumerable<Tuple<string, TimeSpan>> Map(string value)
{
var returnVal = GetLineValue(value);
if (returnVal != null) yield return returnVal;
}
}
public class MobilePhoneMinCombiner : CombinerBase<TimeSpan>
{
public override IEnumerable<Tuple<string, TimeSpan>> Combine(string key, IEnumerable<TimeSpan> value)
{
yield return new Tuple<string, TimeSpan>(key, value.Min());
}
}
public class MobilePhoneMinReducer : ReducerBase<TimeSpan, TimeSpan>
{
public override IEnumerable<Tuple<string, TimeSpan>> Reduce(string key, IEnumerable<TimeSpan> value)
{
yield return new Tuple<string, TimeSpan>(key, value.Min());
}
}
}

MSDN.Hadoop.Submission.Console.exe -input "mobilecsharp/data" -output "mobilecsharp/querytimes"
-mapper "MSDN.Hadoop.MapReduceCSharp.MobilePhoneMinMapper, MSDN.Hadoop.MapReduceCSharp"
-reducer "MSDN.Hadoop.MapReduceCSharp.MobilePhoneMinReducer, MSDN.Hadoop.MapReduceCSharp"
-combiner "MSDN.Hadoop.MapReduceCSharp.MobilePhoneMinCombiner, MSDN.Hadoop.MapReduceCSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduceCSharp\Release\MSDN.Hadoop.MapReduceCSharp.dll"

F# Mobile Phone Query

Calculates the mobile phone range and average time for a device:

namespace MSDN.Hadoop.MapReduceFSharp
open System
open MSDN.Hadoop.MapReduceBase
type MobilePhoneQueryMapper() =
inherit MapperBaseText<TimeSpan>()
// Performs the split into key/value
let splitInput (value:string) =
try
let splits = value.Split('\t')
let devicePlatform = splits.[3]
let queryTime = TimeSpan.Parse(splits.[1])
Some(devicePlatform, queryTime)
with
| :? System.ArgumentException -> None
// Map the data from input name/value to output name/value
override self.Map (value:string) =
seq {
let result = splitInput value
if result.IsSome then
yield result.Value
}
type MobilePhoneQueryReducer() =
inherit ReducerBase<TimeSpan, (TimeSpan*TimeSpan*TimeSpan)>()
override self.Reduce (key:string) (values:seq<TimeSpan>) =
let initState = (TimeSpan.MaxValue, TimeSpan.MinValue, 0L, 0L)
let (minValue, maxValue, totalValue, totalCount) =
values |>
Seq.fold (fun (minValue, maxValue, totalValue, totalCount) value ->
(min minValue value, max maxValue value, totalValue + (int64)(value.TotalSeconds), totalCount + 1L) ) initState
Seq.singleton (key, (minValue, TimeSpan.FromSeconds((float)(totalValue/totalCount)), maxValue))

MSDN.Hadoop.Submission.Console.exe -input "mobile/data" -output "mobile/querytimes"
-mapper "MSDN.Hadoop.MapReduceFSharp.MobilePhoneQueryMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.MobilePhoneQueryReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduceFSharp\Release\MSDN.Hadoop.MapReduceFSharp.dll"

F# Store XML (XML in Samples)

Calculates the total revenue, within the store XML, based on demographic data; also demonstrating multiple keys:

namespace MSDN.Hadoop.MapReduceFSharp
open System
open System.Collections.Generic
open System.Linq
open System.IO
open System.Text
open System.Xml
open System.Xml.Linq
open MSDN.Hadoop.MapReduceBase
type StoreXmlElementMapper() =
inherit MapperBaseXml<decimal>()
override self.Map (element:XElement) =
let aw = "http://schemas.microsoft.com/sqlserver/2004/07/adventure-works/StoreSurvey"
let demographics = element.Element(XName.Get("Demographics")).Element(XName.Get("StoreSurvey", aw))
seq {
if not(demographics = null) then
let business = demographics.Element(XName.Get("BusinessType", aw)).Value
let bank = demographics.Element(XName.Get("BankName", aw)).Value
let key = Utilities.FormatKeys(business, bank)
let sales = Decimal.Parse(demographics.Element(XName.Get("AnnualSales", aw)).Value)
yield (key, sales)
}
type StoreXmlElementReducer() =
inherit ReducerBase<decimal, int>()
override self.Reduce (key:string) (values:seq<decimal>) =
let totalRevenue =
values |> Seq.sum
Seq.singleton (key, int totalRevenue)

MSDN.Hadoop.Submission.Console.exe -input "stores/demographics" -output "stores/banking"
-mapper "MSDN.Hadoop.MapReduceFSharp.StoreXmlElementMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.StoreXmlElementReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduceFSharp\bin\Release\MSDN.Hadoop.MapReduceFSharp.dll"
-nodename Store -format Xml

F# Binary Document (Word and PDF Documents)

Calculates the pages per author for a combination of Office Word and PDF documents:

namespace MSDN.Hadoop.MapReduceFSharp
open System
open System.Collections.Generic
open System.Linq
open System.IO
open System.Text
open System.Xml
open System.Xml.Linq
open DocumentFormat.OpenXml
open DocumentFormat.OpenXml.Packaging
open DocumentFormat.OpenXml.Wordprocessing
open iTextSharp.text
open iTextSharp.text.pdf
open MSDN.Hadoop.MapReduceBase
type OfficePageMapper() =
inherit MapperBaseBinary<int>()
let (|WordDocument|PdfDocument|UnsupportedDocument|) extension =
if String.Equals(extension, ".docx", StringComparison.InvariantCultureIgnoreCase) then
WordDocument
else if String.Equals(extension, ".pdf", StringComparison.InvariantCultureIgnoreCase) then
PdfDocument
else
UnsupportedDocument
let dc = XNamespace.Get("http://purl.org/dc/elements/1.1/")
let cp = XNamespace.Get("http://schemas.openxmlformats.org/package/2006/metadata/core-properties")
let unknownAuthor = "unknown author"
let authorKey = "Author"
let getAuthorsWord (document:WordprocessingDocument) =
let coreFilePropertiesXDoc = XElement.Load(document.CoreFilePropertiesPart.GetStream())
// Take the first dc:creator element and split based on a ";"
let creators = coreFilePropertiesXDoc.Elements(dc + "creator")
if Seq.isEmpty creators then
[| unknownAuthor |]
else
let creator = (Seq.head creators).Value
if String.IsNullOrWhiteSpace(creator) then
[| unknownAuthor |]
else
creator.Split(';')
let getPagesWord (document:WordprocessingDocument) =
// return page count
Int32.Parse(document.ExtendedFilePropertiesPart.Properties.Pages.Text)
let getAuthorsPdf (document:PdfReader) =
// For PDF documents perform the split on a ","
if document.Info.ContainsKey(authorKey) then
let creators = document.Info.[authorKey]
if String.IsNullOrWhiteSpace(creators) then
[| unknownAuthor |]
else
creators.Split(',')
else
[| unknownAuthor |]
let getPagesPdf (document:PdfReader) =
// return page count
document.NumberOfPages
// Map the data from input name/value to output name/value
override self.Map (filename:string) (document:Stream) =
let result =
match Path.GetExtension(filename) with
| WordDocument ->
// Get access to the word processing document from the input stream
use document = WordprocessingDocument.Open(document, false)
// Process the word document with the mapper
let pages = getPagesWord document
let authors = (getAuthorsWord document)
// close document
document.Close()
Some(pages, authors)
| PdfDocument ->
// Get access to the pdf processing document from the input stream
let document = new PdfReader(document)
// Process the pdf document with the mapper
let pages = getPagesPdf document
let authors = (getAuthorsPdf document)
// close document
document.Close()
Some(pages, authors)
| UnsupportedDocument ->
None
if result.IsSome then
snd result.Value
|> Seq.map (fun author -> (author, fst result.Value))
else
Seq.empty
type OfficePageReducer() =
inherit ReducerBase<int, int>()
override self.Reduce (key:string) (values:seq<int>) =
let totalPages =
values |> Seq.sum
Seq.singleton (key, totalPages)

MSDN.Hadoop.Submission.Console.exe -input "office/documents" -output "office/authors"
-mapper "MSDN.Hadoop.MapReduceFSharp.OfficePageMapper, MSDN.Hadoop.MapReduceFSharp"
-reducer "MSDN.Hadoop.MapReduceFSharp.OfficePageReducer, MSDN.Hadoop.MapReduceFSharp"
-combiner "MSDN.Hadoop.MapReduceFSharp.OfficePageReducer, MSDN.Hadoop.MapReduceFSharp"
-file "%HOMEPATH%\MSDN.Hadoop.MapReduceFSharp\bin\Release\MSDN.Hadoop.MapReduceFSharp.dll"
-file "C:\Reference Assemblies\itextsharp.dll" -format Binary

Optional Parameters

To support some additional Hadoop Streaming options a few optional parameters are supported.

-numberReducers X

As expected this specifies the maximum number of reducers to use.

-debug

The option turns on verbose mode and specifies a job configuration to keep failed task outputs.

To view the the supported options one can use a help parameters, displaying:

Command Arguments:
-input (Required=true) : Input Directory or Files
-output (Required=true) : Output Directory
-mapper (Required=true) : Mapper Class
-reducer (Required=true) : Reducer Class
-combiner (Required=false) : Combiner Class (Optional)
-format (Required=false) : Input Format |Text(Default)|Binary|Xml|
-numberReducers (Required=false) : Number of Reduce Tasks (Optional)
-numberKeys (Required=false) : Number of MapReduce Keys (Optional)
-file (Required=true) : Processing Files (Must include Map and Reduce Class files)
-nodename (Required=false) : XML Processing Nodename (Optional)
-debug (Required=false) : Turns on Debugging Options

UI Submission

The provided submission framework works from a command-line. However there is nothing to stop one submitting the job using a UI; albeit a command console is opened. To this end I have put together a simple UI that supports submitting Hadoop jobs.

This simple UI supports all the necessary options for submitting jobs.

Code Download

As mentioned the actual Executables and Source code can be downloaded from:

http://code.msdn.microsoft.com/Framework-for-Composing-af656ef7

The source includes, not only the .Net submission framework, but also all necessary Java classes for supporting the Binary and XML job submissions. This relies on a custom Streaming JAR which should be copied to the Hadoop lib directory, there are two versions of the Streaming jar; one for running in azure and one for when running local. The difference is that they have been compiled with different versions of the Java compiler. Just remember to use the appropriate version (dropping the –local and –azure prefixes) when copying to your Hadoop lib folder.

To use the code one just needs to reference the EXE’s in the Release directory. This folder also contains the MSDN.Hadoop.MapReduceBase.dll that contains the abstract base class definitions.

Moving Forward

In a separate post I will cover what is actually happening under the covers.

As always if you find the code useful and/or use this for your MapReduce jobs, or just have some comments, please do let me know.

Written by Carl Nolan

Framework for Composing and Submitting .Net Hadoop MapReduce Jobs

MCS UK Solution Development — Mon, 16 Apr 2012 19:07:00 GMT

An updated version of this post can be found at:

http://blogs.msdn.com/b/carlnol/archive/2012/04/29/generic-based-framework-for-net-hadoop-mapreduce-job-submission.aspx

If you have been following my blog you will see that I have been putting together samples for writing .Net Hadoop MapReduce jobs; using Hadoop Streaming. However one thing that became apparent is that the samples could be reconstructed in a composable framework to enable one to submit .Net based MapReduce jobs whilst only writing Mappers and Reducers types.

To this end I have put together a framework that allows one to submit MapReduce jobs using the following command line syntax:

Where the mapper and reducer parameters are .Net types that derive from a base Map and Reduce abstract classes. The input, output, and files options are analogous to the standard Hadoop streaming submissions. The mapper and reducer options (more on a combiner option later) allow one to define a .Net type derived from the appropriate abstract base classes.

Under the covers standard Hadoop Streaming is being used, where controlling executables are used to handle the StdIn and StdOut operations and activating the required .Net types. The “file” parameter is required to specify the DLL for the .Net type to be loaded at runtime, in addition to any other required files.

As an aside the framework and base classes are all written in F#; with sample Mappers and Reducers, and abstract base classes being provided both in C# and F#. The code is based off the F# Streaming samples in my previous blog posts. I will cover more of the semantics of the code in a later post, but I wanted to provide some usage samples of the code.

As always the source can be downloaded from:

http://code.msdn.microsoft.com/Framework-for-Composing-af656ef7

Mapper and Reducer Base Classes

The following definitions outline the abstract base classes from which one needs to derive.

C# Base

[AbstractClass]
public abstract class MapReduceBase
{
    protected MapReduceBase();
 
    public override IEnumerable<Tuple<string, object>> Cleanup();
    public override void Setup();
}

C# Base Mapper

public abstract class MapperBaseText : MapReduceBase
{
    protected MapperBaseText();
 
    public override IEnumerable<Tuple<string, object>> Cleanup();
    public abstract override IEnumerable<Tuple<string, object>> Map(string value);
}

C# Base Reducer

[AbstractClass]
public abstract class ReducerBase : MapReduceBase
{
    protected ReducerBase();
 
    public abstract override IEnumerable<Tuple<string, object>> Reduce(string key, IEnumerable<string> values);
}

F# Base

[<AbstractClass>]
type MapReduceBase() =
 
    abstract member Setup: unit -> unit
    default this.Setup() = ()
 
    abstract member Cleanup: unit -> IEnumerable<string * obj>
    default this.Cleanup() = Seq.empty

F# Base Mapper

[<AbstractClass>]
type MapperBaseText() =
    inherit MapReduceBase()
 
    abstract member Map: string -> IEnumerable<string * obj>
 
    abstract member Cleanup: unit -> IEnumerable<string * obj>
    default this.Cleanup() = Seq.empty

F# Base Reducer

[<AbstractClass>]
type ReducerBase() =
    inherit MapReduceBase()
 
    abstract member Reduce: key:string -> values:IEnumerable<string> -> IEnumerable<string * obj>

The objective in defining these base classes was to not only support creating .Net Mapper and Reducers but also to provide a means for Setup and Cleanup operations to support In-Place Mapper optimizations, utilize IEnumerable and sequences for publishing data from the Mappers and Reducers, and finally provide a simple submission mechanism analogous to submitting Java based jobs.

For each class a Setup function is provided to allow one to perform tasks related to the instantiation of each Mapper and/or Reducer. The Mapper’s Map and Cleanup functions return an IEnumerable consisting of tuples with a a Key/Value pair. It is these tuples that represent the mappers output. Currently the types of the key and value’s are respectively a String and an Object. These are then converted to strings for the streaming output.

The Reducer takes in an IEnumerable of the Object String representations, created by the Mapper output, and reduces this into a Object value enumerable. Once again the Cleanup allows for return values to allow for In-Reducer optimizations.

Combiners

The support for Combiners is provided through one of two means. As is often the case, support is provided so one can reuse a Reducer as a Combiner. In addition explicit support is provided for a Combiner using the following abstract class definition:

C# Base Combiner

public class MobilePhoneMinCombiner : CombinerBase
{
    public override IEnumerable<Tuple<string, object>> Combine(string key, IEnumerable<string> value)
    {
        yield return new Tuple<string, object>(key, value.Select(timespan => TimeSpan.Parse(timespan)).Min());
    }
}

F# Base Combiner

[<AbstractClass>]
type CombinerBase() =
    inherit MapReduceBase()
 
    abstract member Combine: key:string -> values:IEnumerable<string> -> IEnumerable<string * obj>

Using a Combiner follows exactly the same pattern for using mappers and reducers, as example being:

-combiner "MSDN.Hadoop.MapReduceCSharp.MobilePhoneMinCombiner, MSDN.Hadoop.MapReduceCSharp"

The prototype for the Combiner is essentially the same as that of the Reducer except the function called for each row of data is Combine, rather than Reduce.

Binary and XML Processing

In my previous posts on Hadoop Streaming I provided samples that allowed one to perform Binary and XML based Mappers. The composable framework also provides support for submitting jobs that support Binary and XML based Mappers. To support this the following additional abstract classes have been defined:

C# Base Binary Mapper

[AbstractClass]
public abstract class MapperBaseBinary : MapReduceBase
{
    protected MapperBaseBinary();
 
    public abstract override IEnumerable<Tuple<string, object>> Map(string filename, Stream document);
}

C# Base XML Mapper

[AbstractClass]
public abstract class MapperBaseXml : MapReduceBase
{
    protected MapperBaseXml();
 
    public abstract override IEnumerable<Tuple<string, object>> Map(XElement element);
}

F# Base Binary Mapper

[<AbstractClass>]
type MapperBaseBinary() =
    inherit MapReduceBase()
 
    abstract member Map: filename:string -> document:Stream -> IEnumerable<string * obj>

F# Base XML Mapper

[<AbstractClass>]
type MapperBaseXml() =
    inherit MapReduceBase()
 
    abstract member Map: element:XElement -> IEnumerable<string * obj>

To support using Mappers and Reducers derived from these types a “format” submission parameter is required. Supported values being Text, Binary, and XML; the default value being “Text”.

To submit a binary streaming job one just has to use a Mapper derived from the MapperBaseBinary abstract class and use the binary format specification:

-format Binary

In this case the input into the Mapper will be a Stream object that represents a complete binary document instance.

-format XML –nodename Node

In this case the input into the Mapper will be an XElement node derived from the XML document based on the nodename parameter.

Samples

To demonstrate the submission framework here are some sample Mappers and Reducers with the corresponding command line submissions:

C# Mobile Phone Range (with In-Mapper optimization)

namespace MSDN.Hadoop.MapReduceCSharp
{
    public class MobilePhoneRangeMapper : MapperBaseText
    {
        private Dictionary<string, Tuple<TimeSpan, TimeSpan>> ranges;
 
        private Tuple<string, TimeSpan> GetLineValue(string value)
        {
            try
            {
                string[] splits = value.Split('\t');
                string devicePlatform = splits[3];
                TimeSpan queryTime = TimeSpan.Parse(splits[1]);
                return new Tuple<string, TimeSpan>(devicePlatform, queryTime);
            }
            catch (Exception)
            {
                return null;
            }
        }
 
        /// <summary>
        /// Define a Dictionary to hold the (Min, Max) tuple for each device platform.
        /// </summary>
        public override void Setup()
        {
            this.ranges = new Dictionary<string, Tuple<TimeSpan, TimeSpan>>();
        }
 
        /// <summary>
        /// Build the Dictionary of the (Min, Max) tuple for each device platform.
        /// </summary>
        public override IEnumerable<Tuple<string, object>> Map(string value)
        {
            var range = GetLineValue(value);
            if (range != null)
            {
                if (ranges.ContainsKey(range.Item1))
                {
                    var original = ranges[range.Item1];
                    if (range.Item2 < original.Item1)
                    {
                        // Update Min amount
                        ranges[range.Item1] = new Tuple<TimeSpan, TimeSpan>(range.Item2, original.Item2);
                    }
                    if (range.Item2 > original.Item2)
                    {
                        //Update Max amount
                        ranges[range.Item1] = new Tuple<TimeSpan, TimeSpan>(original.Item1, range.Item2);
                    }
                }
                else
                {
                    ranges.Add(range.Item1, new Tuple<TimeSpan, TimeSpan>(range.Item2, range.Item2));
                }
            }
 
            return Enumerable.Empty<Tuple<string, object>>();
        }
 
        /// <summary>
        /// Return the Dictionary of the Min and Max values for each device platform.
        /// </summary>
        public override IEnumerable<Tuple<string, object>> Cleanup()
        {
            foreach (var range in ranges)
            {
                yield return new Tuple<string, object>(range.Key, range.Value.Item1);
                yield return new Tuple<string, object>(range.Key, range.Value.Item2);
            }
        }
    }
 
    public class MobilePhoneRangeReducer : ReducerBase
    {
        public override IEnumerable<Tuple<string, object>> Reduce(string key, IEnumerable<string> value)
        {
            var baseRange = new Tuple<TimeSpan, TimeSpan>(TimeSpan.MaxValue, TimeSpan.MinValue);
            var rangeValue = value.Select(stringspan => TimeSpan.Parse(stringspan)).Aggregate(baseRange, (accSpan, timespan) =>
                    new Tuple<TimeSpan, TimeSpan>((timespan < accSpan.Item1) ? timespan : accSpan.Item1, (timespan > accSpan.Item2) ? timespan : accSpan.Item2));
 
            yield return new Tuple<string, object>(key, rangeValue);
        }
    }
}

C# Mobile Min (with Mapper, Combiner, Reducer)

namespace MSDN.Hadoop.MapReduceCSharp
{
    public class MobilePhoneMinMapper : MapperBaseText
    {
        private Tuple<string, object> GetLineValue(string value)
        {
            try
            {
                string[] splits = value.Split('\t');
                string devicePlatform = splits[3];
                TimeSpan queryTime = TimeSpan.Parse(splits[1]);
                return new Tuple<string, object>(devicePlatform, queryTime);
            }
            catch (Exception)
            {
                return null;
            }
        }
 
        public override IEnumerable<Tuple<string, object>> Map(string value)
        {
            var returnVal = GetLineValue(value);
            if (returnVal != null) yield return returnVal;
        }
    }
 
    public class MobilePhoneMinCombiner : CombinerBase
    {
        public override IEnumerable<Tuple<string, object>> Combine(string key, IEnumerable<string> value)
        {
            yield return new Tuple<string, object>(key, value.Select(timespan => TimeSpan.Parse(timespan)).Min());
        }
    }
 
    public class MobilePhoneMinReducer : ReducerBase
    {
        public override IEnumerable<Tuple<string, object>> Reduce(string key, IEnumerable<string> value)
        {
            yield return new Tuple<string, object>(key, value.Select(timespan => TimeSpan.Parse(timespan)).Min());
        }
    }
}

F# Mobile Phone Query

namespace MSDN.Hadoop.MapReduceFSharp
 
open System
open MSDN.Hadoop.MapReduceBase
 
// Extracts the QueryTime for each Platform Device
type MobilePhoneQueryMapper() =
    inherit MapperBaseText()
 
    // Performs the split into key/value
    let splitInput (value:string) =
        try
            let splits = value.Split('\t')
            let devicePlatform = splits.[3]
            let queryTime = TimeSpan.Parse(splits.[1])
            Some(devicePlatform, box queryTime)
        with
        | :? System.ArgumentException -> None
 
    // Map the data from input name/value to output name/value
    override self.Map (value:string) =
        seq {
            let result = splitInput value
            if result.IsSome then
                yield result.Value
        }
 
// Calculates the (Min, Avg, Max) of the input stream query time (based on Platform Device)
type MobilePhoneQueryReducer() =
    inherit ReducerBase()
 
    override self.Reduce (key:string) (values:seq<string>) =
        let initState = (TimeSpan.MaxValue, TimeSpan.MinValue, 0L, 0L)
        let (minValue, maxValue, totalValue, totalCount) =
            values |>
            Seq.fold (fun (minValue, maxValue, totalValue, totalCount) value ->
                (min minValue (TimeSpan.Parse(value)), max maxValue (TimeSpan.Parse(value)), totalValue + (int64)(TimeSpan.Parse(value).TotalSeconds), totalCount + 1L) ) initState
 
        Seq.singleton (key, box (minValue, TimeSpan.FromSeconds((float)(totalValue/totalCount)), maxValue))

F# Store XML (XML in Samples)

namespace MSDN.Hadoop.MapReduceFSharp
 
open System
open System.Collections.Generic
open System.Linq
open System.IO
open System.Text
open System.Xml
open System.Xml.Linq
 
open MSDN.Hadoop.MapReduceBase
 
// Extracts the QueryTime for each Platform Device
type StoreXmlElementMapper() =    
    inherit MapperBaseXml()
 
    override self.Map (element:XElement) =
 
        let aw = "http://schemas.microsoft.com/sqlserver/2004/07/adventure-works/StoreSurvey"
        let demographics = element.Element(XName.Get("Demographics")).Element(XName.Get("StoreSurvey", aw))
 
        seq {
            if not(demographics = null) then
                let business = demographics.Element(XName.Get("BusinessType", aw)).Value
                let sales = Decimal.Parse(demographics.Element(XName.Get("AnnualSales", aw)).Value) |> box
                yield (business, sales)
            }
 
// Calculates the Total Revenue of the store demographics
type StoreXmlElementReducer() =
    inherit ReducerBase()
 
    override self.Reduce (key:string) (values:seq<string>) =
        let totalRevenue =
            values |>
            Seq.fold (fun revenue value -> revenue + Int32.Parse(value)) 0
 
        Seq.singleton (key, box totalRevenue)

F# Binary Document (Word and PDF Documents)

namespace MSDN.Hadoop.MapReduceFSharp
 
open System
open System.Collections.Generic
open System.Linq
open System.IO
open System.Text
open System.Xml
open System.Xml.Linq
 
open DocumentFormat.OpenXml
open DocumentFormat.OpenXml.Packaging
open DocumentFormat.OpenXml.Wordprocessing
 
open iTextSharp.text
open iTextSharp.text.pdf
 
open MSDN.Hadoop.MapReduceBase
 
// Calculates the pages per author for a Word document
type OfficePageMapper() =
    inherit MapperBaseBinary()
 
    let (|WordDocument|PdfDocument|UnsupportedDocument|) extension =
        if String.Equals(extension, ".docx", StringComparison.InvariantCultureIgnoreCase) then
            WordDocument
        else if String.Equals(extension, ".pdf", StringComparison.InvariantCultureIgnoreCase) then
            PdfDocument
        else
            UnsupportedDocument
 
    let dc = XNamespace.Get("http://purl.org/dc/elements/1.1/")
    let cp = XNamespace.Get("http://schemas.openxmlformats.org/package/2006/metadata/core-properties")
    let unknownAuthor = "unknown author"
    let authorKey = "Author"
 
    let getAuthorsWord (document:WordprocessingDocument) =
        let coreFilePropertiesXDoc = XElement.Load(document.CoreFilePropertiesPart.GetStream())
          
        // Take the first dc:creator element and split based on a ";"
        let creators = coreFilePropertiesXDoc.Elements(dc + "creator")
        if Seq.isEmpty creators then
            [| unknownAuthor |]
        else
            let creator = (Seq.head creators).Value
            if String.IsNullOrWhiteSpace(creator) then
                [| unknownAuthor |]
            else
                creator.Split(';')
 
    let getPagesWord (document:WordprocessingDocument) =
        // return page count
        Int32.Parse(document.ExtendedFilePropertiesPart.Properties.Pages.Text)
 
    let getAuthorsPdf (document:PdfReader) =          
        // For PDF documents perform the split on a ","
        if document.Info.ContainsKey(authorKey) then
            let creators = document.Info.[authorKey]
            if String.IsNullOrWhiteSpace(creators) then
                [| unknownAuthor |]
            else
                creators.Split(',')
        else
            [| unknownAuthor |]
 
    let getPagesPdf (document:PdfReader) =
        // return page count
        document.NumberOfPages
 
    // Map the data from input name/value to output name/value
    override self.Map (filename:string) (document:Stream) =
 
        let result =
            match Path.GetExtension(filename) with
            | WordDocument ->
                // Get access to the word processing document from the input stream
                use document = WordprocessingDocument.Open(document, false)
                // Process the word document with the mapper
                let pages = getPagesWord document
                let authors = (getAuthorsWord document)
                // close document
                document.Close()
                Some(pages, authors)
            | PdfDocument ->
                // Get access to the pdf processing document from the input stream
                let document = new PdfReader(document)
                // Process the pdf document with the mapper
                let pages = getPagesPdf document
                let authors = (getAuthorsPdf document)       
                // close document
                document.Close()
                Some(pages, authors)
            | UnsupportedDocument ->
                None
 
        if result.IsSome then
            snd result.Value
            |> Seq.map (fun author -> (author, (box << fst) result.Value))
        else
            Seq.empty
 
// Calculates the total pages per author
type OfficePageReducer() =
    inherit ReducerBase()
 
    override self.Reduce (key:string) (values:seq<string>) =
        let totalPages =
            values |>
            Seq.fold (fun pages value -> pages + Int32.Parse(value)) 0
 
        Seq.singleton (key, box totalPages)

Optional Parameters

To support some additional Hadoop Streaming options a few optional parameters are supported.

-numberReducers X

As expected this specifies the maximum number of reducers to use.

-debug

The option turns on verbose mode and specifies a job configuration to keep failed task outputs.

To view the the supported options one can use a help parameters, displaying:

Command Arguments:
-input (Required=true) : Input Directory or Files
-output (Required=true) : Output Directory
-mapper (Required=true) : Mapper Class
-reducer (Required=true) : Reducer Class
-combiner (Required=false) : Combiner Class (Optional)
-format (Required=false) : Input Format |Text(Default)|Binary|Xml|
-numberReducers (Required=false) : Number of Reduce Tasks (Optional)
-file (Required=true) : Processing Files (Must include Map and Reduce Class files)
-nodename (Required=false) : XML Processing Nodename (Optional)
-debug (Required=false) : Turns on Debugging Options

UI Submission

This simple UI supports all the necessary options for submitting jobs.

Code Download

As mentioned the actual Executables and Source code can be downloaded from:

http://code.msdn.microsoft.com/Framework-for-Composing-af656ef7

To use the code one just needs to reference the EXE’s in the Release directory. This folder also contains the MSDN.Hadoop.MapReduceBase.dll that contains the abstract base class definitions.

Moving Forward

Moving forward there a few considerations for the code, that I will be looking at over time:

Currently the abstract interfaces are all based on Object return types. Moving forward it would be beneficial if the types were based on Generics. This would allow a better serialization process. Currently value serialization is based on string representation of an objects value and the key is restricted to s string. Better serialization processes, such as Protocol Buffers, or .Net Serialization, would improve performance.

Currently the code only supports a single key value, although the multiple keys are supported by the streaming interface. Various options are available for dealing with multiple keys which will next be on my investigation list.

In a separate post I will cover what is actually happening under the covers.

Written by Carl Nolan

Workflow Foundation (WF4) – Rehosting The Workflow Designer

MCS UK Solution Development — Mon, 26 Mar 2012 08:13:00 GMT

Rehosting the WF designer in an application outside of Visual Studio is nothing new and since WF4, nothing particularly difficult. In fact the WF product team has gone out of its way to make rehosting the designer as easy an experience as possible and they’ve done a great job. The WCF and WF Samples for .NET Framework 4 provides code samples for rehosting the designer with its corresponding toolbox and properties grid. The samples also cover handling validation errors, executing workflows and providing some level of debugging through workflow tracking – most of what you need to write your own rehosted designer.

But to get a useable app you’re going to have to write some boiler plate code to stitch all these concepts together. You’ll need code to open, save and execute workflows. You’ll need to manage window layout, display workflow output and handle runtime exceptions. And wouldn't it be nice to support working with multiple workflows at the same time all in a windowing environment that supports docking and pinning?

Enter Workflow Studio – a simple, generic application that allows you to design and execute multiple XAML based workflows in a windowed environment akin with Visual Studio. Use it to design and test your workflows in environments where Visual Studio is not the right tool. You can even use it as a simple hosting environment. I’ve developed Workflow Studio as an application you can use out of the box, or you can use it as the basis for your own specialised implementation. You can download the source code here.

Features

The following image shows the Workflow Studio environment:

Here’s a rundown of its features and where appropriate I've provided references to sample code should you want to understand more about the implementation detail.

Develop XAML Based Workflows And Workflow Services
Use the fully fledged WF designer to develop XAML based workflows and workflow services just as you would in Visual Studio. Interaction between the toolbox and properties box is just as in Visual Studio. For workflow services, add any WCF configuration to the Workflow Studio app config file. The solution comes with a test workflow service and client with appropriate configuration in the app config file as an example. Look at the Designer Rehosting example in WCF and WF Samples for .NET Framework 4 that demonstrates the basis for this implementation. Also, Pro WF Windows Workflow in .NET 4 by Bruce Bukovics has a great chapter on designer rehosting. Should you wish to execute other XAML based child workflows from a workflow then you can use the ExecuteXamlWorkflow custom activity I described in my previous post.
Work On More Than One Workflow At The Same Time
Workflow Studio allows you to develop multiple workflows in a single application.
Toolbox Support For All Standard WF Activities
All standard WF 4.0 activities are supported.
Add Custom Activities To The Toolbox
To add custom activities to the toolbox, copy the custom activity DLL to the bin folder of Workflow Studio, select the "Add Reference …" option and locate the DLL file. The custom activity will then become available in the toolbox. If your workflow references custom activities then these will be automatically loaded and added to the toolbox when the workflow is loaded providing the referenced DLL has been previously copied to the bin folder. The section on designer rehosting in Pro WF Windows Workflow in .NET 4 by Bruce Bukovics, providing the details of toolbox manipulation that I used as basis for Workflow Studio.
Execute Workflows Concurrently
You can execute one or more workflows or workflow services concurrently by either selecting "Start Debugging" or "Start Without Debugging" from the debug menu. Running workflows can be stopped by selecting "Abort" from the same menu. Each workflow has its own independent output window where WriteLine activity or exception stack trace output is written. You can also capture output through a trace source and associate a listener to log the output. We're using standard .NET diagnostics here, so it's totally flexible.
Show Workflow Validation Errors
Any validation errors detected by the designer are displayed in the error window along with error code and severity. Each workflow has its own independent window.
Debug Workflows
This isn't fully fledged Visual Studio debugging - it's workflow tracking. You'll see the currently executing activity highlighted in the designer so you'll be able to visually track the execution path of your workflow in real time. As each activity executes the activity name, id, state and workflow instance id are written to the workflow's debug window. Clicking on a row in the debug window will focus the designer on the corresponding activity. Similar to workflow output, the debug window output is also captured through a trace source so that output can be logged. Check out Visual Workflow Tracking in WCF and WF Samples for .NET Framework 4 which demonstrates how to implement visual tracking in the designer.
Visual Studio Like Window Docking And Pinning
One of the key features of Workflow Studio is to offer full window docking and pinning functionality that you find in Visual Studio. This functionality is provided by the AvalonDock open source WPF docking framework. This is a great framework but took some jumping through hoops to get working. I started out wanting to make this a pure MVVM application but soon discovered this wasn't easy, if at all possible. Apparently this is addressed with the up and coming 2.0 release along with other improvements. Note that you'll need to download and install AvalonDock independently of Workflow Studio since I don't redistribute it here - don't worry this is simple. Unfortunately there's no NuGet package available as yet so please follow instructions below.

Other Features

Persistence

Persistence for Workflows other than workflow services is not currently supported. Workflows are hosted using the WorkflowApplication class and persistence makes little sense unless you can implement code that interacts with the host to load and unload instances and this is not possible with Workflow Studio currently.

On the other hand, persistence for Workflow services is supported. Here, interaction with the host is not required since persistence is triggered through service interaction. Workflow services are hosted using the WorkflowServiceHost class and this allows all persistence behaviour to be configured in the app config file. I used Leon Welicki's great post on long running workflows with WCF to see how this would work and whether the debugging mode could help trace workflow execution. Here's how it looked:

Sending the debug output to a file (see below) really comes into its own here as you can sort and filter the output in an application such as Excel to look at the execution sequence of individual workflow instances.

Localization

Workflow studio is easily localizable. All text strings are defined in the resources.resx file that can be found in the solution properties folder. Simply translate this to your required language and rename using the naming convention for resource files. Resources are selected according to your current locale, so if you need the ability to select locales other than your current one you will need to implement this feature yourself.

Output Logging

As mentioned previously, all workflow output and debug tracking output is captured using trace sources that can be associated with listeners that log output. Output can be captured either per workflow or collectively.

Here's an example from the supplied app config file:

<!-- for output, add new sources here where name must match workflow name without file extension, suffixed with "Output" -->
           <source name="TestLoopWithDelayOutput"
             switchType="System.Diagnostics.SourceSwitch"
             switchName="verboseSwitch">
               <listeners>
                   <add name="testLoopWithDelayOutputListener"
                     type="System.Diagnostics.TextWriterTraceListener"
                     initializeData="TestLoopWithDelayOutput.log">
                   </add>
                   <remove name="Default"/>
               </listeners>
           </source>

In the above example, the workflow document tab with the name "TestLoopWithDelay.xaml" will write its output to the file TestLoopWithDelayOutput.log. The source name must match the workflow document name, excluding the extension and suffixed with “Output”.

Each workflow will also write its output to a single shared output file, so you don’t have to configure the above if your don’t require the output in a separate file:

<!-- The AllOutput source captures output for all workflows -->
            <source name="AllOutput"
              switchType="System.Diagnostics.SourceSwitch"
              switchName="verboseSwitch">
                <listeners>
                    <add name="allOutputListener"
                      type="System.Diagnostics.TextWriterTraceListener"
                      initializeData="AllOutput.log">
                    </add>
                    <remove name="Default"/>
                </listeners>
            </source>

A trace source is also associated with exception handling. In the same config section you'll notice that a listener logs all exceptions to the Error.log file of the bin folder.

Similar to workflow output, debug tracking output is captured using trace sources and each workflow can have it’s tracking captured independently or collectively. Where specifying the trace source for debug tracking the source name must match the workflow document name, excluding the extension and suffixed with “Debug”. Collective output is written to “AllDebug.csv”. The configuration uses the System.Diagnostics.DelimitedListTraceListener that writes output to a comma delimited file. This is ideal for loading into say Excel, where you can filter and sort to see the execution path of a specific workflow instance.

Controlling Debug Output

The following settings in the Workflow Studio app config file allow you to control the tracking output behaviour:

PauseBetweenDebugStepsInMilliseconds	The number of milliseconds to pause before moving onto the next activity. This allows you to see the execution path more easily as the workflow executes in real time.
DisableDebugViewOutput	You may only be interested in tracking output sent to a file in which case you can disable interactive tracking by setting this value to true.

Installing AvalonDock

As previously mentioned, AvalonDock is not distributed with Workflow Studio so you need to download and install it first. This version of Workflow Studio has been developed with version 1.3.3571 of AvalonDock so please ensure you install this version. The steps are simple:

Download AvalonDock version 1.3.3571 from http://avalondock.codeplex.com/releases/48794/download/131885
Run the installer. This will GAC the AvalonDock assembly.
Build the solution. You should not need to modify any references.

Conclusion

Hopefully you’ll find Workflow Studio a useful application for designing and debugging workflows in environments where Visual Studio is not a viable option. You can even use it for simple workflow hosting. It’ll save you writing a lot of boiler plate code and can serve as a starting point for your own implementation should you require something more specialized. I’d be interested to hear if you find Workflow Studio useful and welcome any feedback for possible enhancements. Also, please let me know of any bugs you may find.

Written by Christopher Owczarek

Stream join aggregates using SQL Server Stream Insight

MCS UK Solution Development — Fri, 16 Mar 2012 12:34:00 GMT

Lately I have been working with StreamInsight and thought it was about time to share some of my observations and key points.

This is loosely based upon a requirement to aggregate Financial risk vectors and deliver changes to clients in an event based manner. When working through this I was primarily interested in the following factors:

Joining the two streams in a timely manner, with the results being ‘lively’ (that is released as quickly as possible)
What sort of performance could be achieved (without much effort)
How I could inject queries into the engine at runtime to enable a ‘service based model’ for clients.

Think of risk vectors simply as curves - a series of values plotted over a set of time points (arrays of doubles over distinct points in time). For example, look at the following 'curve' which shows the Bank Of England base rates I took at a specific point in time.

Normally over time longer term rates are higher than shorter term rates.

Essentially I have two streams of data which I want to correlate. One stream is of risk vector arrays - which have been calculated based upon a number of inputs upstream, usually by some form of compute grid.

The second stream is the attribution of the vectors - the things we may want to analyse or aggregate by - Trade, Portfolio etc. with a join key which relates the Trade to the Risk. Once combined I multiply a nominal amount by the corresponding rate and that is my risk. The basic idea for each Trade is:

Join to the array of doubles;
Multiply a nominal amount by each double;
Dequeue the result

This is summarized in the below diagram.

The data sources are as shown below.

Basic Plumbing

The first thing to do is to build out the 'infrastructure' using the StreamInsight programming model.

This means creating my adapters (Risk, Trade and Output). I won't describe the details of the programming model. You can review the code sample and the Stream Insight documentation for this. However, for this scenario there a couple of key points:

Both Trade and Risk must be either Interval or Edge event types. This is because I am joining streams across time and the cardinality of the relationship is many to many. A vector will apply to many Trades, and a Trade can apply to many vectors. A join on two point events would not support this (once the Point event is released it is gone).
For aggregate queries I want to release the events whenever there is an update (new source event which contributes to it). This means using the SnapshotWindow – using Hopping or Tumbling will not support this.

Joining Streams

It is worth briefly describing the management of application time. This is a fundamental concept to understand in Stream Insight. Common Time Increments (CTI) events are used as a basis for advancing application time. The CTI is effectively an assertion that there will be no more events beyond this time - and this tells the engine that it is ok to dequeue events. For example if I enqueue 3 events with times of T1, T2 and T3 (assuming Point events) and enqueue them to the engine, they will be dequeued when I assert a CTI of T4.

There are a few ways of managing application time in Stream Insight:

Declaratively - using AdvancedTimeSettings

Using a heartbeat mechanism - such as a timer

Declaratively using an AdvancedTimeImport

When joining streams what is important to realize is that a ‘join event’ is only released to the output after a CTI has been inserted into each of the source streams.

In my example I used the following input files:

Trades – 5000 trades, each with a join to a single array
Risk – 40 vectors, each an array of 20 values

This means that we will produce 100,000 events (5000 x 20).

Firstly, I looked at using AdvancedTimeSettings, defining the same time policy for each stream. The problem here is that - unless your streams have the same number of events which are enqueued with the same occurrence of CTIs (very unlikely!) - at a certain point in time the slower stream will enqueue all of its events. As mentioned above, this means that there are no more events to enqueue and, as we are using declarative approach to time management, there will be no further CTIs produced. So we could enqueue all of the Risk, more Trades could arrive which have a matching join already enqueued, but they will not be released – because there is no CTI inserted to the Risk stream.

The second option is to manage things manually using a ‘heartbeat’ CTI injection approach – essentially each adapter injecting CTIs using a Timer. This is an approach which does work but you have to be careful to make sure your streams are in sync and you do not end up with CTI violations – whereby you try to insert a CTI with a timestamp greater than an already enqueued event. I took a simple approach with a singleton to hold state which both streams would call a method of to advance time.

The third option is to use AdvancedTimeImport. This allows us to synchronize two streams – the ‘slower’ stream importing CTIs from the ‘faster’ stream. An example is shown below.

#region Time Settings
 
var atgs = new AdvanceTimeGenerationSettings(1, TimeSpan.FromTicks(1),true);
var ats = new AdvanceTimeSettings(atgs, null, AdvanceTimePolicy.Adjust);
 
//These settings will import CTIs from the Risk Stream
var timeImportSettings = new AdvanceTimeSettings(null, 
                            new AdvanceTimeImportSettings("TradeStream"), AdvanceTimePolicy.Adjust);
#endregion
 
#region Streams
 
//Create the inputstreams
 
var tStream = CepStream<Trade>.Create(
             "TradeStream",
             typeof(TradeAdapteFactory),
             tconfig,
             EventShape.Interval
             ,ats
             );
 
//Import timeimport from trade
var rStream = CepStream<Risk>.Create(
             "RiskStream",
             typeof(RiskAdapteFactory),
             rconfig,
             EventShape.Interval//,timeImportSettings
             ,timeImportSettings
 
             );
 
#endregion

In my example there is a subtlety here – which stream is the ‘slow’ one? From a logical perspective I should expect Trade events to arrive before Risk (Risk events are produced elsewhere using Trade as an input). If I consider the cardinality and volume of my data I have 5000 Trades and 800 Risk points. This means that Trade is the fast stream because I will have finished enqueuing Risk before I have finished enqueing all my Trade events. Thus, I generate CTIs in the Risk stream for any new Trades imported, thereby releasing any ‘join events’. Whereas this is fine, and works completely as expected, it forced me to think about the timeliness of the arrival of data in the source streams in real life. The two streams are from different sources and completely independent. Event though Trade is the ‘fast’ stream here what happens if there is a gap in the enqueuing of Trades, perhaps due to a system failure? I may get new Risk events arrive which have an existing Trade match but the join events will not be released. If this is not desirable, and you do not want gaps/delays in the fast stream causing delays on valid events being released then a manual approach to time management is probably preferable.

Performance (and ‘null’ adapters)

One of my goals was to see what sort of ‘raw’ performance I could get with StreamInsight. As I was interested in the raw performance of the engine rather than how well (or badly) I could write input and output adapters I used simple csv source files and accessed them using a StreamReader. For the output adapter I came up with the idea of a ‘NULL adapter’ whereby I dequeue events but then do not do anything further with them (outputting them to console for example, which is what I started with, is just far too slow). I could use the Event Flow Debugger and server.GetDiagnosticView to validate that all of the events had been enqueued and released.

So with two StreamReader Input Adapters, and one ‘Null Adapter’ what sort of performance could I get on my laptop?

Without much (ok, any!) effort I was able to process and release all events within 2.5 seconds. That is 40,000 a second or, from a ‘released vectors’ perspective, 2000! This was way in excess of my expected target (400). This hardly represents a soak test with a steady state but there is an important point I took away from this. The StreamInsight engine is very performant – it is up to you as the adapter writer to try to keep up! I think the ‘null adapter’ approach is a useful way to benchmark the kind of output performance you should be trying to get near too. The reality will always be at least a little slower depending on your event sink (Database, Service, Client etc.).

Injecting Queries

StreamInsight has LINQ support for writing queries.

An aggregate query based upon my join query is shown below. As noted above, it must use SnapshotWindow in order to produce the ‘event based result’ I needed.

//SUM by CounterParty over a snapshot to get some aggregates
            var CPartyQuery = from c in joinedrisk
                              group c by c.CounterParty into CpGroup
                              from window in CpGroup.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
                              select new { Sum = window.Sum(e => e.Amount), CounterParty=CpGroup.Key };

This is great, but it quickly led me to think about how can you bind queries to the engine – to support a scenario of having CEP ‘as a service’ which clients can bind to, for example. After some poking around there is some support for this. There is a property of a QueryTemplate (Definition) which returns an XML definition of the query. An example query is shown below.

<QueryTemplate Name="cep:/Server/Application/RiskAggregateService/QueryTemplate/filterLogic" Description="filtered" xmlns="http://schemas.microsoft.com/ComplexEventProcessing/2010/01/Metadata">
  - <Import Name="filterInput" Type="cep:/Server/Application/RiskAggregateService/EventType/RealTimeRiskCEPApplication.Events.Risk%2C%20RealTimeRiskCEPApplication%2C%20Version%3D1.0.0.0%2C%20Culture%3Dneutral%2C%20PublicKeyToken%3Dnull">
    <OutputStream Name="Import.2.0" />
  </Import>
  - <Export Name="OutputAdapter">
    <InputStream Name="Select.1.1" />
  </Export>
  - <Select Name="Select.1.1">
    <InputStream Name="Import.2.0" />
    <OutputStream Name="Select.1.1" />
    - <FilterExpression>
      - <Equal>
        <InputField Name="RiskId" StreamName="Import.2.0" />
        <Constant Type="System.Int32" Nullable="false" Value="1" />
      </Equal>
    </FilterExpression>
  </Select>
</QueryTemplate>

Once you have this, you can bind queries from XML, as per the below.

//Create the query template, and bind it
 
var dummyTemplate1 = application.CreateQueryTemplate("dummyTemplate1", "Description...", CepStream<TradeDetail>.Create("input1"));
 
var dummyTemplate2 = application.CreateQueryTemplate("dummyTemplate2", "Description...", CepStream<Risk>.Create("input2"));
 
XmlReader riskFilter = XmlReader.Create(fileDir + "RealTimeRiskCEPApplication\\RiskJoin.xml");
 
QueryTemplate filterQT = application.CreateQueryTemplate(riskFilter);
 
QueryBinder queryBinder = new QueryBinder(filterQT);
 
//Bind to the input adapters
 
queryBinder.BindProducer<Risk>("RiskStream", RiskInputAdapter, rconfig, EventShape.Interval, timeImportSettings);
 
queryBinder.BindProducer<TradeDetail>("TradeStream", TradeInputAdapter, tconfig, EventShape.Interval);          
 
//Bind to the output
queryBinder.AddConsumer("ConsoleOutput", ConsoleOutAdapter, conbaseriskconfig, EventShape.Point, StreamEventOrder.FullyOrdered);
 
Query basequery = application.CreateQuery("BaseRisk", "Base risk join query", queryBinder);

One wrinkle I found is that you have to have a couple of dummy query templates with the same event type as you use in BindProducer. Otherwise it does not work.

It is also worth noting that there are some limitations to this query binding technique – lambda expressions in the query being a notable one. It is also worth noting that none of this (including the XML structure for queries) was documented, even through it is in the API.

In Summary

In summary, SQL Server StreamInsight is a highly performant CEP engine. The adapter model is relatively straightforward. Generally speaking the engine will be as performant as the adapters you write for it - The ‘null adapter’ approach is worth considering just to get a baseline on the hardware you are running. The semantics of time, and understanding the affects of CTI injection on result liveliness are probably the most important concepts to get to grips with. And I found the Event Flow Debugger essential in understanding what was going on in the engine when results were not as I was expecting.

It is possible to inject queries at runtime, though this was not that obvious. Nonetheless the support is there for developing a service layer which allows clients bind the queries which there are interested in. The results to these could then be published using a message based technology to clients.

Written by Terry Room

Hadoop Binary Streaming and F# MapReduce

MCS UK Solution Development — Sun, 08 Jan 2012 10:38:40 GMT

As mentioned in my previous post Hadoop Streaming not only supports text streaming, but it also supports Binary Streaming. As such I wanted to put together a sample that supports processing Office documents. As before the code can be downloaded from:

http://code.msdn.microsoft.com/Hadoop-Streaming-and-F-f2e76850

Putting together this sample involved a bit more work than the text streaming case as one needed to put together an implementation of the Java classes to support binary streaming; namely FileInputFormat and RecordReader. The purpose of these implementations is to support Binary Streaming of the document such that it is not split on the usual line boundaries. More on the Java code later.

The implemented Java classes are written, with a key value type pairing of <Text, BytesWritable>, which writes out the data for the mapper in the following format:

Filename
Tab character
UTF8 Encoded Document
Linefeed character

The Mapper code will get called with this format for each document in the specified input directories.

Map and Reduce Classes

The goal of the sample code is to support a Map and Reduce with the following prototypes:

Map : WordprocessingDocument –> seq<string * obj>

Reduce : string -> seq<string> -> obj option

The idea is the Mapper takes in a Word document and projects this into a sequence of keys and values. The Reducer, as in the text streaming case, takes in a key value pair and returns an optional reduced value.

The use of the obj type is, as in the text streaming case, to support serializing the output values.

As an example, I have put together a MapReduce to process Word documents, where the word document is mapped into the number of pages per author and where multiple authors are credited with the same number of pages.

The sample Mapper code is as follows:

module OfficeWordPageMapper =

    let dc = XNamespace.Get("http://purl.org/dc/elements/1.1/")

    let cp = XNamespace.Get("http://schemas.openxmlformats.org/package/2006/metadata/core-properties")

    let unknownAuthor = "unknown author"

    let getAuthors (document:WordprocessingDocument) =

        let coreFilePropertiesXDoc = XElement.Load(document.CoreFilePropertiesPart.GetStream()); 

        // Take the first dc:creator element and split based on a ";"

        let creators = coreFilePropertiesXDoc.Elements(dc + "creator")

        if Seq.isEmpty creators then

            [| unknownAuthor |]

        else

            let creator = (Seq.head creators).Value

            if String.IsNullOrWhiteSpace(creator) then

                [| unknownAuthor |]

            else

                creator.Split(';')

    let getPages (document:WordprocessingDocument) =

        // return page count

        Int32.Parse(document.ExtendedFilePropertiesPart.Properties.Pages.Text)

    // Map the data from input name/value to output name/value

    let Map (document:WordprocessingDocument) =

        let pages = getPages document

        (getAuthors document)

        |> Seq.map (fun author -> (author, pages))

As you can see the majority code is needed to pull out the document properties. If one wanted to process the words within the document one would use the following code:

document.MainDocumentPart.Document.Body.InnerText

To run this code it is worth noting there is a dependency on installing the Open XML SDK 2.0 for Microsoft Office.

Finally, the Reducer code is as follows:

module OfficePageReducer = 

    let Reduce (key:string) (values:seq<string>) =

        let totalPages =

            values |> 

            Seq.fold (fun pages value -> pages + Int32.Parse(value)) 0

        Some(totalPages)

Again, as in the text streaming application, both the mapper and the reducer are executables that read the input from StdIn and emit the output to StdOut. In the case of the mapper the console application will need to do multiple Console.ReadByte() calls to get the data, and then perform a Console.WriteLine() to emit the output. The reducer will do a Console.ReadLine() to get the data, and perform a Console.WriteLine() to emit the output.

The previous post covers the schematics of creating console applications for F#; so I will not cover this again but assume the same program structure.

Mapper Executable

As previously mentioned the purpose of the Mapper code is to perform Input Format Parsing, Projection, and Filtering. In the Mapper for a Word document the bytes are used to create a WordprocessingDocument, with invalid documents ignored, these are then projected into a key/value sequence using the Map function:

module Controller = 

    let Run (args:string array) =    

        // Define what arguments are expected

        let defs = [

            {ArgInfo.Command="input"; Description="Input File"; Required=false };

            {ArgInfo.Command="output"; Description="Output File"; Required=false } ]

        // Parse Arguments into a Dictionary

        let parsedArgs = Arguments.ParseArgs args defs        

        // Ensure Standard Input/Output and allow for debug configuration

        let builder = new StringBuilder()

        let encoder = Text.UTF8Encoding()

        use reader = 

            if parsedArgs.ContainsKey("input") then

                builder.Append(Path.GetFileName(parsedArgs.["input"])) |> ignore

                File.Open(Path.GetFullPath(parsedArgs.["input"]), FileMode.Open) :> Stream

            else

                let stream = new MemoryStream()

                // Ignore bytes until one hits a tab

                let rec readTab() = 

                    let inByte = Console.OpenStandardInput().ReadByte()

                    if inByte <> 0x09 then                        

                        builder.Append(encoder.GetString([| (byte)inByte |])) |> ignore

                        readTab()

                readTab()

                // Copy the rest of the stream and truncate the last linefeed char

                Console.OpenStandardInput().CopyTo(stream)

                if (stream.Length > 0L) then

                    stream.SetLength(stream.Length - 1L)

                stream.Position <- 0L

                stream :> Stream

        use writer =

            if parsedArgs.ContainsKey("output") then

                new StreamWriter(Path.GetFullPath(parsedArgs.["output"]))

            else

                new StreamWriter(Console.OpenStandardOutput(), AutoFlush = true)

        let filename = builder.ToString()

        // Combine the name/value output into a string

        let outputCollector (outputKey, outputValue) =

            let output = sprintf "%s\t%O" outputKey outputValue

            writer.WriteLine(output)

        // Check we do not have a null document

        if (reader.Length > 0L) then

            try

                // Get access to the word processing document from the input stream

                use document = WordprocessingDocument.Open(reader, false)

                // Process the word document with the mapper

                OfficeWordPageMapper.Map document

                |> Seq.iter (fun value -> outputCollector value)

                // close document

                document.Close()

            with

            | :? System.IO.FileFormatException ->

                // Ignore invalid files formats

                ()

        // Close the streams

        reader.Close()

        writer.Close()

There are a few things of note in the code.

When pulling out the filename of the document that is being processed, it is assumed that UTF8 encoding has been used and that a Tab character is used as a delimiter between the filename and the documents bytes.

In building the Stream that is to be used for creating the WordprocessingDocument a MemoryStream is used. There are several reasons for this. Firstly one needs to remove the last Newline character from the Stream, and secondly when creating a WordprocessingDocument a Stream is needed that supports Seek operations; unfortunately this is not the case for StdIn.

At the moment the code does not use the Filename. However in future posts I will extend the code to also support processing PDF documents.

Reducer Executable

After running the Mapper, the data being parsed into the Reducer will be a key/value pair delimited with a Tab. Using the above Map, a sample input dataset for the Reducer would be:

Brad Sarsfield    44
Carl Nolan    1
Marie West    1
Carl Nolan 9

Thus in this case, the code for the Reducer will be the same as in the text streaming case:

module Controller = 

    let Run (args:string array) = 

        // Define what arguments are expected

        let defs = [

            {ArgInfo.Command="input"; Description="Input File"; Required=false };

            {ArgInfo.Command="output"; Description="Output File"; Required=false } ]

        // Parse Arguments into a Dictionary

        let parsedArgs = Arguments.ParseArgs args defs

        // Ensure Standard Input/Output and allow for debug configuration

        use reader = 

            if parsedArgs.ContainsKey("input") then

                new StreamReader(Path.GetFullPath(parsedArgs.["input"]))

            else

                new StreamReader(Console.OpenStandardInput())

        use writer =

            if parsedArgs.ContainsKey("output") then

                new StreamWriter(Path.GetFullPath(parsedArgs.["output"]))

            else

                new StreamWriter(Console.OpenStandardOutput(), AutoFlush = true)

        // Combine the name/value output into a string

        let outputCollector outputKey outputValue =            

            let output = sprintf "%s\t%O" outputKey outputValue

            writer.WriteLine(output)

        // Read the next line of the input stream

        let readLine() = 

            reader.ReadLine()

        // Parse the input into the required name/value pair

        let parseLine (input:string) = 

            let keyValue = input.Split('\t')

            (keyValue.[0].Trim(), keyValue.[1].Trim())

        // Converts a input line into an option

        let getInput() = 

            let input = readLine()

            if not(String.IsNullOrWhiteSpace(input)) then

                Some(parseLine input)

            else

                None

        // Creates a sequence of the input based on the provided key

        let lastInput = ref None

        let continueDo = ref false

        let inputsByKey key firstValue = seq {

            // Yield any value from previous read

            yield firstValue

            continueDo := true

            while !continueDo do

                match getInput() with

                | Some(input) when (fst input) = key ->

                    // Yield found value and remainder of sequence

                    yield (snd input)                    

                | Some(input) ->

                    // Have a value but different key

                    lastInput := Some(fst input, snd input)

                    continueDo := false

                | None ->

                    // Have no more entries

                    lastInput := None

                    continueDo := false

        }

        // Controls the calling of the reducer

        let rec processInput (input:(string*string) option) =

            if input.IsSome then

                let key = fst input.Value

                let value = OfficePageReducer.Reduce key (inputsByKey key (snd input.Value))

                if value.IsSome then

                    outputCollector key value.Value

                if lastInput.contents.IsSome then

                    processInput lastInput.contents

        processInput (getInput())

Once run, the reduced output would be:

Brad Sarsfield    44
Carl Nolan    10
Marie West    1

Once again the only complexity in the code is the fact the Seq.groupBy function cannot be used. Also, as in the text streaming case there is a fair amount of code controlling the input and output streams for calling the Map and Reduce functions, that can be reused for all Hadoop Binary Streaming jobs.

Running the Executables

In the case of Binary Streaming the command parameters are a little different to the text streaming case:

C:\Apps\dist\hadoop.cmd jar lib/hadoop-streaming-ms.jar

 -input "/office/documents" 

 -output "/office/authors" 

 -mapper "..\..\jars\FSharp.Hadoop.MapperOffice.exe" 

 -combiner "..\..\jars\FSharp.Hadoop.ReducerOffice.exe" 

 -reducer "..\..\jars\FSharp.Hadoop.ReducerOffice.exe" 

 -file "C:\Users\carlnol\Projects\FSharp.Hadoop.MapReduce\FSharp.Hadoop.MapperOffice\bin\Release\FSharp.Hadoop.MapperOffice.exe" 

 -file "C:\Users\carlnol\Projects\FSharp.Hadoop.MapReduce\FSharp.Hadoop.ReducerOffice\bin\Release\FSharp.Hadoop.ReducerOffice.exe" 

 -file "C:\Users\carlnol\Projects\FSharp.Hadoop.MapReduce\FSharp.Hadoop.MapReduce\bin\Release\FSharp.Hadoop.MapReduce.dll" 

 -file "C:\Users\carlnol\Projects\FSharp.Hadoop.MapReduce\FSharp.Hadoop.Utilities\bin\Release\FSharp.Hadoop.Utilities.dll" 

 -inputformat com.microsoft.hadoop.mapreduce.lib.input.BinaryDocumentWithNameInputFormat

The first difference is the use of the hadoop-streaming-ms.jar. This file contains the InputFormat class specified by the inputformat parameter; the later Java Classes section discusses how this is created. This is needed to support Binary Streaming.

One other difference to my previous text streaming case is the use of the Reducer class as a Combiner. As the output types from the mapper are the same as the reducer then this is possible.

Tester Application

For completeness I have included the base code for the tester application; albeit the full code is included in the download. The code is very similar to the tester application mentioned in my previous post, with the exception of how the mapper executable is called:

module MapReduceConsole =

    let Run args =

        // Define what arguments are expected

        let defs = [            

            {ArgInfo.Command="input"; Description="Input File"; Required=true };

            {ArgInfo.Command="output"; Description="Output File"; Required=true };

            {ArgInfo.Command="tempPath"; Description="Temp File Path"; Required=true };

            {ArgInfo.Command="mapper"; Description="Mapper EXE"; Required=true };

            {ArgInfo.Command="reducer"; Description="Reducer EXE"; Required=true }; ]

        // Parse Arguments into a Dictionary

        let parsedArgs = Arguments.ParseArgs args defs

        Arguments.DisplayArgs parsedArgs

        // define the executables

        let mapperExe = Path.GetFullPath(parsedArgs.["mapper"])

        let reducerExe = Path.GetFullPath(parsedArgs.["reducer"])

        Console.WriteLine()

        Console.WriteLine (sprintf "The Mapper file is:\t%O" mapperExe)

        Console.WriteLine (sprintf "The Reducer file is:\t%O" reducerExe)

        // Get the file names

        let inputpath = Path.GetFullPath(parsedArgs.["input"])

        let outputfile = Path.GetFullPath(parsedArgs.["output"])

        let tempPath = Path.GetFullPath(parsedArgs.["tempPath"])

        let tempFile = Path.Combine(tempPath, Path.GetFileName(outputfile))

        let mappedfile = Path.ChangeExtension(tempFile, "mapped")

        let reducefile = Path.ChangeExtension(tempFile, "reduced")

        Console.WriteLine()

        Console.WriteLine (sprintf "The input path is:\t\t%O" inputpath)

        Console.WriteLine (sprintf "The mapped temp file is:\t%O" mappedfile)

        Console.WriteLine (sprintf "The reduced temp file is:\t%O" reducefile)

        Console.WriteLine (sprintf "The output file is:\t\t%O" outputfile)

        // Give the user an option to continue

        Console.WriteLine()

        Console.WriteLine("Hit ENTER to continue...")

        Console.ReadLine() |> ignore

        let CHUNK = 1024

        let buffer:byte array = Array.zeroCreate CHUNK

        // Call the mapper with the input file

        let mapperProcessLoop inputfile =

            use mapper = new Process()

            mapper.StartInfo.FileName <- mapperExe

            mapper.StartInfo.UseShellExecute <- false

            mapper.StartInfo.RedirectStandardInput <- true

            mapper.StartInfo.RedirectStandardOutput <- true

            mapper.Start() |> ignore

            use mapperInput = mapper.StandardInput.BaseStream

            use mapperOutput = mapper.StandardOutput 

            // Map the reader to a background thread so processing can happen in parallel

            let taskMapperFunc() = 

                use mapperWriter = File.AppendText(mappedfile)

                while not mapperOutput.EndOfStream do

                    mapperWriter.WriteLine(mapperOutput.ReadLine())

            let taskMapperWriting = Task.Factory.StartNew(Action(taskMapperFunc)) 

            // Pass the file into the mapper process and close input stream when done          

            use mapperReader = new BinaryReader(File.OpenRead(inputfile))

            let rec readLoop() = 

                let bytesread = mapperReader.Read(buffer, 0, CHUNK)

                if bytesread > 0 then

                    mapperInput.Write(buffer, 0, bytesread)

                    readLoop()

            // Write out a Filename/Tab/Document/LineFeed

            let filename = Path.GetFileName(inputfile)

            let encoding = Text.UTF8Encoding();

            mapperInput.Write(encoding.GetBytes(filename), 0, encoding.GetByteCount(filename))

            mapperInput.Write([| 0x09uy |], 0, 1)

            readLoop()

            mapperInput.Write([| 0x0Auy |], 0, 1)

            mapperInput.Close()

            taskMapperWriting.Wait()

            mapperOutput.Close()

            mapper.WaitForExit()

            let result = match mapper.ExitCode with | 0 -> true | _ -> false

            mapper.Close()

            result

        let mapperProcess() = 

            Console.WriteLine "Mapper Processing Starting..."  

            // Create the output file

            if File.Exists(mappedfile) then File.Delete(mappedfile)

            use mapperWriter = File.CreateText(mappedfile)

            mapperWriter.Close()

            // function to determine if valid document extension

            let isValidFile inputfile = 

                if String.Equals(Path.GetExtension(inputfile), ".docx", StringComparison.InvariantCultureIgnoreCase) ||

                   String.Equals(Path.GetExtension(inputfile), ".pdf", StringComparison.InvariantCultureIgnoreCase) then

                    true

                else

                    false

            // Process the files in the directory

            Directory.GetFiles(inputpath)

            |> Array.filter isValidFile

            |> Array.fold (fun result inputfile -> result && (mapperProcessLoop inputfile)) true

        // Sort the mapped file by the first field - mimic the role of Hadoop

        let hadoopProcess() = 

            Console.WriteLine "Hadoop Processing Starting..."

            let unsortedValues = seq {

                use reader = new StreamReader(File.OpenRead(mappedfile))

                while not reader.EndOfStream do

                    let input = reader.ReadLine()

                    let keyValue = input.Split('\t')

                    yield (keyValue.[0].Trim(), keyValue.[1].Trim())

                reader.Close()

                }

            use writer = File.CreateText(reducefile)

            unsortedValues

            |> Seq.sortBy fst

            |> Seq.iter (fun (key, value) -> writer.WriteLine (sprintf "%O\t%O" key value))

            writer.Close()

        // Finally call the reducer process

        let reducerProcess() =

            use reducer = new Process()

            reducer.StartInfo.FileName <- reducerExe

            reducer.StartInfo.UseShellExecute <- false

            reducer.StartInfo.RedirectStandardInput <- true

            reducer.StartInfo.RedirectStandardOutput <- true

            reducer.Start() |> ignore

            use reducerInput = reducer.StandardInput

            use reducerOutput = reducer.StandardOutput 

            // Map the reader to a background thread so processing can happen in parallel

            Console.WriteLine "Reducer Processing Starting..." 

            let taskReducerFunc() = 

                use reducerWriter = File.CreateText(outputfile)

                while not reducerOutput.EndOfStream do

                    reducerWriter.WriteLine(reducerOutput.ReadLine())

            let taskReducerWriting = Task.Factory.StartNew(Action(taskReducerFunc)) 

            // Pass the file into the mapper process and close input stream when done

            use reducerReader = new StreamReader(File.OpenRead(reducefile))

            while not reducerReader.EndOfStream do

                reducerInput.WriteLine(reducerReader.ReadLine())

            reducerInput.Close()

            taskReducerWriting.Wait()

            reducerOutput.Close()

            reducer.WaitForExit()

            let result = match reducer.ExitCode with | 0 -> true | _ -> false

            reducer.Close()

            result

        // Finish test

        if mapperProcess() then

            Console.WriteLine "Mapper Processing Complete..."  

            hadoopProcess()

            Console.WriteLine "Hadoop Processing Complete..."

            if reducerProcess() then

                Console.WriteLine "Reducer Processing Complete..."

                Console.WriteLine "Processing Complete..."     

        Console.ReadLine() |> ignore

When calling the mapper it is no longer the case that the mapper executable is called once, with each line being redirected into the executables StdIn. In the binary case the mapper executable is called once for each document, where for each document the data is then redirected into the executables StdIn. As mentioned the format defined is the filename, delimitated with a Tab, followed by the documents bytes, and finally the Newline character:

// Pass the file into the mapper process and close input stream when done          

use mapperReader = new BinaryReader(File.OpenRead(inputfile))

let rec readLoop() = 

    let bytesread = mapperReader.Read(buffer, 0, CHUNK)

    if bytesread > 0 then

        mapperInput.Write(buffer, 0, bytesread)

        readLoop()

// Write out a Filename/Tab/Document/LineFeed

let filename = Path.GetFileName(inputfile)

let encoding = Text.UTF8Encoding();

mapperInput.Write(encoding.GetBytes(filename), 0, encoding.GetByteCount(filename))

mapperInput.Write([| 0x09uy |], 0, 1)

readLoop()

mapperInput.Write([| 0x0Auy |], 0, 1)

This code is code for each document found in the directory specified in the input argument.

The previous post discussed the threading and stream processing needed for testing in a little more detail.

Java Classes

To complete the post here is the full listing for the Java class implementations:

FileInputFormat
package com.microsoft.hadoop.mapreduce.lib.input;

import java.io.IOException;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.io.BytesWritable;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.FileSplit;

import org.apache.hadoop.mapred.FileInputFormat;

import org.apache.hadoop.mapred.InputSplit;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.RecordReader;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapreduce.InputFormat;

import org.apache.hadoop.mapreduce.JobContext;

public class BinaryDocumentWithNameInputFormat

    extends FileInputFormat<Text, BytesWritable> {

    public BinaryDocumentWithNameInputFormat() {

        super();

    }

    protected boolean isSplitable(FileSystem fs, Path filename) {

        return false;

    }

    @Override

    public RecordReader<Text, BytesWritable> getRecordReader(

            InputSplit split, JobConf job, Reporter reporter) throws IOException {

        return new BinaryDocumentWithNameRecordReader((FileSplit) split, job);        

    }

}

RecordReader
package com.microsoft.hadoop.mapreduce.lib.input;

import java.io.IOException;

import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.BytesWritable;

import org.apache.hadoop.io.DataOutputBuffer;

import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.RecordReader;

import org.apache.hadoop.mapred.FileSplit;

import org.apache.hadoop.mapred.Reporter;

public class BinaryDocumentWithNameRecordReader implements RecordReader<Text, BytesWritable> {

    private FileSplit fileSplit;

    private Configuration conf;

    private boolean processed = false;

    public BinaryDocumentWithNameRecordReader(FileSplit fileSplit, Configuration conf)

        throws IOException {

        this.fileSplit = fileSplit;

        this.conf = conf;

    }

    @Override

    public Text createKey() {

        return new Text();

    }

    @Override

    public BytesWritable createValue() {

        return new BytesWritable();

    }

    @Override

    public long getPos() throws IOException {

        return this.processed ? this.fileSplit.getLength() : 0;

    }

    @Override

    public float getProgress() throws IOException {

        return this.processed ? 1.0f : 0.0f;

    }

    @Override

    public boolean next(Text key, BytesWritable value) throws IOException {

        if (!this.processed) {

            byte[] contents = new byte[(int) this.fileSplit.getLength()];

            Path file = this.fileSplit.getPath();

            FileSystem fs = file.getFileSystem(this.conf);

            FSDataInputStream in = null;

            try {

                in = fs.open(file);

                in.readFully(contents, 0, contents.length);

                key.set(file.getName());

                value.set(contents, 0, contents.length);

            }

            finally {

                in.close();

            }

            this.processed = true;

            return true;

        }

        else {

            return false;

        }

    }

    @Override

    public void close() throws IOException {

    }

}

Once the Java classes have been compiled they need to be merged into a copy of the hadoop-streaming.jar file. In my case I have created a file called hadoop-streaming-ms.jar. This file is a copy of the base file into which I have copied the classes, as the JAR file is just a ZIP file; although one can also use the JAR tool. One just has to remember to use the package path:

com\microsoft\hadoop\mapreduce\lib\input

To use this new streaming package the file needs to be copied to the Hadoop install lib directory; in my case:

C:\Apps\dist\lib

If you download the code there is also an implementation of these classes that support a key value type pairing of <NullWritable, BytesWritable>; rather than the <Text, BytesWritable> key value type pairing that is used to pass in the documents filename. This can used used if the document’s filename is not needed.

Conclusion

Hopefully the code is once again useful if you intend to write any Binary Streaming applications of documents. If you download the sample code, in addition to support for processing of Microsoft Word documents there is support for processing PDF documents.

Written by Carl Nolan

Hadoop Streaming and F# MapReduce

MCS UK Solution Development — Thu, 22 Dec 2011 22:50:32 GMT

As you may know Microsoft has recently announced plans for a Hadoop adoption for both Windows Server and Windows Azure. You can find out more about Hadoop and Windows Azure at Apache Hadoop-based Services for Windows Azure and Availability of Community Technology Preview (CTP) of Hadoop based Service on Windows Azure.

If you are not familiar with MapReduce then there are some useful resources for F# and MapReduce that would also be worth reading Exploring MapReduce with F# and Parsing Log Files with F#, MapReduce and Windows Azure.

As Hadoop is written in Java the main integration point for MapReduce for .Net developers is through Hadoop Streaming. Hadoop streaming being a utility that allows you to create and run MapReduce jobs with any executable or script as the mapper and/or the reducer. You can find out everything you want to know about Hadoop Streaming at: http://hadoop.apache.org/common/docs/current/streaming.html

MapReduce Code

Before starting it is worth noting that the complete code for this blog post can be found at:

http://code.msdn.microsoft.com/Hadoop-Streaming-and-F-f2e76850

When writing a Hadoop Streaming MapReduce job it will quickly become apparent that there is a codebase that one can reuse. As such, I thought it would be useful to put together this codebase to enable F# developers to write MapReduce libraries through a simple API. The idea was to provide reusable code such that one only needed to be concerned with implementing the MapReduce code with the following function prototype’s:

Map : string –> (string * obj) option

Reduce : string -> seq<string> –> obj option

The idea is that the Hadoop text input is processed and each input line is passed into the Map function which parses and filters the key/value pair for the data. The values are then sorted and merged, by Hadoop. The processed mapped data is then passed into the Reduce function, as a key and corresponding sequence of strings, which then defines the optional output value.

So why the use of the obj type? Hadoop Streaming is based on text data, albeit a binary interface is also available. Thus the inputs into the MapReduce are strings. However, when performing the MapReduce operations strings are not always suitable, but they do need to be able to be represented as strings. This is normally handled through serialization but in this case I have used sprintf with a “%O” pattern. Thus any type needs to have a meaningful ToString() implementation such that the data can once again be parsed back into a workable type.

As a sample consider the following data:

11075    19:07:56    en-US    iPhone OS    Apple    iPhone 4.2.1    Georgia    United States        0    0

11081    01:46:19    en-US    Android    Samsung    SCH-i500    Colorado    United States    4.2620937    0    0

11086    04:07:25    en-US    Android    Unknown    Android 2.3    California    United States        0    0

11090    03:34:59    en-US    iPhone OS    Apple    iPod Touch 4.3.x    Hawaii    United States        0    0

11095    19:34:47    en-US    Android    Samsung    SCH-i500    Illinois    United States    0.4621525    1    0

11095    02:31:19    en-US    Android    Samsung    SCH-i500    Nebraska    United States    1.2662282    0    0

11095    02:31:21    en-US    Android    Samsung    SCH-i500    Nebraska    United States    0.2905647    0    1

11095    19:34:49    en-US    Android    Samsung    SCH-i500    Illinois    United States    1.3336967    1    1

11097    10:22:41    en-US    iPhone OS    Apple    iPhone 4.0    Illinois    United States        0    0

11102    12:54:27    en-US    Android    Samsung    SCH-i400    Florida    United States        0    0

11106    12:54:25    en-GB    Windows Phone    HTC    7 Trophy    Greater London    United Kingdom    9.94    2    0

11106    08:50:46    en-GB    Windows Phone    HTC    7 Trophy    Greater London    United Kingdom    3.12    0    0

11106    11:07:31    en-GB    Windows Phone    HTC    7 Trophy    Greater London    United Kingdom    15.538    1    0

11106    11:13:27    en-GB    Windows Phone    HTC    7 Trophy    Greater London    United Kingdom    1.5066558    1    1

11106    11:13:28    en-GB    Windows Phone    HTC    7 Trophy    Greater London    United Kingdom        1    2

11112    00:42:52    en-US    iPhone OS    Apple    iPhone 4.2.x    Illinois    United States    18.1075543    0    0

11112    00:43:14    en-US    iPhone OS    Apple    iPhone 4.2.x    Illinois    United States    2.6342826    0    1

11130    10:48:20    en-GB    iPhone OS    Apple    iPhone 4.2.1    Greater London    United Kingdom        0    0

11131    12:19:52    en-US    Android    Unknown    Android 2.3    Massachusetts    United States        0    0

This represents some mobile phone data with the 2nd column representing the query time and the 4th column representing the platform device. A simple MapReduce could be:

Pull the Device Platform and Query Time from the data:

Mapper
module MobilePhoneQueryMapper =

    // Performs the split into key/value

    let private splitInput (value:string) =

        try

            let splits = value.Split('\t')

            let devicePlatform = splits.[3]

            let queryTime = TimeSpan.Parse(splits.[1])

            Some(devicePlatform, box queryTime)

        with

        | :? System.ArgumentException -> None

    // Map the data from input name/value to output name/value

    let Map (value:string) =

        splitInput value

Calculate the Min, Average, and Max Query Times.

Reducer
module MobilePhoneQueryRangeReducer = 

    let Reduce (key:string) (values:seq<string>) =

        let initState = (TimeSpan.MaxValue, TimeSpan.MinValue, 0L, 0L)

        let (minValue, maxValue, totalValue, totalCount) =

            values |> 

            Seq.fold (fun (minValue, maxValue, totalValue, totalCount) value ->

                (min minValue (TimeSpan.Parse(value)), max maxValue (TimeSpan.Parse(value)), totalValue + (int64)(TimeSpan.Parse(value).TotalSeconds), totalCount + 1L) ) initState

        Some(box (minValue, TimeSpan.FromSeconds((float)(totalValue/totalCount)), maxValue))

The box function is used to ensure the types returned from the MapReduce calls is of the obj type.

So onto the executable code.

As this is Hadoop Streaming applications, both the mapper and the reducer are executables that read the input from StdIn (line by line) and emit the output to StdOut. Thus one just needs a console application that does a Console.ReadLine() to get the data, and perform a Console.WriteLine() to emit the output.

As with all F# Console applications the MainEntry point is defined as:

module Program =   

    [<EntryPoint>]

    let Main(args) = 

        Controller.Run args

        // main entry point return

        0

Each executable then contains a Controller module that is run to process the data.

Mapper Executable

The purpose of the Mapper code is to perform Input Format Parsing, Projection, and Filtering. The code to perform this is as follows:

module Controller = 

    let Run (args:string array) =    

        // Define what arguments are expected

        let defs = [

            {ArgInfo.Command="input"; Description="Input File"; Required=false };

            {ArgInfo.Command="output"; Description="Output File"; Required=false } ]

        // Parse Arguments into a Dictionary

        let parsedArgs = Arguments.ParseArgs args defs              

        // Ensure Standard Input/Output and allow for debug configuration

        use reader = 

            if parsedArgs.ContainsKey("input") then

                new StreamReader(Path.GetFullPath(parsedArgs.["input"]))

            else

                new StreamReader(Console.OpenStandardInput())

        use writer =

            if parsedArgs.ContainsKey("output") then

                new StreamWriter(Path.GetFullPath(parsedArgs.["output"]))

            else

                new StreamWriter(Console.OpenStandardOutput(), AutoFlush = true)

        // Combine the name/value output into a string

        let outputCollector (outputKey, outputValue) =

            let output = sprintf "%s\t%O" outputKey outputValue

            writer.WriteLine(output)

        // Read the next line of the input stream

        let readLine() = 

            reader.ReadLine()

        // Define the input sequence

        let rec inputs() = seq {

            let input = readLine()

            if not(String.IsNullOrWhiteSpace(input)) then

                // Yield the input and the remainder of the sequence

                yield input

                yield! inputs()

        }

        // Process the lines from the stream and pass into the mapper

        inputs()

        |> Seq.map MobilePhoneQueryMapper.Map

        |> Seq.filter Option.isSome

        |> Seq.iter (fun value -> outputCollector value.Value)

A majority of this code provides a means for specifying input and output files to better aid testing (more on this in a bit).

The code boils down to the last three lines which performs the main functions:

inputs()

|> Seq.map MobilePhoneQueryMapper.Map

|> Seq.filter Option.isSome

|> Seq.iter (fun value -> outputCollector value.Value)

This parses the input, maps the data into a key value pairs, filters out unwanted rows, and outputs the data using the outputCollector function.

The outputCollector function takes the processed key/value pair and outputs them to the correct stream. In the Java API the types used for output are based on a Writable interface; rather than Java Serialization. In this implementation the outputCollector takes the obj.ToString(); through the use of sprintf(). Thus if the provided types does not have a useful ToString() you will have to create the string representation before calling the outputCollector. The outputCollector performs the formatting of the key/value pair into a Tab delimited string; as required for Hadoop Streaming.

Reducer Executable

After running the Mapper, the data being parsed into the Reducer will be a key/value pair delimited with a Tab. Using the aforementioned sample data and Map, a sample, but selective, input dataset for a Reducer would be:

Android    18:54:20
Android    19:19:44
Android    19:19:46
RIM OS    17:19:36
RIM OS    17:17:18
RIM OS    00:44:41
Windows Phone    12:54:25
Windows Phone    08:50:46
Windows Phone    11:13:28
iPhone OS    19:07:56
iPhone OS    03:34:59
proprietary development    14:29:20
proprietary development    14:30:17

The processing of the mapped data within the Reducer is a little more complex than the Mapper. The idea is that the data is grouped by the input key and the resulting sequences are passed into the Reduce function; a function call for each key along with the corresponding sequence.

Whereas the Seq.groupBy can perform this operation, the groupBy function makes no assumption on the ordering of the original sequence. As a consequence the resulting sequence is not lazily evaluated, and is thus not suitable for large sequences; an absolute must for Hadoop MapReduce jobs. The code thus has to create a lazily evaluated sequence for each input key. This can be achieved as one knows the input data is sorted.

To achieve this the processing has to be state dependant to handle the transition from one key value to the next. The input data is processed in such a fashion that any change in key value causes a transition to the next sequence. The state persistence is needed to track the key change, and to ensure the first yield of the new sequence is the value that caused this transition; transition values are not lost.

The code for the Reducer is as follows:

module Controller = 

    let Run (args:string array) = 

        // Define what arguments are expected

        let defs = [

            {ArgInfo.Command="input"; Description="Input File"; Required=false };

            {ArgInfo.Command="output"; Description="Output File"; Required=false } ]

        // Parse Arguments into a Dictionary

        let parsedArgs = Arguments.ParseArgs args defs

        // Ensure Standard Input/Output and allow for debug configuration

        use reader = 

            if parsedArgs.ContainsKey("input") then

                new StreamReader(Path.GetFullPath(parsedArgs.["input"]))

            else

                new StreamReader(Console.OpenStandardInput())

        use writer =

            if parsedArgs.ContainsKey("output") then

                new StreamWriter(Path.GetFullPath(parsedArgs.["output"]))

            else

                new StreamWriter(Console.OpenStandardOutput(), AutoFlush = true)

        // Combine the name/value output into a string

        let outputCollector outputKey outputValue =            

            let output = sprintf "%s\t%O" outputKey outputValue

            writer.WriteLine(output)

        // Read the next line of the input stream

        let readLine() = 

            reader.ReadLine()

        // Parse the input into the required name/value pair

        let parseLine (input:string) = 

            let keyValue = input.Split('\t')

            (keyValue.[0].Trim(), keyValue.[1].Trim())

        // Converts a input line into an option

        let getInput() = 

            let input = readLine()

            if not(String.IsNullOrWhiteSpace(input)) then

                Some(parseLine input)

            else

                None

        // Creates a sequence of the input based on the provided key

        let lastInput = ref None

        let continueDo = ref false

        let inputsByKey key (firstValue:string option) = seq {

            // Yield any value from previous read

            if firstValue.IsSome then

                yield firstValue.Value

            continueDo := true

            while !continueDo do

                match getInput() with

                | Some(input) when (fst input) = key ->

                    // Yield found value and remainder of sequence

                    yield (snd input)                    

                | Some(input) ->

                    // Have a value but different key

                    lastInput := Some(fst input, snd input)

                    continueDo := false

                | None ->

                    // Have no more entries

                    lastInput := None

                    continueDo := false

        }

        // Controls the calling of the reducer

        let rec processInput (input:(string*string) option) =

            if input.IsSome then

                let key = fst input.Value

                let value = MobilePhoneQueryReducer.Reduce key (inputsByKey key (Some(snd input.Value)))

                if value.IsSome then

                    outputCollector key value.Value

                if lastInput.contents.IsSome then

                    processInput lastInput.contents

        processInput (getInput())

The code also contains the same testing structure to allow one to pass in an input and output file; once again to aid in debugging.

The code uses a string.Split() to derive the tuple of key/value pair; as this is how the Mapper writes out the data. Once the data has been Reduced, the outputCollector once again performs the formatting of the key/value data into a Tab delimited string, sending the text to the StdOut stream. The value in this case is the reduced data corresponding to the key.

The output for the sample data and Reduce sample code, listed above, would be:

Android    (00:00:06, 12:54:39, 23:59:54)
RIM OS    (00:01:07, 13:52:56, 23:59:58)
Unknown    (00:00:36, 10:29:27, 23:52:36)
Windows Phone    (00:00:32, 12:38:31, 23:55:17)
iPhone OS    (00:00:01, 11:51:53, 23:59:50)
proprietary development    (14:29:20, 14:29:44, 14:30:17)

The value corresponding to the key is the string representing of a tuple of type (TimeSpan*TimeSpan*TimeSpan).

As you can see, there is a fair amount of code controlling the input and output streams for calling the Map and reduce functions; that can be reused for all Hadoop Streaming jobs.

Testing the Code

So now the code has been put together how can it be tested? Debugging a job once it has been deployed to a Hadoop cluster is not an easy task. As such one is much better off if testing can be performed without Hadoop in the picture. Whereas this will not cover all test cases, as some issues will only be found when deployed to a cluster, it does provide a means to cover most test scenarios.

Unit Testing the individual map and Reduce functions is relatively straight forward. However performing testing on sample input data is a little trickier.

To assist in testing with data files I have put together a Tester application. This application:

Defines and executes a Process for the Mapper executable in which
- The StdIn is modified to be the file specified in the “–input” command line argument
- the StdOut is modified to be a file with a “mapper” extension
When the Mapper has completed, Sorts the output file from the Mapper into a file with a “reducer” extension
When the Sort is complete, defines and executes a Process for the Reducer executable in which
- The StdIn is modified to be the sorted “reducer” file
- The StdOut is modified to be the file specified in the “–output” command line argument

Running the Tester application allows one to check inputs and outputs, in a flow similar to running within Hadoop. The full listing for the Tester application can be found in the download code:

http://code.msdn.microsoft.com/Hadoop-Streaming-and-F-f2e76850

Whereas this allows for simple data testing there is an easier option when one needs to attach a debugger. Bother the Mapper and Reducer executables accept a “–input” and “–output” command line arguments. These options define the files to be used, rather than the Standard Input/Output streams, for data processing. Using these arguments one can easily debug the executable with a set of test data and view the output. To create a Reducer input, for testing, one can run the Tester application with the required input data, and use the output file with a “reducer” extension as the input.

One final thing that has been useful for testing is the concept of a Null controller. This is merely an executable that just takes the input and passes it to the output stream:

module ControllerNull = 

    let Run (args:string array) = 

        let rec inputs() =

            let input = Console.ReadLine()

            if not(String.IsNullOrWhiteSpace(input)) then

                Console.WriteLine input

                inputs()

        inputs()

This Null controller is useful when deploying your application to a Hadoop cluster to ensure data is being input and output correctly.

Hadoop and F#

I am not going to say much about running the Hadoop job other than to show the command to run the Streaming job:

C:\Apps\dist\bin>hadoop.cmd jar ../lib/hadoop-streaming.jar -D keep.failed.task.file=true

 -input "/mobile/debug/sampledata.txt"

 -output "/mobile/querytimesdebug"

 -mapper "..\..\jars\FSharp.Hadoop.Mapper.exe"

 -reducer "..\..\jars\FSharp.Hadoop.Reducer.exe"

 -file "C:\MyDirectory\FSharp.Hadoop.MapReduce\FSharp.Hadoop.Mapper\bin\Debug\FSharp.Hadoop.Mapper.exe"

 -file "C:\MyDirectory\FSharp.Hadoop.MapReduce\FSharp.Hadoop.Reducer\bin\Debug\FSharp.Hadoop.Reducer.exe"

 -file "C:\MyDirectory\FSharp.Hadoop.MapReduce\FSharp.Hadoop.MapReduce\bin\Debug\FSharp.Hadoop.MapReduce.dll"

The sample download code includes all the necessary command line code to run the job, and also includes a sample data file and some commands to copy this data to the cluster.

There is however a point to be made about using F#, namely the dependency on the F# Runtime. Importantly the F# Runtime contains many useful functions and types, including APIs for collections such as lists, arrays, maps, sets and sequences. One should install the F# Runtime on all the nodes in the cluster. The installer can be found at:

Visual Studio 2010 F# 2.0 Runtime SP1

If one is running the job on a server on which the FSharp Runtime has not been deployed there is another option. One can copy the appropriate runtime file to the temp execution directory. To achieve this one needs to add the following file option to the execution command:

-file C:\Program Files (x86)\Reference Assemblies\Microsoft\FSharp\2.0\Runtime\v4.0\FSharp.Core.dll

Conclusion

To conclude, don’t forget to check out Channel9 for more information:

http://channel9.msdn.com/Events/windowsazure/learn/Learn-about-Hadoop-on-Windows-Azure-with-Alex-Stojanovic

Written by Carl Nolan

Mutual authentication with a IIS hosted WCF data service installed in a workgroup environment

MCS UK Solution Development — Wed, 14 Dec 2011 14:41:13 GMT

This post covers the steps required to secure communication between a WCF client and a WCF data service using mutual certificate authentication.

The client/service topology is depicted below:

Both the client and server run on a Windows Server 2008 R2 virtual machine with Windows SDK 7.1, Visual Studio 2010, SQL Server Express 2008 R2 and .NET framework 4.0.

Mutual authentication explained

A detailed discussion of public key cryptography is outside the scope of this article. However, I highly recommend that you read more here to get a better understanding of the underlying principles, so that you can better understand the ‘How-To’ steps that follow.

In essence the client possesses a public key certificate (hereafter known as a client certificate) with associated private key. Similarly, the service has possession of another public key certificate (hereafter known as a service certificate) with associated private key. For a given pair of keys, either the public or private key can be used to encrypt data but ONLY the corresponding key can decrypt the data. The public key is available to the world at large but the private key MUST remain with the owner alone. In this way the owner is able to encrypt data using the private key and a party in receipt of encrypted data can decrypt it using the public key. The fact that the receiving party can successfully decrypt the data using the public key means that the data MUST have been encrypted by the corresponding private key and, as only the owner has access to the private key, the data MUST have originated from the owner; this is known as nonrepudiation as the owner cannot claim that encrypted data did not originate from them (of course, the private key may have been stolen but that is a different matter).

Translating this to the current scenario, in order for the client and service to successfully converse using certificates, the WCF client application must have access to the client’s private key and IIS (the receiving party) must have access to the client’s public key. Conversely, IIS must have access to the service private key and the client must have access to the service public key.

In an Active Directory domain, the public key certificate is ordinarily assigned to a domain account such that IIS can locate the certificate using Active Directory Domain Services (ADDS). However, ADDS does not exist in a workgroup environment and so the client certificate public key list must be maintained within IIS itself.

Methodology

1) Create a trusted root certificate

The certificates in this article are issued by a certificate authority, in this case a trusted root certificate created using the makecert tool that ships with the Windows SDK.

Create a folder somewhere on the file system and, in a command window, change the directory to match the newly created folder
Run the following command to create a root certificate (do not define a password)
- "C:\Program Files (x86)\Microsoft SDKs\Windows\v7.0A\Bin\makecert" -r -pe -n "CN=Test Root Authority" -ss CA -sr CurrentUser -a sha1 -sky signature -cy authority -sv TestRoot.pvk TestRoot.cer

2) Install the client certificate with private key

The client application, as owner of the client certificate, needs access to the private key at runtime. We are going to create an interactive client (e.g. – Console, WinForm, Silverlight, Internet Explorer etc.), and so the certificate will be installed in the Current User – Personal store. If the client was non-interactive (e.g. – NT service) the certificate would be installed in the Local Machine – Personal store.

Additionally, as the client certificate has been issued from a trusted certificate authority (CA), the root certificate created in 1) must also be installed.

From the folder where TestRoot.cer is located, create and install a client certificate by running the following in a command window
- "C:\Program Files (x86)\Microsoft SDKs\Windows\v7.0A\Bin\makecert" -n "CN=Test Client" -ic "TestRoot.cer" -iv "TestRoot.pvk" -a sha1 -sky exchange -pe -sr currentuser -ss my "TestClient.cer"

Open a snap-in window
- Start – Run – mmc
- Add/Remove snap-in
- Ctrl-M
- Add a Certificates snap-in for ‘My user account’

Expand the (Trusted Root Certification Authorities)/Certificates node

Right click the Certificates folder and choose All Tasks – Import

Browse to TestRoot.cer

Click through the remaining windows and finish

3) Install the service certificate with private key

The service, as owner of a service certificate, needs access to the private key at runtime. As the service is non-interactive (i.e. – it is not running under the context of a logged-on user), the certificate will be installed in the Local Machine – Personal store.

Additionally, as the service certificate has been issued from a trusted certificate authority (CA), the root certificate for the CA must also be installed so that the chain of trust can be verified.

From the folder where TestRoot.cer is located, install the service certificate by running the following in a command window
- "C:\Program Files (x86)\Microsoft SDKs\Windows\v7.0A\Bin\makecert" -n "CN=www.mydataservice.com" -ic "TestRoot.cer" -iv "TestRoot.pvk" -a sha1 -sky exchange -eku 1.3.6.1.5.5.7.3.1 -pe -sr LocalMachine -ss my TestDataService.cer

Open a snap-in window
- Start – Run – mmc
- Add/Remove snap-in
- Ctrl-M
- Add a Certificates snap-in for ‘Computer account’

Expand the (Trusted Root Certification Authorities)/Certificates node

Right click the Certificates folder and choose All Tasks – Import

Browse to TestRoot.cer

Click through the remaining windows and finish

4) Export the client public key certificate

IIS needs access to the client’s public key certificate. In this section the certificate is exported for later use.

Open a snap-in window (or use the same window as 2))
- Start – Run – mmc
- Add/Remove snap-in
- Ctrl-M
- Add a Certificates snap-in for ‘My user account’

Expand the (Certificates – Current User)/Personal/Certificates node

Right click the Test Client certificate and choose All Tasks – Export (Do not export the primary key)
- N.B. - Exporting the private key will not cause problems but remember that the private key is for the client ALONE

Save the certificate as TestClient_Public.cer (Base-64 encoded X.509)

5) Export the service public key certificate

The WCF client needs access to the service public key certificate. In this section the certificate is exported for later use.

Open a snap-in window (or use the same window as 3))

Start – Run – mmc

Add/Remove snap-in

Ctrl-M

Add a Certificates snap-in for ‘Computer account’

Expand the (Certificates – Local Computer)/Personal/Certificates node

Right click the www.testdataservice.com certificate and choose All Tasks – Export (Do not export the primary key)
- Exporting the private key will not cause problems but remember that the private key is for the service ALONE

Save the certificate as TestDataService_Public.cer (Base-64 encoded X.509)

6) Create a WCF data service

In this section a basic WCF data service is created in order to secure it using mutual authentication.

In Visual Studio 2010, create a new empty solution called MyDataServices

Add a new ASP.NET empty web application called MyDataService

Add a new empty ADO.NET Entity Data Model item to MyDataService called MyDataServiceModel.edmx (empty model)

Add a new WCF Data Service item to MyDataService called MyDataServiceImp.svc

Open the code behind file for MyDataServiceImp.svc and add MyDataServiceModelContainer into the class definition

Open MyDataServiceModel.edmx

Right click anywhere on the screen and choose to Generate Database from Model

Create a new data connection to SQL Server if one does not already exist

Build the solution

7) Host the data service in IIS

Open IIS manager

Start – Administrative Tools – Internet Information Services (IIS) Manager

Create a new website called ‘My Data Service’

Point the physical path to the data service created in 6)

Choose a binding type of https and select the www.mydataservice.com SSL certificate

Modify the app pool under which the website is running so that the .NET framework version is v4.0.30319

Add a hosts file entry for ‘127.0.0.1 www.mydataservice.com’

Via Windows Explorer, give ‘Everyone’ full control of the MyDataService folder
- N.B. – this is a shortcut for keeping the blog post succinct. Such permissions should not be an option for a production deployment

8) Create a WCF client

In order to make life a little easier, we will create and configure the WCF client before locking down the service.

Add a new Console Application project called MyDataServiceClient to MyDataServices

Right click the project and choose to ‘Add Service Reference’

Enter an address of https://www.mydataservice.com/mydataserviceimp.svc

Click Go

Once the service has been discovered give it the namespace MyDataServiceEntities

9) Enable client certificate authentication in IIS

Both IIS and the WCF data service must be configured for client certificate authentication, otherwise there is a mismatch and a service activation failure similar to ‘The SSL settings for the service 'SslRequireCert' does not match those of the IIS 'Ssl'’ will occur.

Open IIS manager
- Start – Administrative Tools – Internet Information Services (IIS) Manager

Highlight the WCF data service website and double click ‘SSL Settings’

Configure IIS as below

10) Register the client public key certificate with the service

In this section the public key certificate from 4) is registered with the data service website.

Client certificates can be mapped in a ‘one to one’ or ‘many to one’ relationship with Windows accounts. ‘one to one’ means that a client certificate is linked to a single Windows account. ‘many to one’ means that many client certificates are assigned to a single Windows account.

In this post we will use the ‘one to one’ relationship but the ‘many to one’ relationship does not differ greatly

Highlight the MyDataService website and double click ‘Configuration Editor’

Choose the ‘oneToOneMappings’ section

Click Add

Set the entry as ‘enabled’ = true

Set the username with associated password as that of the local administrator

In production this would most likely be a service account with reduced privileges

Right click the TestClient_Public.cer file from 4) and choose to edit in Notepad

Ensure that the block between, but not including, ‘-----BEGIN CERTIFICATE-----‘ and ‘-----END CERTIFICATE-----‘, is all on a single line (it is not by default)

Copy and paste the block into the ‘certificate’ field so that the resulting entry is as below

Close the window

Click Apply

11) Mandate client certificate authentication for the data service

The data service must also be configured to mandate client certificate authentication.

In web.config, overwrite the <system.serviceModel> element with the following

Code Snippet
 <system.serviceModel>
  <bindings>
    <webHttpBinding>
      <binding name="mutual">
        <security mode="Transport">
          <transport clientCredentialType="Certificate" />
        </security>
      </binding>
    </webHttpBinding>
  </bindings>
  <behaviors>
    <endpointBehaviors>
      <behavior name="mutual">
        <webHttp/>
      </behavior>
    </endpointBehaviors>
  </behaviors>
  <serviceHostingEnvironment aspNetCompatibilityEnabled="true" />
  <services>
    <service name="MyDataService.MyDataServiceImp">
      <endpoint contract="System.Data.Services.IRequestHandler"
                binding="webHttpBinding"
                bindingConfiguration="mutual"
                behaviorConfiguration="mutual">
      </endpoint>
    </service>
  </services>
</system.serviceModel>
 

The contents of the <webHttpBinding> element mandate that client certificate authentication at the transport level is required for the binding. The <services>/<service>/<endpoint> element describes the contract that all WCF data services must adhere to.

12) Register the service public key certificate with the client

As this article is concerned with mutual authentication then it is not sufficient for only the service to authenticate the client. In addition, the client must be certain that it is calling the right endpoint. One way to achieve this is to store the service public key certificate with the client so that during runtime, the SSL certificate presented by the service can be compared against the locally stored version. In this way the client can be sure that it is calling the correct SSL endpoint located at www.mydataservice.com and that it has not been phished into calling another endpoint with the same URL but different SSL certificate.

Open a snap-in window
- Start – Run – mmc
- Add/Remove snap-in
- Ctrl-M
- Add a Certificates snap-in for ‘My user account’

Expand the (Certificates – Current User)/Personal/Certificates node

Right click the Certificates folder and choose All Tasks – Import

Browse to TestDataService_Public.cer that was created in 6)

Click through the remaining windows and finish

13) Update the WCF client to verify the service certificate

As the service is now locked down, the client will not be able to use it without undergoing some modification.

Overwrite Program.cs with the following

Code Snippet
 namespace MyDataServiceClient
{
    using System;
    using System.Collections.Generic;
    using System.Net;
    using System.Net.Security;
    using System.Security.Cryptography.X509Certificates;
 
    class Program
    {
        private static MyDataServiceEntities.MyDataServiceModelContainer entities = new MyDataServiceEntities.MyDataServiceModelContainer(new Uri(@"https://www.mydataservice.com/mydataserviceimp.svc"));
 
        static void Main(string[] args)
        {
            X509Store store = new X509Store(StoreName.My, StoreLocation.CurrentUser);
 
            store.Open(OpenFlags.ReadOnly);
 
            X509Certificate2Collection certColl = store.Certificates.Find(X509FindType.FindBySubjectDistinguishedName, "CN=Test Client", false);
 
            entities.ClientCertificate = certColl[0];
 
            store.Close();
 
            ServicePointManager.ServerCertificateValidationCallback +=
                delegate(object sender, X509Certificate cert, X509Chain chain, SslPolicyErrors sslError)
                {
                    // Validate the certificate
                    store = new X509Store(StoreName.My, StoreLocation.CurrentUser);
 
                    store.Open(OpenFlags.ReadOnly);
 
                    X509Certificate2Collection serverCertColl = store.Certificates.Find(X509FindType.FindBySubjectDistinguishedName, "CN=www.mydataservice.com", false);
 
                    X509Certificate2 serverCert = serverCertColl[0];
 
                    store.Close();
 
                    // Make this as rigorous as necessary
                    return serverCert.Thumbprint == (new X509Certificate2(cert)).Thumbprint;
                };
 
            // Send the query to the data service context
            IEnumerable<object> result = entities.Execute<object>(entities.BaseUri);
 
            Console.WriteLine();
            Console.WriteLine("**********");
            Console.WriteLine("Success!!!");
            Console.WriteLine("**********");
 
            Console.ReadKey();
        }
    }
}
 

Note that the ClientCertificate property of the proxy is not recognised; this represents the last piece of the puzzle. As there is no declarative configuration for the WCF client (i.e. - no <system.serviceModel> element was created in the client config file, unlike for, say, a standard SOAP based web service), the client certificate is hooked in programmatically by intercepting the underlying HttpWebRequest instance.

Overwrite Reference.cs contained within the MyDataServiceEntities service reference, with the following

Code Snippet

namespace MyDataServiceClient.MyDataServiceEntities
{
    using System.Data.Services.Client;
    using System.Net;
    using System.Security.Cryptography.X509Certificates;
 
    public partial class MyDataServiceModelContainer : global::System.Data.Services.Client.DataServiceContext
    {
        private X509Certificate clientCertificate = null;
 
        public MyDataServiceModelContainer(global::System.Uri serviceRoot) :
            base(serviceRoot)
        {
            this.OnContextCreated();
        }
 
        partial void OnContextCreated();
 
        public X509Certificate ClientCertificate
        {
            get
            {
                return clientCertificate;
            }
            set
            {
                if (value == null)
                {
                    // if the event has been hooked up before, we should remove it
                    if (clientCertificate != null)
                        this.SendingRequest -= this.OnSendingRequest_AddCertificate;
                }
                else
                {
                    // hook up the event if its being set to something non-null
                    if (clientCertificate == null)
                        this.SendingRequest += this.OnSendingRequest_AddCertificate;
                }
 
                clientCertificate = value;
            }
        }
 
        private void OnSendingRequest_AddCertificate(object sender, SendingRequestEventArgs args)
        {
            if (null != ClientCertificate)
                (args.Request as HttpWebRequest).ClientCertificates.Add(ClientCertificate);
        }
    }
}

Set MyDataServiceClient as the startup project

Press F5

Conclusion

In this article I have demonstrated a methodology for securing communication between a WCF client and a IIS hosted WCF data service, installed on a workgroup server, using mutual certificate authentication. While in many ways similar to mutual authentication scenarios described elsewhere, securing a WCF data service exhibits some unique characteristics that warranted documenting.

I hope you find this useful.

Written by Bradley Cotier

Adventures in TSQL: SQL Server Query Performance Analysis using DMVs

MCS UK Solution Development — Sun, 27 Nov 2011 09:50:16 GMT

From the development perspective I often have to perform an analysis of a database application. More often than not this entails looking at a running system and ensuring that the application queries are behaving as expected. As such, I thought it would be worthwhile sharing some TSQL scripts that I have been using over the years for tuning SQL Server queries. So what I am generally looking for are:

Long Running Queries – which queries are taking the longest to run
CPU Intensive Queries – which queries are consuming the most CPU per execution
Worst Case Total CPU – which queries are consuming the most CPU across all executions
Most IO Intensive Queries – which queries consume the most read/write operations

The idea behind looking at these queries is to analyse the execution plan and review whether the query would benefit from being refactored or whether there are indexes that would enable the query to perform better.

The basis of this analysis are some queries based on the SQL Server Dynamic Management Views (DMVs). For performing CPU based analysis, I have been using the following DMV based query:

CPU Query
-- Which Queries are taking the most time/cpu to execute
 SELECT TOP 20
     total_worker_time, total_elapsed_time,
     total_worker_time/execution_count AS avg_cpu_cost, execution_count,
     (SELECT DB_NAME(dbid) + ISNULL('..' + OBJECT_NAME(objectid), '')
         FROM sys.dm_exec_sql_text([sql_handle])) AS query_database,
     (SELECT SUBSTRING(est.[text], statement_start_offset/2 + 1,
         (CASE WHEN statement_end_offset = -1
             THEN LEN(CONVERT(nvarchar(max), est.[text])) * 2
             ELSE statement_end_offset
             END - statement_start_offset) / 2
         )
         FROM sys.dm_exec_sql_text([sql_handle]) AS est) AS query_text,
     total_logical_reads/execution_count AS avg_logical_reads,
     total_logical_writes/execution_count AS avg_logical_writes,
     last_worker_time, min_worker_time, max_worker_time,
     last_elapsed_time, min_elapsed_time, max_elapsed_time,
     plan_generation_num, qp.query_plan
 FROM sys.dm_exec_query_stats
     OUTER APPLY sys.dm_exec_query_plan([plan_handle]) AS qp
 WHERE [dbid] >= 5 AND DB_NAME(dbid) IS NOT NULL
   AND (total_worker_time/execution_count) > 100
 --ORDER BY avg_cpu_cost DESC;
 --ORDER BY execution_count DESC;
 ORDER BY total_worker_time DESC;

The basis of this query is that it shows the top 20 queries that consume the most CPU resources. The query can be modified to show queries that have consumed the most worker time or the most worker time per execution. It is also sometimes useful to see which queries have been executed the most, as you will want this to be highly performant. Just changing the ORDER BY allows one to tweak the queries one wants to analyse.

This query provides 2 critical pieces of information for query analysis. Firstly the query_text is returned so one can see what has been executed. The other critical piece of information is the actual query plan. This is returned as an XML document. Having the XML is useful in itself, however the fact one can save the XML as a file with a .sqlplan extension makes this XML invaluable.

Once the .sqlplan file has been saved, opening this file in SQL Server Enterprise Manager provides one with a graphical interpretation of the executed query. This enables one to easily analyse the actual query execution plan.

The other useful, but similar query, is the one to analyse IO based metrics:

Disk IO Query
SELECT TOP 20
     total_logical_reads/execution_count AS avg_logical_reads,
     total_logical_writes/execution_count AS avg_logical_writes,
     total_worker_time/execution_count AS avg_cpu_cost, execution_count,
     total_worker_time, total_logical_reads, total_logical_writes, 
     (SELECT DB_NAME(dbid) + ISNULL('..' + OBJECT_NAME(objectid), '')
         FROM sys.dm_exec_sql_text([sql_handle])) AS query_database,
     (SELECT SUBSTRING(est.[text], statement_start_offset/2 + 1,
         (CASE WHEN statement_end_offset = -1
             THEN LEN(CONVERT(nvarchar(max), est.[text])) * 2
             ELSE statement_end_offset
             END - statement_start_offset
         ) / 2)
         FROM sys.dm_exec_sql_text(sql_handle) AS est) AS query_text,
     last_logical_reads, min_logical_reads, max_logical_reads,
     last_logical_writes, min_logical_writes, max_logical_writes,
     total_physical_reads, last_physical_reads, min_physical_reads, max_physical_reads,
     (total_logical_reads + (total_logical_writes * 5))/execution_count AS io_weighting,
     plan_generation_num, qp.query_plan
 FROM sys.dm_exec_query_stats
     OUTER APPLY sys.dm_exec_query_plan([plan_handle]) AS qp
 WHERE [dbid] >= 5 AND DB_NAME(dbid) IS NOT NULL
   and (total_worker_time/execution_count) > 100
 ORDER BY io_weighting DESC;
 --ORDER BY avg_logical_reads DESC;
 --ORDER BY avg_logical_writes DESC;
 --ORDER BY avg_cpu_cost DESC;

Once again one can tweak the ORDER BY to analyse queries which consume the most read or write operations. For performing a quick IO based query analysis I have found it useful to calculate the total query IO based on a weighting of reads to writes; weighted IO = (writes * weighting) + reads. In this case I have provided a weighting that assumes a read is 5 times more expensive than a write.

Hope you find these DMV based queries useful, as they have held me in good stead over the years.

Written by Carl Nolan

Creating a Partitioned View in the BAM Archiving Database

MCS UK Solution Development — Tue, 15 Nov 2011 22:09:36 GMT

When you run the BAM data maintenance package (BAM_DM_<activity name>) BAM copies each partition in the BAM Primary Import database to a separate table in the BAM Archive database. You can create partitioned views in the BAM Archive database to facilitate locating the data. However one is left to create these partitioned views oneself.

A version of a script that can be used to create these partitioned views can be found on the MSDN site:

http://msdn.microsoft.com/en-us/library/aa562047.aspx

Whereas this version of the script will work when the BAM activities are first deployed, one will get issues if one modifies the BAM configuration by adding new activity items. The sample script works by creating a view that does a UNION ALL of SELECT * FROM BAM_DM_<activity partition>.

This works fine as long as the activity definition is unchanged, but the view falls over if a new item is added. This is because the schema of the archive partitions is no longer consistent. There are also issues for failed partitions as they are not excluded from the partitioned view.

However, one can easily resolve this issue by using explicit column names. As activities are augmented with new items the base table definition in the BAM Archive database is modified such that it represents the full items list for the activity. Using this one is able to determine the columns required for the partitioned view.

Thus when creating a column list for an archived partition one has to determine one can match the columns with those needed for the view. If a column is not present in a partitioned table then the column definition is modified to be:

NULL AS [column_name]

This ensures consistency for all SELECT statements that make up the partitioned view.

Here is a full listing of a stored procedure one can use to create a new partitioned view:

USE [BAMArchive]
 GO
 
 CREATE PROCEDURE [dbo].[CreateBamActivityView]
 (
     @activityName        nvarchar(64),
     @viewType            nvarchar(64)    = 'Instances'
 )
 AS
 BEGIN
     SET NOCOUNT ON;
    
     DECLARE @partitionName  nvarchar(128);
     DECLARE @tableName      nvarchar(128);
     DECLARE @viewName       nvarchar(128);
     DECLARE @templateName   nvarchar(128);
     DECLARE @columnNames    nvarchar(max);
     DECLARE @schema         nvarchar(12);
     DECLARE @isFirstTable   bit;
 
     DECLARE @dropScript        nvarchar(max);
     DECLARE @createScript    nvarchar(max);
     DECLARE @newLine        nvarchar(12);
 
     DECLARE @likeData        nvarchar(128);
     DECLARE @likeDefault    nvarchar(128);
     DECLARE @likeFailed        nvarchar(128);
     DECLARE @likeUsage        nvarchar(128);
 
     SET @newLine = CHAR(13) + CHAR(10);
     SET @schema = 'dbo';
 
     SET @templateName = N'bam_' + @activityName + N'_' + @viewType;
     SET @viewName = N'[' + @schema + '].[bam_' + @activityName + '_' + @viewType + 'View]';
     SET @likeDefault = N'bam[_]' + @activityName + N'[_]' + @viewType;
 
     SET @likeData =  @likeDefault + N'[_]%';
     SET @likeFailed = @likeDefault + N'[_]%[_]Failed';
 
     -- Define a table of the expected columns based on the table template
     DECLARE @activityColumns TABLE (activity_columnname varchar(256));
 
     INSERT INTO @activityColumns
     SELECT [COLUMN_NAME] FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = @schema AND TABLE_NAME = @templateName;
 
     -- See if activity has data otherwise create default view
     IF EXISTS(SELECT 1 FROM sys.sysobjects WHERE [name] LIKE @likeData AND [name] NOT LIKE @likeFailed AND [type] = N'U')
     BEGIN
         SET @likeUsage = @likeData;
     END
     ELSE
     BEGIN
         SET @likeUsage = @likeDefault;
     END;
 
     SET @isFirstTable = 1;
 
     SET @dropScript = N'IF  EXISTS (SELECT * FROM sys.views WHERE object_id = OBJECT_ID(N''' + @viewName + '''))' 
         + @newLine + '    DROP VIEW ' + @viewName + ';'
 
     -- Define a cursor to iterate through created table
     DECLARE instance_cursor CURSOR LOCAL FOR
     SELECT [name] FROM sys.sysobjects 
     WHERE [name] LIKE @likeUsage AND [name] NOT LIKE @likeFailed AND [type] = N'U';
 
     OPEN instance_cursor;
     FETCH NEXT FROM instance_cursor INTO @partitionName;
 
     WHILE @@fetch_status = 0 
     BEGIN
         IF (@partitionName IS NOT NULL)
         BEGIN
             SET @tableName = N'[' + @schema + '].[' + @partitionName + N']';
         
             IF (@isFirstTable = 1)
             BEGIN
                 SET @createScript = N'CREATE VIEW ' +  @viewName + N' AS ' + @newLine;
                 SET @isFirstTable = 0;
             END
             ELSE
             BEGIN
                 SET @createScript = @createScript + @newLine + N'    UNION ALL' + @newLine;
             END
 
             -- Calculate the column defintions based on the working partition
             SET @columnNames = '';
             SELECT @columnNames = @columnNames +
                 CASE
                     WHEN (SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = @schema AND TABLE_NAME = @partitionName AND COLUMN_NAME = activity_columnname) IS NOT NULL THEN '[' + RTRIM(activity_columnname) + ']'
                     ELSE 'NULL AS [' + RTRIM(activity_columnname) + ']'
                 END + ', '
                 FROM @activityColumns
                 OPTION (FAST 1);
 
             IF (LEN(@columnNames) > 0) SET @columnNames = SUBSTRING(@columnNames, 1, LEN(@columnNames) -1);
             
             -- Add the table into the view definition
             SET @createScript = @createScript + N'    SELECT ' + @columnNames + ' FROM ' + @tableName + ' WITH (NOLOCK)';
         END;        
 
         FETCH NEXT FROM instance_cursor INTO @partitionName;
     END
 
     IF (@createScript IS NOT NULL)
     BEGIN
         SELECT @createScript = @createScript + ';';
     END;
 
     CLOSE instance_cursor;
     DEALLOCATE instance_cursor;
 
     -- Display commands to be executed
     PRINT @dropScript;
     PRINT @newLine;
     PRINT @createScript;
 
     -- Execute the creation
     EXEC(@dropScript);
 
     IF (@createScript IS NOT NULL)
     BEGIN
         EXEC(@createScript);
     END;
         
 END
 GO

To create a new partitioned view for an activity one merely has to call the stored procedure with the activity name:

Consider a simple activity definition for Audit:

CREATE TABLE [dbo].[bam_Audit_Instances] (
     [RecordID] [bigint] NOT NULL,
     [ActivityID] [nvarchar](128) NOT NULL,
     [Audited] [datetime] NULL,
     [AuditReference] [nvarchar](38) NULL,
     [AuditAction] [nvarchar](50) NULL,
     [Caller] [nvarchar](50) NULL,
     [AuditPreChange] [nvarchar](1000) NULL,
     [AuditPostChange] [nvarchar](1000) NULL,
     [LastModified] [datetime] NULL,
     
     PRIMARY KEY CLUSTERED 
     (
         [RecordID] ASC
     )
 )

A partitioned view could be created for the activity using:

EXEC dbo.CreateBamActivityView 'Audit'

If the activity was changed over time, with the addition of the pre and post change columns, the view that would be created would be:

SELECT [RecordID], [ActivityID], [Audited], [AuditReference], [AuditAction], [Caller],
     [AuditPreChange], [AuditPostChange], [LastModified]
     FROM [dbo].[bam_Audit_Instances_20111111] WITH (NOLOCK)
 UNION ALL
 SELECT [RecordID], [ActivityID], [Audited], [AuditReference], [AuditAction], [Caller],
     [AuditPreChange], [AuditPostChange], [LastModified]
     FROM [dbo].[bam_Audit_Instances_20111112] WITH (NOLOCK)
 UNION ALL
 SELECT [RecordID], [ActivityID], [Audited], [AuditReference], [AuditAction], [Caller],
     NULL AS [AuditPreChange], NULL AS [AuditPostChange], [LastModified]
     FROM [dbo].[bam_Audit_Instances_20111012] WITH (NOLOCK)
 UNION ALL
 SELECT [RecordID], [ActivityID], [Audited], [AuditReference], [AuditAction], [Caller],
     NULL AS [AuditPreChange], NULL AS [AuditPostChange], [LastModified]
     FROM [dbo].[bam_Audit_Instances_20111011] WITH (NOLOCK);

To use the stored procedure that creates this view one merely has to schedule the procedures execution after each archive run.

Hope you find this useful.

Written by Carl Nolan

Adventures in TSQL: Comma separated string from column values

MCS UK Solution Development — Fri, 04 Nov 2011 10:42:54 GMT

It seems that several times now I have had the requirement to create a comma separated string from column values. The latest reason for doing this was to build up a list of table column names such that a view could be generated for the table.

This is surprisingly easy to do in TSQL. One merely has to define a varchar variable and build up the comma separated string from within a SELECT statement; as this sample demonstrates:

USE [AdventureWorks]
 GO
 
 DECLARE @schema varchar(128) = 'Sales';
 DECLARE @tableName varchar (128) = 'SalesOrderHeader';
 
 DECLARE @columnNames varchar(max) = '';
 
 SELECT @columnNames = @columnNames + '[' + COLUMN_NAME + '], '
     FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = @schema AND TABLE_NAME = @tableName
     OPTION (FAST 1);
     
 IF (LEN(@columnNames) > 0) SET @columnNames = SUBSTRING(@columnNames, 1, LEN(@columnNames) -1);
 
 PRINT @columnNames

The output from this is the @columnNames variable is:

[SalesOrderID], [RevisionNumber], [OrderDate], [DueDate], [ShipDate], [Status], [OnlineOrderFlag], [SalesOrderNumber], [PurchaseOrderNumber], [AccountNumber], [CustomerID], [ContactID], [SalesPersonID], [TerritoryID], [BillToAddressID], [ShipToAddressID], [ShipMethodID], [CreditCardID], [CreditCardApprovalCode], [CurrencyRateID], [SubTotal], [TaxAmt], [Freight], [TotalDue], [Comment], [rowguid], [ModifiedDate]

The logic for this can even get more complicated where only certain values from the table are selected for inclusion in the comma separated string.

In the case of the AdventureWorks database many tables have a column named “rowguid”. If one wanted to exclude this from the list one would write:

SELECT @columnNames = @columnNames +
     CASE
         WHEN COLUMN_NAME = 'rowguid' THEN ''
         ELSE '[' + COLUMN_NAME + '], '
     END
     FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = @schema AND TABLE_NAME = @tableName
     OPTION (FAST 1);

The ability to define a variable value in this fashion is not well known, but nevertheless a useful feature.

Written by Carl Nolan

Preventing Build Hangs and File Locking on Build Servers When Running Unit Tests (revised)

MCS UK Solution Development — Wed, 02 Nov 2011 21:02:41 GMT

Recently I was tasked with changing our daily and Continuous Integration builds so that they would also execute our unit tests. This seemed like a straightforward task, and indeed it was. Except for one thing: every so often a build would fail because another process was locking the test results directory.

When we looked at the build server we saw a number of instances of a process called DW20 running on the server. This process was locking the test results directory. Terminating these processes allowed further builds that included unit tests to run.

DW20 is the Windows Error Reporting program. What was happening was that Windows Error Reporting was asking for permission to send error reports to Microsoft, and waiting for a response from a user. Being on a server this was never going to happen, especially as the prompt could not be seen!

So, two questions needed to be addressed. The first was to find out what was causing the errors that were triggering Windows Error Reporting in the first place, and the second was to see if there was a way to stop DW20 waiting for a prompt when there was an error.

The easiest one to solve was to prevent DW20 waiting and locking the test results directory. On Windows Server 2008 you can configure Windows Error Reporting to not wait for input from the user. To find it, run Server Manager and then click on Turn on Windows Error Reporting (you may need to scroll down to find it). As you can see from the screenshot below, on our build server it was configured to ask about sending reports every time there is an error. Choose either the first or the last option to be sure that DW20 will not wait and lock your test results directory.

On Windows Server 2003 you get to a similar setting through the System applet in Control Panel, as shown in the screenshot below (with thanks to Jean-Pierre de Tiege from Charteris):

The other problem was to work out why we were getting these errors in the first place. For this I used the Windows Sysinternals Process Explorer tool. Using this I hovered the mouse over the DW20 processes, this showed me the command line parameter that was being passed to each DW20 process. This pointed me to a file in a temporary directory which contained the following:

       Version=131072

      General_AppName=QTAgent32.exe

      EventType=VSTEExecutionFrameworkUE

      LoggingFlags=0

      UIFlags=1

      EventLogSource=Team Test Error Reporting

      UI LCID=127

      P1=QTAgent32.exe

      P2=v2.0.50727

      P3=10.0.0.0

      P4=Unknown

      P5=ArgumentNullException

      P6=27F84F9B

      FilesToDelete=C:\Users\build\AppData\Local\Temp\tmp7C26.tmp|C:\Users\build\AppData\Local\Temp\tmp7C29.tmp|C:\Users\build\AppData\Local\Temp\tmp7C3A.tmp

      ReportingFlags=15

      Main_Intro_Bold=An unexpected condition has occurred.

      Main_Intro_Reg=An unexpected condition has occurred in the test execution framework. Information about the condition has been gathered.

      Main_Plea_Bold=Please tell Microsoft about this problem.

      Main_Plea_Reg=We have created an error report that you can send to help us fix bugs. We will treat this report as confidential and anonymous.

      Queued_EventDescription=An exception has occurred in the test execution framework component: Value cannot be null.

      Parameter name: certificateFindKey

The important thing to note in this file is that the process which crashed, was running QTAgent32.exe. This is the test runner used by the TestToolsTask task to run unit tests on the build server without requiring Visual Studio to be installed.

This immediately reminded me that two of the unit tests were failing with an “Error” state rather than the more usual “Failed” state. When the test runner reports an Error state for a test it means that the test caused an error in the test runner itself. The most common reason I have seen for this is in tests that use threads, and indeed in this case the problem tests were using threads. Removing those tests stopped the QTAgent32 errors from happening in the first place.

So, the conclusion is simple.

Watch out for unit tests which fail with an Error state, an fix them as a matter of urgency.
Make sure that on all of your build servers Windows Error Reporting is not configured to prompt.

I hope this has been useful and helps you to avoid problems running unit tests as part of your daily and Continuous Integration builds.

Written by Rob Jarratt

FSharpChart release supporting Stacked Charts and Markers (version 0.60)

MCS UK Solution Development — Tue, 01 Nov 2011 22:19:15 GMT

FSharpChart now supports binding for Stacked Charts and for modifying the Marker associated with a series.

As always one can download the latest release from:

http://code.msdn.microsoft.com/FSharpChart-b59073f5

In previous versions of FSharpChart it was not very intuitive how a Stacked Chart could be plotted. What one had to do was define a CombinedChart where one defines the charts to be combined with the same StackedGroupName property value:

let rndStacked = new System.Random()
 let dataX() = [ for f in 1 .. 10 -> rndStacked.NextDouble() * 10.0]
 
 FSharpChart.Combine
   [ FSharpChart.StackedBar100(dataX(), StackedGroupName = "g1")
     FSharpChart.StackedBar100(dataX(), StackedGroupName = "g1")
     FSharpChart.StackedBar100(dataX(), StackedGroupName = "g1") ]

This displays the following chart:

(Thanks to Tomas Petricek for this sample)

This approach also allows one to easily combine multiple Stacked Charts using different StackedGroupName properties:

FSharpChart.Combine
   [ FSharpChart.StackedBar(dataX(), StackedGroupName = "g1")
     FSharpChart.StackedBar(dataX(), StackedGroupName = "g1")
     FSharpChart.StackedBar(dataX(), StackedGroupName = "g1")
     FSharpChart.StackedBar(dataX(), StackedGroupName = "g2")
     FSharpChart.StackedBar(dataX(), StackedGroupName = "g2") ]

This displays the following chart:

However, for Stacked Charts, a more intuitive approach would be just to specify a list of data series. With this new release one can now specify the data series using the following formats:

list<list<'TY>>
 list<list<'TX * 'TY>>

This now allows one to define, say a StackedColumn chart, using the much simpler expression:

let rndStacked = new System.Random()
 let dataXY() = [ for f in 1 .. 10 -> (f, round (rndStacked.NextDouble() * 100.0))]
 
 [dataXY(); dataXY(); dataXY()]
 |> FSharpChart.StackedColumn
 |> FSharpChart.WithSeries.DataPoint(Label="#VAL")

This displays the following chart:

Hopefully you will agree this is much simpler and more intuitive. The only current limitation of this binding approach is that one cannot use a CombinedChart for multiple Stacked Charts when binding in this fashion. If this is the desire then the previous approach with distinct StackedGroupName properties is needed.

The download also includes a new WinForms sample demonstrating how bindings can be managed for these Stacked Charts.

thanks to a community suggestion I have also added support for modifying the Marker associated with a series. To modify the Marker for a Data Series one can now write:

[ for i in 1. .. 20. .. 1000.0 -> i, rnd.NextDouble() ]
 |> FSharpChart.Line
 |> FSharpChart.WithSeries.Marker(Color=Color.Red, Style=MarkerStyle.Cross)

Again, this displays the chart:

Don’t forget the package is also available from NuGet:

http://www.nuget.org/List/Packages/MSDN.FSharpChart.dll

Once again enjoy!

Written by Carl Nolan

String extension methods rather than using RegEx

MCS UK Solution Development — Wed, 19 Oct 2011 20:31:58 GMT

In a recent project I was using regular expressions a lot for validating string expressions such that statements like the following were true:

Must be all characters
Must be all alphanumeric’s and be of length 6
Must be all numeric digits (e.g. “3648”)
Must be all upper case characters

However in reflection the char data type checks, such as IsDigit(), along with extension methods offer an alternative implementation. What I wanted to write was validations such as:

mystring.IsAllLetterOrDigit();
 mystring.IsAllLetter(3);
 mystring.IsAllLower();

The implementation to support such expressions is:

/// <summary>
 /// Determines if all characters are letters.
 /// </summary>
 public static bool IsAllLetter(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsLetter, length);
 }
 
 /// <summary>
 /// Determines if all characters are digits.
 /// </summary>
 public static bool IsAllDigit(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsDigit, length);
 }
 
 /// <summary>
 /// Determines if all characters are letters or digits.
 /// </summary>
 public static bool IsAllLetterOrDigit(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsLetterOrDigit, length);
 }
 
 /// <summary>
 /// Determines if all characters are numbers.
 /// </summary>
 public static bool IsAllNumber(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsNumber, length);
 }
 
 
 /// <summary>
 /// Determines if all characters are lower characters.
 /// </summary>
 public static bool IsAllLower(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsLower, length);
 }
 
 /// <summary>
 /// Determines if all characters are upper characters.
 /// </summary>
 public static bool IsAllUpper(this string value, [Optional] int length)
 {
     return IsAllValidation(value, char.IsUpper, length);
 }
 
 /// <summary>
 /// Determines if all characters in a string satisfy the character check.
 /// </summary>
 private static bool IsAllValidation(string value, Func<char, bool> charFunc, int length)
 {
     // Check have values
     if (length < 0) throw new ArgumentOutOfRangeException("Length must be positive.");
 
     // Check have more than whitespace
     if (string.IsNullOrWhiteSpace(value)) return false;
 
     // Check all string characters satisfy condition
     if (!value.All<char>(charFunc)) return false;
     
     // If neccessary check strign length
     if (length > 0 && value.Length != length) return false;
 
     return true;
 }

This provides a variety of checks all based on the char IsXXX() methods.

The IsAllValidation() is the actual implementation for all extension methods. As a parameter it takes the char based method for which each char in the string is to be checked. This works based on the fact that the String Type supports IEnumerable of Type Char. We can then use the LINQ All<char>() method to ensure that each char in the string satisfies the required condition.

The length check is an optional check specified using the Optional Attribute on the length parameter. If a value greater than zero is specified then the string is also validated to ensure it has the specified length.

As you can see, with a small amount of code one can very easily extend char methods to support checking each char in a string.

Enjoy!

Written by Carl Nolan

Preventing Build Hangs and File Locking on Build Servers When Running Unit Tests

MCS UK Solution Development — Wed, 12 Oct 2011 13:07:02 GMT

Note : An updated version of this post can be found here.

       Version=131072

      General_AppName=QTAgent32.exe

      EventType=VSTEExecutionFrameworkUE

      LoggingFlags=0

      UIFlags=1

      EventLogSource=Team Test Error Reporting

      UI LCID=127

      P1=QTAgent32.exe

      P2=v2.0.50727

      P3=10.0.0.0

      P4=Unknown

      P5=ArgumentNullException

      P6=27F84F9B

      FilesToDelete=C:\Users\build\AppData\Local\Temp\tmp7C26.tmp|C:\Users\build\AppData\Local\Temp\tmp7C29.tmp|C:\Users\build\AppData\Local\Temp\tmp7C3A.tmp

      ReportingFlags=15

      Main_Intro_Bold=An unexpected condition has occurred.

      Main_Intro_Reg=An unexpected condition has occurred in the test execution framework. Information about the condition has been gathered.

      Main_Plea_Bold=Please tell Microsoft about this problem.

      Main_Plea_Reg=We have created an error report that you can send to help us fix bugs. We will treat this report as confidential and anonymous.

      Queued_EventDescription=An exception has occurred in the test execution framework component: Value cannot be null.

      Parameter name: certificateFindKey

So, the conclusion is simple.

Watch out for unit tests which fail with an Error state, an fix them as a matter of urgency.
Make sure that on all of your build servers Windows Error Reporting is not configured to prompt.

I hope this has been useful and helps you to avoid problems running unit tests as part of your daily and Continuous Integration builds.

Written by Rob Jarratt

FSharpChart new release available (version 0.55)

MCS UK Solution Development — Tue, 11 Oct 2011 22:17:20 GMT

If you have been using FSharpChart you will have seen some recent activities around documentation. I am happy to say that a new code drop has been made available. This release deals with some minor bug fixes and restructures the code download, hopefully making it easier to navigate, in addition to the following changes:

Defines a default font for all charts and chart properties
Provides Style helpers to make it easier to define Labels, Fonts, etc.
Provides a mechanism for specifying global, chart, and type defaults (some of which are defined)
Provides a mechanism for DataPoint tooltips and sets a default for each chart type

First and foremost as a reminder the code can be downloaded from here:

http://code.msdn.microsoft.com/FSharpChart-b59073f5

Before drilling down into the code changes it is worth mentioning the new code structure. Hopefully this will make it simpler in navigating the source and samples. The folders and file contents for the download are:

bin: Contains a compiled DLL for FSharpChart; called MSDN.FSharpChart.dll
scripts: Contains the FSharpChart.fsx script file
source\fsharpchart: Contains all the source files for FSharpChart
source\generator: Contains some fsx files used to generate the original source code
- The Generator.Script.fsx file can be used to create a single FSharpChart.fsx file from all the individual source control files
samples\scripts: Contains some sample script files that demonstrate creating various charts
samples\winforms: Contains a WinForms application that displays a chart and demonstrates how to change properties at runtime
samples\wpf: Contains a WPF application that display a chart using Interop

Default Font

As some folks have expressed a desire to have charts rendered using more modern fonts, the code now contains a default font:

let DefaultFont = 
     new Font("Calibri", 9.0f, FontStyle.Regular)

Style Helpers

When defining a Font the normal definition is as follows:

new Font("Calibri", 9.0f, FontStyle.Regular)

However, with the definition of a default Font it makes sense to allow one to define a new Font where one expresses the differences to the default Font. Thus a StyleHelper method has been provided allowing one to write:

StyleHelper.Font(FontSize=12.0f)

Within the provided FSharpChart samples there are often times where one has to define a LabelStyle, Title, or Legend. These elements all support fonts to be defined. To simplify this process a series of StyleHelper methods are available where one can also just express the Font characteristic that differs from the default. The definition for these helpers is:

type StyleHelper =
 
     static member LabelStyle
         ( ?Angle, ?Color, ?Format, ?Interval, ?IntervalOffset, ?IntervalOffsetType,
           ?IntervalType, ?IsEndLabelVisible, ?IsStaggered, ?TruncatedLabels,
           ?FontName:string, ?FontFamily:FontFamily, ?FontStyle:FontStyle, ?FontSize:float32)
 
     static member Legend
         ( ?Title, ?Background, ?Alignment, ?Docking, ?InsideArea,
           ?FontName:string, ?FontFamily:FontFamily, ?FontStyle:FontStyle, ?FontSize:float32)
 
     static member Title
         ( ?Text, ?TextStyle, ?Background, ?Color, ?BorderColor, ?BorderWidth, ?BorderDashStyle, 
           ?TextOrientation, ?Alignment, ?Docking, ?InsideArea,
           ?FontName:string, ?FontFamily:FontFamily, ?FontStyle:FontStyle, ?FontSize:float32)
 
     static member Font(?FamilyName:string, ?FontFamily:FontFamily, ?FontStyle:FontStyle, ?FontSize:float32)

This allows code such as:

chart.Title <- StyleHelper.Title("Chart Sin/Cosine", FontSize = 12.0f, FontStyle = FontStyle.Bold)
 chart.Legend <- StyleHelper.Legend(InsideArea = false, Alignment = StringAlignment.Center, Docking = Docking.Top, FontName = "Arial Narrow")

Hopefully folks this is an easier syntax for Titles, Legends, LabelStyles, and Fonts.

Charting Defaults

In allowing the modification of all chart Font properties to use the newly defined default a mechanism was defined that allowed any chart property to be overridden with a default. Thus the code now supports the ability to specify a default value for any property which can be restricted to either a particular type (such as Series or Label) or to a particular chart type.

To demonstrate this I have defined a few simple properties that folks have requested:

// Type used for defining defaults
 type internal ChartStyleDefault =
     { ChartType:Charting.SeriesChartType option; ParentType:Type option; PropertyName:string; PropertyDefault:obj }
 
 // Definition of defaults for the chart
 let internal PropertyDefaults = 
     [ // Define type specific defaults
       { ChartStyleDefault.ChartType = None; ParentType = Some(typeof<Charting.LabelStyle>); PropertyName="Font"; PropertyDefault=(box (new Font("Arial Narrow", 10.0f, FontStyle.Regular))) }
       { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Line); ParentType = Some(typeof<Charting.Series>); PropertyName="BorderWidth"; PropertyDefault=(box 2) }
       // Define global defaults
       { ChartStyleDefault.ChartType = None; ParentType = None; PropertyName="Font"; PropertyDefault=(box DefaultFont) } ]

Hopefully this will allow folks to customize FSharpChart with a set of defaults that render charts with desirable visuals.

DataPoint ToolTips

Last but not least ToolTip’s have now been defined for DataPoint’s on a chart.

Support for these ToolTips are provided in several ways.

Firstly these ToolTips are enabled by default with a specific default ToolTip being defined for each chart type:

{ ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Line); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Spline); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Bar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Column); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Area); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedBar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedColumn); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedArea); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedBar100); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedColumn100); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StackedArea100); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.SplineArea); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Range); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, High=#VALY1, Low=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.RangeBar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "Y=#VALX, High=#VALY1, Low=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.RangeColumn); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, High=#VALY1, Low=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.SplineRange); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, High=#VALY1, Low=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Point); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.PointAndFigure); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, High=#VALY1, Low=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.ThreeLineBreak); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.StepLine); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Pie); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Doughnut); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.BoxPlot); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "Lower Whisker=#VALY1, Upper Whisker=#VALY2, Lower Box=#VALY3, Upper Box=#VALY4") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Candlestick); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "High=#VALY1, Low=#VALY2, Open=#VALY3, Close=#VALY4") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Stock); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "High=#VALY1, Low=#VALY2, Open=#VALY3, Close=#VALY4") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Renko); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Bubble); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VALY1, Size=#VALY2") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.ErrorBar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VALY1, Lower=#VALY2, Upper=#VALY3") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Funnel); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Pyramid); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Kagi); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "X=#VALX, Y=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Polar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "Angle=#VALX, Distance=#VAL") }
 { ChartStyleDefault.ChartType = Some(Charting.SeriesChartType.Radar); ParentType = Some(typeof<Charting.Series>); PropertyName="ToolTip"; PropertyDefault=(box "Point=#VALX, Distance=#VAL") }

This ability is supported through the Series ToolTip property. A good set of documentation is available in the charting Technical Reference:

http://msdn.microsoft.com/en-us/library/dd456726.aspx

The Keywords section covers the syntax for the property and the Chart Types section defines what values can and should be used.

Finally, the DataPoint properties for a Series can now be specified using the WithSeries.DataPoint method:

FSharpChart.Line [ for f in 1.0 .. 0.1 .. 10.0 -> f, sin f ]
 |> FSharpChart.WithSeries.DataPoint(ToolTip = "(#VALX{#0.000}, #VAL{#,##0.000;-#,##0.000})")

The supported properties that can be defined are Label, LabelToolTip, and ToolTip.

In Conclusion

Hopefully folks will find these enhancements useful; the ToolTip support having been a highly requested feature. As always please do let me know if there are features that you would like to be included in FSharpChart.

Written by Carl Nolan

Federated Security: How to setup and call a WCF service secured by ADFS 2.0

MCS UK Solution Development — Wed, 17 Aug 2011 13:20:28 GMT

In this blog post I am going to walkthrough the steps required to secure a WCF service with ADFS 2.0, as well as how to call this service from a client. The aim of this article is to provide an outline of the general principals and steps, not to explain how to set up a fully secured scenario (everything in this post was tested across a couple of VMs on a development machine).

With federated security, there are two modes of authentication:

Passive. In this scenario, a client application (i.e. browser) tries to access a web resource (e.g. a webpage). If the client isn’t authenticated for that resource, it is redirected to a login page. On successful authentication, the client is redirected back to the original web resource and authorized as appropriate. Note that in this scenario, the client browser just needs to follow redirects etc. and does not need to use any special logic to authenticate. A more comprehensive explanation can be found here: http://msdn.microsoft.com/en-us/magazine/ff872350.aspx
Active. In this scenario, it is probable that the resource being accessed does not have a user interface (e.g a WCF / Rest service). The client application therefore needs to contain logic to authenticate against a Secure Token Store. Note that WCF does make provision for the active scenario through config (tutorial here: http://msdn.microsoft.com/en-us/gg557876). In our walkthrough below, however, we are going to authenticate programmatically.

This walkthrough was carried out on two VMs.

Server 1: Server 2008R2. Active Directory Domain Controller with IIS7 installed. It is assumed that an AD domain has been set up. This machine will be used to host ADFS2.0.
Server 2: Server 2008R2. IIS7 installed. This machine will be used to host our WCF service (note that this does not need to be joined to our domain).

Server 1: Prepare the default website to host the ADFS 2.0 endpoints.

When installed (later in this post), ADFS 2.0 creates endpoints under the default website. On the first VM (with AD installed), launch IIS Manager and ensure that the default website is running.
Ensure that you have a url that resolves to this machine (to be used for the ADFS 2.0 endpoints).
- In my environment, I added the following entry to your hosts file (C:\Windows\System32\drivers\etc): 127.0.0.1 adfs.testdomain.dev
- On a live environment, this should be done via DNS.
- Note that if you’re LAN connection uses a proxy server, you may need to bypass the proxy for the above address (i.e. IE –> Tools > Options >
As ADFS 2.0 endpoints should be run over https, ensure that you have a certificate setup (and it’s trusted). On a production environment this should be properly created and trusted, however on my dev environment I followed the instructions on http://www.robbagby.com/iis/self-signed-certificates-on-iis-7-the-easy-way-and-the-most-effective-way/ using SelfSSL.exe (note that selfssl is part of the IIS Resource Kit Toolkit).
- My specific command was “SelfSSL /N:CN=adfs.testdomain.dev /V:365 /S:1” (this create the certificate and configure the default website to use this certificate).

Server 1: Setting up ADFS 2.0

To set up ADFS 2.0, you first need to download it from http://www.microsoft.com/download/en/details.aspx?displaylang=en&id=10909. Once download, run through the following steps…

Run the setup wizard
On the first screen click next
Read and accept the license agreement
On the server role screen, select “Federation Server”
Click next to install.
Once it has run, you will need to restart the machine.

After it has installed, you now need to configure it

Launch AD FS 2.0 Management Tool from Start > Administrative Tools > AD FS 2.0 Management
Once loaded, select the AD FS 2.0 Federation Server Configuration Wizard (from the main panel)

Select create a new federation service and click next

Select “Stand-alone federation server” and click next

Accept the defaults for SSL certificates (this will be the certificate of you created earlier). Press next

Confirm the settings and press next.
Wait for the results.

Server 2: Setting up the website to host the WCF Service

On server 2, the WIF run time and SDK first needs to be installed.

WIF 4.0 Runtime - http://www.microsoft.com/download/en/details.aspx?displaylang=en&id=17331
WIF 4.0 SDK - http://www.microsoft.com/download/en/details.aspx?displaylang=en&id=4451

Next, if your not using a DNS server (i.e. you’re on a dev enviroment), edit C:\Windows\System32\drivers\etc\hosts and add the following entries:

adfs.testdomain.dev – 10.10.10.1 (set this to the IP address of the ADFS VM, aka Server 1)
services.testdomain.dev – 127.0.0.1

Next, create a website to host the WCF service:

In IIS Manager, right click on the site node in the connections pane and select “Add New Site”. Set the website settings as follows:
- SiteName: wcftestsite (lowercase)
- Physical Path: C:\inetpub\wwwroot\wcftestsite (you will need to create this folder)
- Binding (we will change this to https later)
  - Type: http
  - IP Address: All Unassigned
  - Port: 80
  - Host name: services.testdomain.dev (lowercase)
Once created, in IIS Manager, select the “Application Pool” node in the connections pane. Locate and select the newly created wcftestsite application pool.
- In the basic settings ensure that the application pool .NET framework version is set to v4.0
- In the “advanced settings” dialog. Under the process model section, change the identity to LocalSystem (this is so that it can access the ADFS certificate in the trusted root store). Note, on a production environment you would use a specific named account as opposed to this.
- In the “advanced settings” dialog. Under the process model section, set “Load User Profile” to true.

Next, we need to set up a trusted certificate for our new site. We will follow the steps as per server 1 to do this. My SelfSSL command is as follows: SelfSSL /N:CN=services.testdomain.dev /V:365 /S:2

Also, set IE to accept certificates as above to both services.testdomain.dev and adfs.testdomain.dev (as per server 1).

Server 2: Creating the WCF service

Open Visual Studio 2010
Select file > new > website
Select WCF Service from the project types
Set the web location to http://services.testdomain.dev/wcfservice and click ok (all lowercase)
Alter / add the following code to the GetData method on your WCF service. Browse to the service to ensure it no errors are thrown.

          1: public string GetData()

         2: {

         3:     var id = Thread.CurrentPrincipal.Identity as IClaimsIdentity;

         4:     return string.Format("Hello {0}", id.Claims[0].Value);

         5: }

In IIS Manager, select the newly created website and change the app pool to “wcftestsite”

Now the service has been created, it now needs to be altered to use ADFS 2.0.

In Visual Studio, right click on the project node for your WCF service website and select “Add STS Reference” (this is a new option provided by the WIF SDK)

Click next

Select the “Use an Existing STS” radio button
Set the url to https://adfs.testdomain.dev
Click test location (this should autocomplete the rest of the URL as per the screenshot below).
Click next

Set “Disable certificate chain validation” as we have created our own certificates.
Click next.

Select “Enable encryption” (wcf does not allow this to be disabled)
Select and existing certificate from store and select you “services.testdomain.dev” cert.
Click next
Leave offered claim as-is and click next
Verify the summary and click finish.
Once complete, a folder structure and a file named FederationMetadata.xml will be added to you website.

Server 1: Setup a trust relationship for our WCF service

Copy the newly created FederationMetadata.xml file create above to the ADFS VM
In ADFS2.0 Management console, from the tree right click on “Trust Relationships” and select “Add Relying Party Trust”.

Click start

Select “Import data about the relying party from a file” and select the FederationMetadata.xml file copied from the other server. Press next
Type the display name as “servicestestdomain” then click next
On the next screen, leave permit all users to access this relying party selected. Click next
Verify settings and click next
Leave the checkbox select for “Open the Edit Claim Rules dialog for this relying party trust when the wizard closes”. Click close.

Edit Claim Rules

On the edit claims rules dialog, verify that “Permit Access to All users” is set on the Issuance Authorization Rules tab
On the Issuance Tranform Rules tab, click add rule. Select “Pass Through or Filter an Incoming Claim”.
Set the claim rule name to “WCFServicePassthroughIncomingClaim”, on incoming claim type, select “Windows Account Name”.
Leave the radio button set as “pass through all claim values”
Click finish, then ok to save.

Server 2: Setup a test application to call the service

Run Visual Studio 2010 as an administrator (this should ensure that we have rights to see our newly created service). Open the solution that contains our WCF Service.
In our Visual Studio solution, create a console app named “WCFClient”. Set this as the startup project. Note that this can be totally separate to our solution, even on a different server – it just makes it easier to have them in the same place for this walkthrough.
Ensure that the target framework for this project is “.NET Framework 4.0” and not the “.NET Framework 4.0 Client Profile”.

Right click on this project and select “Add Service Reference”.
Set the address to https://services.testdomain.dev/wcfservice/service.svc and click discover. Note that we can still see the service as we are running as the user who created this account.

Set the namespace to MyTestService and click ok.
Add a couple of references:
- System.IdentityModel (.NET 4.0)
- Microsoft.IdentityModel
Add a new class called Token.cs to the project. This actively obtains the token from ADFS 2.0 (the secure token store). Add the following code to this class…

         1: using System;

         2: using System.Collections.Generic;

         3: using System.Linq;

         4: using System.Text;

         5: using System.IdentityModel.Tokens;

         6: using System.Net;

         7: using Microsoft.IdentityModel.Protocols.WSTrust;

         8: using System.ServiceModel;

         9: using System.ServiceModel.Security;

        10:  

        11: namespace WCFClient

        12: {

        13:     internal class Token

        14:     {

        15:         public static SecurityToken GetToken(string username, string password, string appliesTo, out RequestSecurityTokenResponse rsts)

        16:         {

        17:             WS2007HttpBinding binding = new WS2007HttpBinding();

        18:             binding.Security.Message.EstablishSecurityContext = false;

        19:             binding.Security.Transport.ClientCredentialType = HttpClientCredentialType.None;

        20:             binding.Security.Message.ClientCredentialType = MessageCredentialType.UserName;

        21:             binding.Security.Mode = SecurityMode.TransportWithMessageCredential; 

        22:  

        23:             WSTrustChannelFactory trustChannelFactory = new WSTrustChannelFactory(binding, new EndpointAddress("https://adfs.testdomain.dev/adfs/services/trust/13/usernamemixed"));

        24:             trustChannelFactory.TrustVersion = TrustVersion.WSTrust13;

        25:             trustChannelFactory.Credentials.UserName.UserName = username;

        26:             trustChannelFactory.Credentials.UserName.Password = password;

        27:             trustChannelFactory.ConfigureChannelFactory();

        28:  

        29:             // Create issuance issuance and get security token 

        30:             RequestSecurityToken requestToken = new RequestSecurityToken(WSTrust13Constants.RequestTypes.Issue);

        31:             requestToken.AppliesTo = new EndpointAddress(appliesTo);

        32:             WSTrustChannel tokenClient = (WSTrustChannel)trustChannelFactory.CreateChannel();

        33:             SecurityToken token = tokenClient.Issue(requestToken, out rsts);

        34:             return token;

        35:         }

        36:     }

        37: }

Add another class called FederatedWCFClient.cs to the project. This effectively wraps our service reference proxy class (autogenerated when we add our service reference), and allows it pass in a token to authenticate. Code for this is as follows:

         1: using System;

         2: using System.Collections.Generic;

         3: using System.Linq;

         4: using System.Text;

         5: using System.ServiceModel;

         6: using System.IdentityModel.Tokens;

         7: using Microsoft.IdentityModel.Protocols.WSTrust;

         8:  

         9: namespace WCFClient

        10: {

        11:     public class FederatedWCFClient<T>

        12:     {

        13:         private SecurityToken token;

        14:         private ChannelFactory<T> factory;

        15:         

        16:         public FederatedWCFClient(SecurityToken securityToken, string binding) 

        17:         {

        18:             this.token = securityToken;

        19:             this.factory = new ChannelFactory<T>(binding);

        20:             this.factory.ConfigureChannelFactory<T>();

        21:         }

        22:  

        23:         public void Close()

        24:         {

        25:             this.factory.Close();

        26:         }

        27:  

        28:         public T Client

        29:         {

        30:             get

        31:             {

        32:                 return this.factory.CreateChannelWithIssuedToken(token);

        33:             }

        34:         }

        35:  

        36:         public T ClientActAs

        37:         {

        38:             get

        39:             {

        40:                 return this.factory.CreateChannelActingAs(this.token);

        41:             }

        42:         }

        43:     }

        44: }

Finally, update our main method in Program.cs as follows (ensure that the binding name, endpoint and credentials are set up correctly).

         1: using System;

         2: using System.Collections.Generic;

         3: using System.Linq;

         4: using System.Text;

         5: using System.Net;

         6: using Microsoft.IdentityModel.Protocols.WSTrust;

         7:  

         8: namespace WCFClient

         9: {

        10:     class Program

        11:     {

        12:         static void Main(string[] args)

        13:         {

        14:             var requestTokenResponse = new RequestSecurityTokenResponse();

        15:             var token = Token.GetToken(@"mydomain\testuser", "p@ssw0rd", "http://services.testdomain.dev/wcfservice/Service.svc", out requestTokenResponse);

        16:             

        17:             var wcfClient = new FederatedWCFClient<MyTestService.IService>(token, "WS2007FederationHttpBinding_IService");   // This must match the app.config

        18:             var client = wcfClient.Client as MyTestService.IService;

        19:             var result = client.GetData();

        20:             Console.WriteLine(result);

        21:             wcfClient.Close();

        22:         }

        23:     }

        24: }

Now when you run the program, the client app will do the following

Obtain a token from ADFS
Call the WCF service method
Write the output of the WCF service method

Troubleshooting

The most common cause of error when creating this application was due to incorrect / inconsistent casing of the wcf service endpoint url. Verify case consistency of these.
ADFS 2.0 logs are visible in the Event Viewer (under “application and service logs”)
It can be useful turn on WCF logging WCF http://msdn.microsoft.com/en-us/library/aa702726.aspx. Logs can be opened in the Microsoft Service Trace Viewer (note that I normally use the message view within this – I locate errors, then view the exception tree for that error).
If when browsing to the WCF service endpoint, you receive “CryptographicException: Keyset does not exist” – this usually means the application pool does not trust the certificate. Check the WCF service is running under the correct application pool.

Thanks to Bradley Cotier for his debugging assistance

Written by Rob Nowik

Writing a gesture service with the Kinect for Windows SDK

MCS UK Solution Development — Mon, 08 Aug 2011 10:53:00 GMT

After further experimenting with the Kinect SDK, it became obvious what needed to come next. If you were to create an application using the Kinect SDK, you will want to be able to control the application using gestures (i.e. waving, swiping, motions to access menus, etc.). From this, we decided to write a gesture service in c# that would analyse the gestures. This blog post outlines how we did this and how you can implement the same functionality.

To be able to recognize gestures, it is first important to understand what makes a gesture.

We concluded that gestures were made up of parts. Each part of a gesture is a specific movement that, when combined with other gesture parts, makes up the whole gesture. For example the diagram below shows the two parts of a simple wave gesture and how they can be identified:

However this is not quite enough to be able to recognize multiple gestures with any degree of accuracy. The problem occurs when you think about multiple gestures being recognized at once.

It’s not as simple as looking for the next part of the gesture. . For example, consider the wave gesture shown above. If I was to drop my hand between the two parts, then it would still be recognized as a wave as both parts of the gesture were completed in the order they were defined; yet I clearly did not perform a wave. To solve this problem we came up with three results that a gesture part can return when it checks to see if it has been completed or not. The diagram below shows these three results and the impact of returning each of them:

A result of ‘Pausing’ allows the system to identify a movement that does not fulfil the gesture but could be a result of the user moving slowly. In short the three results mean the following:

Fail – The gesture failed. The user moved in a way that was inconsistent with the gesture and as such the gesture will start again at the beginning.
Pausing – The user did not fail the gesture but they did not perform the next part either. The system will check again for this part after a short pause. A result of pausing can only be returned a maximum of 100 times before the gesture will fail and recognition will start again at the beginning.
Succeed – the user performed this part of the gesture. After a short pause the system will start looking for the next part of the gesture.

The Solution

The overall gesture service is made up of three main parts each of which is detailed below:

The Gesture Controller:

The gesture controller is a way of controlling all of the gestures that a user can perform. The code for this can be seen below:

#region using...
using System;
using System.Collections.Generic;
using Microsoft.Research.Kinect.Nui;
#endregion
  
/// <summary>
/// The gesture controller
/// </summary>
public class GestureControler
{
  
    /// <summary>
    /// The list of all gestures we are currently looking for
    /// </summary>
    private List<Gesture> gestures = new List<Gesture>();
  
    /// <summary>
    /// Initializes a new instance of the <see cref="GestureControler"/> class.
    /// </summary>
    public GestureControler()
    {
    }
  
    /// <summary>
    /// Occurs when [gesture recognized].
    /// </summary>
    public event EventHandler<GestureEventArgs> GestureRecognised;
  
    /// <summary>
    /// Updates all gestures.
    /// </summary>
    /// <param name="data">The skeleton data.</param>
    public void UpdateAllGestures(SkeletonData data)
    {
        foreach (Gesture gesture in this.gestures)
        {
            gesture.UpdateGesture(data);
        }
    }
  
    /// <summary>
    /// Adds the gesture.
    /// </summary>
    /// <param name="type">The gesture type.</param>
    /// <param name="gestureDefinition">The gesture definition.</param>
    public void AddGesture(GestureType type, IRelativeGestureSegment[] gestureDefinition)
    {
        Gesture gesture = new Gesture(type, gestureDefinition);
        gesture.GestureRecognised += new EventHandler<GestureEventArgs>(this.Gesture_GestureRecognised);
        this.gestures.Add(gesture);
    }
  
    /// <summary>
    /// Handles the GestureRecognised event of the g control.
    /// </summary>
    /// <param name="sender">The source of the event.</param>
    /// <param name="e">The <see cref="KinectSkeltonTracker.GestureEventArgs"/> instance containing the event data.</param>
    private void Gesture_GestureRecognised(object sender, GestureEventArgs e)
    {
        if (this.GestureRecognised != null)
        {
        this.GestureRecognised(this, e);
        }
  
        foreach (Gesture g in this.gestures)
        {
            g.Reset();
        }
    }
}

A Gesture:

This controls all of the parts of a gesture and which one is currently being checked. It contains an array of IRelativeGestureSegment which are individual implementations of the IRelativeGestureSegment interface (which I will mention later). When a skeleton frame is created it is passed through to each Gesture which then passes it through to the current gesture segment. When the final segment returns a result of ‘Succeed’ it raises a gesture recognized event which is caught by the gesture controller. The code for the Gesture class can be seen below:

#region using...
using System;
using Microsoft.Research.Kinect.Nui;
#endregion
  
<summary>
/// A single gesture
/// </summary>
public class Gesture
{
    /// <summary>
    /// The parts that make up this gesture
    /// </summary>
    private IRelativeGestureSegment[] gestureParts;
  
    /// <summary>
    /// The current gesture part that we are matching against
    /// </summary>
    private int currentGesturePart = 0;
  
    /// <summary>
    /// the number of frames to pause for when a pause is initiated
    /// </summary>
    private int pausedFrameCount = 10;
    
    /// <summary>
    /// The current frame that we are on
    /// </summary>
    private int frameCount = 0;
  
    /// <summary>
    /// Are we paused?
    /// </summary>
    private bool paused = false;
  
    /// <summary>
    /// The type of gesture that this is
    /// </summary>
    private GestureType type;
  
    /// <summary>
    /// Initializes a new instance of the <see cref="Gesture"/> class.
    /// </summary>
    /// <param name="type">The type of gesture.</param>
    /// <param name="gestureParts">The gesture parts.</param>
    public Gesture(GestureType type, IRelativeGestureSegment[] gestureParts)
    {
        this.gestureParts = gestureParts;
        this.type = type;
    }
  
    /// <summary>
    /// Occurs when [gesture recognised].
    /// </summary>
    public event EventHandler<GestureEventArgs> GestureRecognised;
  
    /// <summary>
    /// Updates the gesture.
    /// </summary>
    /// <param name="data">The skeleton data.</param>
    public void UpdateGesture(SkeletonData data)
    {
        if (this.paused)
        {
            if (this.frameCount == this.pausedFrameCount)
            {
                this.paused = false;
            }
  
            this.frameCount++;
        }
  
        GesturePartResult result = this.gestureParts[this.currentGesturePart].CheckGesture(data);
        if (result == GesturePartResult.Suceed)
        {
            if (this.currentGesturePart + 1 < this.gestureParts.Length)
            {
                this.currentGesturePart++;
                this.frameCount = 0;
                this.pausedFrameCount = 10;
                this.paused = true;
            }
            else
            {
                if (this.GestureRecognised != null)
                {
                    this.GestureRecognised(this, new GestureEventArgs(this.type, data.TrackingID, data.UserIndex));
                    this.Reset();
                }
            }
        }
        else if (result == GesturePartResult.Fail || this.frameCount == 50)
        {
            this.currentGesturePart = 0;
            this.frameCount = 0;
            this.pausedFrameCount = 5;
            this.paused = true;
        }
        else
        {
            this.frameCount++;
            this.pausedFrameCount = 5;
            this.paused = true;
        }
    }
  
    /// <summary>
    /// Resets this instance.
    /// </summary>
    public void Reset()
    {
        this.currentGesturePart = 0;
        this.frameCount = 0;
        this.pausedFrameCount = 5;
        this.paused = true;
    }
}

The IRelativeGestureSegment:

This is the final part of a gesture. It is essentially the individual segments that make up a gesture. Below is the IRelativeGestureSegment class and the implementations of this class for a wave gesture

#region using...
using Microsoft.Research.Kinect.Nui;
#endregion
  
/// <summary>
/// Defines a single gesture segment which uses relative positioning 
/// of body parts to detect a gesture
/// </summary>
public interface IRelativeGestureSegment
{
    /// <summary>
    /// Checks the gesture.
    /// </summary>
    /// <param name="skeleton">The skeleton.</param>
    /// <returns>GesturePartResult based on if the gesture part has been completed</returns>
    GesturePartResult CheckGesture(SkeletonData skeleton);
}

Wave gesture

#region using...
using Microsoft.Research.Kinect.Nui;
#endregion
  
/// <summary>
/// the first part of the wave left gesture
/// </summary>
public class WaveLeftSegment1 : IRelativeGestureSegment
{
    /// <summary>
    /// Checks the gesture.
    /// </summary>
    /// <param name="skeleton">The skeleton.</param>
    /// <returns>GesturePartResult based on if the gesture part has been completed</returns>
    public GesturePartResult CheckGesture(SkeletonData skeleton)
    {
        // hand above elbow
        if (skeleton.Joints[JointID.HandLeft].Position.Y > skeleton.Joints[JointID.ElbowLeft].Position.Y)
        {
            // hand right of elbow
            if (skeleton.Joints[JointID.HandLeft].Position.X > skeleton.Joints[JointID.ElbowLeft].Position.X)
            {
                return GesturePartResult.Suceed;
            }
            // hand has not dropped but is not quite where we expect it to be, pausing till next frame
            return GesturePartResult.Pausing;
        }
    
        // hand dropped - no gesture fails
        return GesturePartResult.Fail;
        }
    }
  
    /// <summary>
    /// The second part of the wave left gesture
    /// </summary>
    public class WaveLeftSegment2 : IRelativeGestureSegment
    {
        /// <summary>
        /// Checks the gesture.
        /// </summary>
        /// <param name="skeleton">The skeleton.</param>
        /// <returns>GesturePartResult based on if the gesture part has been completed</returns>
        public GesturePartResult CheckGesture(SkeletonData skeleton)
        {
            // hand above elbow
            if (skeleton.Joints[JointID.HandLeft].Position.Y > skeleton.Joints[JointID.ElbowLeft].Position.Y)
            {
                // hand right of elbow
                if (skeleton.Joints[JointID.HandLeft].Position.X < skeleton.Joints[JointID.ElbowLeft].Position.X)
                {
                    return GesturePartResult.Suceed;
                }
                // hand has not dropped but is not quite where we expect it to be, pausing till next frame
                return GesturePartResult.Pausing;
            }
            // hand dropped - no gesture fails
            return GesturePartResult.Fail;
        }
    }
}

NOTE: a wave gesture is made up of two parts that are repeated three times. For example the code to create a new Wave gesture would look like this (gestures is the gesture controller):

IRelativeGestureSegment[] waveLeftSegments = new IRelativeGestureSegment[6];
WaveLeftSegment1 waveLeftSegment1 = new WaveLeftSegment1();
WaveLeftSegment2 waveLeftSegment2 = new WaveLeftSegment2();
waveLeftSegments[0] = waveLeftSegment1;
waveLeftSegments[1] = waveLeftSegment2;
waveLeftSegments[2] = waveLeftSegment1;
waveLeftSegments[3] = waveLeftSegment2;
waveLeftSegments[4] = waveLeftSegment1;
waveLeftSegments[5] = waveLeftSegment2;
this.gestures.AddGesture(GestureType.WaveLeft, waveLeftSegments);

The full source code for this example (and for skeleton tracking) can be downloaded here. It contains a wave gestures with both hands as well as swipe left, swipe right and a menu gesture. When writing your own gestures it is important to consider the amount of checking that is required and optimize this for each of the parts. Generally smaller segments work better as there is less checking to be done which improves performance.

Written by Michael Tsikkos and James Glading

First Look at the Kinect for Windows SDK

MCS UK Solution Development — Mon, 08 Aug 2011 10:39:13 GMT

The Kinect for Windows SDK has finally arrived, and we couldn’t wait to get our hands on it!

After downloading, installing, and actually getting a Kinect with a USB connector for our PC’s, we were good to go…

Step 1 – What does the SDK give us?

When we started, we didn’t really know what to expect from the SDK.

What you get are a set of 20 points, each corresponding to a specific joint in the body. These points come in standard ‘x, y, z’ coordinates, with an extra coordinate ‘w’, which we haven’t got round to figuring out what it is there for just yet.

Here comes problem one. The points that are given are within the range -1 and 1, going from left to right for x, and bottom to top with y. Therefore, you have to first convert these to work with WPF applications so that if fits within the coordinates of a standard page.

This part was just a formality and had a very simple fix with a couple of converters.

One for X:

          1: /// <summary>

         2: /// Joint converter

         3: /// </summary>

         4: public class KinectValueToScreenCoOrdinatesConverterX : IValueConverter

         5: {

         6:     /// <summary>

         7:     /// Converts a value.

         8:     /// </summary>

         9:     /// <param name="value">The value produced by the binding source.</param>

        10:     /// <param name="targetType">The type of the binding target property.</param>

        11:     /// <param name="parameter">The converter parameter to use.</param>

        12:     /// <param name="culture">The culture to use in the converter.</param>

        13:     /// <returns>

        14:     /// A converted value. If the method returns null, the valid null value is used.

        15:     /// </returns>

        16:     public object Convert(object value, Type targetType, object parameter, System.Globalization.CultureInfo culture)

        17:     {

        18:         double multiplier = Application.Current.MainWindow.Width;

        19:  

        20:         double translation = 0;

        21:         try

        22:         {

        23:             if (parameter != null)

        24:             {

        25:                 translation = double.Parse(parameter as string);

        26:             }

        27:  

        28:             float fl = (float)value;

        29:             return translation + (multiplier + (fl * multiplier));

        30:        }

        31:        catch (InvalidCastException)

        32:        {

        33:             return 0;

        34:         }

        35:     }

        36:  

        37:     /// <summary>

        38:     /// Converts a value.

        39:     /// </summary>

        40:     /// <param name="value">The value that is produced by the binding target.</param>

        41:     /// <param name="targetType">The type to convert to.</param>

        42:     /// <param name="parameter">The converter parameter to use.</param>

        43:     /// <param name="culture">The culture to use in the converter.</param>

        44:     /// <returns>

        45:     /// A converted value. If the method returns null, the valid null value is used.

        46:     /// </returns>

        47:     public object ConvertBack(object value, Type targetType, object parameter, System.Globalization.CultureInfo culture)

        48:     {

        49:         throw new NotImplementedException();

        50:     }

        51: }

And one for Y:

         1: /// <summary>

         2: /// Joint converter Y

         3: /// </summary>

         4: public class KinectValueToScreenCoOrdinatesConverterY : IValueConverter

         5: {

         6:     /// <summary>

         7:     /// Converts a value.

         8:     /// </summary>

         9:     /// <param name="value">The value produced by the binding source.</param>

        10:     /// <param name="targetType">The type of the binding target property.</param>

        11:     /// <param name="parameter">The converter parameter to use.</param>

        12:     /// <param name="culture">The culture to use in the converter.</param>

        13:     /// <returns>

        14:     /// A converted value. If the method returns null, the valid null value is used.

        15:     /// </returns>

        16:     public object Convert(object value, Type targetType, object parameter, System.Globalization.CultureInfo culture)

        17:     {

        18:         double multiplier = Application.Current.MainWindow.Height;

        19:  

        20:             double translation = 0;

        21:             try

        22:             {

        23:                 if (parameter != null)

        24:                 {

        25:                     translation = double.Parse(parameter as string);

        26:                 }

        27:  

        28:                 float fl = (float)value;

        29:                 return translation + (multiplier + (-fl * multiplier));

        30:             }

        31:             catch (InvalidCastException)

        32:             {

        33:                 return 0;

        34:             }

        35:         }

        36:  

        37:         /// <summary>

        38:         /// Converts a value.

        39:         /// </summary>

        40:         /// <param name="value">The value that is produced by the binding target.</param>

        41:         /// <param name="targetType">The type to convert to.</param>

        42:         /// <param name="parameter">The converter parameter to use.</param>

        43:         /// <param name="culture">The culture to use in the converter.</param>

        44:         /// <returns>

        45:         /// A converted value. If the method returns null, the valid null value is used.

        46:         /// </returns>

        47:         public object ConvertBack(object value, Type targetType, object parameter, System.Globalization.CultureInfo culture)

        48:         {

        49:             throw new NotImplementedException();

        50:         }

        51:     }

A converter is assigned to every part of the skeleton that is displayed on the UI. Since Kinect returns a series of skeletons (each one representing a person) with the points already mapped to each joint, we decided to drive the MVVM pattern directly off of the SkeletonData that Kinect returns.

Step 2 – Putting everything into the MVVM pattern

To simplify things and make it easier to create multiple skeletons on screen at any one time, we decided to use the MVVM pattern. There are four main parts to this:

1. The Kinect connection

2. The Main View Model

3. The Skeleton View Model

4. The UI

Kinect Connection

The Kinect connection is an intermediate class between our view models and the Kinect sensor. It fires Skeleton ready events for each skeleton that is returned and skeleton frame complete events when an entire frame has been processed. As it processes the skeletons in a frame, it keeps track of their ID value so we can track all the skeletons that were in a frame during the Skeleton Frame Complete event.

One of the main parts of this class is the InitializeNui() method:

         1: /// <summary>

         2:         /// Initializes the Kinect sensor.

         3:         /// </summary>

         4:         /// <returns>bool value true if the sensor initialised correctly</returns>

         5:         private bool InitializeNui()

         6:         {

         7:             if (this.kinectRunTime == null)

         8:             {

         9:                 return false;

        10:             }

        11:  

        12:             try

        13:             {

        14:                 this.kinectRunTime.Initialize(RuntimeOptions.UseDepth | RuntimeOptions.UseSkeletalTracking | RuntimeOptions.UseColor);

        15:             }

        16:             catch (Exception exception)

        17:             {

        18:                 Console.WriteLine(exception.ToString());

        19:                 return false;

        20:             }

        21:  

        22:             this.kinectRunTime.VideoStream.Open(ImageStreamType.Video, 2, ImageResolution.Resolution640x480, ImageType.Color);

        23:  

        24:             this.kinectRunTime.SkeletonEngine.TransformSmooth = true;

        25:  

        26:             var parameters = new TransformSmoothParameters

        27:             {

        28:                 Smoothing = 0.75f,

        29:                 Correction = 0.0f,

        30:                 Prediction = 0.0f,

        31:                 JitterRadius = 0.05f,

        32:                 MaxDeviationRadius = 0.04f

        33:             };

        34:  

        35:             this.kinectRunTime.SkeletonEngine.SmoothParameters = parameters;

        36:  

        37:             return true;

        38:         }

        39:     }

This method initialises the Kinect sensor with a set of pre define arguments. The actual initialization of the sensor happens in the line :

         1: this.kinectRunTime.Initialize(RuntimeOptions.UseDepth | RuntimeOptions.UseSkeletalTracking | RuntimeOptions.UseColor);
    

The runtime options specify what you want the Kinect sensor to look for. In our example we were looking for skeleton frames and images from the camera.

Main View Model

The main view model controls the allocation of skeletons to the skeleton view models. It contains a list of Skeleton View models and a dictionary of skeleton view models with the skeleton id as the key. Each time a skeleton ready event is fired in the Kinect connection it looks up the skeleton ID value in the dictionary and assigns the new skeleton data to the skeleton in that view model.

Originally the Main view model was also going to sort out the removal of skeletons that are no longer active. However the Kinect sensor does not always return all of the active skeletons in each of the Skelton frame ready events that the Kinect SDK produce. This means that a simple removal of any skeletons not in the current frame will fail and the view model associated with a skeleton will be continuously deleted and re-created. To get around this we placed a timer in each of the skeleton view models. If the view model is not updated within a set period of time the view model raises an event and the main view model then catches it and deleted the skeleton.

The Skeleton View Model

The skeleton view model is essentially a way of exposing the properties of a SkeletonData object so that the UI can bind to them. When the skeleton property is changed a NotifyPropertyChanged event is raised on all of the exposed properties which updates all the bindings on the UI.

The skeleton view model also controls the Gesture Service which we will discuss in our next blog post. This is again updated when the skeleton property changes.

The full code for the skeleton view model can be found in the source code below:

         1: /// <summary>

         2:     /// The main view model

         3:     /// </summary>

         4:     public class SkeletonViewModel : INotifyPropertyChanged

         5:     {

         6:         /// <summary>

         7:         /// The gesture controler for this skeleton

         8:         /// </summary>

         9:         private GestureControler gestures = new GestureControler();

        10:  

        11:         #region fields

        12:  

        13:         /// <summary>

        14:         /// backing field for the gesture text

        15:         /// </summary>

        16:         private string gestureText;

        17:  

        18:         /// <summary>

        19:         /// backing field for the gesture property

        20:         /// </summary>

        21:         private bool gestureDetected;

        22:  

        23:         /// <summary>

        24:         /// The skeleton data

        25:         /// This is used as the backing field for all properties

        26:         /// </summary>

        27:         private SkeletonData skeleton;

        28:  

        29:         /// <summary>

        30:         /// backing field for the joint color

        31:         /// </summary>

        32:         private Color jointColor;

        33:  

        34:         /// <summary>

        35:         /// Occurs when [delete skeleton].

        36:         /// </summary>

        37:         public event EventHandler DeleteSkeleton;

        38:         

        39:         /// <summary>

        40:         /// the timer for the on screen gesture text

        41:         /// </summary>

        42:         private DispatcherTimer textTimer = new DispatcherTimer() { Interval = TimeSpan.FromSeconds(3) };

        43:  

        44:         #endregion

        45:  

        46:         #region constructors

        47:  

        48:         /// <summary>

        49:         /// Initializes a new instance of the <see cref="SkeletonViewModel"/> class.

        50:         /// </summary>

        51:         public SkeletonViewModel()

        52:         {

        53:             this.ChangeTimer = new DispatcherTimer();

        54:             this.ChangeTimer.Interval = TimeSpan.FromSeconds(0.5);

        55:             this.ChangeTimer.Tick += new EventHandler(this.ChangeTimer_Tick);

        56:             this.textTimer.Tick += new EventHandler(this.TextTimer_Tick);

        57:             ChangeTimer.Start();

        58:             DefineGestures();

        59:             this.gestures.GestureRecognised += new EventHandler<GestureEventArgs>(this.Gestures_GestureRecognised);

        60:         }

        61:  

        62:         #endregion

        63:  

        64:         #region events

        65:  

        66:         /// <summary>

        67:         /// Occurs when a property value changes.

        68:         /// </summary>

        69:         public event PropertyChangedEventHandler PropertyChanged;

        70:  

        71:         #endregion

        72:  

        73:         #region properties

        74:  

        75:         /// <summary>

        76:         /// Gets or sets the change timer.

        77:         /// </summary>

        78:         /// <value>

        79:         /// The change timer.

        80:         /// </value>

        81:         public DispatcherTimer ChangeTimer

        82:         {

        83:             get;

        84:             set;

        85:         }

        86:  

        87:         /// <summary>

        88:         /// Gets or sets the gesture text.

        89:         /// </summary>

        90:         /// <value>

        91:         /// The gesture text.

        92:         /// </value>

        93:         public string GestureText

        94:         {

        95:             get

        96:             {

        97:                 return this.gestureText;

        98:             }

        99:  

       100:             set

       101:             {

       102:                 if (this.gestureText != value)

       103:                 {

       104:                     this.gestureText = value;

       105:                     this.NotifyPropertyChanged("GestureText");

       106:                 }

       107:             }

       108:         }

       109:  

       110:         /// <summary>

       111:         /// Gets or sets a value indicating whether this <see cref="SkeletonViewModel"/> is waved.

       112:         /// </summary>

       113:         /// <value>

       114:         ///   <c>true</c> if waved; otherwise, <c>false</c>.

       115:         /// </value>

       116:         public bool GestureDetected

       117:         {

       118:             get

       119:             {

       120:                 return this.gestureDetected;

       121:             }

       122:  

       123:             set

       124:             {

       125:                 if (this.gestureDetected != value)

       126:                 {

       127:                     this.gestureDetected = value;

       128:                     this.NotifyPropertyChanged("GestureDetected");

       129:                 }

       130:             }

       131:         }

       132:  

       133:         /// <summary>

       134:         /// Gets or sets the color of the joint.

       135:         /// </summary>

       136:         /// <value>

       137:         /// The color of the joint.

       138:         /// </value>

       139:         public Color JointColor

       140:         {

       141:             get

       142:             {

       143:                 return this.jointColor;

       144:             }

       145:  

       146:             set

       147:             {

       148:                 if (this.jointColor != value)

       149:                 {

       150:                     this.jointColor = value;

       151:                     this.NotifyPropertyChanged("JointColor");

       152:                 }

       153:             }

       154:         }

       155:  

       156:         /// <summary>

       157:         /// Gets or sets the skeleton.

       158:         /// </summary>

       159:         /// <value>

       160:         /// The skeleton.

       161:         /// </value>

       162:         public SkeletonData Skeleton

       163:         {

       164:             get

       165:             {

       166:                 return this.skeleton;

       167:             }

       168:  

       169:             set

       170:             {

       171:                 if (this.skeleton != value)

       172:                 {

       173:                     this.skeleton = value;

       174:                     this.NotifyPropertyChanged("Skeleton");

       175:                     this.NotifyAllChange();

       176:                     ResetTimer();

       177:                     this.gestures.UpdateAllGestures(this.skeleton);

       178:                 }

       179:             }

       180:         }

       181:  

       182:         #region KinectBodyParts

       183:  

       184:         /// <summary>

       185:         /// Gets the head.

       186:         /// </summary>

       187:         public Joint Head

       188:         {

       189:             get

       190:             {

       191:                 if (this.Skeleton != null)

       192:                 {

       193:                     return this.Skeleton.Joints[JointID.Head];

       194:                 }

       195:                 else

       196:                 {

       197:                     return new Joint();

       198:                 }

       199:             }

       200:         }

       201:  

       202:         /// <summary>

       203:         /// Gets the left hand.

       204:         /// </summary>

       205:         public Joint LeftHand

       206:         {

       207:             get

       208:             {

       209:                 if (this.Skeleton != null)

       210:                 {

       211:                     return this.Skeleton.Joints[JointID.HandLeft];

       212:                 }

       213:                 else

       214:                 {

       215:                     return new Joint();

       216:                 }

       217:             }

       218:         }

       219:  

       220:         /// <summary>

       221:         /// Gets the right hand.

       222:         /// </summary>

       223:         public Joint RightHand

       224:         {

       225:             get

       226:             {

       227:                 if (this.Skeleton != null)

       228:                 {

       229:                     return this.Skeleton.Joints[JointID.HandRight];

       230:                 }

       231:                 else

       232:                 {

       233:                     return new Joint();

       234:                 }

       235:             }

       236:         }

       237:  

       238:         /// <summary>

       239:         /// Gets the left wrist.

       240:         /// </summary>

       241:         public Joint LeftWrist

       242:         {

       243:             get

       244:             {

       245:                 if (this.Skeleton != null)

       246:                 {

       247:                     return this.Skeleton.Joints[JointID.WristLeft];

       248:                 }

       249:                 else

       250:                 {

       251:                     return new Joint();

       252:                 }

       253:             }

       254:         }

       255:  

       256:         /// <summary>

       257:         /// Gets the right wrist.

       258:         /// </summary>

       259:         public Joint RightWrist

       260:         {

       261:             get

       262:             {

       263:                 if (this.Skeleton != null)

       264:                 {

       265:                     return this.Skeleton.Joints[JointID.WristRight];

       266:                 }

       267:                 else

       268:                 {

       269:                     return new Joint();

       270:                 }

       271:             }

       272:         }

       273:  

       274:         /// <summary>

       275:         /// Gets the left elbow.

       276:         /// </summary>

       277:         public Joint LeftElbow

       278:         {

       279:             get

       280:             {

       281:                 if (this.Skeleton != null)

       282:                 {

       283:                     return this.Skeleton.Joints[JointID.ElbowLeft];

       284:                 }

       285:                 else

       286:                 {

       287:                     return new Joint();

       288:                 }

       289:             }

       290:         }

       291:  

       292:         /// <summary>

       293:         /// Gets the right elbow.

       294:         /// </summary>

       295:         public Joint RightElbow

       296:         {

       297:             get

       298:             {

       299:                 if (this.Skeleton != null)

       300:                 {

       301:                     return this.Skeleton.Joints[JointID.ElbowRight];

       302:                 }

       303:                 else

       304:                 {

       305:                     return new Joint();

       306:                 }

       307:             }

       308:         }

       309:  

       310:         /// <summary>

       311:         /// Gets the shoulder center.

       312:         /// </summary>

       313:         public Joint ShoulderCenter

       314:         {

       315:             get

       316:             {

       317:                 if (this.Skeleton != null)

       318:                 {

       319:                     return this.Skeleton.Joints[JointID.ShoulderCenter];

       320:                 }

       321:                 else

       322:                 {

       323:                     return new Joint();

       324:                 }

       325:             }

       326:         }

       327:  

       328:         /// <summary>

       329:         /// Gets the spine.

       330:         /// </summary>

       331:         public Joint Spine

       332:         {

       333:             get

       334:             {

       335:                 if (this.Skeleton != null)

       336:                 {

       337:                     return this.Skeleton.Joints[JointID.Spine];

       338:                 }

       339:                 else

       340:                 {

       341:                     return new Joint();

       342:                 }

       343:             }

       344:         }

       345:  

       346:         /// <summary>

       347:         /// Gets the hip center.

       348:         /// </summary>

       349:         public Joint HipCenter

       350:         {

       351:             get

       352:             {

       353:                 if (this.Skeleton != null)

       354:                 {

       355:                     return this.Skeleton.Joints[JointID.HipCenter];

       356:                 }

       357:                 else

       358:                 {

       359:                     return new Joint();

       360:                 }

       361:             }

       362:         }

       363:  

       364:         /// <summary>

       365:         /// Gets the left knee.

       366:         /// </summary>

       367:         public Joint LeftKnee

       368:         {

       369:             get

       370:             {

       371:                 if (this.Skeleton != null)

       372:                 {

       373:                     return this.Skeleton.Joints[JointID.KneeLeft];

       374:                 }

       375:                 else

       376:                 {

       377:                     return new Joint();

       378:                 }

       379:             }

       380:         }

       381:  

       382:         /// <summary>

       383:         /// Gets the right knee.

       384:         /// </summary>

       385:         public Joint RightKnee

       386:         {

       387:             get

       388:             {

       389:                 if (this.Skeleton != null)

       390:                 {

       391:                     return this.Skeleton.Joints[JointID.KneeRight];

       392:                 }

       393:                 else

       394:                 {

       395:                     return new Joint();

       396:                 }

       397:             }

       398:         }

       399:  

       400:         /// <summary>

       401:         /// Gets the left ankle.

       402:         /// </summary>

       403:         public Joint LeftAnkle

       404:         {

       405:             get

       406:             {

       407:                 if (this.Skeleton != null)

       408:                 {

       409:                     return this.Skeleton.Joints[JointID.AnkleLeft];

       410:                 }

       411:                 else

       412:                 {

       413:                     return new Joint();

       414:                 }

       415:             }

       416:         }

       417:  

       418:         /// <summary>

       419:         /// Gets the right ankle.

       420:         /// </summary>

       421:         public Joint RightAnkle

       422:         {

       423:             get

       424:             {

       425:                 if (this.Skeleton != null)

       426:                 {

       427:                     return this.Skeleton.Joints[JointID.AnkleRight];

       428:                 }

       429:                 else

       430:                 {

       431:                     return new Joint();

       432:                 }

       433:             }

       434:         }

       435:  

       436:         /// <summary>

       437:         /// Gets the left foot.

       438:         /// </summary>

       439:         public Joint LeftFoot

       440:         {

       441:             get

       442:             {

       443:                 if (this.Skeleton != null)

       444:                 {

       445:                     return this.Skeleton.Joints[JointID.FootLeft];

       446:                 }

       447:                 else

       448:                 {

       449:                     return new Joint();

       450:                 }

       451:             }

       452:         }

       453:  

       454:         /// <summary>

       455:         /// Gets the right foot.

       456:         /// </summary>

       457:         public Joint RightFoot

       458:         {

       459:             get

       460:             {

       461:                 if (this.Skeleton != null)

       462:                 {

       463:                     return this.Skeleton.Joints[JointID.FootRight];

       464:                 }

       465:                 else

       466:                 {

       467:                     return new Joint();

       468:                 }

       469:             }

       470:         }

       471:  

       472:         /// <summary>

       473:         /// Gets the Left Hip

       474:         /// </summary>

       475:         public Joint RightHip

       476:         {

       477:             get

       478:             {

       479:                 if (this.Skeleton != null)

       480:                 {

       481:                     return this.Skeleton.Joints[JointID.HipRight];

       482:                 }

       483:                 else

       484:                 {

       485:                     return new Joint();

       486:                 }

       487:             }

       488:         }

       489:  

       490:         /// <summary>

       491:         /// Gets the Left Hip

       492:         /// </summary>

       493:         public Joint LeftHip

       494:         {

       495:             get

       496:             {

       497:                 if (this.Skeleton != null)

       498:                 {

       499:                     return this.Skeleton.Joints[JointID.HipLeft];

       500:                 }

       501:                 else

       502:                 {

       503:                     return new Joint();

       504:                 }

       505:             }

       506:         }

       507:  

       508:         /// <summary>

       509:         /// Gets the Right Shoulder

       510:         /// </summary>

       511:         public Joint RightShoulder

       512:         {

       513:             get

       514:             {

       515:                 if (this.Skeleton != null)

       516:                 {

       517:                     return this.Skeleton.Joints[JointID.ShoulderRight];

       518:                 }

       519:                 else

       520:                 {

       521:                     return new Joint();

       522:                 }

       523:             }

       524:         }

       525:  

       526:         /// <summary>

       527:         /// Gets the Left Shoulder

       528:         /// </summary>

       529:         public Joint LeftShoulder

       530:         {

       531:             get

       532:             {

       533:                 if (this.Skeleton != null)

       534:                 {

       535:                     return this.Skeleton.Joints[JointID.ShoulderLeft];

       536:                 }

       537:                 else

       538:                 {

       539:                     return new Joint();

       540:                 }

       541:             }

       542:         }

       543:         #endregion

       544:  

       545:         #endregion

       546:  

       547:         #region methods

       548:  

       549:         /// <summary>

       550:         /// Handles the Tick event of the textTimer control.

       551:         /// </summary>

       552:         /// <param name="sender">The source of the event.</param>

       553:         /// <param name="e">The <see cref="System.EventArgs"/> instance containing the event data.</param>

       554:         private void TextTimer_Tick(object sender, EventArgs e)

       555:         {

       556:             this.GestureDetected = false;

       557:             this.textTimer.Stop();

       558:         }

       559:  

       560:         /// <summary>

       561:         /// Handles the GestureRecognised event of the Gestures control.

       562:         /// </summary>

       563:         /// <param name="sender">The source of the event.</param>

       564:         /// <param name="e">The <see cref="KinectSkeltonTracker.GestureEventArgs"/> instance containing the event data.</param>

       565:         private void Gestures_GestureRecognised(object sender, GestureEventArgs e)

       566:         {

       567:             if (e.GestureType == GestureType.WaveRight)

       568:             {

       569:                 this.GestureDetected = true;

       570:                 this.GestureText = "Waved with right hand";

       571:                 this.textTimer.Start();

       572:             }

       573:             else if (e.GestureType == GestureType.WaveLeft)

       574:             {

       575:                 this.GestureDetected = true;

       576:                 this.GestureText = "Waved with left hand";

       577:                 this.textTimer.Start();

       578:             }

       579:             else if (e.GestureType == GestureType.LeftSwipe)

       580:             {

       581:                 this.GestureDetected = true;

       582:                 this.GestureText = "Swiped left";

       583:                 this.textTimer.Start();

       584:             }

       585:             else if (e.GestureType == GestureType.RightSwipe)

       586:             {

       587:                 this.GestureDetected = true;

       588:                 this.GestureText = "Swiped right";

       589:                 this.textTimer.Start();

       590:             }

       591:             else if (e.GestureType == GestureType.Menu)

       592:             {

       593:                 this.GestureDetected = true;

       594:                 this.GestureText = "Menu";

       595:                 this.textTimer.Start();

       596:             }

       597:         }

       598:  

       599:         /// <summary>

       600:         /// Defines the gestures.

       601:         /// </summary>

       602:         private void DefineGestures()

       603:         {

       604:             IRelativeGestureSegment[] waveRightSegments = new IRelativeGestureSegment[6];

       605:             WaveRightSegment1 waveRightSegment1 = new WaveRightSegment1();

       606:             WaveRightSegment2 waveRightSegment2 = new WaveRightSegment2();

       607:             waveRightSegments[0] = waveRightSegment1;

       608:             waveRightSegments[1] = waveRightSegment2;

       609:             waveRightSegments[2] = waveRightSegment1;

       610:             waveRightSegments[3] = waveRightSegment2;

       611:             waveRightSegments[4] = waveRightSegment1;

       612:             waveRightSegments[5] = waveRightSegment2;

       613:             this.gestures.AddGesture(GestureType.WaveRight, waveRightSegments);

       614:  

       615:             IRelativeGestureSegment[] waveLeftSegments = new IRelativeGestureSegment[6];

       616:             WaveLeftSegment1 waveLeftSegment1 = new WaveLeftSegment1();

       617:             WaveLeftSegment2 waveLeftSegment2 = new WaveLeftSegment2();

       618:             waveLeftSegments[0] = waveLeftSegment1;

       619:             waveLeftSegments[1] = waveLeftSegment2;

       620:             waveLeftSegments[2] = waveLeftSegment1;

       621:             waveLeftSegments[3] = waveLeftSegment2;

       622:             waveLeftSegments[4] = waveLeftSegment1;

       623:             waveLeftSegments[5] = waveLeftSegment2;

       624:             this.gestures.AddGesture(GestureType.WaveLeft, waveLeftSegments);

       625:  

       626:             IRelativeGestureSegment[] swipeleftSegments = new IRelativeGestureSegment[3];

       627:             swipeleftSegments[0] = new SwipeLeftSegment1();

       628:             swipeleftSegments[1] = new SwipeLeftSegment2();

       629:             swipeleftSegments[2] = new SwipeLeftSegment3();

       630:             this.gestures.AddGesture(GestureType.LeftSwipe, swipeleftSegments);

       631:  

       632:             IRelativeGestureSegment[] swiperightSegments = new IRelativeGestureSegment[3];

       633:             swiperightSegments[0] = new SwipeRightSegment1();

       634:             swiperightSegments[1] = new SwipeRightSegment2();

       635:             swiperightSegments[2] = new SwipeRightSegment3();

       636:             this.gestures.AddGesture(GestureType.RightSwipe, swiperightSegments);

       637:  

       638:             IRelativeGestureSegment[] menuSegments = new IRelativeGestureSegment[20];

       639:             MenuSegments1 menuSegment = new MenuSegments1();

       640:             for (int i = 0; i < 20; i++)

       641:             {

       642:                 //gesture consists of the same thing 20 times 

       643:                 menuSegments[i] = menuSegment;

       644:             }

       645:  

       646:             this.gestures.AddGesture(GestureType.Menu, menuSegments);

       647:         }

       648:  

       649:         /// <summary>

       650:         /// Handles the Tick event of the ChangeTimer control.

       651:         /// </summary>

       652:         /// <param name="sender">The source of the event.</param>

       653:         /// <param name="e">The <see cref="System.EventArgs"/> instance containing the event data.</param>

       654:         void ChangeTimer_Tick(object sender, EventArgs e)

       655:         {

       656:             if (this.DeleteSkeleton != null)

       657:             {

       658:                 this.DeleteSkeleton(this, null);

       659:             }

       660:         }

       661:  

       662:         /// <summary>

       663:         /// Resets the timer.

       664:         /// </summary>

       665:         public void ResetTimer()

       666:         {

       667:             this.ChangeTimer.Stop();

       668:             this.ChangeTimer.Start();

       669:         }

       670:  

       671:         /// <summary>

       672:         /// Notifies all change.

       673:         /// </summary>

       674:         private void NotifyAllChange()

       675:         {

       676:             this.NotifyPropertyChanged("Head");

       677:             this.NotifyPropertyChanged("LeftHand");

       678:             this.NotifyPropertyChanged("LeftHandX");

       679:             this.NotifyPropertyChanged("LeftHandY");

       680:             this.NotifyPropertyChanged("RightHand");

       681:             this.NotifyPropertyChanged("LeftWrist");

       682:             this.NotifyPropertyChanged("RightWrist");

       683:             this.NotifyPropertyChanged("LeftElbow");

       684:             this.NotifyPropertyChanged("RightElbow");

       685:             this.NotifyPropertyChanged("ShoulderCenter");

       686:             this.NotifyPropertyChanged("HipCenter");

       687:             this.NotifyPropertyChanged("LeftKnee");

       688:             this.NotifyPropertyChanged("RightKnee");

       689:             this.NotifyPropertyChanged("LeftAnkle");

       690:             this.NotifyPropertyChanged("RightAnkle");

       691:             this.NotifyPropertyChanged("LeftFoot");

       692:             this.NotifyPropertyChanged("RightFoot");

       693:             this.NotifyPropertyChanged("Spine");

       694:         }

       695:  

       696:         /// <summary>

       697:         /// Notifies the property changed.

       698:         /// </summary>

       699:         /// <param name="propertyName">Name of the property.</param>

       700:         private void NotifyPropertyChanged(string propertyName)

       701:         {

       702:             if (this.PropertyChanged != null)

       703:             {

       704:                 this.PropertyChanged(this, new PropertyChangedEventArgs(propertyName));

       705:             }

       706:         }

       707:  

       708:         #endregion

       709:     }

       710: }

Step 3 – Getting some sort of visual representation.

Before we do anything with the points that are coming through, the first thing to do is to simply display the camera feed from the Kinect. This is very easy to do.

In the XAML, add an image control (i.e. called “cameraFeed”). In the code, when the main window is created, set up an event handler for the Kinect’s ImageFrameReady property; and it will be called once the SDK has picked up an image.

Within the handler, create a ‘PlanarImage’ from the event args, and then set the image source of the image control to a new BitmapSource as seen below:

         1: public MainWindow()

         2: {

         3:     InitializeComponent();

         4:     //other stuff here        

         5:     model.Kinect.ImageFrameReady += new EventHandler<Microsoft.Research.Kinect.Nui.ImageFrameReadyEventArgs>(kinect_ImageFrameReady);

         6: }

         7:  

         8: void kinect_ImageFrameReady(object sender, Microsoft.Research.Kinect.Nui.ImageFrameReadyEventArgs e)

         9: {

        10:     PlanarImage image = e.ImageFrame.Image;

        11:     cameraFeed.Source = BitmapSource.Create(image.Width, image.Height, 96, 96, PixelFormats.Bgr32, null, image.Bits, image.Width * image.BytesPerPixel);

        12: }

Now let’s do something with the actual points we have obtained.

What we wanted to do here was to create a very simple “image” of a skeleton on the screen.

One thing we did here was to disregard some points from the skeleton. We found it irrelevant to display a few points (i.e. shoulder blades, and sides of the hips etc.), as it didn’t really make the skeleton look or perform any differently, and by excluding them, improved the performance of the entire application.

So, with what remained, we used a combination of ellipses and lines to represent the skeleton.

Simple enough to start off with, add an items control to the page, and within its Data Template, place the 16 ellipses (representing each joint), and a line to connect to each set of relevant ellipses (sort of like the song…”the wrist bone is connected to the elbow”, and so forth).

Each ellipse was created as per below:

         1: <Ellipse Height="14" Name="LeftHand" Stroke="Black" Fill="{Binding JointColor, Converter={StaticResource ColorToSolidColorBrushConverter}}" Width="14" Visibility="{Binding LeftHand, Converter={StaticResource JointToVisibilityConverter}}">

         2: <Ellipse.RenderTransform>

         3: <TranslateTransform X="{Binding LeftHand.Position.X, Converter={StaticResource                KinectValueToScreenCoOrdinatesConverterX}, ConverterParameter=0}" Y="{Binding LeftHand.Position.Y, Converter={StaticResource KinectValueToScreenCoOrdinatesConverterY}, ConverterParameter=0}" />

         4:        </Ellipse.RenderTransform>

         5: </Ellipse>

The lines were also very simple:

         1: <Line StrokeThickness="3" Stroke="Black" StrokeEndLineCap="Round" 

         2: X1="{Binding LeftHand.Position.X, Converter={StaticResource KinectValueToScreenCoOrdinatesConverterX}, ConverterParameter=7}" 

         3: Y1="{Binding LeftHand.Position.Y, Converter={StaticResource KinectValueToScreenCoOrdinatesConverterY}, ConverterParameter=7}" 

         4: X2="{Binding LeftWrist.Position.X, Converter={StaticResource KinectValueToScreenCoOrdinatesConverterX}, ConverterParameter=7}" 

         5: Y2="{Binding LeftWrist.Position.Y, Converter={StaticResource KinectValueToScreenCoOrdinatesConverterY}, ConverterParameter=7}" 

         6: Fill="Black" />

By binding the ellipses to each relevant point in the Main View Model, and using the converter that we created earlier, you should obtain something similar to this:-

The Kinect sensor has picked up the skeletons of the two people standing in front of it. By then updating the collection of skeletons and updating the bindings of the items control source, the skeletons should be displayed correctly on the screen.

Now, as the frames come in, and the bindings update, the ellipses will seamlessly move around the screen as you move, and there you have it, you’re very own living skeleton!

In our next post, we will talk about how to write a gesture service with the SDK, which will recognize the gestures the skeletons on screen are making.

Written by Michael Tsikkos and James Glading

F# Console Application Template

MCS UK Solution Development — Fri, 29 Jul 2011 18:03:54 GMT

If you are like me and often use console applications for a variety of purposes you would have found the F# template not much use (in fact a blank code file). As such I decided to put together a more complete Project Template that I could use.

The template can be found on the Visual Studio Gallery:

http://visualstudiogallery.msdn.microsoft.com/031891a9-06c3-47db-9c7d-8c9d4a32546a

When the template is installed you get the following template added to your F# folder when creating a new F# project:

The Project Template creates a Windows Console Application with the following components:

Program.fs: The application entry point defined
Arguments.fs: A simple command line parser placing all arguments into a dictionary
MyConsole.fs: A separate module for the custom console application code

The application entry point is quite simple:

module Program =   
 
     [<EntryPoint>]
     let Main(args) = 
 
         MyConsole.Run args
 
         // main entry point return
         0

The custom console application handles the argument parsing, in addition to providing a TODO for the custom code:

module MyConsole =
         
     let Run args =
 
         // Define what arguments are expected
         let defs = [
             {ArgInfo.Command="a1"; Description="Argument 1"; Required=true };
             {ArgInfo.Command="a2"; Description="Argument 2"; Required=false } ]
 
         // Parse Arguments into a Dictionary
         let parsedArgs = Arguments.ParseArgs args defs
         
         // TODO add your code here 
         Arguments.DisplayArgs parsedArgs
         Console.ReadLine() |> ignore

If one does not wish to use the argument parsing module the ArgInfo list can be removed. This list provides a simple means for ensuring that the start-up arguments are as expected. The returned parsed parameters is merely a Dictionary of the found arguments.

If one so desires one could map this over to the argument parsing provided in the FSharp PowerPack.

For completeness here is the argument parsing code:

module Arguments =
 
     // Type for simple argument checking
     type ArgInfo = { Command:string; Description:string; Required:bool }
 
     // Displays the help arguments
     let DisplayHelp (defs:ArgInfo list) =
         match defs with
         | [] -> Console.WriteLine "No help text defined."
         | _ ->
             Console.WriteLine "Command Arguments:"
             defs
             |> List.iter (fun def ->
                 let helpText = sprintf "-%s (Required=%b) : %s" def.Command def.Required def.Description
                 Console.WriteLine helpText )
 
     // Displays the found arguments
     let DisplayArgs (args:Dictionary<string, string>) =
         match args.Keys.Count with
         | 0 -> Console.WriteLine "No arguments found."
         | _ ->
             Console.WriteLine "Arguments Found:"
             for arg in args.Keys do
                 if String.IsNullOrEmpty(args.[arg]) then
                     Console.WriteLine (sprintf "-%s" arg)
                 else
                     Console.WriteLine (sprintf "-%s '%s'" arg args.[arg])
 
     // Parse the input arguments
     let ParseArgs (args:string array) (defs:ArgInfo list) =
 
         let parsedArgs = new Dictionary<string, string>()
 
         // Ensure help is supported if defintions provided
         let fullDefs = 
             if not (List.exists (fun def -> String.Equals(def.Command, "help")) defs) then
                 {ArgInfo.Command="help"; Description="Display Help Text"; Required=false } :: defs
             else
                 defs
 
         // Report errors
         let reportError errorText =        
             DisplayArgs parsedArgs
             DisplayHelp fullDefs
             let errMessage = sprintf "Error occured: %A" errorText
             Console.Error.WriteLine errMessage
             Console.Error.Flush()
             Environment.Exit(1)
 
         // Capture variables
         let captureArg command value =
             match defs with
             | [] -> parsedArgs.Add(command, value)
             | _ ->                
                 if not (List.exists (fun def -> String.Equals(def.Command, command)) fullDefs) then
                     reportError (sprintf "Command '%s' Not in definition list." command)
                 else
                     parsedArgs.Add(command, value)            
 
         let (|IsCommand|_|) (command:string) =            
             let m = Regex.Match(command, "^(?:-{1,2}|\/)(?<command>.*)$", RegexOptions.IgnoreCase)
             if m.Success then Some(m.Groups.["command"].Value.ToLower()) else None
 
         let rec loop (argList:string list) =
             match argList with
             | [] -> ()
             | head::tail ->
                 match head with
                 | IsCommand command ->
                     match tail with
                     | [] -> captureArg command String.Empty
                     | iHead::iTail -> 
                         match iHead with
                         | IsCommand iCommand ->
                             captureArg command String.Empty
                             loop tail
                         | _ ->
                             captureArg command iHead
                             loop iTail
                 | _ -> reportError (sprintf "Expected a command but got '%s'" head)
         loop (Array.toList args)
         
         // Look to see if help has been requested if not check for required
         if (parsedArgs.ContainsKey("help")) then
             DisplayHelp defs
         else
             defs
             |> List.filter (fun def -> def.Required) 
             |> List.iter ( fun def ->
                 if not (parsedArgs.ContainsKey(def.Command)) then
                     reportError (sprintf "Command '%s' found but in argument list." def.Command))
                   
         parsedArgs

As always, hopefully you will find this template useful.

Written by Carl Nolan

HTML 5 / JavaScript: Creating a Carousel

MCS UK Solution Development — Fri, 29 Jul 2011 17:33:39 GMT

Following on from my previous posts on html5 animation and reflection effects, I thought I'd put it all together and create a carousel.

Firstly, I create a JavaScript pseudo class called CarouselImage. This takes an image url and draws it at a given co-ordinate, scaled by a co-efficient (used to simulate perspective). This class also applies a reflection effect (as per my previous post).

       function CarouselImage(url, x, y, scale) {

          this.imageUrl = url;

          this.xPosition = x;

          this.yPosition = y;

          this.scaleCoeff = scale;

          this.DrawImageWithReflection = function(currentContext) {

              var mainImage = new Image();

              mainImage.src = this.imageUrl;

              var xPos = this.xPosition;

              var yPos = this.yPosition;

              var scale = this.scaleCoeff;

              mainImage.onload = function () {

                  var imgWidth = mainImage.width * scale;

                  var imgHeight = mainImage.height * scale;

                  // Draw main image

                  currentContext.drawImage(mainImage, xPos, yPos, imgWidth, imgHeight);

                  // Setup a reflection (via reversing scale in y-direction around an axis that is two times the height of the image)  

                  currentContext.translate(0, yPos + (2 * imgHeight));

                  currentContext.scale(1, -1);

                  currentContext.drawImage(mainImage, xPos, 0, imgWidth, imgHeight);

                  // Revert transform and scale  

                  currentContext.translate(0, yPos + (2 * imgHeight));

                  currentContext.scale(1, -1);

                  // Reflection image overlay, to created fade out effect (solid colour with increasing opacity).

                  var alphaGradient = currentContext.createLinearGradient(xPos, yPos + imgHeight, xPos, yPos + (2 * imgHeight));

                  alphaGradient.addColorStop(0, "rgba(255, 255, 255, 0.75)");

                  alphaGradient.addColorStop(0.75, "rgba(255, 255, 255, 1)");

                  currentContext.fillStyle = alphaGradient;

                  currentContext.fillRect(xPos - 1, yPos + imgHeight + 1, imgWidth + 2, imgHeight + 2); // Rectangle made 1px bigger in all directions otherwise scaling leave artifacts

              }

          }

      }

I then create a pseudo class to calculate the position of images in the carousel. I work out the angle of separation between each image by dividing 360 degrees (2 * Math.PI) by the number of images . I use sine and cosine functions to calculate the image positions on an oval (with co-efficients so that the oval is wider than it is tall to create perspective). I finally scale images, so that images higher up the page are smaller, thus appearing further away.

      function CarouselPosition(centerX, centerY, ovalWidth, ovalHeight) {

          this.Angle = 0;

          this.CenterX = centerX;

          this.CenterY = centerY;

          this.OvalWidth = ovalWidth;

          this.OvalHeight = ovalHeight;

          this.CalculateDivisions = function (numberOfItems) {

              this.Angle = (2 * Math.PI / numberOfItems);

          }

          this.CalculateXPosition = function (itemPosition, offsetAngle) {

              return this.OvalWidth * Math.sin((this.Angle * itemPosition) + offsetAngle) + this.CenterX;

          }

          this.CalculateYPosition = function (itemPosition, offsetAngle) {

              return this.OvalHeight * Math.cos((this.Angle * itemPosition) + offsetAngle) + this.CenterY;

          }

          this.CalculateScale = function (itemPosition, offsetAngle) {

              return 1 - (0.2 * -Math.cos((this.Angle * itemPosition) + offsetAngle)) - 0.2;

          }

      }

I create a top level class is called Carousel. This takes an array of image urls and plots them at calculated locations via the above classes. An important point to note is that as setting the z-position does not appear to be possible in the Canvas element, I work out all of the images positions and store in array before plotting them . This allows me to sort based on y-position, so images higher up the page appear at the back of the carousel (thus simulating z-position).

      function Carousel(currentContext, images) 

      {

          this.drawingContext = currentContext;

          this.carouselItems = images;

          this.offsetAngle = 0;

          this.DrawCarousel = function () {

              // Clear the screen

              this.drawingContext.clearRect(0, 0, 1000, 800);

              // Set the position of the caraousel and calculate the angle between each image

              var position = new CarouselPosition(400, 200, 350, 150);

              position.CalculateDivisions(this.carouselItems.length);

              // Update offset angle - this is updated each frame for animation

              this.offsetAngle += 0.01;  

              // We first calculate the positions of all images and put into an array

              var zSortedArray = new Array();

              for (i = 0; i < this.carouselItems.length; i++) {

                  var x = position.CalculateXPosition(i, this.offsetAngle);

                  var y = position.CalculateYPosition(i, this.offsetAngle);

                  var scale = position.CalculateScale(i, this.offsetAngle);

                  var imgUrl = this.carouselItems[i];

                  var currentImage = new CarouselImage(imgUrl, x, y, scale);

                  zSortedArray.push(currentImage);

              }

              // Sort by y position (to simulate z-index)

              zSortedArray.sort(this.PeformSort);

              // Draw images that have been sorted, so images higher on the screen appear in the background

              for (i = 0; i < zSortedArray.length; i++) {

                  var image = zSortedArray[i];

                  image.DrawImageWithReflection(this.drawingContext);

              }

          }

          this.PeformSort = function(a, b) {

              if (a.yPosition > b.yPosition) {

                  return 1;

              }

              return -1;

          }

      }

In the page onload event, I create an instance of my Carousel object, passing in an array of images paths (these make up the nodes on the carousel). I finally set a setInterval method to redraw the canvas every 40ms (i.e. to produce 25 frames per second).

      this.window.onload = function () {

          var canvas = document.getElementById('myCanvas');

          var context = canvas.getContext('2d');

          var images = new Array("images/image1.jpg", "images/image2.jpg", "images/image3.jpg", "images/image4.jpg", "images/image5.jpg", "images/image6.jpg", "images/image7.jpg", "images/image8.jpg")

          this.carousel = new Carousel(context, images);

          setInterval('this.carousel.DrawCarousel();', 40);

      }

Note that at present, I haven’t set up any event handling (e.g. responding to mouse click, or arrow key presses).

Also note that I have tested this on IE9, Firefox and Chrome. It runs smoothly on IE9, but there’s a reasonable amount of flicker on the other two.

Click here to see it in action.

Please feel free to get in contact if you have any questions.

Written by Rob Nowik

FSharpChart and consuming within a WinForms application

MCS UK Solution Development — Sun, 24 Jul 2011 09:04:01 GMT

In all the previous discussions around FSharpChart, samples have been shown within F# Interactive. I wanted to spend a few moments and demonstrate how FSharpChart can be consumed within a WinForms application, and how chart properties can be modified at runtime.

Before getting started a quick word is warranted about F# WinForms applications. By default no project template is installed with Visual Studio. However I have posted a Project Template onto the Visual Studio Gallery that can be used for WinForms applications:

http://visualstudiogallery.msdn.microsoft.com/eba78049-a17e-4868-9ead-065da1421052

Using this template here is some sample code that defines a BoxPlot chart, uses this to create a ChartControl, adds the ChartControl to the form’s Control collection, and renders the Form:

type public BoxplotForm() as form =
     inherit Form()
 
     let chart =
         let rnd = new Random()
 
         FSharpChart.BoxPlot
              ( [ "Eggs", [| for i in 0 .. 20 -> float (rnd.Next 20) |]
                  "Bacon", [| for i in 0 .. 20 -> float (rnd.Next 15 + 2) |]
                  "Sausage", [| for i in 0 .. 20 -> float (rnd.Next 5 + 5) |]
                  "Beans", [| for i in 0 .. 20 -> float (rnd.Next 10 + 3) |]
                  "Mushroom", [| for i in 0 .. 20 -> float (rnd.Next 15 + 5) |] ],
                  BoxPlotShowUnusualValues = true, BoxPlotShowMedian = true,
                  BoxPlotShowAverage = true, BoxPlotWhiskerPercentile = 10,
                  Name = "Breakfast Food BoxPlot")
         |> FSharpChart.WithMargin(1.0f, 5.0f, 1.0f, 1.0f) 
         |> FSharpChart.WithSeries.Style(Color = Color.SeaGreen)
         |> FSharpChart.WithLegend(InsideArea = false, Font = new Font("Arial", 8.0f), Alignment = StringAlignment.Center, Docking = Docking.Top)
 
     // initialize your controls
     let chartControl = new ChartControl(chart, Dock = DockStyle.Fill) 
     let series = new MultiValue()  
 
     do
         // get access to the series
         series.BindSeries(chartControl.ChartSeries)
 
         // add controls to the form
         form.SuspendLayout();
         form.Controls.Add(chartControl)
 
         // define form properties
         form.ClientSize <- new Size(600, 600)
         form.Text <- "F# BoxPlot Tester Utility"
 
         // render
         form.ResumeLayout(false)
         form.PerformLayout()

As you can see not much code is required. Here is the rendered form:

A quick word is warranted about the use of the series value. This value is bound to the data series and as such can be used to modify the data rendered by the chart. All one has to do is call SetData:

let rnd = new Random()
 series.SetData([ "Cereal", [| for i in 0 .. 20 -> float (rnd.Next 20) |]
                  "Yogurt", [| for i in 0 .. 20 -> float (rnd.Next 15 + 2) |]
                  "Toast", [| for i in 0 .. 20 -> float (rnd.Next 15 + 2) |]
                  "Fruit", [| for i in 0 .. 20 -> float (rnd.Next 10 + 5) |] ],
                  chart.ChartBinder)

In this instance series variable is defined as a MultiValue() type. However for other chart types this type will vary; such as OneValue, TwoValue, ThreeValue, and FourValue.

Normally when setting the series data, using SetData, one just has to provide the new data. For the MultiValue type however a binder property is needed so the binding can optionally modify the chart properties.

Finally, one also has the option of setting other runtime chart properties; such as Name, Title, Margin, Legend, and any Custom property. One just has to set the property value, such as:

chart.Margin <- (2.0f, 12.0f, 2.0f, 2.0f)
 chart.Title <- Helper.CreateTitle("Chart Sin/Cosine", Font = new Font("Arial", 12.0f, FontStyle.Bold))
 chart.Legend <- Helper.CreateLegend(InsideArea = false, Font = new Font("Arial", 8.0f))

Hopefully these code snippets demonstrate that using FSharpChart within WinForms applications is relatively simple.

These samples can be found in the FSharpChart download.

Written by Carl Nolan

Adventures in TSQL: Adding date and time values

MCS UK Solution Development — Tue, 19 Jul 2011 20:23:50 GMT

With the addition of the SQL Server date and time types, I have often found myself needing to create a datetime (or datetime2) value based on the addition of a date and a time value. However, there is no built-in function for such an operation.

There are a few solutions to this problem that encompass processing strings; such as:

DECLARE @nowdate date = SYSDATETIME();        
DECLARE @nowtime time = SYSDATETIME();        
        
SELECT CONVERT(datetime2,CONVERT(varchar, @nowdate, 112) + ' ' + CONVERT(varchar, @nowtime, 108))

However, if one realizes that when a time value is cast to a datetime the date component is set to '1900-01-01', and this date equates to a days equivalent of zero in the SQL date and time functions. Thus the following user-defined function can be used to add a date and time value:

CREATE FUNCTION [dbo].[DateTimeAdd]        
(        
      @datepart         date,        
      @timepart         time        
)        
RETURNS datetime2        
AS        
BEGIN        
      RETURN DATEADD(dd, DATEDIFF(dd, 0, @datepart), CAST(@timepart AS datetime2));        
END

This works as the DATEDIFF function with a zero value and a date returns the number of days from the date '1900-01-01'. Thus adding the DATEDIFF result to a time value cast to a datetime value, effectively adds the date and time values.

Thus one can now write:

DECLARE @nowdate date = SYSDATETIME();        
DECLARE @nowtime time = SYSDATETIME();        
        
SELECT dbo.DateTimeAdd(@nowdate, @nowtime);

In the same manner if one wanted to extract just the date portion of a datetime variable, one option is to perform a cast to and from a date type. However one can also start from day 0 and add the corresponding number of days:

CREATE FUNCTION [dbo].[DatetimeDateOnly]        
(        
    @datetime datetime2        
)        
RETURNS datetime2        
AS        
BEGIN        
    RETURN DATEADD(dd, 0, DATEDIFF(dd, 0, @datetime));        
END

Written by Carl Nolan

F# and Running Parallel Tasks

MCS UK Solution Development — Thu, 14 Jul 2011 20:22:24 GMT

Recently I have been working a lot with the Task Parallel Libraries. In doing so, in F#, you will quickly learn that one has to perform quite a bit of casting. As such I have found it useful to define some wrapper members that abstract these casting and calling schematics:

type ParallelEx =
 
     static member inline Do (actions:(unit -> unit) array, ?cancellationToken:CancellationToken) =
         match actions with
         | [||] -> ()
         | [|action|] -> action()
         | _ ->
             let tasks = 
                 match cancellationToken with
                 | Some c -> Array.map (fun x -> Task.Factory.StartNew(Action(x), c)) actions
                 | None -> Array.map (fun x -> Task.Factory.StartNew(Action(x))) actions
             Task.WaitAll tasks
 
     static member inline Let (funcs:(unit -> 'T) array, ?cancellationToken:CancellationToken) =
         match funcs with
         | [||] -> Array.empty
         | [|func|] -> [| func() |]
         | _ ->
             let tasks = 
                 match cancellationToken with
                 | Some c -> Array.map (fun x -> Task.Factory.StartNew(Func<'T>(x), c)) funcs
                 | None -> Array.map (fun x -> Task.Factory.StartNew(Func<'T>(x))) funcs
             Task.WaitAll [| for task in tasks -> task :> Task |]
             Array.map (fun (t: Task<'T>) -> t.Result) tasks

Each of these members takes an array of functions to execute, in parallel as tasks, only returning when all tasks have completed, performing a WaitAll on the tasks. For the Let member any results are then collected and returned in an array.

These wrappers use Array prototypes, rather than lists, mostly because the TPL uses arrays. This means one does not have to do any type conversions between list and array types; and vice-versa.

One thing you will notice in the Let member is the casting of the tasks to type Task, when performing the WaitAll. This is because the tasks, when created using Func types, are of type Task<’T>. Whereas C# will automatically perform the downcast, this has to be explicitly performed in F#.

These wrappers currently support an optional CancellationToken. They could easily be extended for other Task capabilities such as state and TaskCreationOptions. If I extend these in the near future I will post them up.

A quick word about Exceptions. As these members are just wrappers over the TPL you will need to cater for exceptions thrown by the TPL, including the AggregateException.

To demonstrate using the Do member, here is the call one would use when performing a parallel Quicksort:

ParallelEx.Do [|       
fun () -> quickSortDepth low (pivot - 1) (depth - 1);        
fun () -> quickSortDepth (pivot + 1) high (depth - 1) |]

As you can see, using these wrappers saves one having to always perform Action and Func casts for parallel executions.

If one so desired one could just as easily put a wrapper over Parallel.Invoke, rather than using tasks:

static member inline Invoke (actions:(unit -> unit) array) =
     match actions with
     | [||] -> ()
     | [|action|] -> action()
     | _ ->
         Array.map (fun x -> Action(x)) actions
         |> Parallel.Invoke

The usage schematics for this wrapper would be the same as that for the Do member.

Written by Carl Nolan