Class: ODBA::Cache
- Includes:
- DRb::DRbUndumped, Singleton
- Defined in:
- lib/odba/cache.rb
Constant Summary collapse
- CLEANER_PRIORITY =
:nodoc:
0
- CLEANING_INTERVAL =
:nodoc:
5
- LOCK_FILE =
File lock exclusive control between processes, not threads, to create safely a new odba_id Sometimes several update jobs (processes) to the same database at the same time
'/tmp/lockfile'
- COUNT_FILE =
'/tmp/count'
- @@receiver_name =
RUBY_VERSION >= '1.9' ? :@receiver : '@receiver'
Instance Attribute Summary collapse
-
#cleaner_step ⇒ Object
Returns the value of attribute cleaner_step.
-
#debug ⇒ Object
Returns the value of attribute debug.
-
#destroy_age ⇒ Object
Returns the value of attribute destroy_age.
-
#file_lock ⇒ Object
Returns the value of attribute file_lock.
-
#retire_age ⇒ Object
Returns the value of attribute retire_age.
Instance Method Summary collapse
-
#_clean(retire_time, holder, offset) ⇒ Object
:nodoc:.
-
#bulk_fetch(bulk_fetch_ids, odba_caller) ⇒ Object
Returns all objects designated by bulk_fetch_ids and registers odba_caller for each of them.
-
#bulk_restore(rows, odba_caller = nil) ⇒ Object
:nodoc:.
-
#clean ⇒ Object
:nodoc:.
-
#clean_prefetched(flag = true) ⇒ Object
overrides the ODBA_PREFETCH constant and @odba_prefetch instance variable in Persistable.
-
#clear ⇒ Object
:nodoc:.
-
#count(klass) ⇒ Object
Get number of instances of a class.
-
#create_deferred_indices(drop_existing = false) ⇒ Object
Creates or recreates automatically defined indices.
-
#create_index(index_definition, origin_module = Object) ⇒ Object
Creates a new index according to IndexDefinition.
-
#delete(odba_object) ⇒ Object
Permanently deletes object from the database and deconnects all connected Persistables.
- #delete_cache_entry(key) ⇒ Object
-
#delete_index_element(odba_object) ⇒ Object
:nodoc:.
-
#drop_index(index_name) ⇒ Object
Permanently deletes the index named index_name.
-
#drop_indices ⇒ Object
:nodoc:.
-
#ensure_index_deferred(index_definition) ⇒ Object
Queue an index for creation by #setup.
-
#extent(klass, odba_caller = nil) ⇒ Object
Get all instances of a class (- a limited extent).
-
#fetch(odba_id, odba_caller = nil) ⇒ Object
Fetch a Persistable identified by odba_id.
-
#fetch_cache_entry(odba_id_or_name) ⇒ Object
:nodoc:.
-
#fetch_collection(odba_obj) ⇒ Object
:nodoc:.
-
#fetch_collection_element(odba_id, key) ⇒ Object
:nodoc:.
-
#fetch_named(name, odba_caller, &block) ⇒ Object
:nodoc:.
-
#fetch_or_do(obj_id, odba_caller, &block) ⇒ Object
:nodoc:.
-
#fetch_or_restore(odba_id, dump, odba_caller) ⇒ Object
:nodoc:.
- #fill_index(index_name, targets) ⇒ Object
-
#include?(odba_id) ⇒ Boolean
Checks wether the object identified by odba_id has been loaded.
- #index_keys(index_name, length = nil) ⇒ Object
- #index_matches(index_name, substring, limit = nil, offset = 0) ⇒ Object
-
#indices ⇒ Object
Returns a Hash-table containing all stored indices.
-
#initialize ⇒ Cache
constructor
:nodoc:.
- #invalidate(odba_id) ⇒ Object
- #invalidate!(*odba_ids) ⇒ Object
- #lock(dbname) ⇒ Object
- #new_id(dbname, odba_storage) ⇒ Object
-
#next_id ⇒ Object
Returns the next valid odba_id.
-
#prefetch ⇒ Object
Use this to load all prefetchable Persistables from the db at once.
-
#print_stats ⇒ Object
prints loading statistics to $stdout.
-
#register_peer(peer) ⇒ Object
Register a peer that has access to the same DB backend.
-
#reserve_next_id(id) ⇒ Object
Reserve an id with all registered peers.
-
#reset_stats ⇒ Object
Clears the loading statistics.
-
#retrieve_from_index(index_name, search_term, meta = nil) ⇒ Object
Find objects in an index.
-
#setup ⇒ Object
Create necessary DB-Structure / other storage-setup.
-
#size ⇒ Object
Returns the total number of cached objects.
-
#start_cleaner ⇒ Object
:nodoc:.
-
#store(object) ⇒ Object
Store a Persistable object in the database.
-
#store_cache_entry(odba_id, object, name = nil) ⇒ Object
:nodoc:.
-
#store_collection_elements(odba_obj) ⇒ Object
:nodoc:.
-
#store_object_connections(odba_id, target_ids) ⇒ Object
:nodoc:.
-
#transaction(&block) ⇒ Object
Executes the block in a transaction.
-
#transaction_rollback ⇒ Object
:nodoc:.
-
#unregister_peer(peer) ⇒ Object
Unregister a peer.
-
#update_indices(odba_object) ⇒ Object
:nodoc:.
-
#update_references(target_ids, object) ⇒ Object
:nodoc:.
Constructor Details
#initialize ⇒ Cache
:nodoc:
23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
# File 'lib/odba/cache.rb', line 23 def initialize # :nodoc: if(self::class::CLEANING_INTERVAL > 0) start_cleaner end @retire_age = 300 @receiver = nil @cache_mutex = Mutex.new @deferred_indices = [] @fetched = Hash.new @prefetched = Hash.new @clean_prefetched = false @cleaner_offset = 0 @prefetched_offset = 0 @cleaner_step = 500 @loading_stats = {} @peers = [] @file_lock = false @debug ||= false # Setting @debug to true makes two unit test fail! end |
Instance Attribute Details
#cleaner_step ⇒ Object
Returns the value of attribute cleaner_step.
22 23 24 |
# File 'lib/odba/cache.rb', line 22 def cleaner_step @cleaner_step end |
#debug ⇒ Object
Returns the value of attribute debug.
22 23 24 |
# File 'lib/odba/cache.rb', line 22 def debug @debug end |
#destroy_age ⇒ Object
Returns the value of attribute destroy_age.
22 23 24 |
# File 'lib/odba/cache.rb', line 22 def destroy_age @destroy_age end |
#file_lock ⇒ Object
Returns the value of attribute file_lock.
22 23 24 |
# File 'lib/odba/cache.rb', line 22 def file_lock @file_lock end |
#retire_age ⇒ Object
Returns the value of attribute retire_age.
22 23 24 |
# File 'lib/odba/cache.rb', line 22 def retire_age @retire_age end |
Instance Method Details
#_clean(retire_time, holder, offset) ⇒ Object
:nodoc:
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
# File 'lib/odba/cache.rb', line 97 def _clean(retire_time, holder, offset) # :nodoc: if(offset > holder.size) offset = 0 end counter = 0 cutoff = offset + @cleaner_step @cache_mutex.synchronize { holder.each_value { |value| counter += 1 if(counter > offset && value.odba_old?(retire_time)) value.odba_retire && @cleaned += 1 end return cutoff if(counter > cutoff) } } cutoff # every once in a while we'll get a 'hash modified during iteration'-Error. # not to worry, we'll just try again later. rescue StandardError offset end |
#bulk_fetch(bulk_fetch_ids, odba_caller) ⇒ Object
Returns all objects designated by bulk_fetch_ids and registers odba_caller for each of them. Objects which are not yet loaded are loaded from ODBA#storage.
45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
# File 'lib/odba/cache.rb', line 45 def bulk_fetch(bulk_fetch_ids, odba_caller) instances = [] loaded_ids = [] bulk_fetch_ids.each { |id| if(entry = fetch_cache_entry(id)) entry.odba_add_reference(odba_caller) instances.push(entry.odba_object) loaded_ids.push(id) end } bulk_fetch_ids -= loaded_ids unless(bulk_fetch_ids.empty?) rows = ODBA.storage.bulk_restore(bulk_fetch_ids) instances += bulk_restore(rows, odba_caller) end instances end |
#bulk_restore(rows, odba_caller = nil) ⇒ Object
:nodoc:
62 63 64 65 66 67 68 69 70 71 |
# File 'lib/odba/cache.rb', line 62 def bulk_restore(rows, odba_caller = nil) # :nodoc: retrieved_objects= [] rows.each { |row| obj_id = row.at(0) dump = row.at(1) odba_obj = fetch_or_restore(obj_id, dump, odba_caller) retrieved_objects.push(odba_obj) } retrieved_objects end |
#clean ⇒ Object
:nodoc:
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
# File 'lib/odba/cache.rb', line 72 def clean # :nodoc: now = Time.now start = Time.now if(@debug) @cleaned = 0 if(@debug) puts "starting cleaning cycle" $stdout.flush end retire_horizon = now - @retire_age @cleaner_offset = _clean(retire_horizon, @fetched, @cleaner_offset) if(@clean_prefetched) @prefetched_offset = _clean(retire_horizon, @prefetched, @prefetched_offset) end if(@debug) puts "cleaned: #{@cleaned} objects in #{Time.now - start} seconds" puts "remaining objects in @fetched: #{@fetched.size}" puts "remaining objects in @prefetched: #{@prefetched.size}" mbytes = File.read("/proc/#{$$}/stat").split(' ').at(22).to_i / (2**20) GC.start puts "remaining objects in ObjectSpace: #{ObjectSpace.each_object {}}" puts "memory-usage: #{mbytes}MB" $stdout.flush end end |
#clean_prefetched(flag = true) ⇒ Object
overrides the ODBA_PREFETCH constant and @odba_prefetch instance variable in Persistable. Use this if a secondary client is more memory-bound than performance-bound.
121 122 123 124 125 |
# File 'lib/odba/cache.rb', line 121 def clean_prefetched(flag=true) if(@clean_prefetched = flag) clean end end |
#clear ⇒ Object
:nodoc:
126 127 128 129 |
# File 'lib/odba/cache.rb', line 126 def clear # :nodoc: @fetched.clear @prefetched.clear end |
#count(klass) ⇒ Object
Get number of instances of a class
221 222 223 |
# File 'lib/odba/cache.rb', line 221 def count(klass) ODBA.storage.extent_count(klass) end |
#create_deferred_indices(drop_existing = false) ⇒ Object
Creates or recreates automatically defined indices
131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
# File 'lib/odba/cache.rb', line 131 def create_deferred_indices(drop_existing = false) @deferred_indices.each { |definition| name = definition.index_name if(drop_existing && self.indices.include?(name)) drop_index(name) end unless(self.indices.include?(name)) index = create_index(definition) if(index.target_klass.respond_to?(:odba_extent)) index.fill(index.target_klass.odba_extent) end end } end |
#create_index(index_definition, origin_module = Object) ⇒ Object
Creates a new index according to IndexDefinition
146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
# File 'lib/odba/cache.rb', line 146 def create_index(index_definition, origin_module=Object) transaction { klass = if(index_definition.fulltext) FulltextIndex elsif(index_definition.resolve_search_term.is_a?(Hash)) ConditionIndex else Index end index = klass.new(index_definition, origin_module) indices.store(index_definition.index_name, index) indices.odba_store_unsaved index } end |
#delete(odba_object) ⇒ Object
Permanently deletes object from the database and deconnects all connected Persistables
163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
# File 'lib/odba/cache.rb', line 163 def delete(odba_object) odba_id = odba_object.odba_id name = odba_object.odba_name odba_object.odba_notify_observers(:delete, odba_id, odba_object.object_id) rows = ODBA.storage.retrieve_connected_objects(odba_id) rows.each { |row| id = row.first # Self-Referencing objects don't have to be resaved begin if(connected_object = fetch(id, nil)) connected_object.odba_cut_connection(odba_object) connected_object.odba_isolated_store end rescue OdbaError warn "OdbaError ### deleting #{odba_object.class}:#{odba_id}" warn " ### while looking for connected object #{id}" end } delete_cache_entry(odba_id) delete_cache_entry(name) ODBA.storage.delete_persistable(odba_id) delete_index_element(odba_object) odba_object end |
#delete_cache_entry(key) ⇒ Object
187 188 189 190 191 192 |
# File 'lib/odba/cache.rb', line 187 def delete_cache_entry(key) @cache_mutex.synchronize { @fetched.delete(key) @prefetched.delete(key) } end |
#delete_index_element(odba_object) ⇒ Object
:nodoc:
193 194 195 196 197 198 |
# File 'lib/odba/cache.rb', line 193 def delete_index_element(odba_object) # :nodoc: klass = odba_object.class indices.each_value { |index| index.delete(odba_object) } end |
#drop_index(index_name) ⇒ Object
Permanently deletes the index named index_name
200 201 202 203 204 205 |
# File 'lib/odba/cache.rb', line 200 def drop_index(index_name) transaction { ODBA.storage.drop_index(index_name) self.delete(self.indices[index_name]) } end |
#drop_indices ⇒ Object
:nodoc:
206 207 208 209 210 211 |
# File 'lib/odba/cache.rb', line 206 def drop_indices # :nodoc: keys = self.indices.keys keys.each{ |key| drop_index(key) } end |
#ensure_index_deferred(index_definition) ⇒ Object
Queue an index for creation by #setup
213 214 215 |
# File 'lib/odba/cache.rb', line 213 def ensure_index_deferred(index_definition) @deferred_indices.push(index_definition) end |
#extent(klass, odba_caller = nil) ⇒ Object
Get all instances of a class (- a limited extent)
217 218 219 |
# File 'lib/odba/cache.rb', line 217 def extent(klass, odba_caller=nil) bulk_fetch(ODBA.storage.extent_ids(klass), odba_caller) end |
#fetch(odba_id, odba_caller = nil) ⇒ Object
Fetch a Persistable identified by odba_id. Registers odba_caller with the CacheEntry. Loads the Persistable if it is not already loaded.
226 227 228 229 230 |
# File 'lib/odba/cache.rb', line 226 def fetch(odba_id, odba_caller=nil) fetch_or_do(odba_id, odba_caller) { load_object(odba_id, odba_caller) } end |
#fetch_cache_entry(odba_id_or_name) ⇒ Object
:nodoc:
231 232 233 |
# File 'lib/odba/cache.rb', line 231 def fetch_cache_entry(odba_id_or_name) # :nodoc: @prefetched[odba_id_or_name] || @fetched[odba_id_or_name] end |
#fetch_collection(odba_obj) ⇒ Object
:nodoc:
235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 |
# File 'lib/odba/cache.rb', line 235 def fetch_collection(odba_obj) # :nodoc: collection = [] bulk_fetch_ids = [] rows = ODBA.storage.restore_collection(odba_obj.odba_id) return collection if rows.empty? idx = 0 rows.each { |row| key = row[0].is_a?(Integer) ? row[0] : ODBA.marshaller.load(row[0]) value = row[1].is_a?(Integer) ? row[1] : ODBA.marshaller.load(row[1]) idx += 1 item = nil if([key, value].any? { |item| item.instance_variable_get(@@receiver_name) }) odba_id = odba_obj.odba_id warn "stub for #{item.class}:#{item.odba_id} was saved with receiver in collection of #{odba_obj.class}:#{odba_id}" warn "repair: remove [#{odba_id}, #{row[0]}, #{row[1].length}]" ODBA.storage.collection_remove(odba_id, row[0]) key = key.odba_isolated_stub key_dump = ODBA.marshaller.dump(key) value = value.odba_isolated_stub value_dump = ODBA.marshaller.dump(value) warn "repair: insert [#{odba_id}, #{key_dump}, #{value_dump.length}]" ODBA.storage.collection_store(odba_id, key_dump, value_dump) end bulk_fetch_ids.push(key.odba_id) bulk_fetch_ids.push(value.odba_id) collection.push([key, value]) } bulk_fetch_ids.compact! bulk_fetch_ids.uniq! bulk_fetch(bulk_fetch_ids, odba_obj) collection.each { |pair| pair.collect! { |item| if(item.is_a?(ODBA::Stub)) ## don't fetch: that may result in a conflict when storing. #fetch(item.odba_id, odba_obj) item.odba_container = odba_obj item elsif(ce = fetch_cache_entry(item.odba_id)) warn "collection loaded unstubbed object: #{item.odba_id}" ce.odba_add_reference(odba_obj) ce.odba_object else item end } } collection end |
#fetch_collection_element(odba_id, key) ⇒ Object
:nodoc:
283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 |
# File 'lib/odba/cache.rb', line 283 def fetch_collection_element(odba_id, key) # :nodoc: key_dump = ODBA.marshaller.dump(key.odba_isolated_stub) ## for backward-compatibility and robustness we only attempt ## to load if there was a dump stored in the collection table if(dump = ODBA.storage.collection_fetch(odba_id, key_dump)) item = ODBA.marshaller.load(dump) if(item.is_a?(ODBA::Stub)) fetch(item.odba_id) elsif(item.is_a?(ODBA::Persistable)) warn "collection_element was unstubbed object: #{item.odba_id}" fetch_or_restore(item.odba_id, dump, nil) else item end end end |
#fetch_named(name, odba_caller, &block) ⇒ Object
:nodoc:
299 300 301 302 303 304 305 306 307 308 309 310 311 |
# File 'lib/odba/cache.rb', line 299 def fetch_named(name, odba_caller, &block) # :nodoc: fetch_or_do(name, odba_caller) { dump = ODBA.storage.restore_named(name) if(dump.nil?) odba_obj = block.call odba_obj.odba_name = name odba_obj.odba_store(name) odba_obj else fetch_or_restore(name, dump, odba_caller) end } end |
#fetch_or_do(obj_id, odba_caller, &block) ⇒ Object
:nodoc:
312 313 314 315 316 317 318 319 |
# File 'lib/odba/cache.rb', line 312 def fetch_or_do(obj_id, odba_caller, &block) # :nodoc: if (cache_entry = fetch_cache_entry(obj_id)) && cache_entry._odba_object cache_entry.odba_add_reference(odba_caller) cache_entry.odba_object else block.call end end |
#fetch_or_restore(odba_id, dump, odba_caller) ⇒ Object
:nodoc:
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 |
# File 'lib/odba/cache.rb', line 320 def fetch_or_restore(odba_id, dump, odba_caller) # :nodoc: fetch_or_do(odba_id, odba_caller) { odba_obj, collection = restore(dump) @cache_mutex.synchronize { fetch_or_do(odba_id, odba_caller) { cache_entry = CacheEntry.new(odba_obj) cache_entry.odba_add_reference(odba_caller) hash = odba_obj.odba_prefetch? ? @prefetched : @fetched name = odba_obj.odba_name hash.store(odba_obj.odba_id, cache_entry) if name hash.store(name, cache_entry) end odba_obj } } } end |
#fill_index(index_name, targets) ⇒ Object
338 339 340 |
# File 'lib/odba/cache.rb', line 338 def fill_index(index_name, targets) self.indices[index_name].fill(targets) end |
#include?(odba_id) ⇒ Boolean
Checks wether the object identified by odba_id has been loaded.
342 343 344 |
# File 'lib/odba/cache.rb', line 342 def include?(odba_id) @fetched.include?(odba_id) || @prefetched.include?(odba_id) end |
#index_keys(index_name, length = nil) ⇒ Object
345 346 347 348 |
# File 'lib/odba/cache.rb', line 345 def index_keys(index_name, length=nil) index = indices.fetch(index_name) index.keys(length) end |
#index_matches(index_name, substring, limit = nil, offset = 0) ⇒ Object
349 350 351 352 |
# File 'lib/odba/cache.rb', line 349 def index_matches(index_name, substring, limit=nil, offset=0) index = indices.fetch(index_name) index.matches substring, limit, offset end |
#indices ⇒ Object
Returns a Hash-table containing all stored indices.
354 355 356 357 358 |
# File 'lib/odba/cache.rb', line 354 def indices @indices ||= fetch_named('__cache_server_indices__', self) { {} } end |
#invalidate(odba_id) ⇒ Object
359 360 361 362 363 364 |
# File 'lib/odba/cache.rb', line 359 def invalidate(odba_id) ## when finalizers are run, no other threads will be scheduled, # therefore we don't need to @cache_mutex.synchronize @fetched.delete odba_id @prefetched.delete odba_id end |
#invalidate!(*odba_ids) ⇒ Object
365 366 367 368 369 370 371 372 |
# File 'lib/odba/cache.rb', line 365 def invalidate!(*odba_ids) odba_ids.each do |odba_id| if entry = fetch_cache_entry(odba_id) entry.odba_retire :force => true end invalidate odba_id end end |
#lock(dbname) ⇒ Object
377 378 379 380 381 382 383 384 |
# File 'lib/odba/cache.rb', line 377 def lock(dbname) lock_file = LOCK_FILE + "." + dbname open(lock_file, 'a') do |st| st.flock(File::LOCK_EX) yield st.flock(File::LOCK_UN) end end |
#new_id(dbname, odba_storage) ⇒ Object
385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 |
# File 'lib/odba/cache.rb', line 385 def new_id(dbname, odba_storage) count_file = COUNT_FILE + "." + dbname count = nil lock(dbname) do unless File.exist?(count_file) open(count_file, "w") do |out| out.print odba_storage.max_id end end count = File.read(count_file).to_i count += 1 open(count_file, "w") do |out| out.print count end odba_storage.update_max_id(count) end count end |
#next_id ⇒ Object
Returns the next valid odba_id
404 405 406 407 408 409 410 411 412 413 414 415 416 417 |
# File 'lib/odba/cache.rb', line 404 def next_id if @file_lock dbname = ODBA.storage.instance_variable_get('@dbi').dbi_args.first.split(/:/).last id = new_id(dbname, ODBA.storage) else id = ODBA.storage.next_id end @peers.each do |peer| peer.reserve_next_id id rescue DRb::DRbError end id rescue OdbaDuplicateIdError retry end |
#prefetch ⇒ Object
Use this to load all prefetchable Persistables from the db at once
419 420 421 |
# File 'lib/odba/cache.rb', line 419 def prefetch bulk_restore(ODBA.storage.restore_prefetchable) end |
#print_stats ⇒ Object
prints loading statistics to $stdout
423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 |
# File 'lib/odba/cache.rb', line 423 def print_stats fmh = " %-20s | %10s | %5s | %6s | %6s | %6s | %-20s\n" fmt = " %-20s | %10.3f | %5i | %6.3f | %6.3f | %6.3f | %s\n" head = sprintf(fmh, "class", "total", "count", "min", "max", "avg", "callers") line = "-" * head.length puts line print head puts line @loading_stats.sort_by { |key, val| val[:total_time] }.reverse.each { |key, val| key = key.to_s if(key.length > 20) key = key[-20,20] end avg = val[:total_time] / val[:count] printf(fmt, key, val[:total_time], val[:count], val[:times].min, val[:times].max, avg, val[:callers].join(',')) } puts line $stdout.flush end |
#register_peer(peer) ⇒ Object
Register a peer that has access to the same DB backend
448 449 450 |
# File 'lib/odba/cache.rb', line 448 def register_peer peer @peers.push(peer).uniq! end |
#reserve_next_id(id) ⇒ Object
Reserve an id with all registered peers
452 453 454 |
# File 'lib/odba/cache.rb', line 452 def reserve_next_id id ODBA.storage.reserve_next_id id end |
#reset_stats ⇒ Object
Clears the loading statistics
456 457 458 |
# File 'lib/odba/cache.rb', line 456 def reset_stats @loading_stats.clear end |
#retrieve_from_index(index_name, search_term, meta = nil) ⇒ Object
Find objects in an index
460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 |
# File 'lib/odba/cache.rb', line 460 def retrieve_from_index(index_name, search_term, =nil) index = indices.fetch(index_name) ids = index.fetch_ids(search_term, ) if .respond_to?(:error_limit) && (limit = .error_limit) \ && (size = ids.size) > limit.to_i error = OdbaResultLimitError.new error.limit = limit error.size = size error.index = index_name error.search_term = search_term error. = raise error end bulk_fetch(ids, nil) end |
#setup ⇒ Object
Create necessary DB-Structure / other storage-setup
476 477 478 479 480 481 482 483 |
# File 'lib/odba/cache.rb', line 476 def setup ODBA.storage.setup self.indices.each_key { |index_name| ODBA.storage.ensure_target_id_index(index_name) } create_deferred_indices nil end |
#size ⇒ Object
Returns the total number of cached objects
485 486 487 |
# File 'lib/odba/cache.rb', line 485 def size @prefetched.size + @fetched.size end |
#start_cleaner ⇒ Object
:nodoc:
488 489 490 491 492 493 494 495 496 497 498 499 500 501 |
# File 'lib/odba/cache.rb', line 488 def start_cleaner # :nodoc: @cleaner = Thread.new { Thread.current.priority = self::class::CLEANER_PRIORITY loop { sleep(self::class::CLEANING_INTERVAL) begin clean rescue StandardError => e puts e puts e.backtrace end } } end |
#store(object) ⇒ Object
Store a Persistable object in the database
503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 |
# File 'lib/odba/cache.rb', line 503 def store(object) odba_id = object.odba_id name = object.odba_name object.odba_notify_observers(:store, odba_id, object.object_id) if(ids = Thread.current[:txids]) ids.unshift([odba_id,name]) end ## get target_ids before anything else target_ids = object.odba_target_ids changes = store_collection_elements(object) prefetchable = object.odba_prefetch? dump = object.odba_isolated_dump ODBA.storage.store(odba_id, dump, name, prefetchable, object.class) store_object_connections(odba_id, target_ids) update_references(target_ids, object) object = store_cache_entry(odba_id, object, name) update_indices(object) @peers.each do |peer| peer.invalidate! odba_id rescue DRb::DRbError end object end |
#store_cache_entry(odba_id, object, name = nil) ⇒ Object
:nodoc:
525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 |
# File 'lib/odba/cache.rb', line 525 def store_cache_entry(odba_id, object, name=nil) # :nodoc: @cache_mutex.synchronize { if cache_entry = fetch_cache_entry(odba_id) cache_entry.update object else hash = object.odba_prefetch? ? @prefetched : @fetched cache_entry = CacheEntry.new(object) hash.store(odba_id, cache_entry) unless(name.nil?) hash.store(name, cache_entry) end end cache_entry.odba_object } end |
#store_collection_elements(odba_obj) ⇒ Object
:nodoc:
540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 |
# File 'lib/odba/cache.rb', line 540 def store_collection_elements(odba_obj) # :nodoc: odba_id = odba_obj.odba_id collection = odba_obj.odba_collection.collect { |key, value| [ ODBA.marshaller.dump(key.odba_isolated_stub), ODBA.marshaller.dump(value.odba_isolated_stub) ] } old_collection = ODBA.storage.restore_collection(odba_id).collect { |row| [row[0], row [1]] } changes = (old_collection - collection).each { |key_dump, _| ODBA.storage.collection_remove(odba_id, key_dump) }.size changes + (collection - old_collection).each { |key_dump, value_dump| ODBA.storage.collection_store(odba_id, key_dump, value_dump) }.size end |
#store_object_connections(odba_id, target_ids) ⇒ Object
:nodoc:
556 557 558 |
# File 'lib/odba/cache.rb', line 556 def store_object_connections(odba_id, target_ids) # :nodoc: ODBA.storage.ensure_object_connections(odba_id, target_ids) end |
#transaction(&block) ⇒ Object
Executes the block in a transaction. If the transaction fails, all affected Persistable objects are reloaded from the db (which by then has also performed a rollback). Rollback is quite inefficient at this time.
562 563 564 565 566 567 568 569 570 |
# File 'lib/odba/cache.rb', line 562 def transaction(&block) Thread.current[:txids] = [] ODBA.storage.transaction(&block) rescue Exception => excp transaction_rollback raise excp ensure Thread.current[:txids] = nil end |
#transaction_rollback ⇒ Object
:nodoc:
571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 |
# File 'lib/odba/cache.rb', line 571 def transaction_rollback # :nodoc: if(ids = Thread.current[:txids]) ids.each { |id, name| if(entry = fetch_cache_entry(id)) if(dump = ODBA.storage.restore(id)) odba_obj, collection = restore(dump) entry.odba_replace!(odba_obj) else entry.odba_cut_connections! delete_cache_entry(id) delete_cache_entry(name) end end } end end |
#unregister_peer(peer) ⇒ Object
Unregister a peer
588 589 590 |
# File 'lib/odba/cache.rb', line 588 def unregister_peer peer @peers.delete peer end |
#update_indices(odba_object) ⇒ Object
:nodoc:
591 592 593 594 595 596 597 |
# File 'lib/odba/cache.rb', line 591 def update_indices(odba_object) # :nodoc: if(odba_object.odba_indexable?) indices.each { |index_name, index| index.update(odba_object) } end end |
#update_references(target_ids, object) ⇒ Object
:nodoc:
598 599 600 601 602 603 604 |
# File 'lib/odba/cache.rb', line 598 def update_references(target_ids, object) # :nodoc: target_ids.each { |odba_id| if(entry = fetch_cache_entry(odba_id)) entry.odba_add_reference(object) end } end |