1"use strict";(self.webpackChunkarrow_website=self.webpackChunkarrow_website||[]).push([["4443"],{72859(e,n,i){i.r(n),i.d(n,{metadata:()=>r,default:()=>h,frontMatter:()=>o,contentTitle:()=>c,toc:()=>l,assets:()=>a});var r=JSON.parse('{"id":"learn/coroutines/racing","title":"Racing","description":"The parallelism operators describe the cases in which we are interested in the result","source":"@site/content/docs/learn/coroutines/racing.md","sourceDirName":"learn/coroutines","slug":"/learn/coroutines/racing","permalink":"/learn/coroutines/racing","draft":false,"unlisted":false,"tags":[],"version":"current","sidebarPosition":2,"frontMatter":{"sidebar_position":2},"sidebar":"learnSidebar","previous":{"title":"Parallelism","permalink":"/learn/coroutines/parallel"},"next":{"title":"Resource","permalink":"/learn/coroutines/resource-safety"}}'),t=i(74848),s=i(28453);let o={sidebar_position:2},c="Racing",a={},l=[{value:"Simple racing",id:"simple-racing",level:2},{value:"Using <code>select</code>",id:"using-select",level:2},{value:"Racing DSL (experimental)",id:"racing-dsl-experimental",level:2},{value:"Timeout",id:"timeout",level:3},{value:"Allow exceptions to win",id:"allow-exceptions-to-win",level:3},{value:"Race until a certain condition",id:"race-until-a-certain-condition",level:3},{value:"Custom exception handling",id:"custom-exception-handling",level:3},{value:"Integration with typed errors",id:"integration-with-typed-errors",level:3}];function d(e){let n={a:"a",admonition:"admonition",code:"code",em:"em",h1:"h1",h2:"h2",h3:"h3",header:"header",li:"li",p:"p",pre:"pre",strong:"strong",ul:"ul",...(0,s.R)(),...e.components};return(0,t.jsxs)(t.Fragment,{children:[(0,t.jsx)(n.header,{children:(0,t.jsx)(n.h1,{id:"racing",children:"Racing"})}),"\n",(0,t.jsxs)(n.p,{children:["The ",(0,t.jsx)(n.a,{href:"/learn/coroutines/parallel",children:"parallelism operators"})," describe the cases in which we are interested in the result\nof ",(0,t.jsx)(n.em,{children:"every"})," computation we perform. But imagine the scenario where we want to\ndownload a file, but we try two servers simultaneously for resilience purposes.\nOnce we get the file from one server, we're not really interested in the\nrest. This is an example of ",(0,t.jsx)(n.strong,{children:"racing"})," two computations."]}),"\n",(0,t.jsxs)(n.p,{children:["This is an key point in racing: we only care about the ",(0,t.jsx)(n.strong,{children:"first value"})," to ",(0,t.jsx)(n.strong,{children:"succeed"}),".\nBut that is too abstract, the desired behavior should match the following:"]}),"\n",(0,t.jsxs)(n.ul,{children:["\n",(0,t.jsxs)(n.li,{children:["The first value produced by a racer ",(0,t.jsx)(n.em,{children:"wins"})," the race."]}),"\n",(0,t.jsxs)(n.li,{children:["All exceptions need to be logged but not ",(0,t.jsx)(n.em,{children:"win"})," the race, they need to ",(0,t.jsx)(n.em,{children:"await"})," the race."]}),"\n",(0,t.jsx)(n.li,{children:"When the race is finished, all racers need to be canceled before returning the winning value. This guarantees all\nacquired resources are closed whilst trying to return the success value as soon as possible."}),"\n"]}),"\n",(0,t.jsx)(n.h2,{id:"simple-racing",children:"Simple racing"}),"\n",(0,t.jsx)(n.p,{children:"For the simpler cases,\nArrow provides functions that perform racing over 2 or 3 computations, with the\noption of customizing the coroutine context."}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:"suspend fun file(server1: String, server2: String) =\n raceN(\n { downloadFrom(server1) },\n { downloadFrom(server2) }\n ).merge()\n"})}),"\n",(0,t.jsxs)(n.p,{children:["The example above shows a typical pattern combined with ",(0,t.jsx)(n.code,{children:"raceN"}),".\nThe result of the function above is ",(0,t.jsx)(n.code,{children:"Either<A, B>"}),", with each type\ncorresponding to one branch in ",(0,t.jsx)(n.code,{children:"raceN"}),'. Since we have two computations that\nreturn the same type here and don\'t care which one "wins," we conflate both into\na single value.']}),"\n",(0,t.jsxs)(n.h2,{id:"using-select",children:["Using ",(0,t.jsx)(n.code,{children:"select"})]}),"\n",(0,t.jsxs)(n.p,{children:["The ",(0,t.jsx)(n.a,{href:"https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/",children:'"coroutines standard library"'}),"\noffers ",(0,t.jsx)(n.a,{href:"https://kotlinlang.org/docs/select-expression.html",children:"select expressions"}),", but often a higher level DSL is desired.\nFor example, when ",(0,t.jsx)(n.em,{children:"racing"})," for the ",(0,t.jsx)(n.em,{children:"first success value"})," and cancelling the others. Or, when you want to easily combine\nit with typed errors."]}),"\n",(0,t.jsxs)(n.p,{children:["Let's first see how we'd write this example with ",(0,t.jsx)(n.code,{children:"select"}
1),", and how we can rewrite it with Arrow Fx Coroutines.\nYou don't need to understand the whole code to keep reading, but it provides a good explanation of the difficulties of racing."]}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:'object RemoteCache {\n suspend fun getUser(id: UserId): User =\n if (Random.nextBoolean()) User("$id-remote-user") else throw BadRequestException()\n}\n\nobject LocalCache {\n suspend fun getUser(id: UserId): User =\n if (Random.nextBoolean()) User("$id-local-user") else throw NullPointerException()\n}\n\nsuspend fun <A> awaitAfterError(block: suspend () -> A): A = try {\n block()\n} catch (e: Throwable) {\n if (e is CancellationException || NonFatal(e)) throw e\n e.printStackTrace()\n awaitCancellation()\n}\n\nsuspend fun getRemoteUser(id: UserId): User = coroutineScope {\n try {\n select {\n async { awaitAfterError { RemoteCache.getUser(id) } }.onAwait { it }\n async { awaitAfterError { LocalCache.getUser(id) } }.onAwait { it }\n }\n } finally {\n coroutineContext.job.cancelChildren()\n }\n}\n'})}),"\n",(0,t.jsx)(n.p,{children:"The snippet above handles four important edge cases:"}),"\n",(0,t.jsxs)(n.ul,{children:["\n",(0,t.jsxs)(n.li,{children:[(0,t.jsx)(n.code,{children:"coroutineScope { }"})," ensures that all active coroutines are completed before returning."]}),"\n",(0,t.jsxs)(n.li,{children:[(0,t.jsx)(n.code,{children:"try/finally"}),": When ",(0,t.jsx)(n.code,{children:"select"})," returns we need to always cancel all still active running coroutines instead of waiting\nfor them to be finished. This is crucial otherwise we would still await ",(0,t.jsx)(n.em,{children:"all"})," values, and that would defeat the\npurpose of racing."]}),"\n",(0,t.jsxs)(n.li,{children:[(0,t.jsx)(n.code,{children:"awaitAfterError"}),": When an error occurs in a participant during racing, they are often considered ",(0,t.jsx)(n.em,{children:"losers"}),", and must\nawait the race to be finished. Since the ",(0,t.jsx)(n.em,{children:"winner"})," cancels the race, this can be achieved by ",(0,t.jsx)(n.code,{children:"awaitCancellation"}),".","\n",(0,t.jsxs)(n.ul,{children:["\n",(0,t.jsxs)(n.li,{children:[(0,t.jsx)(n.code,{children:"e.prinStacktrace()"}),": When an error occurs in a participant during racing, they are often considered ",(0,t.jsx)(n.em,{children:"losers"}),",\nhowever, its typically not desired this information disappears so some form of error handling strategy is needed."]}),"\n",(0,t.jsxs)(n.li,{children:[(0,t.jsx)(n.code,{children:"e is CancellationException || e.isFatal()"}),": We should never recover from ",(0,t.jsx)(n.code,{children:"CancellationException"}),", nor should we\nrecover from fatal exceptions like ",(0,t.jsx)(n.code,{children:"OutOfMemoryException"}),"."]}),"\n"]}),"\n"]}),"\n"]}),"\n",(0,t.jsx)(n.h2,{id:"racing-dsl-experimental",children:"Racing DSL (experimental)"}),"\n",(0,t.jsx)(n.admonition,{type:"warning",children:(0,t.jsx)(n.p,{children:"The functionality described in this section is experimental.\nAlthough the basic concepts shall remain, we may tweak the API in the future."})}),"\n",(0,t.jsxs)(n.p,{children:["The process of setting up this racing logic is a bit low-level, and complex. So unless for specific needs, we often\nprefer a higher level DSL. Arrow offers a simpler ",(0,t.jsx)(n.code,{children:"racing"})," DSL on top of ",(0,t.jsx)(n.code,{children:"select"})," that guarantees these semantics."]}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:"suspend fun getUserRacing(id: UserId): User = racing {\n race { RemoteCache.getUser(id) }\n race { LocalCache.getUser(id) }\n}\n"})}),"\n",(0,t.jsx)(n.p,{children:"This snippet is significantly simpler, and less error-prone. Initially, it seems we've lost a lot of flexibility, but\nlet's see what else the Arrow DSL has to offer."}),"\n",(0,t.jsx)(n.admonition,{type:"info",children:(0,t.jsxs)(n.p,{children:["If all racers thrown an exception then the ",(0,t.jsx)(n.code,{children:"racing"})," block will hang ",(0,t.jsx)(n.strong,{children:"forever"})," waiting for a success value."]})}),"\n",(0,t.jsx)(n.h3,{id:"timeout",children:"Timeout"}),"\n",(0,t.jsx)(n.p,{children:"Hanging is common with racing if everything fails. Therefore, it's common to include a timeout."}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:"suspend fun getUserRacing(id: UserId): User = racing {\n race { RemoteCache.getUser(id) }\n race { LocalCache.getUser(id) }\n race {\n delay(10.milliseconds)\n throw TimeoutException()\n }\n}\n"})}),"\n",(0,t.jsxs)(n.p,{children:["Here we simply use a ",(0,t.jsx)(n.code,{children:"delay"})," inside of the ",(0,t.jsx)(n.code,{children:"race { }"})," block, and return a default value after the ",(0,t.jsx)(n.code,{children:"delay"})," (timeout)\nfinishes. In this case we throw ",(0,t.jsx)(n.code,{children:"UserRaceException"})," but we could've also returned ",(0,t.jsx)(n.code,{children:"null"}),", ",(0,t.jsx)(n.code,{children:"Either.Right"}),", or ",(0,t.jsx)(n.code,{children:"raise"}),"."]}),"\n",(0,t.jsx)(n.h3,{id:"allow-exceptions-to-win",children:"Allow exceptions to win"}),"\n",(0,t.jsxs)(n.p,{children:["Sometimes ignoring all exceptions is not desired, and certain participants are allowed to have their exception to win\nthe race. For example, in case when ",(0,t.jsx)(n.code,{children:"LocalCache"})," returns fast in the success cache hit occurs, but slow with a cache\nmiss.",(0,t.jsx)(n.code,{children:"RemoteCache"})," could finish the race faster with ",(0,t.jsx)(n.code,{children:"BadRequest"})," or ",(0,t.jsx)(n.code,{children:"Unauthorized"}),". Beware that ",(0,t.jsx)(n.code,{children:"HttpClient"}
1)," might\nstill fail before a successful cache hit occurs like ",(0,t.jsx)(n.code,{children:"ConnectException"}),"."]}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:"suspend fun getUserRacing(id: UserId): User = racing {\n raceOrFail { RemoteCache.getUser(id) }\n race { LocalCache.getUser(id) }\n}\n"})}),"\n",(0,t.jsx)(n.h3,{id:"race-until-a-certain-condition",children:"Race until a certain condition"}),"\n",(0,t.jsxs)(n.p,{children:["Sometimes we need to race not only until a value is produced, but also that it needs to meet certain conditions.\nFor example, when we need to retrieve a ",(0,t.jsx)(n.code,{children:"NonEmptyList<UserId>"}),"."]}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:"suspend fun LocalCache.getCachedUsers(ids: NonEmptyList<UserId>): List<User> =\n ids.mapNotNull { id -> getUserOrNull(id) }\n\nsuspend fun getUserRacing(ids: NonEmptyList<UserId>): List<User> = racing {\n race { RemoteCache.getUsers(ids) }\n race(condition = { it.size == ids.size }) { LocalCache.getCachedUsers(ids) }\n}\n"})}),"\n",(0,t.jsxs)(n.p,{children:["Since ",(0,t.jsx)(n.code,{children:"getCachedUsers"})," ignores missing cached values we need to make sure the result from our cache matches the amount\nrequested ids. We can easily do so by adding a ",(0,t.jsx)(n.code,{children:"condition"})," check in our ",(0,t.jsx)(n.code,{children:"race"})," to verify ",(0,t.jsx)(n.code,{children:"it.size == ids.size"}),"."]}),"\n",(0,t.jsx)(n.admonition,{type:"info",children:(0,t.jsxs)(n.p,{children:["If all racers fail, or do not meet the ",(0,t.jsx)(n.code,{children:"condition"})," then the ",(0,t.jsx)(n.code,{children:"racing"})," block will hang ",(0,t.jsx)(n.strong,{children:"forever"})," waiting for a success\nvalue."]})}),"\n",(0,t.jsx)(n.h3,{id:"custom-exception-handling",children:"Custom exception handling"}),"\n",(0,t.jsxs)(n.p,{children:["By default, the strategy for unhandled exceptions is ",(0,t.jsx)(n.code,{children:"Throwable::printStackTrace"}),", but it attempts to use any installed\n",(0,t.jsx)(n.code,{children:"CoroutineExceptionHandler"}),". So in the case of Ktor it will be using the Ktor's default",(0,t.jsx)(n.code,{children:"CoroutineExceptionHandler"}),".\nIt can easily be overridden by explicitly using ",(0,t.jsx)(n.code,{children:"withContext"})," to install a ",(0,t.jsx)(n.code,{children:"CoroutineExceptionHandler"}),"."]}),"\n",(0,t.jsx)(n.pre,{children:(0,t.jsx)(n.code,{className:"language-kotlin",children:'suspend fun customErrorHandling(): String =\n withContext(CoroutineExceptionHandler { ctx, t -> t.printStackTrace() }) {\n racing {\n race {\n delay(2.seconds)\n throw RuntimeException("boom!")\n }\n race {\n delay(10.seconds)\n "Winner!"\n }\n }\n }\n'})}),"\n",(0,t.jsx)(n.h3,{id:"integration-with-typed-errors",children:"Integration with typed errors"}),"\n",(0,t.jsxs)(n.p,{children:["For the purposes of racing, using ",(0,t.jsx)(n.code,{children:"raise"})," has the same effects as throwing\nan exception. That means that if you want the error to propagate you need\nto use ",(0,t.jsx)(n.code,{children:"raceOrFail"}),"; otherwise that computation does not count as successful."]})]})}function h(e={}){let{wrapper:n}={...(0,s.R)(),...e.components};return n?(0,t.jsx)(n,{...e,children:(0,t.jsx)(d,{...e})}):d(e)}},28453(e,n,i){i.d(n,{R:()=>o,x:()=>c});var r=i(96540);let t={},s=r.createContext(t);function o(e){let n=r.useContext(s);return r.useMemo(function(){return"function"==typeof e?e(n):{...n,...e}},[n,e])}function c(e){let n;return n=e.disableParentContext?"function"==typeof e.components?e.components(t):e.components||t:o(e.components),r.createElement(s.Provider,{value:n},e.children)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.