fix get mtu (#186)

Reviewed-on: #186

CMakeLists.txt

@@ -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")
+list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
-	set(CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
-endif()
 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})
 

libs/http_client/curl_thread_pool_p.cpp

@@ -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;
+	destroy();
-	for (auto * t: threads)
+	// piCout << "~CurlThreadPool cleanup ...";
-		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;
-	}
 	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";
+}

libs/http_client/curl_thread_pool_p.h

@@ -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};

libs/http_client/pihttpclient.cpp

@@ -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();
+	CurlThreadPool::instance();
-	if (!PRIVATE->handle) return false;
+	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);
+		// CURLcode res    = curl_easy_perform(PRIVATE->handle);
-		// piCout << "done" << res << "in" << tm.elapsed_m() << ", bytes" << buffer_out.size();
+
-		if (res == CURLE_OK) {
+		curl_multi_add_handle(PRIVATE->multi, PRIVATE->handle);
-			reply.setBody(std::move(buffer_out));
+		int running_cnt = 1;
-			if (on_finish) on_finish(reply);
+		do {
-		} else {
+			CURLMcode mc = curl_multi_perform(PRIVATE->multi, &running_cnt);
-			if (res == CURLE_ABORTED_BY_CALLBACK || is_cancel) {
+			// piCout << "curl_multi_perform" << mc << running_cnt;
-				// piCerr << "curl_easy_perform() failed:" << curl_easy_strerror(res);
+			if (!mc) /* wait for activity, timeout or "nothing" */
-				if (on_abort) on_abort(reply);
+				mc = curl_multi_poll(PRIVATE->multi, nullptr, 0, 50, nullptr);
-			} else {
+			if (mc || is_cancel) {
-				last_error = curl_easy_strerror(res);
+				// piCout << "curl_multi_poll() failed";
-				if (on_error) on_error(reply);
+				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);
+	PRIVATE->destroy();
-	curl_easy_cleanup(PRIVATE->handle);
-	PRIVATE->handle      = nullptr;
-	PRIVATE->header_list = nullptr;
 }
 
 
@@ -224,6 +280,7 @@ void PIHTTPClient::start() {
 
 void PIHTTPClient::abort() {
 	is_cancel = true;
+	if (PRIVATE->isInit()) curl_multi_wakeup(PRIVATE->multi);
 }
 
 

libs/main/http_common/pihttpconstants.h

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

libs/main/http_common/pihttptypes.cpp

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

libs/main/http_common/pihttptypes.h

@@ -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; }

libs/main/io_devices/piethernet.cpp

@@ -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;
 				}
 			}

libs/main/thread/pithreadpoolloop.cpp

@@ -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;
+	counter    = index_start;
-	int end = index_start + index_count;
+	wait_count = index_count;
-	for (auto * t: threads)
+	sem_exec.release(index_count);
-		t->start([this, end, t]() {
-			while (1) {
-				int cc = counter.fetch_add(1);
-				if (cc >= end) {
-					t->stop();
-					return;
-				}
-				func(cc);
-			}
-		});
 }
 
 
@@ -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();
-}

libs/main/thread/pithreadpoolloop.h

@@ -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};
 };
 
 

main.cpp

@@ -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>"
+class PIThreadPoolLoopNW {
-						 "<html>"
+public:
-						 "<body>"
+	PIThreadPoolLoopNW(int thread_cnt = -1) {
-						 "<h1>Title</h1>"
+		if (thread_cnt <= 0) thread_cnt = piMaxi(1, PISystemInfo::instance()->processorsCount);
-						 "</body>"
+		piForTimes(thread_cnt) {
-						 "</html>";
+			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.setBasicAuthRealm("pip");
-	server.setBasicAuthEnabled(true);
+	// server.setBasicAuthEnabled(true);
-	server.setBasicAuthCallback([](const PIString & u, const PIString & p) -> bool {
+	// server.setBasicAuthCallback([](const PIString & u, const PIString & p) -> bool {
-		piCout << "basic auth" << u << p;
+	// 	piCout << "basic auth" << u << p;
-		return (u == "u" && p == "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  "))
+						   PIStringList(msg.headers().map<PIString>([](PIString k, PIString v) { return k + " = " + v; })).join("\n\t\t
-					   .arg(PIString::fromUTF8(msg.body()));
+	")) .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();
+	// server.stop();
-	// c->abort();
+	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();