cmhi
/
f-stack
mirror of https://github.com/F-Stack/f-stack.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
f-stack/app/micro_thread/mt_concurrent.cpp

529 lines
14 KiB
C++
Raw Normal View History

init
2017-04-21 10:43:26 +00:00
/**
* Tencent is pleased to support the open source community by making MSEC available.
*
* Copyright (C) 2016 THL A29 Limited, a Tencent company. All rights reserved.
*
* Licensed under the GNU General Public 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
*
* https://opensource.org/licenses/GPL-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.
*/
/**
* @file mt_concurrent.c
* @info <EFBFBD><EFBFBD>·<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ģ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>չ
* @time 20130924
**/
#include "micro_thread.h"
#include "mt_msg.h"
#include "mt_notify.h"
#include "mt_connection.h"
#include "mt_concurrent.h"
using namespace std;
using namespace NS_MICRO_THREAD;
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD>Ĵ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ż<EFBFBD>, <EFBFBD>첽<EFBFBD><EFBFBD><EFBFBD>ȵȴ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
* @param req_list - <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>б<EFBFBD>
* @param how - EPOLLIN EPOLLOUT
* @param timeout - <EFBFBD><EFBFBD>ʱʱ<EFBFBD><EFBFBD> <EFBFBD><EFBFBD><EFBFBD>뵥λ
* @return 0 <EFBFBD>ɹ<EFBFBD>, <0ʧ<EFBFBD><EFBFBD> -3 <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ
*/
int NS_MICRO_THREAD::mt_multi_netfd_poll(IMtActList& req_list, int how, int timeout)
{
KqObjList fdlist;
TAILQ_INIT(&fdlist);
KqueuerObj* obj = NULL;
IMtAction* action = NULL;
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
action = *it;
if (action) {
obj = action->GetNtfyObj();
}
if (!action || !obj)
{
action->SetErrno(ERR_FRAME_ERROR);
MTLOG_ERROR("input action %p, or ntify null, error", action);
return -1;
}
obj->SetRcvEvents(0);
if (how & KQ_EVENT_READ)
{
obj->EnableInput();
}
else
{
obj->DisableInput();
}
if (how & KQ_EVENT_WRITE)
{
obj->EnableOutput();
}
else
{
obj->DisableOutput();
}
TAILQ_INSERT_TAIL(&fdlist, obj, _entry);
}
MtFrame* mtframe = MtFrame::Instance();
if (!mtframe || !mtframe->KqueueSchedule(&fdlist, NULL, (int)timeout))
{
if (errno != ETIME)
{
action->SetErrno(ERR_KQUEUE_FAIL);
MTLOG_ERROR("Mtframe %p, epoll schedule failed, errno %d", mtframe, errno);
return -2;
}
return -3;
}
return 0;
}
/**
* @brief Ϊÿ<EFBFBD><EFBFBD>ITEM<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ĵ<EFBFBD>socket
* @param req_list - <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>б<EFBFBD>
* @return 0 <EFBFBD>ɹ<EFBFBD>, <0ʧ<EFBFBD><EFBFBD>
*/
int NS_MICRO_THREAD::mt_multi_newsock(IMtActList& req_list)
{
int sock = -1, has_ok = 0;
IMtAction* action = NULL;
IMtConnection* net_handler = NULL;
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
action = *it;
if (NULL == action)
{
action->SetErrno(ERR_PARAM_ERROR);
MTLOG_ERROR("Invalid param, conn %p null!!", action);
return -1;
}
if (action->GetErrno() != ERR_NONE) {
continue;
}
net_handler = action->GetIConnection();
if (NULL == net_handler)
{
action->SetErrno(ERR_FRAME_ERROR);
MTLOG_ERROR("Invalid param, conn %p null!!", net_handler);
return -2;
}
sock = net_handler->CreateSocket();
if (sock < 0)
{
action->SetErrno(ERR_SOCKET_FAIL);
MTLOG_ERROR("Get sock data failed, ret %d, errno %d!!", sock, errno);
return -3;
}
has_ok = 1;
if (action->GetProtoType() == MT_UDP)
{
action->SetMsgFlag(MULTI_FLAG_OPEN);
}
else
{
action->SetMsgFlag(MULTI_FLAG_INIT);
}
}
if (has_ok)
{
return 0;
}
else
{
return -4;
}
}
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD>Ĵ<EFBFBD><EFBFBD><EFBFBD>, <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
* @param req_list - <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>б<EFBFBD>
* @param timeout - <EFBFBD><EFBFBD>ʱʱ<EFBFBD><EFBFBD> <EFBFBD><EFBFBD><EFBFBD>뵥λ
* @return 0 <EFBFBD>ɹ<EFBFBD>, <0ʧ<EFBFBD><EFBFBD>
*/
int NS_MICRO_THREAD::mt_multi_open(IMtActList& req_list, int timeout)
{
utime64_t start_ms = MtFrame::Instance()->GetLastClock();
utime64_t end_ms = start_ms + timeout;
utime64_t curr_ms = 0;
int ret = 0, has_open = 0;
IMtAction* action = NULL;
IMtConnection* net_handler = NULL;
IMtActList::iterator it;
while (1)
{
IMtActList wait_list;
for (it = req_list.begin(); it != req_list.end(); ++it)
{
action = *it;
if (action->GetErrno() != ERR_NONE) {
continue;
}
if (action->GetMsgFlag() == MULTI_FLAG_OPEN) {
has_open = 1;
continue;
}
net_handler = action->GetIConnection();
if (NULL == net_handler)
{
action->SetErrno(ERR_FRAME_ERROR);
MTLOG_ERROR("Invalid param, conn %p null!!", net_handler);
return -1;
}
ret = net_handler->OpenCnnect();
if (ret < 0)
{
wait_list.push_back(action);
}
else
{
action->SetMsgFlag(MULTI_FLAG_OPEN);
}
}
curr_ms = MtFrame::Instance()->GetLastClock();
if (curr_ms > end_ms)
{
MTLOG_DEBUG("Open connect timeout, errno %d", errno);
for (IMtActList::iterator it = wait_list.begin(); it != wait_list.end(); ++it)
{
(*it)->SetErrno(ERR_CONNECT_FAIL);
}
if (!has_open)
{
return 0;
}
else
{
return -2;
}
}
if (!wait_list.empty())
{
mt_multi_netfd_poll(wait_list, KQ_EVENT_WRITE, end_ms - curr_ms);
}
else
{
return 0;
}
}
}
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD>Ĵ<EFBFBD><EFBFBD><EFBFBD>, <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
* @param req_list - <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>б<EFBFBD>
* @param timeout - <EFBFBD><EFBFBD>ʱʱ<EFBFBD><EFBFBD> <EFBFBD><EFBFBD><EFBFBD>뵥λ
* @return 0 <EFBFBD>ɹ<EFBFBD>, <0ʧ<EFBFBD><EFBFBD>
*/
int NS_MICRO_THREAD::mt_multi_sendto(IMtActList& req_list, int timeout)
{
utime64_t start_ms = MtFrame::Instance()->GetLastClock();
utime64_t end_ms = start_ms + timeout;
utime64_t curr_ms = 0;
int ret = 0, has_send = 0;
IMtAction* action = NULL;
IMtConnection* net_handler = NULL;
while (1)
{
IMtActList wait_list;
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
action = *it;
if (action->GetErrno() != ERR_NONE) {
continue;
}
if (action->GetMsgFlag() == MULTI_FLAG_SEND) {
has_send = 1;
continue;
}
net_handler = action->GetIConnection();
if (NULL == net_handler)
{
action->SetErrno(ERR_FRAME_ERROR);
MTLOG_ERROR("Invalid param, conn %p null!!", net_handler);
return -2;
}
// 0 -<2D><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>; -1 ֹͣ<CDA3><D6B9><EFBFBD><EFBFBD>; > 0 <20><><EFBFBD><EFBFBD>OK
ret = net_handler->SendData();
if (ret == -1)
{
action->SetErrno(ERR_SEND_FAIL);
MTLOG_ERROR("MultiItem msg send error, %d", errno);
continue;
}
else if (ret == 0)
{
wait_list.push_back(action);
continue;
}
else
{
action->SetMsgFlag(MULTI_FLAG_SEND);
}
}
curr_ms = MtFrame::Instance()->GetLastClock();
if (curr_ms > end_ms)
{
MTLOG_DEBUG("send data timeout");
for (IMtActList::iterator it = wait_list.begin(); it != wait_list.end(); ++it)
{
(*it)->SetErrno(ERR_SEND_FAIL);
}
if (has_send)
{
return 0;
}
else
{
return -5;
}
}
if (!wait_list.empty())
{
mt_multi_netfd_poll(wait_list, KQ_EVENT_WRITE, end_ms - curr_ms);
}
else
{
return 0;
}
}
return 0;
}
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>մ<EFBFBD><EFBFBD><EFBFBD>
*/
int NS_MICRO_THREAD::mt_multi_recvfrom(IMtActList& req_list, int timeout)
{
utime64_t start_ms = MtFrame::Instance()->GetLastClock();
utime64_t end_ms = start_ms + timeout;
utime64_t curr_ms = 0;
int ret = 0;
IMtAction* action = NULL;
IMtConnection* net_handler = NULL;
while (1)
{
IMtActList wait_list;
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
action = *it;
if (action->GetErrno() != ERR_NONE) {
continue;
}
if (MULTI_FLAG_FIN == action->GetMsgFlag()) ///< <20>Ѵ<EFBFBD><D1B4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
{
continue;
}
net_handler = action->GetIConnection();
if (NULL == net_handler)
{
action->SetErrno(ERR_FRAME_ERROR);
MTLOG_ERROR("Invalid param, conn %p null!!", net_handler);
return -2;
}
// <0 ʧ<><CAA7>, 0 <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>, >0 <20>ɹ<EFBFBD>
ret = net_handler->RecvData();
if (ret < 0)
{
action->SetErrno(ERR_RECV_FAIL);
MTLOG_ERROR("MultiItem msg recv failed: %p", net_handler);
continue;
}
else if (ret == 0)
{
wait_list.push_back(action);
continue;
}
else
{
action->SetMsgFlag(MULTI_FLAG_FIN);
action->SetCost(MtFrame::Instance()->GetLastClock() - start_ms);
}
}
curr_ms = MtFrame::Instance()->GetLastClock();
if (curr_ms > end_ms)
{
MTLOG_DEBUG("Recv data timeout, curr %llu, start: %llu", curr_ms, start_ms);
for (IMtActList::iterator it = wait_list.begin(); it != wait_list.end(); ++it)
{
(*it)->SetErrno(ERR_RECV_TIMEOUT);
}
return -5;
}
if (!wait_list.empty())
{
mt_multi_netfd_poll(wait_list, KQ_EVENT_READ, end_ms - curr_ms);
}
else
{
return 0;
}
}
}
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>մ<EFBFBD><EFBFBD><EFBFBD>
*/
int NS_MICRO_THREAD::mt_multi_sendrcv_ex(IMtActList& req_list, int timeout)
{
utime64_t start_ms = MtFrame::Instance()->GetLastClock();
utime64_t curr_ms = 0;
int rc = mt_multi_newsock(req_list); // TODO, <20><><EFBFBD><EFBFBD>ȡconnect<63><74>ʱʱ<CAB1><CAB1><EFBFBD><EFBFBD>
if (rc < 0)
{
MT_ATTR_API(320842, 1); // socketʧ<74><CAA7>
MTLOG_ERROR("mt_multi_sendrcv new sock failed, ret: %d", rc);
return -1;
}
rc = mt_multi_open(req_list, timeout);
if (rc < 0)
{
MT_ATTR_API(320843, 1); // connectʧ<74><CAA7>
MTLOG_ERROR("mt_multi_sendrcv open failed, ret: %d", rc);
return -2;
}
curr_ms = MtFrame::Instance()->GetLastClock();
rc = mt_multi_sendto(req_list, timeout - (curr_ms - start_ms));
if (rc < 0)
{
MT_ATTR_API(320844, 1); // <20><><EFBFBD><EFBFBD>ʧ<EFBFBD><CAA7>
MTLOG_ERROR("mt_multi_sendrcv send failed, ret: %d", rc);
return -3;
}
curr_ms = MtFrame::Instance()->GetLastClock();
rc = mt_multi_recvfrom(req_list, timeout - (curr_ms - start_ms));
if (rc < 0)
{
MT_ATTR_API(320845, 1); // <20><><EFBFBD><EFBFBD>δ<EFBFBD><CEB4>ȫ<EFBFBD>ɹ<EFBFBD>
MTLOG_ERROR("mt_multi_sendrcv recv failed, ret: %d", rc);
return -4;
}
return 0;
}
/**
* @brief <EFBFBD><EFBFBD>·IO<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>մ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӿ<EFBFBD>, <EFBFBD><EFBFBD>װACTON<EFBFBD>ӿ<EFBFBD>ģ<EFBFBD><EFBFBD>, <EFBFBD>ڲ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>msg
* @param req_list -action list ʵ<EFBFBD>ַ<EFBFBD>װ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӿ<EFBFBD>
* @param timeout -<EFBFBD><EFBFBD>ʱʱ<EFBFBD><EFBFBD>, <EFBFBD><EFBFBD>λms
* @return 0 <EFBFBD>ɹ<EFBFBD>, -1 <EFBFBD><EFBFBD>ʼ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʧ<EFBFBD><EFBFBD>, <EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ɹ<EFBFBD><EFBFBD>򲿷ֳɹ<EFBFBD>
*/
int NS_MICRO_THREAD::mt_msg_sendrcv(IMtActList& req_list, int timeout)
{
int iRet = 0;
// <20><>һ<EFBFBD><D2BB>, <20><>ʼ<EFBFBD><CABC>action<6F><6E><EFBFBD><EFBFBD>, <20><>װ<EFBFBD><D7B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
IMtAction* pAction = *it;
if (!pAction || pAction->InitConnEnv() < 0)
{
MTLOG_ERROR("invalid action(%p) or int failed, error", pAction);
return -1;
}
iRet = pAction->DoEncode();
if (iRet < 0)
{
pAction->SetErrno(ERR_ENCODE_ERROR);
MTLOG_ERROR("pack action pkg failed, act %p, ret %d", pAction, iRet);
continue;
}
}
// <20>ڶ<EFBFBD><DAB6><EFBFBD>, ͬ<><CDAC><EFBFBD>շ<EFBFBD><D5B7><EFBFBD>Ϣ, ʧ<><CAA7>Ҳ<EFBFBD><D2B2>Ҫ֪ͨ<CDA8><D6AA><EFBFBD><EFBFBD>
mt_multi_sendrcv_ex(req_list, timeout);
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>, ͬ<><CDAC>֪ͨ<CDA8><D6AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
IMtAction* pAction = *it;
if (pAction->GetMsgFlag() != MULTI_FLAG_FIN)
{
pAction->DoError();
MTLOG_DEBUG("send recv failed: %d", pAction->GetErrno());
continue;
}
iRet = pAction->DoProcess();
if (iRet < 0)
{
MTLOG_DEBUG("action process failed: %d", iRet);
continue;
}
}
// <20><><EFBFBD>IJ<EFBFBD>, <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڲ<EFBFBD><DAB2><EFBFBD>Դ, <20><><EFBFBD>ݸ<EFBFBD><DDB8><EFBFBD><EFBFBD>÷<EFBFBD>
for (IMtActList::iterator it = req_list.begin(); it != req_list.end(); ++it)
{
IMtAction* pAction = *it;
pAction->Reset();
}
return 0;
}