synca: md5 bruteforce with portals, workers and channels

#include <synca/synca.h>
#include <synca/log.h>

#include <fstream>
#include <regex>
#include <synca/channel.h>

#include "md5.h"

using namespace synca;

void next(char& b)
{
    switch (b)
    {
    case 'z': b = '0'; break;
    case '9': b = 'a'; break;
    default: ++ b; break;
    }
}

bool next(std::string& p)
{
    for (auto e = p.rbegin(); e != p.rend(); ++ e)
    {
        next(*e);
        if (*e != '0')
            return true;
    }
    return false;
}

std::vector<std::string> splitLine(const std::string& line)
{
    static const std::regex fmt(R"(^\s*([0-9a-fA-F]+)\s+([0-9a-z]+)\s+([0-9a-z]+)\s*$)");
    std::smatch parts;
    VERIFY(std::regex_match(line, parts, fmt), "Invalid format");
    return {parts.str(1), parts.str(2), parts.str(3)};
}

struct Disk
{
    boost::optional<std::string> readLine(std::istream& is)
    {
        std::string line;
        return std::getline(is, line) ? boost::optional<std::string>{line} : boost::optional<std::string>{};
    }
};

struct Master
{
    void processFile(const char* name)
    {
        JLOG("opening: " << name);
        std::ifstream ifs(name);
        VERIFY(ifs, "File open error");
        while (true)
        {
            auto line = portal<Disk>()->readLine(ifs);
            if (!line)
                break;
            if (line->empty())
                continue;
            lines.put(*line);
            tryStartWorker();
        }
        isClosed = true;
        lines.close();
        LOG("file parsed");
    }

private:
    void onPassword(const std::string& md5, const std::string& pwd)
    {
        RJLOG(md5 << ", found password: " << pwd);
    }

    void onError(const std::string& line, const std::string& e)
    {
        RJLOG(line << ", error: " << e);
    }

    void onNotFound(const std::string& md5)
    {
        RJLOG(md5 << ", not found");
    }

    void onWorkerCompleted()
    {
        ++ availableWorkers;
        tryStartWorker();
    }

    static void processLine(const std::string& line)
    {
        JLOG("process line: " << line);
        auto split = splitLine(line);
        auto&& md5hex = split[0];
        auto&& left = split[1];
        auto&& right = split[2];
        JLOG("process: " << md5hex << "," << left << "," << right);
        VERIFY(left.size() == right.size() && left <= right, "Inconsistent range");
        auto etalon = md5_cpp11::from_hex_string(md5hex);
        while (true)
        {
            if (md5_cpp11::make_digest(left) == etalon)
            {
                portal<Master>()->onPassword(md5hex, left);
                return;
            }
            if (!next(left) || left == right)
            {
                portal<Master>()->onNotFound(md5hex);
                return;
            }
        }
    }

    static void worker(Channel<std::string>& lines)
    {
        std::string line;
        try
        {
            while (lines.get(line))
                processLine(line);
        }
        catch (std::exception& e)
        {
            portal<Master>()->onError(line, e.what());
        }
        portal<Master>()->onWorkerCompleted();
    }

    void tryStartWorker()
    {
        if (isClosed || availableWorkers == 0)
            return;
        -- availableWorkers;
        go([&] {
            worker(lines);
        });
    }

    int availableWorkers = threadConcurrency();
    bool isClosed = false;
    Channel<std::string> lines;
};

int main(int argv, const char* argc[])
{
    try
    {
        VERIFY(argv == 2, "Invalid args");
        ThreadPool cpu(threadConcurrency(), "cpu");
        ThreadPool disk(1, "disk");
        scheduler<DefaultTag>().attach(cpu);
        Alone masterAlone(cpu, "master");
        portal<Master>().attach(masterAlone);
        portal<Disk>().attach(disk);
        go([&] {
            portal<Master>()->processFile(argc[1]);
        });
        waitForAll();
    }
    catch (std::exception& e)
    {
        RLOG("Error: " << e.what());
        return 1;
    }
    return 0;
}