--- /dev/null
+/*************************************************************************
+ *
+ * Copyright 2016 Realm Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ **************************************************************************/
+
+#ifndef REALM_GROUP_SHARED_HPP
+#define REALM_GROUP_SHARED_HPP
+
+#include <functional>
+#include <limits>
+#include <realm/util/features.h>
+#include <realm/util/thread.hpp>
+#include <realm/util/interprocess_condvar.hpp>
+#include <realm/util/interprocess_mutex.hpp>
+#include <realm/group.hpp>
+#include <realm/group_shared_options.hpp>
+#include <realm/handover_defs.hpp>
+#include <realm/impl/transact_log.hpp>
+#include <realm/metrics/metrics.hpp>
+#include <realm/replication.hpp>
+#include <realm/version_id.hpp>
+
+namespace realm {
+
+namespace _impl {
+class SharedGroupFriend;
+class WriteLogCollector;
+}
+
+/// Thrown by SharedGroup::open() if the lock file is already open in another
+/// process which can't share mutexes with this process
+struct IncompatibleLockFile : std::runtime_error {
+ IncompatibleLockFile(const std::string& msg)
+ : std::runtime_error("Incompatible lock file. " + msg)
+ {
+ }
+};
+
+/// Thrown by SharedGroup::open() if the type of history
+/// (Replication::HistoryType) in the opened Realm file is incompatible with the
+/// mode in which the Realm file is opened. For example, if there is a mismatch
+/// between the history type in the file, and the history type associated with
+/// the replication plugin passed to SharedGroup::open().
+///
+/// This exception will also be thrown if the history schema version is lower
+/// than required, and no migration is possible
+/// (Replication::is_upgradable_history_schema()).
+struct IncompatibleHistories : util::File::AccessError {
+ IncompatibleHistories(const std::string& msg, const std::string& path)
+ : util::File::AccessError("Incompatible histories. " + msg, path)
+ {
+ }
+};
+
+/// A SharedGroup facilitates transactions.
+///
+/// When multiple threads or processes need to access a database
+/// concurrently, they must do so using transactions. By design,
+/// Realm does not allow for multiple threads (or processes) to
+/// share a single instance of SharedGroup. Instead, each concurrently
+/// executing thread or process must use a separate instance of
+/// SharedGroup.
+///
+/// Each instance of SharedGroup manages a single transaction at a
+/// time. That transaction can be either a read transaction, or a
+/// write transaction.
+///
+/// Utility classes ReadTransaction and WriteTransaction are provided
+/// to make it safe and easy to work with transactions in a scoped
+/// manner (by means of the RAII idiom). However, transactions can
+/// also be explicitly started (begin_read(), begin_write()) and
+/// stopped (end_read(), commit(), rollback()).
+///
+/// If a transaction is active when the SharedGroup is destroyed, that
+/// transaction is implicitly terminated, either by a call to
+/// end_read() or rollback().
+///
+/// Two processes that want to share a database file must reside on
+/// the same host.
+///
+///
+/// Desired exception behavior (not yet fully implemented)
+/// ------------------------------------------------------
+///
+/// - If any data access API function throws an unexpected exception during a
+/// read transaction, the shared group accessor is left in state "error
+/// during read".
+///
+/// - If any data access API function throws an unexpected exception during a
+/// write transaction, the shared group accessor is left in state "error
+/// during write".
+///
+/// - If SharedGroup::begin_write() or SharedGroup::begin_read() throws an
+/// unexpected exception, the shared group accessor is left in state "no
+/// transaction in progress".
+///
+/// - SharedGroup::end_read() and SharedGroup::rollback() do not throw.
+///
+/// - If SharedGroup::commit() throws an unexpected exception, the shared group
+/// accessor is left in state "error during write" and the transaction was
+/// not committed.
+///
+/// - If SharedGroup::advance_read() or SharedGroup::promote_to_write() throws
+/// an unexpected exception, the shared group accessor is left in state
+/// "error during read".
+///
+/// - If SharedGroup::commit_and_continue_as_read() or
+/// SharedGroup::rollback_and_continue_as_read() throws an unexpected
+/// exception, the shared group accessor is left in state "error during
+/// write".
+///
+/// It has not yet been decided exactly what an "unexpected exception" is, but
+/// `std::bad_alloc` is surely one example. On the other hand, an expected
+/// exception is one that is mentioned in the function specific documentation,
+/// and is used to abort an operation due to a special, but expected condition.
+///
+/// States
+/// ------
+///
+/// - A newly created shared group accessor is in state "no transaction in
+/// progress".
+///
+/// - In state "error during read", almost all Realm API functions are
+/// illegal on the connected group of accessors. The only valid operations
+/// are destruction of the shared group, and SharedGroup::end_read(). If
+/// SharedGroup::end_read() is called, the new state becomes "no transaction
+/// in progress".
+///
+/// - In state "error during write", almost all Realm API functions are
+/// illegal on the connected group of accessors. The only valid operations
+/// are destruction of the shared group, and SharedGroup::rollback(). If
+/// SharedGroup::end_write() is called, the new state becomes "no transaction
+/// in progress"
+class SharedGroup {
+public:
+ /// \brief Same as calling the corresponding version of open() on a instance
+ /// constructed in the unattached state. Exception safety note: if the
+ /// `upgrade_callback` throws, then the file will be closed properly and the
+ /// upgrade will be aborted.
+ explicit SharedGroup(const std::string& file, bool no_create = false,
+ const SharedGroupOptions options = SharedGroupOptions());
+
+ /// \brief Same as calling the corresponding version of open() on a instance
+ /// constructed in the unattached state. Exception safety note: if the
+ /// `upgrade_callback` throws, then the file will be closed properly and
+ /// the upgrade will be aborted.
+ explicit SharedGroup(Replication& repl, const SharedGroupOptions options = SharedGroupOptions());
+
+ struct unattached_tag {
+ };
+
+ /// Create a SharedGroup instance in its unattached state. It may
+ /// then be attached to a database file later by calling
+ /// open(). You may test whether this instance is currently in its
+ /// attached state by calling is_attached(). Calling any other
+ /// function (except the destructor) while in the unattached state
+ /// has undefined behavior.
+ SharedGroup(unattached_tag) noexcept;
+
+ ~SharedGroup() noexcept;
+
+ // Disable copying to prevent accessor errors. If you really want another
+ // instance, open another SharedGroup object on the same file.
+ SharedGroup(const SharedGroup&) = delete;
+ SharedGroup& operator=(const SharedGroup&) = delete;
+
+ /// Attach this SharedGroup instance to the specified database file.
+ ///
+ /// While at least one instance of SharedGroup exists for a specific
+ /// database file, a "lock" file will be present too. The lock file will be
+ /// placed in the same directory as the database file, and its name will be
+ /// derived by appending ".lock" to the name of the database file.
+ ///
+ /// When multiple SharedGroup instances refer to the same file, they must
+ /// specify the same durability level, otherwise an exception will be
+ /// thrown.
+ ///
+ /// \param file Filesystem path to a Realm database file.
+ ///
+ /// \param no_create If the database file does not already exist, it will be
+ /// created (unless this is set to true.) When multiple threads are involved,
+ /// it is safe to let the first thread, that gets to it, create the file.
+ ///
+ /// \param options See SharedGroupOptions for details of each option.
+ /// Sensible defaults are provided if this parameter is left out.
+ ///
+ /// Calling open() on a SharedGroup instance that is already in the attached
+ /// state has undefined behavior.
+ ///
+ /// \throw util::File::AccessError If the file could not be opened. If the
+ /// reason corresponds to one of the exception types that are derived from
+ /// util::File::AccessError, the derived exception type is thrown. Note that
+ /// InvalidDatabase is among these derived exception types.
+ ///
+ /// \throw FileFormatUpgradeRequired only if \a SharedGroupOptions::allow_upgrade
+ /// is `false` and an upgrade is required.
+ void open(const std::string& file, bool no_create = false,
+ const SharedGroupOptions options = SharedGroupOptions());
+
+ /// Open this group in replication mode. The specified Replication instance
+ /// must remain in existence for as long as the SharedGroup.
+ void open(Replication&, const SharedGroupOptions options = SharedGroupOptions());
+
+ /// Close any open database, returning to the unattached state.
+ void close() noexcept;
+
+ /// A SharedGroup may be created in the unattached state, and then
+ /// later attached to a file with a call to open(). Calling any
+ /// function other than open(), is_attached(), and ~SharedGroup()
+ /// on an unattached instance results in undefined behavior.
+ bool is_attached() const noexcept;
+
+ /// Reserve disk space now to avoid allocation errors at a later
+ /// point in time, and to minimize on-disk fragmentation. In some
+ /// cases, less fragmentation translates into improved
+ /// performance.
+ ///
+ /// When supported by the system, a call to this function will
+ /// make the database file at least as big as the specified size,
+ /// and cause space on the target device to be allocated (note
+ /// that on many systems on-disk allocation is done lazily by
+ /// default). If the file is already bigger than the specified
+ /// size, the size will be unchanged, and on-disk allocation will
+ /// occur only for the initial section that corresponds to the
+ /// specified size. On systems that do not support preallocation,
+ /// this function has no effect. To know whether preallocation is
+ /// supported by Realm on your platform, call
+ /// util::File::is_prealloc_supported().
+ ///
+ /// It is an error to call this function on an unattached shared
+ /// group. Doing so will result in undefined behavior.
+ void reserve(size_t size_in_bytes);
+
+ /// Querying for changes:
+ ///
+ /// NOTE:
+ /// "changed" means that one or more commits has been made to the database
+ /// since the SharedGroup (on which wait_for_change() is called) last
+ /// started, committed, promoted or advanced a transaction. If the
+ /// SharedGroup has not yet begun a transaction, "changed" is undefined.
+ ///
+ /// No distinction is made between changes done by another process
+ /// and changes done by another thread in the same process as the caller.
+ ///
+ /// Has db been changed ?
+ bool has_changed();
+
+ /// The calling thread goes to sleep until the database is changed, or
+ /// until wait_for_change_release() is called. After a call to
+ /// wait_for_change_release() further calls to wait_for_change() will return
+ /// immediately. To restore the ability to wait for a change, a call to
+ /// enable_wait_for_change() is required. Return true if the database has
+ /// changed, false if it might have.
+ bool wait_for_change();
+
+ /// release any thread waiting in wait_for_change() on *this* SharedGroup.
+ void wait_for_change_release();
+
+ /// re-enable waiting for change
+ void enable_wait_for_change();
+ // Transactions:
+
+ using version_type = _impl::History::version_type;
+ using VersionID = realm::VersionID;
+
+ /// Thrown by begin_read() if the specified version does not correspond to a
+ /// bound (or tethered) snapshot.
+ struct BadVersion;
+
+ /// \defgroup group_shared_transactions
+ //@{
+
+ /// begin_read() initiates a new read transaction. A read transaction is
+ /// bound to, and provides access to a particular snapshot of the underlying
+ /// Realm (in general the latest snapshot, but see \a version). It cannot be
+ /// used to modify the Realm, and in that sense, a read transaction is not a
+ /// real transaction.
+ ///
+ /// begin_write() initiates a new write transaction. A write transaction
+ /// allows the application to both read and modify the underlying Realm
+ /// file. At most one write transaction can be in progress at any given time
+ /// for a particular underlying Realm file. If another write transaction is
+ /// already in progress, begin_write() will block the caller until the other
+ /// write transaction terminates. No guarantees are made about the order in
+ /// which multiple concurrent requests will be served.
+ ///
+ /// It is an error to call begin_read() or begin_write() on a SharedGroup
+ /// object with an active read or write transaction.
+ ///
+ /// If begin_read() or begin_write() throws, no transaction is initiated,
+ /// and the application may try to initiate a new read or write transaction
+ /// later.
+ ///
+ /// end_read() terminates the active read transaction. If no read
+ /// transaction is active, end_read() does nothing. It is an error to call
+ /// this function on a SharedGroup object with an active write
+ /// transaction. end_read() does not throw.
+ ///
+ /// commit() commits all changes performed in the context of the active
+ /// write transaction, and thereby terminates that transaction. This
+ /// produces a new snapshot in the underlying Realm. commit() returns the
+ /// version associated with the new snapshot. It is an error to call
+ /// commit() when there is no active write transaction. If commit() throws,
+ /// no changes will have been committed, and the transaction will still be
+ /// active, but in a bad state. In that case, the application must either
+ /// call rollback() to terminate the bad transaction (in which case a new
+ /// transaction can be initiated), call close() which also terminates the
+ /// bad transaction, or destroy the SharedGroup object entirely. When the
+ /// transaction is in a bad state, the application is not allowed to call
+ /// any method on the Group accessor or on any of its subordinate accessors
+ /// (Table, Row, Descriptor). Note that the transaction is also left in a
+ /// bad state when a modifying operation on any subordinate accessor throws.
+ ///
+ /// rollback() terminates the active write transaction and discards any
+ /// changes performed in the context of it. If no write transaction is
+ /// active, rollback() does nothing. It is an error to call this function in
+ /// a SharedGroup object with an active read transaction. rollback() does
+ /// not throw.
+ ///
+ /// the Group accessor and all subordinate accessors (Table, Row,
+ /// Descriptor) that are obtained in the context of a particular read or
+ /// write transaction will become detached upon termination of that
+ /// transaction, which means that they can no longer be used to access the
+ /// underlying objects.
+ ///
+ /// Subordinate accessors that were detached at the end of the previous
+ /// read or write transaction will not be automatically reattached when a
+ /// new transaction is initiated. The application must reobtain new
+ /// accessors during a new transaction to regain access to the underlying
+ /// objects.
+ ///
+ /// \param version If specified, this must be the version associated with a
+ /// *bound* snapshot. A snapshot is said to be bound (or tethered) if there
+ /// is at least one active read or write transaction bound to it. A read
+ /// transaction is bound to the snapshot that it provides access to. A write
+ /// transaction is bound to the latest snapshot available at the time of
+ /// initiation of the write transaction. If the specified version is not
+ /// associated with a bound snapshot, this function throws BadVersion.
+ ///
+ /// \throw BadVersion Thrown by begin_read() if the specified version does
+ /// not correspond to a bound (or tethered) snapshot.
+
+ const Group& begin_read(VersionID version = VersionID());
+ void end_read() noexcept;
+ Group& begin_write();
+ // Return true (and take the write lock) if there is no other write
+ // in progress. In case of contention return false immediately.
+ // If the write lock is obtained, also provide the Group associated
+ // with the SharedGroup for further operations.
+ bool try_begin_write(Group*& group);
+ version_type commit();
+ void rollback() noexcept;
+ // report statistics of last commit done on THIS shared group.
+ // The free space reported is what can be expected to be freed
+ // by compact(). This may not correspond to the space which is free
+ // at the point where get_stats() is called, since that will include
+ // memory required to hold older versions of data, which still
+ // needs to be available.
+ void get_stats(size_t& free_space, size_t& used_space);
+ //@}
+
+ enum TransactStage {
+ transact_Ready,
+ transact_Reading,
+ transact_Writing,
+ };
+
+ /// Get the current transaction type
+ TransactStage get_transact_stage() const noexcept;
+
+ /// Get a version id which may be used to request a different SharedGroup
+ /// to start transaction at a specific version.
+ VersionID get_version_of_current_transaction();
+
+ /// Report the number of distinct versions currently stored in the database.
+ /// Note: the database only cleans up versions as part of commit, so ending
+ /// a read transaction will not immediately release any versions.
+ uint_fast64_t get_number_of_versions();
+
+ /// Compact the database file.
+ /// - The method will throw if called inside a transaction.
+ /// - The method will throw if called in unattached state.
+ /// - The method will return false if other SharedGroups are accessing the
+ /// database in which case compaction is not done. This is not
+ /// necessarily an error.
+ /// It will return true following successful compaction.
+ /// While compaction is in progress, attempts by other
+ /// threads or processes to open the database will wait.
+ /// Be warned that resource requirements for compaction is proportional to
+ /// the amount of live data in the database.
+ /// Compaction works by writing the database contents to a temporary
+ /// database file and then replacing the database with the temporary one.
+ /// The name of the temporary file is formed by appending
+ /// ".tmp_compaction_space" to the name of the database
+ ///
+ /// FIXME: This function is not yet implemented in an exception-safe manner,
+ /// therefore, if it throws, the application should not attempt to
+ /// continue. If may not even be safe to destroy the SharedGroup object.
+ ///
+ /// WARNING / FIXME: compact() should NOT be exposed publicly on Windows
+ /// because it's not crash safe! It may corrupt your database if something fails
+ bool compact();
+
+#ifdef REALM_DEBUG
+ void test_ringbuf();
+#endif
+
+ /// To handover a table view, query, linkview or row accessor of type T, you
+ /// must wrap it into a Handover<T> for the transfer. Wrapping and
+ /// unwrapping of a handover object is done by the methods
+ /// 'export_for_handover()' and 'import_from_handover()' declared below.
+ /// 'export_for_handover()' returns a Handover object, and
+ /// 'import_for_handover()' consumes that object, producing a new accessor
+ /// which is ready for use in the context of the importing SharedGroup.
+ ///
+ /// The Handover always creates a new accessor object at the importing side.
+ /// For TableViews, there are 3 forms of handover.
+ ///
+ /// - with payload move: the payload is handed over and ends up as a payload
+ /// held by the accessor at the importing side. The accessor on the
+ /// exporting side will rerun its query and generate a new payload, if
+ /// TableView::sync_if_needed() is called. If the original payload was in
+ /// sync at the exporting side, it will also be in sync at the importing
+ /// side. This is indicated to handover_export() by the argument
+ /// MutableSourcePayload::Move
+ ///
+ /// - with payload copy: a copy of the payload is handed over, so both the
+ /// accessors on the exporting side *and* the accessors created at the
+ /// importing side has their own payload. This is indicated to
+ /// handover_export() by the argument ConstSourcePayload::Copy
+ ///
+ /// - without payload: the payload stays with the accessor on the exporting
+ /// side. On the importing side, the new accessor is created without
+ /// payload. A call to TableView::sync_if_needed() will trigger generation
+ /// of a new payload. This form of handover is indicated to
+ /// handover_export() by the argument ConstSourcePayload::Stay.
+ ///
+ /// For all other (non-TableView) accessors, handover is done with payload
+ /// copy, since the payload is trivial.
+ ///
+ /// Handover *without* payload is useful when you want to ship a tableview
+ /// with its query for execution in a background thread. Handover with
+ /// *payload move* is useful when you want to transfer the result back.
+ ///
+ /// Handover *without* payload or with payload copy is guaranteed *not* to
+ /// change the accessors on the exporting side.
+ ///
+ /// Handover is *not* thread safe and should be carried out
+ /// by the thread that "owns" the involved accessors.
+ ///
+ /// Handover is transitive:
+ /// If the object being handed over depends on other views
+ /// (table- or link- ), those objects will be handed over as well. The mode
+ /// of handover (payload copy, payload move, without payload) is applied
+ /// recursively. Note: If you are handing over a tableview dependent upon
+ /// another tableview and using MutableSourcePayload::Move,
+ /// you are on thin ice!
+ ///
+ /// On the importing side, the top-level accessor being created during
+ /// import takes ownership of all other accessors (if any) being created as
+ /// part of the import.
+
+ /// Type used to support handover of accessors between shared groups.
+ template <typename T>
+ struct Handover;
+
+ /// thread-safe/const export (mode is Stay or Copy)
+ /// during export, the following operations on the shared group is locked:
+ /// - advance_read(), promote_to_write(), commit_and_continue_as_read(),
+ /// rollback_and_continue_as_read(), close()
+ template <typename T>
+ std::unique_ptr<Handover<T>> export_for_handover(const T& accessor, ConstSourcePayload mode);
+
+ // specialization for handover of Rows
+ template <typename T>
+ std::unique_ptr<Handover<BasicRow<T>>> export_for_handover(const BasicRow<T>& accessor);
+
+ // destructive export (mode is Move)
+ template <typename T>
+ std::unique_ptr<Handover<T>> export_for_handover(T& accessor, MutableSourcePayload mode);
+
+ /// Import an accessor wrapped in a handover object. The import will fail
+ /// if the importing SharedGroup is viewing a version of the database that
+ /// is different from the exporting SharedGroup. The call to
+ /// import_from_handover is not thread-safe.
+ template <typename T>
+ std::unique_ptr<T> import_from_handover(std::unique_ptr<Handover<T>> handover);
+
+ // We need two cases for handling of LinkViews, because they are ref counted.
+ std::unique_ptr<Handover<LinkView>> export_linkview_for_handover(const LinkViewRef& accessor);
+ LinkViewRef import_linkview_from_handover(std::unique_ptr<Handover<LinkView>> handover);
+
+ // likewise for Tables.
+ std::unique_ptr<Handover<Table>> export_table_for_handover(const TableRef& accessor);
+ TableRef import_table_from_handover(std::unique_ptr<Handover<Table>> handover);
+
+ /// When doing handover to background tasks that may be run later, we
+ /// may want to momentarily pin the current version until the other thread
+ /// has retrieved it.
+ ///
+ /// Pinning can be done in both read- and write-transactions, but with different
+ /// semantics. When pinning during a read-transaction, the version pinned is the
+ /// one accessible during the read-transaction. When pinning during a write-transaction,
+ /// the version pinned will be the last version that was succesfully committed to the
+ /// realm file at the point in time, when the write-transaction was started.
+ ///
+ /// The release is not thread-safe, so it has to be done on the SharedGroup
+ /// associated with the thread calling unpin_version(), and the SharedGroup
+ /// must be attached to the realm file at the point of unpinning.
+
+ // Pin version for handover (not thread safe)
+ VersionID pin_version();
+
+ // Release pinned version (not thread safe)
+ void unpin_version(VersionID version);
+
+#if REALM_METRICS
+ std::shared_ptr<metrics::Metrics> get_metrics();
+#endif // REALM_METRICS
+
+ // Try to grab a exclusive lock of the given realm path's lock file. If the lock
+ // can be acquired, the callback will be executed with the lock and then return true.
+ // Otherwise false will be returned directly.
+ // The lock taken precludes races with other threads or processes accessing the
+ // files through a SharedGroup.
+ // It is safe to delete/replace realm files inside the callback.
+ // WARNING: It is not safe to delete the lock file in the callback.
+ using CallbackWithLock = std::function<void(const std::string& realm_path)>;
+ static bool call_with_lock(const std::string& realm_path, CallbackWithLock callback);
+
+ // Return a list of files/directories core may use of the given realm file path.
+ // The first element of the pair in the returned list is the path string, the
+ // second one is to indicate the path is a directory or not.
+ // The temporary files are not returned by this function.
+ // It is safe to delete those returned files/directories in the call_with_lock's callback.
+ static std::vector<std::pair<std::string, bool>> get_core_files(const std::string& realm_path);
+
+private:
+ struct SharedInfo;
+ struct ReadCount;
+ struct ReadLockInfo {
+ uint_fast64_t m_version = std::numeric_limits<version_type>::max();
+ uint_fast32_t m_reader_idx = 0;
+ ref_type m_top_ref = 0;
+ size_t m_file_size = 0;
+ };
+ class ReadLockUnlockGuard;
+
+ // Member variables
+ size_t m_free_space = 0;
+ size_t m_used_space = 0;
+ Group m_group;
+ ReadLockInfo m_read_lock;
+ uint_fast32_t m_local_max_entry;
+ util::File m_file;
+ util::File::Map<SharedInfo> m_file_map; // Never remapped
+ util::File::Map<SharedInfo> m_reader_map;
+ bool m_wait_for_change_enabled;
+ std::string m_lockfile_path;
+ std::string m_lockfile_prefix;
+ std::string m_db_path;
+ std::string m_coordination_dir;
+ const char* m_key;
+ TransactStage m_transact_stage;
+ util::InterprocessMutex m_writemutex;
+#ifdef REALM_ASYNC_DAEMON
+ util::InterprocessMutex m_balancemutex;
+#endif
+ util::InterprocessMutex m_controlmutex;
+#ifdef REALM_ASYNC_DAEMON
+ util::InterprocessCondVar m_room_to_write;
+ util::InterprocessCondVar m_work_to_do;
+ util::InterprocessCondVar m_daemon_becomes_ready;
+#endif
+ util::InterprocessCondVar m_new_commit_available;
+ util::InterprocessCondVar m_pick_next_writer;
+ std::function<void(int, int)> m_upgrade_callback;
+
+#if REALM_METRICS
+ std::shared_ptr<metrics::Metrics> m_metrics;
+#endif // REALM_METRICS
+
+ void do_open(const std::string& file, bool no_create, bool is_backend, const SharedGroupOptions options);
+
+ // Ring buffer management
+ bool ringbuf_is_empty() const noexcept;
+ size_t ringbuf_size() const noexcept;
+ size_t ringbuf_capacity() const noexcept;
+ bool ringbuf_is_first(size_t ndx) const noexcept;
+ void ringbuf_remove_first() noexcept;
+ size_t ringbuf_find(uint64_t version) const noexcept;
+ ReadCount& ringbuf_get(size_t ndx) noexcept;
+ ReadCount& ringbuf_get_first() noexcept;
+ ReadCount& ringbuf_get_last() noexcept;
+ void ringbuf_put(const ReadCount& v);
+ void ringbuf_expand();
+
+ /// Grab a read lock on the snapshot associated with the specified
+ /// version. If `version_id == VersionID()`, a read lock will be grabbed on
+ /// the latest available snapshot. Fails if the snapshot is no longer
+ /// available.
+ ///
+ /// As a side effect update memory mapping to ensure that the ringbuffer
+ /// entries referenced in the readlock info is accessible.
+ ///
+ /// FIXME: It needs to be made more clear exactly under which conditions
+ /// this function fails. Also, why is it useful to promise anything about
+ /// detection of bad versions? Can we really promise enough to make such a
+ /// promise useful to the caller?
+ void grab_read_lock(ReadLockInfo&, VersionID);
+
+ // Release a specific read lock. The read lock MUST have been obtained by a
+ // call to grab_read_lock().
+ void release_read_lock(ReadLockInfo&) noexcept;
+
+ void do_begin_read(VersionID, bool writable);
+ void do_end_read() noexcept;
+ /// return true if write transaction can commence, false otherwise.
+ bool do_try_begin_write();
+ void do_begin_write();
+ version_type do_commit();
+ void do_end_write() noexcept;
+ void set_transact_stage(TransactStage stage) noexcept;
+
+ /// Returns the version of the latest snapshot.
+ version_type get_version_of_latest_snapshot();
+
+ /// Returns the version of the snapshot bound in the current read or write
+ /// transaction. It is an error to call this function when no transaction is
+ /// in progress.
+ version_type get_version_of_bound_snapshot() const noexcept;
+
+ // make sure the given index is within the currently mapped area.
+ // if not, expand the mapped area. Returns true if the area is expanded.
+ bool grow_reader_mapping(uint_fast32_t index);
+
+ // Must be called only by someone that has a lock on the write
+ // mutex.
+ void low_level_commit(uint_fast64_t new_version);
+
+ void do_async_commits();
+
+ /// Upgrade file format and/or history schema
+ void upgrade_file_format(bool allow_file_format_upgrade, int target_file_format_version,
+ int current_hist_schema_version, int target_hist_schema_version);
+
+ //@{
+ /// See LangBindHelper.
+ template <class O>
+ void advance_read(O* observer, VersionID);
+ template <class O>
+ void promote_to_write(O* observer);
+ version_type commit_and_continue_as_read();
+ template <class O>
+ void rollback_and_continue_as_read(O* observer);
+ //@}
+
+ /// Returns true if, and only if _impl::History::update_early_from_top_ref()
+ /// was called during the execution of this function.
+ template <class O>
+ bool do_advance_read(O* observer, VersionID, _impl::History&);
+
+ /// If there is an associated \ref Replication object, then this function
+ /// returns `repl->get_history()` where `repl` is that Replication object,
+ /// otherwise this function returns null.
+ _impl::History* get_history();
+
+ int get_file_format_version() const noexcept;
+
+ /// finish up the process of starting a write transaction. Internal use only.
+ void finish_begin_write();
+
+ void close_internal(std::unique_lock<InterprocessMutex>) noexcept;
+ friend class _impl::SharedGroupFriend;
+};
+
+
+inline void SharedGroup::get_stats(size_t& free_space, size_t& used_space) {
+ free_space = m_free_space;
+ used_space = m_used_space;
+}
+
+
+class ReadTransaction {
+public:
+ ReadTransaction(SharedGroup& sg)
+ : m_shared_group(sg)
+ {
+ m_shared_group.begin_read(); // Throws
+ }
+
+ ~ReadTransaction() noexcept
+ {
+ m_shared_group.end_read();
+ }
+
+ bool has_table(StringData name) const noexcept
+ {
+ return get_group().has_table(name);
+ }
+
+ ConstTableRef get_table(size_t table_ndx) const
+ {
+ return get_group().get_table(table_ndx); // Throws
+ }
+
+ ConstTableRef get_table(StringData name) const
+ {
+ return get_group().get_table(name); // Throws
+ }
+
+ const Group& get_group() const noexcept;
+
+ /// Get the version of the snapshot to which this read transaction is bound.
+ SharedGroup::version_type get_version() const noexcept;
+
+private:
+ SharedGroup& m_shared_group;
+};
+
+
+class WriteTransaction {
+public:
+ WriteTransaction(SharedGroup& sg)
+ : m_shared_group(&sg)
+ {
+ m_shared_group->begin_write(); // Throws
+ }
+
+ ~WriteTransaction() noexcept
+ {
+ if (m_shared_group)
+ m_shared_group->rollback();
+ }
+
+ bool has_table(StringData name) const noexcept
+ {
+ return get_group().has_table(name);
+ }
+
+ TableRef get_table(size_t table_ndx) const
+ {
+ return get_group().get_table(table_ndx); // Throws
+ }
+
+ TableRef get_table(StringData name) const
+ {
+ return get_group().get_table(name); // Throws
+ }
+
+ TableRef add_table(StringData name, bool require_unique_name = true) const
+ {
+ return get_group().add_table(name, require_unique_name); // Throws
+ }
+
+ TableRef get_or_add_table(StringData name, bool* was_added = nullptr) const
+ {
+ return get_group().get_or_add_table(name, was_added); // Throws
+ }
+
+ Group& get_group() const noexcept;
+
+ /// Get the version of the snapshot on which this write transaction is
+ /// based.
+ SharedGroup::version_type get_version() const noexcept;
+
+ SharedGroup::version_type commit()
+ {
+ REALM_ASSERT(m_shared_group);
+ SharedGroup::version_type new_version = m_shared_group->commit();
+ m_shared_group = nullptr;
+ return new_version;
+ }
+
+ void rollback() noexcept
+ {
+ REALM_ASSERT(m_shared_group);
+ m_shared_group->rollback();
+ m_shared_group = nullptr;
+ }
+
+private:
+ SharedGroup* m_shared_group;
+};
+
+
+// Implementation:
+
+struct SharedGroup::BadVersion : std::exception {
+};
+
+inline SharedGroup::SharedGroup(const std::string& file, bool no_create, const SharedGroupOptions options)
+ : m_group(Group::shared_tag())
+ , m_upgrade_callback(std::move(options.upgrade_callback))
+{
+ open(file, no_create, options); // Throws
+}
+
+inline SharedGroup::SharedGroup(unattached_tag) noexcept
+ : m_group(Group::shared_tag())
+{
+}
+
+inline SharedGroup::SharedGroup(Replication& repl, const SharedGroupOptions options)
+ : m_group(Group::shared_tag())
+ , m_upgrade_callback(std::move(options.upgrade_callback))
+{
+ open(repl, options); // Throws
+}
+
+inline void SharedGroup::open(const std::string& path, bool no_create_file, const SharedGroupOptions options)
+{
+ // Exception safety: Since open() is called from constructors, if it throws,
+ // it must leave the file closed.
+
+ bool is_backend = false;
+ do_open(path, no_create_file, is_backend, options); // Throws
+}
+
+inline void SharedGroup::open(Replication& repl, const SharedGroupOptions options)
+{
+ // Exception safety: Since open() is called from constructors, if it throws,
+ // it must leave the file closed.
+
+ REALM_ASSERT(!is_attached());
+
+ repl.initialize(*this); // Throws
+
+ typedef _impl::GroupFriend gf;
+ gf::set_replication(m_group, &repl);
+
+ std::string file = repl.get_database_path();
+ bool no_create = false;
+ bool is_backend = false;
+ do_open(file, no_create, is_backend, options); // Throws
+}
+
+inline bool SharedGroup::is_attached() const noexcept
+{
+ return m_file_map.is_attached();
+}
+
+inline SharedGroup::TransactStage SharedGroup::get_transact_stage() const noexcept
+{
+ return m_transact_stage;
+}
+
+inline SharedGroup::version_type SharedGroup::get_version_of_bound_snapshot() const noexcept
+{
+ return m_read_lock.m_version;
+}
+
+class SharedGroup::ReadLockUnlockGuard {
+public:
+ ReadLockUnlockGuard(SharedGroup& shared_group, ReadLockInfo& read_lock) noexcept
+ : m_shared_group(shared_group)
+ , m_read_lock(&read_lock)
+ {
+ }
+ ~ReadLockUnlockGuard() noexcept
+ {
+ if (m_read_lock)
+ m_shared_group.release_read_lock(*m_read_lock);
+ }
+ void release() noexcept
+ {
+ m_read_lock = 0;
+ }
+
+private:
+ SharedGroup& m_shared_group;
+ ReadLockInfo* m_read_lock;
+};
+
+
+template <typename T>
+struct SharedGroup::Handover {
+ std::unique_ptr<typename T::HandoverPatch> patch;
+ std::unique_ptr<T> clone;
+ VersionID version;
+};
+
+template <typename T>
+std::unique_ptr<SharedGroup::Handover<T>> SharedGroup::export_for_handover(const T& accessor, ConstSourcePayload mode)
+{
+ if (m_transact_stage != transact_Reading)
+ throw LogicError(LogicError::wrong_transact_state);
+ std::unique_ptr<Handover<T>> result(new Handover<T>());
+ // Implementation note:
+ // often, the return value from clone will be T*, BUT it may be ptr to some
+ // base of T instead, so we must cast it to T*. This is always safe, because
+ // no matter the type, clone() will clone the actual accessor instance, and
+ // hence return an instance of the same type.
+ result->clone.reset(dynamic_cast<T*>(accessor.clone_for_handover(result->patch, mode).release()));
+ result->version = get_version_of_current_transaction();
+ return move(result);
+}
+
+
+template <typename T>
+std::unique_ptr<SharedGroup::Handover<BasicRow<T>>> SharedGroup::export_for_handover(const BasicRow<T>& accessor)
+{
+ if (m_transact_stage != transact_Reading)
+ throw LogicError(LogicError::wrong_transact_state);
+ std::unique_ptr<Handover<BasicRow<T>>> result(new Handover<BasicRow<T>>());
+ // See implementation note above.
+ result->clone.reset(dynamic_cast<BasicRow<T>*>(accessor.clone_for_handover(result->patch).release()));
+ result->version = get_version_of_current_transaction();
+ return move(result);
+}
+
+
+template <typename T>
+std::unique_ptr<SharedGroup::Handover<T>> SharedGroup::export_for_handover(T& accessor, MutableSourcePayload mode)
+{
+ if (m_transact_stage != transact_Reading)
+ throw LogicError(LogicError::wrong_transact_state);
+ std::unique_ptr<Handover<T>> result(new Handover<T>());
+ // see implementation note above.
+ result->clone.reset(dynamic_cast<T*>(accessor.clone_for_handover(result->patch, mode).release()));
+ result->version = get_version_of_current_transaction();
+ return move(result);
+}
+
+
+template <typename T>
+std::unique_ptr<T> SharedGroup::import_from_handover(std::unique_ptr<SharedGroup::Handover<T>> handover)
+{
+ if (handover->version != get_version_of_current_transaction()) {
+ throw BadVersion();
+ }
+ std::unique_ptr<T> result = move(handover->clone);
+ result->apply_and_consume_patch(handover->patch, m_group);
+ return result;
+}
+
+template <class O>
+inline void SharedGroup::advance_read(O* observer, VersionID version_id)
+{
+ if (m_transact_stage != transact_Reading)
+ throw LogicError(LogicError::wrong_transact_state);
+
+ // It is an error if the new version precedes the currently bound one.
+ if (version_id.version < m_read_lock.m_version)
+ throw LogicError(LogicError::bad_version);
+
+ _impl::History* hist = get_history(); // Throws
+ if (!hist)
+ throw LogicError(LogicError::no_history);
+
+ do_advance_read(observer, version_id, *hist); // Throws
+}
+
+template <class O>
+inline void SharedGroup::promote_to_write(O* observer)
+{
+ if (m_transact_stage != transact_Reading)
+ throw LogicError(LogicError::wrong_transact_state);
+
+ _impl::History* hist = get_history(); // Throws
+ if (!hist)
+ throw LogicError(LogicError::no_history);
+
+ do_begin_write(); // Throws
+ try {
+ VersionID version = VersionID(); // Latest
+ bool history_updated = do_advance_read(observer, version, *hist); // Throws
+
+ Replication* repl = m_group.get_replication();
+ REALM_ASSERT(repl); // Presence of `repl` follows from the presence of `hist`
+ version_type current_version = m_read_lock.m_version;
+ repl->initiate_transact(current_version, history_updated); // Throws
+
+ // If the group has no top array (top_ref == 0), create a new node
+ // structure for an empty group now, to be ready for modifications. See
+ // also Group::attach_shared().
+ using gf = _impl::GroupFriend;
+ gf::create_empty_group_when_missing(m_group); // Throws
+ }
+ catch (...) {
+ do_end_write();
+ throw;
+ }
+
+ set_transact_stage(transact_Writing);
+}
+
+template <class O>
+inline void SharedGroup::rollback_and_continue_as_read(O* observer)
+{
+ if (m_transact_stage != transact_Writing)
+ throw LogicError(LogicError::wrong_transact_state);
+
+ _impl::History* hist = get_history(); // Throws
+ if (!hist)
+ throw LogicError(LogicError::no_history);
+
+ // Mark all managed space (beyond the attached file) as free.
+ using gf = _impl::GroupFriend;
+ gf::reset_free_space_tracking(m_group); // Throws
+
+ BinaryData uncommitted_changes = hist->get_uncommitted_changes();
+
+ // FIXME: We are currently creating two transaction log parsers, one here,
+ // and one in advance_transact(). That is wasteful as the parser creation is
+ // expensive.
+ _impl::SimpleInputStream in(uncommitted_changes.data(), uncommitted_changes.size());
+ _impl::TransactLogParser parser; // Throws
+ _impl::TransactReverser reverser;
+ parser.parse(in, reverser); // Throws
+
+ if (observer && uncommitted_changes.size()) {
+ _impl::ReversedNoCopyInputStream reversed_in(reverser);
+ parser.parse(reversed_in, *observer); // Throws
+ observer->parse_complete(); // Throws
+ }
+
+ ref_type top_ref = m_read_lock.m_top_ref;
+ size_t file_size = m_read_lock.m_file_size;
+ _impl::ReversedNoCopyInputStream reversed_in(reverser);
+ gf::advance_transact(m_group, top_ref, file_size, reversed_in); // Throws
+
+ do_end_write();
+
+ Replication* repl = gf::get_replication(m_group);
+ REALM_ASSERT(repl); // Presence of `repl` follows from the presence of `hist`
+ repl->abort_transact();
+
+ set_transact_stage(transact_Reading);
+}
+
+template <class O>
+inline bool SharedGroup::do_advance_read(O* observer, VersionID version_id, _impl::History& hist)
+{
+ ReadLockInfo new_read_lock;
+ grab_read_lock(new_read_lock, version_id); // Throws
+ REALM_ASSERT(new_read_lock.m_version >= m_read_lock.m_version);
+ if (new_read_lock.m_version == m_read_lock.m_version) {
+ release_read_lock(new_read_lock);
+ // _impl::History::update_early_from_top_ref() was not called
+ return false;
+ }
+
+ ReadLockUnlockGuard g(*this, new_read_lock);
+ {
+ version_type new_version = new_read_lock.m_version;
+ size_t new_file_size = new_read_lock.m_file_size;
+ ref_type new_top_ref = new_read_lock.m_top_ref;
+
+ // Synchronize readers view of the file
+ SlabAlloc& alloc = m_group.m_alloc;
+ alloc.update_reader_view(new_file_size);
+
+ hist.update_early_from_top_ref(new_version, new_file_size, new_top_ref); // Throws
+ }
+
+ if (observer) {
+ // This has to happen in the context of the originally bound snapshot
+ // and while the read transaction is still in a fully functional state.
+ _impl::TransactLogParser parser;
+ version_type old_version = m_read_lock.m_version;
+ version_type new_version = new_read_lock.m_version;
+ _impl::ChangesetInputStream in(hist, old_version, new_version);
+ parser.parse(in, *observer); // Throws
+ observer->parse_complete(); // Throws
+ }
+
+ // The old read lock must be retained for as long as the change history is
+ // accessed (until Group::advance_transact() returns). This ensures that the
+ // oldest needed changeset remains in the history, even when the history is
+ // implemented as a separate unversioned entity outside the Realm (i.e., the
+ // old implementation and ShortCircuitHistory in
+ // test_lang_Bind_helper.cpp). On the other hand, if it had been the case,
+ // that the history was always implemented as a versioned entity, that was
+ // part of the Realm state, then it would not have been necessary to retain
+ // the old read lock beyond this point.
+
+ {
+ version_type old_version = m_read_lock.m_version;
+ version_type new_version = new_read_lock.m_version;
+ ref_type new_top_ref = new_read_lock.m_top_ref;
+ size_t new_file_size = new_read_lock.m_file_size;
+ _impl::ChangesetInputStream in(hist, old_version, new_version);
+ m_group.advance_transact(new_top_ref, new_file_size, in); // Throws
+ }
+
+ g.release();
+ release_read_lock(m_read_lock);
+ m_read_lock = new_read_lock;
+
+ return true; // _impl::History::update_early_from_top_ref() was called
+}
+
+inline _impl::History* SharedGroup::get_history()
+{
+ using gf = _impl::GroupFriend;
+ if (Replication* repl = gf::get_replication(m_group))
+ return repl->get_history();
+ return 0;
+}
+
+inline int SharedGroup::get_file_format_version() const noexcept
+{
+ using gf = _impl::GroupFriend;
+ return gf::get_file_format_version(m_group);
+}
+
+
+// The purpose of this class is to give internal access to some, but
+// not all of the non-public parts of the SharedGroup class.
+class _impl::SharedGroupFriend {
+public:
+ static Group& get_group(SharedGroup& sg) noexcept
+ {
+ return sg.m_group;
+ }
+
+ template <class O>
+ static void advance_read(SharedGroup& sg, O* obs, SharedGroup::VersionID ver)
+ {
+ sg.advance_read(obs, ver); // Throws
+ }
+
+ template <class O>
+ static void promote_to_write(SharedGroup& sg, O* obs)
+ {
+ sg.promote_to_write(obs); // Throws
+ }
+
+ static SharedGroup::version_type commit_and_continue_as_read(SharedGroup& sg)
+ {
+ return sg.commit_and_continue_as_read(); // Throws
+ }
+
+ template <class O>
+ static void rollback_and_continue_as_read(SharedGroup& sg, O* obs)
+ {
+ sg.rollback_and_continue_as_read(obs); // Throws
+ }
+
+ static void async_daemon_open(SharedGroup& sg, const std::string& file)
+ {
+ bool no_create = true;
+ bool is_backend = true;
+ SharedGroupOptions options;
+ options.durability = SharedGroupOptions::Durability::Async;
+ options.encryption_key = nullptr;
+ options.allow_file_format_upgrade = false;
+ sg.do_open(file, no_create, is_backend, options); // Throws
+ }
+
+ static int get_file_format_version(const SharedGroup& sg) noexcept
+ {
+ return sg.get_file_format_version();
+ }
+
+ static SharedGroup::version_type get_version_of_latest_snapshot(SharedGroup& sg)
+ {
+ return sg.get_version_of_latest_snapshot();
+ }
+
+ static SharedGroup::version_type get_version_of_bound_snapshot(const SharedGroup& sg) noexcept
+ {
+ return sg.get_version_of_bound_snapshot();
+ }
+};
+
+inline const Group& ReadTransaction::get_group() const noexcept
+{
+ using sgf = _impl::SharedGroupFriend;
+ return sgf::get_group(m_shared_group);
+}
+
+inline SharedGroup::version_type ReadTransaction::get_version() const noexcept
+{
+ using sgf = _impl::SharedGroupFriend;
+ return sgf::get_version_of_bound_snapshot(m_shared_group);
+}
+
+inline Group& WriteTransaction::get_group() const noexcept
+{
+ REALM_ASSERT(m_shared_group);
+ using sgf = _impl::SharedGroupFriend;
+ return sgf::get_group(*m_shared_group);
+}
+
+inline SharedGroup::version_type WriteTransaction::get_version() const noexcept
+{
+ using sgf = _impl::SharedGroupFriend;
+ return sgf::get_version_of_bound_snapshot(*m_shared_group);
+}
+
+} // namespace realm
+
+#endif // REALM_GROUP_SHARED_HPP
fnp / wl-app.git / blobdiff