icinga2/lib/base/process-unix.cpp

396 lines
8.9 KiB
C++

/******************************************************************************
* Icinga 2 *
* Copyright (C) 2012 Icinga Development Team (http://www.icinga.org/) *
* *
* This program is free software; you can redistribute it and/or *
* modify it under the terms of the GNU General Public License *
* as published by the Free Software Foundation; either version 2 *
* of the License, or (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU General Public License for more details. *
* *
* You should have received a copy of the GNU General Public License *
* along with this program; if not, write to the Free Software Foundation *
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. *
******************************************************************************/
#include "i2-base.h"
#ifndef _WIN32
#include <execvpe.h>
using namespace icinga;
condition_variable Process::m_CV;
int Process::m_TaskFd;
Timer::Ptr Process::m_StatusTimer;
extern char **environ;
void Process::Initialize(void)
{
int fds[2];
#if HAVE_PIPE2
if (pipe2(fds, O_CLOEXEC) < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("pipe2")
<< errinfo_errno(errno));
}
#else /* HAVE_PIPE2 */
if (pipe(fds) < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("pipe")
<< errinfo_errno(errno));
}
/* Don't bother setting fds[0] to clo-exec as we'll only
* use it in the following dup() call. */
Utility::SetCloExec(fds[1]);
#endif /* HAVE_PIPE2 */
m_TaskFd = fds[1];
unsigned int threads = thread::hardware_concurrency();
if (threads == 0)
threads = 2;
for (unsigned int i = 0; i < threads; i++) {
int childTaskFd = dup(fds[0]);
if (childTaskFd < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("dup")
<< errinfo_errno(errno));
}
Utility::SetNonBlocking(childTaskFd);
Utility::SetCloExec(childTaskFd);
thread t(&Process::WorkerThreadProc, childTaskFd);
t.detach();
}
(void) close(fds[0]);
m_StatusTimer = boost::make_shared<Timer>();
m_StatusTimer->OnTimerExpired.connect(boost::bind(&Process::StatusTimerHandler));
m_StatusTimer->SetInterval(5);
m_StatusTimer->Start();
}
void Process::WorkerThreadProc(int taskFd)
{
map<int, Process::Ptr> tasks;
pollfd *pfds = NULL;
for (;;) {
map<int, Process::Ptr>::iterator it, prev;
pfds = (pollfd *)realloc(pfds, (1 + tasks.size()) * sizeof(pollfd));
if (pfds == NULL) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("realloc")
<< errinfo_errno(errno));
}
int idx = 0;
int fd;
BOOST_FOREACH(tie(fd, tuples::ignore), tasks) {
pfds[idx].fd = fd;
pfds[idx].events = POLLIN | POLLHUP;
idx++;
}
if (tasks.size() < MaxTasksPerThread) {
pfds[idx].fd = taskFd;
pfds[idx].events = POLLIN;
idx++;
}
int rc = poll(pfds, idx, -1);
if (rc < 0 && errno != EINTR) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("poll")
<< errinfo_errno(errno));
}
if (rc == 0)
continue;
for (int i = 0; i < idx; i++) {
if ((pfds[i].revents & (POLLIN|POLLHUP)) == 0)
continue;
if (pfds[i].fd == taskFd) {
vector<Process::Ptr> new_tasks;
unsigned int want = MaxTasksPerThread - tasks.size();
if (want > 0) {
boost::mutex::scoped_lock lock(m_Mutex);
/* Read one byte for every task we take from the pending tasks list. */
char buffer[MaxTasksPerThread];
ASSERT(want <= sizeof(buffer));
int have = read(taskFd, &buffer, want);
if (have < 0) {
if (errno == EAGAIN)
break; /* Someone else was faster and took our task. */
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("read")
<< errinfo_errno(errno));
}
while (have > 0) {
ASSERT(!m_Tasks.empty());
Process::Ptr task = m_Tasks.front();
m_Tasks.pop_front();
new_tasks.push_back(task);
have--;
}
m_CV.notify_all();
}
BOOST_FOREACH(const Process::Ptr& task, new_tasks) {
try {
task->InitTask();
int fd = task->m_FD;
if (fd >= 0)
tasks[fd] = task;
} catch (...) {
task->FinishException(boost::current_exception());
}
}
continue;
}
it = tasks.find(pfds[i].fd);
if (it == tasks.end())
continue;
Process::Ptr task = it->second;
if (!task->RunTask()) {
prev = it;
tasks.erase(prev);
task->FinishResult(task->m_Result);
}
}
}
}
void Process::QueueTask(void)
{
{
boost::mutex::scoped_lock lock(m_Mutex);
while (m_Tasks.size() >= PIPE_BUF)
m_CV.wait(lock);
m_Tasks.push_back(GetSelf());
/**
* This little gem which is commonly known as the "self-pipe trick"
* takes care of waking up the select() call in the worker thread.
*/
if (write(m_TaskFd, "T", 1) < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("write")
<< errinfo_errno(errno));
}
}
}
void Process::InitTask(void)
{
m_Result.ExecutionStart = Utility::GetTime();
ASSERT(m_FD == -1);
int fds[2];
#if HAVE_PIPE2
if (pipe2(fds, O_NONBLOCK | O_CLOEXEC) < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("pipe2")
<< errinfo_errno(errno));
}
#else /* HAVE_PIPE2 */
if (pipe(fds) < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("pipe")
<< errinfo_errno(errno));
}
Utility::SetNonBlocking(fds[0]);
Utility::SetCloExec(fds[0]);
Utility::SetNonBlocking(fds[1]);
Utility::SetCloExec(fds[1]);
#endif /* HAVE_PIPE2 */
// build argv
char **argv = new char *[m_Arguments.size() + 1];
for (unsigned int i = 0; i < m_Arguments.size(); i++)
argv[i] = strdup(m_Arguments[i].CStr());
argv[m_Arguments.size()] = NULL;
m_Arguments.clear();
// build envp
int envc = 0;
/* count existing environment variables */
while (environ[envc] != NULL)
envc++;
char **envp = new char *[envc + (m_ExtraEnvironment ? m_ExtraEnvironment->GetLength() : 0) + 1];
for (int i = 0; i < envc; i++)
envp[i] = strdup(environ[i]);
if (m_ExtraEnvironment) {
ObjectLock olock(m_ExtraEnvironment);
String key;
Value value;
int index = envc;
BOOST_FOREACH(tie(key, value), m_ExtraEnvironment) {
String kv = key + "=" + Convert::ToString(value);
envp[index] = strdup(kv.CStr());
index++;
}
}
envp[envc + (m_ExtraEnvironment ? m_ExtraEnvironment->GetLength() : 0)] = NULL;
m_ExtraEnvironment.reset();
#if HAVE_WORKING_VFORK
m_Pid = vfork();
#else /* HAVE_WORKING_VFORK */
m_Pid = fork();
#endif /* HAVE_WORKING_VFORK */
if (m_Pid < 0) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("fork")
<< errinfo_errno(errno));
}
if (m_Pid == 0) {
// child process
if (dup2(fds[1], STDOUT_FILENO) < 0 || dup2(fds[1], STDERR_FILENO) < 0) {
perror("dup2() failed.");
_exit(128);
}
(void) close(fds[0]);
(void) close(fds[1]);
if (execvpe(argv[0], argv, envp) < 0) {
perror("execvpe() failed.");
_exit(128);
}
_exit(128);
}
// parent process
// free arguments
for (int i = 0; argv[i] != NULL; i++)
free(argv[i]);
delete [] argv;
// free environment
for (int i = 0; envp[i] != NULL; i++)
free(envp[i]);
delete [] envp;
m_FD = fds[0];
(void) close(fds[1]);
}
bool Process::RunTask(void)
{
char buffer[512];
int rc;
do {
rc = read(m_FD, buffer, sizeof(buffer));
if (rc > 0) {
m_OutputStream.write(buffer, rc);
}
} while (rc > 0);
if (rc < 0 && errno == EAGAIN)
return true;
String output = m_OutputStream.str();
int status, exitcode;
(void) close(m_FD);
if (waitpid(m_Pid, &status, 0) != m_Pid) {
BOOST_THROW_EXCEPTION(posix_error()
<< errinfo_api_function("waitpid")
<< errinfo_errno(errno));
}
if (WIFEXITED(status)) {
exitcode = WEXITSTATUS(status);
} else if (WIFSIGNALED(status)) {
stringstream outputbuf;
outputbuf << "Process was terminated by signal " << WTERMSIG(status);
output = outputbuf.str();
exitcode = 128;
} else {
exitcode = 128;
}
m_Result.ExecutionEnd = Utility::GetTime();
m_Result.ExitStatus = exitcode;
m_Result.Output = output;
return false;
}
void Process::StatusTimerHandler(void)
{
boost::mutex::scoped_lock lock(m_Mutex);
if (m_Tasks.size() > 50)
Logger::Write(LogCritical, "base", "More than 50 waiting Process tasks: " +
Convert::ToString(m_Tasks.size()));
}
#endif /* _WIN32 */