<commit_before>-- lua-npge, Nucleotide PanGenome explorer (Lua module)
-- Copyright (C) 2014-2015 <NAME>
-- See the LICENSE file for terms of use.

local Workers = {}

Workers.mapBlocks = function(workers, blockset)
    local mapItems = require 'npge.util.mapItems'
    local buckets = mapItems(workers, blockset:blocks())
    local BlockSet = require 'npge.model.BlockSet'
    local sequences = blockset:sequences()
    local blocksets = {}
    for i = 1, workers do
        local blocks = buckets[i]
        local bs = BlockSet(sequences, blocks)
        table.insert(blocksets, bs)
    end
    return blocksets
end

Workers.mapSequences = function(workers, blockset)
    local mapItems = require 'npge.util.mapItems'
    local buckets = mapItems(workers, blockset:sequences())
    local BlockSet = require 'npge.model.BlockSet'
    local blocksets = {}
    for i = 1, workers do
        local sequences = buckets[i]
        local bs = BlockSet(sequences, {})
        table.insert(blocksets, bs)
    end
    return blocksets
end

local workerCode = [[
local ref = %q
local alg = %q
local BlockSet = require 'npge.model.BlockSet'
local bs = BlockSet.fromRef(ref)
local loadstring = require 'npge.util.loadstring'
local algorithm = loadstring(alg)
bs = assert(algorithm(bs))
local increase_count = true
return BlockSet.toRef(bs, increase_count)
]]

-- Map-reduce for algorithms on blocks.
-- 1. Splits the blockset into buckets using function
--    Workers.mapBlocks or Workers.mapSequences (argument map)
-- 2. apply alg to each bucket in parallel. alg must be a
--    string of Lua code, which gets a bockset and returns
--    a blockset.
Workers.applyToBlockset = function(blockset, alg, map)
    local threads = require 'npge.util.threads'
    return threads(
    -- generator
    function(workers)
        local codes = {}
        local blocksets = map(workers, blockset)
        local BlockSet = require 'npge.model.BlockSet'
        for _, bs in ipairs(blocksets) do
            local ref = BlockSet.toRef(bs)
            table.insert(codes, workerCode:format(ref, alg))
        end
        return codes
    end,
    -- collector
    function(results)
        local blocksets = {}
        local BlockSet = require 'npge.model.BlockSet'
        for _, ref in ipairs(results) do
            local decrease_count = true
            local bs = BlockSet.fromRef(ref, decrease_count)
            table.insert(blocksets, bs)
        end
        local Merge = require 'npge.algo.Merge'
        local unpack = require 'npge.util.unpack'
        return Merge(unpack(blocksets))
    end)
end

Workers.GoodSubblocks = function(blockset)
    local code = [[<commit_msg>Workers: increase refcount when passing to thread<commit_after>-- lua-npge, Nucleotide PanGenome explorer (Lua module)
-- Copyright (C) 2014-2015 <NAME>
-- See the LICENSE file for terms of use.

local Workers = {}

Workers.mapBlocks = function(workers, blockset)
    local mapItems = require 'npge.util.mapItems'
    local buckets = mapItems(workers, blockset:blocks())
    local BlockSet = require 'npge.model.BlockSet'
    local sequences = blockset:sequences()
    local blocksets = {}
    for i = 1, workers do
        local blocks = buckets[i]
        local bs = BlockSet(sequences, blocks)
        table.insert(blocksets, bs)
    end
    return blocksets
end

Workers.mapSequences = function(workers, blockset)
    local mapItems = require 'npge.util.mapItems'
    local buckets = mapItems(workers, blockset:sequences())
    local BlockSet = require 'npge.model.BlockSet'
    local blocksets = {}
    for i = 1, workers do
        local sequences = buckets[i]
        local bs = BlockSet(sequences, {})
        table.insert(blocksets, bs)
    end
    return blocksets
end

local workerCode = [[
local ref = %q
local alg = %q
local BlockSet = require 'npge.model.BlockSet'
local decrease_count = true
local bs = BlockSet.fromRef(ref, decrease_count)
local loadstring = require 'npge.util.loadstring'
local algorithm = loadstring(alg)
bs = assert(algorithm(bs))
local increase_count = true
return BlockSet.toRef(bs, increase_count)
]]

-- Map-reduce for algorithms on blocks.
-- 1. Splits the blockset into buckets using function
--    Workers.mapBlocks or Workers.mapSequences (argument map)
-- 2. apply alg to each bucket in parallel. alg must be a
--    string of Lua code, which gets a bockset and returns
--    a blockset.
Workers.applyToBlockset = function(blockset, alg, map)
    local threads = require 'npge.util.threads'
    return threads(
    -- generator
    function(workers)
        local codes = {}
        local blocksets = map(workers, blockset)
        local BlockSet = require 'npge.model.BlockSet'
        for _, bs in ipairs(blocksets) do
            local increase_count = true
            local ref = BlockSet.toRef(bs, increase_count)
            table.insert(codes, workerCode:format(ref, alg))
        end
        return codes
    end,
    -- collector
    function(results)
        local blocksets = {}
        local BlockSet = require 'npge.model.BlockSet'
        for _, ref in ipairs(results) do
            local decrease_count = true
            local bs = BlockSet.fromRef(ref, decrease_count)
            table.insert(blocksets, bs)
        end
        local Merge = require 'npge.algo.Merge'
        local unpack = require 'npge.util.unpack'
        return Merge(unpack(blocksets))
    end)
end

Workers.GoodSubblocks = function(blockset)
    local code = [[
{'fraction_non_alphanumeric': 0.09392164042475284, 'fraction_numerical': 0.004393994873672647, 'mean_word_length': 2.8662420382165603, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1847598', 'n_tokens_mistral': 1663, 'n_tokens_neox': 1588, 'n_words': 496}