vitalif
/
f-stack
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add_micro_thread_auto_adjust_thread_cnt

dev
woolen 2019-12-26 16:59:57 +08:00
parent 33d130f74c
commit 84bcae25a9
4 changed files with 139 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
app/micro_thread/micro_thread.cpp
View File

@ -49,7 +49,7 @@ Thread::Thread(int stack_size)
memset(&_jmpbuf, 0, sizeof(_jmpbuf));
}
static DefaultLogAdapter def_log_adapt;
/**
* @brief LINUX x86/x86_64's allocated stacks.
*/
@ -75,7 +75,7 @@ bool Thread::InitStack()
void* vaddr = mmap(NULL, memsize, PROT_READ | PROT_WRITE, mmap_flags, zero_fd, 0);
if (vaddr == (void *)MAP_FAILED)
{
MTLOG_ERROR("mmap stack failed, size %d", memsize);
MTLOG_ERROR("mmap stack failed, size %d,errmsg: %s.", memsize,strerror(errno));
free(_stack);
_stack = NULL;
return false;
@ -130,6 +130,12 @@ void Thread::SwitchContext()
}
}
int Thread::SaveContext()
{
return save_context(_jmpbuf);
}
void Thread::RestoreContext()
{
restore_context(_jmpbuf, 1);
@ -375,6 +381,7 @@ void ScheduleObj::ScheduleStartRun()
unsigned int ThreadPool::default_thread_num = DEFAULT_THREAD_NUM; ///< 2000 micro threads.
unsigned int ThreadPool::last_default_thread_num = DEFAULT_THREAD_NUM; ///< 2000 micro threads.
unsigned int ThreadPool::default_stack_size = DEFAULT_STACK_SIZE; ///< 128k stack.
bool ThreadPool::InitialPool(int max_num)
@ -442,6 +449,7 @@ MicroThread* ThreadPool::AllocThread()
if (_total_num >= _max_num)
{
MT_ATTR_API(361140, 1); // no more quota
MTLOG_ERROR("total %d is outof max: %d", _total_num,_max_num);
return NULL;
}
@ -455,6 +463,12 @@ MicroThread* ThreadPool::AllocThread()
}
_total_num++;
_use_num++;
if(_use_num >(int) default_thread_num){
if(((int) default_thread_num * 2 )< _max_num){
last_default_thread_num = default_thread_num;
default_thread_num = default_thread_num * 2;
}
}
return thread;
}
@ -475,6 +489,10 @@ void ThreadPool::FreeThread(MicroThread* thread)
thread->Destroy();
delete thread;
_total_num--;
if(default_thread_num / 2 >= DEFAULT_THREAD_NUM){
last_default_thread_num = default_thread_num;
default_thread_num = default_thread_num / 2;
}
}
}
@ -500,7 +518,11 @@ void MtFrame::SetHookFlag() {
bool MtFrame::InitFrame(LogAdapter* logadpt, int max_thread_num)
{
_log_adpt = logadpt;
if(logadpt == NULL){
_log_adpt = &def_log_adapt;
}else{
_log_adpt = logadpt;
}
if ((this->InitKqueue(max_thread_num) < 0) || !this->InitialPool(max_thread_num))
{
@ -851,6 +873,42 @@ void MtFrame::WaitNotify(utime64_t timeout)
thread->SwitchContext();
}
void MtFrame::NotifyThread(MicroThread* thread)
{
if(thread == NULL){
return;
}
MicroThread* cur_thread = GetActiveThread();
if (thread->HasFlag(MicroThread::IO_LIST))
{
this->RemoveIoWait(thread);
if(cur_thread == this->DaemonThread()){
// 这里不直接切的话,还是不及时,会导致目标线程等待到超时
if(cur_thread->SaveContext() == 0){
this->SetActiveThread(thread);
thread->SetState(MicroThread::RUNNING);
thread->RestoreContext();
}
}else{
this->InsertRunable(thread);
}
}
}
void MtFrame::SwapDaemonThread()
{
MicroThread* thread = GetActiveThread();
MicroThread* daemon_thread = this->DaemonThread();
if(thread != daemon_thread){
if(thread->SaveContext() == 0){
this->InsertRunable(thread);
this->SetActiveThread(daemon_thread);
daemon_thread->SetState(MicroThread::RUNNING);
daemon_thread->RestoreContext();
}
}
}
bool MtFrame::KqueueSchedule(KqObjList* fdlist, KqueuerObj* fd, int timeout)
{
MicroThread* thread = GetActiveThread();

53
app/micro_thread/micro_thread.h
View File

@ -40,6 +40,7 @@
#include <signal.h>
#include <errno.h>
#include <setjmp.h>
#include <stdarg.h>
#include <set>
#include <vector>
@ -56,8 +57,9 @@ namespace NS_MICRO_THREAD {
#define STACK_PAD_SIZE 128
#define MEM_PAGE_SIZE 4096
#define DEFAULT_STACK_SIZE 128*1024
#define DEFAULT_THREAD_NUM 2000
#define DEFAULT_STACK_SIZE STACK_PAD_SIZE * 1024
#define DEFAULT_THREAD_NUM 5000
#define MAX_THREAD_NUM 800000
typedef unsigned long long utime64_t;
typedef void (*ThreadStart)(void*);
@ -119,6 +121,8 @@ public:
void SwitchContext(void);
int SaveContext(void);
void RestoreContext(void);
utime64_t GetWakeupTime(void) {
@ -317,12 +321,51 @@ public:
};
class DefaultLogAdapter :public LogAdapter
{
public:
bool CheckDebug(){ return false;};
bool CheckTrace(){ return false;};
bool CheckError(){ return false;};
inline void LogDebug(char* fmt, ...){
va_list args;
char szBuff[1024];
va_start(args, fmt);
memset(szBuff, 0, sizeof(szBuff));
vsprintf(szBuff, fmt, args);
va_end(args);
printf("%s\n",szBuff);
};
inline void LogTrace(char* fmt, ...){
va_list args;
char szBuff[1024];
va_start(args, fmt);
memset(szBuff, 0, sizeof(szBuff));
vsprintf(szBuff, fmt, args);
va_end(args);
printf("%s\n",szBuff);
};
inline void LogError(char* fmt, ...){
va_list args;
char szBuff[1024];
va_start(args, fmt);
memset(szBuff, 0, sizeof(szBuff));
vsprintf(szBuff, fmt, args);
va_end(args);
printf("%s\n",szBuff);
};
};
class ThreadPool
{
public:
static unsigned int default_thread_num;
static unsigned int last_default_thread_num;
static unsigned int default_stack_size;
static void SetDefaultThreadNum(unsigned int num) {
@ -403,7 +446,7 @@ public:
MicroThread *GetRootThread();
bool InitFrame(LogAdapter* logadpt = NULL, int max_thread_num = 50000);
bool InitFrame(LogAdapter* logadpt = NULL, int max_thread_num = MAX_THREAD_NUM);
void SetHookFlag();
@ -441,6 +484,10 @@ public:
void WaitNotify(utime64_t timeout);
void NotifyThread(MicroThread* thread);
void SwapDaemonThread();
void RemoveIoWait(MicroThread* thread);
void InsertRunable(MicroThread* thread);

20
app/micro_thread/mt_api.cpp
View File

@ -645,6 +645,26 @@ void* mt_start_thread(void* entry, void* args)
return MtFrame::Instance()->CreateThread((ThreadStart)entry, args, true);
}
void* mt_active_thread()
{
return MtFrame::Instance()->GetActiveThread();
}
void mt_thread_wait(int ms)
{
MtFrame::Instance()->WaitNotify(ms);
}
void mt_thread_wakeup_wait(void * thread_p)
{
MtFrame::Instance()->NotifyThread((MicroThread *) thread_p);
}
void mt_swap_thread()
{
return MtFrame::Instance()->SwapDaemonThread();
}
#define BUF_ALIGNMENT_SIZE 4096
#define BUF_ALIGN_SIZE(x) (((x)+BUF_ALIGNMENT_SIZE-1)&~(BUF_ALIGNMENT_SIZE-1))
#define BUF_DEFAULT_SIZE 4096

8
app/micro_thread/mt_api.h
View File

@ -134,6 +134,14 @@ int mt_wait_events(int fd, int events, int timeout);
void* mt_start_thread(void* entry, void* args);
void* mt_active_thread();
void mt_thread_wait(int ms);
void mt_thread_wakeup_wait(void * thread_p);
void mt_swap_thread();
}
#endif