<commit_before>/**
   A parallel word-frequency counting program.

   This is meant primarily to demonstrate Rust's MapReduce framework.

   It takes a list of files on the command line and outputs a list of
   words along with how many times each word is used.

*/

use std;

import std::io;
import option = std::option::t;
import std::option::some;
import std::option::none;
import std::str;
import std::vec;
import std::map;

mod map_reduce {
    export putter;
    export getter;
    export mapper;
    export reducer;
    export map_reduce;

    type putter = fn(str, str) -> ();

    type mapper = fn(str, putter);

    type getter = fn() -> option[str];

    type reducer = fn(str, getter);


    fn map_reduce (vec[str] inputs,
                   mapper f,
                   reducer reduce) {
        auto intermediates = map::new_str_hash[vec[str]]();

        fn emit(&map::hashmap[str, vec[str]] im,
                str key, str val) {
            auto old = [];
            alt(im.remove(key)) {
                case (some(?v)) {
                    old = v;
                }
                case (none) { }
            }

            im.insert(key, old + [val]);
        }

        for (str i in inputs) {
            f(i, bind emit(intermediates, _, _));
        }

        fn get(vec[str] vals, &mutable uint i) -> option[str] {
            i += 1u;
            if(i <= vec::len(vals)) {
                some(vals.(i - 1u))
            }
            else {
                none
            }
        }

        for each (@tup(str, vec[str]) kv in intermediates.items()) {
            auto i = 0u;
            reduce(kv._0, bind get(kv._1, i));
        }
    }
}

fn main(vec[str] argv) {
    if(vec::len(argv) < 2u) {
        auto out = io::stdout();

        out.write_line(#fmt("Usage: %s <filename> ...", argv.(0)));
        fail;
    }

    fn map(str filename, map_reduce::putter emit) {
        auto f = io::file_reader(filename);

        while(true) {
            alt(read_word(f)) {
                case (some(?w)) {
                    emit(w, "1");
                }
                case (none) {
                    break;
                }
            }
        }
    }

    fn reduce(str word, map_reduce::getter get) {
        auto count = 0;

        while(true) {
            alt(get()) {
                case(some(_)) { count += 1 }
                case(none) { break }
            }
        }

        auto out = io::stdout();
        out.write_line(#fmt("%s: %d", word, count));
    }

    map_reduce::map_reduce(vec::slice(argv, 1u, vec::len(argv)), map, reduce);
}

fn read_word(io::reader r) -> option[str] {
    auto w = "";

    while(!r.eof()) {
        auto c = r.read_char();

        if(is_word_char(c)) {
            w += str::from_char(c);
        }
        else {
            if(w != "") {
                ret some(w);
            }<commit_msg>Task-ified the word count program.<commit_after>/**
   A parallel word-frequency counting program.

   This is meant primarily to demonstrate Rust's MapReduce framework.

   It takes a list of files on the command line and outputs a list of
   words along with how many times each word is used.

*/

use std;

import std::io;
import option = std::option::t;
import std::option::some;
import std::option::none;
import std::str;
import std::vec;
import std::map;

fn map(str filename, map_reduce::putter emit) {
    auto f = io::file_reader(filename);

    while(true) {
        alt(read_word(f)) {
            case (some(?w)) {
                emit(w, "1");
            }
            case (none) {
                break;
            }
        }
    }
}

fn reduce(str word, map_reduce::getter get) {
    auto count = 0;

    while(true) {
        alt(get()) {
            case(some(_)) { count += 1 }
            case(none) { break }
        }
    }

    auto out = io::stdout();
    out.write_line(#fmt("%s: %d", word, count));
}

mod map_reduce {
    export putter;
    export getter;
    export mapper;
    export reducer;
    export map_reduce;

    type putter = fn(str, str) -> ();

    type mapper = fn(str, putter);

    type getter = fn() -> option[str];

    type reducer = fn(str, getter);

    tag ctrl_proto {
        find_reducer(str, chan[chan[reduce_proto]]);
        mapper_done;
    }

    tag reduce_proto {
        emit_val(str);
        done;
    }

    fn start_mappers(chan[ctrl_proto] ctrl,
                     vec[str] inputs) {
        for(str i in inputs) {
            spawn map_task(ctrl, i);
        }
    }

    fn map_task(chan[ctrl_proto] ctrl,
                str input) {

        auto intermediates = map::new_str_hash();

        fn emit(&map::hashmap[str, chan[reduce_proto]] im,
                chan[ctrl_proto] ctrl,
                str key, str val) {
            auto c;
            alt(im.find(key)) {
                case(some(?_c)) {
                    c = _c
                }
                case(none) {
                    auto p = port[chan[reduce_proto]]();
                    ctrl <| find_reducer(key, chan(p));
                    p |> c;
                    im.insert(key, c);
                }
            }
            c <| emit_val(val);
        }

        map(input, bind emit(intermediates, ctrl, _, _));
        ctrl <| mapper_done;
    }

    fn reduce_task(str key, chan[chan[reduce_proto]] out) {
        auto p = port();

        out <| chan(p);

        fn get(port[reduce_proto] p) -> option[str] {
            auto m;
            p |> m;

            alt(m) {
                case(emit_val(?v)) { ret some(v); }
                case(done) { ret none; }
            }
        }

        reduce(key, bind get(p));
    }

    fn map_reduce (vec[str] inputs) {
        auto ctrl = port[ctrl_proto]();

        // This task becomes the master control task. It spawns others
        // to do the rest.

        let map::hashmap[str, chan[reduce_proto]] reducers;

        reducers = map::new_str_hash();

        start_mappers(chan(ctrl), inputs);

        auto num_mappers = vec::len(inputs) as int;

        while(num_mappers > 0) {
            auto m;
            ctrl |> m;

            alt(m) {
                case(mapper_done) { num_mappers -= 1; }
                case(find_reducer(?k, ?cc)) {
                    auto c;
                    alt(reducers.find(k)) {
                        case(some(?_c)) { c = _c; }
                        case(none) {
                            auto p = port();
                            spawn reduce_task(k, chan(p));
                            p |> c;
                            reducers.insert(k, c);
                        }
                    }
                    cc <| c;
                }
            }
        }

        for each(@tup(str, chan[reduce_proto]) kv in reducers.items()) {
            kv._1 <| done;
        }
    }
}

fn main(vec[str] argv) {
    if(vec::len(argv) < 2u) {
        auto out = io::stdout();

        out.write_line(#fmt("Usage: %s <filename> ...", argv.(0)));
        fail;
    }

    map_reduce::map_reduce(vec::slice(argv, 1u, vec::len(argv)));
}

fn read_word(io::reader r) -> option[str] {
    auto w = "";

    while(!r.eof()) {
        auto c = r.read_char();

        if(is_word_char(c)) {
            w += str::from_char(c);
        }
        else {
            if(w != "") {
                ret some(w);
            }
{'fraction_non_alphanumeric': 0.13578404774823657, 'fraction_numerical': 0.0027129679869777536, 'mean_word_length': 1.4039778284969024, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 18, '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': '6525680', 'n_tokens_mistral': 2384, 'n_tokens_neox': 2243, 'n_words': 553}