gian/ceph-mirror
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2026-08-02 07:03:18 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #66572 from kotreshhr/mirror-multithreaded

Browse Source 
tools/cephfs_mirror: Multi-threaded Mirroring

Reviewed-by: Venky Shankar <vshankar@redhat.com>
This commit is contained in:
Venky Shankar 2026-02-25 17:54:30 +05:30 committed by GitHub
commit 81cb89b393
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 797 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
PendingReleaseNotes
View File

@ -37,6 +37,19 @@
date-time. The `schedule ls` output displays the start time in UTC, including
the date and time in the format "%Y-%m-%d %H:%M:00". The `schedule status`
output now displays the next schedule time in UTC.
* CephFS Mirroring: Now utilizes a multi-threaded architecture to improve synchronization
performance. The workload is split into two distinct thread pools: a crawler thread pool, which
manages snapshot crawl and a data synchronization thread pool, which handles concurrent file
transfers. Users can fine-tune these operations using configuration parameters:
cephfs_mirror_max_concurrent_directory_syncs (controlling the number of concurrent snapshots being crawled)
and cephfs_mirror_max_datasync_threads (controlling the total threads available for data sync).
For more information, see https://tracker.ceph.com/issues/73452
* CephFS Mirroring: Improved incremental synchronization behavior in CephFS mirroring. Previously,
block-level delta synchronization was used for all files regardless of size. With this change,
blockdiff is applied only to files larger than a configurable threshold, while smaller files are
synchronized using full copy, as blockdiff is not efficient for small files. The threshold is
controlled by the new configuration option cephfs_mirror_blockdiff_min_file_size (default: 16_M).
For more information, see https://tracker.ceph.com/issues/73452
>=20.0.0

13
doc/cephfs/cephfs-mirroring.rst
View File

@ -552,6 +552,7 @@ Configuration Options
---------------------
.. confval:: cephfs_mirror_max_concurrent_directory_syncs
.. confval:: cephfs_mirror_max_datasync_threads
.. confval:: cephfs_mirror_action_update_interval
.. confval:: cephfs_mirror_restart_mirror_on_blocklist_interval
.. confval:: cephfs_mirror_max_snapshot_sync_per_cycle
@ -561,6 +562,7 @@ Configuration Options
.. confval:: cephfs_mirror_restart_mirror_on_failure_interval
.. confval:: cephfs_mirror_mount_timeout
.. confval:: cephfs_mirror_perf_stats_prio
.. confval:: cephfs_mirror_blockdiff_min_file_size
Re-adding Peers
---------------
@ -572,3 +574,14 @@ in the command output). Also, it is recommended to purge synchronized directorie
from the peer before re-adding it to another file system (especially those directories
which might exist in the new primary file system). This is not required if re-adding
a peer to the same primary file system it was earlier synchronized from.
Multi-threaded snapshot sync
----------------------------
CephFS mirroring now utilizes a multi-threaded architecture to improve synchronization
performance. The workload is split into two distinct thread pools: a crawler thread pool, which
manages snapshot crawl and a data synchronization thread pool, which handles concurrent file
transfers. Users can fine-tune these operations using configuration parameters:
- ``cephfs_mirror_max_concurrent_directory_syncs``: controls the number of concurrent snapshots being crawled.
- ``cephfs_mirror_max_datasync_threads``: controls the total threads available for data sync.
For more information, see https://tracker.ceph.com/issues/73452

31
src/common/options/cephfs-mirror.yaml.in
View File

@ -5,13 +5,38 @@ options:
- name: cephfs_mirror_max_concurrent_directory_syncs
type: uint
level: advanced
desc: maximum number of concurrent snapshot synchronization threads
long_desc: maximum number of directory snapshots that can be synchronized concurrently
by cephfs-mirror daemon. Controls the number of synchronization threads.
desc: maximum number of concurrent snapshot synchronization crawler threads
long_desc: maximum number of directory snapshots that can be crawled concurrently
by cephfs-mirror daemon. Controls the number of synchronization crawler threads.
Note that the crawler threads also does entry operations like directory creations,
file deletes and snapshot deletes/renames.
default: 3
services:
- cephfs-mirror
min: 1
- name: cephfs_mirror_max_datasync_threads
type: uint
level: advanced
desc: maximum number of concurrent snapshot data synchronization threads
long_desc: specifies the maximum number of worker threads in the CephFS mirror
data synchronization thread pool. These threads process file synchronization tasks
produced by crawler threads for mirrored directory snapshots.
default: 6
services:
- cephfs-mirror
min: 1
- name: cephfs_mirror_blockdiff_min_file_size
type: size
level: advanced
desc: minimum file size threshold in bytes above which block-level diff is used during CephFS mirroring.
long_desc: defines the minimum file size, in bytes, required for CephFS mirroring to use block-level
delta synchronization instead of performing a full file copy. When a file’s size is greater than to
this threshold, the mirroring engine attempts to synchronize only the modified block extents between
snapshots. For files smaller than or equal to this value, a full file copy is performed instead, as
block-level diff may not provide meaningful performance benefits for small files.
default: 16_M
services:
- cephfs-mirror
- name: cephfs_mirror_action_update_interval
type: secs
level: advanced

16
src/tools/cephfs_mirror/FSMirror.h
View File

@ -58,13 +58,11 @@ public:
}
monotime get_failed_ts() {
std::scoped_lock locker(m_lock);
return m_failed_ts;
return m_failed_ts.load(std::memory_order_relaxed);
}
void set_failed_ts() {
std::scoped_lock locker(m_lock);
m_failed_ts = clock::now();
m_failed_ts.store(clock::now(), std::memory_order_relaxed);
}
bool is_blocklisted() {
@ -73,13 +71,11 @@ public:
}
monotime get_blocklisted_ts() {
std::scoped_lock locker(m_lock);
return m_blocklisted_ts;
return m_blocklisted_ts.load(std::memory_order_relaxed);
}
void set_blocklisted_ts() {
std::scoped_lock locker(m_lock);
m_blocklisted_ts = clock::now();
m_blocklisted_ts.store(clock::now(), std::memory_order_relaxed);
}
Peers get_peers() {
@ -140,8 +136,8 @@ private:
}
};
monotime m_blocklisted_ts;
monotime m_failed_ts;
std::atomic<monotime> m_blocklisted_ts;
std::atomic<monotime> m_failed_ts;
CephContext *m_cct;
Filesystem m_filesystem;
uint64_t m_pool_id;

6
src/tools/cephfs_mirror/InstanceWatcher.cc
View File

@ -115,8 +115,10 @@ void InstanceWatcher::handle_rewatch_complete(int r) {
if (r == -EBLOCKLISTED) {
dout(0) << ": client blocklisted" <<dendl;
std::scoped_lock locker(m_lock);
m_blocklisted = true;
{
std::scoped_lock locker(m_lock);
m_blocklisted = true;
}
m_elistener.set_blocklisted_ts();
} else if (r == -ENOENT) {
derr << ": mirroring object deleted" << dendl;

6
src/tools/cephfs_mirror/MirrorWatcher.cc
View File

@ -91,8 +91,10 @@ void MirrorWatcher::handle_rewatch_complete(int r) {
if (r == -EBLOCKLISTED) {
dout(0) << ": client blocklisted" <<dendl;
std::scoped_lock locker(m_lock);
m_blocklisted = true;
{
std::scoped_lock locker(m_lock);
m_blocklisted = true;
}
m_elistener.set_blocklisted_ts();
} else if (r == -ENOENT) {
derr << ": mirroring object deleted" << dendl;

739
src/tools/cephfs_mirror/PeerReplayer.cc
View File

@ -178,7 +178,8 @@ PeerReplayer::PeerReplayer(CephContext *cct, FSMirror *fs_mirror,
m_local_mount(mount),
m_service_daemon(service_daemon),
m_asok_hook(new PeerReplayerAdminSocketHook(cct, filesystem, peer, this)),
m_lock(ceph::make_mutex("cephfs::mirror::PeerReplayer::" + stringify(peer.uuid))) {
m_lock(ceph::make_mutex("cephfs::mirror::PeerReplayer::" + stringify(peer.uuid))),
smq_lock(ceph::make_mutex("cephfs::mirror::PeerReplayer::smq" + stringify(peer.uuid))) {
// reset sync stats sent via service daemon
m_service_daemon->add_or_update_peer_attribute(m_filesystem.fscid, m_peer,
SERVICE_DAEMON_FAILED_DIR_COUNT_KEY, (uint64_t)0);
@ -298,16 +299,47 @@ int PeerReplayer::init() {
m_replayers.push_back(std::move(replayer));
}
auto nr_data_replayers = g_ceph_context->_conf.get_val<uint64_t>(
"cephfs_mirror_max_datasync_threads");
dout(20) << ": spawning " << nr_data_replayers << " snapshot data replayer(s)" << dendl;
while (nr_data_replayers-- > 0) {
std::unique_ptr<SnapshotDataSyncThread> data_replayer(
new SnapshotDataSyncThread(this));
std::string name("d_replayer-" + stringify(nr_data_replayers));
data_replayer->create(name.c_str());
m_data_replayers.push_back(std::move(data_replayer));
}
return 0;
}
void PeerReplayer::shutdown() {
dout(20) << dendl;
bool expected = false;
if (!m_stopping.compare_exchange_strong(expected, true)) {
dout(1) << ": shutdown is already in progress - return"<< dendl;
return;
}
// wake up all datasync threads waiting on syncm_q
{
std::scoped_lock locker(m_lock);
ceph_assert(!m_stopping);
m_stopping = true;
std::unique_lock smq_l1(smq_lock);
for (auto& syncm : syncm_q) {
std::unique_lock sdq_l1(syncm->get_sdq_lock());
syncm->sdq_cv_notify_all_unlocked();
}
smq_cv.notify_all(); // wake up syncm_q wait
}
// Join data sync threads first
for (auto &replayer : m_data_replayers) {
replayer->join();
}
m_data_replayers.clear();
// Wake up crawler thread shutdown wait after datasync thread die
{
std::scoped_lock lock(m_lock);
m_cond.notify_all();
}
@ -315,6 +347,7 @@ void PeerReplayer::shutdown() {
replayer->join();
}
m_replayers.clear();
ceph_unmount(m_remote_mount);
ceph_release(m_remote_mount);
m_remote_mount = nullptr;
@ -349,6 +382,14 @@ void PeerReplayer::remove_directory(string_view dir_root) {
m_cond.notify_all();
}
void PeerReplayer::enqueue_syncm(const std::shared_ptr<SyncMechanism>& item) {
dout(20) << ": Enqueue syncm object=" << item << dendl;
std::lock_guard lock(smq_lock);
syncm_q.push_back(item);
smq_cv.notify_all();
}
boost::optional<std::string> PeerReplayer::pick_directory() {
dout(20) << dendl;
@ -611,18 +652,17 @@ int PeerReplayer::propagate_snap_renames(
return 0;
}
int PeerReplayer::remote_mkdir(const std::string &epath, const struct ceph_statx &stx,
const FHandles &fh) {
int PeerReplayer::SyncMechanism::remote_mkdir(const std::string &epath, const struct ceph_statx &stx) {
dout(10) << ": remote epath=" << epath << dendl;
int r = ceph_mkdirat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFDIR);
int r = ceph_mkdirat(m_remote, m_fh->r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFDIR);
if (r < 0 && r != -EEXIST) {
derr << ": failed to create remote directory=" << epath << ": " << cpp_strerror(r)
<< dendl;
return r;
}
r = ceph_chownat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_uid, stx.stx_gid,
r = ceph_chownat(m_remote, m_fh->r_fd_dir_root, epath.c_str(), stx.stx_uid, stx.stx_gid,
AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to chown remote directory=" << epath << ": " << cpp_strerror(r)
@ -630,7 +670,7 @@ int PeerReplayer::remote_mkdir(const std::string &epath, const struct ceph_statx
return r;
}
r = ceph_chmodat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFMT,
r = ceph_chmodat(m_remote, m_fh->r_fd_dir_root, epath.c_str(), stx.stx_mode & ~S_IFMT,
AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to chmod remote directory=" << epath << ": " << cpp_strerror(r)
@ -640,7 +680,7 @@ int PeerReplayer::remote_mkdir(const std::string &epath, const struct ceph_statx
struct timespec times[] = {{stx.stx_atime.tv_sec, stx.stx_atime.tv_nsec},
{stx.stx_mtime.tv_sec, stx.stx_mtime.tv_nsec}};
r = ceph_utimensat(m_remote_mount, fh.r_fd_dir_root, epath.c_str(), times, AT_SYMLINK_NOFOLLOW);
r = ceph_utimensat(m_remote, m_fh->r_fd_dir_root, epath.c_str(), times, AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to change [am]time on remote directory=" << epath << ": "
<< cpp_strerror(r) << dendl;
@ -766,11 +806,6 @@ int PeerReplayer::copy_to_remote(const std::string &dir_root, const std::string
<< cpp_strerror(r) << dendl;
goto freeptr;
}
r = ceph_fsync(m_remote_mount, r_fd, 0);
if (r < 0) {
derr << ": failed to sync data for file path=" << epath << ": "
<< cpp_strerror(r) << dendl;
}
}
freeptr:
@ -790,10 +825,12 @@ close_local_fd:
return -EINVAL;
}
dout(20) << ": dir_root=" << dir_root << ", epath=" << epath << " error=" << r << " synced" << dendl;
return r == 0 ? 0 : r;
}
int PeerReplayer::remote_file_op(SyncMechanism *syncm, const std::string &dir_root,
int PeerReplayer::remote_file_op(std::shared_ptr<SyncMechanism>& syncm, const std::string &dir_root,
const std::string &epath, const struct ceph_statx &stx,
bool sync_check, const FHandles &fh, bool need_data_sync, bool need_attr_sync) {
dout(10) << ": dir_root=" << dir_root << ", epath=" << epath << ", need_data_sync=" << need_data_sync
@ -1272,20 +1309,129 @@ int PeerReplayer::sync_perms(const std::string& path) {
return 0;
}
PeerReplayer::SyncMechanism::SyncMechanism(MountRef local, MountRef remote, FHandles *fh,
PeerReplayer::SyncMechanism::SyncMechanism(PeerReplayer& peer_replayer, std::string_view dir_root,
MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, const Snapshot &current,
boost::optional<Snapshot> prev)
: m_local(local),
: m_peer_replayer(peer_replayer),
m_dir_root(dir_root),
m_local(local),
m_remote(remote),
m_fh(fh),
m_peer(peer),
m_current(current),
m_prev(prev) {
m_prev(prev),
sdq_lock(ceph::make_mutex("cephfs::mirror::PeerReplayer::SyncMechanism" + stringify(peer.uuid))) {
}
PeerReplayer::SyncMechanism::~SyncMechanism() {
}
void PeerReplayer::SyncMechanism::push_dataq_entry(SyncEntry e) {
dout(10) << ": snapshot data replayer dataq pushed" << " syncm=" << this
<< " epath=" << e.epath << dendl;
std::unique_lock lock(sdq_lock);
m_sync_dataq.push(std::move(e));
sdq_cv.notify_all();
}
bool PeerReplayer::SyncMechanism::pop_dataq_entry(SyncEntry &out_entry) {
std::unique_lock smq_lock(m_peer_replayer.get_smq_lock());
std::unique_lock lock(sdq_lock);
dout(20) << ": snapshot data replayer waiting on m_sync_dataq, syncm=" << this << dendl;
sdq_cv.wait(lock, [this]{ return !m_sync_dataq.empty() || m_crawl_finished || m_datasync_error || m_crawl_error;});
while (true) {
bool ready = sdq_cv.wait_for(lock, 2s, [this] {
return m_peer_replayer.is_stopping() ||
!m_sync_dataq.empty() ||
m_crawl_finished ||
m_datasync_error ||
m_crawl_error;
});
// check for shutdown/blocklist/cancel
int r = 0;
if (m_peer_replayer.should_backoff(m_dir_root, &r)) {
dout(0) << ": backing off, shutdown/blocklist/cancel r=" << r << dendl;
if (r == -ECANCELED) {
set_datasync_error_unlocked(r);
dout(5) << ": snapshot data replayer, mirroring cancelled for syncm=" << this << dendl;
} else { //shutdown/blocklist
m_peer_replayer.mark_all_syncms_to_backoff_unlocked(r);
}
return false;
}
// predicate true
if (ready)
break;
// otherwise timeout occured, nothing to do - loop again
}
dout(20) << ": snapshot data replayer woke up to process m_syncm_dataq, syncm=" << this
<< " crawl_finished=" << m_crawl_finished << dendl;
if (m_datasync_error || m_crawl_error) {
dout(20) << ": snapshot data replayer, datasync_error="<< m_datasync_error
<< " crawl_error=" << m_crawl_error << " syncm=" << this << dendl;
return false;
}
if (m_sync_dataq.empty() && m_crawl_finished) {
dout(20) << ": snapshot data replayer - finished processing syncm=" << this
<< " Proceed with next syncm job " << dendl;
return false; // no more work
}
out_entry = std::move(m_sync_dataq.front());
m_sync_dataq.pop();
dout(10) << ": snapshot data replayer dataq popped" << " syncm=" << this
<< " epath=" << out_entry.epath << dendl;
return true;
}
bool PeerReplayer::SyncMechanism::has_pending_work() const {
std::unique_lock lock(sdq_lock);
const bool job_done =
m_sync_dataq.empty() && m_crawl_finished;
/* On crawl error, return true even if the queue is empty to
* - Dequeue the syncm object
* - Notify the crawler as it waits after the error for pending jobs to finish.
*/
if (m_crawl_error) {
// If m_in_flight > 0, those threads will take care of dequeue/notify, you just consume next job
if (m_in_flight > 0)
return false;
else
return true;
}
// No more work if datasync failed or everything is done
if (m_datasync_error || job_done)
return false;
return true;
}
void PeerReplayer::SyncMechanism::mark_crawl_finished(int ret) {
std::unique_lock lock(sdq_lock);
m_crawl_finished = true;
if (ret < 0)
m_crawl_error = true;
sdq_cv.notify_all();
}
// Returns false if there is any error during data sync
bool PeerReplayer::SyncMechanism::wait_for_sync() {
std::unique_lock lock(sdq_lock);
dout(20) << ": Waiting for data sync to be done to take snapshot - dir_root=" << m_dir_root
<< " current=" << m_current << " prev=" << (m_prev ? stringify(m_prev) : "")
<< " syncm=" << this << dendl;
sdq_cv.wait(lock, [this]{return m_sync_done || m_datasync_error;});
dout(20) << ": Woke up to take snapshot - dir_root=" << m_dir_root
<< " current=" << m_current << " prev=" << (m_prev ? stringify(m_prev) : "")
<< " syncm=" << this << dendl;
m_sync_done = false;
return m_datasync_error;
}
int PeerReplayer::SyncMechanism::get_changed_blocks(const std::string &epath,
const struct ceph_statx &stx, bool sync_check,
const std::function<int (uint64_t, struct cblock *)> &callback) {
@ -1303,11 +1449,11 @@ int PeerReplayer::SyncMechanism::get_changed_blocks(const std::string &epath,
return callback(block.num_blocks, block.b);
}
PeerReplayer::SnapDiffSync::SnapDiffSync(std::string_view dir_root, MountRef local, MountRef remote,
FHandles *fh, const Peer &peer, const Snapshot &current,
PeerReplayer::SnapDiffSync::SnapDiffSync(PeerReplayer& peer_replayer, std::string_view dir_root,
MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, const Snapshot &current,
boost::optional<Snapshot> prev)
: SyncMechanism(local, remote, fh, peer, current, prev),
m_dir_root(dir_root) {
: SyncMechanism(peer_replayer, dir_root, local, remote, fh, peer, current, prev) {
}
PeerReplayer::SnapDiffSync::~SnapDiffSync() {
@ -1447,10 +1593,86 @@ int PeerReplayer::SnapDiffSync::get_entry(std::string *epath, struct ceph_statx
dout(20) << ": entry=" << e_name << ", snapid=" << snapid << dendl;
if (e_name != "." && e_name != "..") {
break;
auto _epath = entry_path(entry.epath, e_name);
dout(20) << ": epath=" << _epath << dendl;
if (snapid == (*m_prev).second) {
dout(20) << ": epath=" << _epath << " is deleted in current snapshot " << dendl;
// do not depend on d_type reported in struct dirent as the
// delete and create could have been processed and a restart
// of an interrupted sync would use the incorrect unlink API.
// N.B.: snapdiff returns the deleted entry before the newly
// created one.
struct ceph_statx pstx;
r = ceph_statxat(m_remote, m_fh->r_fd_dir_root, _epath.c_str(), &pstx,
CEPH_STATX_MODE, AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0 && r != -ENOENT) {
derr << ": failed to stat remote entry=" << _epath << ", r=" << r << dendl;
return r;
}
if (r == 0) {
if (!S_ISDIR(pstx.stx_mode)) {
r = ceph_unlinkat(m_remote, m_fh->r_fd_dir_root, _epath.c_str(), 0);
} else {
r = purge_func(_epath);
}
if (r < 0) {
derr << ": failed to propagate missing dirs r=" << r << dendl;
return r;
}
}
m_deleted[entry.epath].emplace(e_name);
r = 1; //Continue with the outer loop
break;
}
struct ceph_statx estx;
r = ceph_statxat(m_local, m_fh->c_fd, _epath.c_str(), &estx,
CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID |
CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME,
AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to stat epath=" << epath << ", r=" << r << dendl;
return r;
}
bool pic = entry.is_purged_or_itype_changed() || m_deleted[entry.epath].contains(e_name);
if (S_ISDIR(estx.stx_mode)) {
SyncEntry se;
r = init_directory(_epath, estx, pic, &se);
if (r < 0) {
return r;
}
if (pic) {
dout(10) << ": purge or itype change (including parent) found for entry="
<< se.epath << dendl;
se.set_purged_or_itype_changed();
}
m_sync_stack.emplace(se);
dout(20) << ": Added directory to stack =" << _epath << dendl;
r = remote_mkdir(_epath, estx);
if (r < 0) {
derr << ": mkdir failed on remote. epath=" << _epath << ": " << cpp_strerror(r)
<< dendl;
return r;
}
//Fill epath to avoid caller treat this as failure and breaking the loop early.
*epath = _epath;
*stx = estx;
return r; // New directory added to stack
} else {
push_dataq_entry(SyncEntry(_epath, estx, !pic));
dout(10) << ": sync_check=" << *sync_check << " for epath=" << _epath << dendl;
}
}
}
if (r == 1)
continue;
if (r == 0) {
dout(10) << ": done for directory=" << entry.epath << dendl;
fini_directory(entry);
@ -1461,73 +1683,6 @@ int PeerReplayer::SnapDiffSync::get_entry(std::string *epath, struct ceph_statx
if (r < 0) {
return r;
}
auto _epath = entry_path(entry.epath, e_name);
dout(20) << ": epath=" << _epath << dendl;
if (snapid == (*m_prev).second) {
dout(20) << ": epath=" << _epath << " is deleted in current snapshot " << dendl;
// do not depend on d_type reported in struct dirent as the
// delete and create could have been processed and a restart
// of an interrupted sync would use the incorrect unlink API.
// N.B.: snapdiff returns the deleted entry before the newly
// created one.
struct ceph_statx pstx;
r = ceph_statxat(m_remote, m_fh->r_fd_dir_root, _epath.c_str(), &pstx,
CEPH_STATX_MODE, AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0 && r != -ENOENT) {
derr << ": failed to stat remote entry=" << _epath << ", r=" << r << dendl;
return r;
}
if (r == 0) {
if (!S_ISDIR(pstx.stx_mode)) {
r = ceph_unlinkat(m_remote, m_fh->r_fd_dir_root, _epath.c_str(), 0);
} else {
r = purge_func(_epath);
}
if (r < 0) {
derr << ": failed to propagate missing dirs r=" << r << dendl;
return r;
}
}
m_deleted[entry.epath].emplace(e_name);
continue;
}
struct ceph_statx estx;
r = ceph_statxat(m_local, m_fh->c_fd, _epath.c_str(), &estx,
CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID |
CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME,
AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to stat epath=" << epath << ", r=" << r << dendl;
return r;
}
bool pic = entry.is_purged_or_itype_changed() || m_deleted[entry.epath].contains(e_name);
if (S_ISDIR(estx.stx_mode)) {
SyncEntry se;
r = init_directory(_epath, estx, pic, &se);
if (r < 0) {
return r;
}
if (pic) {
dout(10) << ": purge or itype change (including parent) found for entry="
<< se.epath << dendl;
se.set_purged_or_itype_changed();
}
m_sync_stack.emplace(se);
}
*epath = _epath;
*stx = estx;
*sync_check = !pic;
dout(10) << ": sync_check=" << *sync_check << " for epath=" << *epath << dendl;
return 0;
}
*epath = "";
@ -1540,7 +1695,7 @@ int PeerReplayer::SnapDiffSync::get_changed_blocks(const std::string &epath,
dout(20) << ": dir_root=" << m_dir_root << ", epath=" << epath
<< ", sync_check=" << sync_check << dendl;
if (!sync_check) {
if (!sync_check || stx.stx_size <= m_peer_replayer.blockdiff_min_file_size) {
return SyncMechanism::get_changed_blocks(epath, stx, sync_check, callback);
}
@ -1586,7 +1741,7 @@ int PeerReplayer::SnapDiffSync::get_changed_blocks(const std::string &epath,
return r;
}
void PeerReplayer::SnapDiffSync::finish_sync() {
void PeerReplayer::SnapDiffSync::finish_crawl(int ret) {
dout(20) << dendl;
while (!m_sync_stack.empty()) {
@ -1600,12 +1755,16 @@ void PeerReplayer::SnapDiffSync::finish_sync() {
m_sync_stack.pop();
}
// Crawl and entry operations are done syncing here. So mark crawl finished here
mark_crawl_finished(ret);
}
PeerReplayer::RemoteSync::RemoteSync(MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, const Snapshot &current,
boost::optional<Snapshot> prev)
: SyncMechanism(local, remote, fh, peer, current, prev) {
PeerReplayer::RemoteSync::RemoteSync(PeerReplayer& peer_replayer, std::string_view dir_root,
MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, const Snapshot &current,
boost::optional<Snapshot> prev)
: SyncMechanism(peer_replayer, dir_root, local, remote, fh, peer, current, prev) {
}
PeerReplayer::RemoteSync::~RemoteSync() {
@ -1645,8 +1804,7 @@ int PeerReplayer::RemoteSync::get_entry(std::string *epath, struct ceph_statx *s
dout(20) << ": top of stack path=" << entry.epath << dendl;
if (!entry.is_directory()) {
*epath = entry.epath;
*stx = entry.stx;
push_dataq_entry(std::move(entry));
m_sync_stack.pop();
return 0;
}
@ -1663,7 +1821,6 @@ int PeerReplayer::RemoteSync::get_entry(std::string *epath, struct ceph_statx *s
}
int r;
std::string e_name;
while (true) {
struct dirent de;
r = ceph_readdirplus_r(m_local, entry.dirp, &de, NULL,
@ -1680,8 +1837,42 @@ int PeerReplayer::RemoteSync::get_entry(std::string *epath, struct ceph_statx *s
auto d_name = std::string(de.d_name);
if (d_name != "." && d_name != "..") {
e_name = d_name;
break;
struct ceph_statx cstx;
auto _epath = entry_path(entry.epath, d_name);
r = ceph_statxat(m_local, m_fh->c_fd, _epath.c_str(), &cstx,
CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID |
CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME,
AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to stat epath=" << _epath << ": " << cpp_strerror(r)
<< dendl;
return r;
}
if (S_ISDIR(cstx.stx_mode)) {
ceph_dir_result *dirp;
r = opendirat(m_local, m_fh->c_fd, _epath, AT_SYMLINK_NOFOLLOW, &dirp);
if (r < 0) {
derr << ": failed to open local directory=" << _epath << ": "
<< cpp_strerror(r) << dendl;
return r;
}
m_sync_stack.emplace(SyncEntry(_epath, dirp, cstx));
dout(20) << ": Added directory to stack =" << _epath << dendl;
r = remote_mkdir(_epath, cstx);
if (r < 0) {
derr << ": mkdir failed on remote. epath=" << _epath << ": " << cpp_strerror(r)
<< dendl;
return r;
}
//Fill epath to avoid caller treat this as failure and breaking the loop early.
*epath = _epath;
*stx = cstx;
return r; // New directory added to stack
} else {
push_dataq_entry(SyncEntry(_epath, cstx));
}
}
}
@ -1697,42 +1888,13 @@ int PeerReplayer::RemoteSync::get_entry(std::string *epath, struct ceph_statx *s
if (r < 0) {
return r;
}
struct ceph_statx cstx;
auto _epath = entry_path(entry.epath, e_name);
r = ceph_statxat(m_local, m_fh->c_fd, _epath.c_str(), &cstx,
CEPH_STATX_MODE | CEPH_STATX_UID | CEPH_STATX_GID |
CEPH_STATX_SIZE | CEPH_STATX_ATIME | CEPH_STATX_MTIME,
AT_STATX_DONT_SYNC | AT_SYMLINK_NOFOLLOW);
if (r < 0) {
derr << ": failed to stat epath=" << _epath << ": " << cpp_strerror(r)
<< dendl;
return r;
}
if (S_ISDIR(cstx.stx_mode)) {
ceph_dir_result *dirp;
r = opendirat(m_local, m_fh->c_fd, _epath, AT_SYMLINK_NOFOLLOW, &dirp);
if (r < 0) {
derr << ": failed to open local directory=" << _epath << ": "
<< cpp_strerror(r) << dendl;
break;
}
m_sync_stack.emplace(SyncEntry(_epath, dirp, cstx));
}
*epath = _epath;
*stx = cstx;
return 0;
}
*epath = "";
return 0;
}
void PeerReplayer::RemoteSync::finish_sync() {
void PeerReplayer::RemoteSync::finish_crawl(int ret) {
dout(20) << dendl;
while (!m_sync_stack.empty()) {
@ -1746,6 +1908,9 @@ void PeerReplayer::RemoteSync::finish_sync() {
m_sync_stack.pop();
}
// Crawl and entry operations are done syncing here. So mark stack finished here
mark_crawl_finished(ret);
}
int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot &current,
@ -1771,13 +1936,13 @@ int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot &cu
return r;
}
SyncMechanism *syncm;
std::shared_ptr<SyncMechanism> syncm;
if (fh.p_mnt == m_local_mount) {
syncm = new SnapDiffSync(dir_root, m_local_mount, m_remote_mount, &fh,
m_peer, current, prev);
syncm = std::make_shared<SnapDiffSync>(*this, dir_root, m_local_mount, m_remote_mount,
&fh, m_peer, current, prev);
} else {
syncm = new RemoteSync(m_local_mount, m_remote_mount, &fh,
m_peer, current, boost::none);
syncm = std::make_shared<RemoteSync>(*this, dir_root, m_local_mount, m_remote_mount,
&fh, m_peer, current, boost::none);
}
r = syncm->init_sync();
@ -1785,10 +1950,11 @@ int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot &cu
derr << ": failed to initialize sync mechanism" << dendl;
ceph_close(m_local_mount, fh.c_fd);
ceph_close(fh.p_mnt, fh.p_fd);
delete syncm;
return r;
}
enqueue_syncm(syncm);
// starting from this point we shouldn't care about manual closing of fh.c_fd,
// it will be closed automatically when bound tdirp is closed.
while (true) {
@ -1817,37 +1983,9 @@ int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot &cu
dout(10) << ": tree traversal done for dir_root=" << dir_root << dendl;
break;
}
if (S_ISDIR(stx.stx_mode)) {
r = remote_mkdir(epath, stx, fh);
if (r < 0) {
break;
}
} else {
bool need_data_sync = true;
bool need_attr_sync = true;
if (sync_check) {
r = should_sync_entry(epath, stx, fh,
&need_data_sync, &need_attr_sync);
if (r < 0) {
break;
}
}
dout(5) << ": entry=" << epath << ", data_sync=" << need_data_sync
<< ", attr_sync=" << need_attr_sync << dendl;
if (need_data_sync || need_attr_sync) {
r = remote_file_op(syncm, dir_root, epath, stx, sync_check, fh, need_data_sync, need_attr_sync);
if (r < 0) {
break;
}
}
dout(10) << ": done for epath=" << epath << dendl;
}
}
syncm->finish_sync();
delete syncm;
syncm->finish_crawl(r);
dout(20) << " cur:" << fh.c_fd
<< " prev:" << fh.p_fd
@ -1861,6 +1999,36 @@ int PeerReplayer::do_synchronize(const std::string &dir_root, const Snapshot &cu
// there is no need to close this fd manually.
ceph_close(fh.p_mnt, fh.p_fd);
// Wait for datasync threads to finish syncing
bool datasync_err = syncm->wait_for_sync();
if (r == 0 && !datasync_err) {
// All good, fsync remote fs and take the snapshot
dout(20) << ": syncing remote filesystem, dir_root=" << dir_root << dendl;
r = ceph_sync_fs(m_remote_mount);
if (r < 0) {
derr << ": failed to sync remote filesystem, dir_root=" << dir_root
<< ": " << cpp_strerror(r) << dendl;
return r;
}
auto cur_snap_id_str{stringify(current.second)};
snap_metadata snap_meta[] = {{PRIMARY_SNAP_ID_KEY.c_str(), cur_snap_id_str.c_str()}};
r = ceph_mksnap(m_remote_mount, dir_root.c_str(), current.first.c_str(), 0755,
snap_meta, sizeof(snap_meta)/sizeof(snap_metadata));
if (r < 0) {
derr << ": failed to snap remote directory dir_root=" << dir_root
<< ": " << cpp_strerror(r) << dendl;
return r;
}
} else if (datasync_err) {
r = syncm->get_datasync_errno();
derr << ": Datasync thread failed, bailing out without taking snap. dir_root="
<< dir_root << ": " << cpp_strerror(r) << dendl;
} else {
derr << ": Crawler thread failed, bailing out without taking snap. dir_root="
<< dir_root << ": " << cpp_strerror(r) << dendl;
}
return r;
}
@ -1906,21 +2074,11 @@ int PeerReplayer::synchronize(const std::string &dir_root, const Snapshot &curre
r = do_synchronize(dir_root, current);
}
}
// snap sync failed -- bail out!
if (r < 0) {
return r;
}
auto cur_snap_id_str{stringify(current.second)};
snap_metadata snap_meta[] = {{PRIMARY_SNAP_ID_KEY.c_str(), cur_snap_id_str.c_str()}};
r = ceph_mksnap(m_remote_mount, dir_root.c_str(), current.first.c_str(), 0755,
snap_meta, sizeof(snap_meta)/sizeof(snap_metadata));
if (r < 0) {
derr << ": failed to snap remote directory dir_root=" << dir_root
<< ": " << cpp_strerror(r) << dendl;
}
return r;
}
@ -1991,6 +2149,7 @@ int PeerReplayer::do_sync_snaps(const std::string &dir_root) {
double start = 0;
double end = 0;
double duration = 0;
uint64_t blockdiff_min_file_size_conf = 0;
for (; it != local_snap_map.end(); ++it) {
if (m_perf_counters) {
start = std::chrono::duration_cast<std::chrono::seconds>(clock::now().time_since_epoch()).count();
@ -1999,6 +2158,17 @@ int PeerReplayer::do_sync_snaps(const std::string &dir_root) {
m_perf_counters->tset(l_cephfs_mirror_peer_replayer_last_synced_start, t);
}
set_current_syncing_snap(dir_root, it->first, it->second);
// Check for blockdiff_min_file_size config change at the beginning of snapshot sync
blockdiff_min_file_size_conf = g_ceph_context->_conf.get_val<Option::size_t>(
"cephfs_mirror_blockdiff_min_file_size");
{
std::scoped_lock locker(m_lock);
if (blockdiff_min_file_size != blockdiff_min_file_size_conf) {
dout(10) << ": blockdiff_min_file_size changed" << " old=" << blockdiff_min_file_size
<< " new=" << blockdiff_min_file_size_conf << dendl;
blockdiff_min_file_size = blockdiff_min_file_size_conf;
}
}
boost::optional<Snapshot> prev = boost::none;
if (last_snap_id != 0) {
prev = std::make_pair(last_snap_name, last_snap_id);
@ -2107,6 +2277,221 @@ void PeerReplayer::run(SnapshotReplayerThread *replayer) {
}
}
bool PeerReplayer::is_syncm_active(const std::shared_ptr<PeerReplayer::SyncMechanism>& syncm_obj) {
return std::find(syncm_q.begin(), syncm_q.end(), syncm_obj) != syncm_q.end();
}
void PeerReplayer::remove_syncm(const std::shared_ptr<PeerReplayer::SyncMechanism>& syncm_obj)
{
// caller holds lock
auto it = std::find(syncm_q.begin(), syncm_q.end(), syncm_obj);
if (it != syncm_q.end()) {
syncm_q.erase(it);
}
}
/* The data sync threads should consume the next syncm job if the present syncm has no
* pending work. This can evidently happen if the last file being synced in the present
* syncm job is a large file. In this case, one data sync thread is busy syncing the
* large file, the rest of data sync threads could start consuming the next syncm job
* instead of being idle waiting for the last file to be synced from present syncm job.
*/
std::shared_ptr<PeerReplayer::SyncMechanism> PeerReplayer::pick_next_syncm_and_mark() {
// caller holds lock
for (auto& syncm : syncm_q) {
if (syncm->has_pending_work()) {
syncm->inc_in_flight();
return syncm;
}
}
return nullptr;
}
void PeerReplayer::mark_and_notify_syncms_to_backoff(int err) {
// caller holds the smq_lock
ceph_assert(ceph_mutex_is_locked_by_me(smq_lock));
for (auto& syncm : syncm_q) {
std::unique_lock sdq_lock(syncm->get_sdq_lock());
syncm->set_datasync_error_unlocked(err);
syncm->mark_backoff_unlocked();
if (get_active_datasync_threads() == 1) { //Last thread
// To wake up crawler thread whose dataq process is not started
syncm->sdq_cv_notify_all_unlocked();
}
}
// To wake up other datasync threads to speed up exit
smq_cv.notify_all();
}
void PeerReplayer::mark_all_syncms_to_backoff_unlocked(int err) {
// caller holds the smq_lock and sdq_lock
ceph_assert(ceph_mutex_is_locked_by_me(smq_lock));
for (auto& syncm : syncm_q) {
ceph_assert(ceph_mutex_is_locked_by_me(syncm->get_sdq_lock()));
syncm->set_datasync_error_unlocked(err);
syncm->mark_backoff_unlocked();
}
}
void PeerReplayer::notify_all_syncms_to_backoff() {
// caller holds the smq_lock and sdq_lock
ceph_assert(ceph_mutex_is_locked_by_me(smq_lock));
if (get_active_datasync_threads() == 1) { //Last thread
for (auto& syncm : syncm_q) {
ceph_assert(ceph_mutex_is_locked_by_me(syncm->get_sdq_lock()));
// To wake up crawler thread
syncm->sdq_cv_notify_all_unlocked();
}
}
// To wake up other datasync threads to speed up exit
smq_cv.notify_all();
}
void PeerReplayer::run_datasync(SnapshotDataSyncThread *data_replayer) {
dout(10) << ": snapshot datasync replayer=" << data_replayer << dendl;
/* The m_stopping is made atomic and m_lock is no longer required for state
* change or access. Hence data sync thread can get rid of waiting for is_stopping
* using m_lock
*/
while (true) {
bool shutdown = false;
bool blocklist = false;
std::shared_ptr<SyncMechanism> syncm;
{
std::unique_lock lock(smq_lock);
dout(20) << ": snapshot data replayer waiting for syncm to process" << dendl;
while (true) {
bool ready = smq_cv.wait_for(lock, 2s, [this, &syncm] {
syncm = pick_next_syncm_and_mark();
return is_stopping() || syncm != nullptr;
});
// immediate shutdown - predicate is true
if (is_stopping()) {
dout(5) << ": exiting snapshot data replayer=" << data_replayer
<< " as mirroring is shutting down" << dendl;
shutdown = true;
mark_and_notify_syncms_to_backoff(-EINPROGRESS);
break;
}
// work available - predicate is true
if (ready)
break;
// Timed wake up path - blocklist check
if (m_fs_mirror->is_blocklisted()) {
dout(5) << ": exiting snapshot data replayer=" << data_replayer
<< " as client is blocklisted" << dendl;
blocklist = true;
mark_and_notify_syncms_to_backoff(-EBLOCKLISTED);
break;
}
// otherwise timeout occured, nothing to do - loop again
}
if (shutdown || blocklist) {
break; //exit
}
// syncm is gauranteed to be non-null because of the predicate used in above wait.
dout(20) << ": snapshot data replayer woke up! syncm=" << syncm << dendl;
}
/* - FHandles are not thread safe, so don't use FHandles from SyncMechanism, open them locally here.
* - On crawl error, don't open handles as other threads could have dequeued syncm and notified
* crawler resulting in unregister of dir_root.
*/
int r = 0;
FHandles fh;
bool handles_opened = false;
const bool crawl_error = syncm->get_crawl_error();
if (!crawl_error) {
r = pre_sync_check_and_open_handles(std::string(syncm->get_m_dir_root()),
syncm->get_m_current(), syncm->get_m_prev(), &fh);
if (r < 0) {
dout(5) << ": open_handles failed, cannot proceed sync: " << cpp_strerror(r)
<< " dir_root=" << syncm->get_m_dir_root() << "syncm=" << syncm << dendl;
syncm->set_datasync_error(r); //don't do continue, as it needs to go through dequeue/notify logic
} else {
handles_opened = true;
}
}
// Wait on data sync queue for entries to process
SyncEntry entry;
while (syncm->pop_dataq_entry(entry)) {
bool need_data_sync = true;
bool need_attr_sync = true;
if (entry.sync_check) {
r = should_sync_entry(entry.epath, entry.stx, fh,
&need_data_sync, &need_attr_sync);
if (r < 0) {
dout(5) << ": should_sync_entry failed, cannot proceed sync: " << cpp_strerror(r)
<< " dir_root=" << syncm->get_m_dir_root() << " epath=" << entry.epath << dendl;
syncm->set_datasync_error(r);
break;
}
}
dout(5) << ": syncm=" << syncm << " entry=" << entry.epath << " data_sync="
<< need_data_sync << " attr_sync=" << need_attr_sync << dendl;
if (need_data_sync || need_attr_sync) {
r = remote_file_op(syncm, std::string(syncm->get_m_dir_root()),
entry.epath, entry.stx, entry.sync_check, fh,
need_data_sync, need_attr_sync);
if (r < 0) {
dout(5) << ": remote_file_op failed, cannot proceed sync: " << cpp_strerror(r)
<< " dir_root=" << syncm->get_m_dir_root() << " epath=" << entry.epath << dendl;
syncm->set_datasync_error(r);
break;
}
}
dout(10) << ": done for epath=" << entry.epath << " syncm=" << syncm << dendl;
}
// Close fds
if (handles_opened) {
ceph_close(m_local_mount, fh.c_fd);
ceph_close(fh.p_mnt, fh.p_fd);
}
// Dequeue syncm object after processing
{
std::unique_lock smq_l1(smq_lock);
std::unique_lock sdq_l1(syncm->get_sdq_lock());
// backoff ?
const bool syncm_backoff = syncm->get_backoff_unlocked();
int syncm_errno = syncm->get_datasync_errno_unlocked();
if (syncm_backoff && syncm_errno != -ECANCELED) {
notify_all_syncms_to_backoff(); //shutdown/blocklist
dout(5) << ": exiting snapshot data replayer=" << data_replayer
<< " as client is blocklisted or mirroring is shutting down, error=" << syncm_errno << dendl;
break; // exit
}
const bool last_in_flight_syncm = syncm->get_in_flight_unlocked() == 1;
const bool crawl_finished = syncm->get_crawl_finished_unlocked();
const bool sync_error =
syncm->get_datasync_error_unlocked() ||
syncm->get_crawl_error_unlocked();
if (!syncm_q.empty() && last_in_flight_syncm && (crawl_finished || sync_error)) {
if (sync_error && !is_syncm_active(syncm)){
dout(20) << ": syncm object=" << syncm << " already dequeued" << dendl;
} else {
dout(20) << ": Dequeue syncm object=" << syncm << dendl;
syncm->set_sync_finished_and_notify_unlocked(); // To wake up crawler thread waiting to take snapshot
if (syncm_q.front() == syncm) {
syncm_q.pop_front();
} else { // if syncms in the middle finishes first
remove_syncm(syncm);
}
dout(20) << ": syncm_q after removal " << syncm_q << dendl;
smq_cv.notify_all();
}
}
// Decrement should be within this locked block where the comparison happens, moving outside faults inc/dec logic
syncm->dec_in_flight_unlocked();
}
} // outer while
}
void PeerReplayer::peer_status(Formatter *f) {
std::scoped_lock locker(m_lock);
f->open_object_section("stats");

181
src/tools/cephfs_mirror/PeerReplayer.h
View File

@ -76,7 +76,7 @@ private:
};
bool is_stopping() {
return m_stopping;
return m_stopping.load(std::memory_order_acquire);
}
struct Replayer;
@ -95,6 +95,40 @@ private:
PeerReplayer *m_peer_replayer;
};
class SnapshotDataSyncThreadGuard {
public:
explicit SnapshotDataSyncThreadGuard(PeerReplayer *peer_replayer)
: m_peer_replayer(peer_replayer) {
m_peer_replayer->m_active_datasync_threads.fetch_add(1, std::memory_order_relaxed);
}
~SnapshotDataSyncThreadGuard() {
m_peer_replayer->m_active_datasync_threads.fetch_sub(1, std::memory_order_relaxed);
}
SnapshotDataSyncThreadGuard(const SnapshotDataSyncThreadGuard&) = delete;
SnapshotDataSyncThreadGuard& operator=(const SnapshotDataSyncThreadGuard&) = delete;
private:
PeerReplayer* m_peer_replayer;
};
class SnapshotDataSyncThread : public Thread {
public:
SnapshotDataSyncThread(PeerReplayer *peer_replayer)
: m_peer_replayer(peer_replayer) {
}
void *entry() override {
SnapshotDataSyncThreadGuard guard(m_peer_replayer); //active thread counter
m_peer_replayer->run_datasync(this);
return 0;
}
private:
PeerReplayer *m_peer_replayer;
};
struct DirRegistry {
int fd;
bool canceled = false;
@ -113,6 +147,7 @@ private:
// includes parent dentry purge
bool purged_or_itype_changed = false;
bool is_snapdiff = false;
bool sync_check = true;
SyncEntry() {
}
@ -122,6 +157,13 @@ private:
: epath(path),
stx(stx) {
}
SyncEntry(std::string_view path,
const struct ceph_statx &stx,
bool sync_check)
: epath(path),
stx(stx),
sync_check(sync_check) {
}
SyncEntry(std::string_view path,
ceph_dir_result *dirp,
const struct ceph_statx &stx)
@ -163,9 +205,10 @@ private:
class SyncMechanism {
public:
SyncMechanism(MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, /* keep dout happy */
const Snapshot &current, boost::optional<Snapshot> prev);
explicit SyncMechanism(PeerReplayer& peer_replayer, std::string_view dir_root,
MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, /* keep dout happy */
const Snapshot &current, boost::optional<Snapshot> prev);
virtual ~SyncMechanism() = 0;
virtual int init_sync() = 0;
@ -178,9 +221,83 @@ private:
const struct ceph_statx &stx, bool sync_check,
const std::function<int (uint64_t, struct cblock *)> &callback);
virtual void finish_sync() = 0;
virtual void finish_crawl(int ret) = 0;
void push_dataq_entry(PeerReplayer::SyncEntry e);
bool pop_dataq_entry(PeerReplayer::SyncEntry &out);
bool has_pending_work() const;
void mark_crawl_finished(int ret);
bool get_crawl_finished_unlocked() {
return m_crawl_finished;
}
void set_datasync_error(int err) {
std::unique_lock lock(sdq_lock);
m_datasync_error = true;
m_datasync_errno = err;
}
void set_datasync_error_unlocked(int err) {
m_datasync_error = true;
m_datasync_errno = err;
}
void mark_backoff_unlocked() {
m_backoff = true;
}
bool get_backoff_unlocked() {
return m_backoff;
}
bool get_datasync_error_unlocked() {
return m_datasync_error;
}
int get_datasync_errno() {
std::unique_lock lock(sdq_lock);
return m_datasync_errno;
}
int get_datasync_errno_unlocked() {
return m_datasync_errno;
}
bool get_crawl_error() {
std::unique_lock lock(sdq_lock);
return m_crawl_error;
}
bool get_crawl_error_unlocked() {
return m_crawl_error;
}
void inc_in_flight() {
std::unique_lock lock(sdq_lock);
++m_in_flight;
}
void dec_in_flight_unlocked() {
--m_in_flight;
}
int get_in_flight_unlocked() {
return m_in_flight;
}
ceph::mutex& get_sdq_lock() {
return sdq_lock;
}
std::string_view get_m_dir_root() {
return m_dir_root;
}
Snapshot get_m_current() const {
return m_current;
}
boost::optional<Snapshot> get_m_prev() const {
return m_prev;
}
void set_sync_finished_and_notify_unlocked() {
m_sync_done = true;
sdq_cv.notify_all();
}
void sdq_cv_notify_all_unlocked() {
sdq_cv.notify_all();
}
bool wait_for_sync();
int remote_mkdir(const std::string &epath, const struct ceph_statx &stx);
protected:
PeerReplayer& m_peer_replayer;
// It's not used in RemoteSync but required to be accessed in datasync threads
std::string m_dir_root;
MountRef m_local;
MountRef m_remote;
FHandles *m_fh;
@ -188,11 +305,23 @@ private:
Snapshot m_current;
boost::optional<Snapshot> m_prev;
std::stack<PeerReplayer::SyncEntry> m_sync_stack;
mutable ceph::mutex sdq_lock;
ceph::condition_variable sdq_cv;
std::queue<PeerReplayer::SyncEntry> m_sync_dataq;
int m_in_flight = 0;
bool m_crawl_finished = false;
bool m_crawl_error = false;
bool m_sync_done = false;
bool m_datasync_error = false;
int m_datasync_errno = 0;
bool m_backoff = false;
};
class RemoteSync : public SyncMechanism {
public:
RemoteSync(MountRef local, MountRef remote, FHandles *fh,
RemoteSync(PeerReplayer& peer_replayer, std::string_view dir_root,
MountRef local, MountRef remote, FHandles *fh,
const Peer &peer, /* keep dout happy */
const Snapshot &current, boost::optional<Snapshot> prev);
~RemoteSync();
@ -203,13 +332,13 @@ private:
const std::function<int (const std::string&)> &dirsync_func,
const std::function<int (const std::string&)> &purge_func);
void finish_sync();
void finish_crawl(int ret);
};
class SnapDiffSync : public SyncMechanism {
public:
SnapDiffSync(std::string_view dir_root, MountRef local, MountRef remote,
FHandles *fh, const Peer &peer, const Snapshot &current,
SnapDiffSync(PeerReplayer& peer_replayer, std::string_view dir_root, MountRef local,
MountRef remote, FHandles *fh, const Peer &peer, const Snapshot &current,
boost::optional<Snapshot> prev);
~SnapDiffSync();
@ -223,7 +352,7 @@ private:
const struct ceph_statx &stx, bool sync_check,
const std::function<int (uint64_t, struct cblock *)> &callback);
void finish_sync();
void finish_crawl(int ret);
private:
int init_directory(const std::string &epath,
@ -231,7 +360,6 @@ private:
int next_entry(SyncEntry &entry, std::string *e_name, snapid_t *snapid);
void fini_directory(SyncEntry &entry);
std::string m_dir_root;
std::map<std::string, std::set<std::string>> m_deleted;
};
@ -357,6 +485,7 @@ private:
}
typedef std::vector<std::unique_ptr<SnapshotReplayerThread>> SnapshotReplayers;
typedef std::vector<std::unique_ptr<SnapshotDataSyncThread>> SnapshotDataReplayers;
CephContext *m_cct;
FSMirror *m_fs_mirror;
@ -375,14 +504,33 @@ private:
ceph::condition_variable m_cond;
RadosRef m_remote_cluster;
MountRef m_remote_mount;
bool m_stopping = false;
std::atomic<bool> m_stopping{false};
SnapshotReplayers m_replayers;
SnapshotDataReplayers m_data_replayers;
std::atomic<int> m_active_datasync_threads{0};
ceph::mutex smq_lock;
ceph::condition_variable smq_cv;
std::deque<std::shared_ptr<SyncMechanism>> syncm_q;
uint64_t blockdiff_min_file_size = 0;
ServiceDaemonStats m_service_daemon_stats;
PerfCounters *m_perf_counters;
void run(SnapshotReplayerThread *replayer);
void run_datasync(SnapshotDataSyncThread *data_replayer);
void remove_syncm(const std::shared_ptr<SyncMechanism>& syncm_obj);
bool is_syncm_active(const std::shared_ptr<SyncMechanism>& syncm_obj);
std::shared_ptr<SyncMechanism> pick_next_syncm_and_mark();
int get_active_datasync_threads() const {
return m_active_datasync_threads.load(std::memory_order_relaxed);
}
void mark_and_notify_syncms_to_backoff(int err);
void mark_all_syncms_to_backoff_unlocked(int err);
void notify_all_syncms_to_backoff();
boost::optional<std::string> pick_directory();
int register_directory(const std::string &dir_root, SnapshotReplayerThread *replayer);
@ -421,13 +569,18 @@ private:
boost::optional<Snapshot> prev);
int do_sync_snaps(const std::string &dir_root);
int remote_mkdir(const std::string &epath, const struct ceph_statx &stx, const FHandles &fh);
int remote_file_op(SyncMechanism *syncm, const std::string &dir_root,
int remote_file_op(std::shared_ptr<SyncMechanism>& syncm, const std::string &dir_root,
const std::string &epath, const struct ceph_statx &stx,
bool sync_check, const FHandles &fh, bool need_data_sync, bool need_attr_sync);
int copy_to_remote(const std::string &dir_root, const std::string &epath, const struct ceph_statx &stx,
const FHandles &fh, uint64_t num_blocks, struct cblock *b);
int sync_perms(const std::string& path);
// add syncm to syncm_q
void enqueue_syncm(const std::shared_ptr<SyncMechanism>& item);
ceph::mutex& get_smq_lock() {
return smq_lock;
}
};
} // namespace mirror