Big blockmerge rehaul started

Author: meggart

devscripts/Project.toml

@@ -2,6 +2,7 @@
 ArchGDAL = "c9ce4bd3-c3d5-55b8-8973-c0e20141b8c3"
 BenchmarkTools = "6e4b80f9-dd63-53aa-95a3-0cdb28fa8baf"
 CFTime = "179af706-886a-5703-950a-314cd64e0468"
+CairoMakie = "13f3f980-e62b-5c42-98c6-ff1f3baf88f0"
 Cthulhu = "f68482b8-f384-11e8-15f7-abe071a5a75f"
 Dagger = "d58978e5-989f-55fb-8d15-ea34adc7bf54"
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"
@@ -22,6 +23,7 @@ NetworkLayout = "46757867-2c16-5918-afeb-47bfcb05e46a"
 OnlineStats = "a15396b6-48d5-5d58-9928-6d29437db91e"
 PkgTemplates = "14b8a8f1-9102-5b29-a752-f990bacb7fe1"
 Primes = "27ebfcd6-29c5-5fa9-bf4b-fb8fc14df3ae"
+ProgressMeter = "92933f4c-e287-5a05-a399-4b506db050ca"
 Proj = "c94c279d-25a6-4763-9509-64d165bea63e"
 Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
 StatsBase = "2913bbd2-ae8a-5f71-8c99-4fb6c76f3a91"

devscripts/maintest.jl

@@ -2,24 +2,85 @@ using DiskArrayEngine
 import DiskArrayEngine as DAE
 using DiskArrays: ChunkType, RegularChunks
 using Statistics
-using Zarr, DiskArrays, OffsetArrays
-#using DiskArrayEngine: MWOp, internal_size, ProductArray, InputArray, getloopinds, UserOp, mysub, ArrayBuffer, NoFilter, AllMissing,
-#  create_buffers, read_range, generate_inbuffers, generate_outbuffers, get_bufferindices, offset_from_range, generate_outbuffer_collection, put_buffer, 
-#  Output, _view, Input, applyfilter, apply_function, LoopWindows, GMDWop, results_as_diskarrays, create_userfunction, steps_per_chunk, apparent_chunksize,
-#  find_adjust_candidates, generate_LoopRange, get_loopsplitter, split_loopranges_threads, merge_loopranges_threads, LocalRunner, 
-#  merge_outbuffer_collection, DistributedRunner
+using Zarr, DiskArrays
 using StatsBase: rle, mode
 using CFTime: timedecode
 using Dates
 using OnlineStats
 using Logging
 using Distributed
-#global_logger(SimpleLogger(stdout,Logging.Debug))
-#global_logger(SimpleLogger(stdout))
 using LoggingExtras
 using Test
 using DataStructures: OrderedSet
 
+
+
+
+inwindows1 = DAE.MovingWindow(1, 5, 5, 4)
+outwindows1 = 1:4
+inwindows2 = DAE.MovingWindow(1, 2, 2, 2)
+outwindows2 = 1:2
+
+inar = DAE.InputArray(1:20, windows=(inwindows1,))
+outspecs = (DAE.create_outwindows(4, windows=(outwindows1,)),)
+f = create_userfunction(sum, Float64)
+op1 = DAE.GMDWop((inar,), outspecs, f)
+r = results_as_diskarrays(op1)[1]
+
+inar2 = DAE.InputArray(r, windows=(inwindows2,))
+outspecs2 = (DAE.create_outwindows(4, windows=(outwindows2,)),)
+f2 = create_userfunction(sum, Float64)
+op2 = DAE.GMDWop((inar2,), outspecs2, f2)
+r2 = results_as_diskarrays(op2)[1]
+
+g = DAE.result_to_graph(r2)
+
+@test length(g.nodes) == 3
+@test g.nodes[1] == DAE.MwopOutNode(false, nothing, (2,), Float64)
+@test g.nodes[2] == DAE.MwopOutNode(false, nothing, (4,), Float64)
+@test g.nodes[3] == 1:20
+
+@test length(g.connections) == 2
+conn1, conn2 = g.connections
+@test conn1.inputids == [3]
+@test conn1.outputids == [2]
+@test conn2.inputids == [2]
+@test conn2.outputids == [1]
+
+nodemergestrategies = DAE.collect_strategies(g)
+
+@test only(nodemergestrategies[2]) isa DAE.BlockMerge
+@test nodemergestrategies[1] == [nothing]
+@test nodemergestrategies[3] == [nothing]
+
+dimmap = DAE.create_loopdimmap(conn1, conn2, 2)
+@test dimmap isa DAE.DimMap
+@test dimmap.d == Dict(1 => 1)
+
+newop = DAE.merge_operations(DAE.BlockMerge, conn1, conn2, 2, dimmap)
+@test newop isa DAE.UserOp
+@test newop.f isa DAE.BlockFunctionChain
+@test newop.f.funcs[1] === f.f
+@test newop.f.funcs[2] === f2.f
+@test newop.f.args == [((1,), (1,)), ((2,), (2,))]
+@test newop.f.transfers == [1 => [2]]
+
+newconn, newnodes = DAE.merged_connection(DAE.BlockMerge, g, conn1, conn2, 2, newop, nodemergestrategies, dimmap)
+
+@test newconn isa DAE.MwopConnection
+@test newconn.f === newop
+@test newconn.inputids == [3, 4]
+@test newconn.outputids == [2, 1]
+
+newconn.inwindows[1].windows.members[1]
+newconn.inwindows[2].windows.members[1]
+newconn.outwindows[1].windows.members[1]
+newconn.outwindows[2].windows.members[1]
+
+
+
+
+
 g = zopen("https://s3.bgc-jena.mpg.de:9000/esdl-esdc-v2.1.1/esdc-8d-0.25deg-184x90x90-2.1.1.zarr")
 
 
@@ -30,29 +91,121 @@ t = g["time"]
 tvec = timedecode(t[:], t.attrs["units"]);
 groups = yearmonth.(tvec)
 
-r = aggregate_diskarray(a, mean, (1 => nothing, 2 => 8, 3 => groups), strategy=:reduce)
+r = aggregate_diskarray(a, mean, (1 => nothing, 2 => 8, 3 => groups), strategy=:direct)
 
 #a = compute(r)
 
 r2 = aggregate_diskarray(r, maximum, (2 => nothing,))
 
+
+
 r3 = r .+ 273.15
 
 finalres = r2 .+ r3
 
-finalres[45, 100]
+finalres[1, 45, 100]
 
 g = DAE.MwopGraph()
 outnode = DAE.to_graph!(g, r2);
+
 DAE.remove_aliases!(g)
+
 using CairoMakie, GraphMakie
-#p = graphplot(g,elabels=DAE.edgenames(g),ilabels=DAE.nodenames(g))
+p = graphplot(g, elabels=DAE.edgenames(g), ilabels=DAE.nodenames(g))
+
+#DAE.fuse_step_direct!(g)
 
-dg = DAE.DimensionGraph(g)
-dg.concomps
 
+nodemergestrategies = DAE.collect_strategies(g)
+i_eliminate = findfirst(nodemergestrategies) do strat
+  !isempty(strat) && !all(isnothing, strat)
+end
+### DAE.eliminate_node(g, i_eliminate, nodemergestrategies[i_eliminate], BlockMerge)
+nodegraph = g
+inconids = DAE.inconnections(nodegraph, i_eliminate)
+outconids = DAE.outconnections(nodegraph, i_eliminate)
+inconns = nodegraph.connections[inconids]
+outconns = nodegraph.connections[outconids]
+
+inconn = only(inconns)
+outconn = only(outconns)
+
+dimmap = DAE.create_loopdimmap(inconn, outconn, i_eliminate)
+
+newop = DAE.merge_operations(DAE.BlockMerge, inconn, outconn, i_eliminate, dimmap)
+
+newconn = DAE.merged_connection(DAE.BlockMerge, nodegraph, inconn, outconn, i_eliminate, newop, nodemergestrategies, dimmap)
+
+newconn.inputids
+newconn.outputids
+newconn.inwindows[2].windows.members[2]
+
+
+nodemergestrategies = DAE.collect_strategies(g)
+i_eliminate = findfirst(nodemergestrategies) do strat
+  !isempty(strat) && !all(isnothing, strat)
+end
+
+nodegraph = g;
+inconids = DAE.inconnections(nodegraph, i_eliminate)
+outconids = DAE.outconnections(nodegraph, i_eliminate)
+inconns = nodegraph.connections[inconids]
+outconns = nodegraph.connections[outconids]
+
+inconn = only(inconns)
+outconn = only(outconns)
+
+dimmap = DAE.create_loopdimmap(inconn, outconn, i_eliminate)
+
+chain1 = DAE.BlockFunctionChain(inconn)
+chain2 = DAE.BlockFunctionChain(outconn)
+
+to_eliminate = i_eliminate
+
+chain1.args
+chain2.args
+ifrom = findfirst(==(to_eliminate), inconn.outputids)
+ito = findall(==(to_eliminate), outconn.inputids)
+transfer = ifrom => ito
+
+newfunc = DAE.build_chain(chain1, chain2, dimmap, transfer)
+
+
+newop = DAE.merge_operations(DAE.BlockMerge, inconn, outconn, i_eliminate, dimmap)
+
+newconn = DAE.merged_connection(DAE.BlockMerge, nodegraph, inconn, outconn, i_eliminate, newop, nodemergestrategies, dimmap)
+
+deleteat!(nodegraph.connections, [inconids; outconids])
+push!(nodegraph.connections, newconn)
+
+nodegraph.connections
+
+conn = only(nodegraph.connections)
+op = conn.f
+inputs = InputArray.(g.nodes[conn.inputids], conn.inwindows)
+outspecs = map(g.nodes[conn.outputids], conn.outwindows) do outnode, outwindow
+  (; lw=outwindow, chunks=outnode.chunks, ismem=outnode.ismem)
+end
+
+
+function gmwop_from_conn(conn)
+  op = conn.f
+  inputs = InputArray.(g.nodes[conn.inputids], conn.inwindows)
+  outspecs = map(g.nodes[conn.outputids], conn.outwindows) do outnode, outwindow
+    (; lw=outwindow, chunks=outnode.chunks, ismem=outnode.ismem)
+  end
+  DAE.GMDWop(inputs, outspecs, op)
+end
+
+
+using Graphs: nv, outneighbors, inneighbors
+ioutnodes = (findall(n -> !isempty(inneighbors(g, n)) && isempty(outneighbors(g, n)), 1:nv(g)))
+lastop = findall(conn -> all(in(conn.outputids), ioutnodes), g.connections) |> only
+op = gmwop_from_conn(g.connections[lastop])
+rgraph = results_as_diskarrays(op)[1]
+
+runner = rgraph[1, 45, 100]
 
-DAE.fuse_step_direct!(g)
 
 
 
@@ -98,13 +251,7 @@ remaining_conn.outputids
 remaining_conn.outwindows[1]
 
 
-op = remaining_conn.f
-inputs = InputArray.(g.nodes[remaining_conn.inputids], remaining_conn.inwindows)
-outspecs = map(g.nodes[remaining_conn.outputids], remaining_conn.outwindows) do outnode, outwindow
-  (; lw=outwindow, chunks=outnode.chunks, ismem=outnode.ismem)
-end
 
-mergedop = DAE.GMDWop(inputs, outspecs, op)
 
 rnow = DAE.results_as_diskarrays(mergedop)[2]
 

src/buffers.jl

@@ -31,8 +31,6 @@ windowbuffersize(looprange, window) = maximum(c -> internal_size(inner_index(win
 function generate_inbuffers(inars, loopranges)
     map(inars) do ia
         et = eltype(ia.a)
-        #@show loopranges
-
         Array{et}(undef, getbufsize(ia, loopranges))
     end
 end

src/disjointranges.jl

@@ -25,7 +25,7 @@ windowmaximum(w) = maximum(maximum,w)
 last_contains_value(w::AbstractVector{<:Number},i) = findlast(<=(i),w)
 last_contains_value(w::AbstractRange,i) = searchsortedlast(w,i)
 function last_contains_value(w,i)
-    ii = findlast(r->maximum(r)<=i,w)
+    ii = findlast(r -> i in r, w)
     if ii === nothing
         length(w)+1
     else

src/enginearrays.jl

@@ -39,7 +39,7 @@ function collect_bcdims(A)
                     newsize
                 elseif newsize == 1
                     oldsize
-                elseif odlsize == 1
+                elseif oldsize == 1
                     newsize
                 else
                     error("Dimension length do not match $newsize and $oldsize")

src/executionplan.jl

@@ -198,6 +198,8 @@ function get_chunkspec(outspec,ot)
   si = map(m->last(last(m))-first(first(m))+1,outspec.lw.windows.members)
   if cs isa GridChunks
     cs = cs.chunks
+  elseif cs === nothing
+    cs = map(_->nothing,si)
   end
   cs = map(cs,si) do csnow,s
     if csnow === nothing

src/graphs.jl

@@ -22,6 +22,7 @@ describe(z::ZArray, _) = Zarr.zname(z)
 describe(z::Array{<:Any,0}, _) = z[]
 describe(z, i) = "Input $i"
 EmptyInput(n::MwopOutNode) = EmptyInput{n.eltype,length(n.size)}(n.size)
+Base.eltype(n::MwopOutNode) = n.eltype
 
 mutable struct MwopGraph <: AbstractGraph{Int}
     dims::UnitRange{Int}
@@ -257,7 +258,9 @@ function eliminate_node(nodegraph, i_eliminate, strategies, appliedstrat)
 
     newop = merge_operations(appliedstrat, inconn, outconn, i_eliminate, dimmap)
 
-    newconn = merged_connection(appliedstrat, nodegraph, inconn, outconn, i_eliminate, newop, strategies, dimmap)
+    newconn, newnodes = merged_connection(appliedstrat, nodegraph, inconn, outconn, i_eliminate, newop, strategies, dimmap)
+
+    append!(nodegraph.nodes, newnodes)
 
     deleteat!(nodegraph.connections, [inconids; outconids])
     push!(nodegraph.connections, newconn)
@@ -319,4 +322,10 @@ function fuse_graph!(g::MwopGraph)
     while length(g.connections) > 1
         fuse_step_block!(g::MwopGraph)
     end
+end
+
+function result_to_graph(res)
+    g = MwopGraph()
+    to_graph!(g, res)
+    g
 end