From 34594d7c8f724782b3da220a8ef2da5f1bbbd533 Mon Sep 17 00:00:00 2001 From: Fankux Date: Tue, 19 Apr 2022 11:52:43 +0800 Subject: [PATCH] fix compile version --- CMakeLists.txt | 12 ++++++++++++ src/arranger/arranger.cpp | 5 ++--- src/arranger/arranger.h | 2 +- src/arranger/source_invoke.cpp | 11 +++++------ src/codec/legacy_encoder.cpp | 5 +++-- src/communication/channel.cpp | 1 + src/communication/communicator.cpp | 8 ++------ src/entry.cpp | 4 ++++ src/oblogreader/reader_routine.cpp | 1 + 9 files changed, 31 insertions(+), 18 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index f020324..b45efff 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -194,6 +194,18 @@ execute_process( WORKING_DIRECTORY ${PROJECT_SOURCE_DIR} ) +if (NOT GIT_VERSION) + message(WARNING "oblogproxy fetch git version empty, use current time as program version") + STRING(TIMESTAMP GIT_VERSION "%Y-%m-%d %H:%M:%S") +endif () + +if (NOT GIT_VERSION) + message(WARNING "oblogproxy fetch current time failed") + SET(GIT_VERSION "2.0.0") +endif () + +message("oblogproxy version: ${GIT_VERSION}") + if (WITH_DEBUG) SET(DEBUG_SYMBOL "-ggdb") else () diff --git a/src/arranger/arranger.cpp b/src/arranger/arranger.cpp index bd51843..56016fd 100644 --- a/src/arranger/arranger.cpp +++ b/src/arranger/arranger.cpp @@ -52,7 +52,7 @@ EventResult Arranger::on_msg(const PeerInfo& peer, const Message& msg) { OMS_INFO << "Arranger on_msg fired: " << peer.to_string(); if (msg.type() == MessageType::HANDSHAKE_REQUEST_CLIENT) { - ClientHandshakeRequestMessage& handshake = (ClientHandshakeRequestMessage&)msg; + auto& handshake = (ClientHandshakeRequestMessage&)msg; OMS_INFO << "Handshake request from peer: " << peer.to_string() << ", msg: " << handshake.to_string(); ClientMeta client = ClientMeta::from_handshake(peer, handshake); @@ -128,7 +128,7 @@ int Arranger::create(const ClientMeta& client) int ret = start_source(client, client.configuration); if (ret != OMS_OK) { - close_client_locked(client, ""); + close_client_locked(client, "failed to invoke"); return ret; } @@ -182,7 +182,6 @@ int Arranger::close_client_locked(const ClientMeta& client, const std::string& m if (ret != OMS_OK) { OMS_WARN << "Failed to send error response message. client=" << client.peer.id(); } - _accepter.remove_channel(channel_entry->second); _client_peers.erase(channel_entry); } diff --git a/src/arranger/arranger.h b/src/arranger/arranger.h index 1f9ed0c..c5d271f 100644 --- a/src/arranger/arranger.h +++ b/src/arranger/arranger.h @@ -37,7 +37,7 @@ public: private: EventResult on_msg(const PeerInfo&, const Message&); - int auth(ClientMeta& client, std::string& errmsg); + static int auth(ClientMeta& client, std::string& errmsg); int start_source(const ClientMeta& client, const std::string& configuration); diff --git a/src/arranger/source_invoke.cpp b/src/arranger/source_invoke.cpp index c18f9ac..81114a9 100644 --- a/src/arranger/source_invoke.cpp +++ b/src/arranger/source_invoke.cpp @@ -99,7 +99,6 @@ public: ::exit(-1); } else { // parent; - OMS_INFO << "+++ create oblogreader with pid: " << pid; SourceWaiter::instance().add(pid, _client); } @@ -116,7 +115,6 @@ private: static int start_oblogreader(Communicator& communicator, const ClientMeta& client, const std::string& configuration) { communicator.fork_prepare(); - // we create new thread for fork() acting as children process's main thread ForkThread fork_thd(communicator, client, configuration); fork_thd.start(); @@ -168,15 +166,16 @@ void SourceWaiter::WaitThread::run() { int retval = OMS_OK; waitpid(_pid, &retval, 0); - OMS_WARN << "--- oblogreader exit with ret: " << retval << ", try to close fd: " << _client.peer.file_desc; - if (retval != OMS_OK) { + if (WIFEXITED(retval)) { + OMS_INFO << "--- oblogreader exit succeed, try to close fd: " << _client.peer.file_desc; + } else { + OMS_ERROR << "oblogreader exit failed:" << WEXITSTATUS(retval); // TODO... response to client with _client.channel } - shutdown(_client.peer.file_desc, SHUT_RDWR); // use a thread to remove avoid join dead lock - Arranger::instance().close_client(_client); + Arranger::instance().close_client(_client, "oblogreader exit"); SourceWaiter::instance().remove(_pid); OMS_WARN << "--- oblogreader WaiterThread(" << tid() << ") exit for pid: " << _pid; diff --git a/src/codec/legacy_encoder.cpp b/src/codec/legacy_encoder.cpp index ddb9ff0..539485d 100644 --- a/src/codec/legacy_encoder.cpp +++ b/src/codec/legacy_encoder.cpp @@ -212,7 +212,9 @@ LegacyEncoder::LegacyEncoder() memcpy(buf, &code_be, 4); uint32_t varlen_be = cpu_to_be(msg.message.size()); memcpy(buf + 4, &varlen_be, 4); - memcpy(buf + 8, msg.message.c_str(), msg.message.size()); + if (msg.message.size() != 0) { + memcpy(buf + 8, msg.message.c_str(), msg.message.size()); + } // buf's ownership transfered to buffer buffer.push_back(buf, len); @@ -238,7 +240,6 @@ int LegacyEncoder::encode(const Message& msg, MsgBuf& buffer) uint32_t msg_type_be = cpu_to_be((uint32_t)msg.type()); memcpy(buf + 2, &msg_type_be, 4); buffer.push_front(buf, len); - return ret; } diff --git a/src/communication/channel.cpp b/src/communication/channel.cpp index 464c338..c2ebbc8 100644 --- a/src/communication/channel.cpp +++ b/src/communication/channel.cpp @@ -46,6 +46,7 @@ Channel* Channel::get() void Channel::put() { if (1 == _refcount.fetch_sub(1)) { + OMS_DEBUG << "delete Channel"; delete this; } } diff --git a/src/communication/communicator.cpp b/src/communication/communicator.cpp index cbcc87a..be06f4c 100644 --- a/src/communication/communicator.cpp +++ b/src/communication/communicator.cpp @@ -186,7 +186,6 @@ int Communicator::add_channel(const PeerInfo& peer, Channel* ch /* = nullptr */) } std::lock_guard lock_guard(_lock); - auto iter = _channels.find(peer); if (iter != _channels.end()) { OMS_WARN << "Add channel twice: " << peer.file_desc; @@ -263,7 +262,6 @@ int Communicator::remove_channel(const PeerInfo& peer, bool steal /* = false */) Channel* ch = nullptr; { std::lock_guard lock_guard(_lock); - auto iter = _channels.find(peer); if (iter == _channels.end()) { OMS_WARN << "No channel found of peer: " << peer.to_string(); @@ -284,7 +282,6 @@ int Communicator::remove_channel(const PeerInfo& peer, bool steal /* = false */) int Communicator::clear_channels() { std::lock_guard lock_guard(_lock); - for (auto& channel : _channels) { Channel* ch = channel.second; release_channel_event(*ch, false); @@ -296,7 +293,6 @@ int Communicator::clear_channels() Channel* Communicator::get_channel(const PeerInfo& peer) { const std::lock_guard lock_guard(_lock); - return get_channel_locked(peer); } @@ -468,8 +464,7 @@ void Communicator::on_event(int fd, short event, void* arg) delete msg; } } - - ch->put(); + //对于ER_CLOSE_CHANNEL,先处理错误再释放内存 switch (err) { case EventResult::ER_CLOSE_CHANNEL: c.remove_channel(ch->_peer); @@ -478,6 +473,7 @@ void Communicator::on_event(int fd, short event, void* arg) // do nothing break; } + ch->put(); } void Communicator::close_listen() diff --git a/src/entry.cpp b/src/entry.cpp index 28f14e3..8955f52 100644 --- a/src/entry.cpp +++ b/src/entry.cpp @@ -29,6 +29,10 @@ int main(int argc, char** argv) options.usage(); exit(0); })); + options.add(OmsOption('v', "version", false, "program version", [&](const std::string&) { + printf("version: " __OMS_VERSION__ "\n"); + exit(0); + })); options.add(OmsOption('f', "file", true, "configuration json file", [&](const std::string& optarg) { if (conf.load(optarg) != OMS_OK) { OMS_INFO << "failed to load config: " << optarg; diff --git a/src/oblogreader/reader_routine.cpp b/src/oblogreader/reader_routine.cpp index 026271e..8acf38b 100644 --- a/src/oblogreader/reader_routine.cpp +++ b/src/oblogreader/reader_routine.cpp @@ -46,6 +46,7 @@ void ReaderRoutine::stop() void ReaderRoutine::run() { if (_oblog.start() != OMS_OK) { + OMS_ERROR << "Failed to start ReaderRoutine"; return; } -- GitLab