Caching streams in Functional Reactive Programming

Question

I have an application which is written entirely using the FRP paradigm and I think I am having performance issues due to the way that I am creating the streams. It is written in Haxe but the problem is not language specific.

For example, I have this function which returns a stream that resolves every time a config file is updated for that specific section like the following:

function getConfigSection(section:String) : Stream<Map<String, String>> {
    return configFileUpdated()
        .then(filterForSectionChanged(section))
        .then(readFile)
        .then(parseYaml);
}

In the reactive programming library I am using called promhx each step of the chain should remember its last resolved value but I think every time I call this function I am recreating the stream and reprocessing each step. This is a problem with the way I am using it rather than the library.

Since this function is called everywhere parsing the YAML every time it is needed is killing the performance and is taking up over 50% of the CPU time according to profiling.

As a fix I have done something like the following using a Map stored as an instance variable that caches the streams:

function getConfigSection(section:String) : Stream<Map<String, String>> {
    var cachedStream = this._streamCache.get(section);
    if (cachedStream != null) {
        return cachedStream;
    }

    var stream = configFileUpdated()
        .filter(sectionFilter(section))
        .then(readFile)
        .then(parseYaml);

    this._streamCache.set(section, stream);
    return stream;
}

This might be a good solution to the problem but it doesn't feel right to me. I am wondering if anyone can think of a cleaner solution that maybe uses a more functional approach (closures etc.) or even an extension I can add to the stream like a cache function.

Another way I could do it is to create the streams before hand and store them in fields that can be accessed by consumers. I don't like this approach because I don't want to make a field for every config section, I like being able to call a function with a specific section and get a stream back.

I'd love any ideas that could give me a fresh perspective!

Answer 1

Well, I think one answer is to just abstract away the caching like so:

class Test {
    static function main() {
        var sideeffects = 0;
        var cached = memoize(function (x) return x + sideeffects++);
        cached(1);
        trace(sideeffects);//1
        cached(1);
        trace(sideeffects);//1
        cached(3);
        trace(sideeffects);//2
        cached(3);
        trace(sideeffects);//2
    }
    @:generic static function memoize<In, Out>(f:In->Out):In->Out {
        var m = new Map<In, Out>();
        return
            function (input:In) 
                return switch m[input] {
                    case null: m[input] = f(input);
                    case output: output;
                }
    }
}

You may be able to find a more "functional" implementation for memoize down the road. But the important thing is that it is a separate thing now and you can use it at will.

You may choose to memoize(parseYaml) so that toggling two states in the file actually becomes very cheap after both have been parsed once. You can also tweak memoize to manage the cache size according to whatever strategy proves the most valuable.