Add support for parsing chunked data

[kou]

This is similar to #854 but a different feature request. Both of #854 and this are for migrating to the json gem from yajl-ruby (in Fluentd. #854's description doesn't mention it but it's mentioned in a comment #854 (comment) ) but this proposes different API. We can't migrate from yajl-ruby with API proposed in #854.

#854 proposed parser.parse(io) like API but what we need is io.each_chunk {|chunk| parser.feed(chunk)} like API. An application wants to read data not the json gem read data. parser.parse(io) like API reads data by the json gem. Fluentd can't migrate to yajl-ruby with the API.

Fluentd that is a event based log collector reads data from multiple sockets efficiently by IO multiplexing. We may not be able to read a whole JSON (e.g. ["a", 2, true]) at a time with the IO multiplexing. We may get a whole JSON as multiple chunks (e.g. ["a", , 2, and true]). yajl-ruby can parse chunks without concatenating all of them:

yajl-ruby:

require "yajl"
parser = Yajl::Parser.new
parser.on_parse_complete = lambda {|x| pp x}
parser << "[\"a\""
parser << ", 2, "
parser << "true]" # lambda {|x| pp x} is called with ["a", 2, true]

The json gem:

require "json"
JSON.parse(["[\"a\"", ", 2, ", "true]"].join)

If an application can extract a whole JSON from an input without parsing input (*), we don't need yajl-ruby style API. But an input may have multiple JSONs (like Fluentd's use case), we need yajl-ruby style API. Because we need to parse JSON to split data in an input to JSONs for this case.

(*) For example:

1. Only one JSON exists in an input
2. Each JSON is delimited by new line (JSON lines)
3. ...

API

JSON::Parser.new requires a source. So it may be better that we define a new class such as JSON::StreamParser instead of reusing existing JSON::Parser.

JSON::StreamParser accepts a chunk not a whole JSON:

parser = JSON::StreamParser.new
parser.parse_chunk("[\"a\"")
parser.parse_chunk(", 2, ")
parser.parse_chunk("true]")

Other candidate for parse_chunk: consume

We need to define callback style API to pass parsed data to application. Because one parse_chunk call may parse 0 or more JSONs. For example, if chunk is [1][2], [1] and [2] are parsed at once.

JSON::Parser accepts on_load callback. We may resuse the name for JSON::StreamParser but on_load is for JSON value not JSON. It may be better that we use different name. For example:

parser = JSON::StreamParser.new
parser.on_load = lambda {|x| pp x}
parser.parse_chunk("[\"a\"")
parser.parse_chunk(", 2, ")
parser.parse_chunk("true]") # lambda {x| pp x} is called with ["a", 2, true]
parser.parse_chunk("[1][2]") # lambda {x| pp x} is called twice with [1] and [2]

[byroot]

Thanks, I better understand the ask now (the streaming term is used for different things in JSON parsing context).

I'm open to such feature, as I think it can be useful, however I suspect it would require a fairly large refactoring of the parser.

Right now the parser use recursive C function calls, so it can't really be interrupted or resumed.

So the parser would first need to be changed quite a bit to look more like msgpack's parser: https://github.com/msgpack/msgpack-ruby/blob/677e6b888f1512550d9b3cdb4b5aca28f9fdb859/ext/msgpack/unpacker.c, which is something I considered in the past, but didn't act on.

The best approach is likely to keep a dedicated stack to represent the parser state. (e.g. .stack = { ARRAY, ARRAY, OBJECT, OBJECT_KEY... }).

And that's assuming such change wouldn't degrade performance too much, if it does, then we'd need to consider whether we have to maintain two parsers so that the classic usage keep performing well.

As for the API, I think I'd prefer something that doesn't use callback, like MessagePack:

parser = JSON::StreamParser.new
parser.feed("[\"a\"")
parser.read # JSON::EOFError

parser.feed(", 2, ")
parser.read # JSON::EOFError

parser.feed("true]")
parser.read # => ["a", 2, true]

parser.feed("[1][2]")
parser.read # => [1]
parser.read # => [2]

But perhaps exceptions aren't ideal. The problem is we can't just return nil given that null is a valid document in some contexts. So perhaps something like:

parser = JSON::StreamParser.new
parser.feed("[\"a\"")
parser.read? # => false

parser.feed(", 2, ")
parser.read? # => false

parser.feed("true]")
parser.read? # => true
parser.read # => ["a", 2, true]

parser.feed("[1][2]")
parser.read? # => true
parser.read # => [1]

parser.read? # => true
parser.read # => [2]

parser.read? # => false

Also this is just for the C parser, which I think can reasonably be converted to this new architecture. The JRuby parser still uses ragel and would be significantly more work. But I'd be OK with only having this feature in the C extension until someone offer to make significant changes to the Jave version.

[eregon]

This sounds likely to degrade performance significantly to me and result in a much more complicated parser.
Maybe one approach could be to basically:

buf = +""
buf << "[\"a\""
json = JSON.parse(buf) rescue nil # => nil
buf << ", 2, "
json = JSON.parse(buf) rescue nil # => nil
buf << "true]"
json = JSON.parse(buf) rescue nil # => ["a", 2, true]

That would also require the input to be given in chunks that terminate, would that be enough?
If not maybe JSON could have a mode where it accepts extra trailing bytes, and give a way to know how many bytes it parsed so we can restart parsing after that.
That would be needed for the case of buf = "[1][2]".

[eregon]

Another thought is if this is strict JSON with only native JSON types a Regexp (using Subexpression Calls: \g) might be enough to:

• Find if an input is a complete JSON expression
• And if so find the end of it

[byroot]

This sounds likely to degrade performance significantly to me and result in a much more complicated parser.

Not necessarily. It's an idea I toyed with for a while, but ultimately it just replace direct calls with a stack based dispatch, not that different from a bytecode interpreter.

Whether it can be as fast is hard to be certain until it's implemented, and it's a significant amount of work so...

My worry would be more about the mere concept of a significant rewrite, as it's potential for new bugs and such.

As for your suggestion, it may be fine if you only occasionally have partial documents, but from my limited understading of fluentd, it's a very common case, so having to restart parsing from 0 every time would be prohibitive I fear.

[headius]

I originally proposed the streaming dumping and I agree this would be good to add.

[kou]

That would also require the input to be given in chunks that terminate, would that be enough?

It's suboptimal. It parses one JSON multiple times. If all use cases don't require performance, it's acceptable. But Fluentd is a log collector and log collector needs performance because it needs to process many logs.

[eregon]

Not necessarily. It's an idea I toyed with for a while, but ultimately it just replace direct calls with a stack based dispatch, not that different from a bytecode interpreter.

Using that analogy the current version has no dispatch, like JITed Ruby code, and this would have dispatch overhead, like a bytecode interpreter. So that seems pretty clear there will be some overhead. Also it means the CPU can't predict the next function to be called, it will be just a function pointer I guess and that's well known to be slower.

Maybe it's worth asking the reverse question, what are the problems with yajl? It seems purpose-built for this, so seems a good fit, no? Most usages of JSON don't need this, but supporting it would slow parsing down and add significant complexity.

[kou]

yajl-ruby may not be maintained actively. See also: https://github.com/brianmario/yajl-ruby

[byroot]

Yeah, yajl has maintenance concerns.

So that seems pretty clear there will be some overhead.

I suspect it would be negligible with the proper optimizations (direct threading & al), especially since from experience the dispatch really isn't where the most time is spent.

But regardless, if it's indeed a perf regression, it's always possible to have two parsers that would share a lot of code.

Most usages of JSON don't need this,

Indeed, which is my biggest hesitation to be honest. So far my stance has been that I wanted to support the 95% use case, and not grow the API too much, leaving more specialized use cases to alternative gems.

Something I wonder, and sorry if it's obvious because I know very little about fluentd, is why documents can't be separated by a newline or some other marker? Are the JSON blobs purely black boxes and you can't assume they're on a single line? Not trying to say you are holding it wrong, mostly just curious.

[kou]

is why documents can't be separated by a newline or some other marker?

Because Fluentd's Forward Protocol Specification v1 https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1 didn't define it... There are tools that implement this specification. So Fluentd can't assume that JSONs are separated by something to keep backward compatibility.

[kddnewton]

I have a JSON parser that doesn't use recursion here: https://gist.github.com/kddnewton/3e90d0dceda633641933e98bd4ee8694. The original commit was what I pushed up for the JSON rewrite, I don't mind porting the second commit here so we can compare perf.

[headius]

No concerns with chunked parser from the JRuby side. Dispatches from Java back into Ruby will soon be inlinable and should not impact performance.

A rewrite for JRuby may be coming soon.

[kddnewton]

This starts the work: #989 in case you're interested. Feel free to close otherwise.

[enebo]

I will be submitting a PR for a new JRuby JSON parser this week. At worst, a new parser will need to be added for streaming but it will use 95+% of the same logic and just need to do a little extra tracking and it might not because the new parser uses a stack like @byroot mentions above with state like:

    enum ParserState {
        ELEMENT,        // Start of document (also to never have empty stack)
        OBJECT_KEY,     // "key"
        OBJECT_COLON,   // ':'
        OBJECT_VALUE,   // (string, number, object, array, true, false, null)
        OBJECT_COMMA,   // ','
        ARRAY_VALUE,    // (string, number, object, array, true, false, null)
        ARRAY_COMMA,    // ','
    }

in a stack:

    protected static class StateElement {
        ParserState state;
        IRubyObject key;
        Object value;
    }

So at a fail it knows if state is at OBJECT_COMMA, then it is expecting a ','. So the next entry into parse might just continue where it left off. Anyways, just pointing out I do not think this will be a big problem with the new parser.