Hier ist mein fork() Sample. Es forkt und überwacht einen Subprozess. Ich habe auch einen tty_dummy für den string gebastelt der dein Protokoll nachahmt bzw. parst. Er sollte so auch wieder auf syncronisieren können (tty_dummy). Der Trick hier ist "find('\n')" und wenn der regex nicht matcht, das wegzuwerfen.
Die Funktion die im Subprozess aufgerufen wird, öffnet dann den tty und spricht mit dem Master nur über die pipes. Die tty_dummy() ist nur ein ungetestetes Skelett. Auf Arduino Seite würde ich das auch so ähnlich machen.
Das ganze wurde auf meinem Debianrechner getestet.
Das Testprogram zeigt:
./restarttest
Started worker: 5433
Slave got command: 14
Got data from worker 4
Slave got command: 11
Got data from worker 3
Slave got command: 11
Got data from worker 3
Slave got command: 11
Got data from worker 3
Slave got command: 11
Got data from worker 2
Slave got command: 13
worker abort
Terminate worker: 5433
Started worker: 5434
Slave got command: 12
Got data from worker 1
Slave got command: 13
Got data from worker 1
Slave got command: 16
Got data from worker 3
Slave got command: 16
Got data from worker 3
Slave got command: 16
...
/************************************************** ****************************
DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
Version 2, December 2004
Copyright (C) 2019 Georg Gast <georg@schorsch-tech.de>
Everyone is permitted to copy and distribute verbatim or modified
copies of this license document, and changing it is allowed as long
as the name is changed.
DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
0. You just DO WHAT THE FUCK YOU WANT TO.
************************************************** ****************************/
/************************************************** ****************************
Sample for Roboternetz.de:
pthread: was genau macht "joinable" und was macht "detached"?
https://www.roboternetz.de/community/threads/73509-pthread-was-genau-macht-joinable-und-was-macht-detached
Compile with:
-------------
g++ --std=c++11 restart.cpp -o restart
Explanation:
------------
This program creates a master process and forks a child process. These
two processes are connected by pipes. Just use plain POSIX.
+--------------------+
| Master |
+--------------------+
| ^
v |
+--------------------+
| worker |
+--------------------+
They exchange random integer bytes.
The master sends to the child random commands. If it does not get a response
within 100 millisec it sends SIGKILL to the worker that it can not prevent
from termination. After the worker is reaped, close the old pipes, create
new pipes and forks a new worker.
The worker starts to sleep, if its rng generates a 6, if it gets a 7
it calls std::abort(). In any other case it sends it to the master.
but just if SIMULATE_FAIL is 1.
The child rng uses this distribution scheme:
std::normal_distribution<> dis(3,2);
-2
-1 *
0 ***
1 *****
2 ********
3 **********
4 ********
5 ******
6 ***
7 *
8
Comments:
---------
If we need to send more data than PIPE_BUF (linux 64 bit 64kb, 32bit 4kb)
through the pipes, we need to use select(),poll() or epoll() and create a
multiplexed application then we could make sure the while loop in the
read_loop/write_loop can be avoided. It would also make an application
more expandable.
see: man 7 pipe
boost::asio would also be a nice solution to this problem and make it
platform independent.
The regex in the tty_dummy could be optimized to detect automaticly how
many ix=xxx are received.
************************************************** ****************************/
// if SIMULATE_FAIL is 1, the child simulates random failing
// if SIMULATE_FAIL is 0, the child should run forever
#define SIMULATE_FAIL 1
// std c++
#include <cassert>
#include <chrono> // clock
#include <cmath> // std::round
#include <functional>
#include <iostream>
#include <memory> // shared_ptr for FDHandle
#include <random>
#include <regex>
#include <string>
#include <thread> // sleep_for
#include <tuple>
// abort and runtime_error
#include <exception>
#include <stdexcept>
// linux + posix
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <fcntl.h>
#include <limits.h> // PIPE_BUF
#include <signal.h>
#include <unistd.h>
// make sure we always close the fd (we MUST ensure
// no fd leak as the process is very long running)
using FDHandle = std::shared_ptr<void>;
FDHandle MakeFDHandle(int fd)
{
return FDHandle(reinterpret_cast<void *>(fd), [fd](void *) {
if (fd >= 0)
{
close(fd);
}
});
}
inline int GetFD(FDHandle handle)
{
size_t p = reinterpret_cast<size_t>(handle.get());
return static_cast<int>(p);
}
// chrono helper
using clk = std::chrono::system_clock;
using timepoint_t = clk::time_point;
inline timepoint_t now() { return clk::now(); }
// blocking write until we wrote all data
size_t write_loop(FDHandle fd, const void *buffer, size_t len)
{
assert(fd.get() != nullptr && GetFD(fd) >= 0);
assert(buffer != nullptr);
assert(len > 0);
assert(len < PIPE_BUF &&
"Because in case we got a signal and interrupt the write, we must "
"be able to get the data into the pipe");
ssize_t wrote_bytes = 0;
size_t total_wrote = 0;
while (total_wrote < len)
{
int f = GetFD(fd);
while ((wrote_bytes =
write(GetFD(fd),
reinterpret_cast<const char *>(buffer) + total_wrote,
len - total_wrote)) < 0 &&
errno == EINTR)
;
// EOF
if (wrote_bytes == 0)
{
throw std::runtime_error("write_loop EOF");
}
else if (wrote_bytes < 0)
{
perror("write()");
throw std::runtime_error("write() failed");
}
total_wrote += wrote_bytes;
}
assert(total_wrote == len);
return total_wrote;
}
// blocking read until we get some data
int read_loop(FDHandle fd, void *buffer, size_t len)
{
assert(fd.get() != nullptr && GetFD(fd) >= 0);
assert(buffer != nullptr);
assert(len > 0);
int read_bytes;
while ((read_bytes = read(GetFD(fd), buffer, len)) < 0 && errno == EINTR)
;
// EOF
if (read_bytes == 0)
{
throw std::runtime_error("read_loop EOF");
}
else if (read_bytes < 0)
{
perror("read()");
throw std::runtime_error("read() failed");
}
assert(read_bytes > 0);
return read_bytes;
}
// peek into the pipe and get the available bytes
int get_avail_bytes(FDHandle fd)
{
assert(fd.get() != nullptr && GetFD(fd) >= 0);
int avail_bytes = 0;
if (ioctl(GetFD(fd), FIONREAD, &avail_bytes) != 0)
{
perror("ioctl()");
throw std::runtime_error("ioctl() failed");
}
return avail_bytes;
}
// start the worker and give back the FDHandles to and from the worker
std::tuple<pid_t, FDHandle /* rfd */, FDHandle /* wfd*/>
start_worker(std::function<int(FDHandle rfd, FDHandle wfd)> fo)
{
assert(static_cast<bool>(fo) == true && "Function object must be valid");
int master_to_worker[2];
int worker_to_master[2];
if (pipe(master_to_worker) != 0)
{
perror("pipe() master_to_worker");
exit(EXIT_FAILURE);
}
if (pipe(worker_to_master) != 0)
{
perror("pipe() worker_to_master");
exit(EXIT_FAILURE);
}
pid_t pid = fork();
switch (pid)
{
// fork failed:
case -1:
perror("fork()");
exit(EXIT_FAILURE);
// child
case 0:
{
// as the child we should close the not needed fds
close(master_to_worker[1]);
close(worker_to_master[0]);
exit(fo(MakeFDHandle(master_to_worker[0]),
MakeFDHandle(worker_to_master[1])));
}
// master
default:
{
// as a master we should close the not needed fds
close(master_to_worker[0]);
close(worker_to_master[1]);
return std::make_tuple(pid, MakeFDHandle(worker_to_master[0]),
MakeFDHandle(master_to_worker[1]));
}
}
// just to silence compiler warning: we never reach this point
return std::make_tuple(pid, MakeFDHandle(-1), MakeFDHandle(-1));
}
void kill_worker(pid_t &worker)
{
if (worker <= 0)
{
return;
}
// SIGKILL _CAN NOT_ be prohibited by the worker. It kills in any case
// the worker (if we have the right to send it this signal)
std::cout << "Terminate worker: " << worker << std::endl;
if (kill(worker, SIGKILL) != 0)
{
perror("kill()");
exit(EXIT_FAILURE);
}
// Just wait for this worker and reap the child process (prevent zombie
// process)
if (waitpid(worker, NULL, 0) != worker)
{
perror("waitpid()");
exit(EXIT_FAILURE);
}
worker = 0;
}
// signal handler for master
// here we can just use very limited number of functions
// see man signal
int selfpipe[2];
void onsignal(int signal)
{
switch (signal)
{
case SIGTERM:
case SIGQUIT:
write(selfpipe[1], " ", 1);
break;
case SIGCHLD:
break;
default:
break;
}
}
// master function. It should not fail or hang.
int master(std::function<int(FDHandle rfd, FDHandle wfd)> worker_fo)
{
assert(static_cast<bool>(worker_fo) == true &&
"Function object must be valid");
// install sighandler for SIGTERM, SIGQUIT, SIGCHLD
// with self pipe trick.
if (pipe2(selfpipe, O_NONBLOCK) != 0)
{
perror("pipe() selfpipe");
exit(EXIT_FAILURE);
}
auto rd_selfpipe = MakeFDHandle(selfpipe[0]);
auto wr_selfpipe = MakeFDHandle(selfpipe[1]);
signal(SIGCHLD, &onsignal);
signal(SIGTERM, &onsignal);
signal(SIGQUIT, &onsignal);
// make sure master and worker dont share the same rng sequence
std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<> dis(11, 16);
auto last_bytes_read_at = now();
pid_t worker = 0;
FDHandle rfd;
FDHandle wfd;
bool run = true;
while (run)
{
// start the worker
if (worker == 0)
{
std::tie(worker, rfd, wfd) = start_worker(worker_fo);
last_bytes_read_at = now();
std::cout << "Started worker: " << worker << std::endl;
}
// generate a pseudo command
int cmd = dis(gen);
int wrote_bytes = write_loop(wfd, &cmd, sizeof(int));
assert(wrote_bytes == sizeof(int));
// check if the worker has wrote some data
int avail_bytes = 0;
while (avail_bytes == 0 &&
now() - last_bytes_read_at < std::chrono::milliseconds(100))
{
avail_bytes = get_avail_bytes(rfd);
if (avail_bytes == 0)
{
std::this_thread::sleep_for(std::chrono::milliseco nds(10));
}
}
// check if worker failed
if (avail_bytes == 0)
{
kill_worker(worker);
continue;
}
// got data from the worker
int data;
int read_bytes = read_loop(rfd, &data, sizeof(int));
assert(read_bytes > 0);
std::cout << "Got data from worker " << data << std::endl;
last_bytes_read_at = now();
// check self pipe and reset run flag
if (get_avail_bytes(rd_selfpipe) > 0)
{
run = false;
}
}
// kill the worker, the pipes get closed by the handles
kill_worker(worker);
return EXIT_SUCCESS;
}
int failing_worker(FDHandle rfd, FDHandle wfd)
{
// worker dont get a signal handler as we want to manipulate it by the user
assert(rfd.get() != nullptr && GetFD(rfd) >= 0);
assert(wfd.get() != nullptr && GetFD(wfd) >= 0);
// make sure master and worker dont share the same sequence
std::random_device rd;
std::mt19937 gen(rd());
std::normal_distribution<> dis(3, 2);
/*
-2
-1 *
0 ***
1 *****
2 ********
3 **********
4 ********
5 ******
6 ***
7 *
8
*/
while (1)
{
// read our command from the master
int cmd;
int read_bytes = read_loop(rfd, &cmd, sizeof(cmd));
assert(read_bytes > 0);
std::cout << "Slave got command: " << cmd << std::endl;
// send a response to the master (or sleep or abort)
int rnd = std::round(dis(gen));
#if SIMULATE_FAIL
if (rnd == 6)
{
// sleep forever: simulate hang
std::cout << "worker sleeping" << std::endl;
std::this_thread::sleep_until(timepoint_t::max());
}
else if (rnd == 7)
{
// just die: abnormal program termination
std::cout << "worker abort" << std::endl;
std::abort();
}
#endif
// simulate a read from uart and give it to master
int wrote = write_loop(wfd, &rnd, sizeof(rnd));
assert(wrote > 0);
}
return EXIT_FAILURE;
}
// an example of the tty receiver on the pi
int tty_dummy(FDHandle rfd, FDHandle wfd)
{
assert(rfd.get() != nullptr && GetFD(rfd) >= 0);
assert(wfd.get() != nullptr && GetFD(wfd) >= 0);
std::string rcv_buffer;
while (1)
{
// read data from the master
std::array<char, 256> tmp;
const int read_bytes = read_loop(rfd, &tmp[0], tmp.size());
for (int i = 0; i < read_bytes; ++i)
{
rcv_buffer += tmp[i];
}
// is there a /n
auto pos = rcv_buffer.find('\n');
if (pos != std::string::npos)
{
std::string line(rcv_buffer.begin(), rcv_buffer.begin() + pos);
// remove the line from the rcv buffer
rcv_buffer.erase(rcv_buffer.begin(), rcv_buffer.begin() + pos + 1);
// math this:
// §&i0=17628;&i1=1;&i2=17434;&i3=33;&i4=-444;&i5=5555;&i6=43690;
std::regex ex(
"^§&i0=([\\-0-9]++);&i1=([\\-0-9]+);&i2=([\\-0-9]+);&i3=([\\-0-"
"9]+);&i4=([\\-0-9]+);&i5=([\\-0-9]+);&i6=([\\-0-9]+);$");
std::smatch match;
if (!std::regex_match(line, match, ex))
{
std::cerr << "Unmatched line: " << line << std::endl;
continue;
}
std::cout << "Matched: ";
for (int i = 0; i < 7; ++i)
{
std::cout << "i" << i << "=" << match[i].str() << std::endl;
}
}
}
}
int main(int argc, char *argv[])
{
try
{
// start the master function
return master(&failing_worker);
}
catch (const std::exception &ex)
{
std::cerr << ex.what() << std::endl;
}
catch (...)
{
std::cerr << "Unknown exception" << std::endl;
}
return EXIT_FAILURE;
}
- - - Aktualisiert - - -
@HaWe:
man 3 pthread_kill
NOTES
Signal dispositions are process-wide: if a signal handler is installed, the handler will be invoked in the thread thread, but if the disposition of
the signal is "stop", "continue", or "terminate", this action will affect the whole process.
Das bedeutet, das du nicht den thread töten kannst. Du rufst nur "kill" auf im Kontext des thread thread.
- - - Aktualisiert - - -
Das hier zeigt es ganz gut:
http://openbook.rheinwerk-verlag.de/linux_unix_programmierung/Kap10-011.htm