From c5453f34f76cb0413ce9dac2b4bb150f514b29c6 Mon Sep 17 00:00:00 2001 From: XieHan Date: Tue, 30 Mar 2021 18:22:55 +0800 Subject: [PATCH] update code --- src/kernel/Communicator.cc | 50 +++++++++++++++++++++----------------- src/kernel/Communicator.h | 24 ++++++++++++++---- tutorial/CMakeLists.txt | 1 + 3 files changed, 48 insertions(+), 27 deletions(-) diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index 934d59e6..7e4843ef 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -314,11 +314,6 @@ private: friend class Communicator; }; -CommConnection *CommSession::get_connection() const -{ - return this->entry ? entry->conn : NULL; -} - CommSession::~CommSession() { struct CommConnEntry *entry; @@ -363,6 +358,18 @@ CommMessageOut *CommChannel::message_out() return ∅ } +CommMessageOut *CommSessionOut::message_out() +{ + errno = EPERM; + return NULL; +} + +CommMessageIn *CommSessionOut::message_in() +{ + errno = EPERM; + return NULL; +} + inline int Communicator::first_timeout(CommSession *session) { int timeout = session->target->response_timeout; @@ -1225,7 +1232,7 @@ int Communicator::create_service_session(struct CommConnEntry *entry) session->passive = 1; entry->session = session; session->target = target; - session->entry = entry; + session->conn = entry->conn; session->seq = entry->seq++; session->out = NULL; session->in = NULL; @@ -1359,7 +1366,7 @@ void Communicator::callback(struct poller_result *res, void *context) } free(entry); - entry = session->entry; + entry = ((CommSessionOut *)session)->entry; session->handle(state, res->error); session = entry->session; } @@ -1622,7 +1629,7 @@ int Communicator::request_idle_conn(CommSession *session, CommTarget *target) if (entry) { entry->session = session; - session->entry = entry; + session->conn = entry->conn; session->seq = entry->seq++; session->out = session->message_out(); if (session->out) @@ -1665,7 +1672,7 @@ int Communicator::request(CommSession *session, CommTarget *target) entry = this->launch_conn(session, target); if (entry) { - session->entry = entry; + session->conn = entry->conn; session->seq = entry->seq++; data.operation = PD_OP_CONNECT; data.fd = entry->sockfd; @@ -1677,7 +1684,7 @@ int Communicator::request(CommSession *session, CommTarget *target) this->release_conn(entry); } - session->entry = NULL; + session->conn = NULL; session->seq = 0; return -1; } @@ -1838,15 +1845,7 @@ int Communicator::establish(CommChannel *channel, CommTarget *target) return -1; } -void Communicator::prep_send(CommSession *session, CommChannel *channel) -{ - session->target = channel->target; - session->entry = channel->entry; - session->in = NULL; - session->seq = 0; -} - -int Communicator::send(CommMessageOut *msg, CommSession *session, +int Communicator::send(CommMessageOut *msg, CommSessionOut *session, CommChannel *channel) { struct CommConnEntry *entry = channel->entry; @@ -1854,6 +1853,11 @@ int Communicator::send(CommMessageOut *msg, CommSession *session, struct iovec *end; int cnt; + session->target = channel->target; + session->entry = channel->entry; + session->in = NULL; + session->seq = 0; + session->out = msg; cnt = msg->encode(vectors, ENCODE_IOV_MAX); if ((unsigned int)cnt > ENCODE_IOV_MAX) @@ -1867,8 +1871,10 @@ int Communicator::send(CommMessageOut *msg, CommSession *session, if (!entry->ssl || cnt == 0) { cnt = __send_vectors(vectors, cnt, entry); - if (cnt <= 0) - return cnt; + if (cnt == 0) + return 1; + else if (cnt < 0) + return -1; } entry = (struct CommConnEntry *)malloc(sizeof (struct CommConnEntry)); @@ -1884,7 +1890,7 @@ int Communicator::send(CommMessageOut *msg, CommSession *session, if (this->send_message_async(end - cnt, cnt, entry) >= 0) { __sync_add_and_fetch(&channel->entry->ref, 1); - return 1; + return 0; } close(entry->sockfd); diff --git a/src/kernel/Communicator.h b/src/kernel/Communicator.h index 1254116c..a5803e19 100644 --- a/src/kernel/Communicator.h +++ b/src/kernel/Communicator.h @@ -134,14 +134,14 @@ private: protected: CommTarget *get_target() const { return this->target; } + CommConnection *get_connection() const { return this->conn; } CommMessageOut *get_message_out() const { return this->out; } CommMessageIn *get_message_in() const { return this->in; } long long get_seq() const { return this->seq; } - CommConnection *get_connection() const; private: CommTarget *target; - struct CommConnEntry *entry; + CommConnection *conn; CommMessageOut *out; CommMessageIn *in; long long seq; @@ -243,8 +243,22 @@ private: private: virtual CommMessageOut *message_out(); /* final */ - virtual CommMessageOut *get_message_out() { return NULL; } /* deleted */ + CommMessageOut *get_message_out() { return NULL; } /* deleted */ +private: + struct CommConnEntry *entry; + friend class Communicator; +}; + +class CommSessionOut : public CommSession +{ +private: + virtual CommMessageOut *message_out(); /* final */ + virtual CommMessageIn *message_in(); /* final */ + CommMessageIn *get_message_in() { return NULL; } /* deleted */ + +private: + struct CommConnEntry *entry; friend class Communicator; }; @@ -282,8 +296,8 @@ public: void unbind(CommService *service); int establish(CommChannel *channel, CommTarget *target); - void prep_send(CommSession *session, CommChannel *channel); - int send(CommMessageOut *msg, CommSession *session, CommChannel *channel); + int send(CommMessageOut *msg, CommSessionOut *session, + CommChannel *channel); void shutdown(CommChannel *channel); int sleep(SleepSession *session); diff --git a/tutorial/CMakeLists.txt b/tutorial/CMakeLists.txt index c3c08722..7415daf1 100644 --- a/tutorial/CMakeLists.txt +++ b/tutorial/CMakeLists.txt @@ -23,6 +23,7 @@ else () endif () set(TUTORIAL_LIST + test_channel tutorial-01-wget tutorial-02-redis_cli tutorial-03-wget_to_redis