In mein WF Activity Toolkit gehört unter anderem die RetryActivity von Matt Milner. Wie bereits beschrieben kann mit dieser Custom Activity die Ausführung einer Sequenz wiederholt werden, falls innerhalb der Sequenz ein Fehler auftritt. Dabei können die Anzahl der erneuten Versuche und ein Intervall zwischen den Versuchen angegeben werden. Nun habe ich diese Activity noch ein wenig erweitert, um das Intervall zwischen den Versuchen etwas flexibler konfigurieren zu können.

Erweiterte RetryActivity

Dabei werden zwischen zwei Intervall Modi unterschieden:

• Linear (IntervalExponential = false)
  Die Zeitabstände zwischen den einzelnen Intervallen sind gleichmässig, zB. immer 1 Minute.
• Exponentiell (IntervalExponential = true)
  Die Zeitabstände zwischen den einzelnen Intervallen wird immer verdoppelt bis alle Versuche ausgeschöpft sind. Also wird zB. mit einer Sekunde begonnen, und das nächste Intervall dauert zwei Sekunden. Nach 10 Intervall Schritten wird bereits 512 Sekunden gewartet, also etwa achteinhalb Minuten.

Dies kann nützlich sein, wenn zB. ein Aufruf an ein externes System fehlerhaft sein kann, man aber bei vielen parallelen Workflows das externe System nicht durch zahlreiche gleichzeitige Aufrufe belasten will.

Hier ist der der Source Code der erweiterten RetryActivity: RetryActivityEx.zip

Nachdem ein grösseres Projekt nun beinahe abgeschlossen ist, komme ich auch wieder dazu, ein paar Erfahrungen der letzten Monate zu beschreiben. Die Basis Library der WF 3 enthält ja einige Activities, welche benutzt werden können, um den Ablauf zu modellieren. In einem typischen Projekt kommen dann sofort eigene Custom Activities dazu, welche sich um die Prozess relevanten Punkte kümmern wie zB. den Zugriff auf eine externe DB.

Folgende 3 Custom Activities haben sich als unentbehrlich erwiesen und gehören fest in mein WF Toolkit:

CallWorkflow

Die InvokeWorkflowActivity der WF kann zwar auch andere Workflows ausführen, sie macht dies jedoch asynchron. Das bedeutet, sofort nach dem Aufruf kehrt die Ausführung in den aufrufenden  Workflow zurück. Die CallWorkflowActivity mit dazugehörigem Runtime Service des WF Spezialisten Jon Flanders bietet die synchrone Variante an. Die Ausführung im aufrufenden Workflow wartet solange, bis die aufgerufene Workflow Instanz beendet ist.

Retry

Die RetryActivity von Matt Milner kann immer dann verwendet werden, wenn der Aufruf einer Activity fehlschlagen kann (werfen einer Exception), und man dann das ganze noch einmal ausführen lassen will. Es kann eine Anzahl der Wiederholungsversuche und ein Intervall zwischen den Versuchen angegeben werden.

RetryActivity

PersistencePoint

Die letzte Activity ist eine wirklich simple, aber sehr nützliche. Sie kann im Workflow eingesetzt werden, um einen Speicherpunkt im WF Store zu erzwingen auch wenn die WF dies an dieser Stelle nicht von sich aus tun würde. Wenn im Ablauf ein wichtiger Punkt erreicht wurde, kann dies in die Persitenz DB gespeichert werden, und falls der WF unterbrochen wird, fährt die Ausführung beim letzten Speicherpunkt weiter.

Jede Activity, welche mit dem Attribut [PersistOnClose] ausgezeichnet wird, erreicht dieses Verhalten. Die PersistencePointActivity wird somit trivial:

[PersistOnClose]
public partial class PersistencePointActivity : Activity
{
    public PersistencePointActivity()
    {
        InitializeComponent();
    }
}

Gestern habe ich einen kleinen, aber nützlichen Hack gezeigt, mit welchem man die Service ID des Persistenzdienstes setzen kann. Dies funktioniert auch ganz prima, solange man nicht mit Workflow Services arbeitet.

Wie man es auch versucht, man erhält keine gültige Referenz auf den Persistenz Service.

Also was tun? Reflector zeigt uns die Interna von WorkflowService.ApplyDispatchBehavior():

static class WorkflowExtensions
...
WorkflowPersistenceService service = item.WorkflowRuntime.GetService<WorkflowPersistenceService>();
if (service != null)
{
    bool isStarted = item.WorkflowRuntime.IsStarted;
    if (isStarted)
    {
        item.WorkflowRuntime.StopRuntime();
    }
    item.WorkflowRuntime.RemoveService(service);
    item.WorkflowRuntime.AddService(new SkipUnloadOnFirstIdleWorkflowPersistenceService(service));
    if (isStarted)
    {
        item.WorkflowRuntime.StartRuntime();
    }
}
...

Was geschieht hier genau?
Wenn via Config oder Code ein Persistenz Service hinzugefügt wurde, wird die Runtime gestoppt. Der Standard Persistenz Service wird entfernt, und ein neuer Service mit dem handlichen Namen SkipUnloadOnFirstIdleWorkflowPersistenceService wird hinzugefügt. Danach wird die Runtime wieder gestartet.
Die Workflow Runtime eines Workflow Services verwendet also intern nicht den SqlWorkflowPersistenceService, sondern einen speziellen, öffentlich nicht zugänglichen und gekapselten Dienst.

Wie also kann nun in einem solchen Kontext die Service ID gesetzt werden?

Die gestern gezeigte Erweiterungsmethode kann wie folgt ergänzt werden:

static class WorkflowExtensions
{
    public static void SetServiceInstanceId(this WorkflowPersistenceService workflowPersistenceService, Guid serviceId)
    {
        Type persistenceServiceType = workflowPersistenceService.GetType();
        FieldInfo instanceIdField = persistenceServiceType.GetField("_serviceInstanceId", BindingFlags.NonPublic | BindingFlags.Instance);
        instanceIdField.SetValue(workflowPersistenceService, serviceId);
    }

    public static WorkflowRuntime GetRuntime(this WorkflowServiceHost host)
    {
        return host.Description.Behaviors.Find<WorkflowRuntimeBehavior>().WorkflowRuntime;
    }

    public static T GetWFService<T>(this WorkflowRuntime runtime) where T : class
    {
        if (typeof(T) != typeof(SqlWorkflowPersistenceService))
        {
            return runtime.GetService<T>();
        }
        else
        {
            WorkflowPersistenceService service = runtime.GetService<WorkflowPersistenceService>();
            Type sqps = service.GetType();

            FieldInfo field = (from fieldinfo in sqps.GetFields(BindingFlags.NonPublic | BindingFlags.Instance)
                               where fieldinfo.FieldType == typeof (WorkflowPersistenceService)
                               select fieldinfo).First();
            return field.GetValue(service) as T;
        }
    }
}

Mit der Erweiterungsmethode GetRuntime() erhält man die WF Runtime vom WorkflowServiceHost.
Mittels wfRuntime.GetWFService<SqlWorkflowPersistenceService>() kann danach durch den Zugriff via Reflection auf den intern als Feld gekapselten Persistenzdienst zugegriffen werden.

Eines der mächtigsten Features der WF, der Persistenz Service (SqlWorkflowPersistenceService), erlaubt es einem, den aktuellen Status von Workflow Instanzen in der Workflow DB zu speichern. Microsoft stellt dabei alles, was benötigt wird, zur Verfügung, mindestens für den hauseigenen SQL Server.

Eine Applikation, die beispielsweise auf mehreren Servern gleichzeitig läuft, kann auf dieselbe WF Persistenz Datenbank zugreifen. Dazu wurde das relativ einfache Konzept der Workflow Ownership geschaffen. Dies ist eigentlich nichts anderes als ein einfacher Lock mit einem Timeout.

In der WF Persistenz Datenbank dienen dazu die beiden Felder ownerId und ownedUntil der Tabelle InstanceState:

Aufbau der Tabelle InstanceState

Im Feld ownerId wird die GUID des Persistenz Services der jeweiligen WF Runtime eingetragen, und in ownedUntil steht (im UTC Format!), wie lange die Instanz gesperrt ist.
Wenn diese beiden Felder nicht null sind, wird die entsprechende WF Instanz also gerade von einem Host bearbeitet.
So wird sichergestellt, dass eine Workflow Instanz immer nur von einem Host gleichzeitig ausgeführt wird.

Dies funktioniert grundsätzlich auch prima, der Persistenz Service hat allerdings einige Unschönheiten.

Bei jedem Neustart der WorkflowRuntime wird der Persistenz Service neu initialisiert. Dabei erhält er auch immer eine neu GUID, mit welcher dann die WF Instanzen in der WF DB gelockt werden.
Wenn nun zum Beispiel die Ausführung einer Workflow Instanz abgebrochen wird, bleibt der Lock bestehen.

Nun würde man erwarten, dass die Instanz nach dem Ablaufen des Lock Timeouts von einem anderen Host geladen und weiterbearbeitet werden kann. Leider ist dies nicht der Fall. Der Persistenz Service lädt beim Pollen nach freien Workflow Instanzen nur solche mit abgelaufenem Timer. Diejenigen mit abgelaufenem Lock werden nur beim Start des Services geladen. Solange also der Service nicht neu gestartet wird, kann die Workflow Instanz nicht mehr geladen werden.

Es wäre doch schön, wenn man dem Persistenz Service eine ID zuweisen könnte, damit wäre nämlich das Problem gelöst. In einer Umgebung mit mehreren Hosts, könnte jeder Service seine eigene ID erhalten.
Leider ist dies nicht vorgesehen.

Dem kann man aber abhelfen. Nach einem kurzen Blick auf die Implementation des SqlWorkflowPersistenceService im Reflector kann man dank der in C# 3.0 eingeführten Extension Methods den Service wie folgt erweitern:

static class WorkflowExtensions
{
    public static void SetServiceInstanceId(this WorkflowPersistenceService workflowPersistenceService, Guid serviceId)
    {
        Type persistenceServiceType = workflowPersistenceService.GetType();
        FieldInfo instanceIdField = persistenceServiceType.GetField("_serviceInstanceId",
                                                                    BindingFlags.NonPublic |
                                                                    BindingFlags.Instance);
        instanceIdField.SetValue(workflowPersistenceService, serviceId);
    }
}

Die Service ID wird einfach via Reflection gesetzt. Verwendung auf eigene Gefahr, bei mir hat’s jedenfalls hervorragend funktioniert.