diff --git a/CMakeLists.txt b/CMakeLists.txt index 9e24f15..3235870 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -22,6 +22,8 @@ option(WITH_INDEXED_HASHES "Save memory used for storing transaction hashes and option(WITH_MERGE_MINING_DONATION "Merge mine donations to the author. This doesn't affect your hashrate or payouts in any way - only unused merge mining capacity will be utilised. If you merge mine yourself, your settings will take priority." ON) +option(WITH_REDIS "Use Redis for storage instead of files. Requires libhiredis-dev." OFF) + option(DEV_TEST_SYNC "[Developer only] Sync test, stop p2pool after sync is complete" OFF) option(DEV_WITH_TSAN "[Developer only] Compile with thread sanitizer" OFF) option(DEV_WITH_MSAN "[Developer only] Compile with memory sanitizer" OFF) @@ -97,6 +99,35 @@ if (DEV_DEBUG) add_definitions(-DDEV_DEBUG) endif() +if (WITH_REDIS) + message(STATUS "Redis storage enabled") + add_definitions(-DWITH_REDIS) + + find_path(HIREDIS_INCLUDE_DIR NAMES hiredis/hiredis.h PATH_SUFFIXES "include") + + # Prefer static library for portable binary + find_library(HIREDIS_LIBRARY_STATIC NAMES libhiredis.a PATHS /usr/lib /usr/lib/x86_64-linux-gnu /usr/local/lib) + find_library(HIREDIS_LIBRARY_DYNAMIC NAMES hiredis) + + if (HIREDIS_LIBRARY_STATIC) + set(HIREDIS_LIBRARY ${HIREDIS_LIBRARY_STATIC}) + message(STATUS " Using static hiredis library") + elseif (HIREDIS_LIBRARY_DYNAMIC) + set(HIREDIS_LIBRARY ${HIREDIS_LIBRARY_DYNAMIC}) + message(STATUS " Using dynamic hiredis library (static not found)") + endif() + + if (NOT HIREDIS_INCLUDE_DIR OR NOT HIREDIS_LIBRARY) + message(FATAL_ERROR "hiredis library not found. Install with: apt install libhiredis-dev") + endif() + + message(STATUS " hiredis include: ${HIREDIS_INCLUDE_DIR}") + message(STATUS " hiredis library: ${HIREDIS_LIBRARY}") + + include_directories(${HIREDIS_INCLUDE_DIR}) + set(LIBS ${LIBS} ${HIREDIS_LIBRARY}) +endif() + add_subdirectory(external/src/mx25519) set(LIBS ${LIBS} mx25519) @@ -204,6 +235,11 @@ if (WITH_INDEXED_HASHES) set(SOURCES ${SOURCES} src/indexed_hash.cpp) endif() +if (WITH_REDIS) + set(HEADERS ${HEADERS} src/redis_storage.h) + set(SOURCES ${SOURCES} src/redis_storage.cpp) +endif() + source_group(TREE "${CMAKE_CURRENT_SOURCE_DIR}" PREFIX "Header Files" FILES ${HEADERS}) source_group(TREE "${CMAKE_CURRENT_SOURCE_DIR}" PREFIX "Source Files" FILES ${SOURCES}) diff --git a/src/block_cache.cpp b/src/block_cache.cpp index 9a680a6..49f7f9d 100644 --- a/src/block_cache.cpp +++ b/src/block_cache.cpp @@ -20,7 +20,12 @@ #include "pool_block.h" #include "p2p_server.h" #include "side_chain.h" + +#ifdef WITH_REDIS +#include "redis_storage.h" +#else #include +#endif LOG_CATEGORY(BlockCache) @@ -32,6 +37,145 @@ static const uint64_t CACHE_VERSION = std::hash{}(__DATE__ __TIME__ namespace p2pool { +#ifdef WITH_REDIS + +// Redis-based block cache implementation +struct BlockCache::Impl : public nocopy_nomove +{ + Impl() : m_data(nullptr) {} + ~Impl() {} + void flush() {} // Redis handles persistence + uint8_t* m_data; // Not used in Redis mode +}; + +BlockCache::BlockCache() + : m_impl(new Impl()) + , m_flushRunning(0) + , m_storeIndex(0) + , m_loadingStarted(0) +{ +} + +BlockCache::~BlockCache() +{ + delete m_impl; +} + +void BlockCache::store(const PoolBlock& block) +{ + const std::vector mainchain_data = block.serialize_mainchain_data(); + const std::vector sidechain_data = block.serialize_sidechain_data(); + + const size_t n1 = mainchain_data.size(); + const size_t n2 = sidechain_data.size(); + + if (n1 + n2 + sizeof(uint32_t) + HASH_SIZE > BLOCK_SIZE) { + LOGERR(3, "Block too large to cache: " << (n1 + n2) << " bytes"); + return; + } + + // Create buffer with: sidechain_id (32 bytes) + size prefix (4 bytes) + data + // This allows Go observer to read the sidechain ID directly without recalculating + std::vector buffer; + buffer.reserve(HASH_SIZE + sizeof(uint32_t) + n1 + n2); + + // Prepend sidechain ID (32 bytes) - this is the key for parent chain walking + buffer.insert(buffer.end(), block.m_sidechainId.h, block.m_sidechainId.h + HASH_SIZE); + + // Then the original format: size prefix + data + uint32_t total_size = static_cast(n1 + n2); + buffer.insert(buffer.end(), + reinterpret_cast(&total_size), + reinterpret_cast(&total_size) + sizeof(uint32_t)); + buffer.insert(buffer.end(), mainchain_data.begin(), mainchain_data.end()); + buffer.insert(buffer.end(), sidechain_data.begin(), sidechain_data.end()); + + // Store in Redis hash with circular index as field + uint32_t index = (m_storeIndex++) % NUM_BLOCKS; + std::string field = std::to_string(index); + + RedisStorage& redis = get_redis_storage(); + if (!redis.hset("cache", field, buffer.data(), buffer.size())) { + LOGERR(3, "Failed to store block in Redis cache, index=" << index); + } +} + +void BlockCache::load_all(SideChain& side_chain, P2PServer& server) +{ + RedisStorage& redis = get_redis_storage(); + + // Check cache version + std::string version_str; + bool version_ok = false; + if (redis.get("cache:version", version_str)) { + try { + uint64_t v = std::stoull(version_str); + if (v == CACHE_VERSION) { + version_ok = true; + } + } catch (...) {} + } + + if (!version_ok) { + LOGINFO(1, "Cache version mismatch (recompiled binary), clearing Redis cache"); + for (uint32_t i = 0; i < NUM_BLOCKS; ++i) { + redis.hdel("cache", std::to_string(i)); + } + redis.set("cache:version", std::to_string(CACHE_VERSION)); + } + + if (m_loadingStarted.exchange(1)) { + return; + } + + LOGINFO(1, "Loading cached blocks from Redis..."); + + std::vector>> cache_entries; + if (!redis.hgetall("cache", cache_entries)) { + LOGINFO(1, "No cache entries found in Redis"); + return; + } + + PoolBlock block; + uint32_t blocks_loaded = 0; + + for (const auto& entry : cache_entries) { + const std::vector& data = entry.second; + + // New format: 32-byte sidechain_id + 4-byte length + block data + if (data.size() < HASH_SIZE + sizeof(uint32_t)) { + continue; + } + + // Skip the sidechain_id (32 bytes) - it's only needed by Go observer + const uint8_t* block_data = data.data() + HASH_SIZE; + const size_t block_data_size = data.size() - HASH_SIZE; + + const uint32_t n = *reinterpret_cast(block_data); + if (n == 0 || n + sizeof(uint32_t) > block_data_size) { + continue; + } + + if (block.deserialize(block_data + sizeof(uint32_t), n, side_chain, uv_default_loop_checked(), false) == 0) { + server.add_cached_block(block); + ++blocks_loaded; + } + } + + if (blocks_loaded > 0) { + LOGINFO(1, "Loaded " << blocks_loaded << " blocks from Redis cache"); + } else { + LOGINFO(1, "Redis cache empty, will sync from peers"); + } +} + +void BlockCache::flush() +{ + // Redis handles persistence automatically +} + +#else // !WITH_REDIS - Original file-based implementation + struct BlockCache::Impl : public nocopy_nomove { #if defined(__linux__) || defined(__unix__) || defined(_POSIX_VERSION) || defined(__MACH__) @@ -255,4 +399,6 @@ void BlockCache::flush() } } +#endif // WITH_REDIS + } // namespace p2pool diff --git a/src/main.cpp b/src/main.cpp index 66c8e1b..4e07cc9 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -108,6 +108,11 @@ void p2pool_usage() "--full-validation Enables full share validation / increases CPU usage\n" "--onion-address Tell other peers to use this .onion address to connect to this node through TOR\n" "--no-clearnet-p2p Forces P2P server to listen on 127.0.0.1 and to not connect to clearnet IPs\n" +#ifdef WITH_REDIS + "--redis-host Redis server host, default is 127.0.0.1\n" + "--redis-port Redis server port, default is 6379\n" + "--redis-db Redis database number, default is 0\n" +#endif "--help Show this help message\n\n" "Example command line:\n\n" "%s-salvium --host 127.0.0.1 --rpc-port 19081 --zmq-port 19083 --wallet YOUR_WALLET_ADDRESS --stratum 0.0.0.0:%d --p2p 0.0.0.0:%d\n\n", diff --git a/src/p2p_server.cpp b/src/p2p_server.cpp index cb22f8b..ebc8b87 100644 --- a/src/p2p_server.cpp +++ b/src/p2p_server.cpp @@ -41,7 +41,11 @@ #define ED25519_SIGNATURE_LEN 64 #endif +#ifdef WITH_REDIS +#include "redis_storage.h" +#else #include +#endif #include LOG_CATEGORY(P2PServer) @@ -584,6 +588,80 @@ void P2PServer::save_peer_list_async() } } +#ifdef WITH_REDIS + +void P2PServer::save_peer_list() +{ + std::vector peer_list; + { + MutexLock lock(m_peerListLock); + peer_list = m_peerList; + } + + RedisStorage& redis = get_redis_storage(); + + // Build peer list as newline-separated string + std::string peer_data; + for (const Peer& p : peer_list) { + const char* addr_str; + char addr_str_buf[64]; + + if (p.m_isV6) { + in6_addr addr{}; + memcpy(addr.s6_addr, p.m_addr.data, sizeof(addr.s6_addr)); + addr_str = inet_ntop(AF_INET6, &addr, addr_str_buf, sizeof(addr_str_buf)); + if (addr_str) { + peer_data += '['; + peer_data += addr_str; + peer_data += "]:"; + peer_data += std::to_string(p.m_port); + peer_data += '\n'; + } + } + else { + in_addr addr{}; + memcpy(&addr.s_addr, p.m_addr.data + sizeof(raw_ip::ipv4_prefix), sizeof(addr.s_addr)); + addr_str = inet_ntop(AF_INET, &addr, addr_str_buf, sizeof(addr_str_buf)); + if (addr_str) { + peer_data += addr_str; + peer_data += ':'; + peer_data += std::to_string(p.m_port); + peer_data += '\n'; + } + } + } + + if (!redis.set("peers", peer_data)) { + LOGERR(1, "failed to save peer list to Redis"); + } + else { + LOGINFO(5, "peer list saved to Redis (" << peer_list.size() << " peers)"); + } + + // Save onion peers + const SideChain& s = m_pool->side_chain(); + if (s.onion_pubkeys_count() > 0) { + const std::vector pubkeys = s.seen_onion_pubkeys(); + + std::string onion_data; + for (const hash& h : pubkeys) { + onion_data += to_onion_v3(h); + onion_data += '\n'; + } + + if (!redis.set("onion_peers", onion_data)) { + LOGERR(1, "failed to save onion peer list to Redis"); + } + else { + LOGINFO(5, "onion peer list saved to Redis (" << pubkeys.size() << " peers)"); + } + } + + m_peerListLastSaved = seconds_since_epoch(); +} + +#else // !WITH_REDIS + void P2PServer::save_peer_list() { const std::string path = DATA_DIR + saved_peer_list_file_name; @@ -655,6 +733,8 @@ void P2PServer::save_peer_list() m_peerListLastSaved = seconds_since_epoch(); } +#endif // WITH_REDIS + void P2PServer::load_peer_list() { size_t old_size; @@ -749,7 +829,59 @@ void P2PServer::load_peer_list() } } - // Finally load peers from p2pool_peers.txt and p2pool_onion_peers.txt + // Finally load peers from saved list (file or Redis) +#ifdef WITH_REDIS + { + RedisStorage& redis = get_redis_storage(); + std::string peer_data; + + // Load regular peers from Redis + if (redis.get("peers", peer_data) && !peer_data.empty()) { + // Parse newline-separated addresses + size_t pos = 0; + while (pos < peer_data.size()) { + size_t end = peer_data.find('\n', pos); + if (end == std::string::npos) end = peer_data.size(); + + if (end > pos) { + std::string address = peer_data.substr(pos, end - pos); + if (!address.empty()) { + if (!saved_list.empty()) { + saved_list += ','; + } + saved_list += address; + } + } + pos = end + 1; + } + } + + // Load onion peers from Redis + if (!m_socks5Proxy.empty()) { + std::string onion_data; + if (redis.get("onion_peers", onion_data) && !onion_data.empty()) { + std::vector pubkeys; + size_t pos = 0; + while (pos < onion_data.size()) { + size_t end = onion_data.find('\n', pos); + if (end == std::string::npos) end = onion_data.size(); + + if (end > pos) { + std::string address = onion_data.substr(pos, end - pos); + if (!address.empty()) { + const hash h = from_onion_v3(address); + if (!h.empty()) { + pubkeys.emplace_back(h); + } + } + } + pos = end + 1; + } + s.add_onion_pubkeys(pubkeys); + } + } + } +#else // !WITH_REDIS for (size_t i = 0, n = (m_socks5Proxy.empty() ? 1 : 2); i < n; ++i) { const std::string path = DATA_DIR + (i ? saved_onion_peer_list_file_name : saved_peer_list_file_name); @@ -784,6 +916,7 @@ void P2PServer::load_peer_list() } } } +#endif // WITH_REDIS if (saved_list.empty()) { return; diff --git a/src/p2pool.cpp b/src/p2pool.cpp index b26077b..e1c5b2a 100644 --- a/src/p2pool.cpp +++ b/src/p2pool.cpp @@ -51,6 +51,10 @@ #include #include +#ifdef WITH_REDIS +#include "redis_storage.h" +#include +#endif #include LOG_CATEGORY(P2Pool) @@ -84,6 +88,16 @@ p2pool::p2pool(int argc, char* argv[]) m_params = p; +#ifdef WITH_REDIS + // Initialize Redis connection + RedisStorage& redis = get_redis_storage(); + if (!redis.connect(p->m_redisHost, p->m_redisPort, p->m_redisDb)) { + LOGERR(1, "Failed to connect to Redis at " << p->m_redisHost << ":" << p->m_redisPort); + throw std::exception(); + } + LOGINFO(1, "Connected to Redis at " << p->m_redisHost << ":" << p->m_redisPort << " (db " << p->m_redisDb << ")"); +#endif + // P2Pool-nano requires more Monero blocks for the initial sync if (m_params->m_nano) { BLOCK_HEADERS_REQUIRED = 1440; @@ -197,12 +211,17 @@ p2pool::p2pool(int argc, char* argv[]) } m_timer.data = this; +#ifdef WITH_REDIS + // Always create API in Redis mode - data goes to Redis, not files + m_api = new p2pool_api("redis", p->m_localStats); +#else m_api = p->m_apiPath.empty() ? nullptr : new p2pool_api(p->m_apiPath, p->m_localStats); if (p->m_localStats && !m_api) { LOGERR(1, "--local-api and --stratum-api command line parameters can't be used without --data-api"); throw std::exception(); } +#endif m_sideChain = new SideChain(this, type, p->m_mini ? "mini" : (p->m_nano ? "nano" : nullptr), &p->m_devWallet); @@ -258,6 +277,10 @@ p2pool::~p2pool() } #endif +#ifdef WITH_REDIS + get_redis_storage().disconnect(); +#endif + std::vector merge_mining_clients; { WriteLock lock(m_mergeMiningClientsLock); @@ -1610,6 +1633,42 @@ void p2pool::load_found_blocks() return; } +#ifdef WITH_REDIS + RedisStorage& redis = get_redis_storage(); + std::vector entries; + + // Load all found blocks from Redis list + if (!redis.lrange("found_blocks", 0, -1, entries) || entries.empty()) { + api_update_block_found(nullptr, nullptr); + return; + } + + for (const std::string& entry : entries) { + std::istringstream iss(entry); + + time_t timestamp; + iss >> timestamp; + if (iss.fail()) continue; + + uint64_t height; + iss >> height; + if (iss.fail()) continue; + + hash id; + iss >> id; + if (iss.fail()) continue; + + difficulty_type block_difficulty; + iss >> block_difficulty; + if (iss.fail()) continue; + + difficulty_type cumulative_difficulty; + iss >> cumulative_difficulty; + if (iss.fail()) continue; + + m_foundBlocks.emplace_back(timestamp, height, id, block_difficulty, cumulative_difficulty); + } +#else std::ifstream f(DATA_DIR + FOUND_BLOCKS_FILE); if (!f.is_open()) { return; @@ -1638,6 +1697,7 @@ void p2pool::load_found_blocks() m_foundBlocks.emplace_back(timestamp, height, id, block_difficulty, cumulative_difficulty); } +#endif api_update_block_found(nullptr, nullptr); } @@ -2223,6 +2283,16 @@ void p2pool::api_update_block_found(const ChainMain* data, const PoolBlock* bloc difficulty_type diff; if (data && get_difficulty_at_height(data->height, diff)) { +#ifdef WITH_REDIS + // Append found block to Redis list + char buf[512]; + log::Stream s(buf); + s << cur_time << ' ' << data->height << ' ' << data->id << ' ' << diff << ' ' << total_hashes << '\0'; + RedisStorage& redis = get_redis_storage(); + if (!redis.rpush("found_blocks", buf)) { + LOGERR(1, "Failed to save found block to Redis"); + } +#else const std::string path = DATA_DIR + FOUND_BLOCKS_FILE; std::ofstream f(path, std::ios::app); if (f.is_open()) { @@ -2233,6 +2303,7 @@ void p2pool::api_update_block_found(const ChainMain* data, const PoolBlock* bloc else { LOGERR(1, "Failed to update " << path << ": error " << errno); } +#endif } std::vector found_blocks; diff --git a/src/p2pool_api.cpp b/src/p2pool_api.cpp index 8da9f5e..2afff61 100644 --- a/src/p2pool_api.cpp +++ b/src/p2pool_api.cpp @@ -18,16 +18,96 @@ #include "common.h" #include "p2pool_api.h" +#ifdef WITH_REDIS +#include "redis_storage.h" +#else #ifdef _MSC_VER #include #else #include #endif +#endif LOG_CATEGORY(P2Pool API) namespace p2pool { +#ifdef WITH_REDIS + +// Redis-based API implementation - much simpler than file-based + +p2pool_api::p2pool_api(const std::string& api_path, const bool local_stats) + : m_apiPath(api_path) + , m_counter(0) +{ + (void)local_stats; + // Redis doesn't need directory creation or async file handles + uv_mutex_init_checked(&m_dumpDataLock); + LOGINFO(1, "P2Pool API using Redis storage"); +} + +p2pool_api::~p2pool_api() +{ + uv_mutex_destroy(&m_dumpDataLock); +} + +void p2pool_api::create_dir(const std::string& path, bool is_restricted) +{ + (void)path; + (void)is_restricted; + // No directories needed for Redis +} + +void p2pool_api::on_stop() +{ + // Nothing to clean up for Redis +} + +void p2pool_api::dump_to_file_async_internal(Category category, const char* filename, const Callback::Base& callback) +{ + std::vector buf(1024); + log::Stream s(buf.data(), buf.size()); + callback(s); + + // If the buffer was too small, try again with big enough buffer + if (s.m_spilled) { + buf.resize((static_cast(s.m_pos) + s.m_spilled) * 2 + 1); + s.reset(buf.data(), buf.size()); + callback(s); + } + + buf.resize(s.m_pos); + + // Construct Redis key based on category + std::string key = "api:"; + switch (category) { + case Category::GLOBAL: key += "global:"; break; + case Category::NETWORK: key += "network:"; break; + case Category::POOL: key += "pool:"; break; + case Category::LOCAL: key += "local:"; break; + } + key += filename; + + // Write directly to Redis + RedisStorage& redis = get_redis_storage(); + if (!redis.set(key, reinterpret_cast(buf.data()), buf.size())) { + LOGWARN(4, "Failed to write API data to Redis key: " << key); + } +} + +void p2pool_api::dump_to_file() +{ + // Not used in Redis mode - writes happen synchronously in dump_to_file_async_internal +} + +void p2pool_api::on_fs_open(uv_fs_t*) {} +void p2pool_api::on_fs_write(uv_fs_t*) {} +void p2pool_api::on_fs_close(uv_fs_t*) {} +void p2pool_api::on_fs_rename(uv_fs_t*) {} +void p2pool_api::on_fs_error_cleanup(uv_fs_t*) {} + +#else // !WITH_REDIS - Original file-based implementation + p2pool_api::p2pool_api(const std::string& api_path, const bool local_stats) : m_apiPath(api_path) , m_counter(0) @@ -303,4 +383,6 @@ void p2pool_api::on_fs_error_cleanup(uv_fs_t* req) delete work; } +#endif // WITH_REDIS + } // namespace p2pool diff --git a/src/params.cpp b/src/params.cpp index 522f5ea..38ba26a 100644 --- a/src/params.cpp +++ b/src/params.cpp @@ -291,6 +291,23 @@ Params::Params(int argc, char* const argv[]) ok = true; } +#ifdef WITH_REDIS + if ((strcmp(argv[i], "--redis-host") == 0) && (i + 1 < argc)) { + m_redisHost = argv[++i]; + ok = true; + } + + if ((strcmp(argv[i], "--redis-port") == 0) && (i + 1 < argc)) { + m_redisPort = static_cast(std::min(std::max(strtoul(argv[++i], nullptr, 10), 1UL), 65535UL)); + ok = true; + } + + if ((strcmp(argv[i], "--redis-db") == 0) && (i + 1 < argc)) { + m_redisDb = static_cast(strtoul(argv[++i], nullptr, 10)); + ok = true; + } +#endif + if (!ok) { // Wait to avoid log messages overlapping with printf() calls and making a mess on screen std::this_thread::sleep_for(std::chrono::milliseconds(10)); diff --git a/src/params.h b/src/params.h index efd6152..26634a9 100644 --- a/src/params.h +++ b/src/params.h @@ -139,6 +139,12 @@ struct Params std::string m_onionAddress; hash m_onionPubkey; bool m_noClearnetP2P = false; + +#ifdef WITH_REDIS + std::string m_redisHost = "127.0.0.1"; + int32_t m_redisPort = 6379; + int32_t m_redisDb = 0; +#endif }; } // namespace p2pool diff --git a/src/redis_storage.cpp b/src/redis_storage.cpp new file mode 100644 index 0000000..3229b2b --- /dev/null +++ b/src/redis_storage.cpp @@ -0,0 +1,542 @@ +/* + * This file is part of p2pool-salvium-redis + * Redis storage backend for p2pool-salvium observer mode + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, version 3. + */ + +#ifdef WITH_REDIS + +#include "common.h" +#include "redis_storage.h" +#include +#include + +LOG_CATEGORY(RedisStorage) + +namespace p2pool { + +// Global instance +static RedisStorage* g_redis_storage = nullptr; + +RedisStorage& get_redis_storage() { + if (!g_redis_storage) { + g_redis_storage = new RedisStorage(); + } + return *g_redis_storage; +} + +RedisStorage::RedisStorage() + : m_ctx(nullptr) + , m_port(6379) + , m_db(0) + , m_prefix("p2pool:") + , m_connected(false) +{ +} + +RedisStorage::~RedisStorage() { + disconnect(); +} + +bool RedisStorage::connect(const std::string& host, int port, int db) { + std::lock_guard lock(m_mutex); + + if (m_connected && m_ctx) { + return true; + } + + m_host = host; + m_port = port; + m_db = db; + + struct timeval timeout = { 5, 0 }; // 5 seconds + m_ctx = redisConnectWithTimeout(host.c_str(), port, timeout); + + if (!m_ctx) { + LOGERR(1, "Failed to allocate Redis context"); + return false; + } + + if (m_ctx->err) { + LOGERR(1, "Redis connection error: " << static_cast(m_ctx->errstr)); + redisFree(m_ctx); + m_ctx = nullptr; + return false; + } + + // Select database + if (db != 0) { + redisReply* reply = static_cast(redisCommand(m_ctx, "SELECT %d", db)); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LOGERR(1, "Failed to select Redis database " << db); + if (reply) freeReplyObject(reply); + redisFree(m_ctx); + m_ctx = nullptr; + return false; + } + freeReplyObject(reply); + } + + m_connected = true; + LOGINFO(1, "Connected to Redis at " << host << ":" << port << " db=" << db); + return true; +} + +void RedisStorage::disconnect() { + std::lock_guard lock(m_mutex); + + if (m_ctx) { + redisFree(m_ctx); + m_ctx = nullptr; + } + m_connected = false; +} + +bool RedisStorage::reconnect_if_needed() { + if (m_connected && m_ctx && m_ctx->err == 0) { + return true; + } + + LOGWARN(3, "Redis connection lost, reconnecting..."); + + if (m_ctx) { + redisFree(m_ctx); + m_ctx = nullptr; + } + m_connected = false; + + // Try to reconnect (without lock - caller holds it) + struct timeval timeout = { 5, 0 }; + m_ctx = redisConnectWithTimeout(m_host.c_str(), m_port, timeout); + + if (!m_ctx || m_ctx->err) { + LOGERR(1, "Redis reconnection failed"); + if (m_ctx) { + redisFree(m_ctx); + m_ctx = nullptr; + } + return false; + } + + // Select database + if (m_db != 0) { + redisReply* reply = static_cast(redisCommand(m_ctx, "SELECT %d", m_db)); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + if (reply) freeReplyObject(reply); + redisFree(m_ctx); + m_ctx = nullptr; + return false; + } + freeReplyObject(reply); + } + + m_connected = true; + LOGINFO(1, "Redis reconnected successfully"); + return true; +} + +std::string RedisStorage::prefixed_key(const std::string& key) const { + return m_prefix + key; +} + +bool RedisStorage::ping() { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + redisReply* reply = static_cast(redisCommand(m_ctx, "PING")); + if (!reply) return false; + + bool ok = (reply->type == REDIS_REPLY_STATUS && strcmp(reply->str, "PONG") == 0); + freeReplyObject(reply); + return ok; +} + +bool RedisStorage::set(const std::string& key, const std::string& value) { + return set(key, reinterpret_cast(value.data()), value.size()); +} + +bool RedisStorage::set(const std::string& key, const uint8_t* data, size_t size) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "SET %s %b", pkey.c_str(), data, size) + ); + + if (!reply) { + LOGERR(3, "Redis SET failed for key: " << key); + return false; + } + + bool ok = (reply->type == REDIS_REPLY_STATUS && strcmp(reply->str, "OK") == 0); + freeReplyObject(reply); + return ok; +} + +bool RedisStorage::get(const std::string& key, std::string& value) { + std::vector data; + if (!get(key, data)) return false; + value.assign(data.begin(), data.end()); + return true; +} + +bool RedisStorage::get(const std::string& key, std::vector& data) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "GET %s", pkey.c_str()) + ); + + if (!reply) { + return false; + } + + if (reply->type == REDIS_REPLY_NIL) { + freeReplyObject(reply); + return false; + } + + if (reply->type != REDIS_REPLY_STRING) { + freeReplyObject(reply); + return false; + } + + data.assign( + reinterpret_cast(reply->str), + reinterpret_cast(reply->str) + reply->len + ); + freeReplyObject(reply); + return true; +} + +bool RedisStorage::del(const std::string& key) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "DEL %s", pkey.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::exists(const std::string& key) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "EXISTS %s", pkey.c_str()) + ); + + if (!reply) return false; + + bool exists = (reply->type == REDIS_REPLY_INTEGER && reply->integer > 0); + freeReplyObject(reply); + return exists; +} + +bool RedisStorage::hset(const std::string& key, const std::string& field, const uint8_t* data, size_t size) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "HSET %s %s %b", pkey.c_str(), field.c_str(), data, size) + ); + + if (!reply) { + LOGERR(3, "Redis HSET failed for key: " << key << " field: " << field); + return false; + } + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::hget(const std::string& key, const std::string& field, std::vector& data) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "HGET %s %s", pkey.c_str(), field.c_str()) + ); + + if (!reply || reply->type == REDIS_REPLY_NIL) { + if (reply) freeReplyObject(reply); + return false; + } + + if (reply->type != REDIS_REPLY_STRING) { + freeReplyObject(reply); + return false; + } + + data.assign( + reinterpret_cast(reply->str), + reinterpret_cast(reply->str) + reply->len + ); + freeReplyObject(reply); + return true; +} + +bool RedisStorage::hdel(const std::string& key, const std::string& field) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "HDEL %s %s", pkey.c_str(), field.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::hgetall(const std::string& key, std::vector>>& result) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "HGETALL %s", pkey.c_str()) + ); + + if (!reply || reply->type != REDIS_REPLY_ARRAY) { + if (reply) freeReplyObject(reply); + return false; + } + + result.clear(); + for (size_t i = 0; i + 1 < reply->elements; i += 2) { + redisReply* field = reply->element[i]; + redisReply* value = reply->element[i + 1]; + + if (field->type == REDIS_REPLY_STRING && value->type == REDIS_REPLY_STRING) { + std::string fieldStr(field->str, field->len); + std::vector valueData( + reinterpret_cast(value->str), + reinterpret_cast(value->str) + value->len + ); + result.emplace_back(std::move(fieldStr), std::move(valueData)); + } + } + + freeReplyObject(reply); + return true; +} + +int64_t RedisStorage::hlen(const std::string& key) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return -1; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "HLEN %s", pkey.c_str()) + ); + + if (!reply || reply->type != REDIS_REPLY_INTEGER) { + if (reply) freeReplyObject(reply); + return -1; + } + + int64_t len = reply->integer; + freeReplyObject(reply); + return len; +} + +bool RedisStorage::lpush(const std::string& key, const std::string& value) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "LPUSH %s %s", pkey.c_str(), value.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::rpush(const std::string& key, const std::string& value) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "RPUSH %s %s", pkey.c_str(), value.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::lrange(const std::string& key, int64_t start, int64_t stop, std::vector& result) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "LRANGE %s %lld %lld", pkey.c_str(), start, stop) + ); + + if (!reply || reply->type != REDIS_REPLY_ARRAY) { + if (reply) freeReplyObject(reply); + return false; + } + + result.clear(); + for (size_t i = 0; i < reply->elements; ++i) { + if (reply->element[i]->type == REDIS_REPLY_STRING) { + result.emplace_back(reply->element[i]->str, reply->element[i]->len); + } + } + + freeReplyObject(reply); + return true; +} + +int64_t RedisStorage::llen(const std::string& key) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return -1; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "LLEN %s", pkey.c_str()) + ); + + if (!reply || reply->type != REDIS_REPLY_INTEGER) { + if (reply) freeReplyObject(reply); + return -1; + } + + int64_t len = reply->integer; + freeReplyObject(reply); + return len; +} + +bool RedisStorage::sadd(const std::string& key, const std::string& member) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "SADD %s %s", pkey.c_str(), member.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::srem(const std::string& key, const std::string& member) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "SREM %s %s", pkey.c_str(), member.c_str()) + ); + + if (!reply) return false; + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::smembers(const std::string& key, std::vector& result) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "SMEMBERS %s", pkey.c_str()) + ); + + if (!reply || reply->type != REDIS_REPLY_ARRAY) { + if (reply) freeReplyObject(reply); + return false; + } + + result.clear(); + for (size_t i = 0; i < reply->elements; ++i) { + if (reply->element[i]->type == REDIS_REPLY_STRING) { + result.emplace_back(reply->element[i]->str, reply->element[i]->len); + } + } + + freeReplyObject(reply); + return true; +} + +bool RedisStorage::sismember(const std::string& key, const std::string& member) { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + std::string pkey = prefixed_key(key); + redisReply* reply = static_cast( + redisCommand(m_ctx, "SISMEMBER %s %s", pkey.c_str(), member.c_str()) + ); + + if (!reply || reply->type != REDIS_REPLY_INTEGER) { + if (reply) freeReplyObject(reply); + return false; + } + + bool is_member = (reply->integer > 0); + freeReplyObject(reply); + return is_member; +} + +bool RedisStorage::flushdb() { + std::lock_guard lock(m_mutex); + + if (!reconnect_if_needed()) return false; + + redisReply* reply = static_cast(redisCommand(m_ctx, "FLUSHDB")); + if (!reply) return false; + + freeReplyObject(reply); + LOGWARN(1, "Redis database flushed!"); + return true; +} + +} // namespace p2pool + +#endif // WITH_REDIS diff --git a/src/redis_storage.h b/src/redis_storage.h new file mode 100644 index 0000000..80e6f93 --- /dev/null +++ b/src/redis_storage.h @@ -0,0 +1,90 @@ +/* + * This file is part of p2pool-salvium-redis + * Redis storage backend for p2pool-salvium observer mode + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, version 3. + */ + +#pragma once + +#ifdef WITH_REDIS + +#include +#include +#include +#include + +// Forward declaration - hiredis types +struct redisContext; + +namespace p2pool { + +class RedisStorage { +public: + RedisStorage(); + ~RedisStorage(); + + // Initialize connection to Redis + // host: Redis server host (default: "127.0.0.1") + // port: Redis server port (default: 6379) + // db: Redis database number (default: 0) + bool connect(const std::string& host = "127.0.0.1", int port = 6379, int db = 0); + void disconnect(); + bool is_connected() const { return m_connected; } + + // Key-value operations + bool set(const std::string& key, const std::string& value); + bool set(const std::string& key, const uint8_t* data, size_t size); + bool get(const std::string& key, std::string& value); + bool get(const std::string& key, std::vector& data); + bool del(const std::string& key); + bool exists(const std::string& key); + + // Hash operations (for block cache) + bool hset(const std::string& key, const std::string& field, const uint8_t* data, size_t size); + bool hget(const std::string& key, const std::string& field, std::vector& data); + bool hdel(const std::string& key, const std::string& field); + bool hgetall(const std::string& key, std::vector>>& result); + int64_t hlen(const std::string& key); + + // List operations (for peers, blocks list) + bool lpush(const std::string& key, const std::string& value); + bool rpush(const std::string& key, const std::string& value); + bool lrange(const std::string& key, int64_t start, int64_t stop, std::vector& result); + int64_t llen(const std::string& key); + + // Set operations + bool sadd(const std::string& key, const std::string& member); + bool srem(const std::string& key, const std::string& member); + bool smembers(const std::string& key, std::vector& result); + bool sismember(const std::string& key, const std::string& member); + + // Utility + bool ping(); + bool flushdb(); // Use with caution! + + // Key prefix for namespacing + void set_prefix(const std::string& prefix) { m_prefix = prefix; } + std::string get_prefix() const { return m_prefix; } + +private: + std::string prefixed_key(const std::string& key) const; + bool reconnect_if_needed(); + + redisContext* m_ctx; + std::string m_host; + int m_port; + int m_db; + std::string m_prefix; + bool m_connected; + mutable std::mutex m_mutex; +}; + +// Global Redis storage instance +RedisStorage& get_redis_storage(); + +} // namespace p2pool + +#endif // WITH_REDIS diff --git a/src/side_chain.cpp b/src/side_chain.cpp index 492b744..7c63768 100644 --- a/src/side_chain.cpp +++ b/src/side_chain.cpp @@ -34,6 +34,9 @@ #include "json_parsers.h" #include "crypto.h" #include "hardforks/hardforks.h" +#ifdef WITH_REDIS +#include "redis_storage.h" +#endif #include #if !defined(_MSC_VER) || !defined(__cppcheck__) @@ -3298,51 +3301,143 @@ void SideChain::clear_checkpoints() void SideChain::save_checkpoints() const { - const std::string path = DATA_DIR + "p2pool_checkpoints.dat"; - +#ifdef WITH_REDIS ReadLock lock(m_checkpointsLock); - + + if (m_checkpoints.empty()) { + get_redis_storage().del("checkpoints"); + return; + } + + // Build binary buffer matching file format + std::vector buf; + const uint32_t version = 1; + const uint32_t count = static_cast(m_checkpoints.size()); + + buf.insert(buf.end(), reinterpret_cast(&version), + reinterpret_cast(&version) + sizeof(version)); + buf.insert(buf.end(), reinterpret_cast(&count), + reinterpret_cast(&count) + sizeof(count)); + + for (const Checkpoint& cp : m_checkpoints) { + buf.insert(buf.end(), reinterpret_cast(&cp.height), + reinterpret_cast(&cp.height) + sizeof(cp.height)); + buf.insert(buf.end(), cp.id.h, cp.id.h + HASH_SIZE); + buf.insert(buf.end(), reinterpret_cast(&cp.cumulative_difficulty), + reinterpret_cast(&cp.cumulative_difficulty) + sizeof(cp.cumulative_difficulty)); + } + + if (!get_redis_storage().set("checkpoints", buf.data(), buf.size())) { + LOGWARN(1, "Failed to save checkpoints to Redis"); + return; + } + LOGINFO(3, "Saved " << count << " checkpoints to Redis"); +#else + const std::string path = DATA_DIR + "p2pool_checkpoints.dat"; + + ReadLock lock(m_checkpointsLock); + if (m_checkpoints.empty()) { // No checkpoints to save, remove old file if exists remove(path.c_str()); return; } - + std::ofstream f(path, std::ios::binary); if (!f.is_open()) { LOGWARN(1, "Failed to save checkpoints to " << path); return; } - + // Write version marker for future compatibility const uint32_t version = 1; f.write(reinterpret_cast(&version), sizeof(version)); - + // Write checkpoint count const uint32_t count = static_cast(m_checkpoints.size()); f.write(reinterpret_cast(&count), sizeof(count)); - + // Write each checkpoint for (const Checkpoint& cp : m_checkpoints) { f.write(reinterpret_cast(&cp.height), sizeof(cp.height)); f.write(reinterpret_cast(cp.id.h), HASH_SIZE); f.write(reinterpret_cast(&cp.cumulative_difficulty), sizeof(cp.cumulative_difficulty)); } - + f.close(); LOGINFO(3, "Saved " << count << " checkpoints to " << path); +#endif } void SideChain::load_checkpoints() { +#ifdef WITH_REDIS + std::vector buf; + if (!get_redis_storage().get("checkpoints", buf)) { + LOGINFO(3, "No checkpoints found in Redis (normal for first run)"); + return; + } + + if (buf.size() < 8) { + LOGWARN(1, "Checkpoint data too small, ignoring"); + return; + } + + size_t offset = 0; + + // Read version + uint32_t version = *reinterpret_cast(buf.data() + offset); + offset += sizeof(version); + if (version != 1) { + LOGWARN(1, "Unknown checkpoint version " << version << ", ignoring"); + return; + } + + // Read count + uint32_t count = *reinterpret_cast(buf.data() + offset); + offset += sizeof(count); + if (count > 100) { + LOGWARN(1, "Suspicious checkpoint count " << count << ", ignoring"); + return; + } + + WriteLock lock(m_checkpointsLock); + m_checkpoints.clear(); + + for (uint32_t i = 0; i < count; ++i) { + if (offset + sizeof(uint64_t) + HASH_SIZE + sizeof(difficulty_type) > buf.size()) { + LOGWARN(1, "Checkpoint data corrupted at entry " << i << ", discarding"); + m_checkpoints.clear(); + return; + } + + Checkpoint cp; + memcpy(&cp.height, buf.data() + offset, sizeof(cp.height)); + offset += sizeof(cp.height); + memcpy(cp.id.h, buf.data() + offset, HASH_SIZE); + offset += HASH_SIZE; + memcpy(&cp.cumulative_difficulty, buf.data() + offset, sizeof(cp.cumulative_difficulty)); + offset += sizeof(cp.cumulative_difficulty); + + m_checkpoints.push_back(cp); + } + + LOGINFO(1, "Loaded " << count << " checkpoints from Redis"); + + if (!m_checkpoints.empty()) { + LOGINFO(1, "Latest anchor point: height " << m_checkpoints.back().height << + ", id " << m_checkpoints.back().id); + m_checkpointsNeedValidation = true; + } +#else const std::string path = DATA_DIR + "p2pool_checkpoints.dat"; - + std::ifstream f(path, std::ios::binary); if (!f.is_open()) { LOGINFO(3, "No checkpoint file found at " << path << " (normal for first run)"); return; } - + // Read version uint32_t version = 0; f.read(reinterpret_cast(&version), sizeof(version)); @@ -3350,42 +3445,43 @@ void SideChain::load_checkpoints() LOGWARN(1, "Unknown checkpoint file version " << version << ", ignoring"); return; } - + // Read checkpoint count uint32_t count = 0; f.read(reinterpret_cast(&count), sizeof(count)); - + if (count > 100) { LOGWARN(1, "Checkpoint file has suspicious count " << count << ", ignoring"); return; } - + WriteLock lock(m_checkpointsLock); m_checkpoints.clear(); - + for (uint32_t i = 0; i < count; ++i) { Checkpoint cp; f.read(reinterpret_cast(&cp.height), sizeof(cp.height)); f.read(reinterpret_cast(cp.id.h), HASH_SIZE); f.read(reinterpret_cast(&cp.cumulative_difficulty), sizeof(cp.cumulative_difficulty)); - + if (f.fail()) { LOGWARN(1, "Checkpoint file corrupted at entry " << i << ", discarding"); m_checkpoints.clear(); return; } - + m_checkpoints.push_back(cp); } - + f.close(); LOGINFO(1, "Loaded " << count << " checkpoints from " << path); if (!m_checkpoints.empty()) { - LOGINFO(1, "Latest anchor point: height " << m_checkpoints.back().height << + LOGINFO(1, "Latest anchor point: height " << m_checkpoints.back().height << ", id " << m_checkpoints.back().id); m_checkpointsNeedValidation = true; } +#endif } bool SideChain::validate_loaded_checkpoints()