Compare commits

8 Commits

Author SHA1 Message Date
7b52f6d70d fix get mtu (#186)
Reviewed-on: #186
2025-06-26 10:52:36 +03:00
19fe33383a Fix cmake module path (#185)
Если в cmake передаётся кастомная папка с cmake-модулями, например в toolchain файле, то pip перестаёт добавлять ./cmake в путь поиска модулей. Это приводит к ошибке `can't find PIPMacros.cmake`. Не  очень понимаю зачем так было сделано, поэтому просто убрал условие пустого CMAKE_MODULE_PATH.

Reviewed-on: #185
Co-authored-by: andrey.bychkov <andrey@signalmodelling.ru>
Co-committed-by: andrey.bychkov <andrey@signalmodelling.ru>
2025-06-26 10:52:04 +03:00
e57719c118 version 4.7.2
PIThreadPoolLoop reworked, now threads starts on constructor and immediately run after start()
2025-06-07 10:58:11 +03:00
edb077b400 PIThreadPoolLoopNW test 2025-06-03 20:15:00 +03:00
bad51db0c8 remove export 2025-05-18 18:53:28 +03:00
87ab4c4c28 version 4.7.1
PIHTTP::MessageConst code categories
PIHTTPClient can abort on program exit
2025-05-14 19:45:58 +03:00
8f3456a650 fix CurlThreadPool destroy logic 2025-05-14 19:29:07 +03:00
ce1aee1553 PIHTTP::MessageConst code categories
PIHTTPClient migrated to curl_multi_api
2025-05-14 16:21:38 +03:00
11 changed files with 360 additions and 90 deletions

View File

@@ -6,7 +6,7 @@ endif()
project(PIP)
set(PIP_MAJOR 4)
set(PIP_MINOR 7)
set(PIP_REVISION 0)
set(PIP_REVISION 2)
set(PIP_SUFFIX )
set(PIP_COMPANY SHS)
set(PIP_DOMAIN org.SHS)
@@ -45,9 +45,7 @@ ExternalProject_Add(cmake
set(GIT_CMAKE_DIR "${CMAKE_CURRENT_BINARY_DIR}/cmake-src")
endif()
if ("x${CMAKE_MODULE_PATH}" STREQUAL "x")
set(CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
endif()
list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
if (NOT "x${GIT_CMAKE_DIR}" STREQUAL "x")
list(APPEND CMAKE_MODULE_PATH "${GIT_CMAKE_DIR}")
endif()
@@ -117,7 +115,6 @@ macro(pip_module NAME LIBS LABEL INCLUDES SOURCES MSG)
file(GLOB_RECURSE ASRC "${SOURCES}/*.cpp" "${SOURCES}/*.c")
list(APPEND CPPS ${ASRC})
endif()
#message("${NAME} HS = ${HS}")
list(APPEND HDRS ${HS})
list(APPEND PHDRS ${PHS})

View File

@@ -7,7 +7,9 @@
#include <curl/curl.h>
CurlThreadPool::CurlThreadPool() {
piForTimes(10) {
curl_global_init(CURL_GLOBAL_DEFAULT);
const int count = 10;
for (int i = 0; i < count; ++i) {
auto * t = new PIThread([this]() { threadFunc(); }, true);
threads << t;
}
@@ -15,27 +17,10 @@ CurlThreadPool::CurlThreadPool() {
CurlThreadPool::~CurlThreadPool() {
exiting = true;
for (auto * t: threads)
t->stop();
{
auto cr = clients.getRef();
for (auto c: *cr)
c->abort();
}
sem.release(threads.size());
for (auto * t: threads) {
t->waitForFinish();
t->setDebug(false);
t->terminate();
}
piDeleteAllAndClear(threads);
{
auto cr = clients.getRef();
for (auto c: *cr)
delete c;
}
destroy();
// piCout << "~CurlThreadPool cleanup ...";
curl_global_cleanup();
// piCout << "~CurlThreadPool cleanup ok";
}
@@ -59,7 +44,9 @@ void CurlThreadPool::threadFunc() {
void CurlThreadPool::procClient(PIHTTPClient * c) {
{ clients_in_proc.getRef()->append(c); }
if (c->init()) c->perform();
{ clients_in_proc.getRef()->removeOne(c); }
delete c;
}
@@ -75,3 +62,35 @@ CurlThreadPool * CurlThreadPool::instance() {
static CurlThreadPool ret;
return &ret;
}
void CurlThreadPool::destroy() {
// piCout << "~CurlThreadPool";
exiting = true;
for (auto * t: threads)
t->stop();
{
auto cr = clients.getRef();
for (auto c: *cr)
c->abort();
}
{
auto cr = clients_in_proc.getRef();
for (auto c: *cr)
c->abort();
}
// piCout << "~CurlThreadPool release ...";
sem.release(threads.size());
for (auto * t: threads) {
t->waitForFinish();
t->setDebug(false);
t->terminate();
}
// piCout << "~CurlThreadPool delete ...";
piDeleteAllAndClear(threads);
{
auto cr = clients.getRef();
for (auto c: *cr)
delete c;
}
// piCout << "~CurlThreadPool ok";
}

View File

@@ -18,10 +18,12 @@ private:
CurlThreadPool();
~CurlThreadPool();
void destroy();
void threadFunc();
void procClient(PIHTTPClient * c);
PIProtectedVariable<PIQueue<PIHTTPClient *>> clients;
PIProtectedVariable<PIVector<PIHTTPClient *>> clients_in_proc;
PISemaphore sem;
PIVector<PIThread *> threads;
std::atomic_bool exiting = {false};

View File

@@ -22,8 +22,30 @@ int debug_callback(CURL * handle, curl_infotype type, char * data, size_t size,
PRIVATE_DEFINITION_START(PIHTTPClient)
CURLM * multi = nullptr;
CURL * handle = nullptr;
curl_slist * header_list = nullptr;
bool isInit() const {
return multi && handle;
}
bool init() {
multi = curl_multi_init();
handle = curl_easy_init();
if (!multi || !handle) {
destroy();
return false;
}
return true;
}
void destroy() {
if (multi && handle) curl_multi_remove_handle(multi, handle);
if (header_list) curl_slist_free_all(header_list);
if (handle) curl_easy_cleanup(handle);
if (multi) curl_multi_cleanup(multi);
multi = nullptr;
handle = nullptr;
header_list = nullptr;
}
PRIVATE_DEFINITION_END(PIHTTPClient)
@@ -35,8 +57,8 @@ PIHTTPClient::~PIHTTPClient() {}
bool PIHTTPClient::init() {
if (is_cancel) return false;
PRIVATE->handle = curl_easy_init();
if (!PRIVATE->handle) return false;
CurlThreadPool::instance();
if (!PRIVATE->init()) return false;
auto ait = request.arguments().makeIterator();
while (ait.next()) {
if (!url.contains('?'))
@@ -81,30 +103,64 @@ bool PIHTTPClient::init() {
void PIHTTPClient::perform() {
if (!PRIVATE->handle) return;
if (!PRIVATE->isInit()) return;
if (!is_cancel) {
// piCout << "perform ...";
PITimeMeasurer tm;
CURLcode res = curl_easy_perform(PRIVATE->handle);
// piCout << "done" << res << "in" << tm.elapsed_m() << ", bytes" << buffer_out.size();
if (res == CURLE_OK) {
reply.setBody(std::move(buffer_out));
if (on_finish) on_finish(reply);
} else {
if (res == CURLE_ABORTED_BY_CALLBACK || is_cancel) {
// piCerr << "curl_easy_perform() failed:" << curl_easy_strerror(res);
if (on_abort) on_abort(reply);
} else {
last_error = curl_easy_strerror(res);
if (on_error) on_error(reply);
// CURLcode res = curl_easy_perform(PRIVATE->handle);
curl_multi_add_handle(PRIVATE->multi, PRIVATE->handle);
int running_cnt = 1;
do {
CURLMcode mc = curl_multi_perform(PRIVATE->multi, &running_cnt);
// piCout << "curl_multi_perform" << mc << running_cnt;
if (!mc) /* wait for activity, timeout or "nothing" */
mc = curl_multi_poll(PRIVATE->multi, nullptr, 0, 50, nullptr);
if (mc || is_cancel) {
// piCout << "curl_multi_poll() failed";
break;
}
} while (running_cnt > 0 && !is_cancel);
if (is_cancel) {
if (on_abort) on_abort(reply);
PRIVATE->destroy();
return;
}
CURLMsg * m = nullptr;
CURLcode res = (CURLcode)-1;
do {
int msgq = 0;
m = curl_multi_info_read(PRIVATE->multi, &msgq);
if (m && (m->msg == CURLMSG_DONE)) {
res = m->data.result;
if (res == CURLE_OK) {
reply.setBody(std::move(buffer_out));
if (on_finish) on_finish(reply);
} else {
if (res == CURLE_ABORTED_BY_CALLBACK || is_cancel) {
// piCerr << "curl_easy_perform() failed:" << curl_easy_strerror(res);
if (on_abort) on_abort(reply);
} else {
last_error = curl_easy_strerror(res);
if (on_error) on_error(reply);
}
}
break;
}
} while (m && !is_cancel);
if (res == (CURLcode)-1) {
last_error = "CURL error";
if (on_error) on_error(reply);
}
// piCout << "done" << (int)res << "in" << tm.elapsed_m() << ", bytes" << buffer_out.size();
}
// piCout << last_error;
if (PRIVATE->header_list) curl_slist_free_all(PRIVATE->header_list);
curl_easy_cleanup(PRIVATE->handle);
PRIVATE->handle = nullptr;
PRIVATE->header_list = nullptr;
PRIVATE->destroy();
}
@@ -224,6 +280,7 @@ void PIHTTPClient::start() {
void PIHTTPClient::abort() {
is_cancel = true;
if (PRIVATE->isInit()) curl_multi_wakeup(PRIVATE->multi);
}

View File

@@ -18,7 +18,7 @@ enum class Method {
};
enum class Code {
Unknown,
Unknown = -1,
Continue = 100,
SwitchingProtocols = 101,
Processing = 102,

View File

@@ -18,6 +18,35 @@ const char * PIHTTP::methodName(Method m) {
}
// PIHTTP::MessageConst
bool PIHTTP::MessageConst::isCodeInformational() const {
return (int)code() >= 100 && (int)code() < 200;
}
bool PIHTTP::MessageConst::isCodeSuccess() const {
return (int)code() >= 200 && (int)code() < 300;
}
bool PIHTTP::MessageConst::isCodeRedirection() const {
return (int)code() >= 300 && (int)code() < 400;
}
bool PIHTTP::MessageConst::isCodeClientError() const {
return (int)code() >= 400 && (int)code() < 500;
}
bool PIHTTP::MessageConst::isCodeServerError() const {
return (int)code() >= 500 && (int)code() < 600;
}
// PIHTTP::MessageMutable
PIHTTP::MessageMutable & PIHTTP::MessageMutable::addHeader(const PIString & header, const PIString & value) {
m_headers[header] = value;
return *this;

View File

@@ -13,6 +13,12 @@ class PIP_EXPORT MessageConst {
public:
PIHTTP::Method method() const { return m_method; }
PIHTTP::Code code() const { return m_code; }
bool isCodeInformational() const;
bool isCodeSuccess() const;
bool isCodeRedirection() const;
bool isCodeClientError() const;
bool isCodeServerError() const;
bool isCodeError() const { return isCodeClientError() || isCodeServerError(); }
const PIString & path() const { return m_path; }
PIStringList pathList() const { return m_path.split('/').removeAll({}); }
const PIByteArray & body() const { return m_body; }

View File

@@ -1243,9 +1243,12 @@ PIEthernet::InterfaceList PIEthernet::interfaces() {
# else
if (s != -1) {
struct ifreq ir;
strcpy(ir.ifr_name, cif->ifa_name);
memset(&ir, 0, sizeof(ir));
strncpy(ir.ifr_name, cif->ifa_name, sizeof(ir.ifr_name));
if (ioctl(s, SIOCGIFHWADDR, &ir) == 0) {
ci.mac = macFromBytes(PIByteArray(ir.ifr_hwaddr.sa_data, 6));
}
if (ioctl(s, SIOCGIFMTU, &ir) == 0) {
ci.mtu = ir.ifr_mtu;
}
}

View File

@@ -106,16 +106,29 @@
PIThreadPoolLoop::PIThreadPoolLoop(int thread_cnt) {
if (thread_cnt <= 0) thread_cnt = piMaxi(1, PISystemInfo::instance()->processorsCount);
piForTimes(thread_cnt) {
auto * t = new PIThread();
auto * t = new PIThread([this]() {
while (true) {
sem_exec.acquire();
if (is_destroy) return;
int cc = counter.fetch_add(1);
func(cc);
sem_done.release();
}
});
threads << t;
}
for (auto * t: threads)
t->start();
// piCout << "PIThreadPoolLoop" << proc_cnt << "threads";
}
PIThreadPoolLoop::~PIThreadPoolLoop() {
for (auto * t: threads) {
is_destroy = true;
for (auto * t: threads)
t->stop();
sem_exec.release(threads.size());
for (auto * t: threads) {
if (!t->waitForFinish(100_ms)) t->terminate();
delete t;
}
@@ -127,20 +140,19 @@ void PIThreadPoolLoop::setFunction(std::function<void(int)> f) {
}
void PIThreadPoolLoop::wait() {
// piCout << "wait" << wait_count;
if (wait_count <= 0) return;
sem_done.acquire(wait_count);
wait_count = 0;
// piCout << "wait done";
}
void PIThreadPoolLoop::start(int index_start, int index_count) {
counter = index_start;
int end = index_start + index_count;
for (auto * t: threads)
t->start([this, end, t]() {
while (1) {
int cc = counter.fetch_add(1);
if (cc >= end) {
t->stop();
return;
}
func(cc);
}
});
counter = index_start;
wait_count = index_count;
sem_exec.release(index_count);
}
@@ -154,9 +166,3 @@ void PIThreadPoolLoop::exec(int index_start, int index_count, std::function<void
setFunction(f);
exec(index_start, index_count);
}
void PIThreadPoolLoop::wait() {
for (auto * t: threads)
t->waitForFinish();
}

View File

@@ -26,6 +26,7 @@
#ifndef PITHREADPOOLLOOP_H
#define PITHREADPOOLLOOP_H
#include "pisemaphore.h"
#include "pivector.h"
class PIThread;
@@ -83,7 +84,9 @@ public:
private:
PIVector<PIThread *> threads;
std::function<void(int)> func;
std::atomic_int counter = {0};
PISemaphore sem_exec, sem_done;
std::atomic_bool is_destroy = {false};
std::atomic_int counter = {0}, wait_count = {0};
};

190
main.cpp
View File

@@ -1,24 +1,159 @@
#include "libs/http_client/curl_thread_pool_p.h"
#include "pidigest.h"
#include "pihttpclient.h"
#include "pip.h"
using namespace PICoutManipulators;
using namespace PIHTTP;
PIKbdListener kbd;
const char * pageTitle = "<!DOCTYPE html>"
"<html>"
"<body>"
"<h1>Title</h1>"
"</body>"
"</html>";
class PIThreadPoolLoopNW {
public:
PIThreadPoolLoopNW(int thread_cnt = -1) {
if (thread_cnt <= 0) thread_cnt = piMaxi(1, PISystemInfo::instance()->processorsCount);
piForTimes(thread_cnt) {
auto * t = new PIThread([this]() {
while (true) {
sem_exec.acquire();
if (is_destroy) return;
int cc = counter.fetch_add(1);
func(cc);
sem_done.release();
}
});
threads << t;
}
for (auto * t: threads)
t->start();
// piCout << "PIThreadPoolLoop" << proc_cnt << "threads";
}
virtual ~PIThreadPoolLoopNW() {
is_destroy = true;
for (auto * t: threads)
t->stop();
sem_exec.release(threads.size());
for (auto * t: threads) {
if (!t->waitForFinish(100_ms)) t->terminate();
delete t;
}
}
void setFunction(std::function<void(int)> f) { func = f; }
void wait() {
// piCout << "wait" << wait_count;
if (wait_count <= 0) return;
sem_done.acquire(wait_count);
wait_count = 0;
// piCout << "wait done";
}
void start(int index_start, int index_count) {
counter = index_start;
wait_count = index_count;
sem_exec.release(index_count);
}
void exec(int index_start, int index_count) {
start(index_start, index_count);
wait();
}
void exec(int index_start, int index_count, std::function<void(int)> f) {
setFunction(f);
exec(index_start, index_count);
}
private:
PIVector<PIThread *> threads;
std::function<void(int)> func;
PISemaphore sem_exec, sem_done;
std::atomic_bool is_destroy = {false};
std::atomic_int counter = {0}, wait_count = {0};
};
// PIKbdListener kbd;
PIVector<int> vec;
int main(int argc, char * argv[]) {
PIHTTPServer server;
vec.resize(16);
vec.fill([](int i) { return i; });
piCout << vec;
PIThreadPoolLoop tpl(8);
tpl.setFunction([](int i) { vec[i]++; });
const int count = 10000;
PITimeMeasurer tm;
piForTimes(count) {
tpl.exec(0, 16);
}
// tpl.exec(0, 16);
auto el = tm.elapsed().toMilliseconds();
piCout << "el" << el << "ms," << (el / count * 1000) << "us per round";
// tpl.wait();
piCout << vec;
return 0;
/*piForTimes(10) {
PIThread t;
t.setName("thread____");
t.startOnce([]() {
// piCout << "thread";
piMSleep(2.);
});
PITimeMeasurer tm;
t.stopAndWait();
auto el = tm.elapsed();
piCout << el.toMilliseconds();
}
return 0;*/
/*auto src = PIByteArray::fromAscii("The quick brown fox jumps over the lazy dog");
auto key = PIByteArray::fromAscii("key");
PIStringList tnl;
int max_size = 0;
for (int t = 0; t < (int)PIDigest::Type::C ount; ++t) {
tnl << PIDigest::typeName((PIDigest::Type)t);
max_size = piMaxi(max_size, tnl.back().size_s());
}
PIByteArray hs;
piCout << PIString::fromAscii(src);
for (int t = 0; t < (int)PIDigest::Type::Count; ++t) {
hs = PIDigest::calculate(src, (PIDigest::Type)t);
piCout << tnl[t].expandLeftTo(max_size, ' ') << "->" << hs.toHex();
}
for (int t = 0; t < (int)PIDigest::Type::Count; ++t) {
const int bench_count = 100000;
PITimeMeasurer tm;
piForTimes(bench_count) {
hs = PIDigest::calculate(src, (PIDigest::Type)t);
}
auto el = tm.elapsed();
piCout << tnl[t].expandLeftTo(max_size, ' ') << "time" << el.toMilliseconds();
}
// src.clear();
// crypto_hash_sha512(sout.data(), src.data(), src.size());
// piCout << "sod:" << sout.toHex();
// piCout << "512:" << sha5xx(src, initial_512, 64).toHex();
return 0;*/
/*PIHTTPServer server;
server.listen({"127.0.0.1:7777"});
server.setBasicAuthRealm("pip");
server.setBasicAuthEnabled(true);
server.setBasicAuthCallback([](const PIString & u, const PIString & p) -> bool {
piCout << "basic auth" << u << p;
return (u == "u" && p == "p");
// server.setBasicAuthRealm("pip");
// server.setBasicAuthEnabled(true);
// server.setBasicAuthCallback([](const PIString & u, const PIString & p) -> bool {
// piCout << "basic auth" << u << p;
// return (u == "u" && p == "p");
// });
server.registerPath("sendMessage", Method::Post, [](const PIHTTP::MessageConst & msg) -> PIHTTP::MessageMutable {
return MessageMutable().setCode(Code::Accepted);
});
server.registerUnhandled([](const PIHTTP::MessageConst & msg) -> PIHTTP::MessageMutable {
PIHTTP::MessageMutable ret;
@@ -29,10 +164,13 @@ int main(int argc, char * argv[]) {
.arg(PIString::fromUTF8(msg.body()));
ret.setCode(PIHTTP::Code::BadRequest);
ret.setBody(PIByteArray::fromAscii("hello client! 0123456789"));
piSleep(5.);
return ret;
});
kbd.waitForFinish();
return 0;*/
PIHTTP::MessageMutable req;
/*PIHTTP::MessageMutable req;
req.setBody(PIByteArray::fromAscii("hello server!")).addArgument("a0", "val.0").addArgument("a~r1", "знач,1"_u8);
auto * c = PIHTTPClient::create("http://u:p@127.0.0.1:7777/api", PIHTTP::Method::Get, req);
c->onFinish([](PIHTTP::MessageConst msg) {
@@ -40,8 +178,8 @@ int main(int argc, char * argv[]) {
.arg(PIHTTP::methodName(msg.method()))
.arg(piStringify(msg.arguments()))
.arg(
PIStringList(msg.headers().map<PIString>([](PIString k, PIString v) { return k + " = " + v; })).join("\n\t\t "))
.arg(PIString::fromUTF8(msg.body()));
PIStringList(msg.headers().map<PIString>([](PIString k, PIString v) { return k + " = " + v; })).join("\n\t\t
")) .arg(PIString::fromUTF8(msg.body()));
})
->onError([c](PIHTTP::MessageConst r) {
piCout << "error" << (int)r.code();
@@ -51,15 +189,25 @@ int main(int argc, char * argv[]) {
piCout << "abort" << (int)r.code();
piCout << "msg" << c->lastError();
})
->start();*/
auto * c = PIHTTPClient::create(
PIString("127.0.0.1:7777/%1").arg("sendMessag"),
Method::Post,
MessageMutable().addHeader(Header::ContentType, "application/json").setBody(PIByteArray::fromAscii("{hello}")));
c->onFinish([](const PIHTTP::MessageConst & msg) { piCout << "message finish" << (int)msg.code() << PIString::fromUTF8(msg.body()); })
->onError([c](const PIHTTP::MessageConst & msg) { piCout << "message error" << c->lastError(); })
->onAbort([c](const PIHTTP::MessageConst & msg) { piCout << "aborted"; })
->start();
piMSleep(500);
piMSleep(1000);
// CurlThreadPool::instance()->destroy();
// kbd.enableExitCapture();
// WAIT_FOR_EXIT
kbd.stopAndWait();
// kbd.stopAndWait();
server.stop();
// c->abort();
// server.stop();
c->abort();
piMSleep(10);
return 0;
@@ -122,7 +270,7 @@ int main(int argc, char * argv[]) {
piCout << " body" << r.body;
piCout << "";
rep.setBody(PIByteArray::fromAscii("[{\"value1\": true, \"value2\": \"ыекштп\"}]"));
return rep;
return rep;
});*/
/*piCout << "start" << server.isListen();