F# Transaction Monad. Examples.

Ich habe jetzt bei der Arbeit so ein Fall, wo Transaction Monad alternativ zur aktuellen Programmlogik eingesetzt werden könnte, vorausgesetzt wir verwenden F#.
Es sind zwei unterschiedliche Datenbanksysteme im Einsatz und die Programmlogik sollte für die richtige Transaktionabwicklung sorgen.
Daraus ergeben sich zwei Transaktion Fällen.

1. Innere Transaktion.

open TransactionM.Type
open TransactionM.Builder
open TransactionM.Helpers
open System.Data
open System.Data.SqlClient
open System.Threading.Tasks


type Params = {conn : SqlConnection; statements : string list; simulateCancel : bool}
type Either<'a,'b> = 
    | Result of 'a
    | Fail of 'b

let dbCommit (tr:SqlTransaction) =
    tr.Commit()
    tr

let dbRollback (tr:SqlTransaction) =
    tr.Rollback()
    tr

let beginDBTransaction (cnn:SqlConnection) cancel = 
    match cancel with
    | false -> cnn.BeginTransaction()
    | true -> failwith "error beginDBTransaction"

//check : string -> string -> string -> string -> seq<'a * 'b>
// select data from table to the seq.
let inline check dataSource db userId pswd = seq{
    let connStr =   
        new SqlConnectionStringBuilder(DataSource = dataSource,
                InitialCatalog = db, UserID = userId, Password = pswd)
    use conn = new SqlConnection(connStr.ConnectionString)
    conn.Open()
    use comm = new SqlCommand("SELECT Value1, Value2 FROM TEST_TABLE", conn)
    use reader = comm.ExecuteReader()
    while reader.Read() do
        yield unbox reader.["Value1"], unbox reader.["Value2"] }

let inline execNonQuery conn tr s=
    use comm = new SqlCommand(s, conn, tr)
    let res = comm.ExecuteNonQuery() 
    res

let inline deleteAll dataSource db userId pswd = 
    let connStr =   
        new SqlConnectionStringBuilder(DataSource = dataSource,
                InitialCatalog = db, UserID = userId, Password = pswd)
    use conn = new SqlConnection(connStr.ConnectionString)
    conn.Open()
    use comm = new SqlCommand("DELETE FROM TEST_TABLE", conn)
    comm.ExecuteNonQuery()    

//createTransaction: Params -> TransactionM<'a,'b,TransactionState<string,SqlTransaction>>
let createTransaction p =
    transaction {
                    try 
                        let tr = beginDBTransaction p.conn p.simulateCancel
                        try
                            match Seq.forall (((<) 0) << execNonQuery p.conn tr) p.statements with
                            | true -> 
                                return Commit tr
                            | false ->                
                                return Rollback tr
                        with
                        | e ->   
                            printfn " catch %A" (e.Message)
                            return Rollback tr
                    with
                        | e ->   
                            printfn " catch %A" (e.Message)
                            return Abort (Some e.Message)
                }

//createTransactionWithResult: Params -> TransactionM<'a,'b,TransactionState<string,(SqlTransaction * int)>>
let createTransactionWithResult p  = transaction {
        try 
            let tr = beginDBTransaction p.conn p.simulateCancel
            try
                let l = List.map (execNonQuery p.conn tr) p.statements
                match Seq.forall ((<) 0) l with
                | true -> 
                    return Commit (tr, List.rev l |> List.head)
                | _ ->             
                    return Rollback (tr, 0)
            with
            | e ->   
                printfn " catch %A" (e.Message)
                return Rollback (tr, 0)
        with
            | e ->   
                printfn " catch %A" (e.Message)
                return Abort (Some e.Message)

    }

//createOuterTransaction: 
//  TransactionM<'a,'b,TransactionState<'c,('d * 'e)>> -> 
//      Params -> 
//          TransactionHandle<'a,'b,TransactionState<string,(Either<SqlTransaction,'f> * Either<'d,'c option>)>> -> 
//              TransactionM<'a,'b,('g -> 'g)>
let createOuterTransaction inner parms handle =
    transaction 
        {
            try
                //begin outer DB transaction.
                let outer = (beginDBTransaction parms.conn parms.simulateCancel)
                try
                    // start statement from outer transaction. 
                    match execNonQuery parms.conn outer (parms.statements.Item 0)  with
                    | r when r > 0 -> 
                        //get inner transaction monad.
                        let! innerTransactionState = inner
                        match innerTransactionState with
                        | Commit (innerTransaction, result) -> 
                            // final statement from outer transaction.
                            match execNonQuery parms.conn outer ((parms.statements.Item 1).Replace("$$", result.ToString())) with
                            |  r when r > 0 -> 
                                //commit outer and inner transaction.
                                return! commit handle (Result outer, Result innerTransaction)
                            | _ -> 
                                return! rollback handle (Result outer, Result innerTransaction)
                        | Abort m ->
                            return! rollback handle (Result outer, Fail m)
                        | Rollback (innerTransaction, _) -> 
                            return! rollback handle (Result outer, Result innerTransaction)
                        | _ ->
                            return! rollback handle (Result outer, Fail None)
                    | _ -> 
                        return! rollback handle (Result outer, Fail None)
                with
                | e ->   
                    printfn " catch outer %A" (e.Message)
                    return! rollback handle (Result outer, Fail None)
            with
            | e ->   
                printfn " catch %A" (e.Message)
                return! abort handle (Some e.Message)
        }
//runOuterInnerTransaction : string list -> bool -> string list -> bool -> string
let runOuterInnerTransaction innerStatements innerCancel outeStatements outerCancel =
    let connStr =   
        new SqlConnectionStringBuilder(DataSource = "SOURCE1",
                InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
    let connStr1 =   
        new SqlConnectionStringBuilder(DataSource = "SOURCE2",
                InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
    use con = new SqlConnection(connStr.ConnectionString)
    con.Open()

    use con1 = new SqlConnection(connStr1.ConnectionString)
    con1.Open()

    let innerTr = createTransactionWithResult {conn     =       con ; 
                            statements =     innerStatements;
                            simulateCancel = innerCancel }  
    let outerTr = beginT (createOuterTransaction innerTr {conn     =       con1 ; 
                                                           statements =     outeStatements;
                                                           simulateCancel = outerCancel })
    match runTransactionState outerTr () with
    | Commit (Result outer, Result inner)-> 
        dbCommit inner |> ignore
        dbCommit outer |> ignore  
        "commit."
    | Rollback (Result outer, Result inner)-> 
        dbRollback inner |> ignore
        dbRollback outer |> ignore
        "rollback inner and outer."
    | Rollback (Result outer, Fail (Some m)) ->
        dbRollback outer |> ignore
        "rollback outer." + m
    | Rollback (Result outer, Fail None) ->
        dbRollback outer |> ignore
        "rollback outer." 
    | Rollback (Fail _, Result inner) ->
        failwith "rollback inner without outer."  
    | Abort (Some message) -> 
        message + " abort outer." 
    | Abort None ->
        "abort outer." 
    | _ -> failwith "error."

2. Parallele Transaktionen.

//parallelTasksTransaction : Params -> Params -> bool * string
let parallelTasksTransaction p1 p2 =
    let task p = Task.Factory.StartNew(fun () -> runTransactionState (createTransaction p) ())
    let tasks = [task p1; task p2] |> List.toArray
    let result = 
        Task.Factory.ContinueWhenAll(
                    tasks,
                    (fun (ts:Task<TransactionState<string, SqlTransaction>> []) -> 
                        match ts.[0].Result, ts.[1].Result with
                        | Commit t1, Commit t2 -> 
                            dbCommit t1|>ignore
                            dbCommit t2|>ignore
                            true, "commit."
                        | Commit t1, Rollback t2 -> 
                            dbRollback t1|>ignore
                            dbRollback t2|>ignore
                            false, "rollback. (commit task 1, rollback task 2)"
                        | Rollback t1, Commit t2 -> 
                            dbRollback t1|>ignore
                            dbRollback t2|>ignore
                            false, "rollback. (rollback task 1, commit task 2)"
                        | Rollback t1, Rollback t2 -> 
                            dbRollback t1|>ignore
                            dbRollback t2|>ignore
                            false, "rollback. (rollback task 1, rollback task 2)"
                        | Rollback t1, Abort (Some m) -> 
                            dbRollback t1|>ignore
                            false, "rollback. (rollback task 1, abort task 2)" + m
                        | Abort (Some m), Rollback t2 -> 
                            dbRollback t2|>ignore    
                            false, "rollback. (abort task 1, rollback task 2)." + m
                        | Commit t1, Abort (Some m) -> 
                            dbRollback t1|>ignore
                            false, "rollback. (Commit task 1, abort task 2)" + m
                        | Abort (Some m), Commit t2 -> 
                            dbRollback t2|>ignore    
                            false, "rollback. (abort task 1, Commit task 2)." + m
                        | Abort (Some m1), Abort (Some m2) ->   
                            false, "abort." + m1 + ". " + m2
                        | _ -> failwith "execution error."))
    result.Result

//runParallelTasksTransaction : string list -> bool -> string list -> bool -> bool * string
let runParallelTasksTransaction statements1 cancel1 statements2 cancel2= 
    let connStr =   
        new SqlConnectionStringBuilder(DataSource = "SOURCE1",
                InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
    let connStr1 =   
        new SqlConnectionStringBuilder(DataSource = "SOURCE2",
                InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
    use con = new SqlConnection(connStr.ConnectionString)
    con.Open()

    use con1 = new SqlConnection(connStr1.ConnectionString)
    con1.Open()
    parallelTasksTransaction {conn = con; statements = statements1; simulateCancel = cancel1} 
                             {conn = con1 ; statements = statements2; simulateCancel = cancel2}

Ein Paar Tests.

//test : ('a -> 'b -> 'c -> 'd -> 'e) -> 'a -> 'b -> 'c -> 'd -> unit
let test f a b c d =
    deleteAll "SOURCE1" "TESTDB" "USER" "PWD" |>ignore
    deleteAll "SOURCE2" "TESTDB" "USER" "PWD"  |>ignore
    let res = f a b c d
    let check1 = check "SOURCE1" "TESTDB" "USER" "PWD"
    let check2 = check "SOURCE2" "TESTDB" "USER" "PWD"
    printfn "result: %A" res
    printfn "check data in db1: %A" check1
    printfn "check data in db2: %A" check2
    printfn "%s" (String.replicate 30 "+")

printfn "%s" (String.replicate 50 "-")
printfn "run inner transaction tests."
printfn "test Commit." 
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'start inner 2') ";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (3, 'start inner 3')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'start inner 4')";
                               "UPDATE TEST_TABLE  SET Value1 = 5, Value2 ='end inner 5'  WHERE Value1 =2"] 
                              false
                              ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (10, 'start outer')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES ($$, 'end outer. inner updated $$.') " ] 
                              false
printfn "test Rollback 1." 
test runOuterInnerTransaction ["INSERT INTO NONE"] 
                              false
                              ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (10, 'outer 10')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (20, 'outer 20') " ] 
                              false


printfn "test Rollback 2." 
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'inner 2') "] false
                              ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (10, 'outer 10')";
                               "INSERT INTO NONE " ] false
// simulate error in BeginTransaction.
printfn "test Abort inner." 
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'inner 2') "] true
                              ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (10, 'outer 10')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (20, 'outer 20') " ] false
// simulate error in BeginTransaction.
printfn "test Abort outer." 
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'start inner 2') ";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (3, 'start inner 3')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (2, 'start inner 4')";
                               "UPDATE TEST_TABLE  SET Value1 =5, Value2 ='end inner 5'  WHERE Value1 =2"] 
                              false
                              ["INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (10, 'start outer')";
                               "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES ($$, 'end outer. inner updated $$.') " ] 
                              true

printfn "%s" (String.replicate 50 "-")
printfn "run parallel tasks tests."
printfn "test Commit." 
test runParallelTasksTransaction 
        [for i in 0..100 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"] 
        false
        [for i in 100..150 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"] 
        false

printfn "test Rollback 1." 
test runParallelTasksTransaction 
        ["INSERT INTO NONE"] 
        false
        [for i in 100..150 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"] 
        false

printfn "test Rollback 2." 
test runParallelTasksTransaction 
        [for i in 100..200 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"] 
        false
        ["INSERT INTO NONE"] 
        false
// simulate error in BeginTransaction.
printfn "test Abort." 
test runParallelTasksTransaction 
        [for i in 0..100 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"] 
        false
        [for i in 100..150 -> 
            "INSERT INTO TEST_TABLE  (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"] 
        true

--------------------------------------------------
run inner transaction tests.
test Commit.

result: "commit."
check data in db1: seq [(5, "end inner 5"); (3, "start inner 3"); (5, "end inner 5")]
check data in db2: seq [(10, "start outer"); (2, "end outer. inner updated 2.")]

++++++++++++++++++++++++++++++
test Rollback 1.

 catch "Falsche Syntax in der Nähe von 'NONE'."
result: "rollback inner and outer."
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
test Rollback 2.

 catch "Falsche Syntax in der Nähe von 'NONE'."
result: "rollback inner and outer."
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
test Abort inner.

 catch "error beginDBTransaction"
result: "rollback outer.error beginDBTransaction"
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
test Abort outer.

 catch "error beginDBTransaction"
result: "error beginDBTransaction abort outer."
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
--------------------------------------------------
run parallel tasks tests.
test Commit.

result: (true, "commit.")
check data in db1: seq
  [(0, "Task 1 (0)"); (1, "Task 1 (1)"); (2, "Task 1 (2)"); (3, "Task 1 (3)");
   ...]
check data in db2: seq
  [(100, "Task 2 (100)"); (101, "Task 2 (101)"); (102, "Task 2 (102)");
   (103, "Task 2 (103)"); ...]

++++++++++++++++++++++++++++++
test Rollback 1.

 catch "Falsche Syntax in der Nähe von 'NONE'."
result: (false, "rollback. (rollback task 1, commit task 2)")
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
test Rollback 2.

 catch "Falsche Syntax in der Nähe von 'NONE'."
result: (false, "rollback. (commit task 1, rollback task 2)")
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++
test Abort.

 catch "error beginDBTransaction"
result: (false, "rollback. (Commit task 1, abort task 2)error beginDBTransaction
")
check data in db1: seq []
check data in db2: seq []

++++++++++++++++++++++++++++++

F# Transaction Monad.

Nachdem ich sehr interessante Beiträge über die verschiedensten F# Monads begeistert gelesen habe, überlegte ich mir, ob man so was wie eine Transaction Monad implementieren kann.
Und tatsächlich gibt es bereits eine Haskell Version. Hundertprozentig ist die Funktionsweise von der Monad für mich noch nicht klar, aber soweit ich es beurteilen kann ist die Transaktion ein Hybrid aus der Continuation und der State Monad.

Erstmal ein paar Tests:

open TransactionM
// 5 ways you can leave the monad.
// handle : transaction handle.
let test0 handle = 
    transaction {
      let! s = get
      do! set 99
      match s with
        | 0 -> return id
        | 1 -> return! abort    handle (Some s)
        | 2 -> return! dirty    handle (Some s.)
        | 3 -> return! rollback handle  "rollback!"
        | _ -> return! commit   handle  "commit"
    }
// return TransactionState<int,string> * int. second item is result of transaction.
let runTest0  = 
    let run = runTransaction_ (beginT test0)
    List.map run [0..4]

val test0 :
  TransactionM.TransactionHandle<'a,int,
                                 TransactionM.TransactionState<int,string>> ->
    TransactionM.TransactionM<'a,int,('b -> 'b)>

val runTest0 : (TransactionM.TransactionState<int,string> * int) list =

  [ (Abort null, 0); 
    (Abort (Some 1), 1); 
    (Dirty (Some 2), 99);
    (Rollback "rollback!", 3);
    (Commit "commit", 99) ]

Einfache Listenmanipulation als eine Transaktion.

// Simple list manipulation as transaction.
// p : some condition
// l : init list
// handle : transaction handle.
let testList p l handle = transaction {
        do! set l
        do! modify (fun xs-> 6::xs)
        do! modify (fun xs-> 7::xs)       
        match p  with
        | false -> return! rollback handle  "rollback!" 
        | true -> return! commit   handle   "commit."
    }
// Only if both transactions are successful then concatenate the two lists and commit all transactions.
// rollback otherwise.
// m1, m2 - transactions.
// handle : transaction handle.
let merge m1 m2 handle = 
    transaction {
            let! state1 = m1
            match state1 with
            | Commit a ->  
                let! firstList = get
                printfn "    first list: %A" firstList 

                let! state2 = m2
                match state2 with
                | Commit b   ->  
                    let! secondList = get
                    printfn "    second list: %A" secondList
                    do! set (firstList @ secondList)
                    return! commit   handle  b

                | Rollback b ->  return! rollback   handle  b
                | _          ->  return! abort      handle (Some "abort")

            | Rollback a    ->   return! rollback   handle  a                        
            | _             ->   return! abort      handle (Some "abort")
        }
//return TransactionState<string,'b> * 'c list. second item is result of transaction.
let runMerge i m1 m2 = 
    printfn "Start runMerge %A." i
    let m = beginT (merge m1 m2)
    runTransaction_ m [] 
// ls : list of list.
// return TransactionState<string,string> * int list. second item is result of transaction.
let runList ls = 
    printfn "Start runList."
    let m = List.fold (fun acc (l, p) -> beginT (merge acc (beginT (testList p l)))) (alwaysCommit "commit") ls
    runTransaction_ m []

printfn "runMerge 1: %A " (runMerge 1 (beginT (testList true  [0..3] )) (beginT (testList true  [10..13])))
printfn "runMerge 2: %A " (runMerge 2 (beginT (testList false [0..3] )) (beginT (testList true  [10..13])))

printfn "%A" (runList  [([0..3], true); ([10..13], true); ([20..23], true)])
printfn "%A" (runList  [([0..3], true); ([10..13], true); ([20..23], false)])

val testList :
  bool ->
    int list ->
      TransactionM.TransactionHandle<'a,int list,
                                     TransactionM.TransactionState<'b,string>> ->
        TransactionM.TransactionM<'a,int list,('c -> 'c)>
val merge :
  TransactionM.TransactionM<'a,'b list,TransactionM.TransactionState<'c,'d>> ->
    TransactionM.TransactionM<'a,'b list,TransactionM.TransactionState<'e,'d>> ->
      TransactionM.TransactionHandle<'a,'b list,
                                     TransactionM.TransactionState<string,'d>> ->
        TransactionM.TransactionM<'a,'b list,('f -> 'f)>
val runMerge :
  'a ->
    TransactionM.TransactionM<(TransactionM.TransactionState<string,'b> *
                               'c list),'c list,
                              TransactionM.TransactionState<'d,'b>> ->
      TransactionM.TransactionM<(TransactionM.TransactionState<string,'b> *
                                 'c list),'c list,
                                TransactionM.TransactionState<'e,'b>> ->
        TransactionM.TransactionState<string,'b> * 'c list
val runList :
  (int list * bool) list ->
    TransactionM.TransactionState<string,string> * int list

Start runMerge 1.
    first list: [7; 6; 0; 1; 2; 3]
    second list: [7; 6; 10; 11; 12; 13]
runMerge 1: (Commit "commit.", [7; 6; 0; 1; 2; 3; 7; 6; 10; 11; 12; 13]) 

Start runMerge 2.
runMerge 2: (Rollback "rollback!", []) 

Start runList.
    first list: []
    second list: [7; 6; 0; 1; 2; 3]
    first list: [7; 6; 0; 1; 2; 3]
    second list: [7; 6; 10; 11; 12; 13]
    first list: [7; 6; 0; 1; 2; 3; 7; 6; 10; 11; 12; 13]
    second list: [7; 6; 20; 21; 22; 23]
(Commit "commit.",
 [7; 6; 0; 1; 2; 3; 7; 6; 10; 11; 12; 13; 7; 6; 20; 21; 22; 23])

Start runList.
    first list: []
    second list: [7; 6; 0; 1; 2; 3]
    first list: [7; 6; 0; 1; 2; 3]
    second list: [7; 6; 10; 11; 12; 13]
    first list: [7; 6; 0; 1; 2; 3; 7; 6; 10; 11; 12; 13]
(Rollback "rollback!", [])

Interessant ist ob die Transaktionen asynchron ausgeführt werden können. Ich habe es leider nicht hingekriegt.

Hier ist meine F# Implementierung von der Transaktion Monad.

// from http://hackage.haskell.org/packages/archive/monad-tx/0.0.1/doc/html/Control-Monad-Tx.html
module TransactionM 

open System

// 'e : error type
// 'a : transaction state type
type TransactionState<'e,'a> =
    | Begin
    | Abort of ('e option)
    | Dirty of ('e option)
    | Rollback of 'a
    | Commit of 'a

// 's : state
// 'a : TransactionState
// 'r : result 
// ('s -> 'a -> 'r) : continuation
type TransactionM<'r, 's, 'a> = TransactionM of ('s -> ('s -> 'a -> 'r) -> 'r)

type TransactionHandle<'r, 's, 'a> = TransactionHandle of (('a * TransactionHandle<'r, 's, 'a>) -> TransactionM<'r, 's, unit>)

let inline runTransaction (TransactionM g) s k = g s k

// result is of type (TransactionState * state)
let inline runTransaction_ (TransactionM g) s = g s (fun s' a ->  (a, s'))
// result is of type TransactionState
let inline runTransactionState (TransactionM g) s = g s (fun _ a ->  a)

let inline withCommit f = 
    TransactionM (fun s k -> 
                    let (TransactionM g) = f (fun a -> TransactionM (fun s' _ ->  k s' a)) 
                    g s k)

let inline withRollback f = 
    TransactionM (fun s k -> 
                    let (TransactionM g) = (f (fun a -> TransactionM (fun _ _ -> k s a))) 
                    g s k)

let inline bind (TransactionM g) f = 
    TransactionM(fun s k -> 
                    g s (fun s' a ->
                            let (TransactionM g') = f a
                            g' s' k))
//computation workflow builder.
type TransactionBuilder() =
    member this.Return(a)                               = TransactionM(fun s k -> k s a) 
    member this.Bind(m, k)                              = bind m k
    member this.Zero ()                                 = this.Return ()
    member this.Combine(r1, r2)                         = this.Bind(r1, fun _ -> r2) 
    member this.ReturnFrom(m : TransactionM<_,_,_>)     = m
    member this.Delay(f)                                = this.Bind(this.Return (), f)
    
    member this.TryFinally(computation, compensation) =
        TransactionM(fun s k -> 
            try
                runTransaction computation s k
            finally
                compensation())

    member this.Using(res: #IDisposable, body) =
        this.TryFinally(body res,
            (fun () -> match res with null -> () | disp -> disp.Dispose()))

    member this.TryWith(computation, handler) =
        TransactionM(fun s k ->
            try
                runTransaction computation s k
            with e -> runTransaction (handler e) s k)

let transaction = new TransactionBuilder()

let inline bindM builder m f = (^M: (member Bind: 'd -> ('e -> 'c) -> 'c) (builder, m, f))

let inline (>>.) m n = bindM transaction m (fun _ ->  n)

let inline isBegin t =
    match t with
    | Begin -> true
    | _ -> false

let inline fmap f (TransactionM g) = TransactionM (fun s k -> g s (fun s' a -> k s' (f a)))
//begin transaction.
let inline beginT f =
    let checkpoint  = 
        withCommit (fun fcommit ->
            withRollback (fun frollback ->
                transaction {
                    let go (transactionState, handle) =
                        match transactionState with
                        | Begin         -> failwith     "nested"
                        | Abort e       -> frollback    (Abort e,       handle)
                        | Dirty e       -> fcommit      (Dirty e,       handle)
                        | Rollback a    -> frollback    (Rollback a,    handle)
                        | Commit a      -> fcommit      (Commit a,      handle)
                    return (Begin, TransactionHandle go) 
                    } ))

    withRollback (fun fabort ->
        transaction {
                    let! (transactionState, handle)  = checkpoint
                    if isBegin transactionState then
                        return!  (f  handle >>. fabort (Abort None))
                    return transactionState
                 })

//a bunch of helpers, which allow to access and manipulate transaction.

let inline alwaysCommit a = TransactionM(fun s k -> k s (Commit a))

let inline jump (TransactionHandle k) stat = 
    (k (stat, TransactionHandle k)) >>. TransactionM(fun s k -> k s id)

let inline abort    handle e = jump handle (Abort e)

let inline dirty    handle e = jump handle (Dirty e)

let inline rollback handle a = jump handle (Rollback a)

let inline commit   handle a = jump handle (Commit a)

let get = TransactionM (fun s k -> k s s)

let inline gets f = TransactionM (fun s k -> k s (f s))

let inline set s = 
    TransactionM (fun _ k -> k s ())

let inline modify f = TransactionM(fun s k -> k (f s) ())