DragonSpirit/NXST
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

finish refactor, add docs and CI
CI / Build NRO (push) Failing after 3s
Details
CI / Format check (push) Successful in 14s
Details
CI / Layering check (push) Successful in 1s
Details

Browse Source
This commit is contained in:
Nikolai Fedorov
2026-04-27 01:49:41 +03:00
parent dc65a4c8a9
commit 31b7a4401e
47 changed files with 1964 additions and 1470 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/transfer_service.cpp
+128 -71
View File
@@ -1,5 +1,3 @@
#include <nxst/service/transfer_service.hpp>
#include <arpa/inet.h>
#include <chrono>
#include <cstring>
@@ -12,11 +10,14 @@
#include <unistd.h>
#include <vector>
#include <nxst/service/transfer_service.hpp>
#ifdef __SWITCH__
#include <switch.h>
#include <nxst/infra/fs/io.hpp>
#include <nxst/domain/util.hpp>
#include <nxst/domain/account.hpp>
#include <nxst/domain/util.hpp>
#include <nxst/infra/fs/io.hpp>
#endif
#include <nxst/domain/protocol.hpp>
@@ -32,7 +33,8 @@ static bool sendAll(int sock, const void* buf, size_t len) {
size_t sent = 0;
while (sent < len) {
ssize_t n = send(sock, static_cast<const char*>(buf) + sent, len - sent, 0);
if (n <= 0) return false;
if (n <= 0)
return false;
sent += n;
}
return true;
@@ -42,7 +44,8 @@ static bool recvAll(int sock, void* buf, size_t len) {
size_t got = 0;
while (got < len) {
ssize_t n = read(sock, static_cast<char*>(buf) + got, len - got);
if (n <= 0) return false;
if (n <= 0)
return false;
got += n;
}
return true;
@@ -50,15 +53,19 @@ static bool recvAll(int sock, void* buf, size_t len) {
static bool sendFile(int sock, const fs::path& filepath, TransferState& state) {
std::ifstream infile(filepath, std::ios::binary | std::ios::ate);
if (!infile.is_open()) return false;
if (!infile.is_open())
return false;
uint32_t filename_len = (uint32_t)filepath.string().size();
uint64_t file_size = (uint64_t)infile.tellg();
uint64_t file_size = (uint64_t)infile.tellg();
infile.seekg(0, std::ios::beg);
if (!sendAll(sock, &filename_len, sizeof(filename_len))) return false;
if (!sendAll(sock, filepath.c_str(), filename_len)) return false;
if (!sendAll(sock, &file_size, sizeof(file_size))) return false;
if (!sendAll(sock, &filename_len, sizeof(filename_len)))
return false;
if (!sendAll(sock, filepath.c_str(), filename_len))
return false;
if (!sendAll(sock, &file_size, sizeof(file_size)))
return false;
std::vector<char> buffer(proto::BUF_SIZE);
uint64_t remaining = file_size;
@@ -66,8 +73,10 @@ static bool sendFile(int sock, const fs::path& filepath, TransferState& state) {
size_t to_read = (size_t)std::min(remaining, (uint64_t)proto::BUF_SIZE);
infile.read(buffer.data(), (std::streamsize)to_read);
std::streamsize count = infile.gcount();
if (count <= 0) break;
if (!sendAll(sock, buffer.data(), (size_t)count)) return false;
if (count <= 0)
break;
if (!sendAll(sock, buffer.data(), (size_t)count))
return false;
state.bytes_done.fetch_add((uint64_t)count);
remaining -= (uint64_t)count;
}
@@ -84,12 +93,12 @@ static void mkdirs(const std::string& path) {
mkdir(path.c_str(), 0777);
}
static void receiveFile(int sock, const std::string& rel_path, uint64_t file_size,
TransferState& state) {
static void receiveFile(int sock, const std::string& rel_path, uint64_t file_size, TransferState& state) {
size_t last_slash = rel_path.rfind('/');
if (last_slash != std::string::npos) {
std::string dir = rel_path.substr(0, last_slash);
if (!dir.empty()) mkdirs(dir);
if (!dir.empty())
mkdirs(dir);
}
FILE* outfile = fopen(rel_path.c_str(), "wb");
@@ -99,7 +108,8 @@ static void receiveFile(int sock, const std::string& rel_path, uint64_t file_siz
while (remaining > 0) {
size_t to_read = (size_t)std::min(remaining, (uint64_t)proto::BUF_SIZE);
ssize_t n = read(sock, drain.data(), to_read);
if (n <= 0) break;
if (n <= 0)
break;
remaining -= (uint64_t)n;
}
return;
@@ -113,7 +123,8 @@ static void receiveFile(int sock, const std::string& rel_path, uint64_t file_siz
while (total < file_size) {
size_t to_read = (size_t)std::min(file_size - total, (uint64_t)proto::BUF_SIZE);
ssize_t n = read(sock, buffer.data(), to_read);
if (n <= 0) break;
if (n <= 0)
break;
fwrite(buffer.data(), 1, (size_t)n, outfile);
total += (uint64_t)n;
state.bytes_done.store(total);
@@ -131,17 +142,19 @@ void TransferService::failSend(const std::string& reason) {
int TransferService::findServer(char* out_ip) {
int udp_fd = socket(AF_INET, SOCK_DGRAM, 0);
if (udp_fd < 0) return -1;
if (udp_fd < 0)
return -1;
sender_udp_sock.store(udp_fd);
auto releaseUdp = [&]() {
int owned = sender_udp_sock.exchange(-1);
if (owned == udp_fd) close(udp_fd);
if (owned == udp_fd)
close(udp_fd);
};
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_port = htons(proto::MULTICAST_PORT);
addr.sin_family = AF_INET;
addr.sin_port = htons(proto::MULTICAST_PORT);
addr.sin_addr.s_addr = inet_addr(proto::MULTICAST_GROUP);
if (sendto(udp_fd, "DISCOVER_SERVER", 15, 0, (sockaddr*)&addr, sizeof(addr)) < 0) {
@@ -152,16 +165,19 @@ int TransferService::findServer(char* out_ip) {
// Poll in 100ms slices so cancel races within 100ms
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
while (std::chrono::steady_clock::now() < deadline) {
if (sender_state.cancelled.load()) { releaseUdp(); return -1; }
if (sender_state.cancelled.load()) {
releaseUdp();
return -1;
}
struct timeval tv{0, 100000};
fd_set fds;
FD_ZERO(&fds);
FD_SET(udp_fd, &fds);
if (select(udp_fd + 1, &fds, nullptr, nullptr, &tv) > 0) {
sockaddr_in from{};
socklen_t fromlen = sizeof(from);
char buf[256];
ssize_t n = recvfrom(udp_fd, buf, sizeof(buf) - 1, 0, (sockaddr*)&from, &fromlen);
socklen_t fromlen = sizeof(from);
char buf[256];
ssize_t n = recvfrom(udp_fd, buf, sizeof(buf) - 1, 0, (sockaddr*)&from, &fromlen);
if (n > 0) {
buf[n] = '\0';
if (strcmp(buf, "SERVER_HERE") == 0) {
@@ -179,8 +195,8 @@ int TransferService::findServer(char* out_ip) {
void* TransferService::senderEntry(void* arg) {
auto* a = static_cast<SenderArgs*>(arg);
TransferService* svc = a->svc;
size_t idx = a->title_index;
AccountUid uid = a->uid;
size_t idx = a->title_index;
AccountUid uid = a->uid;
delete a;
svc->runSender(idx, uid);
return nullptr;
@@ -200,7 +216,8 @@ void TransferService::runSender(size_t title_index, AccountUid uid) {
failSend("No receiver found.\nMake sure the other Switch is in Receive mode.");
return finish();
}
if (sender_state.cancelled.load()) return finish();
if (sender_state.cancelled.load())
return finish();
sender_state.setStatus("Creating backup...");
#ifdef __SWITCH__
@@ -212,24 +229,30 @@ void TransferService::runSender(size_t title_index, AccountUid uid) {
fs::path directory = backup_result.value();
#else
fs::path directory = ".";
(void)title_index; (void)uid;
(void)title_index;
(void)uid;
#endif
if (sender_state.cancelled.load()) return finish();
if (sender_state.cancelled.load())
return finish();
sender_state.setStatus("Connecting...");
int tcp_fd = socket(AF_INET, SOCK_STREAM, 0);
if (tcp_fd < 0) { failSend("Failed to open socket."); return finish(); }
if (tcp_fd < 0) {
failSend("Failed to open socket.");
return finish();
}
sender_tcp_sock.store(tcp_fd);
auto releaseTcp = [&]() {
int owned = sender_tcp_sock.exchange(-1);
if (owned == tcp_fd) close(tcp_fd);
if (owned == tcp_fd)
close(tcp_fd);
};
sockaddr_in serv{};
serv.sin_family = AF_INET;
serv.sin_port = htons(proto::TCP_PORT);
serv.sin_port = htons(proto::TCP_PORT);
if (inet_pton(AF_INET, server_ip, &serv.sin_addr) <= 0 ||
connect(tcp_fd, (sockaddr*)&serv, sizeof(serv)) < 0) {
if (!sender_state.cancelled.load())
@@ -245,11 +268,13 @@ void TransferService::runSender(size_t title_index, AccountUid uid) {
sender_state.bytes_total.store(total);
for (const auto& entry : fs::recursive_directory_iterator(directory)) {
if (sender_state.cancelled.load()) break;
if (sender_state.cancelled.load())
break;
const fs::path& p = entry.path();
if (fs::is_regular_file(p)) {
sender_state.setStatus(p.filename().string());
if (!sendFile(tcp_fd, p, sender_state)) break;
if (!sendFile(tcp_fd, p, sender_state))
break;
}
}
@@ -278,19 +303,26 @@ int TransferService::startSend(size_t title_index, AccountUid uid) {
void TransferService::cancelSend() {
sender_state.cancelled.store(true);
int udp = sender_udp_sock.exchange(-1);
if (udp >= 0) { shutdown(udp, SHUT_RDWR); close(udp); }
if (udp >= 0) {
shutdown(udp, SHUT_RDWR);
close(udp);
}
int tcp = sender_tcp_sock.exchange(-1);
if (tcp >= 0) { shutdown(tcp, SHUT_RDWR); close(tcp); }
if (tcp >= 0) {
shutdown(tcp, SHUT_RDWR);
close(tcp);
}
}
// ─── Receiver ────────────────────────────────────────────────────────────────
std::string TransferService::replaceUsername(const std::string& file_path) const {
#ifdef __SWITCH__
std::string username = StringUtils::removeNotAscii(
StringUtils::removeAccents(Account::username(restore_uid)));
std::string username =
StringUtils::removeNotAscii(StringUtils::removeAccents(Account::username(restore_uid)));
size_t last_slash = file_path.rfind('/');
if (last_slash == std::string::npos) return file_path;
if (last_slash == std::string::npos)
return file_path;
size_t prev_slash = file_path.rfind('/', last_slash - 1);
if (prev_slash == std::string::npos)
return username + file_path.substr(last_slash);
@@ -311,21 +343,25 @@ void TransferService::runBroadcast() {
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, nullptr);
int udp = socket(AF_INET, SOCK_DGRAM, 0);
if (udp < 0) { receiver_broadcast_active.store(false); return; }
if (udp < 0) {
receiver_broadcast_active.store(false);
return;
}
receiver_bcast_sock.store(udp);
auto releaseUdp = [&]() {
int owned = receiver_bcast_sock.exchange(-1);
if (owned == udp) close(udp);
if (owned == udp)
close(udp);
};
struct timeval tv{0, 20000}; // 20ms poll so cancel/exit wins race with socketExit
setsockopt(udp, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_ANY);
addr.sin_port = htons(proto::MULTICAST_PORT);
addr.sin_port = htons(proto::MULTICAST_PORT);
if (bind(udp, (sockaddr*)&addr, sizeof(addr)) < 0) {
releaseUdp();
@@ -348,7 +384,8 @@ void TransferService::runBroadcast() {
while (true) {
ssize_t n = recvfrom(udp, buf, sizeof(buf) - 1, 0, (sockaddr*)&from, &fromlen);
if (n < 0) {
if (receiver_state.cancelled.load()) break;
if (receiver_state.cancelled.load())
break;
continue;
}
buf[n] = '\0';
@@ -365,7 +402,7 @@ void TransferService::runBroadcast() {
void* TransferService::acceptEntry(void* arg) {
auto* a = static_cast<AcceptArgs*>(arg);
TransferService* svc = a->svc;
int server_fd = a->server_fd;
int server_fd = a->server_fd;
delete a;
svc->runAccept(server_fd);
return nullptr;
@@ -380,41 +417,48 @@ void TransferService::runAccept(int server_fd) {
int client_sock = accept(server_fd, (sockaddr*)&client_addr, &client_len);
int owned_listen = receiver_listen_sock.exchange(-1);
if (owned_listen == server_fd) close(server_fd);
if (owned_listen == server_fd)
close(server_fd);
if (client_sock >= 0) {
receiver_client_sock.store(client_sock);
while (true) {
uint32_t filename_len = 0;
if (!recvAll(client_sock, &filename_len, sizeof(filename_len))) break;
if (filename_len == proto::EOF_SENTINEL) break;
if (filename_len > proto::MAX_FILENAME) break;
if (!recvAll(client_sock, &filename_len, sizeof(filename_len)))
break;
if (filename_len == proto::EOF_SENTINEL)
break;
if (filename_len > proto::MAX_FILENAME)
break;
std::vector<char> filename_buf(filename_len + 1, '\0');
if (!recvAll(client_sock, filename_buf.data(), filename_len)) break;
if (!recvAll(client_sock, filename_buf.data(), filename_len))
break;
std::string filename_str(filename_buf.data(), filename_len);
filename_str = replaceUsername(filename_str);
{
size_t sl = filename_str.rfind('/');
receiver_state.setStatus(
sl != std::string::npos ? filename_str.substr(sl + 1) : filename_str);
receiver_state.setStatus(sl != std::string::npos ? filename_str.substr(sl + 1)
: filename_str);
}
uint64_t file_size = 0;
if (!recvAll(client_sock, &file_size, sizeof(file_size))) break;
if (!recvAll(client_sock, &file_size, sizeof(file_size)))
break;
receiveFile(client_sock, filename_str, file_size, receiver_state);
}
int owned = receiver_client_sock.exchange(-1);
if (owned == client_sock) close(client_sock);
if (owned == client_sock)
close(client_sock);
if (!receiver_state.cancelled.load()) {
#ifdef __SWITCH__
receiver_state.setStatus("Restoring...");
auto result = io::restore(restore_title_index, restore_uid, 0, restore_title_name);
restore_ok = result.isOk();
restore_ok = result.isOk();
restore_error = result.isOk() ? "" : result.error();
#else
restore_ok = true;
@@ -430,29 +474,32 @@ int TransferService::startReceive(size_t title_index, AccountUid uid, std::strin
receiver_state.reset();
receiver_state.setStatus("Waiting for connection...");
restore_title_index = title_index;
restore_uid = uid;
restore_title_name = std::move(title_name);
restore_ok = false;
restore_uid = uid;
restore_title_name = std::move(title_name);
restore_ok = false;
restore_error.clear();
pthread_t bcast_thread;
if (pthread_create(&bcast_thread, nullptr, broadcastEntry, this) != 0) return 1;
if (pthread_create(&bcast_thread, nullptr, broadcastEntry, this) != 0)
return 1;
receiver_bcast_thread = bcast_thread;
pthread_detach(bcast_thread);
Socket server(socket(AF_INET, SOCK_STREAM, 0));
if (!server.valid()) { cancelReceive(); return 1; }
if (!server.valid()) {
cancelReceive();
return 1;
}
int yes = 1;
setsockopt(server, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(proto::TCP_PORT);
addr.sin_port = htons(proto::TCP_PORT);
if (bind(server, (sockaddr*)&addr, sizeof(addr)) < 0 ||
listen(server, 3) < 0) {
if (bind(server, (sockaddr*)&addr, sizeof(addr)) < 0 || listen(server, 3) < 0) {
cancelReceive();
return 1;
}
@@ -472,12 +519,22 @@ int TransferService::startReceive(size_t title_index, AccountUid uid, std::strin
void TransferService::cancelReceive() {
receiver_state.cancelled.store(true);
int sock = receiver_client_sock.exchange(-1);
if (sock >= 0) { shutdown(sock, SHUT_RDWR); close(sock); }
if (sock >= 0) {
shutdown(sock, SHUT_RDWR);
close(sock);
}
int lsock = receiver_listen_sock.exchange(-1);
if (lsock >= 0) { shutdown(lsock, SHUT_RDWR); close(lsock); }
if (lsock >= 0) {
shutdown(lsock, SHUT_RDWR);
close(lsock);
}
int bsock = receiver_bcast_sock.exchange(-1);
if (bsock >= 0) { shutdown(bsock, SHUT_RDWR); close(bsock); }
if (receiver_broadcast_active.load()) pthread_cancel(receiver_bcast_thread);
if (bsock >= 0) {
shutdown(bsock, SHUT_RDWR);
close(bsock);
}
if (receiver_broadcast_active.load())
pthread_cancel(receiver_bcast_thread);
}
} // namespace nxst
} // namespace nxst