# Do I need @async?

**URL:** <https://discourse.julialang.org/t/do-i-need-async/3993>\
**Category:** General Usage\
**Tags:** macros, reactiveprogramming, parallel\
**Created:** [May 30, 2017, 6:04pm UTC](https://discourse.julialang.org/t/do-i-need-async/3993 "2017-05-30T18:04:32Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![pyrex41](https://avatars.discourse-cdn.com/v4/letter/p/4bbf92/32.png) [@pyrex41](https://discourse.julialang.org/u/pyrex41)\
**Post date:** [May 30, 2017, 6:04pm UTC](https://discourse.julialang.org/t/do-i-need-async/3993/1 "2017-05-30T18:04:32Z")

</div>

I am working on a some simple helper functions to make simple backtesting very easy in julia. I am extending the ReactiveBasics packge. Here is some code for an example using a simple moving average crossover:

```julia
using ReactiveBasics

# my helper macros / functions
macro window(sig, wind::Int)
    quote
        arr=[]
        len = ($wind)
        flatmap($sig) do s
            push!(arr,s)
            length(arr) > len ? shift!(arr) : false
            arr |> Signal
        end
    end
end

function smap(f,s1::Signal, s2::Signal)
    flatmap(s1) do v1
        v2 = s2.value
        f(v1,v2) |> Signal
    end
end

function smap(f,s1::Signal, s2::Signal, s3::Signal...)
    flatmap(s1) do v1
        v2 = s2.value
        v3 = map(value, s3)
        f(v1,v2, v3...) |> Signal
    end
end

macro myfoldp(f, v0, sig)
    quote
        arr = []
        flatmap($sig) do s
            push!(arr,s)
            foldl($f, $v0, arr) |> Signal
        end
    end
end

macro myfoldp(f, v0, sig, args...)
    quote
        arr = []
        flatmap($sig) do s
            push!(arr,s)
            func(a,b) = ($f)(a,b,($args)...)
            foldl(func, $v0, arr) |> Signal
        end
    end
end

macro collect(sig)
    quote
        arr = []
        flatmap($sig) do s
            push!(arr,s) |> Signal
        end
    end
end

# setup model with Signals
setup = quote
    # generate some random data with positive edge
    returns = .5 + randn(10000)
    returns = normalize(returns)
    
    #crate return and price signals
    func(a,b) = a*(1+b)
    ret = Signal(0.0)
    price = @myfoldp func 50 ret
    
    # 50 and 200 day moving average signals
    ma50 = map(mean, @window(price,50))
    ma200 = map(mean, @window(price, 200))
    
    # crossover logic
    trend = smap(ma50,ma200) do f,s
        f >= s ? 1 : -1
    end
    
    # store data in arrays
    collecs1 = map(x->(@collect x), [ret, price, ma50, ma200, trend])
end

# initial run to get first complete 200 day window
function train(x=200)
    for i=1:x
        push!(ret, returns[i])
    end
end

# run simulation for 400 days
function runsim(xi=201,xf=600)
    for i=xi:xf
        push!(ret,returns[i])
    end
end

# save results in csv
function store_results(filename)
    c1 = map(value,collecs1)
    ba = foldl(hcat, c1)
    cols = ["ret", "price", "ma50", "ma200", "trend"]
    aba = hcat(cols,transpose(ba))
    naba = permutedims(aba,[2,1])
    writecsv(filename, naba)
end

# quickly setup, train, run, and save results
function run_all(filename, x=200, y=600)
    eval(setup)
    train(x)
    runsim(x+1,y)
    store_results(filename)
end

# run it:
run_all("data.csv")

```

This all works as expected / desired. My question concerns this bit right here:

```julia
    ma50 = map(mean, @window(price,50))
    ma200 = map(mean, @window(price, 200))
    
    trend = smap(ma50,ma200) do f,s
        f >= s ? 1 : -1
    end

```

For the rest of the code, each map on a signal is chained sequentially. However, these two moving average signals are not chained together, but are both needed to be updated for the `trend` signal to give the correct value. the `smap` function updates on `ma50`, sampling the current value at that time of `ma200`. However, if the chain from `price` → `ma50` → `trend` were to somehow update more quickly than `price` → `ma200`, then the calculation in the logic of the `trend` signal would be sampling the wrong value. I am not sure if this is possible; if it is, I want to make sure that I can explicitly prevent this undesired behavior.

In this example case, it would be trivial to write a Signal function that calculated both the 50day and 200 day averages off of the same window; more generally, however, I want to be able to conveniently group calculations together that are needed simultaneously, and potentially run the associated functions in parallel. I thought this might do the trick:

```julia
@sync begin
    @async ma50 = map(mean, @window(price,50))
    @async ma200 = map(mean, @window(price, 200))
end
trend = smap(ma50,ma200) do f,s
    f >= s ? 1 : -1
end

```

However, when I tried this, it ended up that none of these Signals updated properly when updates were pushed to `ret`. What would be a good, syntactically clean way of ensuring that my Signals are correctly related to each other?

---

<div class="post-metadata">

**Author:** ![amit.murthy](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/amit.murthy/32/47_2.png) [@amit.murthy](https://discourse.julialang.org/u/amit.murthy)\
**Post date:** [June 2, 2017, 6:16am UTC](https://discourse.julialang.org/t/do-i-need-async/3993/2 "2017-06-02T06:16:10Z")

</div>

Do you want the two async tasks to interleave computation? If that is the case you will need the two tasks to explicitly yield to each other - something like `rand(Bool) && yield()` at an appropriate place ?

---

<div class="post-metadata">

**Author:** ![pyrex41](https://avatars.discourse-cdn.com/v4/letter/p/4bbf92/32.png) [@pyrex41](https://discourse.julialang.org/u/pyrex41)\
**Post date:** [June 2, 2017, 12:16pm UTC](https://discourse.julialang.org/t/do-i-need-async/3993/3 "2017-06-02T12:16:50Z")

</div>

I am not sure. I was able to write a macro to get the behavior I want, but it doesn’t utilize any of the task/scheduling features of julia at all, and seems like a rather inelegant solution. Perhaps there is a better approach?

```julia
macro bundle(f,g...)
  quote
    coll = map(eval,($g)) |> collect
    unshift!(coll, ($f))
    queues,sigs = [],[]
    for c in coll
      push!(queues,[])
      push!(sigs,Signal(c.value))
    end
    function eq()
      len = map(length,queues)
      if len[1]>0 && foldl(*, len / len[1]) == 1
        for (i,q) in enumerate(queues)
          v = pop!(q)
          push!(sigs[i], v)
        end
      end
    end
    for (i,gsig) in enumerate(coll)
      subscribe!(gsig) do s
        push!(queues[i], s)
        eq()
      end
    end
    flatmap(last(sigs)) do u
      out = map(value, sigs[1:end-1])
      push!(out, u)
      Signal(out)
    end
  end
end

```

Now, using this `@bundle` macro, we can create a signal that doesn’t update until all the bundled signals themselves update:

```julia
f = Signal(1)
g = Signal(-2)
h = @bundle f g # h.value = [1,-2]

push!(f, 5); h.value # h.value = [1,-2]
push!(g, -7); h.value # h.value = [5,-7]

```

Using @sync and @async when mapping the relationships between different signals doesn’t seem to work; I suspect that this is because instead of having asynchronous updates to Signals, it only asynchronously create the Signals. I would probably have to rewrite the `map` and `flatmap` functions to incorporate asynchronous structure. I don’t really care about the tasks running asynchronously; what I really want is to make sure that particular tasks happen only after (multiple) previous tasks, in an event-driven way. Is this a good use case for scheduled tasks, or am I barking up the wrong tree?