SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

PIThreadPoolLoopNW test

Browse Source
This commit is contained in:
Пелипенко Иван
2025-06-03 20:15:00 +03:00
parent bad51db0c8
commit edb077b400
1 changed files with 169 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
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;