some modifications

Author: fantasy-peak

CMakeLists.txt

@@ -1,5 +1,7 @@
 cmake_minimum_required(VERSION 3.15)
 
+project(example)
+
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_EXTENSIONS FALSE)
 
@@ -32,4 +34,4 @@ foreach(mainfile IN LISTS mains)
 	add_executable(${mainname} ${mainfile})
 	target_include_directories(${mainname} PUBLIC ${DEPS_DEPLOY_DIR}/spdlog-src/include/)
 	target_link_libraries(spdlog)
-endforeach()
+endforeach()

README.md

@@ -11,14 +11,10 @@ fantasy::Reactor reactor;
 reactor.run();
 
 // It will run on the reactor thread, do not block the current thread
-reactor.callLater([&] {
-    printf("task");
-});
+reactor.callLater([&] { spdlog::info("task"); });
 
 // It will run on the reactor thread, block the current thread
-reactor.callNow([&] {
-    printf("task");
-});
+reactor.callNow([&] { spdlog::info("task"); });
 
 ```
 
@@ -28,24 +24,20 @@ fantasy::Reactor reactor;
 reactor.run();
 
 // It will run in one second
-reactor.callAt(std::chrono::system_clock::now() + std::chrono::seconds(1), [] {
-    printf("callAt");
-});
+reactor.callAt(std::chrono::system_clock::now() + std::chrono::seconds(1), [] { spdlog::info("callAt"); });
 
 // It will run in five second
-reactor.callAfter(std::chrono::seconds(5), [] {
-    printf("callAfter");
-});
+reactor.callAfter(std::chrono::seconds(5), [] { spdlog::info("callAfter"); });
 
 // Run every three seconds
 reactor.callEvery(std::chrono::seconds(3), [] {
-    printf("callEvery");
+    spdlog::info("callEvery");
     return fantasy::Reactor::CallStatus::Ok;
 });
 
 // Run every day 05:30:00
 auto id = reactor.callEveryDay(fantasy::Time{5, 30, 0, 0}, [] {
-    printf("callEveryDay");
+    spdlog::info("callEveryDay");
     return fantasy::Reactor::CallStatus::Ok;
 });
 
@@ -57,7 +49,23 @@ reactor.cancel(id);
 ```c++
 fantasy::Reactor reactor;
 reactor.run();
-...
+std::string recv_buffer;
+int servfd;
+struct sockaddr_in servaddr, cliaddr;
+if ((servfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+    spdlog::info("create socket error!");
+    exit(1);
+}
+bzero(&servaddr, sizeof(servaddr));
+servaddr.sin_family = AF_INET;
+servaddr.sin_port = htons(SERVER_PORT);
+servaddr.sin_addr.s_addr = htons(INADDR_ANY);
+int opt = 1;
+setsockopt(servfd, SOL_SOCKET, SO_REUSEADDR, (const void*)&opt, sizeof(opt));
+if (bind(servfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0) {
+    spdlog::info("bind to port {} failure!", SERVER_PORT);
+    exit(1);
+}
 if (listen(servfd, LENGTH_OF_LISTEN_QUEUE) < 0) {
     spdlog::info("call listen failure!");
     exit(1);
@@ -88,9 +96,12 @@ reactor.callOnRead(servfd, [&](int fd, const std::weak_ptr<fantasy::Reactor::Cha
         return fantasy::Reactor::CallStatus::Ok;
     });
     reactor.callOnWrite(clifd, [&](int fd, const std::weak_ptr<fantasy::Reactor::Channel>& channel_ptr) {
+        if (recv_buffer.empty())
+            return fantasy::Reactor::CallStatus::Ok;
         spdlog::info("callOnWrite");
         char buffer[BUFFER_SIZE] = {};
         memcpy(buffer, recv_buffer.c_str(), recv_buffer.size());
+        recv_buffer.clear();
         spdlog::info("buffer: {}", buffer);
         auto n = write(fd, buffer, strlen(buffer));
         if (n < 0) {

example/echo_server.cpp

@@ -61,9 +61,12 @@ int main() {
 			return fantasy::Reactor::CallStatus::Ok;
 		});
 		reactor.callOnWrite(clifd, [&](int fd, const std::weak_ptr<fantasy::Reactor::Channel>& channel_ptr) {
+			if (recv_buffer.empty())
+				return fantasy::Reactor::CallStatus::Ok;
 			spdlog::info("callOnWrite");
 			char buffer[BUFFER_SIZE] = {};
 			memcpy(buffer, recv_buffer.c_str(), recv_buffer.size());
+			recv_buffer.clear();
 			spdlog::info("buffer: {}", buffer);
 			auto n = write(fd, buffer, strlen(buffer));
 			if (n < 0) {

example/test_reactor.cpp

@@ -47,7 +47,8 @@ int main() {
 		std::this_thread::sleep_for(std::chrono::seconds(3));
 	});
 	spdlog::info("end callNow");
-	while (true)
-		std::this_thread::sleep_for(std::chrono::seconds(1));
+	while (true) {
+		std::this_thread::sleep_for(std::chrono::seconds(2));
+	}
 	return 0;
 }

reactor.hpp

@@ -551,7 +551,7 @@ class Reactor : public NonCopyable {
 		}
 
 		void cancel(const TimerId& timer_id) {
-			m_reactor.callNow([=] {
+			m_reactor.callNow([&] {
 				auto i = std::find_if(m_timed_callbacks.begin(), m_timed_callbacks.end(), [&](auto& callback) {
 					return callback.second->id() == timer_id;
 				});
@@ -584,7 +584,6 @@ class Reactor : public NonCopyable {
 		});
 		if (m_thread.joinable())
 			m_thread.join();
-		m_wakeup_channel_ptr->disableAll();
 	}
 
 	auto size() const {
@@ -662,19 +661,19 @@ class Reactor : public NonCopyable {
 			return p.second->fd() == fd;
 		});
 		if (it != m_work_channels.end())
-			return std::pair(it->first, it->second);
+			return it;
 		auto channel_ptr = std::make_shared<Channel>(*this, fd);
 		CallId call_id;
-		m_work_channels.emplace(call_id, channel_ptr);
-		return std::pair(std::move(call_id), std::move(channel_ptr));
+		auto ret = m_work_channels.emplace(std::move(call_id), std::move(channel_ptr));
+		return ret.first;
 	}
 
 public:
 	CallId callOnRead(int fd, IoCall&& io_call, bool is_enable_reading = true) {
 		return callNow([&] {
-			auto [call_id, channel_ptr] = get(fd);
+			auto& [call_id, channel_ptr] = *get(fd);
 			std::weak_ptr<Channel> weak_channel_ptr = channel_ptr;
-			channel_ptr->setReadCallback([this, weak_channel_ptr = std::move(weak_channel_ptr), fd, call_id, io_call = std::move(io_call)] {
+			channel_ptr->setReadCallback([this, weak_channel_ptr = std::move(weak_channel_ptr), call_id, fd, io_call = std::move(io_call)] {
 				auto call_status = io_call(fd, weak_channel_ptr);
 				if (call_status == fantasy::Reactor::CallStatus::Remove)
 					cancel(call_id);
@@ -689,27 +688,9 @@ class Reactor : public NonCopyable {
 		return callOnRead(fd, IoCall(call));
 	}
 
-	void callOnEnableReading(const CallId& call_id) {
-		callNow([&] {
-			auto it = m_work_channels.find(call_id);
-			if (it == m_work_channels.end())
-				return;
-			it->second->enableReading();
-		});
-	}
-
-	void callOnEnableWriting(const CallId& call_id) {
-		callNow([&] {
-			auto it = m_work_channels.find(call_id);
-			if (it == m_work_channels.end())
-				return;
-			it->second->enableWriting();
-		});
-	}
-
 	CallId callOnWrite(int fd, IoCall&& io_call, bool is_enable_writing = false) {
 		return callNow([&] {
-			auto [call_id, channel_ptr] = get(fd);
+			auto& [call_id, channel_ptr] = *get(fd);
 			std::weak_ptr<Channel> weak_channel_ptr = channel_ptr;
 			channel_ptr->setWriteCallback([this, weak_channel_ptr = std::move(weak_channel_ptr), fd, call_id, io_call = std::move(io_call)] {
 				auto call_status = io_call(fd, weak_channel_ptr);
@@ -728,7 +709,7 @@ class Reactor : public NonCopyable {
 
 	CallId callOnClose(int fd, IoCall&& io_call) {
 		return callNow([&] {
-			auto [call_id, channel_ptr] = get(fd);
+			auto& [call_id, channel_ptr] = *get(fd);
 			std::weak_ptr<Channel> weak_channel_ptr = channel_ptr;
 			channel_ptr->setCloseCallback([this, weak_channel_ptr = std::move(weak_channel_ptr), fd, call_id, io_call = std::move(io_call)] {
 				auto call_status = io_call(fd, weak_channel_ptr);
@@ -745,7 +726,7 @@ class Reactor : public NonCopyable {
 
 	CallId callOnError(int fd, IoCall&& io_call) {
 		return callNow([&] {
-			auto [call_id, channel_ptr] = get(fd);
+			auto& [call_id, channel_ptr] = *get(fd);
 			std::weak_ptr<Channel> weak_channel_ptr = channel_ptr;
 			channel_ptr->setErrorCallback([this, weak_channel_ptr = std::move(weak_channel_ptr), fd, call_id, io_call = std::move(io_call)] {
 				auto call_status = io_call(fd, weak_channel_ptr);
@@ -763,7 +744,7 @@ class Reactor : public NonCopyable {
 	template <typename T>
 	void cancel(const T& id) {
 		if constexpr (std::is_same_v<T, CallId>) {
-			callNow([this, id] {
+			callNow([&] {
 				if (auto it = m_work_channels.find(id); it != m_work_channels.end()) {
 					it->second->disableAll();
 					m_release_channel.emplace_back(std::move(it->second));
@@ -776,12 +757,10 @@ class Reactor : public NonCopyable {
 	}
 
 	void run() {
-		m_timer_queue_ptr = std::make_unique<TimerManager>(*this);
-		m_wakeup_channel_ptr = std::make_unique<Channel>(*this, m_wakeup_fd_ptr[0]);
-		m_wakeup_channel_ptr->setReadCallback(std::bind(&Reactor::wakeupRead, this));
-		m_wakeup_channel_ptr->enableReading();
 		m_thread = std::thread([&] {
 			m_thread_id = std::this_thread::get_id();
+			m_timer_queue_ptr = std::make_unique<TimerManager>(*this);
+			auto call_id = callOnRead(m_wakeup_fd_ptr[0], [&](int fd, const std::weak_ptr<Channel>&) { wakeupRead(fd); return CallStatus::Ok; });
 			try {
 				for (;;) {
 					auto active_channels = m_epoller_ptr->poll(-1);
@@ -793,6 +772,7 @@ class Reactor : public NonCopyable {
 				}
 			} catch (const Shutdown&) {
 				std::lock_guard<std::mutex> lk(m_mtx);
+				cancel(call_id);
 				for (auto& func : m_pending_functors)
 					func();
 			}
@@ -815,9 +795,9 @@ class Reactor : public NonCopyable {
 		m_epoller_ptr->updateChannel(channel);
 	}
 
-	void wakeupRead() {
+	void wakeupRead(const int& fd) {
 		char c;
-		if (read(m_wakeup_fd_ptr[0], &c, sizeof(c)) == -1 && m_error_callback)
+		if (read(fd, &c, sizeof(c)) == -1 && m_error_callback)
 			m_error_callback(__FILE__, __LINE__, errno);
 		std::unique_lock<std::mutex> lk(m_mtx);
 		auto pending_functors = std::move(m_pending_functors);
@@ -828,7 +808,6 @@ class Reactor : public NonCopyable {
 
 	std::unique_ptr<Epoll> m_epoller_ptr;
 	std::unique_ptr<int[], std::function<void(int*)>> m_wakeup_fd_ptr;
-	std::unique_ptr<Channel> m_wakeup_channel_ptr;
 	std::thread::id m_thread_id;
 	std::thread m_thread;
 	std::vector<Functor> m_pending_functors;
@@ -846,4 +825,4 @@ class Reactor : public NonCopyable {
 
 } // namespace fantasy
 
-#endif // _REACTOR_HPP_INCLUD_ED
+#endif // _REACTOR_HPP_INCLUD_ED