Pure-Ruby gleichzeitige Hash

Okay, jetzt, dass Sie angegeben haben, die tatsächlich Bedeutung von "threadsicher", hier werden zwei mögliche Implementierungen. Der folgende code wird ausgeführt für immer im MRT und JRuby. Die lockless Implementierung folgt eine eventual consistency Modell, bei dem jeder thread verwendet es die eigene Ansicht des hash-wenn der Meister im Wandel. Es ist etwas Trickserei erforderlich, um sicher zu speichern alle Informationen, die in dem thread nicht ein Speicherleck, aber das ist behandelt und getestet ― Prozess die Größe nicht wachsen, läuft dieser code. Beide Implementierungen müssten mehr arbeiten, um sein "complete", was bedeutet, delete, update, etc. müssten einige denken, aber eines der beiden Konzepte unter Ihren Anforderungen gerecht wird.

Es ist sehr wichtig für Menschen, die Lesen dieses Threads klar, das ganze Thema ist exklusiv für JRuby ― im MRT der built-in Hash ist ausreichend.

module Cash
  def Cash.new(*args, &block)
    env = ENV['CASH_IMPL']
    impl = env ? Cash.const_get(env) : LocklessImpl
    klass = defined?(JRUBY_VERSION) ? impl : ::Hash
    klass.new(*args)
  end

  class LocklessImpl
    def initialize
      @hash = {}
    end

    def thread_hash
      thread = Thread.current
      thread[:cash] ||= {}
      hash = thread[:cash][thread_key]
      if hash
        hash
      else
        hash = thread[:cash][thread_key] = {}
        ObjectSpace.define_finalizer(self){ thread[:cash].delete(thread_key) }
        hash
      end
    end

    def thread_key
      [Thread.current.object_id, object_id]
    end

    def []=(key, val)
      time = Time.now.to_f
      tuple = [time, val]
      @hash[key] = tuple
      thread_hash[key] = tuple
      val
    end

    def [](key)
    # check the master value
    #
      val = @hash[key]

    # someone else is either writing the key or it has never been set.  we
    # need to invalidate our own copy in either case
    #
      if val.nil?
        thread_val = thread_hash.delete(key)
        return(thread_val ? thread_val.last : nil)
      end

    # check our own thread local value
    #
      thread_val = thread_hash[key]

    # in this case someone else has written a value that we have never seen so
    # simply return it
    #
      if thread_val.nil?
        return(val.last)
      end

    # in this case there is a master *and* a thread local value, if the master
    # is newer juke our own cached copy
    #
      if val.first > thread_val.first
        thread_hash.delete(key)
        return val.last
      else
        return thread_val.last
      end
    end
  end

  class LockingImpl < ::Hash
    require 'sync'

    def initialize(*args, &block)
      super
    ensure
      extend Sync_m
    end

    def sync(*args, &block)
      sync_synchronize(*args, &block)
    end

    def [](key)
      sync(:SH){ super }
    end

    def []=(key, val)
      sync(:EX){ super }
    end
  end
end



if $0 == __FILE__
  iteration = 0

  loop do
    n = 42
    hash = Cash.new

    threads =
      Array.new(10) {
        Thread.new do
          Thread.current.abort_on_exception = true
          n.times do |key|
            hash[key] = key
            raise "#{ key }=nil" if hash[key].nil?
          end
        end
      }

    threads.map{|thread| thread.join}

    puts "THREADSAFE: #{ iteration += 1 }"
  end
end

Yehuda, ich denke, Sie erwähnten ivar Einstellung war atomic? Was ist mit einem einfachen kopieren-und tauschen dann?

require 'thread'

class ConcurrentHash
  def initialize
    @reader, @writer = {}, {}
    @lock = Mutex.new
  end

  def [](key)
    @reader[key]
  end

  def []=(key, value)
    @lock.synchronize {
      @writer[key] = value
      @reader, @writer = @writer, @reader
      @writer[key] = value
    }
  end
end

Könnte dies ein Fall für das hamster gem

Hamster implementiert Hash-Array Abgebildet Versucht (HAMT), sowie einige andere persistente Datenstrukturen, in reinem Ruby.

Persistente Datenstrukturen sind unveränderlich, und anstatt mutiert (verändert) die Struktur, wie durch hinzufügen oder ersetzen eines Schlüssel-Wert-paar in einem Hash, der Sie, statt zurück eine neue Daten-Struktur enthält die änderung. Der trick, mit dem beständigen unveränderlichen Datenstrukturen, ist, dass die neu zurückgegebenen Daten-Struktur wieder verwendet, wie viel von den Vorgänger als möglich.

Ich denke, zu implementieren, über hamster, die Sie verwenden würden, Ihre veränderliche hash-wrapper, die er alle liest, um den aktuellen Wert der persistenten immutable hash (dh, sollte schnell sein), während die Bewachung schreibt alles mit einer mutex und austauschen, um den neuen Wert der persistenten immutable hash nach dem schreiben.

Beispiel:

require 'hamster'
require 'hamster/experimental/mutable_hash'    
hsh = Hamster.mutable_hash(:name => "Simon", :gender => :male)

# reading goes directly to hash
puts hsh[:name] # Simon

# writing is actually swapping to new value of underlying persistent data structure
hsh.put(:name, "Joe")
puts hsh[:name] # Joe

So, lassen Sie uns dieses für ein ähnliches problem beschrieben:

(gist hier)

require 'hamster'
require 'hamster/experimental/mutable_hash'

# a bunch of threads with a read/write ratio of 10:1
num_threads = 100
num_reads_per_write = 10
num_loops = 100 
hsh = Hamster.mutable_hash

puts RUBY_DESCRIPTION
puts "#{num_threads} threads x #{num_loops} loops, #{num_reads_per_write}:1 R/W ratio"

t0 = Time.now
Thread.abort_on_exception = true
threads = (0...num_threads).map do |n|
  Thread.new do
    write_key = n % num_reads_per_write
    read_keys = (0...num_reads_per_write).to_a.shuffle # random order
    last_read = nil

    num_loops.times do
      read_keys.each do |k|
        # Reads
        last_read = hsh[k]

        Thread.pass

        # Atomic increments in the correct ratio to reads
        hsh.put(k) { |v| (v || 0) + 1 } if k == write_key
      end
    end
  end
end

threads.map { |t| t.join }
t1 = Time.now

puts "Error in keys" unless (0...num_reads_per_write).to_a == hsh.keys.sort.to_a
puts "Error in values" unless hsh.values.all? { |v| v == (num_loops * num_threads) / num_reads_per_write }
puts "Time elapsed: #{t1 - t0} s"

Ich bin immer folgende Ausgänge:

ruby 1.9.2p320 (2012-04-20 revision 35421) [x86_64-linux]
100 threads x 100 loops, 10:1 R/W ratio
Time elapsed: 5.763414627 s

jruby 1.7.0 (1.9.3p203) 2012-10-22 ff1ebbe on Java HotSpot(TM) 64-Bit Server VM 1.6.0_26-b03 [linux-amd64]
100 threads x 100 loops, 10:1 R/W ratio
Time elapsed: 1.697 s

Was haltet Ihr von dieser?

Diese Lösung ist ähnlich, wie könnte man dies lösen, in Scala oder Clojure, obwohl in diesen Sprachen würde man eher mit software transactional memory mit low-level-CPU-Unterstützung für die Atomare compare-and-swap-Operationen, die implementiert sind.

Bearbeiten: Es ist erwähnenswert, dass ein Grund, warum der hamster Umsetzung ist schnell ist, dass es verfügt über eine lock-free Lesen Pfad. Bitte Antworten Sie in den Kommentaren, wenn Sie Fragen haben, oder wie es funktioniert.

Nicht getestet, und einen naiven stechen um die Optimierung für liest. Es wird davon ausgegangen, dass die meisten der Zeit, wird der Wert nicht gesperrt werden. Wenn Sie es ist, die enge Schleife wird versuchen, bis es ist. Ich legte Thread.critical dort zu helfen, sicherzustellen, dass die gelesenen threads nicht ausgeführt werden, bis das schreiben beendet ist. Nicht sicher, ob der kritische Teil ist gebraucht, es hängt wirklich davon ab, wie gelesen-schwere meinst du, so einige benchmarking ist in Ordnung.

class ConcurrentHash < Hash

  def initialize(*args)
    @semaphore = Mutex.new
    super
  end

  def []=(k,v)
    begin
      old_crit = Thread.critical
      Thread.critical = true unless old_crit
      @semaphore.synchronize { super }
    ensure
      Thread.critical = old_crit
    end
  end

  def [](k)
    while(true)
      return super unless @semaphore.locked?
    end
  end

end

Gibt es vielleicht ein paar anderen lese-Methoden, müssen Sie die @semaphore sperren, ich weiß nicht, ob alles andere ist implementiert #[].

Leider kann ich nicht hinzufügen, einen Kommentar zu Michael Sofaer Antwort, wo er eingeführt: Klasse RWLock und Klasse LockedHash mit @reader_count etc. (die haben nicht genug karma noch)

Dass die Lösung nicht funktioniert. Gibt es einen Fehler:
in `unlock': Versuch zum entsperren eines mutex, die nicht gesperrt ist (ThreadError)

Aufgrund der logischen Fehler: wenn es Zeit zu entsperren Dinge wieder entsperren, passiert 1 extra Zeit (wegen fehlender überprüfen my_block?(). Stattdessen verstopfte es auch, wenn die Entsperrung nicht notwendig war "mein block") und so 2. freischalten auf schon freigeschaltet stumm wirft eine exception. (Ich werde einfügen vollständigen code, wie Sie diesen Fehler zu reproduzieren, die am Ende von diesem post).

Auch Michael erwähnt, "die jede Methode ist unsicher (können Mutationen durch andere threads während einer iteration)" das war wichtig für mich, so dass ich am Ende mit dieser vereinfachten Lösung, die funktioniert für alle meine Anwendungsfälle und es einfach sperren mutex auf jedem Aufruf alle hash-Methode aufgerufen, wenn aus dem anderen thread (Anrufe aus dem gleichen thread, in dessen Besitz sich das Schloss nicht blockieren, um deadlocks zu vermeiden):

#
# This TrulyThreadSafeHash works!
#
# Note if one thread iterating the hash by #each method
# then the hash will be locked for all other threads (they will not be 
# able to even read from it)
#
class TrulyThreadSafeHash
  def initialize
    @mutex = Mutex.new
    @hash = Hash.new
  end

  def method_missing(method_sym, *arguments, &block)

    if !@mutex.owned?  # Returns true if this lock is currently held by current thread
        # We're trying to lock only if mutex is not owned by the current thread (is not locked or is locked by some other thread).
        # Following call will be blocking if mutex locked by other thread:
        @mutex.synchronize{
            return lambda{@hash.send(method_sym,*arguments, &block)}.call
        }
    end

    # We already own the lock (from current thread perspective).
    # We don't even check if @hash.respond_to?(method_sym), let's make Hash
    # respond properly on all calls (including bad calls (example: wrong method names))
    lambda{@hash.send(method_sym,*arguments, &block)}.call
  end

  # since we're tyring to mimic Hash we'll pretend to respond as Hash would
  def self.respond_to?(method_sym, include_private = false)
    Hash.respond_to(method_sym, include_private)
  end

  # override Object's to_s because our method_missing won't be called for to_s
  def to_s(*arguments)
      @mutex.synchronize{
        return @hash.to_s
      }
  end

  # And for those, who want to run extra mile:
  # to make our class json-friendly we shoud require 'json' and uncomment this:
  #def to_json(*options)
  #    @mutex.synchronize{
  #        return @hash.to_json(*options)
  #    }
  #end

end

Und nun das vollständige Beispiel zu veranschaulichen bzw. zu reproduzieren, den Fehler der doppelten Erschließung in Michael Sofaer Lösung:

#!/usr/bin/env ruby

# ======= unchanged copy-paste part from Michael Sofaer answer (begin) =======

class LockedHash
  def initialize
    @hash = Hash.new
    @lock = ThreadAwareLock.new()
    @reader_count = 0
  end

  def [](key)
    @lock.lock_read
    ret = @hash[key]
    @lock.unlock_read
    ret
  end

  def []=(key, value)
    @lock.lock_write
    @hash[key] = value
    @lock.unlock_write
  end

  def method_missing(method_sym, *arguments, &block)
    if @hash.respond_to? method_sym
      @lock.lock_block
      val = lambda{@hash.send(method_sym,*arguments, &block)}.call
      @lock.unlock_block
      return val
    end
    super
  end
end



class RWLock
  def initialize
    @outer = Mutex.new
    @inner = Mutex.new
    @reader_count = 0
  end
  def lock_read
    @outer.synchronize{@inner.synchronize{@reader_count += 1}}
  end
  def unlock_read
    @inner.synchronize{@reader_count -= 1}
  end
  def lock_write
    @outer.lock
    while @reader_count > 0 ;end
  end
  def unlock_write
    @outer.unlock
  end
end

class ThreadAwareLock < RWLock
  def initialize
    @owner = nil
    super
  end
  def lock_block
    lock_write
    @owner = Thread.current.object_id
  end
  def unlock_block
    @owner = nil
    unlock_write
  end
  def lock_read
    super unless my_block?
  end
  def unlock_read
    super unless my_block?
  end
  def lock_write
    super unless my_block?
  end
  def unlock_write
    super unless my_block?
  end
  def my_block?
    @owner == Thread.current.object_id
  end
end

# ======= unchanged copy-paste part from Michael Sofaer answer (end) =======


# global hash object, which will be 'shared' across threads
$h = LockedHash.new

# hash_reader is just iterating through the 'shared' hash $h
# and prints specified delimeter (capitalized when last hash item read)
def hash_reader(delim)
    loop{
        count = 0
        $h.each{
            count += 1
            if count != $h.size
                $stderr.print delim
            else
                $stderr.puts delim.upcase
            end
        }
    }
end

# fill hash with 10 items
10.times{|i|
    $h[i] = i
}

# create a thread which will read $h hash
t1 = Thread.new(){
    hash_reader("o")
}

t1.join  # will never happen, but for completeness

die gibt die folgende Fehlermeldung:

./LockedHash_fails_to_unlock.rb
oooooooooO
./LockedHash_fails_to_unlock.rb:55:in `unlock': Attempt to unlock a mutex which is not locked (ThreadError)
        from ./LockedHash_fails_to_unlock.rb:55:in `unlock_write'
        from ./LockedHash_fails_to_unlock.rb:82:in `unlock_write'
        from ./LockedHash_fails_to_unlock.rb:70:in `unlock_block'
        from ./LockedHash_fails_to_unlock.rb:29:in `method_missing'
        from ./LockedHash_fails_to_unlock.rb:100:in `block in hash_reader'
        from ./LockedHash_fails_to_unlock.rb:98:in `loop'
        from ./LockedHash_fails_to_unlock.rb:98:in `hash_reader'
        from ./LockedHash_fails_to_unlock.rb:119:in `block in <main>'