Module: PgPipeline::PoolOps

Defined in:
lib/pg_pipeline/pool.rb

Class Method Summary collapse

Class Method Details

.finite_float!(value, name, allow_zero: false) ⇒ Object 



576
577
578
579
580
581
582
583
584
585
 
# File 'lib/pg_pipeline/pool.rb', line 576

def finite_float!(value, name, allow_zero: false)
  number = Float(value)
  bound_ok = allow_zero ? number >= 0 : number.positive?
  raise ArgumentError unless number.finite? && bound_ok

  number
rescue ArgumentError, TypeError
  requirement = allow_zero ? "non-negative finite" : "positive finite"
  raise ArgumentError, "#{name} must be a #{requirement} number (got #{value.inspect})"
end

.new_connection(connection_args) ⇒ Object 



634
635
636
 
# File 'lib/pg_pipeline/pool.rb', line 634

def new_connection(connection_args)
  connection_args.nil? ? PG::Connection.new : PG::Connection.new(connection_args)
end

.nonnegative_integer!(value, name) ⇒ Object 



587
588
589
590
591
592
593
594
 
# File 'lib/pg_pipeline/pool.rb', line 587

def nonnegative_integer!(value, name)
  integer = Integer(value)
  raise ArgumentError, "#{name} must be >= 0" if integer.negative?

  integer
rescue ArgumentError, TypeError
  raise ArgumentError, "#{name} must be an integer >= 0"
end

.positive_integer!(value, name) ⇒ Object 



567
568
569
570
571
572
573
574
 
# File 'lib/pg_pipeline/pool.rb', line 567

def positive_integer!(value, name)
  integer = Integer(value)
  raise ArgumentError, "#{name} must be >= 1" if integer < 1

  integer
rescue ArgumentError, TypeError
  raise ArgumentError, "#{name} must be an integer >= 1"
end

.safe_close(conn) ⇒ Object 



656
657
658
659
660
 
# File 'lib/pg_pipeline/pool.rb', line 656

def safe_close(conn)
  conn.close unless conn.finished?
rescue StandardError
  nil
end

.sanitize_pinned_connection(conn) ⇒ Object

Raises:



638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
 
# File 'lib/pg_pipeline/pool.rb', line 638

def sanitize_pinned_connection(conn)
  raise ConnectionLostError, "pinned connection is closed" if conn.finished?
  raise ConnectionLostError, "pinned connection is bad" unless conn.status == PG::CONNECTION_OK

  case conn.transaction_status
  when PG::PQTRANS_IDLE
    nil
  when PG::PQTRANS_INTRANS, PG::PQTRANS_INERROR
    conn.exec("ROLLBACK")
  else
    raise ConnectionLostError,
          "pinned connection returned in unsafe transaction state #{conn.transaction_status}"
  end

  conn.exec("DISCARD ALL")
  true
end

.select_driver(drivers, rr) ⇒ Object 



628
629
630
631
632
 
# File 'lib/pg_pipeline/pool.rb', line 628

def select_driver(drivers, rr)
  slot = [rr]
  driver = select_driver_into(drivers, rr, slot)
  [driver, slot[0]]
end

.select_driver_into(drivers, rr, slot) ⇒ Object 



596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
 
# File 'lib/pg_pipeline/pool.rb', line 596

def select_driver_into(drivers, rr, slot)
  size = drivers.length
  if size.zero?
    slot[0] = rr
    return nil
  end

  start = rr % size
  best = nil
  best_index = 0
  best_load = 0
  offset = 0

  while offset < size
    index = start + offset
    index -= size if index >= size
    driver = drivers[index]
    offset += 1
    next unless driver.available?

    load = driver.load
    next unless best.nil? || load < best_load

    best = driver
    best_index = index
    best_load = load
  end

  slot[0] = best ? (best_index + 1) % size : rr
  best
end