Class: Neo4j::Driver::Bolt::RecordBuffer

Inherits:
Object
  • Object
show all
Defined in:
lib/neo4j/driver/bolt/record_buffer.rb

Overview

Record buffer for one streaming result, with a high/low watermark autopull policy. Sits between the connection's reader (producer) and the cursor (consumer), and unifies two things under one mutex + condition variable: the record queue and the batch's "has_more promise".

Records are delivered incrementally — pushed the instant the reader decodes them — so the cursor sees record 1 without waiting for the whole batch. The batch's terminating SUCCESS resolves the promise (batch_complete), which says whether the server has more. The cursor drives flow control:

* after each record, once drained to/below the low watermark, it checks
the promise (`pull_ready?`, non-blocking) and, if fulfilled with
has_more, issues the next `PULL {n: fetch_size}` — prefetch overlap;
* when it drains the buffer empty mid-stream, it `await`s — one wait that
wakes on *either* the next record *or* the promise resolving. That
shared wait is the crux: mid-batch it wakes on a record (incremental
delivery preserved), at batch-end it wakes on the promise (so it PULLs
the next batch instead of blocking forever on a record that, the SUCCESS
having no payload, will never come).

The cursor is the sole writer (issues every PULL/DISCARD); the reader only ever fills. All sync primitives are stdlib + scheduler-aware, so a cursor running as a fiber under a host reactor yields rather than blocking the thread. See docs/unified-pipeline.md.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(fetch_size:, high_watermark: nil, low_watermark: nil) ⇒ RecordBuffer

Returns a new instance of RecordBuffer.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 33

def initialize(fetch_size:, high_watermark: nil, low_watermark: nil)
  @fetch_size = fetch_size
  # Defaults: refill under half of ~2 batches held. fetch_size -1 ("pull
  # all in one batch") never paginates, so just size the watermarks for a
  # steady single-batch handoff.
  @high_watermark = high_watermark || (fetch_size.positive? ? fetch_size * 2 : 1000)
  @low_watermark = low_watermark || [@high_watermark / 2, 1].max
  @mutex = Mutex.new
  @cv = ConditionVariable.new
  @records = [] # incrementally filled by the reader, drained by the cursor
  @has_more = true # server may have more records for this stream
  @pull_in_flight = true # the first PULL is pipelined with RUN — promise unfulfilled
  @ended = false # terminal SUCCESS/IGNORED seen → no more records
  @error = nil # stream failed; re-raised to the cursor after buffered records
  @summary = nil # terminating SUCCESS metadata
end

Instance Attribute Details

#fetch_size ⇒ Object (readonly)

Returns the value of attribute fetch_size.



31
32
33
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 31

def fetch_size
  @fetch_size
end

#high_watermark ⇒ Object (readonly)

Returns the value of attribute high_watermark.



31
32
33
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 31

def high_watermark
  @high_watermark
end

#low_watermark ⇒ Object (readonly)

Returns the value of attribute low_watermark.



31
32
33
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 31

def low_watermark
  @low_watermark
end

Instance Method Details

#await ⇒ Object

Block (colorlessly) until there's something to re-evaluate: a record was delivered, the batch completed (promise resolved), or the stream ended/failed. Called only when the cursor has drained the buffer empty and a PULL is still outstanding (records coming, or its SUCCESS pending).



119
120
121
122
123
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 119

def await
  @mutex.synchronize do
    @cv.wait(@mutex) while @records.empty? && !@ended && @error.nil? && @pull_in_flight
  end
end

#batch_complete(has_more:) ⇒ Object

The current batch's terminal SUCCESS arrived — resolve the promise. has_more says whether to keep paging; either way no PULL is in flight, so the cursor may issue the next once it drains past the low watermark. Wake a cursor parked in #await waiting for exactly this.



65
66
67
68
69
70
71
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 65

def batch_complete(has_more:)
  @mutex.synchronize do
    @has_more = has_more
    @pull_in_flight = false
    @cv.broadcast
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


125
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 125

def empty? = @mutex.synchronize { @records.empty? }

#ended? ⇒ Boolean

Returns:

  • (Boolean) 


127
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 127

def ended? = @mutex.synchronize { @ended }

#fail(error) ⇒ Object

The stream failed: arm the error the cursor re-raises after it drains the records already delivered (records that preceded the failure are valid), and wake it.



88
89
90
91
92
93
94
95
96
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 88

def fail(error)
  @mutex.synchronize do
    @error ||= error
    @ended = true
    @has_more = false
    @pull_in_flight = false
    @cv.broadcast
  end
end

#finish(summary = nil) ⇒ Object

Final batch (terminating SUCCESS without has_more, or IGNORED): stash the summary, mark ended, and wake the cursor.



75
76
77
78
79
80
81
82
83
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 75

def finish(summary = nil)
  @mutex.synchronize do
    @summary = summary
    @ended = true
    @has_more = false
    @pull_in_flight = false
    @cv.broadcast
  end
end

#has_more? ⇒ Boolean

Returns:

  • (Boolean) 


128
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 128

def has_more? = @mutex.synchronize { @has_more }

#note_pull_issued ⇒ Object

The cursor issued the next PULL — the promise is unfulfilled again; don't issue another until this batch completes.



139
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 139

def note_pull_issued = @mutex.synchronize { @pull_in_flight = true }

#pull_ready? ⇒ Boolean

True when the cursor should issue the next PULL: the batch promise is fulfilled (none in flight) with has_more, and the buffer has drained to/below the low watermark.

Returns:

  • (Boolean) 


135
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 135

def pull_ready? = @mutex.synchronize { @has_more && !@pull_in_flight && @records.size <= @low_watermark }

#push_record(record) ⇒ Object

Append a decoded record and wake a cursor parked in #await. Never blocks (unbounded); the cursor's watermark bounds how much the server ships.



54
55
56
57
58
59
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 54

def push_record(record)
  @mutex.synchronize do
    @records.push(record)
    @cv.broadcast
  end
end

#size ⇒ Object 



126
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 126

def size = @mutex.synchronize { @records.size }

#summary ⇒ Object

The terminating SUCCESS's metadata (nil until finished / on failure).



99
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 99

def summary = @mutex.synchronize { @summary }

#try_shift ⇒ Object

Non-blocking: the next buffered record, or :empty when none is buffered yet, or :ended once the stream is drained and terminal. Buffered records are handed out before a stream failure is raised (they preceded it).



106
107
108
109
110
111
112
113
 
# File 'lib/neo4j/driver/bolt/record_buffer.rb', line 106

def try_shift
  @mutex.synchronize do
    return @records.shift unless @records.empty?
    raise @error if @error

    @ended ? :ended : :empty
  end
end