diff --git a/.gitignore b/.gitignore index 675dfa85..ab74fb5b 100644 --- a/.gitignore +++ b/.gitignore @@ -182,3 +182,5 @@ pyrightconfig.json # Casa tables *.table + +wheelhouse/* diff --git a/HISTORY.rst b/HISTORY.rst index 45b47ec2..e939e28c 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -8,6 +8,7 @@ History 0.5.3 (1-07-2026) ------------------ +* Implement write-locking (:pr:`162`) * Upgrade to casacore 3.8.1 (:pr:`218`) * Remove extraneous whitespace in casacore patch (:pr:`215`) * Introduce ``pytest != 9.1.0`` version restriction (:pr:`214`) diff --git a/cpp/arcae/configuration.cc b/cpp/arcae/configuration.cc index 517f7ce1..069630a5 100644 --- a/cpp/arcae/configuration.cc +++ b/cpp/arcae/configuration.cc @@ -10,9 +10,7 @@ using ::arrow::Status; namespace arcae { -Status SafeMultiThreadedWrites() { - return Status::NotImplemented("Safe Multi-threaded write support"); -} +Status SafeMultiThreadedWrites() { return Status::OK(); } Result Configuration::Get(const std::string& key) const { if (auto it = kvmap_.find(key); it != kvmap_.end()) { diff --git a/cpp/arcae/finally.h b/cpp/arcae/finally.h new file mode 100644 index 00000000..74895658 --- /dev/null +++ b/cpp/arcae/finally.h @@ -0,0 +1,26 @@ +#ifndef ARCAE_FINALLY_H +#define ARCAE_FINALLY_H + +#include + +namespace arcae { +namespace detail { + +template +struct Finally { + Fn fn; + bool enabled; + ~Finally() { + if (enabled) fn(); + } +}; + +template +auto finally(Fn&& fn) { + return Finally{std::forward(fn), true}; +} + +} // namespace detail +} // namespace arcae + +#endif // #define ARCAE_FINALLY_H diff --git a/cpp/arcae/isolated_table_proxy.cc b/cpp/arcae/isolated_table_proxy.cc index 28399a97..04223580 100644 --- a/cpp/arcae/isolated_table_proxy.cc +++ b/cpp/arcae/isolated_table_proxy.cc @@ -53,27 +53,60 @@ const std::shared_ptr& IsolatedTableProxy::GetPool( return proxy_pools_[instance].io_pool_; } +std::shared_ptr IsolatedTableProxy::SpawnWriter() { + // Create an IsolatedTableProxy that serialises writes to a single + // table instance (and thread). + // A custom deleter that releases resources (proxies and pools) + // that are actually managed by the parent ITP + std::shared_ptr itp(new IsolatedTableProxy(), [](auto* p) { + p->proxy_pools_.clear(); + p->dependencies_.clear(); + p->is_closed_ = true; + delete p; + }); + itp->dependencies_.emplace_back(shared_from_this()); + // Using the first instance means that writes can still work after + // non-syncable operations like AddColumns + auto instance = 0; // GetInstance(); + itp->proxy_pools_.push_back(proxy_pools_[instance]); + itp->is_closed_ = false; + return itp; +} + Status IsolatedTableProxy::CheckClosed() const { if (!is_closed_) return Status::OK(); return Status::Invalid("TableProxy is closed"); } Result IsolatedTableProxy::Close() { - if (!is_closed_) { - std::shared_ptr defer_close(nullptr, [this](...) { this->is_closed_ = true; }); - std::vector> results; - results.reserve(proxy_pools_.size()); - for (auto& [proxy, pool] : proxy_pools_) { - results.push_back(arrow::DeferNotOk(pool->Submit([tp = proxy]() { + if (is_closed_) return false; + // Mark closed on scope exit, regardless of how the close tasks fare. + std::shared_ptr defer_close(nullptr, [this](...) { this->is_closed_ = true; }); + std::vector> results; + results.reserve(proxy_pools_.size()); + for (auto& [proxy, pool] : proxy_pools_) { + results.push_back(arrow::DeferNotOk(pool->Submit([tp = proxy]() -> Result { + // flush/close may throw casacore::AipsError; an exception escaping a + // pool task would terminate the process, so convert it to a Status. + try { + tp->flush(false); tp->close(); - return true; - }))); - } - auto all_done = arrow::All(results); - all_done.Wait(); - return true; + } catch (const std::exception& e) { + return Status::Invalid("Error closing table: ", e.what()); + } + return true; + }))); + } + auto all_done = arrow::All(results); + ARROW_ASSIGN_OR_RAISE(auto outcomes, all_done.MoveResult()); + // Drain any late-scheduled continuations (e.g. Then callbacks) so that + // members are not destroyed while a pool task may still reference them. + for (auto& pp : proxy_pools_) pp.io_pool_->WaitForIdle(); + // Surface the first close failure, if any (still leaving the proxy closed). + for (const auto& outcome : outcomes) { + ARROW_RETURN_NOT_OK(outcome.status()); } - return false; + return true; } IsolatedTableProxy::~IsolatedTableProxy() { diff --git a/cpp/arcae/isolated_table_proxy.h b/cpp/arcae/isolated_table_proxy.h index e6d7f78f..6d4b7f04 100644 --- a/cpp/arcae/isolated_table_proxy.h +++ b/cpp/arcae/isolated_table_proxy.h @@ -1,17 +1,21 @@ #ifndef ARCAE_ISOLATED_TABLE_PROXY_H #define ARCAE_ISOLATED_TABLE_PROXY_H +#include #include +#include #include #include #include +#include #include #include #include #include #include +#include #include #include "arcae/type_traits.h" @@ -19,6 +23,11 @@ namespace arcae { namespace detail { +using CasaLockType = casacore::FileLocker::LockType; +using CasaTableProxy = casacore::TableProxy; +using ConstTableProxyRef = const casacore::TableProxy&; +using TableProxyRef = casacore::TableProxy&; + // Isolates access to a CASA Table to a single thread class IsolatedTableProxy : public std::enable_shared_from_this { public: @@ -32,61 +41,114 @@ class IsolatedTableProxy : public std::enable_shared_from_this proxy; + CasaLockType lock_type; + bool locked = false; + + MaybeLockAndFinalise(std::shared_ptr proxy_, CasaLockType lock_type_) + : proxy(std::move(proxy_)), lock_type(lock_type_) { + if (lock_type != CasaLockType::None) { + // Honour the acquisition result: TableProxy::lock returns whether + // the lock was taken. Throwing here is caught by the AipsError + // handler in the dispatching task and converted to Status::Invalid, + // so we never run the wrapped functor without a lock. + locked = proxy->lock(lock_type == CasaLockType::Write, 0); + if (!locked) throw casacore::AipsError("Failed to acquire table lock"); + } + } + ~MaybeLockAndFinalise() { + // Never let an exception escape the destructor: it can run during + // stack unwinding of a functor that already threw, and a second + // in-flight exception would call std::terminate. + if (!locked) return; + try { + if (lock_type == CasaLockType::Write) proxy->flush(false); + proxy->unlock(); + } catch (const std::exception& e) { + ARROW_LOG(WARNING) << "Error finalising table lock: " << e.what(); + } + } + }; + // Runs function with signature // ReturnType Function(const TableProxy &) on the isolation thread // returning an arrow::Future - template >> - ArrowFutureType RunAsync(Fn&& functor) const { - using ResultType = ArrowResultType; + template >> + ArrowFutureType RunAsync( + Fn&& functor, CasaLockType lock_type = CasaLockType::Read) const { + using ResultType = ArrowResultType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); - return RunInPool([this, instance = instance, - functor = std::forward(functor)]() mutable -> ResultType { - try { - return std::invoke(functor, *this->GetProxy(instance)); - } catch (casacore::AipsError& e) { - return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); - } - }); + return RunInPool( + instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, + functor = std::forward(functor)]() mutable -> ResultType { + try { + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(functor, *proxy); + } catch (casacore::AipsError& e) { + return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); + } + }); } // Runs functions with signature // ReturnType Function(TableProxy &) on the isolation thread // returning an arrow::Future template >> - ArrowFutureType RunAsync(Fn&& functor) { - using ResultType = ArrowFutureType; + typename = std::enable_if_t>> + ArrowFutureType RunAsync( + Fn&& functor, CasaLockType lock_type = CasaLockType::Read) { + using ResultType = ArrowFutureType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); - return RunInPool(instance, - [this, instance = instance, - functor = std::forward(functor)]() mutable -> ResultType { - try { - return std::invoke(functor, *this->GetProxy(instance)); - } catch (casacore::AipsError& e) { - return arrow::Status::Invalid("Unhandled casacore exception: ", - e.what()); - } - }); + return RunInPool( + instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, + functor = std::forward(functor)]() mutable -> ResultType { + try { + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(functor, *proxy); + } catch (casacore::AipsError& e) { + return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); + } + }); } - template >> - ArrowFutureType Then( - arrow::Future& future, Fn&& functor) const { - using ResultType = ArrowFutureType; + template < + typename Fn, typename R, + typename = std::enable_if_t>> + ArrowFutureType Then( + arrow::Future& future, Fn&& functor, + CasaLockType lock_type = CasaLockType::Read) const { + using ResultType = ArrowFutureType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); return future.Then( - [this, instance = instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, fn = std::forward(functor)](const R& result) mutable -> ResultType { try { - return std::invoke(fn, result, *this->GetProxy(instance)); + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(fn, result, *proxy); } catch (casacore::AipsError& e) { return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); } }, {}, @@ -95,20 +157,26 @@ class IsolatedTableProxy : public std::enable_shared_from_this>> - ArrowFutureType Then(arrow::Future& future, - Fn&& functor) { - using ResultType = ArrowFutureType; + typename = std::enable_if_t>> + ArrowFutureType Then( + arrow::Future& future, Fn&& functor, + CasaLockType lock_type = CasaLockType::Read) { + using ResultType = ArrowFutureType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); return future.Then( - [this, instance = instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, fn = std::forward(functor)](const R& result) mutable -> ResultType { try { - return std::invoke(fn, result, *this->GetProxy(instance)); + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(fn, result, *proxy); } catch (casacore::AipsError& e) { return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); } }, {}, @@ -120,20 +188,29 @@ class IsolatedTableProxy : public std::enable_shared_from_this - template >> - ArrowResultType RunSync(Fn&& functor) const { - using ResultType = ArrowFutureType; + template >> + ArrowResultType RunSync( + Fn&& functor, CasaLockType lock_type = CasaLockType::Read) const { + using ResultType = ArrowFutureType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); - return RunInPoolSync([this, instance = instance, - functor = std::forward(functor)]() mutable -> ResultType { - try { - return std::invoke(functor, *this->GetProxy(instance)); - } catch (casacore::AipsError& e) { - return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); - } - }); + return RunInPoolSync( + instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, + functor = std::forward(functor)]() mutable -> ResultType { + try { + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(functor, *proxy); + } catch (casacore::AipsError& e) { + return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); + } + }); } // Runs functions with signature @@ -141,28 +218,34 @@ class IsolatedTableProxy : public std::enable_shared_from_this template >> - ArrowResultType RunSync(Fn&& functor) { - using ResultType = ArrowResultType; + typename = std::enable_if_t>> + ArrowResultType RunSync( + Fn&& functor, CasaLockType lock_type = CasaLockType::Read) { + using ResultType = ArrowResultType; ARROW_RETURN_NOT_OK(CheckClosed()); auto instance = GetInstance(); - return RunInPoolSync(instance, - [this, instance = instance, - functor = std::forward(functor)]() mutable -> ResultType { - try { - return std::invoke(functor, *this->GetProxy(instance)); - } catch (casacore::AipsError& e) { - return arrow::Status::Invalid( - "Unhandled casacore exception: ", e.what()); - } - }); + return RunInPoolSync( + instance, + [weak_self = weak_from_this(), instance = instance, lock_type = lock_type, + functor = std::forward(functor)]() mutable -> ResultType { + try { + auto self = weak_self.lock(); + if (!self) return arrow::Status::Invalid("TableProxy is closed"); + auto proxy = self->GetProxy(instance); + MaybeLockAndFinalise lock(proxy, lock_type); + return std::invoke(functor, *proxy); + } catch (casacore::AipsError& e) { + return arrow::Status::Invalid("Unhandled casacore exception: ", e.what()); + } catch (std::runtime_error& e) { + return arrow::Status::Invalid("Unhandled exception: ", e.what()); + } + }); } // Construct an IsolatedTableProxy with the supplied function - template < - typename Fn, - typename = std::enable_if, arrow::Result>>>> + template , arrow::Result>>>> static arrow::Result> Make( Fn&& functor, std::size_t ninstances = 1) { if (ninstances < 1) { @@ -197,13 +280,13 @@ class IsolatedTableProxy : public std::enable_shared_from_this && - std::is_same_v, - arrow::Result>>>> + std::is_invocable_v && + std::is_same_v, + arrow::Result>>>> arrow::Result> Spawn(Fn&& functor) { struct enable_make_shared_itp : public IsolatedTableProxy {}; std::shared_ptr itp = std::make_shared(); - using ResultType = arrow::Result>; + using ResultType = arrow::Result>; // Mark as closed so that if construction fails, we don't try to close it itp->is_closed_ = true; @@ -212,7 +295,12 @@ class IsolatedTableProxy : public std::enable_shared_from_thisSubmit([this, i = i, fn = fwd_functor]() -> ResultType { - return std::invoke(fn, *GetProxy(i)); + // Hold a read lock on the source proxy while the functor runs. + // Under user locking, casacore requires the source table to be + // locked while e.g. a TAQL command builds a reference table from it. + auto proxy = this->GetProxy(i); + MaybeLockAndFinalise lock(proxy, CasaLockType::Read); + return std::invoke(fn, *proxy); })); ARROW_ASSIGN_OR_RAISE(auto table_proxy, future.MoveResult()); @@ -225,7 +313,11 @@ class IsolatedTableProxy : public std::enable_shared_from_this Proxy() const { return nullptr; } + // Spawns an IsolatedTableProxy encapsulating a single instance + // from this ITP. Suitable for constraining writes to a single + // thread and instance as concurrent writes issued from multiple + // threads will produce race conditions in the underlying casacore layer + std::shared_ptr SpawnWriter(); std::size_t nInstances() const { return proxy_pools_.size(); } @@ -282,7 +374,10 @@ class IsolatedTableProxy : public std::enable_shared_from_this proxy_pools_; - bool is_closed_; + // Default to closed so a partially-constructed or default-constructed + // proxy is never treated as open. Atomic because it is written/read + // across the isolation pool threads. + std::atomic is_closed_{true}; std::vector> dependencies_; }; diff --git a/cpp/arcae/new_table_proxy.cc b/cpp/arcae/new_table_proxy.cc index 990f7dde..348caf6c 100644 --- a/cpp/arcae/new_table_proxy.cc +++ b/cpp/arcae/new_table_proxy.cc @@ -27,6 +27,7 @@ using ::arrow::Result; using ::arrow::Status; using ::arrow::Table; +using LockType = ::casacore::FileLocker::LockType; using ::casacore::JsonOut; using ::casacore::JsonParser; using ::casacore::Record; @@ -34,6 +35,8 @@ using ::casacore::TableProxy; namespace arcae { +// Table Read Operations + Result NewTableProxy::GetTableDescriptor() const { return itp_ ->RunAsync([](TableProxy& tp) -> std::string { @@ -112,13 +115,6 @@ Result> NewTableProxy::GetRowShapes( .MoveResult(); } -Result NewTableProxy::PutColumn(const std::string& column, - const std::shared_ptr& data, - const detail::Selection& selection) const { - ARROW_RETURN_NOT_OK(SafeMultithreadedWrites()); - return WriteImpl(itp_, column, data, selection).MoveResult(); -} - Result NewTableProxy::Name() const { return itp_ ->RunAsync( @@ -149,38 +145,42 @@ Result NewTableProxy::nRows() const { .MoveResult(); } +// Table Write Operations from this point onwards + Result NewTableProxy::AddRows(std::size_t nrows) { - ARROW_RETURN_NOT_OK(SafeMultithreadedWrites()); - return itp_ - ->RunAsync([nrows = nrows](TableProxy& tp) { - detail::MaybeReopenRW(tp); - tp.addRow(nrows); - return true; - }) + return itp_->SpawnWriter() + ->RunAsync( + [nrows = nrows](TableProxy& tp) { + detail::MaybeReopenRW(tp); + tp.addRow(nrows); + return true; + }, + LockType::Write) .MoveResult(); } Result NewTableProxy::AddColumns(const std::string& json_columndescs, const std::string& json_dminfo) { - ARROW_RETURN_NOT_OK(SafeMultithreadedWrites()); - return itp_ - ->RunAsync([json_columndescs = json_columndescs, - json_dminfo = json_dminfo](TableProxy& tp) { - detail::MaybeReopenRW(tp); - Record columndescs = JsonParser::parse(json_columndescs).toRecord(); - Record dminfo = JsonParser::parse(json_dminfo).toRecord(); - tp.addColumns(columndescs, dminfo, false); - return true; - }) + return itp_->SpawnWriter() + ->RunAsync( + [json_columndescs = json_columndescs, + json_dminfo = json_dminfo](TableProxy& tp) { + detail::MaybeReopenRW(tp); + Record columndescs = JsonParser::parse(json_columndescs).toRecord(); + Record dminfo = JsonParser::parse(json_dminfo).toRecord(); + tp.addColumns(columndescs, dminfo, false); + return true; + }, + LockType::Write) .MoveResult(); } -Result NewTableProxy::Close() { return itp_->Close(); } - -Status NewTableProxy::SafeMultithreadedWrites() const { - if (itp_->nInstances() == 1) return Status::OK(); - return Status::NotImplemented("Write support when number of table instances ", - itp_->nInstances(), " is greater than one"); +Result NewTableProxy::PutColumn(const std::string& column, + const std::shared_ptr& data, + const detail::Selection& selection) const { + return WriteImpl(itp_->SpawnWriter(), column, data, selection).MoveResult(); } +Result NewTableProxy::Close() { return itp_->Close(); } + } // namespace arcae diff --git a/cpp/arcae/new_table_proxy.h b/cpp/arcae/new_table_proxy.h index b91c2c22..b00895a5 100644 --- a/cpp/arcae/new_table_proxy.h +++ b/cpp/arcae/new_table_proxy.h @@ -25,8 +25,8 @@ class NewTableProxy { std::size_t ninstances = 1) { struct enable_make_shared_ntp : public NewTableProxy {}; std::shared_ptr ntp = std::make_shared(); - ARROW_ASSIGN_OR_RAISE( - ntp->itp_, detail::IsolatedTableProxy::Make(std::move(functor), ninstances)); + ARROW_ASSIGN_OR_RAISE(ntp->itp_, detail::IsolatedTableProxy::Make( + std::forward(functor), ninstances)); return ntp; } @@ -98,11 +98,6 @@ class NewTableProxy { // Close the table arrow::Result Close(); - protected: - // Returns true if multithreaded writes are supported - // by this NewTableProxy - arrow::Status SafeMultithreadedWrites() const; - private: std::shared_ptr itp_; }; diff --git a/cpp/arcae/table_factory.cc b/cpp/arcae/table_factory.cc index e18f4008..a147da19 100644 --- a/cpp/arcae/table_factory.cc +++ b/cpp/arcae/table_factory.cc @@ -65,6 +65,16 @@ namespace { /// Table and subtable names static constexpr char kMain[] = "MAIN"; +// arcae confines each casacore Table to its own thread and acquires/releases a +// lock around every operation (see IsolatedTableProxy::MaybeLockAndFinalise). +// This requires user locking: under auto locking casacore retains the lock +// across operations, and a retained reader lock on one instance deadlocks the +// (now in-process) multi-reader/single-writer coordination that serialises a +// writer on instance 0 against readers on the others. Coerce to a user lock. +void CoerceToUserLocking(casacore::Record& lock_record) { + lock_record.define("option", "user"); +} + } // namespace Result> OpenTable(const std::string& filename, @@ -76,6 +86,7 @@ Result> OpenTable(const std::string& filename, [&filename, &readonly, &json_lockoptions, &cache_spec]() -> Result> { auto lock_record = JsonParser::parse(json_lockoptions).toRecord(); + CoerceToUserLocking(lock_record); try { auto proxy = std::make_shared(filename, lock_record, Table::TableOption::Old); @@ -91,6 +102,7 @@ Result> OpenTable(const std::string& filename, Result> DefaultMS(const std::string& name, const std::string& subtable, + std::size_t ninstances, const std::string& json_table_desc, const std::string& json_dminfo, const std::string& json_cache_size) { @@ -107,12 +119,15 @@ Result> DefaultMS(const std::string& name, modname.append(usubtable); } - ARROW_ASSIGN_OR_RAISE( - auto setup_new_table, - DefaultMSFactory(modname, usubtable, json_table_desc, json_dminfo)); ARROW_ASSIGN_OR_RAISE(auto cache_spec, detail::ParseCacheSizeSpec(json_cache_size)); - return NewTableProxy::Make([&]() -> Result> { + auto MakeMS = [name = name, modname = modname, usubtable = usubtable, + json_table_desc = json_table_desc, json_dminfo = json_dminfo, + cache_spec = cache_spec]() -> Result> { + ARROW_ASSIGN_OR_RAISE( + auto setup_new_table, + DefaultMSFactory(modname, usubtable, json_table_desc, json_dminfo)); + // MAIN Measurement Set case if (usubtable.empty() || usubtable == kMain) { auto ms = MeasurementSet(setup_new_table); @@ -182,7 +197,9 @@ Result> DefaultMS(const std::string& name, ms.rwKeywordSet().defineTable(usubtable, subtable->table()); ARROW_RETURN_NOT_OK(detail::ApplyCacheSizes(*subtable, cache_spec)); return subtable; - }); + }; + + return NewTableProxy::Make(std::move(MakeMS), ninstances); } // Execute a TAQL query on the supplied tables diff --git a/cpp/arcae/table_factory.h b/cpp/arcae/table_factory.h index 08d8918b..5361d74e 100644 --- a/cpp/arcae/table_factory.h +++ b/cpp/arcae/table_factory.h @@ -12,12 +12,12 @@ namespace arcae { arrow::Result> OpenTable( const std::string& filename, std::size_t ninstances = 1, bool readonly = true, - const std::string& json_lockoptions = R"({"option": "auto"})", + const std::string& json_lockoptions = R"({"option": "user"})", const std::string& json_cache_size = "{}"); arrow::Result> DefaultMS( const std::string& name, const std::string& subtable = "MAIN", - const std::string& json_table_desc = "{}", const std::string& json_dminfo = "{}", - const std::string& json_cache_size = "{}"); + std::size_t ninstances = 1, const std::string& json_table_desc = "{}", + const std::string& json_dminfo = "{}", const std::string& json_cache_size = "{}"); arrow::Result> Taql( const std::string& taql, const std::vector>& tables = {}); diff --git a/cpp/arcae/write_impl.cc b/cpp/arcae/write_impl.cc index 004135d1..bd058be2 100644 --- a/cpp/arcae/write_impl.cc +++ b/cpp/arcae/write_impl.cc @@ -18,6 +18,7 @@ #include #include +#include #include #include #include @@ -43,6 +44,7 @@ using ::arrow::ShouldSchedule; using ::arrow::Status; using ::arrow::internal::GetCpuThreadPool; +using LockType = casacore::FileLocker::LockType; template using CasaArray = ::casacore::Array; using ::casacore::ArrayColumn; @@ -79,22 +81,24 @@ struct WriteCallback { // If the chunk is contiguous in memory, we can write // from a CASA Array view over that position in the buffer if (chunk.IsContiguous()) { - return itp->RunAsync([column_name = std::move(column), chunk = chunk, - buffer = buffer](const TableProxy& tp) -> bool { - CT* in_ptr = const_cast(buffer->template data_as()); - in_ptr += chunk.FlatOffset(); - auto shape = chunk.GetShape(); - if (shape.size() == 1) { - auto column = ScalarColumn(tp.table(), column_name); - auto vector = CasaVector(shape, in_ptr, casacore::SHARE); - column.putColumnCells(chunk.ReferenceRows(), vector); - return true; - } - auto column = ArrayColumn(tp.table(), column_name); - auto array = CasaArray(shape, in_ptr, casacore::SHARE); - column.putColumnCells(chunk.ReferenceRows(), chunk.SectionSlicer(), array); - return true; - }); + return itp->RunAsync( + [column_name = std::move(column), chunk = chunk, + buffer = buffer](const TableProxy& tp) -> bool { + CT* in_ptr = const_cast(buffer->template data_as()); + in_ptr += chunk.FlatOffset(); + auto shape = chunk.GetShape(); + if (shape.size() == 1) { + auto column = ScalarColumn(tp.table(), column_name); + auto vector = CasaVector(shape, in_ptr, casacore::SHARE); + column.putColumnCells(chunk.ReferenceRows(), vector); + return true; + } + auto column = ArrayColumn(tp.table(), column_name); + auto array = CasaArray(shape, in_ptr, casacore::SHARE); + column.putColumnCells(chunk.ReferenceRows(), chunk.SectionSlicer(), array); + return true; + }, + LockType::Write); } // Transpose the array into the output buffer @@ -135,19 +139,20 @@ struct WriteCallback { return array; })); - return itp->Then(transpose_fut, - [column_name = std::move(column), chunk = chunk]( - const CasaArray& data, const TableProxy& tp) -> bool { - if (chunk.nDim() == 1) { - auto column = ScalarColumn(tp.table(), column_name); - column.putColumnCells(chunk.ReferenceRows(), data); - return true; - } - auto column = ArrayColumn(tp.table(), column_name); - column.putColumnCells(chunk.ReferenceRows(), chunk.SectionSlicer(), - data); - return true; - }); + return itp->Then( + transpose_fut, + [column_name = std::move(column), chunk = chunk](const CasaArray& data, + const TableProxy& tp) -> bool { + if (chunk.nDim() == 1) { + auto column = ScalarColumn(tp.table(), column_name); + column.putColumnCells(chunk.ReferenceRows(), data); + return true; + } + auto column = ArrayColumn(tp.table(), column_name); + column.putColumnCells(chunk.ReferenceRows(), chunk.SectionSlicer(), data); + return true; + }, + LockType::Write); } // Write a chunk of data from the encapsulated buffer @@ -289,16 +294,18 @@ Future WriteImpl(const std::shared_ptr& itp, std::shared_ptr selection; }; - auto shape_fut = itp->RunAsync([column = column, selection = selection, data = data]( - TableProxy& tp) mutable -> Result { - if (!tp.isWritable()) tp.reopenRW(); - ARROW_RETURN_NOT_OK(ColumnExists(tp.table(), column)); - auto table_column = TableColumn(tp.table(), column); - ARROW_ASSIGN_OR_RAISE(auto shape_data, - ResultShapeData::MakeWrite(table_column, data, selection)); - return ShapeResult{std::make_shared(std::move(shape_data)), - std::make_shared(std::move(selection))}; - }); + auto shape_fut = itp->RunAsync( + [column = column, selection = selection, + data = data](TableProxy& tp) mutable -> Result { + MaybeReopenRW(tp); + ARROW_RETURN_NOT_OK(ColumnExists(tp.table(), column)); + auto table_column = TableColumn(tp.table(), column); + ARROW_ASSIGN_OR_RAISE(auto shape_data, + ResultShapeData::MakeWrite(table_column, data, selection)); + return ShapeResult{std::make_shared(std::move(shape_data)), + std::make_shared(std::move(selection))}; + }, + LockType::Write); // Partition the resulting shape into contiguous chunks // of data to write to disk diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index 4a0a904d..d50407f9 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -26,6 +26,10 @@ add_executable(parallel_write_test parallel_write_test.cc) target_link_libraries(parallel_write_test PRIVATE GTest::gtest_main arcae test_utils) add_test(parallel_write_test parallel_write_test) +add_executable(concurrent_rw_test concurrent_rw_test.cc) +target_link_libraries(concurrent_rw_test PRIVATE GTest::gtest_main arcae test_utils) +add_test(concurrent_rw_test concurrent_rw_test) + add_executable(data_partition_test data_partition_test.cc) target_link_libraries(data_partition_test PRIVATE GTest::gtest_main arcae test_utils) add_test(data_partition_test data_partition_test) @@ -53,5 +57,6 @@ set_tests_properties(result_shape_test isolated_table_proxy_test new_table_proxy_test parallel_write_test + concurrent_rw_test selection_test PROPERTIES ENVIRONMENT "LD_LIBRARY_PATH=${VCPKG_LIBDIRS}") diff --git a/cpp/tests/concurrent_rw_test.cc b/cpp/tests/concurrent_rw_test.cc new file mode 100644 index 00000000..aa9d868d --- /dev/null +++ b/cpp/tests/concurrent_rw_test.cc @@ -0,0 +1,519 @@ +// With ninstances > 1, a writer (instance 0, via SpawnWriter) must exclude +// readers running on the *other* instances. The patched FileLocker only +// coordinates multi-reader/single-writer access among threads sharing a +// single FileLocker object; because arcae gives each instance its own +// TableProxy -> Table -> LockFile -> FileLocker, the +// coordination historically did not span instances and a reader could acquire a +// read lock on the same byte range while a writer held the write lock (silently +// downgrading it at the OS level). +// +// WriterExcludesReadersAcrossInstances is deterministic: it fails on the +// pre-fix build (readers observe the writer in its critical section) and passes +// once the lock state is shared across instances. +// +// The *AcrossProcesses tests cover the orthogonal per-process axis. Across +// distinct PIDs, fcntl(F_SETLK) exclusion is casacore's genuine, original +// mechanism (it worked even pre-patch), so these are NOT a regression gate for +// Issue 1. Their job is to guard that the patch's rewrites -- HashThreadId +// reqIds replacing pid, the refcounted shared-fd LockFile cache, the new +// LockState -- did not BREAK the inter-process exclusion casacore always had, +// and that a reader in another process observes a writer's committed data via +// casacore's getInfo/putInfo lock-file handshake (the real cross-process +// freshness path; the thread_local TableCache concern is in-process only). +// +// They fork() before any IsolatedTableProxy (hence any Arrow/casacore thread) +// is created, so the forking process is still effectively single-threaded and +// the threaded-fork lock-inheritance hazard is avoided. Children coordinate via +// an anonymous MAP_SHARED region and communicate failure via exit code; all +// gtest assertions run in the parent after reaping, because ASSERT/EXPECT in a +// forked child do not propagate to the parent's test result. + +#include +#include +#include +#include +#include + +#ifndef _WIN32 +#include +#include +#include + +#include +#include +#endif + +#include "gtest/gtest.h" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "arcae/isolated_table_proxy.h" + +using ::arcae::detail::IsolatedTableProxy; +using ::arrow::Result; + +using LockType = ::casacore::FileLocker::LockType; +using ::casacore::Array; +using ::casacore::ArrayColumn; +using ::casacore::ArrayColumnDesc; +using ::casacore::ColumnDesc; +using CasaInt = ::casacore::Int; +using MS = ::casacore::MeasurementSet; +using ::casacore::Record; +using ::casacore::SetupNewTable; +using ::casacore::Table; +using ::casacore::TableColumn; +using ::casacore::TableDesc; +using ::casacore::TableProxy; +using IPos = ::casacore::IPosition; + +using namespace std::string_literals; +using namespace std::chrono_literals; + +namespace { + +static constexpr std::size_t knrow = 512; +static constexpr std::size_t knchan = 16; +static constexpr std::size_t kncorr = 4; + +class ConcurrentReadWriteTest : public ::testing::Test { + protected: + std::string table_name_; + + void SetUp() override { + auto* ti = ::testing::UnitTest::GetInstance()->current_test_info(); + table_name_ = std::string(ti->name() + "-"s + arcae::hexuuid(4) + ".table"s); + + auto table_desc = TableDesc(MS::requiredTableDesc()); + auto data_shape = IPos({kncorr, knchan}); + table_desc.addColumn( + ArrayColumnDesc("DATA", data_shape, ColumnDesc::FixedShape)); + auto setup = SetupNewTable(table_name_, table_desc, Table::New); + auto ms = MS(setup, knrow); + auto col = ArrayColumn(TableColumn(ms, "DATA")); + col.putColumn(Array(IPos({kncorr, knchan, knrow}), CasaInt(0))); + } + + Result> OpenTable(std::size_t ninstances) { + return IsolatedTableProxy::Make( + [name = table_name_]() { + auto lockoptions = Record(); + lockoptions.define("option", "user"); + auto tp = std::make_shared(name, lockoptions, Table::Old); + tp->reopenRW(); + return tp; + }, + ninstances); + } +}; + +// Deterministic gate: a writer on instance 0 must exclude readers on other +// instances for the duration that it holds the write lock. +TEST_F(ConcurrentReadWriteTest, WriterExcludesReadersAcrossInstances) { + ASSERT_OK_AND_ASSIGN(auto itp, OpenTable(/*ninstances=*/2)); + auto writer = itp->SpawnWriter(); + + std::atomic writer_in_critical{false}; + std::atomic reader_saw_writer{false}; + std::atomic reads_completed{0}; + + // The writer holds the write lock for a fixed window. The flag is only ever + // true while the write lock is held (MaybeLockAndFinalise acquires before the + // functor runs and releases after it returns). + auto wfut = writer->RunAsync( + [&](TableProxy&) -> Result { + writer_in_critical = true; + std::this_thread::sleep_for(400ms); + writer_in_critical = false; + return true; + }, + LockType::Write); + + // Let the writer enter its critical section. + std::this_thread::sleep_for(80ms); + ASSERT_TRUE(writer_in_critical.load()) << "writer did not enter its critical section"; + + // Fan readers out across the remaining instances while the writer holds the + // lock. A reader that runs its functor (i.e. holds a read lock) while the + // writer is still in its critical section proves the exclusion is broken. + std::vector> rfuts; + while (writer_in_critical.load()) { + rfuts.push_back(itp->RunAsync( + [&](const TableProxy&) -> Result { + if (writer_in_critical.load()) reader_saw_writer = true; + return true; + }, + LockType::Read)); + std::this_thread::sleep_for(5ms); + } + + for (auto& f : rfuts) { + auto r = f.MoveResult(); + if (r.ok()) ++reads_completed; + } + ASSERT_OK(wfut.status()); + + EXPECT_FALSE(reader_saw_writer.load()) + << "A reader acquired a read lock while a writer held the write lock on " + "another instance (cross-instance MRSW exclusion failed). " + << reads_completed.load() << " reads completed."; + + // The writer borrows instance 0 from itp; its custom deleter releases the + // borrowed proxy without closing it, so we drop it (rather than Close it, + // which would double-close the shared proxy) before closing the parent. + writer.reset(); + ASSERT_OK(itp->Close()); +} + +// Stress gate: a writer repeatedly writes a uniform value to every row while +// readers fan across the other instances. A correctly excluded read always +// observes an internally uniform column; a torn read (mixed old/new values) +// indicates the writer was not excluded or stale buffers were served. +TEST_F(ConcurrentReadWriteTest, NoTornReadsAcrossInstances) { + static constexpr int knwrites = 200; + ASSERT_OK_AND_ASSIGN(auto itp, OpenTable(/*ninstances=*/4)); + auto writer = itp->SpawnWriter(); + + std::atomic writing{true}; + std::atomic torn_reads{0}; + std::atomic reads_completed{0}; + + std::thread writer_thread([&]() { + for (int w = 1; w <= knwrites; ++w) { + auto status = writer->RunSync( + [w](TableProxy& tp) -> Result { + auto col = ArrayColumn(TableColumn(tp.table(), "DATA")); + col.putColumn(Array(IPos({kncorr, knchan, knrow}), CasaInt(w))); + return true; + }, + LockType::Write); + if (!status.ok()) { + ADD_FAILURE() << "write failed: " << status.status(); + break; + } + } + writing = false; + }); + + while (writing.load()) { + std::vector> rfuts; + for (int i = 0; i < 16; ++i) { + rfuts.push_back(itp->RunAsync( + [](const TableProxy& tp) -> Result { + auto col = ArrayColumn(TableColumn(tp.table(), "DATA")); + auto data = col.getColumn(); + // Uniform write => min == max for a non-torn read. + return casacore::min(data) == casacore::max(data); + }, + LockType::Read)); + } + for (auto& f : rfuts) { + auto r = f.MoveResult(); + if (r.ok()) { + ++reads_completed; + if (!*r) ++torn_reads; + } + } + } + + writer_thread.join(); + EXPECT_EQ(torn_reads.load(), 0) + << "observed torn reads (" << reads_completed.load() << " reads total)"; + + writer.reset(); + ASSERT_OK(itp->Close()); +} + +#ifndef _WIN32 + +// Coordination region shared between the forked children and the parent. All +// fields are lock-free atomics living in anonymous MAP_SHARED memory mapped +// before fork(), so every process sees the same bytes. +struct SharedState { + std::atomic ready; // barrier: children opened their tables + std::atomic writer_in_critical; // writer holds the write lock right now + std::atomic reader_saw_writer; // a reader ran while writer_in_critical + std::atomic writing; // writer loop is still running (torn-read test) + std::atomic torn_reads; // reads observing min != max + std::atomic reads_completed; // successful reads (diagnostic) +}; + +namespace { + +SharedState* MapSharedState() { + void* p = ::mmap(nullptr, sizeof(SharedState), PROT_READ | PROT_WRITE, + MAP_SHARED | MAP_ANONYMOUS, -1, 0); + if (p == MAP_FAILED) return nullptr; + auto* s = new (p) SharedState(); + s->ready.store(0); + s->writer_in_critical.store(false); + s->reader_saw_writer.store(false); + s->writing.store(true); + s->torn_reads.store(0); + s->reads_completed.store(0); + return s; +} + +void UnmapSharedState(SharedState* s) { + if (s) ::munmap(static_cast(s), sizeof(SharedState)); +} + +// Spin until every child has reached the barrier (or a generous deadline), so +// the writer's critical window genuinely overlaps reader activity. +void WaitForBarrier(SharedState* s, int total) { + auto deadline = std::chrono::steady_clock::now() + 5s; + while (s->ready.load() < total && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(1ms); + } +} + +// Reap all children, killing any that outlive the deadline so a cross-process +// deadlock fails the test instead of hanging CI. Returns true iff every child +// exited 0 within the deadline. +bool ReapAll(const std::vector& pids, std::chrono::milliseconds timeout) { + auto deadline = std::chrono::steady_clock::now() + timeout; + std::vector reaped(pids.size(), false); + std::vector codes(pids.size(), -1); + std::size_t outstanding = pids.size(); + + while (outstanding > 0 && std::chrono::steady_clock::now() < deadline) { + for (std::size_t i = 0; i < pids.size(); ++i) { + if (reaped[i]) continue; + int status = 0; + pid_t r = ::waitpid(pids[i], &status, WNOHANG); + if (r == pids[i]) { + reaped[i] = true; + --outstanding; + codes[i] = (WIFEXITED(status) ? WEXITSTATUS(status) : 128); + } + } + if (outstanding > 0) std::this_thread::sleep_for(10ms); + } + + bool all_ok = true; + for (std::size_t i = 0; i < pids.size(); ++i) { + if (!reaped[i]) { + ::kill(pids[i], SIGKILL); + ::waitpid(pids[i], nullptr, 0); + ADD_FAILURE() << "child " << pids[i] + << " did not exit within the deadline " + "(possible cross-process deadlock)"; + all_ok = false; + } else if (codes[i] != 0) { + ADD_FAILURE() << "child " << pids[i] << " exited with code " << codes[i]; + all_ok = false; + } + } + return all_ok; +} + +// Child exit codes (parent reports these on failure). +enum ChildExit { kChildOk = 0, kChildOpenFailed = 2, kChildWriteFailed = 3 }; + +} // namespace + +// A writer process holding the write lock must exclude reader processes for the +// duration it holds it. This is the per-process analogue of +// WriterExcludesReadersAcrossInstances; it guards casacore's inter-process +// fcntl exclusion against regressions in the patch's reqId/shared-fd rewrites. +TEST_F(ConcurrentReadWriteTest, WriterExcludesReadersAcrossProcesses) { + static constexpr int knreaders = 3; + static constexpr int ntotal = knreaders + 1; + + SharedState* s = MapSharedState(); + ASSERT_NE(s, nullptr) << "mmap of shared state failed"; + + std::vector pids; + bool is_child = false; + bool child_is_writer = false; + + for (int i = 0; i < ntotal && !is_child; ++i) { + pid_t pid = ::fork(); + ASSERT_NE(pid, -1) << "fork failed"; + if (pid == 0) { + is_child = true; + child_is_writer = (i == 0); + } else { + pids.push_back(pid); + } + } + + if (is_child) { + // ---- writer child ---- + if (child_is_writer) { + auto itp_res = OpenTable(/*ninstances=*/1); + if (!itp_res.ok()) ::_exit(kChildOpenFailed); + auto itp = *itp_res; + auto writer = itp->SpawnWriter(); + s->ready.fetch_add(1); + WaitForBarrier(s, ntotal); + + // The flag is only ever set while the write lock is held: the lock is + // acquired by MaybeLockAndFinalise before the functor runs and released + // after it returns. + auto fut = writer->RunAsync( + [&](TableProxy&) -> Result { + s->writer_in_critical.store(true); + std::this_thread::sleep_for(400ms); + s->writer_in_critical.store(false); + return true; + }, + LockType::Write); + auto r = fut.MoveResult(); + ::_exit(r.ok() ? kChildOk : kChildWriteFailed); + } + + // ---- reader child ---- + auto itp_res = OpenTable(/*ninstances=*/2); + if (!itp_res.ok()) ::_exit(kChildOpenFailed); + auto itp = *itp_res; + s->ready.fetch_add(1); + WaitForBarrier(s, ntotal); + + // Wait (bounded) for the writer to enter its critical section, then hammer + // it with reads for as long as it stays there. A read that runs its functor + // while the writer flag is up proves the inter-process exclusion is broken. + // A read that blocks until release, or fails fast under contention, is fine. + auto wstart = std::chrono::steady_clock::now(); + while (!s->writer_in_critical.load() && + std::chrono::steady_clock::now() - wstart < 2s) { + std::this_thread::sleep_for(1ms); + } + while (s->writer_in_critical.load()) { + auto fut = itp->RunAsync( + [&](const TableProxy&) -> Result { + if (s->writer_in_critical.load()) s->reader_saw_writer.store(true); + return true; + }, + LockType::Read); + auto r = fut.MoveResult(); + if (r.ok()) s->reads_completed.fetch_add(1); + std::this_thread::sleep_for(2ms); + } + ::_exit(kChildOk); + } + + // ---- parent ---- + bool all_ok = ReapAll(pids, /*timeout=*/30s); + EXPECT_TRUE(all_ok); + EXPECT_FALSE(s->reader_saw_writer.load()) + << "a reader process acquired a read lock while a writer process held the " + "write lock (inter-process MRSW exclusion failed). " + << s->reads_completed.load() << " reads completed."; + UnmapSharedState(s); +} + +// A writer process repeatedly writes a uniform value while reader processes read +// the same column. A correctly synchronised read always observes an internally +// uniform column; a torn read (mixed old/new values) means the reader process +// was not excluded or served stale storage-manager buffers -- i.e. casacore's +// cross-process getInfo/putInfo freshness handshake did not fire. +TEST_F(ConcurrentReadWriteTest, NoTornReadsAcrossProcesses) { + static constexpr int knreaders = 2; + static constexpr int ntotal = knreaders + 1; + static constexpr int knwrites = 50; + + SharedState* s = MapSharedState(); + ASSERT_NE(s, nullptr) << "mmap of shared state failed"; + + std::vector pids; + bool is_child = false; + bool child_is_writer = false; + + for (int i = 0; i < ntotal && !is_child; ++i) { + pid_t pid = ::fork(); + ASSERT_NE(pid, -1) << "fork failed"; + if (pid == 0) { + is_child = true; + child_is_writer = (i == 0); + } else { + pids.push_back(pid); + } + } + + if (is_child) { + // ---- writer child ---- + if (child_is_writer) { + auto itp_res = OpenTable(/*ninstances=*/1); + if (!itp_res.ok()) ::_exit(kChildOpenFailed); + auto itp = *itp_res; + auto writer = itp->SpawnWriter(); + s->ready.fetch_add(1); + WaitForBarrier(s, ntotal); + + int code = kChildOk; + for (int w = 1; w <= knwrites; ++w) { + auto status = writer->RunSync( + [w](TableProxy& tp) -> Result { + auto col = ArrayColumn(TableColumn(tp.table(), "DATA")); + col.putColumn(Array(IPos({kncorr, knchan, knrow}), CasaInt(w))); + return true; + }, + LockType::Write); + if (!status.ok()) { + code = kChildWriteFailed; + break; + } + } + s->writing.store(false); + ::_exit(code); + } + + // ---- reader child ---- + auto itp_res = OpenTable(/*ninstances=*/2); + if (!itp_res.ok()) ::_exit(kChildOpenFailed); + auto itp = *itp_res; + s->ready.fetch_add(1); + WaitForBarrier(s, ntotal); + + // Safety deadline guards against a wedged writer; the parent watchdog backs + // it up by killing survivors. + auto deadline = std::chrono::steady_clock::now() + 60s; + while (s->writing.load() && std::chrono::steady_clock::now() < deadline) { + auto fut = itp->RunAsync( + [](const TableProxy& tp) -> Result { + auto col = ArrayColumn(TableColumn(tp.table(), "DATA")); + auto data = col.getColumn(); + // Uniform write => min == max for a non-torn read. + return casacore::min(data) == casacore::max(data); + }, + LockType::Read); + auto r = fut.MoveResult(); + if (r.ok()) { + s->reads_completed.fetch_add(1); + if (!*r) s->torn_reads.fetch_add(1); + } + // Back off between reads: continuously re-acquired read locks starve the + // cross-process writer (fcntl read/write contention), which both slows the + // test to a crawl and erodes overlap. A short pause keeps ample read/write + // interleaving while letting the writer make progress. + std::this_thread::sleep_for(2ms); + } + ::_exit(kChildOk); + } + + // ---- parent ---- + bool all_ok = ReapAll(pids, /*timeout=*/120s); + EXPECT_TRUE(all_ok); + EXPECT_EQ(s->torn_reads.load(), 0) << "observed torn reads across processes (" + << s->reads_completed.load() << " reads total)"; + UnmapSharedState(s); +} + +#endif // _WIN32 + +} // namespace diff --git a/cpp/tests/isolated_table_proxy_test.cc b/cpp/tests/isolated_table_proxy_test.cc index 979f81c3..bf2367e4 100644 --- a/cpp/tests/isolated_table_proxy_test.cc +++ b/cpp/tests/isolated_table_proxy_test.cc @@ -65,7 +65,7 @@ class IsolatedTableProxyTest : public ::testing::Test { return IsolatedTableProxy::Make([name = table_name_]() { auto lock = TableLock(TableLock::LockOption::AutoLocking); auto lockoptions = Record(); - lockoptions.define("option", "nolock"); + lockoptions.define("option", "user"); lockoptions.define("internal", lock.interval()); lockoptions.define("maxwait", casacore::Int(lock.maxWait())); return std::make_shared(name, lockoptions, Table::Old); diff --git a/cpp/tests/new_table_proxy_test.cc b/cpp/tests/new_table_proxy_test.cc index cb0342c0..8a77155c 100644 --- a/cpp/tests/new_table_proxy_test.cc +++ b/cpp/tests/new_table_proxy_test.cc @@ -107,7 +107,7 @@ class ZeroRowTableProxyTest : public ::testing::Test { return NewTableProxy::Make([name = table_name_]() { auto lock = TableLock(TableLock::LockOption::AutoLocking); auto lockoptions = Record(); - lockoptions.define("option", "nolock"); + lockoptions.define("option", "user"); lockoptions.define("internal", lock.interval()); lockoptions.define("maxwait", casacore::Int(lock.maxWait())); return std::make_shared(name, lockoptions, Table::Old); @@ -236,7 +236,7 @@ class FixedTableProxyTest : public ::testing::TestWithParam { [&, name = table_name_]() { auto lock = TableLock(TableLock::LockOption::AutoLocking); auto lockoptions = Record(); - lockoptions.define("option", "nolock"); + lockoptions.define("option", "user"); lockoptions.define("internal", lock.interval()); lockoptions.define("maxwait", casacore::Int(lock.maxWait())); auto tp = std::make_shared(name, lockoptions, Table::Old); @@ -675,7 +675,7 @@ class VariableProxyTest : public ::testing::TestWithParam { return NewTableProxy::Make([name = table_name_]() { auto lock = TableLock(TableLock::LockOption::AutoLocking); auto lockoptions = Record(); - lockoptions.define("option", "nolock"); + lockoptions.define("option", "user"); lockoptions.define("internal", lock.interval()); lockoptions.define("maxwait", casacore::Int(lock.maxWait())); return std::make_shared(name, lockoptions, Table::Old); diff --git a/cpp/tests/parallel_write_test.cc b/cpp/tests/parallel_write_test.cc index bae1f049..76afe0cc 100644 --- a/cpp/tests/parallel_write_test.cc +++ b/cpp/tests/parallel_write_test.cc @@ -1,3 +1,18 @@ +// This test case writes to a Tiled Storage Manager at tile boundaries from +// multiple threads, each through its own IsolatedTableProxy over the same +// table. (In practice, arcae only writes from a single thread at a time.) +// +// Each write takes an explicit user-locking write lock (LockType::Write), so +// the patched FileLocker's shared LockState serialises the writers in-process +// (multi-reader/single-writer). This both prevents the Tiled Storage Manager +// header corruption that an unsynchronised version exhibits ("FilebufIO:: +// readBlock - incorrect number of bytes" against "table.f0") and exercises the +// cross-instance write-lock coordination. See: +// 1. CAS-13609 in https://casadocs.readthedocs.io/en/v6.4.3/changelog.html +// 2. https://keflavich.github.io/blog/casa-reading-incorrect-number-of-bytes-wmpi.html + +#include + #include #include #include @@ -8,17 +23,18 @@ #include #include +#include #include #include #include #include -#include using ::arrow::Result; using ::arcae::detail::IsolatedTableProxy; +using LockType = ::casacore::FileLocker::LockType; using ::casacore::Array; using ::casacore::ArrayColumn; using ::casacore::ArrayColumnDesc; @@ -78,10 +94,10 @@ class WriteTests : public ::testing::Test { arrow::Result> OpenTable() { return IsolatedTableProxy::Make([name = table_name_]() { - auto lock = TableLock(TableLock::LockOption::AutoLocking); + auto lock = TableLock(TableLock::LockOption::UserLocking); auto lockoptions = Record(); - lockoptions.define("option", "nolock"); - lockoptions.define("internal", lock.interval()); + lockoptions.define("option", "user"); + lockoptions.define("interval", lock.interval()); lockoptions.define("maxwait", casacore::Int(lock.maxWait())); auto tp = std::make_shared(name, lockoptions, Table::Old); tp->reopenRW(); @@ -122,13 +138,13 @@ TEST_F(WriteTests, Parallel) { try { column.putColumnRange(Slice(start, nrow), data); - table.flush(); } catch (std::exception& e) { return arrow::Status::Invalid("Write failed ", e.what()); } return true; - }); + }, + LockType::Write); })); futures.push_back(result); @@ -156,6 +172,8 @@ TEST_F(WriteTests, Parallel) { auto table_column = TableColumn(table, "MODEL_DATA"); const auto& column_desc = table_column.columnDesc(); auto column = ArrayColumn(table_column); + table.lock(false, 0); + std::shared_ptr result(nullptr, [&](...) { table.unlock(); }); try { return column.getColumnRange(Slice(start, nrow)); diff --git a/pyproject.toml b/pyproject.toml index 45dd1ba9..60deb944 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "arcae" -version = "0.5.4" +version = "0.4.0-alpha.8" authors = [ {name = "Simon Perkins", email = "simon.perkins@gmail.com"} ] diff --git a/src/arcae/__init__.py b/src/arcae/__init__.py index 5fa35160..c430132d 100644 --- a/src/arcae/__init__.py +++ b/src/arcae/__init__.py @@ -5,7 +5,7 @@ if TYPE_CHECKING: from arcae.lib.arrow_tables import Table -__version__ = "0.5.4" +__version__ = "0.4.0-alpha.8" def safe_multithreaded_writes() -> bool: diff --git a/src/arcae/lib/arrow_tables.pxd b/src/arcae/lib/arrow_tables.pxd index cd0bcf98..05ff51b0 100644 --- a/src/arcae/lib/arrow_tables.pxd +++ b/src/arcae/lib/arrow_tables.pxd @@ -99,6 +99,7 @@ cdef extern from "arcae/table_factory.h" namespace "arcae" nogil: cdef CResult[shared_ptr[CCasaTable]] CDefaultMS" arcae::DefaultMS"( const string & name, const string & subtable, + size_t ninstances, const string & json_table_desc, const string & json_dminfo, const string & json_cache_size) diff --git a/src/arcae/lib/arrow_tables.pyx b/src/arcae/lib/arrow_tables.pyx index c51067af..3269b460 100644 --- a/src/arcae/lib/arrow_tables.pyx +++ b/src/arcae/lib/arrow_tables.pyx @@ -6,6 +6,7 @@ from collections.abc import MutableMapping, Sequence import cython import json from typing import Any, Dict, List, Union +import warnings from libcpp cimport bool from libcpp.memory cimport shared_ptr @@ -201,6 +202,13 @@ cdef class Table: size_t cninstances = ninstances Table table = Table.__new__(Table) + if lockoptions != "auto": + warnings.warn( + f"lockoptions is deprecated: " + f"'{lockoptions}' will not be applied.", + DeprecationWarning + ) + if isinstance(lockoptions, str): lockoptions = f"{{\"option\": \"{lockoptions}\"}}" elif isinstance(lockoptions, dict): @@ -221,6 +229,7 @@ cdef class Table: def ms_from_descriptor( filename: str, subtable: str = "MAIN", + ninstances: int = 1, table_desc: Dict | None = None, dminfo: Dict | None = None, cache_size: Union[int, dict, None] = None @@ -229,6 +238,7 @@ cdef class Table: Table table = Table.__new__(Table) string cfilename = tobytes(filename) string csubtable = tobytes(subtable) + size_t cninstances = int(ninstances) json_table_desc = json.dumps(table_desc) if table_desc else "{}" json_dminfo = json.dumps(dminfo) if dminfo else "{}" @@ -239,6 +249,7 @@ cdef class Table: with nogil: table.c_table = GetResultValue(CDefaultMS(cfilename, csubtable, + cninstances, cjson_table_desc, cjson_dm_info, cjson_cache_size)) diff --git a/src/arcae/testing.py b/src/arcae/testing.py index ff893731..95348f81 100644 --- a/src/arcae/testing.py +++ b/src/arcae/testing.py @@ -71,7 +71,7 @@ def sanity(): dir = stack.enter_context(TemporaryDirectory(prefix="arcae-sanity")) ms = os.path.join(dir, "sanity.ms") T = stack.enter_context( - Table.ms_from_descriptor(ms, "MAIN", TABLE_DESC, DMINFO) + Table.ms_from_descriptor(ms, "MAIN", table_desc=TABLE_DESC, dminfo=DMINFO) ) T.addrows(time.size) T.putcol("TIME", time) diff --git a/src/arcae/tests/test_descriptor.py b/src/arcae/tests/test_descriptor.py index 6e876977..e10753db 100644 --- a/src/arcae/tests/test_descriptor.py +++ b/src/arcae/tests/test_descriptor.py @@ -31,20 +31,20 @@ def test_ms_and_weather_subtable(tmp_path_factory): # Basic descriptor table_desc = ms_descriptor("WEATHER", complete=False) - with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc) as W: + with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc=table_desc) as W: assert (ms / "WEATHER").exists() assert W.columns() == ["ANTENNA_ID", "INTERVAL", "TIME"] # Add a column to the basic descriptor table_desc = ms_descriptor("WEATHER", complete=False) table_desc["BLAH"] = table_desc["TIME"].copy() - with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc) as W: + with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc=table_desc) as W: assert (ms / "WEATHER").exists() assert W.columns() == ["ANTENNA_ID", "BLAH", "INTERVAL", "TIME"] # Complete descriptor table_desc = ms_descriptor("WEATHER", complete=True) - with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc) as W: + with Table.ms_from_descriptor(str(ms), "WEATHER", table_desc=table_desc) as W: assert (ms / "WEATHER").exists() assert W.columns() == [ "ANTENNA_ID", diff --git a/src/arcae/tests/test_multithreaded_writes.py b/src/arcae/tests/test_multithreaded_writes.py index 476c4717..611c17ab 100644 --- a/src/arcae/tests/test_multithreaded_writes.py +++ b/src/arcae/tests/test_multithreaded_writes.py @@ -1,6 +1,4 @@ -import pytest from numpy.testing import assert_array_equal -from pyarrow.lib import ArrowNotImplementedError import arcae @@ -8,7 +6,7 @@ def test_safe_multithreaded_writes(): """Assert that this version of arcae does not support multithreaded writes""" - assert not arcae.safe_multithreaded_writes() + assert arcae.safe_multithreaded_writes() def test_writes_succeeds_ninstances_1(column_case_table): @@ -20,11 +18,8 @@ def test_writes_succeeds_ninstances_1(column_case_table): def test_writes_fail_ninstances_2(column_case_table): - """Test that attempting to write when ninstances > 1 fails""" + """Test that writing when ninstances > 1 succeeds""" with arcae.table(column_case_table, ninstances=2, readonly=False) as T: data = T.getcol("FIXED") - with pytest.raises( - ArrowNotImplementedError, - match="Write support when number of table instances 2", - ): - T.putcol("FIXED", data + 1) + T.putcol("FIXED", data + 1) + assert_array_equal(T.getcol("FIXED"), data + 1) diff --git a/tbump.toml b/tbump.toml index cdc30867..07154cee 100644 --- a/tbump.toml +++ b/tbump.toml @@ -1,11 +1,14 @@ [version] -current = "0.5.4" +current = "0.4.0-alpha.8" +# https://semver.org/#spec-item-9 regex = ''' (?P\d+) \. (?P\d+) \. (?P\d+) + -? + (?P(alpha|beta|rc)\.(?P\d+))? ''' [git] diff --git a/vcpkg/overlay-ports/casacore/001-casacore-cmake.patch b/vcpkg/overlay-ports/casacore/001-casacore-cmake.patch index 7125bfd4..5c32dd26 100644 --- a/vcpkg/overlay-ports/casacore/001-casacore-cmake.patch +++ b/vcpkg/overlay-ports/casacore/001-casacore-cmake.patch @@ -139,6 +139,960 @@ index 12229f9..a15e508 100644 // Base class for all Casacore library errors +diff --git a/casa/IO/FileLocker.cc b/casa/IO/FileLocker.cc +index b55c080..774a164 100644 +--- a/casa/IO/FileLocker.cc ++++ b/casa/IO/FileLocker.cc +@@ -24,158 +24,363 @@ + //# Charlottesville, VA 22903-2475 USA + + #include +-#include +-#include ++ + #include + #include + #include +-#include ++ ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++ ++#include ++#include ++#include ++ ++ ++using namespace std::chrono_literals; + + //# Locking is not supported on Cray compute nodes. + #if defined(AIPS_CRAY_PGI) && !defined(AIPS_NOFILELOCK) + # define AIPS_NOFILELOCK 1 + #endif + ++namespace { ++ ++template ++struct Finally { ++ Fn fn; ++ bool enabled = true; ++ ~Finally() { if (enabled) fn(); }; ++}; ++ ++template ++auto finally(Fn && fn) { ++ return Finally{std::forward(fn), true}; ++} ++ ++} // namespace + + namespace casacore { //# NAMESPACE CASACORE - BEGIN + ++// Per-(lock-file path, start, length) coordination state. A single process may ++// open the same table from multiple threads/instances, each with its own ++// FileLocker over a shared file descriptor. POSIX advisory locks are owned per ++// (process, inode), so a second lock request on an overlapping range from the ++// same process silently replaces the existing one rather than blocking. To get ++// real multi-reader/single-writer semantics across those FileLockers, they all ++// share one LockState and only the first acquirer takes (and the last releaser ++// drops) the real fcntl lock. ++struct LockState { ++ mutable std::recursive_mutex mutex; ++ std::condition_variable_any condVar; ++ std::unordered_map lockType; ++ bool modifyingFcntlLock = false; ++ int wantsWriteLock = 0; ++ int error = 0; ++}; ++ ++namespace { ++ ++using LockStateKey = std::tuple; ++ ++std::mutex & lock_state_registry_mutex() { ++ static std::mutex m; ++ return m; ++} ++ ++std::map> & lock_state_registry() { ++ static std::map> registry; ++ return registry; ++} ++ ++// Return the LockState shared by every FileLocker locking the given ++// (name, start, length), creating it if necessary. Expired entries are pruned ++// lazily. Keyed on the absolute lock-file path (stable, avoids fd-recycling ++// races). ++std::shared_ptr getOrCreateLockState( ++ const String & name, uInt start, uInt length) ++{ ++ std::lock_guard guard(lock_state_registry_mutex()); ++ auto & registry = lock_state_registry(); ++ auto key = LockStateKey(std::string(name), start, length); ++ if (auto it = registry.find(key); it != std::end(registry)) { ++ if (auto state = it->second.lock()) return state; ++ } ++ auto state = std::make_shared(); ++ registry[key] = state; ++ // Prune expired weak references ++ for (auto it = std::begin(registry); it != std::end(registry); ) { ++ if (it->second.expired()) it = registry.erase(it); ++ else ++it; ++ } ++ return state; ++} ++ ++} // namespace ++ + FileLocker::FileLocker() +-: itsFD (-1), +- itsError (0), +- itsStart (0), +- itsLength (0), +- itsMsgShown (False), +- itsReadLocked (False), +- itsWriteLocked (False) ++: itsFD (-1), ++ itsStart (0), ++ itsLength (0), ++ itsMsgShown (False), ++ itsState (std::make_shared()) + {} + + FileLocker::FileLocker (int fd, uInt start, uInt length) +-: itsFD (fd), +- itsError (0), +- itsStart (start), +- itsLength (length), +- itsMsgShown (False), +- itsReadLocked (False), +- itsWriteLocked (False) ++: itsFD (fd), ++ itsStart (start), ++ itsLength (length), ++ itsMsgShown (False), ++ itsState (std::make_shared()) + {} + +-FileLocker::~FileLocker() ++FileLocker::FileLocker (int fd, const String& name, uInt start, uInt length) ++: itsFD (fd), ++ itsStart (start), ++ itsLength (length), ++ itsMsgShown (False), ++ itsState (getOrCreateLockState(name, start, length)) + {} + +-Bool FileLocker::acquire (LockType type, uInt nattempts) ++FileLocker::FileLocker(FileLocker&& rhs) : ++ itsFD(std::exchange(rhs.itsFD, -1)), ++ itsStart(std::exchange(rhs.itsStart, 0)), ++ itsLength(std::exchange(rhs.itsLength, 0)), ++ itsMsgShown(rhs.itsMsgShown), ++ itsState(std::move(rhs.itsState)) + { +- itsError = 0; +- // Always success if locking is not supported. +-#if defined(AIPS_NOFILELOCK) +- itsReadLocked = True; +- if (!itsWriteLocked && type == Write) { +- itsWriteLocked = True; ++ // The moved-from FileLocker must still be destructible/assignable; give it ++ // a fresh private state rather than leaving it null. ++ rhs.itsState = std::make_shared(); ++} ++ ++FileLocker& FileLocker::operator=(FileLocker&& rhs) { ++ if (this != &rhs) { ++ std::swap(itsFD, rhs.itsFD); ++ std::swap(itsStart, rhs.itsStart); ++ std::swap(itsLength, rhs.itsLength); ++ std::swap(itsMsgShown, rhs.itsMsgShown); ++ std::swap(itsState, rhs.itsState); + } ++ return *this; ++} ++ ++FileLocker::~FileLocker() ++{ ++} ++ ++ ++// A conflicting lock may result in EAGAIN or EACCESS, depending ++// on the POSIX implementation. ENLOCK is deemed to arise from ++// NFS systems where statd or lockd daemons are not running ++bool ShouldRetry(int err) { ++ auto result = err == EAGAIN || err == EACCES; ++ ++ // Treat ENOLCK as success in an NFS context ++ #if defined(AIPS_LINUX) || defined(AIPS_DARWIN) ++ result = result || err == ENOLCK; ++ #endif ++ return result; ++} ++ ++ ++Bool FileLocker::fcntlAquire(RecursiveMutexLock & state_lock, LockType type, uInt nattempts) { ++ #if defined(AIPS_NOFILELOCK) + return True; +-#else +- struct flock ls; +- ls.l_whence = SEEK_SET; +- ls.l_start = itsStart; +- ls.l_len = itsLength; +- ls.l_type = F_WRLCK; +- // When a read-lock is acquired, it may release an existing write-lock. +- // We do not want that to happen, so when it is write-locked, test +- // if the write-lock is still valid. +- if (type == Read) { +- if (itsWriteLocked) { +- if (fcntl (itsFD, F_SETLK, &ls) != -1) { +-/// cout << "kept " << itsReadLocked << ' ' <error = 0; ++ ++ // nattempts == 0 restores the original casacore semantics of waiting until ++ // the lock is acquired (retrying with a 1s backoff); nattempts > 0 makes a ++ // finite number of attempts. In-process conflicts are already resolved by ++ // the LockState coordination before we get here, so any contention fcntl ++ // reports comes from other processes. ++ for (uInt a = 0; nattempts == 0 || a < nattempts; ++a) { ++ if (fcntl(itsFD, F_SETLK, &lock) != -1) return True; ++ itsState->error = errno; ++ if (!ShouldRetry(itsState->error)) break; ++ itsState->error = 0; ++ // Release the coordination lock while sleeping so other threads can ++ // make progress; we still hold the fcntl gate (modifyingFcntlLock). ++ state_lock.unlock(); ++ std::this_thread::sleep_for(1s); ++ state_lock.lock(); + } +- if (nattempts == 0) { +- // Wait until lock succeeds. +- if (fcntl (itsFD, F_SETLKW, &ls) != -1) { +- itsReadLocked = True; +- if (type == Write) { +- itsWriteLocked = True; +- } +-/// cout << "acquired " << itsReadLocked << ' ' <mutex); ++ auto tid = std::this_thread::get_id(); ++ LockType thread_lock_type = None; ++ if (auto it = itsState->lockType.find(tid); it != std::end(itsState->lockType)) thread_lock_type = it->second; ++ switch (thread_lock_type) { ++ // In the original implementation, a write lock implies a read lock ++ // Retain this logic so that other casacore table logic works ++ case Write: return type == Write || type == Read; ++ case Read: return type == Read; ++ default: break; + } +- // Do finite number of attempts. Wait 1 second between each attempt. +- for (uInt i=0; imutex is recursive, so callers already holding it (acquire, ++ // release) re-lock harmlessly. ++ RecursiveMutexLock lock(itsState->mutex); ++ return std::accumulate( ++ std::begin(itsState->lockType), std::end(itsState->lockType), std::size_t{0}, ++ [&](auto i, auto tl) { return i + int(tl.second == type); } ++ ); ++} ++ ++Bool FileLocker::acquire (LockType requested_lock_type, uInt nattempts) ++{ ++ RecursiveMutexLock state_lock(itsState->mutex); ++ // Notify waiting threads of state change on exit ++ auto defer_notify = finally([&](...) { itsState->condVar.notify_all(); }); ++ auto AllLocks = [this]() -> std::size_t { return numLocks(Read) + numLocks(Write); }; ++ auto tid = std::this_thread::get_id(); ++ auto thread_lock_type = None; ++ if (auto it = itsState->lockType.find(tid); it != std::end(itsState->lockType)) thread_lock_type = it->second; ++ ++ switch(requested_lock_type) { ++ case None: ++ // None was patched in, casacore shouldn't be passing it in ++ // and neither should anyone else for that matter. ++ throw std::runtime_error("None passed to FileLocker::acquire"); ++ case Write: { ++ // This thread already has a write lock, return early ++ if (thread_lock_type == Write) return True; ++ // Signal that we want a write lock ++ itsState->wantsWriteLock += 1; ++ // To acquire a Write Lock, we wait until: ++ // 1. no other threads are acquiring a lock, and ++ // 2. no other thread wants a write lock, and ++ // 3. this is the only read-locked thread (upgrade) or ++ // there are no other read or write locked threads ++ itsState->condVar.wait(state_lock, [&]() { ++ return !itsState->modifyingFcntlLock && AllLocks() == (thread_lock_type == Read ? 1 : 0); ++ }); ++ assert(thread_lock_type == None || thread_lock_type == Read); ++ // Gate other threads while we modify the OS lock. Do not record the ++ // Write lock until fcntl actually succeeds (2f). ++ itsState->modifyingFcntlLock = true; ++ break; ++ } ++ case Read: { ++ // This thread already has a read lock, return early ++ if (thread_lock_type == Read) return True; ++ // To acquire a read lock, wait until: ++ // 1. no other threads are acquiring, and ++ // 2. no other thread wants a write lock, and ++ // 3. this is the only write-locked thread (downgrade) or ++ // there are no write-locked threads ++ itsState->condVar.wait(state_lock, [&]() { ++ return !itsState->modifyingFcntlLock && itsState->wantsWriteLock == 0 && ++ numLocks(Write) == (thread_lock_type == Write ? 1 : 0); ++ }); ++ assert(thread_lock_type == None || thread_lock_type == Write); ++ // Another thread has acquired the posix read lock, we simply ++ // register this thread as a reader (shared read lock) and return ++ if (thread_lock_type == None && numLocks(Read) > 0) { ++ itsState->lockType.insert_or_assign(tid, Read); ++ return True; ++ } ++ // Gate other threads while we modify the OS lock. Do not record the ++ // Read lock until fcntl actually succeeds (2f). ++ itsState->modifyingFcntlLock = true; ++ break; ++ } + } +- itsWriteLocked = False; +- // Note that the system keeps a lock per file and not per fd. +- // So if the same file is opened in the same process and unlocked +- // at the same place, the read lock for this fd is also released. +- // If we think we hold a read lock, determine if we still hold it. +- // We certainly do not if we asked for a read lock. +- // If asked for a write lock, we might still hold it. +- // One attempt is enough to see if we indeed can get a read lock. +- if (itsReadLocked) { +- itsReadLocked = False; +- if (type == Write) { +- ls.l_type = F_RDLCK; +- if (fcntl (itsFD, F_SETLK, &ls) != -1) { +- itsReadLocked = True; +- } +- } ++ ++ auto result = fcntlAquire(state_lock, requested_lock_type, nattempts); ++ ++ // Lock acquisition has finished ++ itsState->modifyingFcntlLock = false; ++ ++ switch(requested_lock_type) { ++ case None: ++ throw std::runtime_error("None passed to FileLocker::acquire"); ++ case Write: { ++ // Record the lock only on success (2f); on failure the previous ++ // lock type is left untouched. ++ if (result) itsState->lockType.insert_or_assign(tid, Write); ++ // A write lock is no longer desired ++ itsState->wantsWriteLock -= 1; ++ break; ++ } ++ case Read: { ++ if (result) itsState->lockType.insert_or_assign(tid, Read); ++ break; ++ } + } +-/// cout << "failed " << itsReadLocked << ' ' <mutex); ++ auto tid = std::this_thread::get_id(); ++ auto lock_type = None; ++ if (auto it = itsState->lockType.find(tid); it != std::end(itsState->lockType)) lock_type = it->second; ++ if (lock_type == None) return True; ++ ++ // Notify waiting threads of state change on exit ++ auto defer_notify = finally([&](...) { itsState->condVar.notify_all(); }); ++ ++ assert(numLocks(lock_type) >= 1); ++ ++ // We're not the last lock of this type ++ // Update the lock type and return success ++ if (numLocks(lock_type) > 1) { ++ itsState->lockType.insert_or_assign(tid, None); ++ return True; ++ } ++ ++ // Wait until no one else is modifying the lock ++ itsState->condVar.wait(state_lock, [&]() { return itsState->modifyingFcntlLock == false; }); ++ ++ itsState->modifyingFcntlLock = true; ++ auto result = fcntlRelease(); ++ // On successful lock release, update the lock type ++ if (result) itsState->lockType.insert_or_assign(tid, None); ++ itsState->modifyingFcntlLock = false; ++ return result; + } + ++ + // Release a lock. +-Bool FileLocker::release() ++Bool FileLocker::fcntlRelease() + { + /// cout << "released " << itsReadLocked << ' ' <mutex is held. ++ itsState->error = 0; + #if defined(AIPS_NOFILELOCK) + return True; + #else +@@ -192,7 +397,7 @@ Bool FileLocker::release() + return True; + } + #endif +- itsError = errno; ++ itsState->error = errno; + return False; + #endif + } +@@ -212,13 +417,19 @@ Bool FileLocker::canLock (uInt& pid, LockType type) + #if defined(AIPS_NOFILELOCK) + return True; + #else ++ RecursiveMutexLock lock(itsState->mutex); + pid = 0; +- itsError = 0; ++ itsState->error = 0; + struct flock ls; +- if (type == Write) { +- ls.l_type = F_WRLCK; +- }else{ +- ls.l_type = F_RDLCK; ++ switch (type) { ++ case Write: ++ ls.l_type = F_WRLCK; ++ break; ++ case Read: ++ ls.l_type = F_RDLCK; ++ break; ++ case None: ++ return False; + } + ls.l_whence = SEEK_SET; + ls.l_start = itsStart; +@@ -227,18 +438,39 @@ Bool FileLocker::canLock (uInt& pid, LockType type) + pid = ls.l_pid; + return (ls.l_type == F_UNLCK); + } +- itsError = errno; ++ itsState->error = errno; + return False; + #endif + } + ++int FileLocker::lastError() const ++{ ++ RecursiveMutexLock lock(itsState->mutex); ++ return itsState->error; ++} ++ + String FileLocker::lastMessage() const + { +- if (itsError == 0) { +- return ""; ++ int err; ++ { ++ RecursiveMutexLock lock(itsState->mutex); ++ err = itsState->error; + } +- return strerror(itsError); ++ if (err == 0) return String(); ++ // strerror_r has two incompatible signatures. The GNU variant returns a ++ // char* that may point to a static string rather than the supplied buffer; ++ // the XSI variant fills the buffer and returns an int. Never write past the ++ // end of a std::string's own storage (the previous implementation wrote ++ // into data() of a zero-length string, which was undefined behaviour). ++ char buf[1024]; ++ buf[0] = '\0'; ++#if (defined(__GLIBC__) || defined(__GNU_LIBRARY__)) && defined(_GNU_SOURCE) ++ const char* msg = ::strerror_r(err, buf, sizeof(buf)); ++ return String(msg ? msg : buf); ++#else ++ if (::strerror_r(err, buf, sizeof(buf)) != 0) return String("Unknown error"); ++ return String(buf); ++#endif + } + + } //# NAMESPACE CASACORE - END +- +diff --git a/casa/IO/FileLocker.h b/casa/IO/FileLocker.h +index 250f5bf..c522123 100644 +--- a/casa/IO/FileLocker.h ++++ b/casa/IO/FileLocker.h +@@ -28,12 +28,18 @@ + + + //# Includes ++#include ++#include ++ + #include + + namespace casacore { //# NAMESPACE CASACORE - BEGIN + + //# Forward Declarations + class String; ++// Coordination state shared by all FileLockers that lock the same ++// (lock-file path, start, length). Defined in FileLocker.cc. ++struct LockState; + + + // +@@ -94,7 +100,9 @@ public: + // Acquire a read lock. + Read, + // Acquire a write lock. +- Write ++ Write, ++ // ++ None, + }; + + // Default constructor creates an invalid fd. +@@ -105,6 +113,14 @@ public: + // The segment is given by start and length. Length=0 means till the + // end of the file. + explicit FileLocker (int fd, uInt start=0, uInt length=0); ++ // As above, but coordinate read/write locks in-process with every other ++ // FileLocker locking the same (name, start, length). This is what lets a ++ // single process safely hold MRSW locks on the same table from multiple ++ // threads/instances, each owning its own FileLocker. name is the ++ // absolute lock-file path and keys the shared coordination state. ++ FileLocker (int fd, const String& name, uInt start=0, uInt length=0); ++ FileLocker(FileLocker && rhs); ++ FileLocker& operator=(FileLocker&& rhs); + + ~FileLocker(); + +@@ -129,7 +145,7 @@ public: + // Test if the process has a lock for read or write on the file. + Bool hasLock (LockType = Write) const; + +- // Get the fd in use. ++ // Get the original file descriptor. + int fd() const; + + // Get the last error. +@@ -138,29 +154,27 @@ public: + // Get the message belonging to the last error. + String lastMessage() const; + ++ std::size_t numLocks(LockType type) const; ++ ++protected: ++ using RecursiveMutexLock = std::unique_lock; ++ ++ Bool fcntlAquire(RecursiveMutexLock & state_lock, LockType type = Write, uInt nattempts = 0); ++ Bool fcntlRelease(); + private: + int itsFD; +- int itsError; + int itsStart; + int itsLength; + Bool itsMsgShown; /// temporary for SUSE 6.1 +- Bool itsReadLocked; +- Bool itsWriteLocked; ++ // Coordination state, potentially shared with other FileLockers locking ++ // the same (name, start, length). Never null. ++ std::shared_ptr itsState; + }; + +- +-inline Bool FileLocker::hasLock (LockType type) const +-{ +- return (type == Write ? itsWriteLocked : itsReadLocked); +-} + inline int FileLocker::fd() const + { + return itsFD; + } +-inline int FileLocker::lastError() const +-{ +- return itsError; +-} + + + +diff --git a/casa/IO/LockFile.cc b/casa/IO/LockFile.cc +index 0f853c5..b68ba84 100644 +--- a/casa/IO/LockFile.cc ++++ b/casa/IO/LockFile.cc +@@ -23,6 +23,15 @@ + //# 520 Edgemont Road + //# Charlottesville, VA 22903-2475 USA + ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++#include ++ + + #include + #include +@@ -69,6 +78,51 @@ + + namespace casacore { //# NAMESPACE CASACORE - BEGIN + ++ ++namespace { ++ ++struct FileDescriptorLease { ++ int fd; ++ int counts; ++ std::unique_ptr io_mutex; ++}; ++ ++using FDCache = std::unordered_map; ++ ++std::recursive_mutex & fd_cache_mutex() { ++ static std::recursive_mutex m; ++ return m; ++} ++ ++FDCache & fd_cache() { ++ static FDCache cache; ++ return cache; ++} ++ ++std::mutex & GetIOMutex(const String & name) { ++ std::lock_guard lock(fd_cache_mutex()); ++ auto it = fd_cache().find(name); ++ if (it == std::end(fd_cache())) ++ throw std::runtime_error("io_mutex not available"); ++ return *it->second.io_mutex; ++} ++ ++Int HashThreadId() { ++ auto tid_hash = std::hash()(std::this_thread::get_id()); ++ static_assert(sizeof(tid_hash) % SIZEINT == 0, "size_t isn't a multiple of a word"); ++ static_assert(sizeof(Int) % SIZEINT == 0, "Int isn't a multiple of a word"); ++ constexpr int hash_words = sizeof(tid_hash) / SIZEINT; ++ constexpr int Int_words = sizeof(Int) / SIZEINT; ++ static_assert((hash_words % Int_words) == 0, "sizeof(size_t) doesn't divide sizeof(Int) perfectly"); ++ Int hash = 0; ++ const Int * p = reinterpret_cast(&tid_hash); ++ for (int i = 0; i < hash_words; ++i) hash ^= p[i]; ++ return hash; ++} ++ ++} // namespace { ++ ++ + LockFile::LockFile (const String& fileName, double inspectInterval, + Bool create, Bool setRequestFlag, Bool mustExist, + uInt seqnr, Bool permLocking, Bool noLocking) +@@ -84,50 +138,72 @@ LockFile::LockFile (const String& fileName, double inspectInterval, + AlwaysAssert (SIZEINT == CanonicalConversion::canonicalSize (static_cast(0)), + AipsError); + itsName = Path(fileName).absoluteName(); +- //# If needed, create the file if it does not exist yet. +- //# If the flag is set, it is allowed that the file does not +- //# exist and cannot be created. In that case it is assumed that +- //# later on each locking request is successful (without doing actual +- //# locking). +- if (!noLocking && !create) { +- File f (itsName); +- if (! f.exists()) { +- if (!f.canCreate() && !mustExist) { +- return; // Acceptable that lock file does not exist ++ ++ std::unique_lock cache_lock(fd_cache_mutex()); ++ auto it = fd_cache().find(itsName); ++ if (it == std::end(fd_cache())) { ++ // Create the cache entry with fd set to -1 ++ // This will be updated later ++ auto fd_lease = FileDescriptorLease{-1, 1, std::make_unique()}; ++ it = fd_cache().emplace(itsName, std::move(fd_lease)).first; ++ ++ //# If needed, create the file if it does not exist yet. ++ //# If the flag is set, it is allowed that the file does not ++ //# exist and cannot be created. In that case it is assumed that ++ //# later on each locking request is successful (without doing actual ++ //# locking). ++ if (!noLocking && !create) { ++ File f (itsName); ++ if (! f.exists()) { ++ if (!f.canCreate() && !mustExist) { ++ // Acceptable that the lock file does not exist. Remove the ++ // placeholder cache entry we just emplaced so that a later ++ // open (e.g. once the file can be created) re-attempts ++ // creation instead of finding a stuck fd == -1 entry. ++ fd_cache().erase(it); ++ return; ++ } ++ create = True; ++ } + } +- create = True; +- } +- } +- //# Open the lock file as read/write if it exists. +- //# If it did not succeed, open as readonly. +- //# For noLocking, it does not need to exist. +- int fd = -1; +- if (!create) { +- fd = FiledesIO::open (itsName.chars(), True, False); +- if (fd == -1) { +- fd = FiledesIO::open (itsName.chars(), False, !noLocking); +- itsWritable = False; +- itsAddToList = False; +- } +- } else if (!noLocking) { +- //# Create a new file with world write access. +- //# Initialize the values in it. +- fd = FiledesIO::create (itsName.chars(), 0666); +- putReqId (fd); ++ //# Open the lock file as read/write if it exists. ++ //# If it did not succeed, open as readonly. ++ //# For noLocking, it does not need to exist. ++ if (!create) { ++ it->second.fd = FiledesIO::open (itsName.chars(), True, False); ++ if (it->second.fd == -1) { ++ it->second.fd = FiledesIO::open (itsName.chars(), False, !noLocking); ++ itsWritable = False; ++ itsAddToList = False; ++ } ++ ++ } else if (!noLocking) { ++ //# Create a new file with world write access. ++ //# Initialize the values in it. ++ it->second.fd = FiledesIO::create (itsName.chars(), 0666); ++ putReqId (it->second.fd); ++ } ++ ++ } else { ++ ++it->second.counts; + } +- if (fd >= 0) { ++ ++ if (it->second.fd >= 0) { + //# Create FileLocker objects for this lock file. + //# The first one is for read/write locks. + //# The second one is to set the file to "in use" and to tell if + //# permanent locking is used. +- itsLocker = FileLocker (fd, 4*seqnr, 1); ++ //# Pass itsName so the FileLockers share coordination state with every ++ //# other FileLocker (in any thread/instance) locking the same byte range ++ //# of this lock file. ++ itsLocker = FileLocker (it->second.fd, itsName, 4*seqnr, 1); + if (permLocking) { +- itsUseLocker = FileLocker (fd, 4*seqnr+1, 2); ++ itsUseLocker = FileLocker (it->second.fd, itsName, 4*seqnr+1, 2); + } else { +- itsUseLocker = FileLocker (fd, 4*seqnr+1, 1); ++ itsUseLocker = FileLocker (it->second.fd, itsName, 4*seqnr+1, 1); + } + if (!noLocking) { +- itsFileIO.reset (new FiledesIO (fd, itsName)); ++ itsFileIO.reset (new FiledesIO (it->second.fd, itsName)); + // Set the file to in use by acquiring a read lock. + itsUseLocker.acquire (FileLocker::Read, 1); + } +@@ -136,10 +212,16 @@ LockFile::LockFile (const String& fileName, double inspectInterval, + + LockFile::~LockFile() + { +- int fd = itsLocker.fd(); +- if (fd >= 0) { +- FiledesIO::close (fd); +- } ++ // Close file descriptor once all leases have expired ++ std::lock_guard lock(fd_cache_mutex()); ++ auto it = fd_cache().find(itsName); ++ // No record, just return ++ if (it == std::end(fd_cache())) return; ++ // This isn't the last lease on the fd ++ if (--it->second.counts > 0) return; ++ // Close any valid fd and remove from cache ++ if (it->second.fd >= 0) FiledesIO::close (it->second.fd); ++ fd_cache().erase(it); + } + + Bool LockFile::isMultiUsed() +@@ -239,6 +321,7 @@ void LockFile::getInfo (MemoryIO& info) + { + // Do nothing if no locking. + if (itsLocker.fd() < 0) { ++ info.seek(Int64(0)); + return; + } + // The lock file contains: +@@ -246,6 +329,7 @@ void LockFile::getInfo (MemoryIO& info) + // - thereafter the length of the info (as a uInt) + // - thereafter the entire info + uChar buffer[2048]; ++ std::lock_guard io_lock(GetIOMutex(itsName)); + // Read the first part of the file. + traceLSEEK (itsLocker.fd(), 0, SEEK_SET); + uInt leng = ::read (itsLocker.fd(), buffer, sizeof(buffer)); +@@ -284,6 +368,7 @@ void LockFile::putInfo (const MemoryIO& info) const + if (itsLocker.fd() < 0 || !itsWritable || infoLeng == 0) { + return; + } ++ std::lock_guard io_lock(GetIOMutex(itsName)); + // Write the info into the lock file preceeded by its length. + uChar buffer[1024]; + uInt leng = CanonicalConversion::fromLocal (buffer, infoLeng); +@@ -307,6 +392,7 @@ void LockFile::putInfo (const MemoryIO& info) const + Int LockFile::getNrReqId() const + { + uChar buffer[8]; ++ std::lock_guard io_lock(GetIOMutex(itsName)); + uInt leng = tracePREAD (itsLocker.fd(), buffer, SIZEINT, 0); + return getInt (buffer, leng, 0); + } +@@ -339,7 +425,7 @@ void LockFile::addReqId() + inx = NRREQID-1; + } + itsReqId[0] = inx+1; +- itsReqId[2*inx+1] = itsPid; ++ itsReqId[2*inx+1] = HashThreadId(); + itsReqId[2*inx+2] = itsHostId; + putReqId (itsLocker.fd()); + } +@@ -352,7 +438,7 @@ void LockFile::removeReqId() + //# can happen when a process with an outstanding request died. + Int nr = itsReqId[0]; + for (i=0; i io_lock(GetIOMutex(itsName)); + AlwaysAssert(tracePWRITE(fd, (Char *)buffer, leng, 0) == Int(leng), + AipsError); + fsync (fd); +@@ -382,6 +469,7 @@ void LockFile::getReqId() + { + int fd = itsLocker.fd(); + uChar buffer[SIZEREQID]; ++ std::lock_guard io_lock(GetIOMutex(itsName)); + if (tracePREAD(fd, buffer, SIZEREQID, 0) > 0) { + CanonicalConversion::fromLocal (buffer, + itsReqId.storage(), +@@ -436,4 +524,8 @@ uInt LockFile::showLock (uInt& pid, Bool& permLocked, const String& fileName) + return result; + } + ++std::size_t LockFile::numLocks(FileLocker::LockType lock_type) const { ++ return itsLocker.numLocks(lock_type); ++} ++ + } //# NAMESPACE CASACORE - END +diff --git a/casa/IO/LockFile.h b/casa/IO/LockFile.h +index 57c57db..f21246f 100644 +--- a/casa/IO/LockFile.h ++++ b/casa/IO/LockFile.h +@@ -280,6 +280,10 @@ public: + // be opened. + static uInt showLock (uInt& pid, Bool& permLocked, const String& fileName); + ++ // Returns the number of locks hold of the given type ++ // Not thread-safe ++ std::size_t numLocks(FileLocker::LockType lock_type) const; ++ + private: + // The copy constructor cannot be used (its semantics are too difficult). + LockFile (const LockFile&); diff --git a/casa/Json/JsonOut.h b/casa/Json/JsonOut.h index b5229c5..7250d93 100644 --- a/casa/Json/JsonOut.h @@ -503,6 +1457,18 @@ diff --git a/tables/Tables/TableProxy.cc b/tables/Tables/TableProxy.cc index f65cf26..3f806cd 100644 --- a/tables/Tables/TableProxy.cc +++ b/tables/Tables/TableProxy.cc +@@ -224,9 +224,9 @@ String TableProxy::endianFormat() const + return "little"; + } + +-void TableProxy::lock (Bool mode, Int nattempts) ++Bool TableProxy::lock (Bool mode, Int nattempts) + { +- table_p.lock (mode, nattempts); ++ return table_p.lock (mode, nattempts); + } + + void TableProxy::unlock() @@ -271,6 +271,9 @@ Record TableProxy::lockOptions() option = "autonoread"; } @@ -527,3 +1493,16 @@ index f65cf26..3f806cd 100644 } if (options.nfields() == 1) { return TableLock(opt); +diff --git a/tables/Tables/TableProxy.h b/tables/Tables/TableProxy.h +index 5b4c539..7d885c5 100644 +--- a/tables/Tables/TableProxy.h ++++ b/tables/Tables/TableProxy.h +@@ -198,7 +198,7 @@ public: + String endianFormat() const; + + // Acquire a (read or write) lock on the table. +- void lock (Bool mode, Int nattempts); ++ Bool lock (Bool mode, Int nattempts); + + // Release a lock on the table. + void unlock();