Skip to content

Commit f493794

Browse files
Al2KlimovAl2Klimov
committed
Spawn Configuration::Concurrency process managers
and not just one to increase checks/time.
1 parent a65f2d6 commit f493794

File tree

1 file changed

+67
-38
lines changed

1 file changed

+67
-38
lines changed

‎lib/base/process.cpp

+67-38
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include "base/process.hpp"
44
#include "base/exception.hpp"
55
#include "base/convert.hpp"
6+
#include "base/configuration.hpp"
67
#include "base/array.hpp"
78
#include "base/objectlock.hpp"
89
#include "base/utility.hpp"
@@ -11,8 +12,10 @@
1112
#include "base/utility.hpp"
1213
#include "base/scriptglobal.hpp"
1314
#include "base/json.hpp"
15+
#include <algorithm>
1416
#include <boost/algorithm/string/join.hpp>
1517
#include <boost/thread/once.hpp>
18+
#include <cstddef>
1619
#include <thread>
1720
#include <iostream>
1821

@@ -33,6 +36,21 @@ extern char **environ;
3336
using namespace icinga;
3437

3538
#define IOTHREADS 4
39+
#define MySpawner l_ProcessControl.Spawners[std::hash<decltype(this)>()(this) % l_ProcessControl.Len]
40+
41+
struct Spawner
42+
{
43+
std::mutex Mutex;
44+
int FD = -1;
45+
pid_t PID = -1;
46+
47+
void StartSpawnProcessHelper();
48+
void ProcessHandler();
49+
pid_t ProcessSpawn(const std::vector<String>& arguments, const Dictionary::Ptr& extraEnvironment, bool adjustPriority, int fds[3]);
50+
int ProcessKill(pid_t pid, int signum);
51+
int ProcessWaitPID(pid_t pid, int *status);
52+
Value ProcessSpawnImpl(struct msghdr *msgh, const Dictionary::Ptr& request);
53+
};
3654

3755
static std::mutex l_ProcessMutex[IOTHREADS];
3856
static std::map<Process::ProcessHandle, Process::Ptr> l_Processes[IOTHREADS];
@@ -42,9 +60,10 @@ static HANDLE l_Events[IOTHREADS];
4260
static int l_EventFDs[IOTHREADS][2];
4361
static std::map<Process::ConsoleHandle, Process::ProcessHandle> l_FDs[IOTHREADS];
4462

45-
static std::mutex l_ProcessControlMutex;
46-
static int l_ProcessControlFD = -1;
47-
static pid_t l_ProcessControlPID;
63+
static struct {
64+
Spawner* Spawners = nullptr;
65+
size_t Len = 0;
66+
} l_ProcessControl;
4867
#endif /* _WIN32 */
4968
static boost::once_flag l_ProcessOnceFlag = BOOST_ONCE_INIT;
5069
static boost::once_flag l_SpawnHelperOnceFlag = BOOST_ONCE_INIT;
@@ -72,7 +91,7 @@ Process::~Process()
7291
}
7392

7493
#ifndef _WIN32
75-
static Value ProcessSpawnImpl(struct msghdr *msgh, const Dictionary::Ptr& request)
94+
Value Spawner::ProcessSpawnImpl(struct msghdr *msgh, const Dictionary::Ptr& request)
7695
{
7796
struct cmsghdr *cmsg = CMSG_FIRSTHDR(msgh);
7897

@@ -147,7 +166,7 @@ static Value ProcessSpawnImpl(struct msghdr *msgh, const Dictionary::Ptr& reques
147166
if (pid == 0) {
148167
// child process
149168

150-
(void)close(l_ProcessControlFD);
169+
(void)close(FD);
151170

152171
if (setsid() < 0) {
153172
perror("setsid() failed");
@@ -253,13 +272,13 @@ static Value ProcessWaitPIDImpl(struct msghdr *msgh, const Dictionary::Ptr& requ
253272
return response;
254273
}
255274

256-
static void ProcessHandler()
275+
void Spawner::ProcessHandler()
257276
{
258277
sigset_t mask;
259278
sigfillset(&mask);
260279
sigprocmask(SIG_SETMASK, &mask, nullptr);
261280

262-
Utility::CloseAllFDs({0, 1, 2, l_ProcessControlFD});
281+
Utility::CloseAllFDs({0, 1, 2, FD});
263282

264283
for (;;) {
265284
size_t length;
@@ -278,7 +297,7 @@ static void ProcessHandler()
278297
msg.msg_control = cbuf;
279298
msg.msg_controllen = sizeof(cbuf);
280299

281-
int rc = recvmsg(l_ProcessControlFD, &msg, 0);
300+
int rc = recvmsg(FD, &msg, 0);
282301

283302
if (rc <= 0) {
284303
if (rc < 0 && (errno == EINTR || errno == EAGAIN))
@@ -291,7 +310,7 @@ static void ProcessHandler()
291310

292311
size_t count = 0;
293312
while (count < length) {
294-
rc = recv(l_ProcessControlFD, mbuf + count, length - count, 0);
313+
rc = recv(FD, mbuf + count, length - count, 0);
295314

296315
if (rc <= 0) {
297316
if (rc < 0 && (errno == EINTR || errno == EAGAIN))
@@ -329,7 +348,7 @@ static void ProcessHandler()
329348

330349
String jresponse = JsonEncode(response);
331350

332-
if (send(l_ProcessControlFD, jresponse.CStr(), jresponse.GetLength(), 0) < 0) {
351+
if (send(FD, jresponse.CStr(), jresponse.GetLength(), 0) < 0) {
333352
BOOST_THROW_EXCEPTION(posix_error()
334353
<< boost::errinfo_api_function("send")
335354
<< boost::errinfo_errno(errno));
@@ -339,13 +358,13 @@ static void ProcessHandler()
339358
_exit(0);
340359
}
341360

342-
static void StartSpawnProcessHelper()
361+
void Spawner::StartSpawnProcessHelper()
343362
{
344-
if (l_ProcessControlFD != -1) {
345-
(void)close(l_ProcessControlFD);
363+
if (FD != -1) {
364+
(void)close(FD);
346365

347366
int status;
348-
(void)waitpid(l_ProcessControlPID, &status, 0);
367+
(void)waitpid(PID, &status, 0);
349368
}
350369

351370
int controlFDs[2];
@@ -366,7 +385,7 @@ static void StartSpawnProcessHelper()
366385
if (pid == 0) {
367386
(void)close(controlFDs[1]);
368387

369-
l_ProcessControlFD = controlFDs[0];
388+
FD = controlFDs[0];
370389

371390
ProcessHandler();
372391

@@ -375,11 +394,11 @@ static void StartSpawnProcessHelper()
375394

376395
(void)close(controlFDs[0]);
377396

378-
l_ProcessControlFD = controlFDs[1];
379-
l_ProcessControlPID = pid;
397+
FD = controlFDs[1];
398+
PID = pid;
380399
}
381400

382-
static pid_t ProcessSpawn(const std::vector<String>& arguments, const Dictionary::Ptr& extraEnvironment, bool adjustPriority, int fds[3])
401+
pid_t Spawner::ProcessSpawn(const std::vector<String>& arguments, const Dictionary::Ptr& extraEnvironment, bool adjustPriority, int fds[3])
383402
{
384403
Dictionary::Ptr request = new Dictionary({
385404
{ "command", "spawn" },
@@ -391,7 +410,7 @@ static pid_t ProcessSpawn(const std::vector<String>& arguments, const Dictionary
391410
String jrequest = JsonEncode(request);
392411
size_t length = jrequest.GetLength();
393412

394-
std::unique_lock<std::mutex> lock(l_ProcessControlMutex);
413+
std::unique_lock<std::mutex> lock(Mutex);
395414

396415
struct msghdr msg;
397416
memset(&msg, 0, sizeof(msg));
@@ -417,14 +436,14 @@ static pid_t ProcessSpawn(const std::vector<String>& arguments, const Dictionary
417436
msg.msg_controllen = cmsg->cmsg_len;
418437

419438
do {
420-
while (sendmsg(l_ProcessControlFD, &msg, 0) < 0) {
439+
while (sendmsg(FD, &msg, 0) < 0) {
421440
StartSpawnProcessHelper();
422441
}
423-
} while (send(l_ProcessControlFD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
442+
} while (send(FD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
424443

425444
char buf[4096];
426445

427-
ssize_t rc = recv(l_ProcessControlFD, buf, sizeof(buf), 0);
446+
ssize_t rc = recv(FD, buf, sizeof(buf), 0);
428447

429448
if (rc <= 0)
430449
return -1;
@@ -439,7 +458,7 @@ static pid_t ProcessSpawn(const std::vector<String>& arguments, const Dictionary
439458
return response->Get("rc");
440459
}
441460

442-
static int ProcessKill(pid_t pid, int signum)
461+
int Spawner::ProcessKill(pid_t pid, int signum)
443462
{
444463
Dictionary::Ptr request = new Dictionary({
445464
{ "command", "kill" },
@@ -450,17 +469,17 @@ static int ProcessKill(pid_t pid, int signum)
450469
String jrequest = JsonEncode(request);
451470
size_t length = jrequest.GetLength();
452471

453-
std::unique_lock<std::mutex> lock(l_ProcessControlMutex);
472+
std::unique_lock<std::mutex> lock(Mutex);
454473

455474
do {
456-
while (send(l_ProcessControlFD, &length, sizeof(length), 0) < 0) {
475+
while (send(FD, &length, sizeof(length), 0) < 0) {
457476
StartSpawnProcessHelper();
458477
}
459-
} while (send(l_ProcessControlFD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
478+
} while (send(FD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
460479

461480
char buf[4096];
462481

463-
ssize_t rc = recv(l_ProcessControlFD, buf, sizeof(buf), 0);
482+
ssize_t rc = recv(FD, buf, sizeof(buf), 0);
464483

465484
if (rc <= 0)
466485
return -1;
@@ -471,7 +490,7 @@ static int ProcessKill(pid_t pid, int signum)
471490
return response->Get("errno");
472491
}
473492

474-
static int ProcessWaitPID(pid_t pid, int *status)
493+
int Spawner::ProcessWaitPID(pid_t pid, int *status)
475494
{
476495
Dictionary::Ptr request = new Dictionary({
477496
{ "command", "waitpid" },
@@ -481,17 +500,17 @@ static int ProcessWaitPID(pid_t pid, int *status)
481500
String jrequest = JsonEncode(request);
482501
size_t length = jrequest.GetLength();
483502

484-
std::unique_lock<std::mutex> lock(l_ProcessControlMutex);
503+
std::unique_lock<std::mutex> lock(Mutex);
485504

486505
do {
487-
while (send(l_ProcessControlFD, &length, sizeof(length), 0) < 0) {
506+
while (send(FD, &length, sizeof(length), 0) < 0) {
488507
StartSpawnProcessHelper();
489508
}
490-
} while (send(l_ProcessControlFD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
509+
} while (send(FD, jrequest.CStr(), jrequest.GetLength(), 0) < 0);
491510

492511
char buf[4096];
493512

494-
ssize_t rc = recv(l_ProcessControlFD, buf, sizeof(buf), 0);
513+
ssize_t rc = recv(FD, buf, sizeof(buf), 0);
495514

496515
if (rc <= 0)
497516
return -1;
@@ -505,8 +524,18 @@ static int ProcessWaitPID(pid_t pid, int *status)
505524

506525
void Process::InitializeSpawnHelper()
507526
{
508-
if (l_ProcessControlFD == -1)
509-
StartSpawnProcessHelper();
527+
if (!l_ProcessControl.Spawners) {
528+
auto len (std::max(1, Configuration::Concurrency));
529+
530+
l_ProcessControl.Spawners = new Spawner[len];
531+
l_ProcessControl.Len = len;
532+
}
533+
534+
for (Spawner *current = l_ProcessControl.Spawners, *stop = l_ProcessControl.Spawners + l_ProcessControl.Len; current < stop; ++current) {
535+
if (current->FD == -1) {
536+
current->StartSpawnProcessHelper();
537+
}
538+
}
510539
}
511540
#endif /* _WIN32 */
512541

@@ -992,7 +1021,7 @@ void Process::Run(const std::function<void(const ProcessResult&)>& callback)
9921021
fds[1] = outfds[1];
9931022
fds[2] = outfds[1];
9941023

995-
m_Process = ProcessSpawn(m_Arguments, m_ExtraEnvironment, m_AdjustPriority, fds);
1024+
m_Process = MySpawner.ProcessSpawn(m_Arguments, m_ExtraEnvironment, m_AdjustPriority, fds);
9961025
m_PID = m_Process;
9971026

9981027
if (m_PID == -1) {
@@ -1058,7 +1087,7 @@ bool Process::DoEvents()
10581087

10591088
m_OutputStream << "<Timeout exceeded.>";
10601089

1061-
int error = ProcessKill(m_Process, SIGTERM);
1090+
int error = MySpawner.ProcessKill(m_Process, SIGTERM);
10621091
if (error) {
10631092
Log(LogWarning, "Process")
10641093
<< "Couldn't terminate the process " << m_PID << " (" << PrettyPrintArguments(m_Arguments)
@@ -1082,7 +1111,7 @@ bool Process::DoEvents()
10821111
m_OutputStream << "<Timeout exceeded.>";
10831112
TerminateProcess(m_Process, 3);
10841113
#else /* _WIN32 */
1085-
int error = ProcessKill(-m_Process, SIGKILL);
1114+
int error = MySpawner.ProcessKill(-m_Process, SIGKILL);
10861115
if (error) {
10871116
Log(LogWarning, "Process")
10881117
<< "Couldn't kill the process group " << m_PID << " (" << PrettyPrintArguments(m_Arguments)
@@ -1138,7 +1167,7 @@ bool Process::DoEvents()
11381167
int status, exitcode;
11391168
if (could_not_kill || m_PID == -1) {
11401169
exitcode = 128;
1141-
} else if (ProcessWaitPID(m_Process, &status) != m_Process) {
1170+
} else if (MySpawner.ProcessWaitPID(m_Process, &status) != m_Process) {
11421171
exitcode = 128;
11431172

11441173
Log(LogWarning, "Process")

0 commit comments

Comments
 (0)