Class: PgPipeline::ConnectionDriver

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_pipeline/connection_driver.rb

Constant Summary collapse

DEFAULT_MAX_PENDING = 
256
DEFAULT_MAX_IN_FLIGHT = 
64

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(conn, max_pending: DEFAULT_MAX_PENDING, max_in_flight: DEFAULT_MAX_IN_FLIGHT) ⇒ ConnectionDriver

Returns a new instance of ConnectionDriver.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/pg_pipeline/connection_driver.rb', line 33

def initialize(conn, max_pending: DEFAULT_MAX_PENDING, max_in_flight: DEFAULT_MAX_IN_FLIGHT)
  @max_pending = DriverOps.positive_integer!(max_pending, :max_pending)
  @max_in_flight = DriverOps.positive_integer!(max_in_flight, :max_in_flight)

  @conn = conn
  @caps = ServerCaps.from_connection(conn)
  @caps.assert_supported!
  DriverOps.warn_flush_coupling_once(@caps)
  @requests = BoundedQueue.new(@max_pending)
  @events = Runtime::Queue.new
  @reader_rearm = Runtime::Queue.new
  @writer_commands = Runtime::Queue.new

  @accepting = @running = @draining = @needs_flush = @writer_armed = @request_event_pending = false
  @readable_events = @results_read = @units_completed = @flush_calls =
    @flush_incomplete = @dispatches = @leaked_watchers = @submitting = 0
  @socket = @owner_task = @reader_task = @writer_task = @dispatching = nil
  @inflight = []
end

Instance Attribute Details

#accepting ⇒ Object

Returns the value of attribute accepting.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def accepting
  @accepting
end

#caps ⇒ Object (readonly)

Returns the value of attribute caps.



16
17
18
 
# File 'lib/pg_pipeline/connection_driver.rb', line 16

def caps
  @caps
end

#conn ⇒ Object (readonly)

Returns the value of attribute conn.



16
17
18
 
# File 'lib/pg_pipeline/connection_driver.rb', line 16

def conn
  @conn
end

#dispatches ⇒ Object 



30
 
# File 'lib/pg_pipeline/connection_driver.rb', line 30

def dispatches = @dispatches || 0

#dispatching ⇒ Object

Returns the value of attribute dispatching.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def dispatching
  @dispatching
end

#draining ⇒ Object

Returns the value of attribute draining.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def draining
  @draining
end

#events ⇒ Object

Returns the value of attribute events.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def events
  @events
end

#flush_calls ⇒ Object 



28
 
# File 'lib/pg_pipeline/connection_driver.rb', line 28

def flush_calls = @flush_calls || 0

#flush_incomplete ⇒ Object 



29
 
# File 'lib/pg_pipeline/connection_driver.rb', line 29

def flush_incomplete = @flush_incomplete || 0

#inflight ⇒ Object

Returns the value of attribute inflight.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def inflight
  @inflight
end

#leaked_watchers ⇒ Object 



31
 
# File 'lib/pg_pipeline/connection_driver.rb', line 31

def leaked_watchers = @leaked_watchers || 0

#max_in_flight ⇒ Object (readonly)

Returns the value of attribute max_in_flight.



16
17
18
 
# File 'lib/pg_pipeline/connection_driver.rb', line 16

def max_in_flight
  @max_in_flight
end

#max_pending ⇒ Object (readonly)

Returns the value of attribute max_pending.



16
17
18
 
# File 'lib/pg_pipeline/connection_driver.rb', line 16

def max_pending
  @max_pending
end

#needs_flush ⇒ Object

Returns the value of attribute needs_flush.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def needs_flush
  @needs_flush
end

#owner_task ⇒ Object

Returns the value of attribute owner_task.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def owner_task
  @owner_task
end

#readable_events ⇒ Object 



25
 
# File 'lib/pg_pipeline/connection_driver.rb', line 25

def readable_events = @readable_events || 0

#reader_rearm ⇒ Object

Returns the value of attribute reader_rearm.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def reader_rearm
  @reader_rearm
end

#reader_task ⇒ Object

Returns the value of attribute reader_task.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def reader_task
  @reader_task
end

#request_event_pending ⇒ Object

Returns the value of attribute request_event_pending.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def request_event_pending
  @request_event_pending
end

#requests ⇒ Object

Returns the value of attribute requests.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def requests
  @requests
end

#results_read ⇒ Object 



26
 
# File 'lib/pg_pipeline/connection_driver.rb', line 26

def results_read = @results_read || 0

#running ⇒ Object

Returns the value of attribute running.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def running
  @running
end

#socket ⇒ Object

Returns the value of attribute socket.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def socket
  @socket
end

#submitting ⇒ Object

Returns the value of attribute submitting.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def submitting
  @submitting
end

#units_completed ⇒ Object 



27
 
# File 'lib/pg_pipeline/connection_driver.rb', line 27

def units_completed = @units_completed || 0

#writer_armed ⇒ Object

Returns the value of attribute writer_armed.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def writer_armed
  @writer_armed
end

#writer_commands ⇒ Object

Returns the value of attribute writer_commands.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def writer_commands
  @writer_commands
end

#writer_task ⇒ Object

Returns the value of attribute writer_task.



17
18
19
 
# File 'lib/pg_pipeline/connection_driver.rb', line 17

def writer_task
  @writer_task
end

Instance Method Details

#abort!(error = ConnectionLostError.new("connection aborted")) ⇒ Object 



102
103
104
 
# File 'lib/pg_pipeline/connection_driver.rb', line 102

def abort!(error = ConnectionLostError.new("connection aborted"))
  DriverOps.abort!(self, error)
end

#available? ⇒ Boolean

Returns:

  • (Boolean) 


56
 
# File 'lib/pg_pipeline/connection_driver.rb', line 56

def available? = @accepting && @running

#dead? ⇒ Boolean

Returns:

  • (Boolean) 


57
 
# File 'lib/pg_pipeline/connection_driver.rb', line 57

def dead? = !@running && !@accepting

#graceful_close ⇒ Object 



100
 
# File 'lib/pg_pipeline/connection_driver.rb', line 100

def graceful_close = DriverOps.graceful_close(self)

#health_check(timeout) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/pg_pipeline/connection_driver.rb', line 81

def health_check(timeout)
  return true unless available?

  probe = Request.build("SELECT 1", nil)
  submit(probe)
  Runtime.with_timeout(timeout) { probe.wait }
  true
rescue Runtime::TimeoutError
  DriverOps.abort_timed_out_health_probe(self, probe)
rescue QueryError, PipelineAbortedError
  true
rescue ShutdownError, NotDispatchedError
  true
rescue ConnectionLostError, PG::Error
  false
ensure
  probe.cancel! if probe && !probe.settled?
end

#load ⇒ Object 



55
 
# File 'lib/pg_pipeline/connection_driver.rb', line 55

def load = @requests.size + @inflight.size + @submitting + (@dispatching ? 1 : 0)

#start ⇒ Object 



53
 
# File 'lib/pg_pipeline/connection_driver.rb', line 53

def start = DriverOps.start(self)

#stats ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/pg_pipeline/connection_driver.rb', line 59

def stats
  {
    available: available?,
    load: load,
    pending: @requests.size,
    in_flight: @inflight.size,
    submitting: @submitting,
    needs_flush: @needs_flush,
    fast_sync: @caps.fast_sync?,
    readable_events: readable_events,
    results_read: results_read,
    units_completed: units_completed,
    flush_calls: flush_calls,
    flush_incomplete: flush_incomplete,
    dispatches: dispatches,
    units_per_readable: DriverOps.ratio(units_completed, readable_events),
    results_per_readable: DriverOps.ratio(results_read, readable_events),
    flush_calls_per_unit: DriverOps.ratio(flush_calls, units_completed),
    leaked_watchers: leaked_watchers
  }
end

#submit(request) ⇒ Object 



54
 
# File 'lib/pg_pipeline/connection_driver.rb', line 54

def submit(request) = DriverOps.submit(self, request)