diff --git a/shard.lock b/shard.lock index 1f84f7d..1095185 100644 --- a/shard.lock +++ b/shard.lock @@ -195,7 +195,7 @@ shards: placeos-resource: git: https://github.com/place-labs/resource.git - version: 3.1.3 + version: 3.2.0 pool: git: https://github.com/ysbaddaden/pool.git diff --git a/spec/mappings_spec.cr b/spec/mappings_spec.cr index 175e593..eaa68fe 100644 --- a/spec/mappings_spec.cr +++ b/spec/mappings_spec.cr @@ -33,6 +33,44 @@ module PlaceOS::Source end end + describe "system modules" do + it "keeps each system's modules apart" do + mappings = Mappings.new + mappings.set_system_modules("cs-1", { + "mod-a" => {name: "Display", control_system_id: "cs-1", index: 1}, + "mod-b" => {name: "Display", control_system_id: "cs-1", index: 2}, + }) + mappings.set_system_modules("cs-2", { + "mod-a" => {name: "Screen", control_system_id: "cs-2", index: 1}, + }) + + mappings.modules_in_systems({"cs-1"}).should eq Set{"mod-a", "mod-b"} + mappings.system_modules_for("cs-2").should eq({"mod-a" => {name: "Screen", control_system_id: "cs-2", index: 1}}) + + # replacing a system's modules leaves the other system's alone + mappings.set_system_modules("cs-1", { + "mod-b" => {name: "Display", control_system_id: "cs-1", index: 1}, + }) + mappings.modules_in_systems({"cs-1"}).should eq Set{"mod-b"} + mappings.read(&.system_modules["mod-a"]).should eq [{name: "Screen", control_system_id: "cs-2", index: 1}] + + mappings.remove_system_modules("cs-2") + mappings.read(&.system_modules.has_key?("mod-a")).should be_false + mappings.system_modules_for("cs-2").should be_empty + end + + it "forgets a deleted module in every system" do + mappings = Mappings.new + mappings.set_system_modules("cs-1", {"mod-a" => {name: "Display", control_system_id: "cs-1", index: 1}}) + mappings.set_system_modules("cs-2", {"mod-a" => {name: "Display", control_system_id: "cs-2", index: 1}}) + + mappings.write(&.unmap_module("mod-a")) + + mappings.modules_in_systems({"cs-1", "cs-2"}).should be_empty + mappings.read(&.system_module_ids).should be_empty + end + end + describe "hierarchy_zones" do it "gets scope zones for ControlSystem" do Model::ControlSystem.clear diff --git a/spec/published_spec.cr b/spec/published_spec.cr new file mode 100644 index 0000000..6b3bdcf --- /dev/null +++ b/spec/published_spec.cr @@ -0,0 +1,42 @@ +require "./spec_helper" + +module PlaceOS::Source + def self.published_status(module_id : String, control_system_id : String, status : String) : Mappings::Status + Mappings::Status.new( + status: status, + index: 1, + module_name: "Display", + module_id: module_id, + driver_id: "driver-1", + control_system_id: control_system_id, + zone_mapping: {"org" => "org-1"}, + ) + end + + describe Published do + it "finds what is published by system" do + published = Published.new + a_power = published_status("mod-a", "cs-1", "power") + a_input = published_status("mod-a", "cs-1", "input") + a_other = published_status("mod-a", "cs-2", "power") + b_power = published_status("mod-b", "cs-2", "power") + [a_power, a_input, a_other, b_power].each { |event| published.add(event) } + # adding twice is not counted twice + published.add(a_power) + + published.module_ids_in({"cs-1"}).should eq Set{"mod-a"} + published.module_ids_in({"cs-2"}).should eq Set{"mod-a", "mod-b"} + published.for_system("cs-1").to_set.should eq Set{a_power, a_input} + + published.remove(a_power) + published.module_ids_in({"cs-1"}).should eq Set{"mod-a"} + published.remove(a_input) + published.module_ids_in({"cs-1"}).should be_empty + published.for_system("cs-1").should be_empty + # removing what is not there changes nothing + published.remove(a_input) + published.module_ids_in({"cs-2"}).should eq Set{"mod-a", "mod-b"} + published.for_module("mod-a").should eq Set{a_other} + end + end +end diff --git a/spec/publishing/publish_metadata_spec.cr b/spec/publishing/publish_metadata_spec.cr index 972cb9c..6d0db2c 100644 --- a/spec/publishing/publish_metadata_spec.cr +++ b/spec/publishing/publish_metadata_spec.cr @@ -64,5 +64,18 @@ module PlaceOS::Source message.data.should eq Mappings::Metadata.new("hello", "org") message.payload.should eq payload.to_json end + + # the topic is the scope, so a second top-level zone would publish it twice + it "publishes once however many top-level zones there are" do + zones = Array.new(2) do + zone = Model::Generator.zone + zone.tags = Set{Mappings.scope} + zone + end + router = Dummy.new + router.publish_metadata(zones, MockModel.new(id: "hello", some_data: "edkh")) + + router.@publisher_managers.first.messages.size.should eq 1 + end end end diff --git a/spec/publishing/rename_spec.cr b/spec/publishing/rename_spec.cr index 24a37f5..3047619 100644 --- a/spec/publishing/rename_spec.cr +++ b/spec/publishing/rename_spec.cr @@ -11,7 +11,7 @@ module PlaceOS::Source def self.mapped(module_id : String, name : String, index : Int32, org : String) : Mappings state = Mappings::State.new state.drivers[module_id] = "driver-1234" - state.system_modules[module_id] = [{name: name, control_system_id: "cs-1", index: index}] + state.map_module(module_id, {name: name, control_system_id: "cs-1", index: index}) state.system_zones["cs-1"] = {"org" => org, "building" => "b", "level" => "l", "area" => "a"} Mappings.new(state) end diff --git a/spec/router/control_system_router_spec.cr b/spec/router/control_system_router_spec.cr index 5c98654..9e1e635 100644 --- a/spec/router/control_system_router_spec.cr +++ b/spec/router/control_system_router_spec.cr @@ -87,6 +87,32 @@ module PlaceOS::Source mapping[second.id.as(String)][:index].should eq 2 end + # the initial sync publishes everything, so the load reconciles nothing + it "reconciles a system changed after startup, not one loaded at startup" do + driver = Model::Generator.driver(module_name: "Display").save! + mod = Model::Generator.module(driver) + mod.custom_name = "Display" + mod.save! + other = Model::Generator.module(driver) + other.custom_name = "Display" + other.save! + cs = Model::Generator.control_system + cs.modules = [mod.id.as(String)] + cs.save! + mappings = Mappings.new + router = Router::ControlSystem.new(mappings, [] of PublisherManager) + + router.sync(cs.id.as(String), cs) + mappings.system_modules_for(cs.id.as(String)).keys.should eq [mod.id] + mappings.reconcile_pending?.should be_false + + cs.modules = [other.id.as(String), mod.id.as(String)] + cs.save! + router.sync(cs.id.as(String)) + mappings.system_modules_for(cs.id.as(String)).keys.to_set.should eq Set{mod.id, other.id} + mappings.take_reconcile.should eq Set{mod.id, other.id} + end + it "system_zones" do cs = Model::Generator.control_system cs.id = "cs-1245" diff --git a/spec/router/driver_router_spec.cr b/spec/router/driver_router_spec.cr index 7079337..5c5f171 100644 --- a/spec/router/driver_router_spec.cr +++ b/spec/router/driver_router_spec.cr @@ -21,5 +21,31 @@ module PlaceOS::Source drivers = mappings.read(&.drivers) drivers.should eq({"mod-keep" => "driver-other"}) end + + # while loading, scope is read from the mappings, not the database + it "publishes a loaded Driver's metadata once, if it has a Module in scope" do + state = Mappings::State.new + state.drivers["mod-a"] = "driver-scoped" + state.drivers["mod-b"] = "driver-scoped" + state.drivers["mod-c"] = "driver-unscoped" + state.map_module("mod-a", {name: "Display", control_system_id: "cs-1", index: 1}) + state.map_module("mod-b", {name: "Display", control_system_id: "cs-2", index: 1}) + state.map_module("mod-c", {name: "Display", control_system_id: "cs-3", index: 1}) + state.system_zones["cs-1"] = {Mappings.scope => "org-1", "building" => "b"} + state.system_zones["cs-2"] = {Mappings.scope => "org-2"} + state.system_zones["cs-3"] = {"building" => "b"} + manager = MockManager.new + router = Router::Driver.new(Mappings.new(state), [manager] of PublisherManager) + + scoped = Model::Generator.driver(module_name: "mock") + scoped.id = "driver-scoped" + unscoped = Model::Generator.driver(module_name: "mock") + unscoped.id = "driver-unscoped" + + router.process_resource(:created, scoped).should eq Resource::Result::Success + router.process_resource(:created, unscoped).should eq Resource::Result::Skipped + + manager.messages.map(&.data).should eq [Mappings::Metadata.new("driver-scoped")] + end end end diff --git a/spec/router/module_router_spec.cr b/spec/router/module_router_spec.cr index 826de8a..6b986e2 100644 --- a/spec/router/module_router_spec.cr +++ b/spec/router/module_router_spec.cr @@ -6,9 +6,7 @@ module PlaceOS::Source # the module name is a topic segment, so a rename moves the topic it "updates on changes to Module name" do state = Mappings::State.new - state.system_modules["mod-renamed"] = [ - {name: "Old Name", control_system_id: "cs-1", index: 1}, - ] + state.map_module("mod-renamed", {name: "Old Name", control_system_id: "cs-1", index: 1}) mappings = Mappings.new(state) mappings.set_system_modules("cs-1", { @@ -23,9 +21,7 @@ module PlaceOS::Source it "updates on changes to Module index" do state = Mappings::State.new - state.system_modules["mod-reindexed"] = [ - {name: "M'Odule", control_system_id: "cs-1", index: 1}, - ] + state.map_module("mod-reindexed", {name: "M'Odule", control_system_id: "cs-1", index: 1}) mappings = Mappings.new(state) mappings.set_system_modules("cs-1", { @@ -41,9 +37,7 @@ module PlaceOS::Source state = Mappings::State.new state.drivers["mod-gone"] = "driver-sns" state.drivers["mod-stays"] = "driver-sns" - state.system_modules["mod-gone"] = [ - {name: "M'Odule", control_system_id: "cs-1", index: 1}, - ] + state.map_module("mod-gone", {name: "M'Odule", control_system_id: "cs-1", index: 1}) mappings = Mappings.new(state) driver = Model::Generator.driver(module_name: "mock") @@ -75,6 +69,25 @@ module PlaceOS::Source mappings.@state.drivers["mod-1234"].should eq "driver-sns" end + # while loading, scope is read from the mappings, not the database + it "maps a loaded Module only if a system has it" do + driver = Model::Generator.driver(module_name: "mock") + driver.id = "driver-sns" + mapped = Model::Generator.module(driver) + mapped.id = "mod-mapped" + unmapped = Model::Generator.module(driver) + unmapped.id = "mod-unmapped" + state = Mappings::State.new + state.map_module("mod-mapped", {name: "Display", control_system_id: "cs-1", index: 1}) + mappings = Mappings.new(state) + + router = Router::Module.new(mappings) + router.process_resource(:created, mapped).should eq Resource::Result::Success + router.process_resource(:created, unmapped).should eq Resource::Result::Skipped + + mappings.read(&.drivers).should eq({"mod-mapped" => "driver-sns"}) + end + it "removes module_id reference on destroy" do driver = Model::Generator.driver(module_name: "mock") driver.id = "driver-sns" diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr index aa53be7..ec660ba 100644 --- a/spec/spec_helper.cr +++ b/spec/spec_helper.cr @@ -200,7 +200,7 @@ module PlaceOS::Source org_id = "org-donor", ) state = Mappings::State.new - state.system_modules[module_id] = [{name: module_name, control_system_id: control_system_id, index: index}] + state.map_module(module_id, {name: module_name, control_system_id: control_system_id, index: index}) state.drivers[module_id] = driver_id state.system_zones[control_system_id] = { "area" => area_id, diff --git a/src/source/mappings.cr b/src/source/mappings.cr index ca0022f..7123a64 100644 --- a/src/source/mappings.cr +++ b/src/source/mappings.cr @@ -45,7 +45,7 @@ module PlaceOS::Source ids = control_system_ids.to_set return if ids.empty? reconcile( - modules_in_systems(ids) | published.module_ids { |event| ids.includes?(event.control_system_id) } + modules_in_systems(ids) | published.module_ids_in(ids) ) end @@ -64,10 +64,9 @@ module PlaceOS::Source # Modules mapped into any of `control_system_ids` now def modules_in_systems(control_system_ids : Enumerable(String)) : Set(String) - ids = control_system_ids.to_set read do |state| - state.system_modules.each_with_object(Set(String).new) do |(module_id, mappings), found| - found << module_id if mappings.any? { |mapping| ids.includes?(mapping[:control_system_id]) } + control_system_ids.each_with_object(Set(String).new) do |control_system_id, found| + state.system_module_ids[control_system_id]?.try { |ids| found.concat(ids) } end end end @@ -218,10 +217,10 @@ module PlaceOS::Source def set_system_modules(control_system_id : String, system_modules : Hash(String, SystemModule)) write do |state| # Clear mappings of all references to control_system_id - remove_system_modules(control_system_id) + state.unmap_system(control_system_id) system_modules.each do |module_id, new_mapping| - state.system_modules[module_id] << new_mapping + state.map_module(module_id, new_mapping) end end end @@ -232,20 +231,19 @@ module PlaceOS::Source # its own, and `RWLock` allows nested reads def system_modules_for(control_system_id : String) : Hash(String, SystemModule) read do |state| - state.system_modules.each_with_object({} of String => SystemModule) do |(module_id, list), found| - list.each { |mapping| found[module_id] = mapping if mapping[:control_system_id] == control_system_id } + found = {} of String => SystemModule + state.system_module_ids[control_system_id]?.try &.each do |module_id| + state.system_modules[module_id]?.try &.each do |mapping| + found[module_id] = mapping if mapping[:control_system_id] == control_system_id + end end + found end end # Remove System Module Mappings def remove_system_modules(control_system_id : String) - # Clear mappings of all references to control_system_id - write do |state| - state.system_modules.transform_values! do |mappings| - mappings.reject! { |sys_mod| sys_mod[:control_system_id] == control_system_id } - end - end + write(&.unmap_system(control_system_id)) end # Mappings @@ -258,7 +256,18 @@ module PlaceOS::Source include JSON::Serializable # module_id => [{control_system_id: String, index: Int32}] - getter system_modules : Hash(String, Array(SystemModule)) = Hash(String, Array(SystemModule)).new { |h, k| h[k] = [] of SystemModule } + # + # NOTE:: change it through `map_module`, `unmap_system` and + # `unmap_module`, which keep `system_module_ids` in step with it + getter system_modules : Hash(String, Array(SystemModule)) = Hash(String, Array(SystemModule)).new + + # control_system_id => the module_ids mapped in it. + # + # The reverse of `system_modules`, so working on one system does not + # mean scanning every Module. Systems are mapped one at a time as the + # routers load, so a scan per system made startup quadratic + @[JSON::Field(ignore: true)] + getter system_module_ids : Hash(String, Set(String)) = Hash(String, Set(String)).new # module_id => driver_id getter drivers : Hash(String, String) = {} of String => String @@ -268,6 +277,30 @@ module PlaceOS::Source def initialize end + + def map_module(module_id : String, mapping : SystemModule) : Nil + (system_modules[module_id] ||= [] of SystemModule) << mapping + (system_module_ids[mapping[:control_system_id]] ||= Set(String).new) << module_id + end + + def unmap_system(control_system_id : String) : Nil + return unless module_ids = system_module_ids.delete(control_system_id) + module_ids.each do |module_id| + next unless list = system_modules[module_id]? + list.reject! { |mapping| mapping[:control_system_id] == control_system_id } + system_modules.delete(module_id) if list.empty? + end + end + + def unmap_module(module_id : String) : Nil + return unless list = system_modules.delete(module_id) + list.each do |mapping| + control_system_id = mapping[:control_system_id] + next unless ids = system_module_ids[control_system_id]? + ids.delete(module_id) + system_module_ids.delete(control_system_id) if ids.empty? + end + end end private getter mappings_lock : RWLock = RWLock.new diff --git a/src/source/published.cr b/src/source/published.cr index c16825d..096581c 100644 --- a/src/source/published.cr +++ b/src/source/published.cr @@ -16,17 +16,30 @@ module PlaceOS::Source # module_id => statuses published for it, one per system it is mapped in private getter events = Hash(String, Set(Mappings::Status)).new + # control_system_id => module_id => how many of its statuses are there. + # So a system can be looked up without scanning everything published + private getter by_system = Hash(String, Hash(String, Int32)).new + def add(event : Mappings::Status) : Nil lock.synchronize do - (events[event.module_id] ||= Set(Mappings::Status).new) << event + set = events[event.module_id] ||= Set(Mappings::Status).new + next if set.includes?(event) + set << event + counts = by_system[event.control_system_id] ||= Hash(String, Int32).new(0) + counts[event.module_id] += 1 end end def remove(event : Mappings::Status) : Nil lock.synchronize do - if set = events[event.module_id]? - set.delete(event) - events.delete(event.module_id) if set.empty? + next unless set = events[event.module_id]? + next unless set.delete(event) + events.delete(event.module_id) if set.empty? + + next unless counts = by_system[event.control_system_id]? + if (counts[event.module_id] -= 1) <= 0 + counts.delete(event.module_id) + by_system.delete(event.control_system_id) if counts.empty? end end end @@ -35,18 +48,24 @@ module PlaceOS::Source lock.synchronize { events[module_id]?.try(&.dup) || Set(Mappings::Status).new } end - # Modules with anything published that matches the block - def module_ids(& : Mappings::Status -> Bool) : Set(String) + # Modules with anything published in any of these systems + def module_ids_in(control_system_ids : Enumerable(String)) : Set(String) lock.synchronize do - events.each_with_object(Set(String).new) do |(module_id, set), ids| - ids << module_id if set.any? { |event| yield event } + control_system_ids.each_with_object(Set(String).new) do |control_system_id, ids| + by_system[control_system_id]?.try &.each_key { |module_id| ids << module_id } end end end - # Every event published, for comparing with what a broker holds - def all : Array(Mappings::Status) - lock.synchronize { events.values.flat_map(&.to_a) } + # Everything published in one system + def for_system(control_system_id : String) : Array(Mappings::Status) + lock.synchronize do + found = [] of Mappings::Status + by_system[control_system_id]?.try &.each_key do |module_id| + events[module_id]?.try &.each { |event| found << event if event.control_system_id == control_system_id } + end + found + end end end end diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr index 87d4d21..c47e2ae 100644 --- a/src/source/publishing/mqtt_publisher.cr +++ b/src/source/publishing/mqtt_publisher.cr @@ -196,14 +196,15 @@ module PlaceOS::Source Log.error(exception: e) { "error removing retained value for #{key}" } end - # Every status topic the broker currently retains, as a client connecting - # now would be handed them. + # The status topics the broker currently retains, as a client connecting + # now would be handed them, that `keep` returns true for. Each topic is + # passed to `keep` as it arrives, so only those wanted are held. # # The broker sends retained messages straight after it acknowledges the # subscription, so collection ends once `quiet` passes without one, or at # `limit` regardless. Only retained messages count: a live publication # arriving meanwhile is not part of what was there. - def retained_state_topics(quiet : Time::Span = 2.seconds, limit : Time::Span = 2.minutes) : Set(String) + def retained_state_topics(quiet : Time::Span = 2.seconds, limit : Time::Span = 2.minutes, &keep : String -> Bool) : Set(String) filter = File.join(MQTT_NAMESPACE, "+", "state", "#") topics = Set(String).new lock = Mutex.new @@ -211,8 +212,9 @@ module PlaceOS::Source client.subscribe(filter) do |topic, payload, retained| next unless retained && !payload.empty? + wanted = keep.call(topic) lock.synchronize do - topics << topic + topics << topic if wanted last = Time.instant end end @@ -285,6 +287,15 @@ module PlaceOS::Source ) end + # Where the system id sits in a state topic: after the namespace, the + # scope, `state` and the rest of the hierarchy + STATE_TOPIC_SYSTEM_SEGMENT = MQTT_NAMESPACE.split('/').size + 2 + (HIERARCHY.size - 1) + + # The system a state topic belongs to + def self.state_topic_system_id(topic : String) : String? + topic.split('/')[STATE_TOPIC_SYSTEM_SEGMENT]? + end + def self.extract_zone_hierarchy(zone_mapping : Hash(String, String)) HIERARCHY.map { |key| zone_mapping[key]? || "_" } end diff --git a/src/source/publishing/publish_metadata.cr b/src/source/publishing/publish_metadata.cr index b6f3f35..7f6be7c 100644 --- a/src/source/publishing/publish_metadata.cr +++ b/src/source/publishing/publish_metadata.cr @@ -8,20 +8,35 @@ module PlaceOS::Source module PublishMetadata(Model) abstract def publisher_managers : Array(PublisherManager) + # Publish model metadata if any of `zones` is top-level. + # + # Once, however many of them are: the topic is the scope, not the zone + def publish_metadata(zones : Enumerable(::PlaceOS::Model::Zone), model : Model) + publish_metadata(model) if zones.any? { |zone| Mappings.hierarchy_tag?(zone) == Mappings.scope } + end + # Publish model metadata # # Only models in the top-level hiearchy are published def publish_metadata(zone : ::PlaceOS::Model::Zone, model : Model) - if Mappings.hierarchy_tag?(zone) == Mappings.scope - message = Publisher::Message.new( - data: Mappings::Metadata.new(model.id.as(String)), - payload: model.destroyed? ? nil : model.to_json, - timestamp: Time.utc - ) + publish_metadata({zone}, model) + end - # Fire off broadcast - spawn { publisher_managers.each &.broadcast(message) } - end + # Publish model metadata, for a model known to be in the top-level hierarchy + # + # NOTE:: broadcast here rather than from a fiber of its own. The publisher + # queues are bounded, and as the routers load, every system and zone is + # published at once: a fiber each held every one of those models' JSON in + # memory while it waited its turn. Waiting here holds one + def publish_metadata(model : Model) + message = Publisher::Message.new( + data: Mappings::Metadata.new(model.id.as(String)), + payload: model.destroyed? ? nil : model.to_json, + timestamp: Time.utc + ) + publisher_managers.each &.broadcast(message) + rescue error + Log.warn(exception: error) { "publishing metadata for #{model.id}" } end end end diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr index 77d0cfa..6df1c37 100644 --- a/src/source/router/control_system_router.cr +++ b/src/source/router/control_system_router.cr @@ -35,7 +35,11 @@ module PlaceOS::Source::Router private getter sync_lock = Mutex.new def handle_create(control_system : PlaceOS::Model::ControlSystem) : Resource::Result - sync(control_system.id.as(String)) + # While the routers load, events are only the rows just read, one per + # system, and change events wait until loading is done. So the row can + # be used as it is, rather than read again for every system + loaded = control_system unless startup_finished? + sync(control_system.id.as(String), loaded) end def handle_update(control_system : PlaceOS::Model::ControlSystem) : Resource::Result @@ -53,12 +57,24 @@ module PlaceOS::Source::Router # system's modules of the same name, so a change to either moves topics. # Reconciling removes the old ones and publishes current values at the new # ones — which also covers a module whose state was written before the - # system mapped it, and so was published nowhere - def sync(control_system_id : String) : Resource::Result + # system mapped it, and so was published nowhere. + # + # `loaded` is the system as the initial load read it. Nothing is + # reconciled for it: status events have not started, nothing is published + # yet to take away, and their initial sync publishes every Module against + # the mappings built here. Queueing them all would read the state of every + # Module a second time, in one batch + # + # `loaded_zones` are the system's hierarchy zones, when already read with it + def sync( + control_system_id : String, + loaded : PlaceOS::Model::ControlSystem? = nil, + loaded_zones : Array(PlaceOS::Model::Zone)? = nil, + ) : Resource::Result sync_lock.synchronize do - current = PlaceOS::Model::ControlSystem.find?(control_system_id) + current = loaded || PlaceOS::Model::ControlSystem.find?(control_system_id) - zones = current.try { |cs| Router::ControlSystem.system_zones(cs) } + zones = current.try { |cs| Router::ControlSystem.system_zones(cs, loaded ? loaded_zones : nil) } modules = current.try { |cs| Router::ControlSystem.system_modules(cs) } changed = mappings.read do |state| @@ -74,7 +90,7 @@ module PlaceOS::Source::Router mappings.remove_system_modules(control_system_id) end - mappings.reconcile_systems({control_system_id}) if changed + mappings.reconcile_systems({control_system_id}) if changed && loaded.nil? end Resource::Result::Success @@ -90,11 +106,14 @@ module PlaceOS::Source::Router # while it was in scope has to go return Resource::Result::Skipped if hierarchy_zones.empty? && action.created? - hierarchy_zones.each { |zone| publish_metadata(zone, control_system) } + publish_metadata(hierarchy_zones, control_system) case action in .created? - handle_create(control_system) + # while loading, the zones just read are the system's hierarchy, so + # they are not read again to map it + return handle_create(control_system) if startup_finished? + sync(control_system.id.as(String), control_system, hierarchy_zones) in .updated? handle_update(control_system) in .deleted? diff --git a/src/source/router/driver_router.cr b/src/source/router/driver_router.cr index b5891d7..cf89ccf 100644 --- a/src/source/router/driver_router.cr +++ b/src/source/router/driver_router.cr @@ -24,13 +24,18 @@ module PlaceOS::Source::Router driver = resource driver_id = driver.id.as(String) + if action.created? && !startup_finished? + # loading: only its metadata is published, and only if it is in scope + return Resource::Result::Skipped unless loaded_in_scope.includes?(driver_id) + publish_metadata(driver) + return Resource::Result::Success + end + hierarchy_zones = Mappings.hierarchy_zones(driver) return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted? - hierarchy_zones.each do |zone| - publish_metadata(zone, driver) - end + publish_metadata(hierarchy_zones, driver) if action.deleted? # driver_id is a topic segment, so everything published by a Module of @@ -48,5 +53,32 @@ module PlaceOS::Source::Router Resource::Result::Success end + + def start(timeout : Time::Span = LOAD_TIMEOUT) : self + @loaded_in_scope = nil + super + end + + @loaded_in_scope : Set(String)? = nil + private getter loaded_in_scope_lock = Mutex.new + + # The Drivers with a Module in a system under a top-level zone, from the + # mappings, which hold every system and Module by the time Drivers load. + # + # Worked out once, for loading. Asking the database meant reading every + # Module of each Driver, then each Module's systems and their zones + private def loaded_in_scope : Set(String) + loaded_in_scope_lock.synchronize do + @loaded_in_scope ||= mappings.read do |state| + state.drivers.each_with_object(Set(String).new) do |(module_id, driver_id), found| + next if found.includes?(driver_id) + scoped = state.system_modules[module_id]?.try &.any? do |mapping| + state.system_zones[mapping[:control_system_id]]?.try(&.has_key?(Mappings.scope)) + end + found << driver_id if scoped + end + end + end + end end end diff --git a/src/source/router/module_router.cr b/src/source/router/module_router.cr index 53aa8b5..74f0953 100644 --- a/src/source/router/module_router.cr +++ b/src/source/router/module_router.cr @@ -52,7 +52,7 @@ module PlaceOS::Source::Router mappings.write do |state| # Remove reference in drivers state.drivers.delete(module_id) - state.system_modules.delete(module_id) + state.unmap_module(module_id) end # What it published is known from the registry, not from its storage, @@ -66,8 +66,7 @@ module PlaceOS::Source::Router def process_resource(action : Resource::Action, resource : Model::Module) : Resource::Result mod = resource - hierarchy_zones = Mappings.hierarchy_zones(mod) - return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted? + return Resource::Result::Skipped unless action.deleted? || in_scope?(action, mod) case action in .created? @@ -78,5 +77,20 @@ module PlaceOS::Source::Router handle_delete(mod) end end + + # Whether the Module is in a system with a hierarchy. + # + # While loading, the systems are already mapped (the ControlSystem router + # loads first, and maps exactly the systems with a hierarchy), so that is + # asked of the mappings. Asking the database meant reading each Module's + # systems, and each of their zones, one Module at a time + private def in_scope?(action : Resource::Action, mod : Model::Module) : Bool + if action.created? && !startup_finished? + module_id = mod.id.as(String) + mappings.read(&.system_modules.has_key?(module_id)) + else + !Mappings.hierarchy_zones(mod).empty? + end + end end end diff --git a/src/source/router/zone_router.cr b/src/source/router/zone_router.cr index bd0ffa6..21ff71a 100644 --- a/src/source/router/zone_router.cr +++ b/src/source/router/zone_router.cr @@ -37,9 +37,7 @@ module PlaceOS::Source::Router hierarchy_zones = Mappings.hierarchy_zones(zone) return Resource::Result::Skipped if hierarchy_zones.empty? && !action.deleted? - hierarchy_zones.each do |parent_zone| - publish_metadata(parent_zone, zone) - end + publish_metadata(hierarchy_zones, zone) # Update/destroy Zone mappings if zone.tags_changed? || zone.destroyed? diff --git a/src/source/status_events.cr b/src/source/status_events.cr index 7cfe69a..6f8d49d 100644 --- a/src/source/status_events.cr +++ b/src/source/status_events.cr @@ -290,27 +290,32 @@ module PlaceOS::Source # event: one that arrives meanwhile is processed afterwards, against the # new mappings, with a value at least as new as the one read here. # + # NOTE:: a zone or driver change can take in thousands of Modules. Only + # their status names are read up front; values are read a Module at a time + # as they are published, so their size never adds up in memory + # # Public for specs; otherwise only the processing fiber calls it def reconcile_now(module_ids : Set(String)) : Nil return if module_ids.empty? removals = [] of Mappings::Status - additions = [] of {Mappings::Status, String} + # module_id => what it should now be published at, and is not + additions = Hash(String, Array(Mappings::Status)).new # topics something maps to now, which a removal must leave alone: in a # reorder, Display_1 stops being one Module's and becomes another's valid = Set(String).new module_ids.each do |module_id| before = mappings.published.for_module(module_id) - values = begin - PlaceOS::Driver::RedisStorage.new(module_id).to_h + statuses = begin + PlaceOS::Driver::RedisStorage.new(module_id).keys rescue error Log.warn(exception: error) { "reading state of Module<#{module_id}> to reconcile" } next end after = Set(Mappings::Status).new - values.each_key do |status| + statuses.each do |status| mappings.status_events?(module_id, status).try &.each { |event| after << event } end @@ -319,8 +324,9 @@ module PlaceOS::Source valid << key end end - (before - after).each { |event| removals << event } - (after - before).each { |event| additions << {event, values[event.status]} } + removals.concat(before - after) + added = after - before + additions[module_id] = added.to_a unless added.empty? end timestamp = Time.utc @@ -331,19 +337,33 @@ module PlaceOS::Source message = Publisher::Message.new(event, nil, timestamp) publisher_managers.each(&.broadcast_delete(message)) end + valid.clear - additions.each do |(event, payload)| - mappings.published.add(event) - message = Publisher::Message.new(event, payload, timestamp) - publisher_managers.each(&.broadcast(message)) + added = 0 + additions.each do |module_id, events| + values = begin + PlaceOS::Driver::RedisStorage.new(module_id).to_h + rescue error + Log.warn(exception: error) { "reading state of Module<#{module_id}> to reconcile" } + next + end + + events.each do |event| + # cleared since its name was read: its removal is on the way + next unless payload = values[event.status]? + mappings.published.add(event) + message = Publisher::Message.new(event, payload, timestamp) + publisher_managers.each(&.broadcast(message)) + added += 1 + end end - unless removals.empty? && additions.empty? + unless removals.empty? && added == 0 Log.info { { message: "reconciled retained state", modules: module_ids.size.to_s, removed: removals.size.to_s, - added: additions.size.to_s, + added: added.to_s, } } end end @@ -354,6 +374,11 @@ module PlaceOS::Source private getter sweep_lock = Mutex.new @pending_sweep : {MqttPublisher, Set(String)}? = nil + # Systems whose published topics are kept while a broker's retained topics + # stream past. The broker sends them in topic order, so each system's + # arrive together and a handful is plenty + SWEEP_SYSTEMS_CACHED = 16 + # Removes every status topic a broker retains that nothing maps to. # # The registry of what is published starts empty, so whatever a broker @@ -361,6 +386,11 @@ module PlaceOS::Source # deleted or moved while source was down, or by a version of source that # did not clean up after them. Once the initial sync has been processed # the registry is complete, and anything else the broker holds is stale. + # + # NOTE:: a broker can retain hundreds of thousands of topics. Each is + # checked against the registry as it arrives and only the ones that look + # stale are kept, rather than holding every retained topic, and every + # published one, to compare as two sets def sweep_brokers : Nil # the registry is only complete once every synced event is processed until stopped? || (synchronize { event_container.empty? } && !mappings.reconcile_pending?) @@ -371,7 +401,9 @@ module PlaceOS::Source next unless manager.is_a?(MqttBrokerManager) manager.publishers.each do |publisher| begin - topics = publisher.retained_state_topics + cache = Hash(String, Set(String)).new + topics = publisher.retained_state_topics { |topic| !published_topic?(topic, cache) } + cache.clear sweep_lock.synchronize { @pending_sweep = {publisher, topics} } # one at a time, so the processing fiber takes each in turn while sweep_lock.synchronize { !@pending_sweep.nil? } && !stopped? @@ -392,19 +424,41 @@ module PlaceOS::Source end end + # Removes the retained topics that are still not published. + # + # They are checked again here, in the processing fiber, because a status + # published while the broker's topics streamed past was retained but + # possibly not yet in the registry when it was first checked + # # Public for specs; otherwise only the processing fiber calls it def sweep_now(publisher : Publisher, retained : Set(String), label : String = "") : Nil - valid = mappings.published.all.compact_map { |event| MqttPublisher.generate_key(event) }.to_set - stale = retained - valid + stale = retained.reject { |topic| published_topic?(topic) } stale.each { |topic| publisher.queue_delete_topic(topic) } Log.info { { - message: "swept retained state", - broker: label, - retained: retained.size.to_s, - removed: stale.size.to_s, + message: "swept retained state", + broker: label, + checked: retained.size.to_s, + removed: stale.size.to_s, } } end + # Whether something is published at this state topic, looked up through + # the topic's system so only that system's topics are generated. `cache` + # keeps the topics of recently seen systems + private def published_topic?(topic : String, cache : Hash(String, Set(String))? = nil) : Bool + return false unless control_system_id = MqttPublisher.state_topic_system_id(topic) + + keys = cache.try &.[control_system_id]? + unless keys + keys = mappings.published.for_system(control_system_id).compact_map { |event| MqttPublisher.generate_key(event) }.to_set + if cache + cache.clear if cache.size >= SWEEP_SYSTEMS_CACHED + cache[control_system_id] = keys + end + end + keys.includes?(topic) + end + def self.parse_channel(channel : String) : {String, String} _, module_id, status = channel.split('/') {module_id, status}