Arrows and Reagents

Guest lecture given at Adv. FP class at University of Cambridge on the practical application of Arrows for static and dynamic analysis using the Reagents library. Reagents are a library for Multicore OCaml for expressing and composing fine-grained lock-free data and synchronization structures.

1. Arrows and Reagents “KC” Sivaramakrishnan Advanced Functional Programming March 3rd, 2016 Slides were borrowed and modified from Aaron Turon’s PLDI 2012 talk: http://www.mpi-sws.org/~turon/pldi-2012-reagents.pdf
2. Arrows module type Arrow = sig type ('a,'b) t val arr : ('a -> ‘b) -> ('a,'b) t val (>>>) : ('a,'b) t -> ('b,'c) t -> ('a,'c) t val first : ('a,'b) t -> ('a * 'c, 'b * 'c) t end 2
3. Arrows module type Arrow = sig type ('a,'b) t val arr : ('a -> ‘b) -> ('a,'b) t val (>>>) : ('a,'b) t -> ('b,'c) t -> ('a,'c) t val first : ('a,'b) t -> ('a * 'c, 'b * 'c) t end Laws arr f >>> arr g ≡ arr (compose g f) (f >>> g) >>> h ≡ f >>> (g >>> h) arr id >>> f ≡ f ... ... 2
4. Functions as Arrows 3 • https://gist.github.com/9eef070c232913121564
5. John Huges, “Generalising Monads to Arrows” “If we think of a library as defining a domain specific 'language', whose constructions are represented as combinators, then the idea is to implement the language via a combination of a static analysis and an optimised dynamic semantics.” 4
6. John Huges, “Generalising Monads to Arrows” “If we think of a library as defining a domain specific 'language', whose constructions are represented as combinators, then the idea is to implement the language via a combination of a static analysis and an optimised dynamic semantics.” 4 val (>>=) : 'a Monad.t -> ('a -> 'b Monad.t) -> 'b Monad.t val (>>>) : ('a, 'b) Arrow.t -> ('b,'c) Arrow.t -> ('a,'c) Arrow.t
7. Functions with cost as Arrows 5 • https://gist.github.com/66fcc8c01b563282ef42 • https://gist.github.com/644fbe3d36f90d98faa1
8. Reagents • DSL for expressing and composing fine-grained concurrency libraries • Aaron Turon, “Reagents: expressing and composing fine- grained concurrency”, PLDI 2012 • Based on Arrows • Enable dynamic optimisations • Built on k-compare-and-swap abstraction 6
9. Compare-and-swap (CAS) module CAS : sig val cas : 'a ref -> expect:'a -> update:'a -> bool end = struct (* atomically... *) let cas r ~expect ~update = if !r = expect then (r:= update; true) else false end 7
10. Compare-and-swap (CAS) module CAS : sig val cas : 'a ref -> expect:'a -> update:'a -> bool end = struct (* atomically... *) let cas r ~expect ~update = if !r = expect then (r:= update; true) else false end • Implemented atomically by processors • x86: CMPXCHG and friends • arm: LDREX, STREX, etc. • ppc: lwarx, stwcx, etc. 7
11. CAS: cost versus contention Threads 2 4 6 8 Conention (log-scale) 100% 0.33% 0.25% 0.2% Throughput Sequential 1.0 0.81 0.62 0.42 0.23 0.04 0.5% 1% 2% 8
12. java.util.concurrent Synchronization Data structures Reentrant locks Semaphores R/W locks Reentrant R/W locks Condition variables Countdown latches Cyclic barriers Phasers Exchangers Queues Nonblocking Blocking (array & list) Synchronous Priority, nonblocking Priority, blocking Deques Sets Maps (hash & skiplist) 9
13. java.util.concurrent Synchronization Data structures Reentrant locks Semaphores R/W locks Reentrant R/W locks Condition variables Countdown latches Cyclic barriers Phasers Exchangers Queues Nonblocking Blocking (array & list) Synchronous Priority, nonblocking Priority, blocking Deques Sets Maps (hash & skiplist) Not Composable 9
14. module type TREIBER_STACK = sig type 'a t val push : 'a t -> 'a -> unit ... end module Treiber_stack : TREIBER_STACK = struct type 'a t = 'a list ref let rec push s t = let cur = !s in if CAS.cas s cur (t::cur) then () else (backoff (); push s t) end 10
15. 3 2 Head 11
16. 3 2 Head 7 11
17. 3 2 Head 7 5 11
18. 3 2 Head 7 5 CAS fail 11
19. 3 2 Head 7 5 11
20. 3 2 Head 7 5 12
21. module type TREIBER_STACK = sig type 'a t val push : 'a t -> 'a -> unit val try_pop : 'a t -> 'a option end module Treiber_stack : TREIBER_STACK = struct type 'a t = 'a list ref let rec push s t = ... let rec try_pop s = match !s with | [] -> None | (x::xs) as cur -> if CAS.cas s cur xs then Some x else (backoff (); try_pop s) end 13
22. Concurrency libraries are indispensable, but hard to build and extend The Problem: let v = Treiber_stack.pop s1 in Treiber_stack.push s2 v is not atomic 14
23. Scalable concurrent algorithms can be built and extended using abstraction and composition The Proposal: Treiber_stack.pop s1 >>> Treiber_stack.push s2 is atomic 15
24. Design 16
25. Lambda: the ultimate abstraction f 'a 'b val f : 'a -> 'b 17
26. Lambda: the ultimate abstraction f 'a 'b g 'b 'c val f : 'a -> 'b val g : 'b -> 'c 18
27. Lambda: the ultimate abstraction f 'a g 'b 'c (compose g f): 'a -> 'c 19
28. f 'a 'b Lambda abstraction: 20
29. f 'a 'b Lambda abstraction: Reagent abstraction: 'a 'b R ('a,'b) Reagent.t 20
30. Reagent combinators module type Reagents = sig type ('a,'b) t val never : ('a,'b) t val constant : 'a -> ('b,'a) t val (>>>) : ('a,'b) t -> ('b,'c) t -> ('a,'c) t module Ref : Ref.S with type ('a,'b) reagent = ('a,'b) t module Channel : Channel.S with type ('a,'b) reagent = ('a,'b) t val run : ('a,'b) t -> 'a -> ‘b ... end 21
31. module type Channel = sig type ('a,'b) endpoint type ('a,'b) reagent val mk_chan : unit -> ('a,'b) endpoint * ('b,'a) endpoint val swap : ('a,'b) endpoint -> ('a,'b) reagent end 22
32. c: ('a,'b) endpoint c swap 'a 'b module type Channel = sig type ('a,'b) endpoint type ('a,'b) reagent val mk_chan : unit -> ('a,'b) endpoint * ('b,'a) endpoint val swap : ('a,'b) endpoint -> ('a,'b) reagent end 22
33. c: ('a,'b) endpoint c swap 'a 'b c swap 'b 'a module type Channel = sig type ('a,'b) endpoint type ('a,'b) reagent val mk_chan : unit -> ('a,'b) endpoint * ('b,'a) endpoint val swap : ('a,'b) endpoint -> ('a,'b) reagent end 22
34. c swap 'a 'b c: ('a,'b) endpoint 23
35. swap Message passing type 'a ref val upd : 'a ref -> f:(‘a -> 'b -> ('a * ‘c) option) -> ('b, 'c) Reagent.t 24
36. swap upd f r 'a 'a 'b 'c Message passing type 'a ref val upd : 'a ref -> f:(‘a -> 'b -> ('a * ‘c) option) -> ('b, 'c) Reagent.t 24
37. swap upd f Message passing Shared state 25
38. swap upd f 'a 'b R 'a 'b S Message passing Shared state 25
39. swap upd f R S <+> 'a 'b Message passing Shared state 25
40. swap upd f R S + Message passing Shared state Disjunction 26
41. swap upd f R S + 'a 'b R 'a 'c S Message passing Shared state Disjunction 26
42. swap upd f R S + R S * 'a ('b * 'c) Message passing Shared state Disjunction 26
43. swap upd f R S + R S * Message passing Shared state Disjunction Conjunction 27
44. module type TREIBER_STACK = sig type 'a t val create : unit -> 'a t val push : 'a t -> ('a, unit) Reagent.t val pop : 'a t -> (unit, 'a) Reagent.t val try_pop : 'a t -> (unit, 'a option) Reagent.t end module Treiber_stack : TREIBER_STACK = struct type 'a t = 'a list Ref.ref let create () = Ref.mk_ref [] let push r x = Ref.upd r (fun xs x -> Some (x::xs,())) let try_pop r = Ref.upd r (fun l () -> match l with | [] -> Some ([], None) | x::xs -> Some (xs, Some x)) let pop r = Ref.upd r (fun l () -> match l with | [] -> None | x::xs -> Some (xs,x)) end 28
45. Composability Treiber_stack.pop s1 >>> Treiber_stack.push s2 Transfer elements atomically 29
46. Composability Treiber_stack.pop s1 >>> Treiber_stack.push s2 Transfer elements atomically Consume elements atomically Treiber_stack.pop s1 <*> Treiber_stack.pop s2 29
47. Composability Treiber_stack.pop s1 >>> Treiber_stack.push s2 Transfer elements atomically Consume elements atomically Treiber_stack.pop s1 <*> Treiber_stack.pop s2 Consume elements from either Treiber_stack.pop s1 <+> Treiber_stack.pop s2 29
48. type fork = {drop : (unit,unit) endpoint; take : (unit,unit) endpoint} let mk_fork () = let drop, take = mk_chan () in {drop; take} let drop f = swap f.drop let take f = swap f.take let init forks = List.iter (fun fork -> Thread.spawn @@ run (drop fork)) forks let eat l_fork r_fork = run (take l_fork <*> take r_fork) (); (* ... * eat * ... *) run (drop l_fork) (); run (drop r_fork) () 30
49. Implementation 31
50. Phase 1 Phase 2 32
51. Phase 1 Phase 2 Accumulate CASes 32
52. Phase 1 Phase 2 Accumulate CASes Attempt k-CAS 32
53. Accumulate CASes Attempt k-CAS 33
54. Accumulate CASes Attempt k-CAS Permanent failure 33
55. Accumulate CASes Attempt k-CAS Permanent failure Transient failure 33
57. Permanent failure 34
58. Permanent failure Transient failure 34
59. Permanent failure Transient failure Transient failure 34
60. Permanent failure Transient failure ? failure Transient failure 34
61. Permanent failure Transient failure ? failure Transient failure P & P = P T & T = T P & T = T T & P = T 34
62. Trouble with k-CAS 35
63. Trouble with k-CAS • Most processors do not support k-CAS 35
64. Trouble with k-CAS • Most processors do not support k-CAS • Implemented as a multi-phase protocol 1. Sort refs 2. Lock refs in order (CAS); rollback if conflicts. 3. Commit refs 35
65. Trouble with k-CAS • Most processors do not support k-CAS • Implemented as a multi-phase protocol 1. Sort refs 2. Lock refs in order (CAS); rollback if conflicts. 3. Commit refs • Additional book-keeping required • CAS list, messages to be consumed, post-commit actions, etc. 35
66. Trouble with k-CAS • Most processors do not support k-CAS • Implemented as a multi-phase protocol 1. Sort refs 2. Lock refs in order (CAS); rollback if conflicts. 3. Commit refs • Additional book-keeping required • CAS list, messages to be consumed, post-commit actions, etc. • Common case is just a single CAS • Identify and optimise with Arrows 35
67. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type 36
68. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure 36
69. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure transient failure 36
70. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure transient failure CAS set 36
71. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure transient failure CAS set Message + thread parking 36
72. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure transient failure CAS set Message + thread parking No CASes 36
73. type 'a result = Block | Retry | Done of 'a type ('a,'b) t = { try_react : 'a -> Reaction.t -> 'b Offer.t option -> 'b result; compose : 'r. ('b,'r) t -> ('a,'r) t; always_commits : bool; may_sync : bool } Reagent type permanent failure transient failure CAS set Message + thread parking No CASes No channel communication 36
74. let rec never : 'a 'b. ('a,'b) t = { try_react = (fun _ _ _ -> Block); may_sync = false; always_commits = false; compose = fun _ -> never } 37
75. let rec never : 'a 'b. ('a,'b) t = { try_react = (fun _ _ _ -> Block); may_sync = false; always_commits = false; compose = fun _ -> never } let rec constant : 'a 'b 'r. 'a -> ('a,'r) t -> ('b, 'r) t = fun x k (* continuation *) -> { may_sync = k.may_sync; always_commits = k.always_commits; try_react = (fun _ rx o -> k.try_react x rx o); compose = (fun next -> constant x (k.compose next)) } 37
76. let rec never : 'a 'b. ('a,'b) t = { try_react = (fun _ _ _ -> Block); may_sync = false; always_commits = false; compose = fun _ -> never } let rec constant : 'a 'b 'r. 'a -> ('a,'r) t -> ('b, 'r) t = fun x k (* continuation *) -> { may_sync = k.may_sync; always_commits = k.always_commits; try_react = (fun _ rx o -> k.try_react x rx o); compose = (fun next -> constant x (k.compose next)) } let rec <+> : 'a 'b 'r. ('a,'b) t -> ('a,'b) t -> ('a,'b) t = fun r1 r2 -> { always_commits = r1.always_commits && r1.always_commits; may_sync = r1.may_sync || r2.may_sync; ... 37
77. let rec cas r ~expect ~update k = let try_react () rx o = if Reaction.has_no_cas rx && k.always_commits then if CAS.cas r.data expect update then ( k.try_react () rx o ) (* Will succeed! *) else Retry else (* slow path with bookkeeping *) in ... Specialising k-CAS 38 rx cas k reagent
78. Optimising Transient Failures let rec without_offer pause r v = match r.try_react v Reaction.empty None with | Done res -> res | Retry -> ( pause (); if r.may_sync then with_offer pause r v else without_offer pause r v) | Block -> with_offer pause r v let run r v = let b = Backoff.create () in let pause () = Backoff.once b in without_offer pause r v 39