1022753986

Committed 02 Oct 2023 10:37AM UTC coverage: 78.168% (-2.0%) from 80.127%

Build # 1022753986

Build Type

push

gitlab-ci

Committed by

oleorhagen

Commit Message

feat: Run the authentication loop once upon bootstrap

Ticket: MEN-6658
Changelog: None

Signed-off-by: Ole Petter <ole.orhagen@northern.tech>

Run Details

32 of 32 new or added lines in 1 file covered. (100.0%)

6996 of 8950 relevant lines covered (78.17%)

10353.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.73

/common/processes/platform/tiny_process_library/tiny_process_library.cpp

// Copyright 2023 Northern.tech AS
//
//    Licensed under the Apache 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
//
//        http://www.apache.org/licenses/LICENSE-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.

#include <common/processes.hpp>

#include <string>
#include <string_view>

#include <common/events_io.hpp>
#include <common/io.hpp>
#include <common/log.hpp>
#include <common/path.hpp>

using namespace std;

namespace mender::common::processes {

const chrono::seconds MAX_TERMINATION_TIME(10);

namespace io = mender::common::io;
namespace log = mender::common::log;
namespace path = mender::common::path;

class ProcessReaderFunctor {
public:
        void operator()(const char *bytes, size_t n);

        // Note: ownership is not held here, but in Process.
        int fd_;

        OutputCallback callback_;
};

Process::Process(vector<string> args) :
        args_ {args},
        max_termination_time_ {MAX_TERMINATION_TIME} {
        async_wait_data_ = make_shared<AsyncWaitData>();
}

Process::~Process() {
        {
                unique_lock lock(async_wait_data_->data_mutex);
                // DoCancel() requires being locked.
                DoCancel();
        }

        EnsureTerminated();

        if (stdout_pipe_ >= 0) {
                close(stdout_pipe_);
        }
        if (stderr_pipe_ >= 0) {
                close(stderr_pipe_);
        }
}

error::Error Process::Start(OutputCallback stdout_callback, OutputCallback stderr_callback) {
        if (proc_) {
                return MakeError(ProcessAlreadyStartedError, "Cannot start process");
        }

        // Tiny-process-library doesn't give a good error if the command isn't found (just returns
        // exit code 1). If the path is absolute, it's pretty easy to check if it exists. This won't
        // cover all errors (non-absolute or unset executable bit, for example), but helps a little,
        // at least.
        if (args_.size() > 0 && path::IsAbsolute(args_[0])) {
                ifstream f(args_[0]);
                if (!f.good()) {
                        int errnum = errno;
                        return error::Error(
                                generic_category().default_error_condition(errnum), "Cannot launch " + args_[0]);
                }
        }

        OutputCallback maybe_stdout_callback;
        if (stdout_pipe_ >= 0 || stdout_callback) {
                maybe_stdout_callback = ProcessReaderFunctor {stdout_pipe_, stdout_callback};
        }
        OutputCallback maybe_stderr_callback;
        if (stderr_pipe_ >= 0 || stderr_callback) {
                maybe_stderr_callback = ProcessReaderFunctor {stderr_pipe_, stderr_callback};
        }

        proc_ =
                make_unique<tpl::Process>(args_, work_dir_, maybe_stdout_callback, maybe_stderr_callback);

        if (proc_->get_id() == -1) {
                proc_.reset();
                return MakeError(
                        ProcessesErrorCode::SpawnError,
                        "Failed to spawn '" + (args_.size() >= 1 ? args_[0] : "<null>") + "'");
        }

        SetupAsyncWait();

        return error::NoError;
}

error::Error Process::Run() {
        auto err = Start();
        if (err != error::NoError) {
                log::Error(err.String());
        }
        return Wait();
}

static error::Error ErrorBasedOnExitStatus(int exit_status) {
        if (exit_status != 0) {
                return MakeError(
                        NonZeroExitStatusError, "Process exited with status " + to_string(exit_status));
        } else {
                return error::NoError;
        }
}

error::Error Process::Wait() {
        if (proc_) {
                exit_status_ = future_exit_status_.get();
                proc_.reset();
                if (stdout_pipe_ >= 0) {
                        close(stdout_pipe_);
                        stdout_pipe_ = -1;
                }
                if (stderr_pipe_ >= 0) {
                        close(stderr_pipe_);
                        stderr_pipe_ = -1;
                }
        }
        return ErrorBasedOnExitStatus(exit_status_);
}

error::Error Process::Wait(chrono::nanoseconds timeout) {
        if (proc_) {
                auto result = future_exit_status_.wait_for(timeout);
                if (result == future_status::timeout) {
                        return error::Error(
                                make_error_condition(errc::timed_out), "Timed out while waiting for process");
                }
                AssertOrReturnError(result == future_status::ready);
                exit_status_ = future_exit_status_.get();
                proc_.reset();
                if (stdout_pipe_ >= 0) {
                        close(stdout_pipe_);
                        stdout_pipe_ = -1;
                }
                if (stderr_pipe_ >= 0) {
                        close(stderr_pipe_);
                        stderr_pipe_ = -1;
                }
        }
        return ErrorBasedOnExitStatus(exit_status_);
}

error::Error Process::AsyncWait(events::EventLoop &loop, AsyncWaitHandler handler) {
        unique_lock lock(async_wait_data_->data_mutex);

        if (async_wait_data_->handler) {
                return error::Error(make_error_condition(errc::operation_in_progress), "Cannot AsyncWait");
        }

        async_wait_data_->event_loop = &loop;
        async_wait_data_->handler = handler;

        if (async_wait_data_->process_ended) {
                // The process has already ended. Schedule the handler immediately.
                auto &async_wait_data = async_wait_data_;
                async_wait_data->event_loop->Post(
                        [async_wait_data, this]() { AsyncWaitInternalHandler(async_wait_data); });
        }

        return error::NoError;
}

error::Error Process::AsyncWait(
        events::EventLoop &loop, AsyncWaitHandler handler, chrono::nanoseconds timeout) {
        timeout_timer_.reset(new events::Timer(loop));

        auto err = AsyncWait(loop, handler);
        if (err != error::NoError) {
                return err;
        }

        timeout_timer_->AsyncWait(timeout, [this, handler](error::Error err) {
                // Move timer here so it gets destroyed after this handler.
                auto timer = std::move(timeout_timer_);
                // Cancel normal AsyncWait() (the process part of it).
                {
                        // DoCancel() requires being locked.
                        unique_lock lock(async_wait_data_->data_mutex);
                        DoCancel();
                }
                if (err != error::NoError) {
                        handler(err.WithContext("Process::Timer"));
                }

                handler(error::Error(make_error_condition(errc::timed_out), "Process::Timer"));
        });

        return error::NoError;
}

void Process::Cancel() {
        unique_lock lock(async_wait_data_->data_mutex);

        if (async_wait_data_->handler && !async_wait_data_->process_ended) {
                auto &async_wait_data = async_wait_data_;
                auto &handler = async_wait_data->handler;
                async_wait_data_->event_loop->Post([handler]() {
                        handler(error::Error(
                                make_error_condition(errc::operation_canceled), "Process::AsyncWait canceled"));
                });
        }

        // DoCancel() requires being locked.
        DoCancel();
}

void Process::DoCancel() {
        // Should already be locked by caller.
        //   unique_lock lock(async_wait_data_->data_mutex);

        timeout_timer_.reset();

        async_wait_data_->event_loop = nullptr;
        async_wait_data_->handler = nullptr;
        async_wait_data_->process_ended = false;
}

void Process::SetupAsyncWait() {
        future_exit_status_ = async(launch::async, [this]() -> int {
                // This function is executed in a separate thread, but the object is guaranteed to
                // still exist while we're in here (because we call `future_exit_status_.get()`
                // during destruction.

                auto status = proc_->get_exit_status();

                auto &async_wait_data = async_wait_data_;

                unique_lock lock(async_wait_data->data_mutex);

                if (async_wait_data->handler) {
                        async_wait_data_->event_loop->Post(
                                [async_wait_data, this]() { AsyncWaitInternalHandler(async_wait_data); });
                }

                async_wait_data->process_ended = true;

                return status;
        });
}

void Process::AsyncWaitInternalHandler(shared_ptr<AsyncWaitData> async_wait_data) {
        timeout_timer_.reset();

        // This function is meant to be executed on the event loop, and the object may have been
        // either cancelled or destroyed before we get here. By having our own copy of the
        // async_wait_data pointer pointer, it survives destruction, and we can test whether we
        // should still proceed here. So note the use of `async_wait_data` instead of
        // `async_wait_data_`.
        unique_lock lock(async_wait_data->data_mutex);

        if (async_wait_data->handler) {
                auto handler = async_wait_data->handler;

                // For next iteration.
                async_wait_data->event_loop = nullptr;
                async_wait_data->handler = nullptr;
                async_wait_data->process_ended = false;

                auto status = GetExitStatus();

                // Unlock in case the handler calls back into this object.
                lock.unlock();
                handler(ErrorBasedOnExitStatus(status));
        }
}

static void CollectLineData(
        string &trailing_line, vector<string> &lines, const char *bytes, size_t len) {
        auto bytes_view = string_view(bytes, len);
        size_t line_start_idx = 0;
        size_t line_end_idx = bytes_view.find("\n", 0);
        if ((trailing_line != "") && (line_end_idx != string_view::npos)) {
                lines.push_back(trailing_line + string(bytes_view, 0, line_end_idx));
                line_start_idx = line_end_idx + 1;
                line_end_idx = bytes_view.find("\n", line_start_idx);
                trailing_line = "";
        }

        while ((line_start_idx < (len - 1)) && (line_end_idx != string_view::npos)) {
                lines.push_back(string(bytes_view, line_start_idx, (line_end_idx - line_start_idx)));
                line_start_idx = line_end_idx + 1;
                line_end_idx = bytes_view.find("\n", line_start_idx);
        }

        if ((line_end_idx == string_view::npos) && (line_start_idx != (len - 1))) {
                trailing_line += string(bytes_view, line_start_idx, (len - line_start_idx));
        }
}

ExpectedLineData Process::GenerateLineData(chrono::nanoseconds timeout) {
        if (proc_) {
                return expected::unexpected(
                        MakeError(ProcessAlreadyStartedError, "Cannot generate line data"));
        }

        if (args_.size() == 0) {
                return expected::unexpected(MakeError(
                        ProcessesErrorCode::SpawnError, "No arguments given, cannot spawn a process"));
        }

        string trailing_line;
        vector<string> ret;
        proc_ = make_unique<tpl::Process>(
                args_, work_dir_, [&trailing_line, &ret](const char *bytes, size_t len) {
                        CollectLineData(trailing_line, ret, bytes, len);
                });

        if (proc_->get_id() == -1) {
                proc_.reset();
                return expected::unexpected(MakeError(
                        ProcessesErrorCode::SpawnError,
                        "Failed to spawn '" + (args_.size() >= 1 ? args_[0] : "<null>") + "'"));
        }

        SetupAsyncWait();

        auto err = Wait(timeout);
        if (err != error::NoError && err.code != MakeError(NonZeroExitStatusError, "").code) {
                return expected::unexpected(err);
        }

        if (trailing_line != "") {
                ret.push_back(trailing_line);
        }

        return ExpectedLineData(ret);
}

io::ExpectedAsyncReaderPtr Process::GetProcessReader(events::EventLoop &loop, int &pipe_ref) {
        if (proc_) {
                return expected::unexpected(
                        MakeError(ProcessAlreadyStartedError, "Cannot get process output"));
        }

        if (pipe_ref >= 0) {
                close(pipe_ref);
                pipe_ref = -1;
        }

        int fds[2];
        int ret = pipe(fds);
        if (ret < 0) {
                int err = errno;
                return expected::unexpected(error::Error(
                        generic_category().default_error_condition(err),
                        "Could not get process stdout reader"));
        }

        pipe_ref = fds[1];

        return make_shared<events::io::AsyncFileDescriptorReader>(loop, fds[0]);
}

io::ExpectedAsyncReaderPtr Process::GetAsyncStdoutReader(events::EventLoop &loop) {
        return GetProcessReader(loop, stdout_pipe_);
}

io::ExpectedAsyncReaderPtr Process::GetAsyncStderrReader(events::EventLoop &loop) {
        return GetProcessReader(loop, stderr_pipe_);
}

int Process::EnsureTerminated() {
        if (!proc_) {
                return exit_status_;
        }

        log::Info("Sending SIGTERM to PID " + to_string(proc_->get_id()));

        Terminate();

        auto result = future_exit_status_.wait_for(max_termination_time_);
        if (result == future_status::timeout) {
                log::Info("Sending SIGKILL to PID " + to_string(proc_->get_id()));
                Kill();
                result = future_exit_status_.wait_for(max_termination_time_);
                if (result != future_status::ready) {
                        // This should not be possible, SIGKILL always terminates.
                        log::Error(
                                "PID " + to_string(proc_->get_id()) + " still not terminated after SIGKILL.");
                        return -1;
                }
        }
        assert(result == future_status::ready);
        exit_status_ = future_exit_status_.get();

        log::Info(
                "PID " + to_string(proc_->get_id()) + " exited with status " + to_string(exit_status_));

        proc_.reset();

        return exit_status_;
}

void Process::Terminate() {
        if (proc_) {
                // At the time of writing, tiny-process-library kills using SIGINT and SIGTERM, for
                // `force = false/true`, respectively. But we want to kill with SIGTERM and SIGKILL,
                // because:
                //
                // 1. SIGINT is not meant to kill interactive processes, whereas SIGTERM is.
                // 2. SIGKILL is required in order to really force, since SIGTERM can be ignored by
                //    the process.
                //
                // If tiny-process-library is fixed, then this can be restored and the part below
                // removed.
                // proc_->kill(false);

                ::kill(proc_->get_id(), SIGTERM);
                ::kill(-proc_->get_id(), SIGTERM);
        }
}

void Process::Kill() {
        if (proc_) {
                // See comment in Terminate().
                // proc_->kill(true);

                ::kill(proc_->get_id(), SIGKILL);
                ::kill(-proc_->get_id(), SIGKILL);
        }
}

void ProcessReaderFunctor::operator()(const char *bytes, size_t n) {
        if (callback_) {
                callback_(bytes, n);
        }

        if (fd_ < 0) {
                return;
        }

        size_t written = 0;
        while (written < n) {
                auto ret = write(fd_, bytes, n);
                if (ret < 0) {
                        int err = errno;
                        log::Error(
                                string {"Error while writing process output to main thread: "} + strerror(err));
                        fd_ = -1;
                        return;
                }

                written += ret;
        }
}

} // namespace mender::common::processes

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	#include <common/processes.hpp>
16
17	#include <string>
18	#include <string_view>
19
20	#include <common/events_io.hpp>
21	#include <common/io.hpp>
22	#include <common/log.hpp>
23	#include <common/path.hpp>
24
25	using namespace std;
26
27	namespace mender::common::processes {
28
29	const chrono::seconds MAX_TERMINATION_TIME(10);
30
31	namespace io = mender::common::io;
32	namespace log = mender::common::log;
33	namespace path = mender::common::path;
34
35	class ProcessReaderFunctor {
36	public:
37	void operator()(const char *bytes, size_t n);
38
39	// Note: ownership is not held here, but in Process.
40	int fd_;
41
42	OutputCallback callback_;
43	};
44
45	Process::Process(vector<string> args) :	×
46	args_ {args},
47	max_termination_time_ {MAX_TERMINATION_TIME} {	×
48	async_wait_data_ = make_shared<AsyncWaitData>();	×
49	}	×
50
51	Process::~Process() {	×
52	{
53	unique_lock lock(async_wait_data_->data_mutex);	×
54	// DoCancel() requires being locked.
55	DoCancel();	×
56	}
57
58	EnsureTerminated();	×
59
60	if (stdout_pipe_ >= 0) {	×
61	close(stdout_pipe_);	×
62	}
63	if (stderr_pipe_ >= 0) {	×
64	close(stderr_pipe_);	×
65	}
66	}	×
67
68	error::Error Process::Start(OutputCallback stdout_callback, OutputCallback stderr_callback) {	1,946✔
69	if (proc_) {	1,946✔
70	return MakeError(ProcessAlreadyStartedError, "Cannot start process");	×
71	}
72
73	// Tiny-process-library doesn't give a good error if the command isn't found (just returns
74	// exit code 1). If the path is absolute, it's pretty easy to check if it exists. This won't
75	// cover all errors (non-absolute or unset executable bit, for example), but helps a little,
76	// at least.
77	if (args_.size() > 0 && path::IsAbsolute(args_[0])) {	1,946✔
78	ifstream f(args_[0]);	1,625✔
79	if (!f.good()) {	1,625✔
80	int errnum = errno;	×
81	return error::Error(
82	generic_category().default_error_condition(errnum), "Cannot launch " + args_[0]);	×
83	}
84	}
85
86	OutputCallback maybe_stdout_callback;	3,892✔
87	if (stdout_pipe_ >= 0 \|\| stdout_callback) {	1,946✔
88	maybe_stdout_callback = ProcessReaderFunctor {stdout_pipe_, stdout_callback};	1,616✔
89	}
90	OutputCallback maybe_stderr_callback;	3,892✔
91	if (stderr_pipe_ >= 0 \|\| stderr_callback) {	1,946✔
92	maybe_stderr_callback = ProcessReaderFunctor {stderr_pipe_, stderr_callback};	1,616✔
93	}
94
95	proc_ =
96	make_unique<tpl::Process>(args_, work_dir_, maybe_stdout_callback, maybe_stderr_callback);	1,946✔
97
98	if (proc_->get_id() == -1) {	1,946✔
99	proc_.reset();	×
100	return MakeError(
101	ProcessesErrorCode::SpawnError,
102	"Failed to spawn '" + (args_.size() >= 1 ? args_[0] : "<null>") + "'");	×
103	}
104
105	SetupAsyncWait();	1,946✔
106
107	return error::NoError;	1,946✔
108	}
109
110	error::Error Process::Run() {	311✔
111	auto err = Start();	933✔
112	if (err != error::NoError) {	311✔
113	log::Error(err.String());	×
114	}
115	return Wait();	622✔
116	}
117
118	static error::Error ErrorBasedOnExitStatus(int exit_status) {	3,699✔
119	if (exit_status != 0) {	3,699✔
120	return MakeError(
121	NonZeroExitStatusError, "Process exited with status " + to_string(exit_status));	562✔
122	} else {
123	return error::NoError;	3,418✔
124	}
125	}
126
127	error::Error Process::Wait() {	2,026✔
128	if (proc_) {	2,026✔
129	exit_status_ = future_exit_status_.get();	1,924✔
130	proc_.reset();	1,924✔
131	if (stdout_pipe_ >= 0) {	1,924✔
132	close(stdout_pipe_);	2✔
133	stdout_pipe_ = -1;	2✔
134	}
135	if (stderr_pipe_ >= 0) {	1,924✔
136	close(stderr_pipe_);	2✔
137	stderr_pipe_ = -1;	2✔
138	}
139	}
140	return ErrorBasedOnExitStatus(exit_status_);	2,026✔
141	}
142
143	error::Error Process::Wait(chrono::nanoseconds timeout) {	63✔
144	if (proc_) {	63✔
145	auto result = future_exit_status_.wait_for(timeout);	63✔
146	if (result == future_status::timeout) {	63✔
147	return error::Error(
148	make_error_condition(errc::timed_out), "Timed out while waiting for process");	×
149	}
150	AssertOrReturnError(result == future_status::ready);	63✔
151	exit_status_ = future_exit_status_.get();	63✔
152	proc_.reset();	63✔
153	if (stdout_pipe_ >= 0) {	63✔
154	close(stdout_pipe_);	×
155	stdout_pipe_ = -1;	×
156	}
157	if (stderr_pipe_ >= 0) {	63✔
158	close(stderr_pipe_);	×
159	stderr_pipe_ = -1;	×
160	}
161	}
162	return ErrorBasedOnExitStatus(exit_status_);	63✔
163	}
164
165	error::Error Process::AsyncWait(events::EventLoop &loop, AsyncWaitHandler handler) {	1,617✔
166	unique_lock lock(async_wait_data_->data_mutex);	3,234✔
167
168	if (async_wait_data_->handler) {	1,617✔
169	return error::Error(make_error_condition(errc::operation_in_progress), "Cannot AsyncWait");	×
170	}
171
172	async_wait_data_->event_loop = &loop;	1,617✔
173	async_wait_data_->handler = handler;	1,617✔
174
175	if (async_wait_data_->process_ended) {	1,617✔
176	// The process has already ended. Schedule the handler immediately.
177	auto &async_wait_data = async_wait_data_;	×
178	async_wait_data->event_loop->Post(	×
179	[async_wait_data, this]() { AsyncWaitInternalHandler(async_wait_data); });	×
180	}
181
182	return error::NoError;	1,617✔
183	}
184
185	error::Error Process::AsyncWait(	1,614✔
186	events::EventLoop &loop, AsyncWaitHandler handler, chrono::nanoseconds timeout) {
187	timeout_timer_.reset(new events::Timer(loop));	1,614✔
188
189	auto err = AsyncWait(loop, handler);	3,228✔
190	if (err != error::NoError) {	1,614✔
191	return err;	×
192	}
193
194	timeout_timer_->AsyncWait(timeout, [this, handler](error::Error err) {	3,228✔
195	// Move timer here so it gets destroyed after this handler.
196	auto timer = std::move(timeout_timer_);	8✔
197	// Cancel normal AsyncWait() (the process part of it).
198	{
199	// DoCancel() requires being locked.
200	unique_lock lock(async_wait_data_->data_mutex);	8✔
201	DoCancel();	4✔
202	}
203	if (err != error::NoError) {	4✔
204	handler(err.WithContext("Process::Timer"));	×
205	}
206
207	handler(error::Error(make_error_condition(errc::timed_out), "Process::Timer"));	4✔
208	});	3,232✔
209
210	return error::NoError;	1,614✔
211	}
212
213	void Process::Cancel() {	1✔
214	unique_lock lock(async_wait_data_->data_mutex);	2✔
215
216	if (async_wait_data_->handler && !async_wait_data_->process_ended) {	1✔
217	auto &async_wait_data = async_wait_data_;	1✔
218	auto &handler = async_wait_data->handler;	1✔
219	async_wait_data_->event_loop->Post([handler]() {	2✔
220	handler(error::Error(	1✔
221	make_error_condition(errc::operation_canceled), "Process::AsyncWait canceled"));	2✔
222	});	3✔
223	}
224
225	// DoCancel() requires being locked.
226	DoCancel();	1✔
227	}	1✔
228
229	void Process::DoCancel() {	2,015✔
230	// Should already be locked by caller.
231	// unique_lock lock(async_wait_data_->data_mutex);
232
233	timeout_timer_.reset();	2,015✔
234
235	async_wait_data_->event_loop = nullptr;	2,015✔
236	async_wait_data_->handler = nullptr;	2,015✔
237	async_wait_data_->process_ended = false;	2,015✔
238	}	2,015✔
239
240	void Process::SetupAsyncWait() {	2,009✔
241	future_exit_status_ = async(launch::async, [this]() -> int {	2,009✔
242	// This function is executed in a separate thread, but the object is guaranteed to
243	// still exist while we're in here (because we call `future_exit_status_.get()`
244	// during destruction.
245
246	auto status = proc_->get_exit_status();	2,009✔
247
248	auto &async_wait_data = async_wait_data_;	2,009✔
249
250	unique_lock lock(async_wait_data->data_mutex);	2,009✔
251
252	if (async_wait_data->handler) {	2,009✔
253	async_wait_data_->event_loop->Post(	3,220✔
254	[async_wait_data, this]() { AsyncWaitInternalHandler(async_wait_data); });	4,830✔
255	}
256
257	async_wait_data->process_ended = true;	2,009✔
258
259	return status;	4,018✔
260	});	2,009✔
261	}	2,009✔
262
263	void Process::AsyncWaitInternalHandler(shared_ptr<AsyncWaitData> async_wait_data) {	1,610✔
264	timeout_timer_.reset();	1,610✔
265
266	// This function is meant to be executed on the event loop, and the object may have been
267	// either cancelled or destroyed before we get here. By having our own copy of the
268	// async_wait_data pointer pointer, it survives destruction, and we can test whether we
269	// should still proceed here. So note the use of `async_wait_data` instead of
270	// `async_wait_data_`.
271	unique_lock lock(async_wait_data->data_mutex);	3,220✔
272
273	if (async_wait_data->handler) {	1,610✔
274	auto handler = async_wait_data->handler;	1,610✔
275
276	// For next iteration.
277	async_wait_data->event_loop = nullptr;	1,610✔
278	async_wait_data->handler = nullptr;	1,610✔
279	async_wait_data->process_ended = false;	1,610✔
280
281	auto status = GetExitStatus();	1,610✔
282
283	// Unlock in case the handler calls back into this object.
284	lock.unlock();	1,610✔
285	handler(ErrorBasedOnExitStatus(status));	1,610✔
286	}
287	}	1,610✔
288
289	static void CollectLineData(	56✔
290	string &trailing_line, vector<string> &lines, const char *bytes, size_t len) {
291	auto bytes_view = string_view(bytes, len);	56✔
292	size_t line_start_idx = 0;	56✔
293	size_t line_end_idx = bytes_view.find("\n", 0);	56✔
294	if ((trailing_line != "") && (line_end_idx != string_view::npos)) {	56✔
295	lines.push_back(trailing_line + string(bytes_view, 0, line_end_idx));	×
296	line_start_idx = line_end_idx + 1;	×
297	line_end_idx = bytes_view.find("\n", line_start_idx);	×
298	trailing_line = "";	×
299	}
300
301	while ((line_start_idx < (len - 1)) && (line_end_idx != string_view::npos)) {	182✔
302	lines.push_back(string(bytes_view, line_start_idx, (line_end_idx - line_start_idx)));	126✔
303	line_start_idx = line_end_idx + 1;	126✔
304	line_end_idx = bytes_view.find("\n", line_start_idx);	126✔
305	}
306
307	if ((line_end_idx == string_view::npos) && (line_start_idx != (len - 1))) {	56✔
308	trailing_line += string(bytes_view, line_start_idx, (len - line_start_idx));	56✔
309	}
310	}	56✔
311
312	ExpectedLineData Process::GenerateLineData(chrono::nanoseconds timeout) {	63✔
313	if (proc_) {	63✔
314	return expected::unexpected(	×
315	MakeError(ProcessAlreadyStartedError, "Cannot generate line data"));	×
316	}
317
318	if (args_.size() == 0) {	63✔
319	return expected::unexpected(MakeError(	×
320	ProcessesErrorCode::SpawnError, "No arguments given, cannot spawn a process"));	×
321	}
322
323	string trailing_line;	126✔
324	vector<string> ret;	126✔
325	proc_ = make_unique<tpl::Process>(	63✔
326	args_, work_dir_, [&trailing_line, &ret](const char *bytes, size_t len) {	119✔
327	CollectLineData(trailing_line, ret, bytes, len);	56✔
328	});	63✔
329
330	if (proc_->get_id() == -1) {	63✔
331	proc_.reset();	×
332	return expected::unexpected(MakeError(	×
333	ProcessesErrorCode::SpawnError,
334	"Failed to spawn '" + (args_.size() >= 1 ? args_[0] : "<null>") + "'"));	×
335	}
336
337	SetupAsyncWait();	63✔
338
339	auto err = Wait(timeout);	126✔
340	if (err != error::NoError && err.code != MakeError(NonZeroExitStatusError, "").code) {	68✔
341	return expected::unexpected(err);	×
342	}
343
344	if (trailing_line != "") {	63✔
345	ret.push_back(trailing_line);	2✔
346	}
347
348	return ExpectedLineData(ret);	63✔
349	}
350
351	io::ExpectedAsyncReaderPtr Process::GetProcessReader(events::EventLoop &loop, int &pipe_ref) {	4✔
352	if (proc_) {	4✔
353	return expected::unexpected(	×
354	MakeError(ProcessAlreadyStartedError, "Cannot get process output"));	×
355	}
356
357	if (pipe_ref >= 0) {	4✔
358	close(pipe_ref);	×
359	pipe_ref = -1;	×
360	}
361
362	int fds[2];
363	int ret = pipe(fds);	4✔
364	if (ret < 0) {	4✔
365	int err = errno;	×
366	return expected::unexpected(error::Error(	×
367	generic_category().default_error_condition(err),	×
368	"Could not get process stdout reader"));	×
369	}
370
371	pipe_ref = fds[1];	4✔
372
373	return make_shared<events::io::AsyncFileDescriptorReader>(loop, fds[0]);	8✔
374	}
375
376	io::ExpectedAsyncReaderPtr Process::GetAsyncStdoutReader(events::EventLoop &loop) {	2✔
377	return GetProcessReader(loop, stdout_pipe_);	2✔
378	}
379
380	io::ExpectedAsyncReaderPtr Process::GetAsyncStderrReader(events::EventLoop &loop) {	2✔
381	return GetProcessReader(loop, stderr_pipe_);	2✔
382	}
383
384	int Process::EnsureTerminated() {	2,023✔
385	if (!proc_) {	2,023✔
386	return exit_status_;	2,001✔
387	}
388
389	log::Info("Sending SIGTERM to PID " + to_string(proc_->get_id()));	22✔
390
391	Terminate();	22✔
392
393	auto result = future_exit_status_.wait_for(max_termination_time_);	22✔
394	if (result == future_status::timeout) {	22✔
395	log::Info("Sending SIGKILL to PID " + to_string(proc_->get_id()));	1✔
396	Kill();	1✔
397	result = future_exit_status_.wait_for(max_termination_time_);	1✔
398	if (result != future_status::ready) {	1✔
399	// This should not be possible, SIGKILL always terminates.
400	log::Error(	×
401	"PID " + to_string(proc_->get_id()) + " still not terminated after SIGKILL.");	×
402	return -1;	×
403	}
404	}
405	assert(result == future_status::ready);	22✔
406	exit_status_ = future_exit_status_.get();	22✔
407
408	log::Info(	22✔
409	"PID " + to_string(proc_->get_id()) + " exited with status " + to_string(exit_status_));	44✔
410
411	proc_.reset();	22✔
412
413	return exit_status_;	22✔
414	}
415
416	void Process::Terminate() {	23✔
417	if (proc_) {	23✔
418	// At the time of writing, tiny-process-library kills using SIGINT and SIGTERM, for
419	// `force = false/true`, respectively. But we want to kill with SIGTERM and SIGKILL,
420	// because:
421	//
422	// 1. SIGINT is not meant to kill interactive processes, whereas SIGTERM is.
423	// 2. SIGKILL is required in order to really force, since SIGTERM can be ignored by
424	// the process.
425	//
426	// If tiny-process-library is fixed, then this can be restored and the part below
427	// removed.
428	// proc_->kill(false);
429
430	::kill(proc_->get_id(), SIGTERM);	23✔
431	::kill(-proc_->get_id(), SIGTERM);	23✔
432	}
433	}	23✔
434
435	void Process::Kill() {	2✔
436	if (proc_) {	2✔
437	// See comment in Terminate().
438	// proc_->kill(true);
439
440	::kill(proc_->get_id(), SIGKILL);	2✔
441	::kill(-proc_->get_id(), SIGKILL);	2✔
442	}
443	}	2✔
444
445	void ProcessReaderFunctor::operator()(const char *bytes, size_t n) {	953✔
446	if (callback_) {	953✔
447	callback_(bytes, n);	911✔
448	}
449
450	if (fd_ < 0) {	953✔
451	return;	945✔
452	}
453
454	size_t written = 0;	8✔
455	while (written < n) {	14✔
456	auto ret = write(fd_, bytes, n);	8✔
457	if (ret < 0) {	8✔
458	int err = errno;	2✔
459	log::Error(	2✔
460	string {"Error while writing process output to main thread: "} + strerror(err));	4✔
461	fd_ = -1;	2✔
462	return;	2✔
463	}
464
465	written += ret;	6✔
466	}
467	}
468
469	} // namespace mender::common::processes

mendersoftware / mender / 1022753986

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous