jobserver/processmanager.cpp at master · Winnerhust/jobserver · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
#ifndef _MPROCESSMANAGER_CPP
#define _MPROCESSMANAGER_CPP

#include <iostream>
#include <list>
#include "processmanager.h"
using namespace std;

list<Worker> Processmanager::m_workers;
int default_adjust_strategy(long jobs_count,Processmanager &pm)
{
  long adj_num = jobs_count/10;

	if (adj_num > pm.get_maxprocnum()){
		adj_num = pm.get_maxprocnum();
	}
	else if (adj_num < pm.get_minprocnum()){
		adj_num = pm.get_minprocnum();
	}

	return adj_num;
}

Processmanager::Processmanager()
:adjust_proc_strategy(NULL),m_maxprocnum(10),m_minprocnum(1),m_jobs_count(0)
{
	signal(SIGCHLD,sig_child);
	signal(SIGTERM,sig_term);
}
Processmanager::~Processmanager()
{
}
void Processmanager::init_num_process(int count)
{
	Worker *w=NULL;
	if (m_workers.empty()){
		for (int i = 0; i < count ; ++i){
			w = new Worker();
			if (w ==NULL){
				throw puts("create Worker failed,memory not enough");
			}
			m_workers.push_back(*w);
		}
	}
}

void Processmanager::start()
{
	for (list<Worker>::iterator i=m_workers.begin();i!=m_workers.end();++i){
		i->start();
	}
}

void Processmanager::sig_child(int sig)
{
	int recode;
	int pid=waitpid(-1,&recode,WNOHANG);

	for (list<Worker>::iterator i=m_workers.begin();i!=m_workers.end();++i){
		if (pid == i->pid()){
			i->set_status(false);
			break;
		}
	}

	signal(SIGCLD,sig_child);
}
void Processmanager::sig_term(int sig)
{
	if (get_one_proc_by_status(true)){
		cout<<"some Worker process is still running"<<endl;
	}
	exit(0);
}
Worker * Processmanager::get_one_proc_by_type(const string &type)
{
	for (list<Worker>::iterator i=m_workers.begin();i!=m_workers.end();++i){
		if (i->type() == type){
			return &(*i);
		}
	 }

	 return NULL;
}

Worker * Processmanager::get_one_proc_by_status(bool status)
{
	for (list<Worker>::iterator i=m_workers.begin();i!=m_workers.end();++i){
        if (i->status() == status){
            return &(*i);
        }
    }
	return NULL;
}

void Processmanager::adjust_proc_num()
{
	if (adjust_proc_strategy==NULL){
		adjust_proc_strategy = default_adjust_strategy;
	}
	int adj_num = adjust_proc_strategy(m_jobs_count,*this);

	Worker * w;

	while (cur_procnum() < adj_num){
		cout<<"+++++"<<endl;
		w=get_one_proc_by_status(false);
		w->start();
	}
//	while (cur_procnum() > adj_num){
	if (cur_procnum() > adj_num){
		cout<<"-----"<<endl;
		w=get_one_proc_by_status(true);
		w->stop();
	}
}
int Processmanager::cur_procnum()
{
	int count=0;
	for(list<Worker>::iterator i=m_workers.begin();i!=m_workers.end();++i){
		if (i->status()){
			++count;
		}
	}

	return count;
}
#endif