jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Add cancellable file tree batch operations

master
evgeny 3 days ago
parent
ac17ffda54
commit
884eb4b412
2 changed files with 338 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 281
      tools/chatgui/src/member_files/batch.cpp
  2. 57
      tools/chatgui/src/member_files/batch.h

281
tools/chatgui/src/member_files/batch.cpp
Unescape View File

@ -0,0 +1,281 @@
#include "batch.h"
#include "../../transport/gui_bridge.h"
#include <QCoreApplication>
#include <QDir>
#include <QFileInfo>
#include <QTemporaryDir>
#include <QRandomGenerator>
#include <QPointer>
#include <QRunnable>
#include <QMessageBox>
#include <algorithm>
#include <cstring>
extern "C" {
#include "member_files/member_files.h"
#include "mem.h"
#include "debug_config.h"
}
static QString text(const char* s) { return QCoreApplication::translate("ChatGui", s); }
static quint64 nextId() { quint64 id; do { id = QRandomGenerator::global()->generate64(); } while (!id); return id; }
static bool pathBytes(const QString& path, bool remote) {
auto bytes = path.toUtf8();
return !bytes.contains('\0') && bytes.size() <= MF_PATH_MAX && (!remote || mf_path_valid(bytes.constData(), 0));
}
MemberFileBatch::MemberFileBatch(quint64 group, quint64 server, quint64 owner, Mode mode, QObject* parent)
: QObject(parent), m_group(group), m_server(server), m_owner(owner), m_mode(mode) {
m_pool.setMaxThreadCount(1);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "file batch: created group=%llu server=%llu owner=%llu mode=%d",
(unsigned long long)group, (unsigned long long)server, (unsigned long long)owner, int(mode));
}
MemberFileBatch::~MemberFileBatch() {
m_cancelled = true; cancelRequest(m_request); cancelRequest(m_transfer); m_pool.waitForDone();
}
void MemberFileBatch::cancelRequest(quint64 id) {
if (!id || !gui_bridge_is_uasync_ready()) return;
auto* q = (mf_cancel_request*)u_calloc(1, sizeof(mf_cancel_request));
if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "file batch: cancel allocation failed id=%llu", (unsigned long long)id); return; }
q->inst = gui_bridge_get_inst(); q->id = id; gui_bridge_post_uasync_fn(member_files_gui_cancel, q);
}
// Сначала проверяем всё локальное дерево; до успешного обхода сервер не изменяется.
void MemberFileBatch::upload(const QStringList& sources, const QString& destination) {
m_work = true;
m_pool.start(QRunnable::create([this, sources, destination] {
QList<MemberFileEntry> entries, pending;
QString error; QStringList roots;
for (const auto& source : sources) roots.append(QDir::cleanPath(source));
roots.removeDuplicates(); std::sort(roots.begin(), roots.end());
QStringList accepted;
for (const auto& root : roots) {
bool covered = false;
for (const auto& parent : accepted) if (root.startsWith(parent + '/')) { covered = true; break; }
if (covered) continue;
accepted.append(root);
QString name = QFileInfo(root).fileName();
pending.append({destination.isEmpty() ? name : destination + '/' + name, root});
}
QSet<QString> targets;
while (!pending.isEmpty() && !m_cancelled) {
auto entry = pending.takeFirst(); QFileInfo info(entry.local);
if (!info.isAbsolute() || !pathBytes(entry.local, false) || !pathBytes(entry.remote, true) ||
info.isSymLink() || !info.isReadable() || (!info.isFile() && !info.isDir()) || targets.contains(entry.remote)) {
error = text("Cannot copy this path: %1").arg(entry.local); break;
}
targets.insert(entry.remote); entry.directory = info.isDir(); entries.append(entry);
if (entries.size() + pending.size() > 100000) { error = text("Too many entries in this operation"); break; }
if (entry.directory) {
QDir dir(entry.local);
if (!dir.isReadable()) { error = text("Cannot read directory: %1").arg(entry.local); break; }
for (const auto& child : dir.entryInfoList(QDir::AllEntries | QDir::NoDotAndDotDot | QDir::Hidden | QDir::System, QDir::Name))
pending.append({entry.remote + '/' + child.fileName(), child.absoluteFilePath()});
}
}
QMetaObject::invokeMethod(this, [this, entries, error] {
m_work = false;
if (m_cancelled || !error.isEmpty()) { finish(error); return; }
m_queue = entries;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "file batch: local scan complete entries=%d", int(m_queue.size()));
next();
}, Qt::QueuedConnection);
}));
}
// destination — каталог назначения либо родитель временного каталога экспорта.
void MemberFileBatch::start(const QList<MemberFileEntry>& entries, const QString& destination, bool temporary) {
m_queue = entries;
if (m_mode == Remove) { next(); return; }
m_work = true;
m_pool.start(QRunnable::create([this, destination, temporary] {
QString root = destination, error;
if (temporary) {
if (!QDir().mkpath(root)) error = text("Cannot create directory: %1").arg(root);
else {
QTemporaryDir dir(root + "/XXXXXX");
if (!dir.isValid()) error = text("Cannot create directory: %1").arg(root);
else { root = dir.path(); dir.setAutoRemove(false); }
}
}
QMetaObject::invokeMethod(this, [this, root, error, temporary] {
m_work = false; if (temporary && error.isEmpty()) m_temporary = root;
if (m_cancelled || !error.isEmpty()) { finish(error); return; }
for (auto& entry : m_queue) {
if (entry.local.isEmpty()) entry.local = QDir(root).filePath(QFileInfo(entry.remote).fileName());
m_roots.append(entry.local); m_seen.insert(entry.remote);
}
next();
}, Qt::QueuedConnection);
}));
}
void MemberFileBatch::request(uint8_t command) {
auto* q = (mf_gui_request*)u_calloc(1, sizeof(mf_gui_request));
if (!q || !gui_bridge_is_uasync_ready()) {
u_free(q); finish(text("Unable to request directory")); return;
}
const auto& entry = m_queue.first();
q->inst = gui_bridge_get_inst(); q->group = m_group; q->server = m_server;
auto& r = q->message; r.kind = MF_REQUEST; r.command = command; r.owner = m_owner; r.request_id = nextId();
memcpy(r.path, entry.remote.toUtf8().constData(), entry.remote.toUtf8().size());
if ((m_mode == Download || command == MF_REMOVE_PATH) && entry.version.size() == 16) memcpy(r.version, entry.version.constData(), 16);
r.cursor = command == MF_LIST ? entry.cursor : 0; m_command = command; m_request = r.request_id;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "file batch: request id=%llu command=%s path=%s cursor=%llu",
(unsigned long long)m_request, mf_command_text(command), r.path, (unsigned long long)r.cursor);
gui_bridge_post_uasync_fn(member_files_gui_request, q);
}
void MemberFileBatch::next() {
if (m_finished || m_work || m_request || m_transfer) return;
if (m_cancelled) { finish(); return; }
if (m_queue.isEmpty() && !m_verify.isEmpty()) { m_queue = m_verify; m_verify.clear(); }
if (m_queue.isEmpty()) { finish(); return; }
const auto entry = m_queue.first();
if (!pathBytes(entry.remote, true) || (m_mode != Remove && !pathBytes(entry.local, false))) {
finish(text("Cannot copy this path: %1").arg(entry.remote)); return;
}
progress(text("Processed %1, skipped %2, remaining %3: %4").arg(m_done).arg(m_skipped).arg(m_queue.size()).arg(entry.remote), -1);
if (m_mode != Download || entry.verify) { request(MF_STAT); return; }
if (entry.directory && entry.cursor) { request(MF_LIST); return; }
m_work = true;
m_pool.start(QRunnable::create([this, entry] {
QFileInfo info(entry.local); int exists = info.exists() || info.isSymLink();
bool same = !info.isSymLink() && (entry.directory ? info.isDir() : info.isFile());
QString error;
if (entry.directory && !exists && !QDir().mkpath(entry.local)) error = text("Cannot create directory: %1").arg(entry.local);
QMetaObject::invokeMethod(this, [this, entry, exists, same, error] {
m_work = false;
if (m_cancelled || !error.isEmpty()) { finish(error); return; }
if (exists && (!entry.directory || !same) && !replace(same)) return;
if (entry.directory) request(MF_LIST); else copyFile(entry.version);
}, Qt::QueuedConnection);
}));
}
bool MemberFileBatch::replace(bool sameType) {
QPointer<MemberFileBatch> guard(this);
int answer = sameType && m_policy ? m_policy : conflict(m_queue.first().remote, sameType);
if (!guard || m_finished) return false;
if (m_cancelled || answer == QMessageBox::Cancel) { cancel(); return false; }
if (sameType && (answer == QMessageBox::YesToAll || answer == QMessageBox::NoToAll)) m_policy = answer;
if (sameType && (answer == QMessageBox::Yes || answer == QMessageBox::YesToAll)) return true;
// Пропуск папки пропускает и всё подготовленное локальное поддерево.
QString prefix = m_queue.first().remote + '/';
for (int i = m_queue.size() - 1; i > 0; --i) if (m_queue[i].remote.startsWith(prefix)) { m_queue.removeAt(i); m_skipped++; }
advance(true); return false;
}
bool MemberFileBatch::response(const mf_message& r) {
if (!m_request || r.request_id != m_request) return false;
m_request = 0;
auto& entry = m_queue.first();
if (r.command != m_command || r.owner != m_owner || QString::fromUtf8(r.path) != entry.remote) {
finish(text("Invalid directory response")); return true;
}
if (m_cancelled) { finish(); return true; }
if (m_mode == Upload && r.command == MF_STAT && (r.status == MF_OK || r.status == MF_NOT_FOUND)) {
if (r.status == MF_OK) {
bool same = r.type == (entry.directory ? MF_DIRECTORY : MF_FILE);
if (entry.directory && same) { advance(); return true; }
if (!replace(same)) return true;
}
if (entry.directory) request(MF_MKDIR);
else copyFile(r.status == MF_NOT_FOUND ? QByteArray(16, '\0') : QByteArray((const char*)r.version, 16));
return true;
}
if (r.status) { finish(QString::fromUtf8(mf_status_text(r.status)) + ": " + entry.remote); return true; }
if (m_mode == Remove && r.command == MF_STAT) {
if (r.type != (entry.directory ? MF_DIRECTORY : MF_FILE) ||
(!entry.directory && entry.version != QByteArray((const char*)r.version, 16))) {
finish(text("Source changed: %1").arg(entry.remote)); return true;
}
entry.version = QByteArray((const char*)r.version, 16); request(MF_REMOVE_PATH); return true;
}
if (r.command != MF_LIST) { advance(); return true; }
if (r.type != MF_DIRECTORY || mf_zero(r.version, 16)) { finish(text("Invalid directory response")); return true; }
QList<MemberFileEntry> children;
for (unsigned at = 0; at < r.data_len;) {
const uint8_t* p = r.data + at;
if (r.data_len - at < MF_ENTRY_HEADER) { finish(text("Invalid directory response")); return true; }
unsigned n = p[0] | (unsigned)p[1] << 8;
if (!n || n > MF_NAME_MAX || n > r.data_len - at - MF_ENTRY_HEADER || (p[2] != MF_FILE && p[2] != MF_DIRECTORY)) {
finish(text("Invalid directory response")); return true;
}
QByteArray bytes((const char*)p + MF_ENTRY_HEADER, int(n)); QString name = QString::fromUtf8(bytes);
QString remote = entry.remote + '/' + name;
if (name.contains('/') || name.toUtf8() != bytes || !pathBytes(remote, true) || m_seen.contains(remote) || m_seen.size() >= 100000) {
finish(text("Invalid directory response")); return true;
}
m_seen.insert(remote);
children.append({remote, entry.local + '/' + name, QByteArray((const char*)p + 19, 16), p[2] == MF_DIRECTORY});
at += MF_ENTRY_HEADER + n;
}
if (r.cursor != entry.cursor + quint64(children.size()) || (r.more && children.isEmpty())) {
finish(text("Invalid directory response")); return true;
}
entry.version = QByteArray((const char*)r.version, 16); entry.cursor = r.cursor;
if (!r.more) { auto verify = entry; verify.verify = true; m_verify.append(verify); }
m_queue.append(children);
if (r.more) request(MF_LIST); else advance();
return true;
}
void MemberFileBatch::copyFile(const QByteArray& version) {
auto* q = (mf_transfer_request*)u_calloc(1, sizeof(mf_transfer_request));
if (!q || !gui_bridge_is_uasync_ready()) { u_free(q); finish(text("Unable to start transfer")); return; }
const auto& e = m_queue.first();
q->inst = gui_bridge_get_inst(); q->id = nextId(); q->group = m_group; q->server = m_server; q->owner = m_owner;
q->upload = m_mode == Upload;
memcpy(q->remote, e.remote.toUtf8().constData(), e.remote.toUtf8().size());
memcpy(q->local, e.local.toUtf8().constData(), e.local.toUtf8().size());
if (version.size() == 16) memcpy(q->version, version.constData(), 16);
m_transfer = q->id; gui_bridge_post_uasync_fn(member_files_gui_transfer, q);
}
bool MemberFileBatch::transfer(const uint8_t* data, int length) {
if (length != 51 || !m_transfer || mf_get64(data) != m_transfer) return false;
quint64 done = mf_get64(data + 32), total = mf_get64(data + 40);
if (data[50] != 2) {
progress(text("Processed %1, skipped %2, remaining %3: %4").arg(m_done).arg(m_skipped).arg(m_queue.size())
.arg(m_queue.first().remote) + text(" · %1 / %2 bytes").arg(done).arg(total), total ? int((long double)done * 1000 / total) : -1);
return true;
}
m_transfer = 0; unsigned status = data[48] | (unsigned)data[49] << 8;
if (m_cancelled) finish();
else if (status) finish(QString::fromUtf8(mf_status_text(status)) + ": " + m_queue.first().remote);
else advance();
return true;
}
void MemberFileBatch::advance(bool skipped) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "file batch: entry completed path=%s skipped=%d", qPrintable(m_queue.first().remote), skipped);
if (!m_queue.first().verify) { if (skipped) m_skipped++; else m_done++; }
m_queue.removeFirst();
QMetaObject::invokeMethod(this, [this] { next(); }, Qt::QueuedConnection);
}
void MemberFileBatch::cancel() {
if (m_finished) return;
m_cancelled = true; cancelRequest(m_request); m_request = 0; cancelRequest(m_transfer);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "file batch: cancelling transfer=%llu worker=%d", (unsigned long long)m_transfer, m_work);
if (!m_transfer && !m_work) finish();
}
void MemberFileBatch::finish(const QString& error) {
if (m_finished) return;
m_finished = true;
bool ok = error.isEmpty() && !m_cancelled;
QString summary = text("Completed: %1, skipped: %2").arg(m_done).arg(m_skipped);
if (!ok) summary = (m_cancelled ? text("Copy cancelled") : error) + " — " + summary;
if (ok) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "file batch: %s", qPrintable(summary));
else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "file batch: %s", qPrintable(summary));
if (!ok && !m_temporary.isEmpty()) {
QString root = m_temporary;
m_pool.start(QRunnable::create([this, root, summary] {
if (!QDir(root).removeRecursively()) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "file batch: cache cleanup failed path=%s", qPrintable(root));
QMetaObject::invokeMethod(this, [this, summary] { finished(false, summary, {}); }, Qt::QueuedConnection);
}));
} else finished(ok, summary, ok ? m_roots : QStringList());
}

57
tools/chatgui/src/member_files/batch.h
Unescape View File

@ -0,0 +1,57 @@
/* Последовательные операции над выбранными файлами/деревьями. Контекст неизменен;
* локальный обход выполняется в worker, сеть — через bridge. Отмена ждёт завершения
* текущего файлового I/O до finished. Временный каталог экспорта удаляется при ошибке. */
#pragma once
#include <QObject>
#include <QList>
#include <QSet>
#include <QThreadPool>
#include <QStringList>
#include <QByteArray>
#include <atomic>
#include <functional>
#include <cstdint>
struct mf_message;
struct MemberFileEntry {
QString remote, local;
QByteArray version;
bool directory = false, verify = false;
quint64 cursor = 0;
};
class MemberFileBatch : public QObject {
public:
enum Mode { Upload, Download, Remove };
MemberFileBatch(quint64 group, quint64 server, quint64 owner, Mode mode, QObject* parent);
~MemberFileBatch() override;
void upload(const QStringList& sources, const QString& destination);
void start(const QList<MemberFileEntry>& entries, const QString& destination = {}, bool temporary = false);
bool response(const mf_message& message);
bool transfer(const uint8_t* data, int length);
void cancel();
std::function<void(const QString&, int)> progress;
std::function<void(bool, const QString&, const QStringList&)> finished;
// Возвращает QMessageBox::Yes/No/YesToAll/NoToAll/Cancel; конфликт разных типов нельзя заменить.
std::function<int(const QString&, bool)> conflict;
private:
void next();
void request(uint8_t command);
void copyFile(const QByteArray& version);
void advance(bool skipped = false);
void finish(const QString& error = {});
bool replace(bool sameType);
void cancelRequest(quint64 id);
const quint64 m_group, m_server, m_owner;
const Mode m_mode;
QList<MemberFileEntry> m_queue, m_verify;
QSet<QString> m_seen;
QStringList m_roots;
QString m_temporary, m_error;
QThreadPool m_pool;
std::atomic_bool m_cancelled{false};
bool m_work = false, m_finished = false;
quint64 m_request = 0, m_transfer = 0;
uint8_t m_command = 0;
int m_done = 0, m_skipped = 0, m_policy = 0;
};
Write Preview
Loading…
Cancel
Save