diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index eb3ffce7..e764067b 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -7,6 +7,7 @@ set(COMMON_KERNEL_HEADERS src/kernel/SleepRequest.h src/kernel/ExecRequest.h src/kernel/IORequest.h + src/kernel/TransRequest.h src/kernel/Executor.h src/kernel/list.h src/kernel/mpoller.h diff --git a/src/kernel/CMakeLists.txt b/src/kernel/CMakeLists.txt index 209f0ec0..683c7829 100644 --- a/src/kernel/CMakeLists.txt +++ b/src/kernel/CMakeLists.txt @@ -17,6 +17,7 @@ set(SRC msgqueue.c thrdpool.c CommRequest.cc + TransRequest.cc CommScheduler.cc Communicator.cc Executor.cc diff --git a/src/kernel/TransRequest.cc b/src/kernel/TransRequest.cc new file mode 100644 index 00000000..950366b0 --- /dev/null +++ b/src/kernel/TransRequest.cc @@ -0,0 +1,59 @@ +/* + Copyright (c) 2021 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com) +*/ + +#include +#include "CommScheduler.h" +#include "TransRequest.h" + +void ChanRequest::dispatch() +{ + if (!this->established) + { + if (this->scheduler->establish(this, this->object, this->wait_timeout, + &this->target) < 0) + { + this->state = CS_STATE_ERROR; + this->error = errno; + this->subtask_done(); + } + } + else + this->scheduler->shutdown(this); +} + +void TransRequest::dispatch() +{ + int ret = this->scheduler->send(this, this->channel); + + if (ret == 0) + return; + + if (ret > 0) + { + this->state = CS_STATE_SUCCESS; + this->error = 0; + } + else + { + this->state = CS_STATE_ERROR; + this->error = errno; + } + + this->subtask_done(); +} + diff --git a/src/kernel/TransRequest.h b/src/kernel/TransRequest.h new file mode 100644 index 00000000..b83549c2 --- /dev/null +++ b/src/kernel/TransRequest.h @@ -0,0 +1,109 @@ +/* + Copyright (c) 2021 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Author: Xie Han (xiehan@sogou-inc.com) +*/ + +#ifndef _TRANSREQUEST_H_ +#define _TRANSREQUEST_H_ + +#include "SubTask.h" +#include "Communicator.h" +#include "CommScheduler.h" + +class ChanRequest : public SubTask, public CommChannel +{ +public: + ChanRequest(CommSchedObject *object, CommScheduler *scheduler) + { + this->scheduler = scheduler; + this->object = object; + this->wait_timeout = 0; + this->established = 0; + } + + CommSchedObject *get_request_object() const { return this->object; } + void set_request_object(CommSchedObject *object) { this->object = object; } + int get_wait_timeout() const { return this->wait_timeout; } + void set_wait_timeout(int timeout) { this->wait_timeout = timeout; } + +public: + virtual void dispatch(); + +protected: + int state; + int error; + +protected: + CommTarget *target; + +protected: + int established; + int wait_timeout; + CommSchedObject *object; + CommScheduler *scheduler; + +protected: + virtual void handle_established() + { + this->state = CS_STATE_SUCCESS; + this->error = 0; + this->established = 1; + this->subtask_done(); + } + + virtual void handle(int state, int error) + { + this->state = state; + this->error = error; + this->established = 0; + this->subtask_done(); + } +}; + +class TransRequest : public SubTask, public TransSession +{ +public: + TransRequest(CommChannel *channel, CommScheduler *scheduler) + { + this->scheduler = scheduler; + this->channel = channel; + } + + CommChannel *get_request_channel() const { return this->channel; } + void set_request_channel(CommChannel *channel) { this->channel = channel; } + +public: + virtual void dispatch(); + +protected: + int state; + int error; + +protected: + CommChannel *channel; + CommScheduler *scheduler; + +protected: + virtual void handle(int state, int error) + { + this->state = state; + this->error = error; + this->subtask_done(); + } +}; + +#endif +