Compare commits

..
3 Commits
19 changed files with 318 additions and 397 deletions

No files matched your search

+3 -3
View File
@@ -13,7 +13,7 @@ jobs:
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev libasio-dev \
libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-dev
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev
- name: Cython
run: sudo pip3 install Cython
- name: llhttp
@@ -56,7 +56,7 @@ jobs:
sudo apt update && \
sudo apt install ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev \
libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip \
libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev
libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-0-dev
- name: Python dependencies
run: sudo pip3 install meson Cython
- name: Configure
@@ -73,7 +73,7 @@ jobs:
run: |
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev libfmt-dev \
libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-dev
libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-0-dev
- name: Configure
run: |
mkdir build && cd build
+1 -1
View File
@@ -12,7 +12,7 @@ jobs:
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-dev libasio-dev \
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev libasio-dev \
llvm llvm-dev clang clang-tools && \
sudo apt remove gcc g++
+9 -10
View File
@@ -33,7 +33,6 @@ option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp"
option (OPENDHT_PROXY_OPENSSL "Build DHT proxy with OpenSSL" ON)
CMAKE_DEPENDENT_OPTION(OPENDHT_HTTP "Build embedded http(s) client" OFF "NOT OPENDHT_PROXY_SERVER;NOT OPENDHT_PROXY_CLIENT" ON)
option (OPENDHT_PEER_DISCOVERY "Enable multicast peer discovery" ON)
option (OPENDHT_IO_URING "Use io_uring if available on the system (Linux only)" OFF)
option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
option (OPENDHT_TESTS_NETWORK "Enable unit tests that require network access" ON)
option (OPENDHT_C "Build C bindings" OFF)
@@ -98,7 +97,7 @@ if (NOT MSVC)
# llhttp
find_path(LLHTTP_INCLUDE_DIR llhttp.h)
find_library(LLHTTP_LIBRARY libllhttp)
find_library(LLHTTP_LIBRARY NAMES llhttp libllhttp)
if (LLHTTP_INCLUDE_DIR AND LLHTTP_LIBRARY)
message(STATUS "Found llhttp ${LLHTTP_INCLUDE_DIR} ${LLHTTP_LIBRARY}")
add_library(llhttp_static STATIC IMPORTED)
@@ -108,7 +107,7 @@ if (NOT MSVC)
)
set(llhttp_target llhttp_static)
else()
FetchContent_Declare(llhttp-local URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.1.tar.gz")
FetchContent_Declare(llhttp-local URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.0.tar.gz")
if (BUILD_SHARED_LIBS)
set(BUILD_SHARED_LIBS ON CACHE INTERNAL "")
else()
@@ -123,7 +122,7 @@ if (NOT MSVC)
endif()
endif()
set(http_lib "-lllhttp")
if (NOT Jsoncpp_FOUND)
message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
endif()
@@ -157,14 +156,14 @@ else ()
${WIN32_DEP_DIR}/openssl/include
${WIN32_DEP_DIR}/restinio/dev
${WIN32_DEP_DIR}/fmt/include
${WIN32_DEP_DIR}/llhttp/include
${WIN32_DEP_DIR}/http_parser
)
endif ()
endif ()
if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
add_definitions(-DASIO_STANDALONE)
if (OPENDHT_IO_URING AND UNIX AND NOT APPLE)
if (UNIX AND NOT APPLE)
pkg_search_module(liburing IMPORTED_TARGET liburing)
endif ()
endif()
@@ -344,12 +343,12 @@ if (MSVC)
)
if (OPENDHT_HTTP)
add_obj_lib (win32_fmt ${WIN32_DEP_DIR}/fmt/msvc/Release/fmt.lib)
add_obj_lib (win32_llhttp ${WIN32_DEP_DIR}/llhttp/build/Release/llhttp.lib)
add_obj_lib (win32_http_parser ${WIN32_DEP_DIR}/http_parser/x64/Release/http-parser.lib)
add_obj_lib (win32_ssl ${WIN32_DEP_DIR}/openssl/libssl_static.lib)
add_obj_lib (win32_crypto ${WIN32_DEP_DIR}/openssl/libcrypto_static.lib)
list (APPEND obj_libs
$<TARGET_OBJECTS:win32_fmt>
$<TARGET_OBJECTS:win32_llhttp>
$<TARGET_OBJECTS:win32_http_parser>
$<TARGET_OBJECTS:win32_ssl>
$<TARGET_OBJECTS:win32_crypto>
)
@@ -362,7 +361,7 @@ if (MSVC)
)
list (APPEND win32_Libs
${PROJECT_SOURCE_DIR}/../fmt/msvc/Release/fmt.lib
${PROJECT_SOURCE_DIR}/../llhttp/build/Release/llhttp.lib
${PROJECT_SOURCE_DIR}/../http_parser/x64/Release/http-parser.lib
${PROJECT_SOURCE_DIR}/../openssl/libssl.lib
${PROJECT_SOURCE_DIR}/../openssl/libcrypto.lib
)
@@ -408,7 +407,7 @@ else()
if (OPENDHT_PROXY_OPENSSL)
target_link_libraries(opendht PUBLIC PkgConfig::OPENSSL)
endif()
if (OPENDHT_IO_URING AND liburing_FOUND)
if (liburing_FOUND)
set(iouring_lib ", liburing")
target_link_libraries(opendht PUBLIC PkgConfig::liburing)
target_compile_definitions(opendht PUBLIC ASIO_HAS_IO_URING ASIO_DISABLE_EPOLL)
+10 -10
View File
@@ -150,22 +150,22 @@ public:
std::shared_ptr<NodeInfo> nodeInfo {};
std::string toString() const {
auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
"Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
"Requests: {} per second.",
listenCount, putCount, pushListenersCount,
print_duration(lastUpdated - serverStartTime),
androidPush.toString(), iosPush.toString(), unifiedPush.toString(),
requestRate);
std::ostringstream ss;
ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
ss << "Push requests in the last " << print_duration(lastUpdated - serverStartTime) << ": "
<< "[Android: " << androidPush.toString() << "], "
<< "[iOS: " << iosPush.toString() << "], "
<< "[Unified: " << unifiedPush.toString() << "]" << std::endl;
ss << "Requests: " << requestRate << " per second." << std::endl;
if (nodeInfo) {
auto& ipv4 = nodeInfo->ipv4;
if (ipv4.table_depth > 1)
ret += fmt::format("IPv4 Network estimation: {}\n", ipv4.getNetworkSizeEstimation());
ss << "IPv4 Network estimation: " << ipv4.getNetworkSizeEstimation() << std::endl;;
auto& ipv6 = nodeInfo->ipv6;
if (ipv6.table_depth > 1)
ret += fmt::format("IPv6 Network estimation: {}\n", ipv6.getNetworkSizeEstimation());
ss << "IPv6 Network estimation: " << ipv6.getNetworkSizeEstimation() << std::endl;;
}
return ret;
return ss.str();
}
/**
+1 -1
View File
@@ -72,7 +72,7 @@ class OPENDHT_PUBLIC Url
{
public:
Url() = default;
Url(std::string_view url);
Url(const std::string& url);
std::string url;
std::string protocol {"http"};
std::string host;
+1 -1
View File
@@ -409,7 +409,7 @@ private:
/**
* This mutlimap contains all prefix insert in the tree in time order
* We could then delete the last one if there are too many nodes
* We could then delete the last one if there is too much node
* The tree will self destroy is branch ( thanks to share_ptr )
*/
std::multimap<time_point, std::shared_ptr<Node>> leaves_;
+1 -1
View File
@@ -24,7 +24,7 @@ namespace proxy {
constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
constexpr const std::chrono::hours OP_MARGIN {2}; // two hours
constexpr const char* const HTTP_PROTO {"http://"};
using ListenToken = uint64_t;
}
+22 -9
View File
@@ -21,7 +21,6 @@
#include "def.h"
#include <msgpack.hpp>
#include <fmt/format.h>
#include <chrono>
#include <random>
@@ -61,7 +60,7 @@ void erase_if(std::map<Key, Item>& map, const Condition& condition)
* Split "[host]:port" or "host:port" to pair<"host", "port">.
*/
OPENDHT_PUBLIC std::pair<std::string, std::string>
splitPort(std::string_view s);
splitPort(const std::string& s);
class OPENDHT_PUBLIC DhtException : public std::runtime_error {
public:
@@ -85,23 +84,37 @@ using duration = clock::duration;
time_point from_time_t(std::time_t t);
std::time_t to_time_t(time_point t);
inline std::string
to_str(double d) {
char buf[16];
auto ret = snprintf(buf, sizeof(buf), "%.3g", d);
return (ret < 0) ? std::to_string(d) : std::string(buf, ret);
}
/**
* Converts std::chrono::duration to floating-point seconds.
*/
template <class DT>
static double
print_dt(DT d) {
return std::chrono::duration_cast<std::chrono::duration<double>>(d).count();
}
template <class DT>
static std::string
print_duration(DT d) {
if (d < std::chrono::seconds(0)) {
return "-" + print_duration(-d);
} else if (d < std::chrono::milliseconds(1)) {
return fmt::format("{:.3g} us", std::chrono::duration_cast<std::chrono::duration<double, std::micro>>(d).count());
return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::micro>>(d).count()) + " us";
} else if (d < std::chrono::seconds(1)) {
return fmt::format("{:.3g} ms", std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(d).count());
return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(d).count()) + " ms";
} else if (d < std::chrono::minutes(1)) {
return fmt::format("{:.3g} s", std::chrono::duration_cast<std::chrono::duration<double>>(d).count());
return to_str(std::chrono::duration_cast<std::chrono::duration<double>>(d).count()) + " s";
} else if (d < std::chrono::hours(1)) {
return fmt::format("{:.3g} min", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<60>>>(d).count());
} else if (d < std::chrono::hours(72)) {
return fmt::format("{:.3g} h", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<3600>>>(d).count());
return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<60>>>(d).count()) + " min";
} else {
return fmt::format("{:.3g} days", std::chrono::duration_cast<std::chrono::duration<double, std::ratio<86400>>>(d).count());
return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<3600>>>(d).count()) + " h";
}
}
+67 -67
View File
@@ -113,7 +113,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
--*remaining;
if (logger_)
logger_->w("Shutting down node: %u ops remaining", *remaining);
logger_->w("shuting down node: %u ops remaining", *remaining);
if (!*remaining && cb) { cb(); }
};
@@ -121,7 +121,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
*remaining += maintainStorage(str, true, str_donecb);
if (logger_)
logger_->w("Shutting down node: after storage, %u ops", *remaining);
logger_->w("shuting down node: after storage, %u ops", *remaining);
if (!*remaining) {
if (cb) cb();
@@ -147,7 +147,7 @@ Dht::sendCachedPing(Bucket& b)
{
if (b.cached)
if (logger_)
logger_->d(b.cached->id, "[node %s] Sending ping to cached node", b.cached->toString().c_str());
logger_->d(b.cached->id, "[node %s] sending ping to cached node", b.cached->toString().c_str());
b.sendCachedPing(network_engine);
}
@@ -280,7 +280,7 @@ Dht::expireSearches()
auto b = sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty() && sr.step_time < t;
if (b) {
if (logger_)
logger_->d(srp.first, "[search %s] Removing search", srp.first.toString().c_str());
logger_->d(srp.first, "[search %s] removing search", srp.first.toString().c_str());
sr.clear();
return b;
} else { return false; }
@@ -360,7 +360,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
auto query_for_vid = std::make_shared<Query>(Select {}, Where {}.id(vid));
sn->pagination_queries[query].push_back(query_for_vid);
if (logger_)
logger_->d(id, sn->node->id, "[search %s] [node %s] Sending %s",
logger_->d(id, sn->node->id, "[search %s] [node %s] sending %s",
id.toString().c_str(), sn->node->toString().c_str(), query_for_vid->toString().c_str());
sn->getStatus[query_for_vid] = network_engine.sendGetValues(status.node,
id,
@@ -371,8 +371,8 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
);
} catch (const std::out_of_range&) {
if (logger_)
logger_->e(id, sn->node->id, "[search %s] [node %s] Received non-id field in response to "\
"'SELECT id' request…",
logger_->e(id, sn->node->id, "[search %s] [node %s] received non-id field in response to "\
"'SELECT id' request...",
id.toString().c_str(), sn->node->toString().c_str());
}
}
@@ -381,7 +381,7 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
n->pagination_queries[query].push_back(select_q);
if (logger_)
logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending %s",
logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending %s",
sr->id.toString().c_str(), n->node->toString().c_str(), select_q->toString().c_str());
n->getStatus[select_q] = network_engine.sendGetValues(n->node,
sr->id,
@@ -426,7 +426,7 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
return nullptr;
/* if (logger_)
logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'find_node'",
logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending 'find_node'",
sr->id.toString().c_str(), n->node->toString().c_str());*/
n->getStatus[query] = network_engine.sendFindNode(n->node,
sr->id,
@@ -439,9 +439,9 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
continue;
if (query and not query->select.empty()) {
/* The request contains a select. No need to paginate… */
/* The request contains a select. No need to paginate... */
/* if (logger_)
logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'get'",
logger_->d(sr->id, n->node->id, "[search %s] [node %s] sending 'get'",
sr->id.toString().c_str(), n->node->toString().c_str());*/
n->getStatus[query] = network_engine.sendGetValues(n->node,
sr->id,
@@ -453,7 +453,7 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
paginate(ws, query, n);
}
/* We only attempt to send one request. Return. */
/* We only try to send one request. return. */
return n;
} while (++cb != sr->callbacks.end());
@@ -519,7 +519,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
/* only put the value if the node doesn't already have it */
if (not hasValue or seq_no < a.value->seq) {
if (logger_)
logger_->d(sr->id, sn->node->id, "[search %s] [node %s] Sending 'put' (vid: %016" PRIx64 ")",
logger_->d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
auto created = a.permanent ? time_point::max() : a.created;
acked = {
@@ -528,7 +528,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
};
} else if (hasValue and a.permanent) {
if (logger_)
logger_->w(sr->id, sn->node->id, "[search %s] [node %s] Sending 'refresh' (vid: %016" PRIx64 ")",
logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %016" PRIx64 ")",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
acked = {
network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
@@ -540,7 +540,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
](const net::Request& /*req*/, net::DhtProtocolException&& e){
if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
if (logger_)
logger_->e(node->id, "[node %s] Returned error 404: storage not found", node->toString().c_str());
logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
if (auto sr = ws.lock()) {
if (auto sn = sr->getNode(node)) {
sn->acked[v->id] = {
@@ -558,7 +558,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
};
} else {
if (logger_)
logger_->w(sr->id, sn->node->id, "[search %s] [node %s] Already has value (vid: %016" PRIx64 "). Aborting.",
logger_->w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %016" PRIx64 "). Aborting.",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
ack_req->reply_time = now;
@@ -594,7 +594,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
sendQuery = true;
} else {
if (logger_)
logger_->w(sr->id, n.node->id, "[search %s] [node %s] Sending 'put' (vid: %016" PRIx64 ")",
logger_->w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id);
n.acked[a.value->id] = {
network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired),
@@ -607,7 +607,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel) {
if (sendQuery) {
n.probe_query = PROBE_QUERY;
if (logger_)
logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending %s",
logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending %s",
sr->id.toString().c_str(), n.node->toString().c_str(), n.probe_query->toString().c_str());
auto req = network_engine.sendGetValues(n.node,
sr->id,
@@ -634,7 +634,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
if (n.getListenTime(r, listenExp) > scheduler.time())
continue;
// if (logger_)
// logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending 'listen'",
// logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
// sr->id.toString().c_str(), n.node->toString().c_str());
if (r == n.listenStatus.end()) {
@@ -706,7 +706,7 @@ Dht::searchStep(std::weak_ptr<Search> ws)
auto syncLevel = preSynced ? level - MARGIN : 0;
/*if (auto req_count = sr->currentlySolicitedNodeCount())
if (logger_)
logger_->d(sr->id, "[search %s IPv%c] Step (%d requests)",
logger_->d(sr->id, "[search %s IPv%c] step (%d requests)",
sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);*/
sr->step_time = now;
@@ -787,7 +787,7 @@ Dht::searchStep(std::weak_ptr<Search> ws)
if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
{
if (logger_)
logger_->w(sr->id, "[search %s IPv%c] Expired", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6');
logger_->w(sr->id, "[search %s IPv%c] expired", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6');
sr->expire();
if (not public_stable)
connectivityChanged(sr->af);
@@ -804,7 +804,7 @@ unsigned Dht::refill(Dht::Search& sr) {
if (cached_nodes.empty()) {
if (logger_)
logger_->e(sr.id, "[search %s IPv%c] No nodes from cache while refilling search",
logger_->e(sr.id, "[search %s IPv%c] no nodes from cache while refilling search",
sr.id.toString().c_str(), (sr.af == AF_INET) ? '4' : '6');
return 0;
}
@@ -816,7 +816,7 @@ unsigned Dht::refill(Dht::Search& sr) {
++inserted;
}
if (logger_)
logger_->d(sr.id, "[search %s IPv%c] Refilled search with %u nodes from node cache",
logger_->d(sr.id, "[search %s IPv%c] refilled search with %u nodes from node cache",
sr.id.toString().c_str(), (sr.af == AF_INET) ? '4' : '6', inserted);
return inserted;
}
@@ -828,7 +828,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
{
if (!isRunning(af)) {
if (logger_)
logger_->e(id, "[search %s IPv%c] Unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
logger_->e(id, "[search %s IPv%c] unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
if (dcb)
dcb(false, {});
return {};
@@ -856,7 +856,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
}
if (not sr) {
if (logger_)
logger_->e(id, "[search %s IPv%c] Maximum number of searches reached!", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
logger_->e(id, "[search %s IPv%c] maximum number of searches reached !", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
return {};
}
}
@@ -870,7 +870,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
sr->nodes.reserve(SEARCH_NODES+1);
sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
if (logger_)
logger_->w(id, "[search %s IPv%c] New search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
logger_->w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
if (search_id == 0)
search_id++;
}
@@ -904,16 +904,16 @@ Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filte
{
if (!isRunning(af))
return 0;
// logger__ERR("[search %s IPv%c] Search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_duration(tm-clock::now()));
// logger__ERR("[search %s IPv%c] search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_dt(tm-clock::now()));
//logger__WARN("listenTo %s", id.toString().c_str());
auto& srs = searches(af);
auto srp = srs.find(id);
Sp<Search> sr = (srp == srs.end()) ? search(id, af) : srp->second;
if (!sr)
throw DhtException("Unable to create search");
throw DhtException("Can't create search");
if (logger_)
logger_->w(id, "[search %s IPv%c] Listen", id.to_c_str(), (af == AF_INET) ? '4' : '6');
logger_->w(id, "[search %s IPv%c] listen", id.to_c_str(), (af == AF_INET) ? '4' : '6');
return sr->listen(cb, std::move(f), q, scheduler);
}
@@ -1022,7 +1022,7 @@ Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point cr
storageStore(id, val, created, {}, permanent);
if (logger_)
logger_->d(id, "put: adding %s → %s", id.toString().c_str(), val->toString().c_str());
logger_->d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());
auto op = std::make_shared<OpStatus>();
auto donecb = [callback](const std::vector<Sp<Node>>& nodes, OpStatus& op) {
@@ -1265,7 +1265,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
vals.push_back(v);
if (not vals.empty()) {
if (logger_)
logger_->d(id, "[store %s] Sending update local listener with token %lu",
logger_->d(id, "[store %s] sending update local listener with token %lu",
id.toString().c_str(),
l.first);
cbs.emplace_back(l.second.get_cb, std::move(vals));
@@ -1286,7 +1286,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
if (f and not f(*v))
continue;
if (logger_)
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending update",
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending update",
id.toString().c_str(),
node_listeners.first->toString().c_str());
std::vector<Sp<Value>> vals {};
@@ -1392,7 +1392,7 @@ void
Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
{
if (logger_)
logger_->d(id, "[store %s] Discarded %ld values (%ld bytes)",
logger_->d(id, "[store %s] discarded %ld values (%ld bytes)",
id.toString().c_str(), values.size(), totalSize);
total_store_size -= totalSize;
@@ -1410,7 +1410,7 @@ Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>
for (const auto& node_listeners : st.listeners) {
for (const auto& l : node_listeners.second) {
if (logger_)
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending expired",
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
id.toString().c_str(),
node_listeners.first->toString().c_str());
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
@@ -1432,7 +1432,7 @@ Dht::expireStore()
if (i->second.empty() && i->second.listeners.empty() && i->second.local_listeners.empty()) {
if (logger_)
logger_->d(i->first, "[store %s] Discarding empty storage", i->first.toString().c_str());
logger_->d(i->first, "[store %s] discarding empty storage", i->first.toString().c_str());
i = store.erase(i);
}
else
@@ -1900,7 +1900,7 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
auto n = q->randomNode(rd);
if (n) {
if (logger_)
logger_->d(id, n->id, "[node %s] Sending [find %s] for neighborhood maintenance",
logger_->d(id, n->id, "[node %s] sending [find %s] for neighborhood maintenance",
n->toString().c_str(), id.toString().c_str());
/* Since our node-id is the same in both DHTs, it's probably
profitable to query both families. */
@@ -1954,14 +1954,14 @@ Dht::bucketMaintenance(RoutingTable& list)
}
if (logger_)
logger_->d(id, n->id, "[node %s] Sending find %s for bucket maintenance", n->toString().c_str(), id.toString().c_str());
logger_->d(id, n->id, "[node %s] sending find %s for bucket maintenance", n->toString().c_str(), id.toString().c_str());
//auto start = scheduler.time();
network_engine.sendFindNode(n, id, want, nullptr, [this,n](const net::Request&, bool over) {
if (over) {
const auto& end = scheduler.time();
// using namespace std::chrono;
// if (logger_)
// logger_->d(n->id, "[node %s] Bucket maintenance op expired after %s", n->toString().c_str(), print_duration(end-start).c_str());
// logger_->d(n->id, "[node %s] bucket maintenance op expired after %s", n->toString().c_str(), print_duration(end-start).c_str());
scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
}
});
@@ -1979,7 +1979,7 @@ Dht::dataPersistence(InfoHash id)
auto str = store.find(id);
if (str != store.end() and now > str->second.maintenance_time) {
if (logger_)
logger_->d(id, "[storage %s] Maintenance (%u values, %u bytes)",
logger_->d(id, "[storage %s] maintenance (%u values, %u bytes)",
id.toString().c_str(), str->second.valueCount(), str->second.totalSize());
maintainStorage(*str);
str->second.maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
@@ -2033,7 +2033,7 @@ Dht::periodic(const uint8_t *buf, size_t buflen, SockAddr from, const time_point
network_engine.processMessage(buf, buflen, std::move(from));
} catch (const std::exception& e) {
if (logger_)
logger_->w("Unable to process message: %s", e.what());
logger_->w("Can't process message: %s", e.what());
}
}
return scheduler.run();
@@ -2087,7 +2087,7 @@ Dht::bootstrap()
}
} catch (const std::exception& e) {
if (logger_)
logger_->e(myid, "Unable to resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
}
}
scheduler.cancel(bootstrapJob);
@@ -2118,12 +2118,12 @@ Dht::confirmNodes()
if (dht4.searches.empty() and dht4.status == NodeStatus::Connected) {
if (logger_)
logger_->d(myid, "[confirm nodes] Initial IPv4 'get' for my id (%s)", myid.toString().c_str());
logger_->d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str());
search(myid, AF_INET);
}
if (dht6.searches.empty() and dht6.status == NodeStatus::Connected) {
if (logger_)
logger_->d(myid, "[confirm nodes] Initial IPv6 'get' for my id (%s)", myid.toString().c_str());
logger_->d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str());
search(myid, AF_INET6);
}
@@ -2281,7 +2281,7 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
const auto& node = req->node;
if (e.getCode() == net::DhtProtocolException::UNAUTHORIZED) {
if (logger_)
logger_->e(node->id, "[node %s] Token flush", node->toString().c_str());
logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
node->authError();
for (auto& srp : searches(node->getFamily())) {
auto& sr = srp.second;
@@ -2296,7 +2296,7 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
}
} else if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
if (logger_)
logger_->e(node->id, "[node %s] Returned error 404: storage not found", node->toString().c_str());
logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
node->cancelRequest(req);
}
}
@@ -2347,7 +2347,7 @@ Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query
if (st != store.end() && not st->second.empty()) {
answer.values = st->second.get(query.where.getFilter());
if (logger_)
logger_->d(hash, "[node %s] Sending %u values", node->toString().c_str(), answer.values.size());
logger_->d(hash, "[node %s] sending %u values", node->toString().c_str(), answer.values.size());
}
return answer;
}
@@ -2359,18 +2359,18 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
{
if (not sr) {
if (logger_)
logger_->w("[search unknown] Got reply to 'get'. Ignoring.");
logger_->w("[search unknown] got reply to 'get'. Ignoring.");
return;
}
/* if (logger_)
logger_->d(sr->id, "[search %s] [node %s] Got reply to 'get' with %u nodes",
logger_->d(sr->id, "[search %s] [node %s] got reply to 'get' with %u nodes",
sr->id.toString().c_str(), node->toString().c_str(), a.nodes4.size()+a.nodes6.size());*/
if (not a.ntoken.empty()) {
if (not a.values.empty() or not a.fields.empty()) {
if (logger_)
logger_->d(sr->id, node->id, "[search %s] [node %s] Found %u values",
logger_->d(sr->id, node->id, "[search %s] [node %s] found %u values",
sr->id.toString().c_str(), node->toString().c_str(), a.values.size());
for (auto& getp : sr->callbacks) { /* call all callbacks for this search */
auto& get = getp.second;
@@ -2419,7 +2419,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
}
} else {
if (logger_)
logger_->w(sr->id, "[node %s] No token provided. Ignoring response content.", node->toString().c_str());
logger_->w(sr->id, "[node %s] no token provided. Ignoring response content.", node->toString().c_str());
network_engine.blacklistNode(node);
}
@@ -2436,7 +2436,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
{
if (not hash) {
if (logger_)
logger_->w(node->id, "[node %s] Listen with no info_hash", node->toString().c_str());
logger_->w(node->id, "[node %s] listen with no info_hash", node->toString().c_str());
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::LISTEN_NO_INFOHASH
@@ -2444,7 +2444,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
}
if (not tokenMatch(token, node->getAddr())) {
if (logger_)
logger_->w(hash, node->id, "[node %s] Incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
logger_->w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
}
Query q = query;
@@ -2456,7 +2456,7 @@ void
Dht::onListenDone(const Sp<Node>& /* node */, net::RequestAnswer& /* answer */, Sp<Search>& sr)
{
// if (logger_)
// logger_->d(sr->id, node->id, "[search %s] [node %s] Got listen confirmation",
// logger_->d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
// sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
if (not sr->done) {
@@ -2476,7 +2476,7 @@ Dht::onAnnounce(Sp<Node> n,
auto& node = *n;
if (not hash) {
if (logger_)
logger_->w(node.id, "Put with no info_hash");
logger_->w(node.id, "put with no info_hash");
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::PUT_NO_INFOHASH
@@ -2484,7 +2484,7 @@ Dht::onAnnounce(Sp<Node> n,
}
if (!tokenMatch(token, node.getAddr())) {
if (logger_)
logger_->w(hash, node.id, "[node %s] Incorrect token %s for 'put'", node.toString().c_str(), hash.toString().c_str());
logger_->w(hash, node.id, "[node %s] incorrect token %s for 'put'", node.toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::PUT_WRONG_TOKEN};
}
{
@@ -2493,7 +2493,7 @@ Dht::onAnnounce(Sp<Node> n,
auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
if (logger_)
logger_->w(hash, node.id, "[node %s] Announce too far from the target. Dropping value.", node.toString().c_str());
logger_->w(hash, node.id, "[node %s] announce too far from the target. Dropping value.", node.toString().c_str());
return {};
}
}
@@ -2502,7 +2502,7 @@ Dht::onAnnounce(Sp<Node> n,
for (const auto& v : values) {
if (v->id == Value::INVALID_ID) {
if (logger_)
logger_->w(hash, node.id, "[value %s] Incorrect value id", hash.toString().c_str());
logger_->w(hash, node.id, "[value %s] incorrect value id", hash.toString().c_str());
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::PUT_INVALID_ID
@@ -2514,17 +2514,17 @@ Dht::onAnnounce(Sp<Node> n,
if (*lv == *vc) {
storageRefresh(hash, v->id);
if (logger_)
logger_->d(hash, node.id, "[store %s] [node %s] Refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
} else {
const auto& type = getType(lv->type);
if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
if (logger_)
logger_->d(hash, node.id, "[store %s] Editing %s",
logger_->d(hash, node.id, "[store %s] editing %s",
hash.toString().c_str(), vc->toString().c_str());
storageStore(hash, vc, created, node.getAddr());
} else {
if (logger_)
logger_->d(hash, node.id, "[store %s] Rejecting edition of %s because of storage policy",
logger_->d(hash, node.id, "[store %s] rejecting edition of %s because of storage policy",
hash.toString().c_str(), vc->toString().c_str());
}
}
@@ -2533,11 +2533,11 @@ Dht::onAnnounce(Sp<Node> n,
const auto& type = getType(vc->type);
if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
// if (logger_)
// logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
// logger_->d(hash, node.id, "[store %s] storing %s", hash.toString().c_str(), std::to_string(vc->id).c_str());
storageStore(hash, vc, created, node.getAddr());
} else {
if (logger_)
logger_->d(hash, node.id, "[store %s] Rejecting storage of %s",
logger_->d(hash, node.id, "[store %s] rejecting storage of %s",
hash.toString().c_str(), vc->toString().c_str());
}
}
@@ -2552,15 +2552,15 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
if (not tokenMatch(token, node->getAddr())) {
if (logger_)
logger_->w(hash, node->id, "[node %s] Incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
logger_->w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
}
if (storageRefresh(hash, vid)) {
if (logger_)
logger_->d(hash, node->id, "[store %s] [node %s] Refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
} else {
if (logger_)
logger_->d(hash, node->id, "[store %s] [node %s] Got refresh for unknown value",
logger_->d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
hash.toString().c_str(), node->toString().c_str());
throw DhtProtocolException {DhtProtocolException::NOT_FOUND, DhtProtocolException::STORAGE_NOT_FOUND};
}
@@ -2583,7 +2583,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
for (const auto& node_listeners : st.listeners) {
for (const auto& l : node_listeners.second) {
if (logger_)
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] Sending refresh",
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending refresh",
id.toString().c_str(),
node_listeners.first->toString().c_str());
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
@@ -2608,7 +2608,7 @@ void
Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
{
if (logger_)
logger_->d(sr->id, node->id, "[search %s] [node %s] Got reply to put!",
logger_->d(sr->id, node->id, "[search %s] [node %s] got reply to put!",
sr->id.toString().c_str(), node->toString().c_str());
searchSendGetValues(sr);
sr->checkAnnounced(answer.vid);
+6 -4
View File
@@ -813,8 +813,8 @@ SockAddr
DhtProxyClient::parsePublicAddress(const Json::Value& val)
{
auto public_ip = val.asString();
auto [host, service] = splitPort(public_ip);
auto sa = SockAddr::resolve(host);
auto hostAndService = splitPort(public_ip);
auto sa = SockAddr::resolve(hostAndService.first);
if (sa.empty()) return {};
return sa.front().getMappedIPv4();
}
@@ -1121,10 +1121,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
std::lock_guard<std::mutex> l(requestLock_);
requests_[reqid] = request;
}
request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) {
request->add_on_status_callback([r=std::weak_ptr(request), seconds = this->listenKeepIdle()] (unsigned status_code) {
if(status_code == 200) {
// increase TCP_KEEPIDLE to save power
request->get_connection()->set_keepalive(seconds);
if (auto request = r.lock()) {
request->get_connection()->set_keepalive(seconds);
}
}
});
request->send();
-2
View File
@@ -631,8 +631,6 @@ DhtProxyServer::createRestRouter()
router->http_get("/key/:hash", std::bind(&DhtProxyServer::get, this, _1, _2));
// key.post
router->http_post("/key/:hash", std::bind(&DhtProxyServer::put, this, _1, _2));
router->add_handler(restinio::http_method_options(),
"/key/:hash/listen", std::bind(&DhtProxyServer::options, this, _1, _2));
// key.listen
router->http_get("/key/:hash/listen", std::bind(&DhtProxyServer::listen, this, _1, _2));
#ifdef OPENDHT_PUSH_NOTIFICATIONS
+5 -4
View File
@@ -54,7 +54,7 @@ constexpr const char HTTPS_PROTOCOL[] = "https://";
constexpr const char ORIGIN_PROTOCOL[] = "//";
constexpr unsigned MAX_REDIRECTS {5};
Url::Url(std::string_view url): url(url)
Url::Url(const std::string& url): url(url)
{
size_t addr_begin = 0;
// protocol
@@ -69,9 +69,10 @@ Url::Url(std::string_view url): url(url)
size_t addr_size = url.substr(addr_begin).find("/");
if (addr_size == std::string::npos)
addr_size = url.size() - addr_begin;
auto [h, s] = splitPort(url.substr(addr_begin, addr_size));
host = std::move(h);
service = std::move(s);
auto host_service = splitPort(url.substr(addr_begin, addr_size));
host = host_service.first;
if (!host_service.second.empty())
service = host_service.second;
// target, query and fragment
size_t query_begin = url.find("?");
auto addr_end = addr_begin + addr_size;
+7 -21
View File
@@ -526,13 +526,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find socket", msg->id};
node->received(now, {});
onNewNode(node, 2);
try {
deserializeNodes(*msg, from);
rsocket->on_receive(node, std::move(*msg));
} catch (const DhtProtocolException& e) {
if (logger_)
logger_->w("Can't deserialize nodes %s", e.what());
}
deserializeNodes(*msg, from);
rsocket->on_receive(node, std::move(*msg));
}
else if (msg->type == MessageType::Error or msg->type == MessageType::Reply) {
auto rsocket = node->getSocket(msg->tid);
@@ -595,22 +590,13 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
r.node->authSuccess();
}
r.reply_time = scheduler.time();
try {
deserializeNodes(*msg, from);
r.setDone(std::move(*msg));
} catch (const DhtProtocolException& e) {
if (logger_)
logger_->w("Can't deserialize nodes %s", e.what());
}
deserializeNodes(*msg, from);
r.setDone(std::move(*msg));
break;
} else { /* request socket data */
try {
deserializeNodes(*msg, from);
rsocket->on_receive(node, std::move(*msg));
} catch (const DhtProtocolException& e) {
if (logger_)
logger_->w("Can't deserialize nodes %s", e.what());
}
deserializeNodes(*msg, from);
rsocket->on_receive(node, std::move(*msg));
}
break;
default:
+9 -9
View File
@@ -44,23 +44,23 @@ const HexMap hex_map = {};
static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
std::pair<std::string, std::string>
splitPort(std::string_view s) {
splitPort(const std::string& s) {
if (s.empty())
return {};
if (s[0] == '[') {
std::size_t closure = s.find_first_of(']');
std::size_t found = s.find_last_of(':');
if (closure == std::string_view::npos)
return {std::string(s), ""};
if (found == std::string_view::npos or found < closure)
return {std::string(s.substr(1,closure-1)), ""};
return {std::string(s.substr(1,closure-1)), std::string(s.substr(found+1))};
if (closure == std::string::npos)
return {s, ""};
if (found == std::string::npos or found < closure)
return {s.substr(1,closure-1), ""};
return {s.substr(1,closure-1), s.substr(found+1)};
}
std::size_t found = s.find_last_of(':');
std::size_t first = s.find_first_of(':');
if (found == std::string_view::npos or found != first)
return {std::string(s), ""};
return {std::string(s.substr(0,found)), std::string(s.substr(found+1))};
if (found == std::string::npos or found != first)
return {s, ""};
return {s.substr(0,found), s.substr(found+1)};
}
std::vector<SockAddr>
+1 -1
View File
@@ -123,7 +123,7 @@ DhtRunnerTester::testPutOverride() {
auto val2 = std::make_shared<dht::Value>("hey");
val2->id = 42;
CPPUNIT_ASSERT_EQUAL(val->id, val2->id);
auto val_data = val2->data;
auto val_data = val->data;
std::promise<bool> p1;
std::promise<bool> p2;
node2.put(key, val, [&](bool ok){
-1
View File
@@ -32,7 +32,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
CPPUNIT_TEST(testConstructors);
CPPUNIT_TEST(testGetPut);
CPPUNIT_TEST(testPutDuplicate);
CPPUNIT_TEST(testPutOverride);
CPPUNIT_TEST(testListen);
CPPUNIT_TEST(testListenLotOfBytes);
CPPUNIT_TEST(testIdOps);
+2 -2
View File
@@ -107,8 +107,8 @@ main(int argc, char **argv)
token = dht.listen<dht::ImMessage>(room, [&](dht::ImMessage&& msg) {
if (msg.from != myid)
std::cout << msg.from.toString() << " at " << printTime(msg.date)
<< " (took " << print_duration(std::chrono::system_clock::now() - std::chrono::system_clock::from_time_t(msg.date))
<< ") " << (msg.to == myid ? "ENCRYPTED ":"") << ": " << msg.id << " - " << msg.msg << std::endl;
<< " (took " << print_dt(std::chrono::system_clock::now() - std::chrono::system_clock::from_time_t(msg.date))
<< "s) " << (msg.to == myid ? "ENCRYPTED ":"") << ": " << msg.id << " - " << msg.msg << std::endl;
return true;
});
connected = true;
+5 -14
View File
@@ -23,7 +23,6 @@
#include <unistd.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <inttypes.h>
#include <getopt.h>
#include <readline/readline.h>
@@ -39,10 +38,6 @@ struct listen_context {
dht_op_token* token;
size_t count;
};
struct put_context {
dht_runner* runner;
dht_value* value;
};
bool dht_value_callback(const dht_value* value, bool expired, void* user_data)
{
@@ -72,10 +67,8 @@ void dht_get_done_callback(bool ok, void* user_data)
void dht_put_done_callback(bool ok, void* user_data)
{
struct put_context* ctx = (struct put_context*)user_data;
printf("Put completed (id: %" PRIx64 "): %s\n", dht_value_get_id(ctx->value), ok ? "success !" : "failure :-(");
dht_value_unref(ctx->value);
free(ctx);
dht_runner* runner = (dht_runner*)user_data;
printf("Put completed: %s\n", ok ? "success !" : "failure :-(");
}
void dht_shutdown_callback(void* user_data)
@@ -268,13 +261,11 @@ int main(int argc, char **argv)
else if (!strcmp(cmd, "p")) {
key = parse_key(arg);
dht_value* val = dht_value_new_from_string(value);
struct put_context* ctx = malloc(sizeof(struct put_context));
ctx->runner = runner;
ctx->value = val;
dht_runner_put(runner, &key, val, dht_put_done_callback, ctx, true);
dht_runner_put(runner, &key, val, dht_put_done_callback, runner, true);
dht_value_unref(val);
}
else {
printf("Unknown command: %s\n", cmd);
printf("Unkown command: %s\n", cmd);
}
}
+168 -236
View File
@@ -1,78 +1,190 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
<title>OpenDHT web node</title>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/5.3.3/css/bootstrap.min.css" />
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js" crossorigin="anonymous"></script>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css">
</head>
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
<title>OpenDHT tester</title>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/css/bootstrap.min.css" />
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.css" />
<script src="https://ajax.googleapis.com/ajax/libs/jquery/3.2.1/jquery.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
<script type="text/javascript">"use strict";
let onGet;
let onPut;
let setServer;
const valueGetElement = function(o) {
const d = window.atob(o.data);
return d;
};
$(function() {
let request = undefined;
let server;
const getTools = $("#getTools");
const getBtn = $("#getBtn");
const getDropdown = $("#getDropdown");
const listenBtn = $("#listenBtn").click(function(){onGet('LISTEN');});
const setGetRequest = function() {
getBtn.button('loading');
getStopBtn.appendTo(getTools);
getDropdown.hide();
}
const clearGetRequest = function() {
if (request === undefined)
return;
request.abort();
request = undefined;
getStopBtn.detach();
getDropdown.show();
getBtn.button('reset');
}
const getStopBtn = $("#getStopBtn").detach().click(clearGetRequest);
const putBtn = $("#putBtn");
const result = $("#dhtResult");
const group = $('<ul class="list-group"/>').appendTo(result);
onGet = function (method) {
if (request !== undefined)
return false;
const input = $("#getKey").val();
group.empty();
let lastAppended = 0;
let start = new Date().getTime();
request = new XMLHttpRequest();
request.onreadystatechange = function(event) {
if (this.readyState >= XMLHttpRequest.LOADING) {
if (this.readyState == XMLHttpRequest.DONE) {
clearGetRequest();
}
if (this.status === 200) {
const elements = this.responseText.split("\n");
const elementsLength = elements.length;
const now = new Date().getTime();
for (let i = lastAppended; i < elementsLength; i++) {
const element = elements[i];
if (!element || element.length == 0)
return;
const o = JSON.parse(element);
if (o.expired) {
$('#value'+o.id).slideUp(100, (e) => $(e.target).remove());
} else {
const d = window.atob(o.data);
const delay = Math.max(0, start-now);
$('<li class="list-group-item" id="value'+o.id+'"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
lastAppended = i+1;
start = Math.max(start, now)+25;
}
}
} else if (this.status !== 0) {
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));
}
}
};
request.onerror = function(event) {
clearGetRequest();
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content."));
};
request.open(method, server + input, true);
request.send(null);
setGetRequest();
return false;
};
onPut = function( ) {
const key = $("#getKey").val();
const value = $("#putValue").val();
$.ajax({
url: server + key,
type: 'POST',
data: JSON.stringify({
data:window.btoa(value)
}),
contentType: 'application/json; charset=utf-8',
dataType: 'json',
success: function( result ) {
putBtn.button('reset');
//$('<li class="list-group-item list-group-item-success"/>').append(valueGetElement(result)).appendTo(group.empty());
},
error: function(result) {
putBtn.button('reset');
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text(result.statusText));
}
});
putBtn.button('loading');
return false;
};
const serverValue = $("#serverValue");
const serverStatus = $("#serverStatus");
const serverBtn = $("#serverBtn");
setServer = function(event) {
server = serverValue.val() + '/';
serverStatus.empty();
serverBtn.button('loading');
$.getJSON(server, function(data){
$('<span><b>Node</b> '+data.node_id+'</span>').appendTo(serverStatus).hide().fadeIn();
}).fail(function(error) {
var message = " Cant' access node."
if (serverValue.val().indexOf("https") != -1){
message += "</br></br>Self-signed certificate must be allowed in browser."
}
serverStatus.html("<div class='alert alert-danger' style='margin-bottom: 0px;'>" +
"<span class='glyphicon glyphicon-remove' aria-hidden='true'></span>" +
message + "</div>");
}).always(function(error) {
serverBtn.button('reset');
});
return false;
};
setServer();
});
</script>
</head>
<body>
<div class="container" style="max-width: 880px;">
<header class="m-3 p-3">
<div class="container" style="max-width: 730px;">
<header class="page-header">
<div class="row">
<div class="col-sm-5 d-flex align-items-center">
<h2>OpenDHT web node</h2>
<div class="col-sm-5">
<h1>OpenDHT tester</h1>
</div>
<div class="col-sm-7">
<div class="card card-body bg-light" style="margin-top:10px; margin-bottom:0px;">
<form id="serverForm" class="form-inline" onsubmit="return setServer();"
style="margin-bottom:4px;">
<div class="well well-sm" style="margin-top:10px; margin-bottom:0px;">
<form id="serverForm" class="form-inline" onsubmit="return setServer();" style="margin-bottom:4px;">
<div class="input-group">
<input type="text" class="form-control" id="serverValue" placeholder="Proxy server" value="http://127.0.0.1:8080" />
<button id="serverBtn" type="submit" class="btn btn-outline-secondary">
<div class='spinner-border spinner-border-sm' role='status'
style="display: none;">
<span class='visually-hidden'>Loading...</span>
</div>
<span class="btn-icon"><i class="bi bi-arrow-clockwise"></i></span>
</button>
<input type="text" class="form-control" id="serverValue" placeholder="Proxy server" value="http://127.0.0.1:8080"/>
<span class="input-group-btn">
<button id="serverBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>"><span class="glyphicon glyphicon-refresh" aria-hidden="true"></span></button>
</span>
</div>
</form>
<div id="serverStatus">
<div class='spinner-border spinner-border-sm' role='status'><span
class='visually-hidden'>Loading...</span></div>
</div>
<div id="serverStatus"><i class='fa fa-circle-o-notch fa-spin'></i></div>
</div>
</div>
</div>
</header>
<div class="list-group" id="dhtResult">
<div class="list-group-item list-group-item-light">
<div class="row">
<div class="col-sm-6">
<form class="form-inline" onsubmit="return onGet('');">
<div class="input-group" id="getTools">
<div class="panel panel-default" id="dhtResult">
<div class="panel-heading">
<div class="row">
<div class="col-xs-6">
<form class="form-inline" onsubmit="return onGet('GET');">
<div class="input-group">
<input type="text" class="form-control" id="getKey" placeholder="Key" aria-label="Key" />
<button id="getBtn" class="btn btn-outline-secondary" type="submit">
<div class='spinner-border spinner-border-sm' role='status' style="display: none;">
<span class='visually-hidden'>Loading...</span>
</div>
<span class="btn-text">Get</span>
</button>
<button id="menuBtn" type="button" class="btn btn-outline-secondary dropdown-toggle dropdown-toggle-split" data-bs-toggle="dropdown" aria-expanded="false">
<span class="visually-hidden">Toggle Dropdown</span>
</button>
<ul class="dropdown-menu dropdown-menu-end">
<li><a class="dropdown-item" id="listenBtn" href="#">Listen</a></li>
</ul>
<button id="getStopBtn" class="btn btn-outline-secondary">Stop</button>
<span class="input-group-btn" id="getTools">
<button id="getBtn" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>" type="submit">Get</button>
<button id="getDropdown"type="button" class="btn btn-default dropdown-toggle" data-toggle="dropdown" aria-haspopup="true" aria-expanded="false"> <span class="caret"></span> <span class="sr-only">Toggle Dropdown</span> </button>
<ul class="dropdown-menu">
<li><a id="listenBtn" href="#">Listen</a></li>
</ul>
<button id="getStopBtn" class="btn btn-default" type="submit"><span class="glyphicon glyphicon-remove" aria-hidden="true"></span></button>
</span>
</div>
</form>
</div>
<div class="col-sm-6">
<div class="col-xs-6">
<form class="form-inline" onsubmit="return onPut();">
<div class="input-group">
<input type="text" class="form-control input-group-input" id="putValue"
placeholder="Value" />
<button id="putBtn" type="submit" class="btn btn-outline-secondary">
<div class='spinner-border spinner-border-sm' role='status' style="display: none;">
<span class='visually-hidden'>Loading...</span>
</div>
<span class="btn-text">Put</span>
</button>
<input type="text" class="form-control input-group-input" id="putValue" placeholder="Value" />
<span class="input-group-btn">
<button id="putBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i> Loading">Put</button>
</span>
</div>
</form>
</div>
@@ -80,185 +192,5 @@
</div>
</div>
</div>
<script type="text/javascript">"use strict";
const valueGetElement = (o) => window.atob(o.data);
const getTools = document.getElementById("getTools");
const getBtn = document.getElementById("getBtn");
const menuBtn = document.getElementById("menuBtn");
const getStopBtn = document.getElementById("getStopBtn");
const listenBtn = document.getElementById("listenBtn");
const putBtn = document.getElementById("putBtn");
const group = document.getElementById("dhtResult");
const serverValue = document.getElementById("serverValue");
const serverStatus = document.getElementById("serverStatus");
const serverBtn = document.getElementById("serverBtn");
var request = undefined;
const setGetRequest = () => {
getBtn.disabled = true;
getBtn.querySelector('.spinner-border').style.display = 'inline-block';
getBtn.querySelector('.btn-text').style.display = 'none';
menuBtn.style.display = 'none';
getTools.appendChild(getStopBtn);
};
const clearGetRequest = () => {
if (request === undefined)
return;
request.abort();
request = undefined;
if (getStopBtn.parentNode) {
getStopBtn.parentNode.removeChild(getStopBtn);
}
menuBtn.style.display = '';
getBtn.disabled = false;
getBtn.querySelector('.spinner-border').style.display = 'none';
getBtn.querySelector('.btn-text').style.display = 'inline';
};
const clearResults = () => {
while (group.children.length > 1) {
group.removeChild(group.lastChild);
}
};
var server;
const setServer = (event) => {
server = serverValue.value + '/';
serverStatus.innerHTML = '';
serverBtn.disabled = true;
serverBtn.querySelector('.spinner-border').style.display = 'inline-block';
serverBtn.querySelector('.btn-icon').style.display = 'none';
var xhr = new XMLHttpRequest();
xhr.open('GET', server, true);
xhr.onreadystatechange = () => {
if (xhr.readyState === XMLHttpRequest.DONE) {
serverBtn.disabled = false;
serverBtn.querySelector('.spinner-border').style.display = 'none';
serverBtn.querySelector('.btn-icon').style.display = 'inline';
if (xhr.status === 200) {
const data = JSON.parse(xhr.responseText);
const span = document.createElement('span');
span.innerHTML = '<b>Node</b> ' + data.node_id;
serverStatus.appendChild(span);
} else {
const message = " Can't access node.";
if (serverValue.value.indexOf("https") != -1) {
message += "</br></br>Self-signed certificate must be allowed in browser.";
}
serverStatus.innerHTML = "<div class='alert alert-danger' style='margin-bottom: 0px;'>" +
"<span class='glyphicon glyphicon-remove' aria-hidden='true'></span>" +
message + "</div>";
}
}
};
xhr.send();
return false;
};
setServer();
const onGet = (path) => {
if (request !== undefined)
return false;
const input = document.getElementById("getKey").value;
clearResults()
let lastAppended = 0;
let start = new Date().getTime();
request = new XMLHttpRequest();
request.onreadystatechange = (event) => {
if (request.readyState >= 2) { // LOADING
if (request.status === 200) {
const elements = request.responseText.split("\n");
const elementsLength = elements.length;
const now = new Date().getTime();
for (let i = lastAppended; i < elementsLength; i++) {
const element = elements[i];
if (!element || element.length == 0)
break;
const o = JSON.parse(element);
if (o.expired) {
const expiredElement = document.getElementById('value' + o.id);
if (expiredElement) {
expiredElement.style.display = 'none';
if (expiredElement.parentNode) {
expiredElement.parentNode.removeChild(expiredElement);
}
}
} else {
const d = window.atob(o.data);
const delay = Math.max(0, start - now);
const li = document.createElement('li');
li.className = 'list-group-item';
li.id = 'value' + o.id;
li.innerHTML = valueGetElement(o);
li.style.opacity = 0;
li.style.transition = 'opacity 0.1s ease-in';
group.appendChild(li);
setTimeout(() => {
li.style.opacity = 1;
}, delay);
lastAppended = i + 1;
start = Math.max(start, now) + 25;
}
}
} else if (request.status !== 0) {
clearResults()
const li = document.createElement('li');
li.className = 'list-group-item list-group-item-danger';
li.textContent = "Error loading content: " + request.statusText;
group.appendChild(li);
}
}
if (request.readyState == 4) { // DONE
clearGetRequest();
}
};
request.onerror = (event) => {
clearGetRequest();
clearResults()
const li = document.createElement('li');
li.className = 'list-group-item list-group-item-danger';
li.textContent = "Error loading content.";
group.appendChild(li);
};
request.open('GET', server + 'key/' + input + path, true);
request.send(null);
setGetRequest();
return false;
};
const onPut = () => {
const key = document.getElementById("getKey").value;
const value = document.getElementById("putValue").value;
putBtn.disabled = true;
putBtn.querySelector('.spinner-border').style.display = 'inline-block';
putBtn.querySelector('.btn-text').style.display = 'none';
fetch(server + key, {
method: 'POST',
headers: { 'Content-Type': 'application/json; charset=utf-8' },
body: JSON.stringify({ data: window.btoa(value) })
})
.then(response => {
if (!response.ok)
throw new Error(response.statusText);
})
.catch(error => {
group.innerHTML = '';
const li = document.createElement('li');
li.className = 'list-group-item list-group-item-danger';
li.textContent = error.message;
group.appendChild(li);
})
.finally(() => {
putBtn.disabled = false;
putBtn.querySelector('.spinner-border').style.display = 'none';
putBtn.querySelector('.btn-text').style.display = 'inline';
});
return false;
};
listenBtn.addEventListener('click', () => { onGet('/listen'); });
getStopBtn.parentNode.removeChild(getStopBtn);
getStopBtn.addEventListener('click', clearGetRequest);
</script>
</body>
</html>
</html>