refactor: improve worker service management and time handling

sniper00 · sniper00 · commit 6f18e38144d1 · 2025-10-31T22:31:13.000+08:00
- Add time offset accessor method to support timestamp synchronization
- Refactor worker service lifecycle management with better error handling
- Extract service ID allocation logic into dedicated method for clarity
- Optimize worker message processing with improved queue handling
- Enhance service scanning with pre-allocated memory and proper JSON formatting
- Improve code quality with const-correctness and structured bindings

Key improvements:
- Replace memory_order_release with memory_order_relaxed for better performance
- Add early return for empty service collections to avoid unnecessary processing
- Use structured bindings for cleaner iteration over service containers
- Separate service ID allocation logic for better maintainability
- Add comprehensive error handling for service creation failures

Performance optimizations:
- Pre-allocate string capacity for service scan responses
- Optimize service removal with streamlined state management
- Improve message queue processing with cached service references
diff --git a/src/common/time.hpp b/src/common/time.hpp
@@ -29,6 +29,10 @@ class time {
         return true;
     }
 
+    static std::time_t offset() {
+        return offset_;
+    }
+
     static std::time_t now() {
         auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(
             std::chrono::steady_clock::now() - start_time_point_
diff --git a/src/moon/core/worker.cpp b/src/moon/core/worker.cpp
@@ -37,9 +37,9 @@ void worker::run() {
 
 void worker::stop() {
     asio::post(io_ctx_, [this] {
-        message m = message { PTYPE_SHUTDOWN, 0, 0, 0 };
-        for (auto& it: services_) {
-            it.second->dispatch(&m);
+        auto m = message { PTYPE_SHUTDOWN, 0, 0, 0 };
+        for (const auto&[k,v]: services_) {
+            v->dispatch(&m);
         }
     });
 }
@@ -51,146 +51,185 @@ void worker::wait() {
     }
 }
 
-void worker::signal(int val) {
+void worker::signal(int val) const{
     if (auto s = current_.load(std::memory_order_acquire); s != nullptr) {
         s->signal(val);
     }
 }
 
-void worker::new_service(std::unique_ptr<service_conf> conf) {
-    count_.fetch_add(1, std::memory_order_release);
-    asio::post(io_ctx_, [this, conf = std::move(conf)]() {
-        do {
-            size_t counter = 0;
-            uint32_t serviceid = conf->opt_service_id;
-            if (serviceid == 0) {
-                do {
-                    if (counter >= WORKER_MAX_SERVICE) {
-                        serviceid = 0;
-                        CONSOLE_ERROR(
-                            "new service failed: can not get more service id. worker[%u] service num[%zu].",
-                            id(),
-                            services_.size()
-                        );
-                        break;
-                    }
+uint32_t worker::allocate_service_id(uint32_t opt_service_id) {
+    // Use specified service ID if provided
+    if (opt_service_id != 0) {
+        if (services_.find(opt_service_id) != services_.end()) {
+            CONSOLE_ERROR(
+                "new service failed: serviceid[%08X] already exists, worker[%u] service num[%zu].",
+                opt_service_id,
+                id(),
+                services_.size()
+            );
+            return 0;
+        }
+        return opt_service_id;
+    }
 
-                    ++nextid_;
-                    if (nextid_ == WORKER_MAX_SERVICE) {
-                        nextid_ = 1;
-                    }
-                    serviceid = nextid_ | (id() << WORKER_ID_SHIFT);
-                    ++counter;
-                } while (services_.find(serviceid) != services_.end());
+    // Auto-allocate service ID
+    size_t counter = 0;
+    uint32_t serviceid = 0;
+    
+    do {
+        if (counter >= WORKER_MAX_SERVICE) {
+            CONSOLE_ERROR(
+                "new service failed: can not get more service id. worker[%u] service num[%zu].",
+                id(),
+                services_.size()
+            );
+            return 0;
+        }
 
-                if (serviceid == 0) {
-                    break;
-                }
-            }else{
-                if (services_.find(serviceid) != services_.end()) {
-                    serviceid = 0;
-                    CONSOLE_ERROR(
-                        "new service failed: serviceid[%08X] already exists, worker[%u] service num[%zu].",
-                        serviceid,
-                        id(),
-                        services_.size()
-                    );
-                    break;
-                }
-            }
+        ++nextid_;
+        if (nextid_ == WORKER_MAX_SERVICE) {
+            nextid_ = 1;
+        }
+        serviceid = nextid_ | (id() << WORKER_ID_SHIFT);
+        ++counter;
+    } while (services_.find(serviceid) != services_.end());
 
-            auto s = server_->make_service(conf->type);
-            MOON_ASSERT(
-                s,
-                moon::format(
-                    "new service failed:service type[%s] was not registered",
-                    conf->type.data()
-                )
-                    .data()
-            );
-            s->set_id(serviceid);
-            s->set_unique(conf->unique);
-            s->set_server_context(server_, this);
+    return serviceid;
+}
 
-            if (!s->init(*conf)) {
-                if (serviceid == BOOTSTRAP_ADDR) {
-                    server_->stop(-1);
-                }
-                break;
+void worker::new_service(std::unique_ptr<service_conf> conf) {
+    count_.fetch_add(1, std::memory_order_relaxed);
+    asio::post(io_ctx_, [this, conf = std::move(conf)]() {
+        // Allocate service ID
+        uint32_t serviceid = allocate_service_id(conf->opt_service_id);
+        if (serviceid == 0) {
+            count_.fetch_sub(1, std::memory_order_relaxed);
+            if (services_.empty()) {
+                shared(true);
+            }
+            if (conf->session != 0) {
+                server_->send_message(message { PTYPE_INTEGER, 0, conf->creator, conf->session, 0 });
             }
-            s->ok(true);
-            services_.emplace(serviceid, std::move(s));
+            return;
+        }
 
-            if (0 != conf->session) {
-                server_->send_message(
-                    message { PTYPE_INTEGER, 0, conf->creator, conf->session, serviceid }
-                );
+        // Create service instance
+        auto s = server_->make_service(conf->type);
+        if (!s) {
+            CONSOLE_ERROR(
+                "new service failed: service type[%s] was not registered",
+                conf->type.data()
+            );
+            count_.fetch_sub(1, std::memory_order_relaxed);
+            if (services_.empty()) {
+                shared(true);
+            }
+            if (conf->session != 0) {
+                server_->send_message(message { PTYPE_INTEGER, 0, conf->creator, conf->session, 0 });
             }
             return;
-        } while (false);
+        }
 
-        count_.fetch_sub(1, std::memory_order_release);
-        if (services_.empty()) {
-            shared(true);
+        // Initialize service
+        s->set_id(serviceid);
+        s->set_unique(conf->unique);
+        s->set_server_context(server_, this);
+
+        if (!s->init(*conf)) {
+            if (serviceid == BOOTSTRAP_ADDR) {
+                server_->stop(-1);
+            }
+            count_.fetch_sub(1, std::memory_order_relaxed);
+            if (services_.empty()) {
+                shared(true);
+            }
+            if (conf->session != 0) {
+                server_->send_message(message { PTYPE_INTEGER, 0, conf->creator, conf->session, 0 });
+            }
+            return;
         }
 
-        if (0 != conf->session) {
-            server_->send_message(message { PTYPE_INTEGER, 0, conf->creator, conf->session, 0 });
+        // Service created successfully
+        s->ok(true);
+        services_.emplace(serviceid, std::move(s));
+
+        if (conf->session != 0) {
+            server_->send_message(
+                message { PTYPE_INTEGER, 0, conf->creator, conf->session, serviceid }
+            );
         }
     });
 }
 
 void worker::remove_service(uint32_t serviceid, uint32_t sender, int64_t sessionid) {
     asio::post(io_ctx_, [this, serviceid, sender, sessionid]() {
-        if (auto s = find_service(serviceid); nullptr != s) {
-            auto name = s->name();
-            auto id = s->id();
-            count_.fetch_sub(1, std::memory_order_release);
-            server_->response(sender, "service destroy"sv, sessionid);
-            services_.erase(serviceid);
-            if (services_.empty())
-                shared(true);
-
-            if (server_->get_state() == state::ready) {
-                auto content =
-                    moon::format("_service_exit,name:%s serviceid:%08X", name.data(), id);
-                auto buf = buffer { content.size() };
-                buf.write_back(content);
-                server_->broadcast(serviceid, buf, PTYPE_SYSTEM);
-            }
-
-            if (serviceid == BOOTSTRAP_ADDR) {
-                server_->set_state(state::stopping);
-            }
-        } else {
+        auto s = find_service(serviceid);
+        if (nullptr == s) {
             server_->response(
                 sender,
                 moon::format("worker::remove_service [%08X] not found", serviceid),
                 sessionid,
                 PTYPE_ERROR
             );
+            return;
+        }
+
+        auto name = s->name();
+        auto id = s->id();
+        count_.fetch_sub(1, std::memory_order_relaxed);
+        services_.erase(serviceid);
+        
+        // Mark worker as shared if no services remain
+        if (services_.empty()) {
+            shared(true);
+        }
+
+        server_->response(sender, "service destroy"sv, sessionid);
+
+        // Broadcast service exit notification (only if server is ready)
+        if (server_->get_state() == state::ready) {
+            auto content = moon::format("_service_exit,name:%s serviceid:%08X", name.data(), id);
+            auto buf = buffer { content.size() };
+            buf.write_back(content);
+            server_->broadcast(serviceid, buf, PTYPE_SYSTEM);
+        }
+
+        // Check if bootstrap service was removed
+        if (serviceid == BOOTSTRAP_ADDR) {
+            server_->set_state(state::stopping);
         }
     });
 }
 
 void worker::scan(uint32_t sender, int64_t sessionid) {
     asio::post(io_ctx_, [this, sender, sessionid] {
+        if (services_.empty()) {
+            server_->response(sender, std::string_view{}, sessionid);
+            return;
+        }
+
         std::string content;
-        for (const auto& it: services_) {
-            if (content.empty())
-                content.append("[");
+        // Pre-allocate approximate space: each entry ~50 chars
+        content.reserve(services_.size() * 50 + 2);
+        content.append("[");
+
+        bool first = true;
+        for (const auto& [k,v]: services_) {
+            if (!first) {
+                content.append(",");
+            }
+            first = false;
 
             content.append(
                 moon::format(
-                    R"({"name":"%s","serviceid":"%X"},)",
-                    it.second->name().data(),
-                    it.second->id()
+                    R"({"name":"%s","serviceid":"%X"})",
+                    v->name().data(),
+                    v->id()
                 )
             );
         }
 
-        if (!content.empty())
-            content.back() = ']';
+        content.append("]");
         server_->response(sender, content, sessionid);
     });
 }
@@ -202,17 +241,23 @@ asio::io_context& worker::io_context() {
 void worker::send(message&& msg) {
     if (mq_.push_back(std::move(msg)) == 1) {
         asio::post(io_ctx_, [this]() {
-            if (auto& read_queue = mq_.swap_on_read(); !read_queue.empty()) {
-                auto size = read_queue.size();
-                swapped_size_.store(size, std::memory_order_relaxed);
-                service* s = nullptr;
-                for (auto& m: read_queue) {
-                    s = handle_one(s, std::move(m));
-                    swapped_size_.store(--size, std::memory_order_relaxed);
-                }
-                read_queue.clear();
-                current_.store(nullptr, std::memory_order_relaxed);
+            auto& read_queue = mq_.swap_on_read();
+            if (read_queue.empty()) {
+                return;
+            }
+
+            auto size = read_queue.size();
+            swapped_size_.store(size, std::memory_order_relaxed);
+            
+            // Process all messages in the queue
+            service* cached_service = nullptr;
+            for (auto& m: read_queue) {
+                cached_service = handle_one(cached_service, std::move(m));
+                swapped_size_.store(--size, std::memory_order_relaxed);
             }
+            
+            read_queue.clear();
+            current_.store(nullptr, std::memory_order_relaxed);
         });
     }
 }
@@ -222,8 +267,7 @@ uint32_t worker::id() const {
 }
 
 service* worker::find_service(uint32_t serviceid) const {
-    auto iter = services_.find(serviceid);
-    if (services_.end() != iter) {
+    if (auto iter = services_.find(serviceid); services_.end() != iter) {
         return iter->second.get();
     }
     return nullptr;
@@ -287,14 +331,14 @@ service* worker::handle_one(service* s, message&& msg) {
         return s->ok() ? s : nullptr;
     }
 
-    for (auto& it: services_) {
-        if (!it.second->unique() && type == PTYPE_SYSTEM) {
+    for (const auto& [k,v]: services_) {
+        if (!v->unique() && type == PTYPE_SYSTEM) {
             continue;
         }
 
-        if (it.second->ok() && it.second->id() != sender) {
-            current_.store(it.second.get(), std::memory_order_release);
-            handle_message(it.second, msg);
+        if (v->ok() && v->id() != sender) {
+            current_.store(v.get(), std::memory_order_release);
+            handle_message(v, msg);
         }
     }
     return nullptr;
diff --git a/src/moon/core/worker.h b/src/moon/core/worker.h
@@ -62,13 +62,15 @@ class worker {
 
     void wait();
 
-    void signal(int val);
+    void signal(int val) const;
 
 private:
     service* handle_one(service* s, message&& msg);
 
     service* find_service(uint32_t serviceid) const;
 
+    uint32_t allocate_service_id(uint32_t opt_service_id);
+
 private:
     std::atomic_bool shared_ = true;
     std::atomic_uint32_t count_ = 0;
