reorganize stuff

src/RecSys.jl

@@ -51,12 +51,17 @@ macro logmsg(s)
 end
 
 
-include("chunk.jl")
-include("mmapsparse.jl")
-include("input.jl")
-include("dist_input.jl")
-include("als_model.jl")
-include("als_dist_model.jl")
+include("chunks/chunk.jl")
+include("chunks/csv.jl")
+include("chunks/mmapsparse.jl")
+include("chunks/mmapdense.jl")
+
+include("inputs/input.jl")
+include("inputs/dist_input.jl")
+
+include("models/als_model.jl")
+include("models/als_dist_model.jl")
+
 include("als-wr.jl")
 include("utils.jl")
 

src/als-wr.jl

@@ -31,12 +31,6 @@ function train(als::ALSWR, niters::Int, nfacts::Int64, lambda::Float64=0.065)
     nothing
 end
 
-function train(als::ALSWR{ParChunk,DistInputs,DistModel}, niters::Int, nfacts::Int64, model_dir::AbstractString, max_cache::Int=10, lambda::Float64=0.065)
-    als.model = prep(als.inp, nfacts, lambda, model_dir, max_cache)
-    fact_iters(als, niters)
-    nothing
-end
-
 fact_iters(als, niters) = fact_iters(als.par, get(als.model), als.inp, niters)
 
 ##
@@ -228,6 +222,34 @@ function fact_iters{TP<:ParShmem,TM<:Model,TI<:Inputs}(::TP, model::TM, inp::TI,
     nothing
 end
 
+function rmse{TP<:Union{ParShmem,ParChunk},TI<:Inputs}(als::ALSWR{TP}, inp::TI)
+    t1 = time()
+
+    model = get(als.model)
+    share!(model)
+    share!(inp)
+    ensure_loaded(inp)
+    ensure_loaded(model)
+
+    cumerr = @parallel (.+) for user in 1:nusers(inp)
+        Uvec = reshape(getU(model,user), 1, nfactors(model))
+        nzrows, nzvals = items_and_ratings(inp, user)
+        predicted = vec(vec_mul_p(model, Uvec))[nzrows]
+        [sum((predicted .- nzvals) .^ 2), length(predicted)]
+    end
+    localize!(model)
+    @logmsg("rmse time $(time()-t1)")
+    sqrt(cumerr[1]/cumerr[2])
+end
+
+##
+# Chunk based distributed memory parallelism
+function train(als::ALSWR{ParChunk,DistInputs,DistModel}, niters::Int, nfacts::Int64, model_dir::AbstractString, max_cache::Int=10, lambda::Float64=0.065)
+    als.model = prep(als.inp, nfacts, lambda, model_dir, max_cache)
+    fact_iters(als, niters)
+    nothing
+end
+
 function fact_iters{TP<:ParChunk,TM<:Model,TI<:Inputs}(::TP, model::TM, inp::TI, niters::Int64)
     t1 = time()
 
@@ -264,26 +286,6 @@ function fact_iters{TP<:ParChunk,TM<:Model,TI<:Inputs}(::TP, model::TM, inp::TI,
     nothing
 end
 
-function rmse{TP<:Union{ParShmem,ParChunk},TI<:Inputs}(als::ALSWR{TP}, inp::TI)
-    t1 = time()
-
-    model = get(als.model)
-    share!(model)
-    share!(inp)
-    ensure_loaded(inp)
-    ensure_loaded(model)
-
-    cumerr = @parallel (.+) for user in 1:nusers(inp)
-        Uvec = reshape(getU(model,user), 1, nfactors(model))
-        nzrows, nzvals = items_and_ratings(inp, user)
-        predicted = vec(vec_mul_p(model, Uvec))[nzrows]
-        [sum((predicted .- nzvals) .^ 2), length(predicted)]
-    end
-    localize!(model)
-    @logmsg("rmse time $(time()-t1)")
-    sqrt(cumerr[1]/cumerr[2])
-end
-
 
 ##
 # Thread parallelism

src/chunk.jl → src/chunks/chunk.jl

@@ -10,13 +10,6 @@ using LRUCache
 using Base.Mmap
 import Base.Mmap: sync!
 
-# D = dimension that is split
-# N = value of the other dimension, which is constant across all splits
-type MemMappedMatrix{T,D,N}
-    val::Matrix{T}
-end
-sync!(m::MemMappedMatrix) = sync!(m.val, Base.MS_SYNC | Base.MS_INVALIDATE)
-
 type Chunk{K,V}
     path::AbstractString
     offset::Int
@@ -37,50 +30,6 @@ function Chunk(path::AbstractString, keyrange, V)
     Chunk{K,V}(path, 0, size, keyrange, V, WeakRef())
 end
 
-function load{T,D,N}(::Type{MemMappedMatrix{T,D,N}}, chunk::Chunk)
-    @logmsg("loading memory mapped chunk $(chunk.path)")
-    ncells = div(chunk.size, sizeof(T))
-    M = Int(ncells/N)
-    dims = (D == 1) ? (M,N) : (N,M)
-    A = Mmap.mmap(chunk.path, Matrix{T}, dims, chunk.offset)
-    MemMappedMatrix{T,D,N}(A)
-end
-
-function load{T<:Vector{UInt8}}(::Type{T}, chunk::Chunk)
-    @logmsg("loading chunk $(chunk.path)")
-    open(chunk.path) do f
-        seek(f, chunk.offset)
-        databytes = Array(UInt8, chunk.size)
-        read!(f, databytes)
-        return databytes::T
-    end
-end
-
-function load{T<:Matrix}(::Type{T}, chunk::Chunk)
-    M = readcsv(load(Vector{UInt8}, chunk))
-    M::T
-end
-
-#=
-function load{T<:SparseMatrixCSC}(::Type{T}, chunk::Chunk)
-    A = load(Matrix{Float64}, chunk)
-    rows = convert(Vector{Int64},   A[:,1]);
-    cols = convert(Vector{Int64},   A[:,2]);
-    vals = convert(Vector{Float64}, A[:,3]);
-
-    # subtract keyrange to keep sparse matrix small
-    cols .-= (first(chunk.keyrange) - 1)
-    sparse(rows, cols, vals)::T
-end
-=#
-function load{T<:SparseMatrixCSC}(::Type{T}, chunk::Chunk)
-    @logmsg("loading memory mapped sparse $(chunk.path)")
-    open(chunk.path) do f
-        seek(f, chunk.offset)
-        return mmap_csc_load(f)::T
-    end
-end
-
 function data{K,V}(chunk::Chunk{K,V}, lrucache::LRU)
     if chunk.data.value == nothing
         data = load(V, chunk)
@@ -97,15 +46,6 @@ function sync!(chunk::Chunk)
     end
 end
 
-#=
-function finalize_chunk_data(chunk::Chunk, data)
-    #@logmsg("unloading chunk $(chunk.path)")
-    # TODO: need to sync! when it gets chucked out of the lrucache
-    #sync!(chunk)
-    chunk.data = nothing
-end
-=#
-
 type ChunkedFile{K,V}
     keyrangetype::Type{K}
     valtype::Type{V}
@@ -119,9 +59,9 @@ keyrange(cf::ChunkedFile) = first(cf.chunks[1].keyrange):last(cf.chunks[end].key
 sync!(cf::ChunkedFile) = empty!(cf.lrucache)
 
 function _unload(cf, k, v)
-    sync!(v)
     chunk = getchunk(cf, first(k))
-    @logmsg("unloading chunk $(chunk.path)")
+    sync!(chunk)
+    #@logmsg("unloading chunk $(chunk.path)")
     chunk.data.value = nothing
     nothing
 end
@@ -146,7 +86,7 @@ function writemeta(cf::ChunkedFile)
         chunks = cf.chunks
         idx = 1
         for chunk in chunks
-            @logmsg("writing chunk: $(chunk.keyrange)")
+            #@logmsg("writing chunk: $(chunk.keyrange)")
             println(meta, first(chunk.keyrange), ",", last(chunk.keyrange), ",", chunkpfx, ".", idx)
             idx += 1
         end

src/chunks/csv.jl

@@ -0,0 +1,27 @@
+function load{T<:Vector{UInt8}}(::Type{T}, chunk::Chunk)
+    #@logmsg("loading chunk $(chunk.path)")
+    open(chunk.path) do f
+        seek(f, chunk.offset)
+        databytes = Array(UInt8, chunk.size)
+        read!(f, databytes)
+        return databytes::T
+    end
+end
+
+function load{T<:Matrix}(::Type{T}, chunk::Chunk)
+    M = readcsv(load(Vector{UInt8}, chunk))
+    M::T
+end
+
+#=
+function load{T<:SparseMatrixCSC}(::Type{T}, chunk::Chunk)
+    A = load(Matrix{Float64}, chunk)
+    rows = convert(Vector{Int64},   A[:,1]);
+    cols = convert(Vector{Int64},   A[:,2]);
+    vals = convert(Vector{Float64}, A[:,3]);
+
+    # subtract keyrange to keep sparse matrix small
+    cols .-= (first(chunk.keyrange) - 1)
+    sparse(rows, cols, vals)::T
+end
+=#

src/chunks/mmapdense.jl

@@ -0,0 +1,15 @@
+# D = dimension that is split
+# N = value of the other dimension, which is constant across all splits
+type MemMappedMatrix{T,D,N}
+    val::Matrix{T}
+end
+sync!(m::MemMappedMatrix) = sync!(m.val, Base.MS_SYNC | Base.MS_INVALIDATE)
+
+function load{T,D,N}(::Type{MemMappedMatrix{T,D,N}}, chunk::Chunk)
+    #@logmsg("loading memory mapped chunk $(chunk.path)")
+    ncells = div(chunk.size, sizeof(T))
+    M = Int(ncells/N)
+    dims = (D == 1) ? (M,N) : (N,M)
+    A = Mmap.mmap(chunk.path, Matrix{T}, dims, chunk.offset)
+    MemMappedMatrix{T,D,N}(A)
+end

src/mmapsparse.jl → src/chunks/mmapsparse.jl

@@ -52,3 +52,11 @@ function sync!(spm::SparseMatrixCSC)
     Mmap.sync!(spm.nzval, Base.MS_SYNC | Base.MS_INVALIDATE)
     nothing
 end
+
+function load{T<:SparseMatrixCSC}(::Type{T}, chunk::Chunk)
+    #@logmsg("loading memory mapped sparse $(chunk.path)")
+    open(chunk.path) do f
+        seek(f, chunk.offset)
+        return mmap_csc_load(f)::T
+    end
+end