From 9990a3ac1b89638c01398900e613987110bc3216 Mon Sep 17 00:00:00 2001 From: Luka Vandervelden Date: Wed, 15 Jul 2020 17:19:27 +0200 Subject: [PATCH 1/3] DODB.no_index, Storage#new_nilable_index. The APIs are still a bit experimental at this point, but the feature is very likely to stay. --- src/dodb.cr | 6 ++++++ src/dodb/index.cr | 9 ++++++++- src/dodb/no_index.cr | 8 ++++++++ 3 files changed, 22 insertions(+), 1 deletion(-) create mode 100644 src/dodb/no_index.cr diff --git a/src/dodb.cr b/src/dodb.cr index 2a34db6..72dabe4 100644 --- a/src/dodb.cr +++ b/src/dodb.cr @@ -76,6 +76,12 @@ class DODB::DataBase(V) end end + def new_nilable_index(name : String, &block : Proc(V, String | DODB::NoIndex)) + Index(V).new(self, @directory_name, name, block).tap do |indexer| + @indexers << indexer + end + end + def new_tags(name : String, &block : Proc(V, Array(String))) Tags(V).new(@directory_name, name, block).tap do |tags| @indexers << tags diff --git a/src/dodb/index.cr b/src/dodb/index.cr index 5c518f3..bb0ac0b 100644 --- a/src/dodb/index.cr +++ b/src/dodb/index.cr @@ -6,7 +6,7 @@ require "./indexer.cr" class DODB::Index(V) < DODB::Indexer(V) property name : String - property key_proc : Proc(V, String) + property key_proc : Proc(V, String | NoIndex) | Proc(V, String) getter storage_root : String @storage : DODB::DataBase(V) @@ -34,6 +34,8 @@ class DODB::Index(V) < DODB::Indexer(V) def index(key, value) index_key = key_proc.call value + return if index_key.is_a? NoIndex + symlink = file_path_index index_key Dir.mkdir_p ::File.dirname symlink @@ -49,6 +51,8 @@ class DODB::Index(V) < DODB::Indexer(V) def deindex(key, value) index_key = key_proc.call value + return if index_key.is_a? NoIndex + symlink = file_path_index index_key ::File.delete symlink @@ -105,6 +109,9 @@ class DODB::Index(V) < DODB::Indexer(V) # in case new_value hasn't changed its index def update(new_value : V) index = key_proc.call new_value + + raise Exception.new "new value is not indexable" if index.is_a? NoIndex + update index, new_value end diff --git a/src/dodb/no_index.cr b/src/dodb/no_index.cr new file mode 100644 index 0000000..37e7a1c --- /dev/null +++ b/src/dodb/no_index.cr @@ -0,0 +1,8 @@ + +class DODB::NoIndex +end + +module DODB + class_getter no_index = NoIndex.new +end + From 2dd42f3a3085d9c346668b956cd6c2c3e7f00874 Mon Sep 17 00:00:00 2001 From: Luka Vandervelden Date: Mon, 20 Jul 2020 14:23:10 +0200 Subject: [PATCH 2/3] Some tests and extra code for parallelization of DODB. --- spec/test.cr | 64 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/dodb.cr | 26 ++++++++++++++++----- 2 files changed, 84 insertions(+), 6 deletions(-) diff --git a/spec/test.cr b/spec/test.cr index 43863d1..31c5778 100644 --- a/spec/test.cr +++ b/spec/test.cr @@ -482,5 +482,69 @@ describe "DODB::DataBase" do end end end + + describe "parallel support" do + # Not sure how many forks would be safe in a test like that. + fork_count = 25 + entries_per_fork = 100 + + it "works for pushing values" do + db = DODB::SpecDataBase.new + + processes = [] of Process + + fork_count.times do |fork_id| + processes << Process.fork do + entries_per_fork.times do |entry_id| + db << Ship.new("entry-#{fork_id}-#{entry_id}", "???") + end + end + end + + processes.each &.wait + + dump = db.to_a + + dump.size.should eq fork_count * entries_per_fork + end + + it "works for updating values" do + db = DODB::SpecDataBase.new + db_entries_by_name = db.new_index "name", &.name + + # First pass, creating data. + processes = [] of Process + fork_count.times do |fork_id| + processes << Process.fork do + entries_per_fork.times do |entry_id| + db << Ship.new("entry-#{fork_id}-#{entry_id}", "???") + end + end + end + processes.each &.wait + + # Second pass, updating data. + processes = [] of Process + fork_count.times do |fork_id| + processes << Process.fork do + entries_per_fork.times do |entry_id| + db_entries_by_name.update Ship.new("entry-#{fork_id}-#{entry_id}", "???", tags: ["updated"]) + end + end + end + processes.each &.wait + + # Third pass, testing database content. + dump = db.to_a + + fork_count.times do |fork_id| + entries_per_fork.times do |entry_id| + entry = db_entries_by_name.get "entry-#{fork_id}-#{entry_id}" + + entry.tags.should eq ["updated"] + end + end + end + end end diff --git a/src/dodb.cr b/src/dodb.cr index 72dabe4..e5f1411 100644 --- a/src/dodb.cr +++ b/src/dodb.cr @@ -44,9 +44,9 @@ class DODB::DataBase(V) end end - def request_lock(name) + def request_lock(name, subname = nil) r = -1 - file_path = get_lock_file_path name + file_path = get_lock_file_path name, subname file_perms = 0o644 flags = LibC::O_EXCL | LibC::O_CREAT @@ -56,8 +56,8 @@ class DODB::DataBase(V) LibC.close r end - def release_lock(name) - File.delete get_lock_file_path name + def release_lock(name, subname = nil) + File.delete get_lock_file_path name, subname end ## @@ -107,11 +107,17 @@ class DODB::DataBase(V) end def <<(item : V) + request_lock "index" + index = last_index + 1 self[index] = item self.last_index = index + + release_lock "index" + + index # FIXME: Should we really return the internal key? end def []?(key : Int32) : V? @@ -160,6 +166,8 @@ class DODB::DataBase(V) end def pop + request_lock "index" + index = last_index # Some entries may have been removed. We’ll skip over those. @@ -178,6 +186,8 @@ class DODB::DataBase(V) last_index = index - 1 + release_lock "index" + poped end @@ -306,8 +316,12 @@ class DODB::DataBase(V) "#{@directory_name}/locks" end - private def get_lock_file_path(name : String) - "#{locks_directory}/#{name}.lock" + private def get_lock_file_path(name : String, subname : String? = nil) + if subname + "#{locks_directory}/#{name}-#{subname}.lock" # FIXME: Separator that causes less collisions? + else + "#{locks_directory}/#{name}.lock" + end end private def read(file_path : String) From 48cf3c23be67c5abde2f40d62a75e9441f0629cb Mon Sep 17 00:00:00 2001 From: Luka Vandervelden Date: Mon, 20 Jul 2020 15:04:17 +0200 Subject: [PATCH 3/3] Fixes issues related to parallel updates. --- spec/test.cr | 25 +++++++++++++++++++++++++ src/dodb/index.cr | 3 +++ 2 files changed, 28 insertions(+) diff --git a/spec/test.cr b/spec/test.cr index 31c5778..8371edf 100644 --- a/spec/test.cr +++ b/spec/test.cr @@ -545,6 +545,31 @@ describe "DODB::DataBase" do end end end + + it "does parallel-safe updates" do + db = DODB::SpecDataBase.new + db_entries_by_name = db.new_index "name", &.name + + # We’ll be storing an integer in the "klass" field, and incrementing + # it in forks in a second time. + db << Ship.new("test", "0") + + processes = [] of Process + fork_count.times do |fork_id| + processes << Process.fork do + entries_per_fork.times do |entry_id| + db_entries_by_name.safe_get "test" do |entry| + entry.klass = (entry.klass.to_i + 1).to_s + + db_entries_by_name.update "test", entry + end + end + end + end + processes.each &.wait + + db_entries_by_name.get("test").klass.should eq((fork_count * entries_per_fork).to_s) + end end end diff --git a/src/dodb/index.cr b/src/dodb/index.cr index bb0ac0b..6fb6c68 100644 --- a/src/dodb/index.cr +++ b/src/dodb/index.cr @@ -72,13 +72,16 @@ class DODB::Index(V) < DODB::Indexer(V) nil end + # FIXME: Unlock on exception. def safe_get(index : String) : Nil + @storage.request_lock @name, index internal_key = get_key(index).to_s @storage.request_lock internal_key yield get index @storage.release_lock internal_key + @storage.release_lock @name, index end def safe_get?(index : String, &block : Proc(V | Nil, Nil)) : Nil