123456789_123456789_123456789_123456789_123456789_

Class: Redis::SubscribedClient

Relationships & Source Files
Inherits: Object
Defined in: lib/redis/subscribe.rb

Constant Summary

  • READ_SLICE = private

    Upper bound on each guarded read (see #next_event) — and therefore on how long a cross-thread write can wait for the monitor.

    # File 'lib/redis/subscribe.rb', line 7 
    0.05

Class Method Summary

Instance Method Summary

Constructor Details

.new(client) ⇒ SubscribedClient

[ GitHub ] 

  


# File 'lib/redis/subscribe.rb', line 10



def initialize(client)
  @client = client
  @write_monitor = Monitor.new
  # Count of writers waiting for (or holding) the monitor; mutations under
  # their own lock so it cannot drift. The read loop deschedules while
  # writers are queued — MRI mutexes barge, and on MRI 3.2 a writer can
  # otherwise starve behind consecutive read slices (Thread.pass is not a
  # handoff there).
  @pending_writes = 0
  @pending_writes_lock = Mutex.new
  @closed = false
end


  







Instance Method Details



  

    #call_v(command)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 23



def call_v(command)
  @pending_writes_lock.synchronize { @pending_writes += 1 }
  @write_monitor.synchronize do
    # A write must never reach a connection whose teardown is freeing it:
    # on the hiredis driver that is a native use-after-free (SIGSEGV), not
    # a rescuable exception.
    raise SubscriptionError, "This client is closed" if @closed

    @client.call_v(command)
  end
ensure
  @pending_writes_lock.synchronize { @pending_writes -= 1 }
end


  








  

    #close  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 73



def close
  # Serialized with every write and guarded read: freeing the connection
  # under either crashes the process on the hiredis driver.
  @pending_writes_lock.synchronize { @pending_writes += 1 }
  @write_monitor.synchronize do
    @closed = true
    @client.close
  end
ensure
  @pending_writes_lock.synchronize { @pending_writes -= 1 }
end


  








  

    #guarded_read(slice)   (protected)
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 139



def guarded_read(slice)
  @write_monitor.synchronize do
    raise SubscriptionError, "This client is closed" if @closed

    @client.next_event(slice)
  end
end


  








  

    #next_event(timeout)   (protected)
  



  

    
Reads the next subscription event in short monitor-guarded slices rather
than one indefinite read: on a connection error the driver's read path
disconnects the connection object in place (hiredis nulls its native
context), so an unguarded read racing a cross-thread write or #close is a
native use-after-free. Slicing under the writers' monitor means reads,
writes and close never overlap; a writer waits at most one slice.
Blocking semantics are preserved (a slice timeout just reads again); the
timed form keeps the caller's overall deadline, nil meaning "no event
within the requested timeout".

  



  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 119



def next_event(timeout)
  if timeout > 0
    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
    loop do
      writer_priority_wait
      remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
      return nil if remaining <= 0

      event = guarded_read([remaining, READ_SLICE].min)
      return event if event
    end
  else
    loop do
      writer_priority_wait
      event = guarded_read(READ_SLICE)
      return event if event
    end
  end
end


  








  

    #psubscribe(*channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 45



def psubscribe(*channels, &block)
  subscription("psubscribe", "punsubscribe", channels, block)
end


  








  

    #psubscribe_with_timeout(timeout, *channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 49



def psubscribe_with_timeout(timeout, *channels, &block)
  subscription("psubscribe", "punsubscribe", channels, block, timeout)
end


  








  

    #punsubscribe(*channels)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 65



def punsubscribe(*channels)
  call_v([:punsubscribe, *channels])
end


  








  

    #ssubscribe(*channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 53



def ssubscribe(*channels, &block)
  subscription("ssubscribe", "sunsubscribe", channels, block)
end


  








  

    #ssubscribe_with_timeout(timeout, *channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 57



def ssubscribe_with_timeout(timeout, *channels, &block)
  subscription("ssubscribe", "sunsubscribe", channels, block, timeout)
end


  








  

    #subscribe(*channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 37



def subscribe(*channels, &block)
  subscription("subscribe", "unsubscribe", channels, block)
end


  








  

    #subscribe_with_timeout(timeout, *channels, &block)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 41



def subscribe_with_timeout(timeout, *channels, &block)
  subscription("subscribe", "unsubscribe", channels, block, timeout)
end


  








  

    #subscription(start, stop, channels, block, timeout = 0)   (protected)
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 87



def subscription(start, stop, channels, block, timeout = 0)
  sub = Subscription.new(&block)

  case start
  when "ssubscribe" then channels.each { |c| call_v([start, c]) } # avoid cross-slot keys
  else call_v([start, *channels])
  end

  while event = next_event(timeout)
    if event.is_a?(::RedisClient::CommandError)
      raise Client::ERROR_MAPPING.fetch(event.class), event.message
    end

    type, *rest = event
    if callback = sub.callbacks[type]
      callback.call(*rest)
    end
    break if type == stop && rest.last == 0
  end
  # No need to unsubscribe here. The real client closes the connection
  # whenever an exception is raised (see #ensure_connected).
end


  








  

    #sunsubscribe(*channels)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 69



def sunsubscribe(*channels)
  call_v([:sunsubscribe, *channels])
end


  








  

    #unsubscribe(*channels)  
  

  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 61



def unsubscribe(*channels)
  call_v([:unsubscribe, *channels])
end


  








  

    #writer_priority_wait   (protected)
  



  

    
Bounded writer priority: a queued writer gets the monitor before the next
read slice. sleep deterministically deschedules this thread (Thread.pass
is a hint MRI 3.2 ignores in favor of the barging re-acquirer); the bound
means a counter defect could only throttle reads, never stop them.

  



  [ GitHub ]


  

  


# File 'lib/redis/subscribe.rb', line 151



def writer_priority_wait
  50.times do
    break if @pending_writes.zero?

    sleep 0.001
  end
end