From b1cba757b8166be33b39e27e3137b01f89913e6a Mon Sep 17 00:00:00 2001 From: Dion Moult Date: Mon, 14 Sep 2026 07:30:18 +1000 Subject: [PATCH] ifcparse: parse instances in parallel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The DATA section is split into one chunk per thread and each worker runs the same per-instance reader as the serial parse over its own reader, storage, inverse records and simple-type list; the results are merged in file order, so instance order, GlobalId precedence and inverse records are identical to the serial parse. Reference resolution then splits over the same threads: each instance's slots are its own and the name table is complete and read-only by then. The default is one thread per core, capped at 16; IFCOPENSHELL_PARSE_THREADS or file::parse_threads() overrides it, and 1 parses as before. The instance headers are read by one loop, for_each_instance_header(), shared with the lazy index: it looks declarations up once per keyword, passes over a bypassed instance's attribute list and slides past a stray keyword the way the serial reader does (the lazy index therefore no longer falls back on one). Finding the split points is the one place that looks at raw bytes rather than tokens, because tokenizing the file serially first would leave nothing to parallelise. It applies three rules: a string starts and ends at a quote and cannot span a line, and a comment runs from /* to */; a split is a '#' that starts a line outside both. Getting a string's end wrong can only lose a candidate, never accept a wrong one, since no string contains a newline. The equality test puts a comment holding a fake instance and a string holding "/*" between the chunks. file_reader gains for_each_span(), which hands a byte range out span by span (one span for a buffer, one per page for the paged reader), and reopen(), a reader over the same file for another thread. TXG 58 MB / 210_King 147 MB / OKgate22 231 MB, 12 threads: 0.44 / 1.24 / 2.01 s against 1.07 / 2.75 / 5.12 s on one thread; memory after the parse within 1–4%, peak +5–4%. This commit was written by an AI coding tool and has not been verified by a human. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013wcN7XquTfUi4vsKQ4KchL --- .../ifcopenshell/ifcopenshell_wrapper.pyi | 6 + src/ifcparse/file.h | 7 + src/ifcparse/file_reader.h | 41 ++ src/ifcparse/parse.cpp | 511 ++++++++++++++---- src/ifcparse/storage.h | 25 + .../tests/test_ifcopenshell_parse.cpp | 94 +++- 6 files changed, 590 insertions(+), 94 deletions(-) diff --git a/src/ifcopenshell-python/ifcopenshell/ifcopenshell_wrapper.pyi b/src/ifcopenshell-python/ifcopenshell/ifcopenshell_wrapper.pyi index e50b76ac7c..877ed50486 100644 --- a/src/ifcopenshell-python/ifcopenshell/ifcopenshell_wrapper.pyi +++ b/src/ifcopenshell-python/ifcopenshell/ifcopenshell_wrapper.pyi @@ -925,6 +925,12 @@ class file(file_mixin): ... def get_max_id(self) -> int: ... + def parse_threads(self, *args: int) -> int: + """Get, or with an argument set, the number of threads ``initialize()`` parses instances with; 0 uses one per core (capped at 16) or honours ``IFCOPENSHELL_PARSE_THREADS``.""" + ... + def effective_parse_threads(self) -> int: + """The thread count ``initialize()`` will use given ``parse_threads()`` and the environment.""" + ... def lazy_loading(self, *args: bool) -> bool: """Get, or with an argument set, whether ``initialize()`` indexes the file with one pass and parses each instance's attributes on first access. Set before ``initialize()``.""" ... diff --git a/src/ifcparse/file.h b/src/ifcparse/file.h index 03dfff068e..92c5d021f8 100644 --- a/src/ifcparse/file.h +++ b/src/ifcparse/file.h @@ -214,6 +214,7 @@ public: private: bool lazy_loading_ = false; + unsigned parse_threads_ = 0; file_open_status good_ = file_open_status::SUCCESS; std::reference_wrapper logger_; @@ -289,6 +290,12 @@ public: // anything it does not handle. void lazy_loading(bool value) { lazy_loading_ = value; } bool lazy_loading() const { return lazy_loading_; } + // Threads used to parse instances; 0 (the default) picks one per core, + // capped at 16, or honours IFCOPENSHELL_PARSE_THREADS. Set before + // initialize(). + void parse_threads(unsigned value) { parse_threads_ = value; } + unsigned parse_threads() const { return parse_threads_; } + unsigned effective_parse_threads() const; #ifdef USE_MMAP bool initialize(const std::string& path, bool use_mmap); #endif diff --git a/src/ifcparse/file_reader.h b/src/ifcparse/file_reader.h index e1beccb302..37d1c9c440 100644 --- a/src/ifcparse/file_reader.h +++ b/src/ifcparse/file_reader.h @@ -144,6 +144,45 @@ public: size_t size() const { return impl_->size(); } IFC_READER_INLINE size_t remaining() const { return size() - cursor_; } + // Hands fn(const char* data, size_t length, size_t offset) contiguous + // spans that together cover [begin, end): a single span for a + // contiguous implementation, one per page for the paged one. This is + // how a pass over the bytes stays independent of how the file is held. + template + void for_each_span(size_t begin, size_t end, Fn&& fn) const { + end = std::min(end, size()); + if (begin >= end) { + return; + } + if constexpr (std::is_same_v) { + const size_t page_size = impl_->page_size(); + for (size_t index = begin / page_size; index * page_size < end; ++index) { + const auto page = impl_->page(index); + const size_t page_begin = index * page_size; + const size_t from = std::max(begin, page_begin) - page_begin; + const size_t to = std::min(end, page_begin + page.second) - page_begin; + if (to > from) { + fn(page.first + from, to - from, page_begin + from); + } + } + } else if constexpr (std::is_same_v) { + throw std::logic_error("A pushed sequential reader has no random access to byte ranges"); + } else { + fn(impl_->data() + begin, end - begin, begin); + } + } + + // A reader over the same file that can be used from another thread: + // the paged implementation gets its own page cache, a contiguous one + // shares the (read-only) content. + file_reader reopen() const { + if constexpr (std::is_same_v) { + return file_reader(impl_->path(), impl_->page_size(), impl_->capacity()); + } else { + return clone(); + } + } + IFC_READER_INLINE char peek() const { if (cursor_ >= size()) { throw std::out_of_range("peek at EOF"); @@ -278,6 +317,7 @@ public: full_buffer_impl(const std::string& content, const caller_fed_tag& tag); size_t size() const { return size_; } + const char* data() const { return buf_.data(); } char get(size_t position) const { if (position >= size_) { throw std::out_of_range("get out of range"); @@ -358,6 +398,7 @@ public: explicit mmap_impl(const std::string& path); size_t size() const { return size_; } + const char* data() const { return map_.data(); } char get(size_t position) const { if (position >= size_) { throw std::out_of_range("get out of range"); diff --git a/src/ifcparse/parse.cpp b/src/ifcparse/parse.cpp index 4a593ce199..0c74b107d3 100644 --- a/src/ifcparse/parse.cpp +++ b/src/ifcparse/parse.cpp @@ -44,6 +44,9 @@ #include #include #include +#include +#include +#include // Apple clang's libc++ has no floating-point std::from_chars overload (it's // =deleted), so on macOS doubles are parsed via strtod_l with a cached "C" @@ -1906,6 +1909,7 @@ bool ifcopenshell::file::initialize(const std::string& fn, bool mmap) { file_reader s(fn); storage_.emplace<1>(this, logger_.get()); header_.reset(new spf_header(this, &logger_.get())); + std::get(storage_).parse_threads = effective_parse_threads(); std::get(storage_).read_from_stream(&s, schema_, max_id_, types_to_bypass_loading_); } @@ -1945,6 +1949,7 @@ bool ifcopenshell::file::initialize(const std::string& path, filetype ty, bool r } if (!indexed) { file_reader s(path); + std::get(storage_).parse_threads = effective_parse_threads(); std::get(storage_).read_from_stream(&s, schema_, max_id_, types_to_bypass_loading_); } @@ -1988,6 +1993,20 @@ bool ifcopenshell::file::initialize(const std::string& path, filetype ty, bool r return good_ == file_open_status::SUCCESS; } +unsigned ifcopenshell::file::effective_parse_threads() const { + if (parse_threads_ != 0) { + return parse_threads_; + } + if (const char* env = std::getenv("IFCOPENSHELL_PARSE_THREADS")) { + const int value = std::atoi(env); + if (value > 0) { + return (unsigned)value; + } + } + const unsigned cores = std::thread::hardware_concurrency(); + return std::min(16u, std::max(1u, cores)); +} + void ifcopenshell::file::bypass_type(const std::string& type_name) { types_to_bypass_loading_.insert(type_name); } @@ -2552,6 +2571,112 @@ void ifcopenshell::impl::in_memory_file_storage::resolve_instance_references(con } } +namespace { + +// Walks the instance headers "#name = KEYWORD(" from the cursor to `end` +// the way instance_streamer::read_instance() does: the declaration is +// looked up once per keyword, unknown and non-entity types are logged and +// skipped, a bypassed instance is collected and its attribute list passed +// over, and every other instance is handed to `visit(name, declaration, +// keyword_offset)` with the lexer just past its opening parenthesis. Stops +// at the ENDSEC that closes the DATA section, at `end`, or when `visit` +// returns false; a header ENDSEC, the DATA keyword and a stray keyword are +// passed over, as the serial reader passes over what it cannot match. +template +void for_each_instance_header(Reader& reader, spf_lexer& lexer, size_t end, const ifcopenshell::schema_definition* schema, const std::vector& bypassed_types, std::vector& bypassed, ifcopenshell::logger& log, Visit visit) { + std::unordered_map declarations; + bool in_data = false; + while (true) { + while (!reader.eof()) { + const char c = reader.peek(); + if (c == ' ' || c == '\t' || c == '\r' || c == '\n') { + reader.increment(); + } else { + break; + } + } + if (reader.eof() || reader.tell() >= end) { + return; + } + token first = lexer.template next(); + if (!first) { + return; + } + if (first.is_keyword()) { + const std::string keyword = first.as_string(); + lexer.reset_pool(); + if (keyword == "ENDSEC") { + if (in_data) { + return; + } + in_data = true; // the header's: the DATA section follows + } + continue; + } + if (!first.is_identifier()) { + lexer.reset_pool(); + continue; + } + in_data = true; + const uint32_t name = (uint32_t)first.as_identifier(); + if (!lexer.template next().is_operator('=')) { + continue; + } + token keyword = lexer.template next(); + if (!keyword.is_keyword()) { + lexer.reset_pool(); + continue; + } + const ifcopenshell::declaration* declaration = nullptr; + const std::string keyword_text = keyword.as_string(); + const size_t keyword_offset = keyword.start_pos; + lexer.reset_pool(); + auto found = declarations.find(keyword_text); + if (found == declarations.end()) { + try { + declaration = schema->declaration_by_name(keyword_text); + if (declaration->as_entity() == nullptr) { + log.message(ifcopenshell::logger::LOG_ERROR, "Non-entity type " + declaration->name() + " at offset " + std::to_string(keyword_offset)); + declaration = nullptr; + } + } catch (const exception& e) { + log.message(ifcopenshell::logger::LOG_ERROR, std::string(e.what()) + " at offset " + std::to_string(keyword_offset)); + } + declarations.emplace(keyword_text, declaration); + } else { + declaration = found->second; + } + if (!lexer.template next().is_operator('(')) { + continue; + } + if (declaration == nullptr || bypassed_types[declaration->index_in_schema()]) { + if (declaration != nullptr) { + bypassed.push_back(name); + } + // Pass over the attribute list, whatever it holds. + int depth = 1; + while (depth > 0) { + token t = lexer.template next(); + if (!t) { + return; + } + if (t.is_operator('(')) { + ++depth; + } else if (t.is_operator(')')) { + --depth; + } + lexer.reset_pool(); + } + continue; + } + if (!visit(name, declaration, keyword_offset)) { + return; + } + } +} + +} + struct ifcopenshell::impl::in_memory_file_storage::lazy_source { file_reader reader; spf_lexer> lexer; @@ -2643,73 +2768,17 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& byref_excl_.reserve(reader.size() / 32); // One pass over the DATA section with the tokenizer's index policy: - // the instance headers as tokens, then the attribute list as tokens - // with only the parentheses, commas and names looked at. Nothing is - // decoded. A keyword where an instance should start, or a token the - // tokenizer rejects, stops the index and the caller parses in full. - std::unordered_map declarations; + // the instance headers through the shared loop, then the attribute list + // as tokens with only the parentheses, commas and names looked at. + // Nothing is decoded. A token the tokenizer rejects, or a structure the + // loop below does not expect, stops the index and the caller parses in + // full. const char* failure = nullptr; size_t failure_offset = 0; - bool in_data = false; try { - while (failure == nullptr) { - token first = lexer.next(); - if (!first) { - break; - } - if (first.is_keyword()) { - const std::string keyword = first.as_string(); - if (keyword == "ENDSEC") { - if (in_data) { - break; - } - in_data = true; // the header's; DATA follows - } else if (keyword != "DATA") { - failure = "keyword where an instance should start"; - failure_offset = first.start_pos; - } - lexer.reset_pool(); - continue; - } - if (!first.is_identifier()) { - continue; - } - in_data = true; - const uint32_t name = (uint32_t)first.as_identifier(); - if (!lexer.next().is_operator('=')) { - continue; - } - token keyword = lexer.next(); - if (!keyword.is_keyword()) { - continue; - } - const ifcopenshell::declaration* declaration = nullptr; - const std::string keyword_text = keyword.as_string(); - lexer.reset_pool(); - auto found = declarations.find(keyword_text); - if (found == declarations.end()) { - try { - declaration = schema->declaration_by_name(keyword_text); - if (declaration->as_entity() == nullptr) { - logger_.get().message(ifcopenshell::logger::LOG_ERROR, "Non-entity type " + declaration->name() + " at offset " + std::to_string(keyword.start_pos)); - declaration = nullptr; - } - } catch (const exception& e) { - logger_.get().message(ifcopenshell::logger::LOG_ERROR, std::string(e.what()) + " at offset " + std::to_string(keyword.start_pos)); - } - declarations.emplace(keyword_text, declaration); - } else { - declaration = found->second; - } - if (!lexer.next().is_operator('(')) { - failure = "expected ( after the type"; - failure_offset = keyword.start_pos; - break; - } + for_each_instance_header(reader, lexer, reader.size(), schema, bypassed_types, lazy_bypassed_, logger_.get(), [&](uint32_t name, const ifcopenshell::declaration* declaration, size_t) { const uint64_t attributes_offset = reader.tell(); - const bool bypassed = declaration != nullptr && bypassed_types[declaration->index_in_schema()]; - const bool indexed = declaration != nullptr && !bypassed; - const uint16_t type_index = indexed ? (uint16_t)declaration->index_in_schema() : 0; + const uint16_t type_index = (uint16_t)declaration->index_in_schema(); int depth = 1; int attribute = 0; bool first_value = true; @@ -2719,7 +2788,7 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& if (!t) { failure = "file ends inside an instance"; failure_offset = attributes_offset; - break; + return false; } if (t.is_operator()) { if (t.value_char == '(') { @@ -2731,12 +2800,10 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& } else if (t.value_char == ';') { failure = "; inside an instance"; failure_offset = t.start_pos; - break; + return false; } } else if (t.is_identifier()) { - if (indexed) { - byref_excl_.add((uint32_t)t.as_identifier(), name, type_index, attribute); - } + byref_excl_.add((uint32_t)t.as_identifier(), name, type_index, attribute); } else if (t.type == token::Token_STRING && depth == 1 && attribute == 0 && first_value) { guid_begin = t.start_pos + 1; guid_end = reader.tell() - 1; @@ -2746,19 +2813,10 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& } lexer.reset_pool(); } - if (failure != nullptr) { - break; - } if (!lexer.next().is_operator(';')) { failure = "expected ; after )"; failure_offset = reader.tell(); - break; - } - if (bypassed) { - lazy_bypassed_.push_back(name); - } - if (!indexed) { - continue; + return false; } auto data = ifcopenshell::make_pointer_type(file, declaration, name, instance_data::lazy_tag{}); if (!byid_.insert({name, data}).second) { @@ -2787,7 +2845,8 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& byguid_[key] = instance; } } - } + return true; + }); } catch (const invalid_token_exception&) { failure = "invalid token"; failure_offset = reader.tell(); @@ -2804,6 +2863,248 @@ bool ifcopenshell::impl::in_memory_file_storage::index_lazily(const std::string& return true; } +namespace { + +// What one parser worker produces from its chunk of the DATA section. +struct parse_worker_output { + std::vector instances; + std::vector bypassed; + unresolved_references mixed_references; + std::unique_ptr storage; + file_open_status status = file_open_status::SUCCESS; + std::exception_ptr error; +}; + +// The instances between `begin` and `end`, both offsets of a '#' that +// starts a line outside any string or comment, through the same reader +// the serial parse uses. +template +void parse_chunk(const Reader& source, size_t begin, size_t end, const ifcopenshell::schema_definition* schema, const std::vector& bypassed_types, parse_worker_output& out) { + try { + Reader reader = source.reopen(); + reader.seek(begin); + auto& storage = *out.storage; + spf_lexer lexer(&reader, storage.logger_.get()); + for_each_instance_header(reader, lexer, end, schema, bypassed_types, out.bypassed, storage.logger_.get(), [&](uint32_t name, const ifcopenshell::declaration* declaration, size_t) { + try { + auto data = storage.load(&lexer, name, declaration, declaration->as_entity(), -1, true); + storage.try_read_semicolon(&lexer); + lexer.reset_pool(); + out.instances.push_back(data); + return true; + } catch (const invalid_token_exception& e) { + out.status = file_open_status::INVALID_SYNTAX; + storage.logger_.get().error(e); + return false; + } + }); + } catch (...) { + out.error = std::current_exception(); + } +} + +// Matches a literal byte by byte, across spans. +class literal_matcher { + const char* literal_; + size_t length_; + size_t matched_ = 0; + public: + explicit literal_matcher(const char* literal) + : literal_(literal), length_(std::strlen(literal)) {} + // True on the byte that completes the literal. + bool feed(char c) { + if (c == literal_[matched_]) { + if (++matched_ == length_) { + matched_ = 0; + return true; + } + } else { + matched_ = c == literal_[0] ? 1 : 0; + } + return false; + } +}; + +} + +template +bool ifcopenshell::impl::in_memory_file_storage::read_instances_parallel(Reader* s, const ifcopenshell::schema_definition* schema, const std::set& types_to_bypass, unsigned int& max_id, unsigned threads, std::vector& bypassed, unresolved_references& mixed_references, std::vector& instances) { + if constexpr (std::is_same_v>) { + return false; + } else { + const size_t n = s->size(); + constexpr size_t min_bytes_per_thread = 2u << 20; + threads = (unsigned)std::min(threads, std::max(1, n / min_bytes_per_thread)); + if (threads < 2) { + return false; + } + + // One pass over the bytes finds the DATA section and, nearest each + // nominal split point, an instance boundary: a '#' that starts a + // line outside any string and any comment. This is the one place + // that looks at raw bytes instead of tokens, because tokenizing the + // file serially to find the split points would leave nothing to + // parallelise. It applies three rules only: a string starts and + // ends at a quote (a doubled quote closes and reopens, which comes + // to the same thing) and cannot span a line; a comment runs from + // /* to */. Getting a string's end wrong can only lose a candidate + // boundary, never accept a wrong one, since no string contains a + // newline. + literal_matcher data_matcher("\nDATA;"), endsec_matcher("\nENDSEC"); + size_t data_begin = 0, data_end = 0; + bool in_string = false, in_comment = false, newline = false; + char previous = 0; + std::vector bounds; + size_t next_split = 0; + s->for_each_span(0, n, [&](const char* data, size_t length, size_t offset) { + for (size_t i = 0; i < length; ++i) { + const char c = data[i]; + const size_t at = offset + i; + if (data_begin == 0) { + if (data_matcher.feed(c)) { + data_begin = at + 1; + bounds.push_back(data_begin); + next_split = data_begin + (n - data_begin) / threads; + } + continue; + } + if (data_end != 0) { + return; + } + if (in_comment) { + if (previous == '*' && c == '/') { + in_comment = false; + } + } else if (in_string) { + if (c == '\'' || c == '\n') { + in_string = false; + } + } else if (c == '\'') { + in_string = true; + } else if (previous == '/' && c == '*') { + in_comment = true; + } else if (endsec_matcher.feed(c)) { + data_end = at + 1 - 7; + return; + } else if (newline && c == '#' && at >= next_split && bounds.size() < threads) { + bounds.push_back(at); + next_split = data_begin + (n - data_begin) * bounds.size() / threads; + } + newline = c == '\n'; + previous = c; + } + }); + if (data_begin == 0) { + return false; + } + if (data_end == 0) { + data_end = n; + } + bounds.push_back(data_end); + if (bounds.size() < 3) { + return false; + } + + std::vector bypassed_types(schema->declarations().size(), 0); + for (const auto& type_name : types_to_bypass) { + const ifcopenshell::declaration* declaration = nullptr; + try { + declaration = schema->declaration_by_name(type_name); + } catch (const ifcopenshell::exception&) { + continue; + } + std::function mark = [&](const ifcopenshell::entity* e) { + bypassed_types[e->index_in_schema()] = 1; + for (const auto* subtype : e->subtypes()) { + mark(subtype); + } + }; + if (const auto* e = declaration->as_entity()) { + mark(e); + } + } + + std::vector> outputs; + for (size_t k = 0; k + 1 < bounds.size(); ++k) { + auto output = std::make_unique(); + output->storage = std::make_unique(file, logger_.get()); + output->storage->schema = schema; + output->storage->resolve_references_in_place = true; + output->storage->references_to_resolve = &output->mixed_references; + output->storage->byref_excl_.reserve((bounds[k + 1] - bounds[k]) / 32); + outputs.push_back(std::move(output)); + } + std::vector workers; + for (size_t k = 0; k < outputs.size(); ++k) { + workers.emplace_back([&, k]() { parse_chunk(*s, bounds[k], bounds[k + 1], schema, bypassed_types, *outputs[k]); }); + } + for (auto& worker : workers) { + worker.join(); + } + for (const auto& output : outputs) { + if (output->error) { + std::rethrow_exception(output->error); + } + } + + // Merge in file order, doing what the serial loop does per instance. + // Everything merged into is sized up front so the transient peak + // stays close to the serial parse's. + size_t instance_count = 0, record_count = 0, simple_type_count = 0; + for (const auto& output : outputs) { + instance_count += output->instances.size(); + record_count += output->storage->byref_excl_.size(); + simple_type_count += output->storage->read_simple_type_instances.size(); + } + instances.reserve(instances.size() + instance_count); + byref_excl_.reserve(byref_excl_.size() + record_count); + read_simple_type_instances.reserve(read_simple_type_instances.size() + simple_type_count); + const auto* ifcroot = schema->declaration_by_name("IfcRoot"); + for (auto& output : outputs) { + for (const auto& data : output->instances) { + const uint32_t name = data->id(); + const auto* declaration = data->declaration(); + express::base instance(data); + if (declaration->is(*ifcroot)) { + try { + const std::string guid = instance.get_attribute_value(0); + std::array key; + if (guid_key(guid, key)) { + if (byguid_.count(key) != 0) { + logger_.get().message(ifcopenshell::logger::LOG_WARNING, "Instance encountered with non-unique GlobalId " + guid); + } + byguid_[key] = instance; + } + } catch (const exception& ex) { + logger_.get().message(ifcopenshell::logger::LOG_ERROR, ex.what()); + } + } + bytype_excl_[declaration].push_back(instance); + if (!byid_.insert({name, data}).second) { + logger_.get().message(ifcopenshell::logger::LOG_WARNING, "Overwriting instance with name #" + std::to_string(name)); + byid_.erase(name); + byid_.insert({name, data}); + } + max_id = (std::max)(max_id, (unsigned int)name); + instances.push_back(data); + } + byref_excl_.append(std::move(output->storage->byref_excl_)); + auto& simple = output->storage->read_simple_type_instances; + read_simple_type_instances.insert(read_simple_type_instances.end(), simple.begin(), simple.end()); + std::vector().swap(simple); + std::vector().swap(output->instances); + bypassed.insert(bypassed.end(), output->bypassed.begin(), output->bypassed.end()); + mixed_references.insert(mixed_references.end(), std::make_move_iterator(output->mixed_references.begin()), std::make_move_iterator(output->mixed_references.end())); + if (output->status != file_open_status::SUCCESS) { + good_ = output->status; + } + } + std::sort(bypassed.begin(), bypassed.end()); + outputs.clear(); + return true; + } +} + template void ifcopenshell::impl::in_memory_file_storage::read_from_stream(Reader* s, const ifcopenshell::schema_definition*& schema, unsigned int& max_id, const std::set& typed_to_bypass) { schema = nullptr; @@ -2854,7 +3155,12 @@ void ifcopenshell::impl::in_memory_file_storage::read_from_stream(Reader* s, con logger_.get().status("Scanning file..."); - while (streamer) { + std::vector bypassed; + unresolved_references mixed_references; + std::vector parsed_instances; + const bool parallel = parse_threads > 1 && read_instances_parallel(s, schema, typed_to_bypass, max_id, parse_threads, bypassed, mixed_references, parsed_instances); + + while (!parallel && streamer) { auto inst = streamer.read_instance(); if (!inst) { @@ -2894,10 +3200,14 @@ void ifcopenshell::impl::in_memory_file_storage::read_from_stream(Reader* s, con max_id = (std::max)(max_id, (unsigned int)current_id); } - good_ = streamer.status(); - byref_excl_ = std::move(streamer.inverses()); + if (!parallel) { + good_ = streamer.status(); + byref_excl_ = std::move(streamer.inverses()); + read_simple_type_instances = streamer.steal_instances(); + bypassed = streamer.bypassed_instances(); + mixed_references = std::move(streamer.references()); + } byref_excl_.sort(); - read_simple_type_instances = streamer.steal_instances(); logger_.get().status("\rDone scanning file "); @@ -2905,19 +3215,38 @@ void ifcopenshell::impl::in_memory_file_storage::read_from_stream(Reader* s, con return; } - const auto& bypassed = streamer.bypassed_instances(); - // The names left in the attribute slots, then those of the simple type // instances read inline (a select such as IfcPropertySetDefinitionSet). - for (auto it = byid_.begin(); it != byid_.end(); ++it) { - resolve_instance_references(it->second, bypassed); - } - for (const auto& data : read_simple_type_instances) { - resolve_instance_references(data, bypassed); + if (parallel) { + // Each instance's slots are its own, byid_ is complete and only read, + // and the logger locks, so the passes split over threads. + for (const auto* list : {&parsed_instances, &read_simple_type_instances}) { + std::vector workers; + const size_t per_thread = (list->size() + parse_threads - 1) / parse_threads; + for (size_t begin = 0; begin < list->size(); begin += per_thread) { + const size_t end = std::min(list->size(), begin + per_thread); + workers.emplace_back([this, list, &bypassed, begin, end]() { + for (size_t i = begin; i < end; ++i) { + resolve_instance_references((*list)[i], bypassed); + } + }); + } + for (auto& worker : workers) { + worker.join(); + } + } + std::vector().swap(parsed_instances); + } else { + for (auto it = byid_.begin(); it != byid_.end(); ++it) { + resolve_instance_references(it->second, bypassed); + } + for (const auto& data : read_simple_type_instances) { + resolve_instance_references(data, bypassed); + } } // What was read with in-place storage off: the header entities. - for (const auto& p : streamer.references()) { + for (const auto& p : mixed_references) { const auto& ref = p.first.name_; const auto& refattr = p.first.index_; diff --git a/src/ifcparse/storage.h b/src/ifcparse/storage.h index 62d6d8e851..9f3f1728f4 100644 --- a/src/ifcparse/storage.h +++ b/src/ifcparse/storage.h @@ -454,6 +454,19 @@ namespace ifcopenshell { } // Finalizes bulk loading. Subsequent add() calls go to the delta. + // Takes over another index's records, e.g. one built by a parser + // worker. Both must still be in bulk-load mode (no delta). + void append(inverse_index&& other) { + if (base_.empty()) { + base_ = std::move(other.base_); + } else { + base_.insert(base_.end(), other.base_.begin(), other.base_.end()); + } + sorted_ = false; + other.clear(); + invalidate_materialized(); + } + void sort() const { if (!sorted_) { std::sort(base_.begin(), base_.end(), record_less); @@ -602,6 +615,18 @@ namespace ifcopenshell { std::vector lazy_bypassed_; std::vector> lazy_offsets_; bool index_lazily(const std::string& path, const ifcopenshell::schema_definition*& schema, unsigned int& max_id, const std::set& types_to_bypass); + + // Number of threads read_from_stream() may use to parse instances; + // 1 parses serially. Set by file::initialize(). + unsigned parse_threads = 1; + + // Parses the DATA section with `threads` workers, each running the + // same per-instance reader over its own chunk, and merges the + // results in file order. Returns false, without side effects, when + // the file is too small to be worth it or no split points were + // found; the caller then parses serially. + template + bool read_instances_parallel(Reader* stream, const ifcopenshell::schema_definition* schema, const std::set& types_to_bypass, unsigned int& max_id, unsigned threads, std::vector& bypassed, unresolved_references& mixed_references, std::vector& instances); void materialize(instance_data* data); typedef std::map> entities_by_type; diff --git a/src/ifcparse/tests/test_ifcopenshell_parse.cpp b/src/ifcparse/tests/test_ifcopenshell_parse.cpp index 13f1cb4fc7..6663cb2b45 100644 --- a/src/ifcparse/tests/test_ifcopenshell_parse.cpp +++ b/src/ifcparse/tests/test_ifcopenshell_parse.cpp @@ -483,19 +483,107 @@ TEST_CASE("Lazy loading yields the same instances, attributes, inverses and Glob std::filesystem::remove(path); } -TEST_CASE("Lazy loading falls back to the full parser on syntax the index pass does not handle", "[ifcparse]") { +TEST_CASE("Lazy loading passes over a stray keyword like the full parser and falls back on what the index pass rejects", "[ifcparse]") { const auto path = std::filesystem::temp_directory_path() / "ifcopenshell_lazy_fallback_test.ifc"; { std::ofstream out(path); - // A stray keyword between instances is something the full parser skips past, but the index pass gives up on. + // A stray keyword between instances: the shared header loop slides past it. std::string spf(reference_resolution_spf); spf.replace(spf.find("#8=IFCWALL"), 0, "STRAY;\n"); out << spf; } + check_lazy_matches_strict(path.string()); + { + std::ofstream out(path); + // A semicolon inside an attribute list is not something the index pass tracks; the full parser takes over. + std::string spf(reference_resolution_spf); + spf.replace(spf.find("(#1,#2,#3)"), 10, "(#1;#2,#3)"); + out << spf; + } ifcopenshell::file lazy(ifcopenshell::uninitialized_tag{}); lazy.lazy_loading(true); lazy.initialize(path.string()); std::filesystem::remove(path); CHECK_FALSE(lazy.lazy_loading()); - CHECK(lazy.instance_by_id(4)); + CHECK(lazy.instance_by_id(8)); +} + +TEST_CASE("Parallel parsing yields the same instances, attributes, inverses and GlobalIds as serial parsing, comments in DATA included", "[ifcparse]") { + // The fixture is small, so the threshold would keep it serial; write a + // file big enough to be chunked by repeating its DATA section under new + // names, with a comment and a string holding '/*' between the copies. + const std::string fixture = std::string(IFCOPENSHELL_TEST_FIXTURES) + "/ColumnPSetsOfSets.ifc"; + std::string source; + { + std::ifstream in(fixture, std::ios::binary); + source.assign(std::istreambuf_iterator(in), std::istreambuf_iterator()); + } + const size_t data_begin = source.find("\nDATA;") + 6; + const size_t data_end = source.find("\nENDSEC", data_begin); + const std::string data = source.substr(data_begin, data_end - data_begin); + // Renumber "#N" to "#N+offset" per copy; every name and reference is offset consistently. + const auto renumber = [](const std::string& block, uint32_t offset) { + std::string out; + out.reserve(block.size() + block.size() / 4); + for (size_t i = 0; i < block.size(); ++i) { + if (block[i] == '#' && i + 1 < block.size() && isdigit((unsigned char)block[i + 1])) { + size_t j = i + 1; + uint32_t name = 0; + while (j < block.size() && isdigit((unsigned char)block[j])) { + name = name * 10 + (uint32_t)(block[j++] - '0'); + } + out += "#" + std::to_string(name + offset); + i = j - 1; + } else { + out += block[i]; + } + } + return out; + }; + std::string big = source.substr(0, data_begin); + uint32_t offset = 0; + while (big.size() < (12u << 20)) { + big += renumber(data, offset); + big += "\n/* a comment between instances\n#1=NOT AN INSTANCE\n*/\n#" + std::to_string(offset + 999999) + "=IFCLABEL('/* not a comment');\n"; + offset += 1000000; + } + big += source.substr(data_end); + const auto path = std::filesystem::temp_directory_path() / "ifcopenshell_parallel_parse_test.ifc"; + { + std::ofstream out(path, std::ios::binary); + out << big; + } + + ifcopenshell::file serial(ifcopenshell::uninitialized_tag{}); + serial.parse_threads(1); + REQUIRE(serial.initialize(path.string())); + ifcopenshell::file parallel(ifcopenshell::uninitialized_tag{}); + parallel.parse_threads(5); + REQUIRE(parallel.initialize(path.string())); + std::filesystem::remove(path); + + size_t count = 0; + for (auto it = serial.begin(); it != serial.end(); ++it) { + const express::base a = it->second; + const express::base b = parallel.instance_by_id((int)a.id()); + REQUIRE(b); + REQUIRE(&b.declaration() == &a.declaration()); + REQUIRE(parallel.instances_by_reference((int)a.id()).size() == serial.instances_by_reference((int)a.id()).size()); + std::ostringstream sa, sb; + a.to_string(sa); + b.to_string(sb); + REQUIRE(sb.str() == sa.str()); + ++count; + } + size_t parallel_count = 0; + for (auto it = parallel.begin(); it != parallel.end(); ++it) { + ++parallel_count; + } + CHECK(parallel_count == count); + CHECK(count > 5000); + CHECK(parallel.get_max_id() == serial.get_max_id()); + for (const auto& rooted : serial.instances_by_type("IfcRoot")) { + const std::string guid = rooted.get_attribute_value(0); + REQUIRE(parallel.instance_by_guid(guid).id() == serial.instance_by_guid(guid).id()); + } }