Concurrent programming with F#

Enabling developers to write concise code in solving complex problems is one of the significant characteristics of functional programming. The conciseness is mostly due to the abstractions provided by the functional programming language. Can we apply these abstractions and write concurrent programs with ease? In this talk, we are going to find the answer to this question by solving tricky concurrent problems in F# using the Hopac library.

Transcript

1. Concurrent Programming with tamizhvendan tamizhvendan Lead Consultant www.ajira.tech Tamizhvendan S Passionate, Pragmatic and Polyglot Programmer https://www.demystifyfp.com
2. A GitHub Profile API
3. The JSON response
4. GitHub APIs https:!//api.github.com/users/tamizhvendan https:!//api.github.com/users/tamizhvendan/repos https:!//api.github.com/repos/tamizhvendan/blog-samples/languages https:!//api.github.com/repos/tamizhvendan/CafeApp/languages https:!//api.github.com/repos/tamizhvendan/fsharp-phonecat/languages
5. The Sequential World time User API call Repo API call Lang API call Lang API call Lang API call
6. The “C” World time User API call Repo API call Lang API call Lang API call Lang API call
7. The Difference time User API call Repo API call Lang API call Lang API call Lang API call time User API call Repo API call Lang API call Lang API call Lang API call V/S
8. Concurrent Programming Concurrency provides a way to structure a solution to solve a problem that may (but not necessarily) be parallelizable. Concurrency is the composition of independently executing computations. Credits: Concurrency is not parallelism Talk by Rob Pike User API call Repo API call Lang API call Lang API call Lang API call User API call Repo API call Lang API call Lang API call Lang API call
9. Sounds like Functional Programming Concurrency provides a way to structure a solution to solve a problem that may (but not necessarily) be parallelizable. Concurrency is the composition of independently executing computations.
10. val getUser: UserName !-> User val getUserRepos: UserName !-> Repo [] val getTopThreeRepos: Repo []!-> Repo [] val getUserRepoLanguages: UserName !-> RepoName !-> Language[] User API call https:!//api.github.com/users/tamizhvendan Repo API call https:!//api.github.com/users/tamizhvendan/repos Lang API call https:!//api.github.com/repos/tamizhvendan/CafeApp/languages Functions!
11. The Sequential World time User API call Repo API call Lang API call Lang API call Lang API call val getGitHubProfile: UserName !-> Profile let getGitHubProfile userName = let user = getUser userName let userRepos = getUserRepos userName let topThreeUserRepos = getTopThreeRepos userRepos let topThreeUserReposLanguages = .... getUserRepoLanguages .... .... .... profile
12. val getUser: UserName !-> C<User> val getUserRepos: UserName !-> C<Repo []> val getTopThreeRepos: Repo []!-> Repo [] val getUserRepoLanguages: UserName !-> RepoName !-> C<Language[]> User API call https:!//api.github.com/users/tamizhvendan Repo API call https:!//api.github.com/users/tamizhvendan/repos Lang API call https:!//api.github.com/repos/tamizhvendan/CafeApp/languages Functions in “C” World!
13. The “C” World val getGitHubProfile: UserName !-> C<Profile> time User API call Repo API call Lang API call Lang API call Lang API call let getGitHubProfile userName = c { let user, userRepos = getUser userName !!<*> getUserRepos userName let topThreeUserRepos = getTopThreeRepos userRepos let topThreeUserReposLanguages = topThreeUserRepos !|> getUserRepoLanguages .... !|> C.conCollect .... .... profile }
14. What is the “C” World? A Programming Model that enables concurrency C<'T> Async<'T> Task<'T> Job<'T> .NET Framework F# Core Hopac Library http:!//tomasp.net/blog/csharp-fsharp-async-intro.aspx/ https:!//neoeinstein.github.io/blog/2016/04-08-hopac-getting-started-with-jobs/index.html
15. F# Library Programming Model inspired by Concurrent ML Job<‘a> - an individual unit of execution (aka lightweight Thread) Ch<‘a> - channels as communicate mechanism Alt<‘a> - coordinates and communicate between jobs Similar to Clojure’s core.async & Golang concurrency model Hopac
16. Job<‘a> !// Job<unit> let longerHelloWorldJob = job { do! timeOutMillis 2000 printfn "Hello, World!" } #time "on" run longerHelloWorldJob #time "off" Hello, World! Real: 00:00:02.003, CPU: 00:00:00.006, GC gen0: 0, gen1: 0
18. Job<‘a> - a real world example type Product = { Id : int Name : string } Product Service type Review = { ProductId : int Author : string Comment : string } Review Service type ProductWithReviews = { Id : int Name : string Reviews : (string * string) list }
19. !// int !-> Job<Product> let getProduct id = job { !// Delay in the place of an external HTTP API call do! timeOutMillis 2000 return {Id = id; Name = "My Awesome Product"} }
20. let getProductReviews id = job { !// Delay in the place of an external HTTP API call do! timeOutMillis 3000 return [ {ProductId = id; Author = "John"; Comment = "It's awesome!"} {ProductId = id; Author = "Sam"; Comment = "Great product"} ] }
21. The Sequential Approach !// int !-> Job<ProductWithReviews> let getProductWithReviews id = job { let! product = getProduct id let! reviews = getProductReviews id let productReviews = reviews !|> List.map (fun r !-> r.Author,r.Comment) return { Id = id Name = product.Name Reviews = productReviews } }
22. The Sequential Approach #time "on" getProductWithReviews 1 !|> run #time "off" Real: 00:00:05.008, CPU: 00:00:00.009, GC gen0: 0, gen1: 0 val it : ProductWithReviews = {Id = 1; Name = "My Awesome Product"; Reviews = [("John", "It's awesome!"); ("Sam", "Great product")];}
23. Introducing Concurrency !// int !-> Job<ProductWithReviews> let getProductWithReviews id = job { let! product, reviews = getProduct id !!<*> getProductReviews id let productReviews = reviews !|> List.map (fun r !-> r.Author,r.Comment) return { Id = id Name = product.Name Reviews = productReviews } } val ( !!<*> ): Job<'x> !-> Job<'y> !-> Job<'x * 'y>
24. Introducing Concurrency Real: 00:00:03.005, CPU: 00:00:00.008, GC gen0: 0, gen1: 0 val it : ProductWithReviews = {Id = 1; Name = "My Awesome Product"; Reviews = [("John", "It's awesome!"); ("Sam", "Great product")];} #time "on" getProductWithReviews 1 !|> run #time "off"
25. time User API call Repo API call Lang API call Lang API call Lang API call let getUserDto username = job { let! user, repos = getUser username !!<*> getTopThreeUserRepos username let! repoDtos = repos !|> Array.map (getRepoDto username) !|> Job.conCollect return { Name = user.Name AvatarUrl = user.AvatarUrl TopThreeRepos = repoDtos.ToArray() } } A GitHub Profile API val conCollect: seq<Job<'x!>> !-> Job<ResizeArray<'x!>>
26. Inter-Thread Communication if (Monitor.TryEnter(lockObject, 300)) { try { !// Place code protected by the Monitor here. } finally { Monitor.Exit(lockObject); } } else { !// Code to execute if the attempt times out. } lock(lockObject) { myField!++; }
27. Share Memory By Communicating
28. Channel & Alt Abstractions to communicate and coordinate between jobs Synchronous by default, Supports asynchronous too! Lightweight First-Class Selective
29. Job 1 Job 2 Job 3 Job 4 time starting job:2 starting job:1 starting job:4 starting job:3 completed job:1 completed job:2 completed job:3 completed job:4 Real: 00:00:04.002, CPU: 00:00:00.013, GC gen0: 0, gen1: 0 Tracking Concurrent Job Status Job.conIgnore [job1;job2;job3;job4]
30. Tracking Concurrent Job Status 4 Jobs 1 Print Job JobStatus type JobStatus = | Started of jobId : int | Completed of jobId : int Ch.give Ch.take Channel<JobStatus> Job x Print Job
31. Job 4 Print Job Job 1 Job 2 Print Job Job 4 Job 1 Print Job Job 3 Print Job Job 3 time Starting Status
32. Job 1 Print Job time Completion Status Job 2 Print Job Job 3 Print Job Job 4 Print Job
33. !// Ch<JobStatus> !-> int !-> int !-> Job<unit> let createJob jobStatusChannel jobId delay = job { do! Ch.give jobStatusChannel (Started jobId) do! timeOutMillis delay do! Ch.give jobStatusChannel (Completed jobId) } Ch.give Channel<JobStatus> Job -> Channel
34. !// Ch<JobStatus> !-> Job<unit> let jobStatusPrinterJob jobStatusChannel = job { let! jobStatus = Ch.take jobStatusChannel match jobStatus with | Started jobId !-> printfn "starting job:%d" jobId | Completed jobId !-> printfn "completed job:%d" jobId } Ch.take Channel<JobStatus> Channel -> Job
35. time !// unit !-> Job<unit> let main () = job { let jobStatusChannel = Ch<JobStatus>() let jobStatusPrinter = jobStatusPrinterJob jobStatusChannel let jobs = [ Job.foreverServer jobStatusPrinter createJob jobStatusChannel 1 1000 createJob jobStatusChannel 2 2000 createJob jobStatusChannel 3 3000 createJob jobStatusChannel 4 4000 ] return! Job.conIgnore jobs }
36. starting job:2 starting job:1 starting job:4 starting job:3 completed job:1 completed job:2 completed job:3 completed job:4 Real: 00:00:04.002, CPU: 00:00:00.013, GC gen0: 0, gen1: 0 #time "on" main () !|> run #time "off"
37. A Slow Printer let jobStatusPrinterJob jobStatusChannel = job { let! jobStatus = Ch.take jobStatusChannel do! timeOutMillis 1000 match jobStatus with | Started jobId !-> printfn "starting job:%d" jobId | Completed jobId !-> printfn "completed job:%d" jobId }
38. #time "on" main () !|> run #time "off" !!--> Timing now on starting job:4 starting job:3 starting job:2 starting job:1 completed job:4 completed job:2 completed job:3 Real: 00:00:07.004, CPU: 00:00:00.014, GC gen0: 0, gen1: 0 val it : unit = () !!--> Timing now off > completed job:1
39. let main () = job { .... .... let jobStatusPrinter2 = jobStatusPrinterJob jobStatusChannel let jobs = [ .... Job.foreverServer jobStatusPrinter2 …. ] return! Job.conIgnore jobs } Scale Up with one more printer!
40. !!--> Timing now on starting job:2 starting job:4 starting job:1 starting job:3 completed job:1 completed job:2 Real: 00:00:04.010, CPU: 00:00:00.018, GC gen0: 0, gen1: 0 val it : unit = () !!--> Timing now off > completed job:4 completed job:3 #time "on" main () !|> run #time "off" It’s faster again!
41. Other Modes BoundedMb<‘x> Ch.send
42. Alt Selective synchronous operation Choose … C1 C2 Cn Alt<‘x> is a subtype of Job<‘X> Ch<‘x> is a subtype of Alt<‘X>
43. Alt in action give me an answer after sometime! type Answer = | FourtyTwo !// int !-> Alt<Answer> let getAnswer delay = timeOutMillis delay !|> Alt.afterFun (fun _ !-> FourtyTwo) Real: 00:00:01.001, CPU: 00:00:00.001, GC gen0: 0, gen1: 0 val it : Answer = FourtyTwo #time "on" getAnswer 1000 !|> run #time "off"
44. Alt in action give me an answer or timeout! type Answer = | FourtyTwo | TimedOut !// int !-> int !-> Alt<Answer> let getAnswerOrTimeOut delay tolerance = let timeOutAlt = timeOutMillis tolerance !|> Alt.afterFun (fun _ !-> TimedOut) Alt.choose [ timeOutAlt getAnswer delay ]
45. Alt in action give me an answer or timeout! #time "on" getAnswerOrTimeOut 1000 2000 !|> run #time "off" Real: 00:00:01.000, CPU: 00:00:00.000, GC gen0: 0, gen1: 0 val it : Answer = FourtyTwo #time "on" getAnswerOrTimeOut 1000 500 !|> run #time "off" Real: 00:00:00.501, CPU: 00:00:00.000, GC gen0: 0, gen1: 0 val it : Answer = TimedOut
46. Fan-out & Fan-in Can we calculate SHA256 hash of all the files in a directory concurrently?
47. A Generic Worker f:’a!->’b id inCh<‘a> outCh<‘b>
48. A Generic Worker type WorkerResponse<'T> = { Result : 'T Id : int } !// int !-> ('a !-> 'b) !-> Ch<'a> !-> Ch<WorkerResponse<'b!>> let worker id f inCh = let outCh = Ch<WorkerResponse<'b!>>() let onInput x = let y = f x Ch.give outCh {Result = y; Id = id} Ch.take inCh !|> Alt.afterJob onInput !|> Job.foreverServer !|> start outCh
49. A Generic Input Feeder !// Ch<'a> !-> 'a list !-> Alt<unit> let rec feedInput inCh inputs = match inputs with | [] !-> Alt.always () | x !:: xs !-> Ch.give inCh x !|> Alt.afterJob (fun _ !-> feedInput inCh xs) inCh<‘a> ‘a list
50. SHA256 Computation The Biz Logic (f) type Sha256Result = { FileName : string Hash : string } !// string !-> Sha256Result let computeSHA256 filePath = use sha256 = new SHA256Managed() use stream = File.OpenRead filePath let hash = sha256.ComputeHash(stream) !|> Seq.map (fun b !-> b.ToString("x2")) !|> String.concat "" { FileName = FileInfo(filePath).Name Hash = hash}
51. Putting things Together !// int !-> string !-> Job<unit> let sha256OfDirFiles workerCount directoryPath = let inCh = Ch<string>() let filePaths = Directory.GetFiles(directoryPath) !|> Array.toList feedInput inCh filePaths !|> start inCh<string> string list let initializeWorker id = worker (id+1) computeSHA256 inCh let workerOutChannels = List.init workerCount initializeWorker SHA256 id inCh<string> outCh<WR<SR!>> SHA256 id inCh<string> outCh<WR<SR!>> … let printResponse r = printfn "Worker %d: %A" r.Id r.Result Alt.choose workerOutChannels !|> Alt.afterFun printResponse !|> Job.forN filePaths.Length Print outCh<WR<SR!>>
52. Stress Testing (532 files - 16.41 GB) sha256OfDirFiles 1 “my_downloads_dir” !|> run Real: 00:06:11.336, CPU: 00:05:19.298, GC gen0: 5, gen1: 0 sha256OfDirFiles 2 “my_downloads_dir” !|> run Real: 00:04:07.560, CPU: 00:05:44.337, GC gen0: 5, gen1: 0 sha256OfDirFiles 3 “my_downloads_dir” !|> run Real: 00:04:11.717, CPU: 00:06:08.536, GC gen0: 4, gen1: 0 sha256OfDirFiles 4 “my_downloads_dir” !|> run Real: 00:03:27.888, CPU: 00:07:36.922, GC gen0: 5, gen1: 0 sha256OfDirFiles 5 “my_downloads_dir” !|> run Real: 00:04:39.357, CPU: 00:06:53.018, GC gen0: 5, gen1: 0 sha256OfDirFiles 6 “my_downloads_dir” !|> run Real: 00:04:01.523, CPU: 00:06:47.747, GC gen0: 5, gen1: 0
53. Concurrency provides a way to structure a solution to solve a problem that may (but not necessarily) be parallelizable. Concurrency is the composition of independently executing computations. SHA256 SHA256 … Print
54. Summary Concurrency !-> Structuring Functional Programming ❤ Concurrent Programming Choose a Right tool for the Job!
55. Deep Dive https://www.demystifyfp.com/tags/hopac
56. References Hopac Documentation Go Concurrency Patterns Concurrency is not Parallelism - Rob Pike