From: Petr Tesarik <ptesarik@suse.cz>
Subject: Rewrite ProcessFilter using generic IO class
References: FATE#318874, bsc#917747
Patch-mainline: v0.8.12
Git-commit: 245b5cef8f0a7b0cfcb74287c2e48d9418e6ce91
Git-commit: b5f19d9b6c4c628304a11eaaf6ffb514a3ddcd63
Git-commit: f25413538774c7c2a794e6019edbc1b363c3edb4
Git-commit: fedbbcaf9b0794d6ed4c368014f0e070aec67e8e
Git-commit: 60a89662a60936a65e955a24ee03f24e918b4e5e
Git-commit: 7e25b99437b587fd9e406523cea79d76f175791e
Git-commit: 8abddc4693c8a955b425e3a650e134ad058f00ea

Backport generic ProcessFilter implementation from kdump-0.8.12.

Note: kdump-0.8.12 uses iostreams instead of byte buffers. I did not
backport the stream handling, but instead implemented a trivial buffer
classes. This code is not yet upstreams.

Signed-off-by: Petr Tesarik <ptesarik@suse.cz>

---
 kdumptool/process.cc |  397 +++++++++++++++++++++------------------------------
 kdumptool/process.h  |   34 ++--
 2 files changed, 186 insertions(+), 245 deletions(-)

--- a/kdumptool/process.cc
+++ b/kdumptool/process.cc
@@ -334,271 +334,214 @@ int MultiplexIO::monitor(int timeout)
 }
 
 //}}}
-//{{{ ProcessFilter ------------------------------------------------------------
+//{{{ Input --------------------------------------------------------------------
+class Input : public ProcessFilter::IO {
+    public:
+
+	Input(int fd)
+	throw ()
+	: ProcessFilter::IO(fd)
+	{ }
+
+	virtual void setupSubProcess(SubProcess &p)
+	throw ();
+};
 
 // -----------------------------------------------------------------------------
-ProcessFilter::ProcessFilter()
+void Input::setupSubProcess(SubProcess &p)
     throw ()
-    : m_stdinBuffer(NULL), m_stdoutBuffer(NULL), m_stderrBuffer(NULL)
-{}
+{
+    p.setPipeDirection(m_fd, SubProcess::ParentToChild);
+}
+
+//}}}
+//{{{ IBuffer ------------------------------------------------------------------
+class IBuffer : public Input {
+    public:
+	IBuffer(int fd, const ByteVector *buffer)
+	throw ()
+	: Input(fd), m_buffer(buffer), m_pos(0)
+        { }
+
+	virtual void setupIO(MultiplexIO &io, SubProcess &p);
+	virtual void handleEvents(MultiplexIO &io, SubProcess &p);
+
+    private:
+	const ByteVector *const m_buffer;
+	size_t m_pos;
+	int pollidx;
+};
 
 // -----------------------------------------------------------------------------
-void ProcessFilter::setStdinBuffer(ByteVector *buffer)
-    throw ()
+void IBuffer::setupIO(MultiplexIO &io, SubProcess &p)
 {
-    m_stdinBuffer = buffer;
+    pollidx = io.add(p.getPipeFD(m_fd), POLLOUT);
 }
 
 // -----------------------------------------------------------------------------
-void ProcessFilter::setStdoutBuffer(ByteVector *buffer)
-    throw ()
+void IBuffer::handleEvents(MultiplexIO &io, SubProcess &p)
 {
-    m_stdoutBuffer = buffer;
+    ssize_t cnt;
+    struct pollfd *poll = &io[pollidx];
+
+    if (poll->revents & POLLOUT) {
+	const unsigned char *bufptr = m_buffer->data() + m_pos;
+	size_t buflen = m_buffer->size() - m_pos;
+
+	cnt = write(poll->fd, bufptr, buflen);
+	if (cnt < 0)
+	    throw KSystemError("Cannot send data to input pipe", errno);
+	m_pos += cnt;
+
+	if (m_pos >= m_buffer->size()) {
+	    p.setPipeDirection(m_fd, SubProcess::None);
+	    io.deactivate(pollidx);
+	}
+    }
 }
 
+//}}}
+//{{{ Output -------------------------------------------------------------------
+
+class Output : public ProcessFilter::IO {
+    public:
+	Output(int fd)
+	throw ()
+	: ProcessFilter::IO(fd)
+	{ }
+
+	virtual void setupSubProcess(SubProcess &p)
+	throw ();
+};
+
 // -----------------------------------------------------------------------------
-void ProcessFilter::setStderrBuffer(ByteVector *buffer)
+void Output::setupSubProcess(SubProcess &p)
     throw ()
 {
-    m_stderrBuffer = buffer;
+    p.setPipeDirection(m_fd, SubProcess::ChildToParent);
 }
 
+//}}}
+//{{{ OBuffer ------------------------------------------------------------------
+class OBuffer : public Output {
+    public:
+	OBuffer(int fd, ByteVector *buffer)
+	throw ()
+	: Output(fd), m_buffer(buffer), m_pos(0)
+        { }
+
+	virtual void setupIO(MultiplexIO &io, SubProcess &p);
+	virtual void handleEvents(MultiplexIO &io, SubProcess &p);
+
+    private:
+	ByteVector *m_buffer;
+	size_t m_pos;
+	int pollidx;
+};
+
 // -----------------------------------------------------------------------------
-uint8_t ProcessFilter::execute(const string &name, const StringVector &args)
-    throw (KError)
+void OBuffer::setupIO(MultiplexIO &io, SubProcess &p)
 {
-    pid_t child;
-    int status = 0;
+    pollidx = io.add(p.getPipeFD(m_fd), POLLIN);
+}
 
-    Debug::debug()->trace("ProcessFilter::execute(%s, %s)",
-        name.c_str(), Stringutil::vector2string(args, ":").c_str());
+// -----------------------------------------------------------------------------
+void OBuffer::handleEvents(MultiplexIO &io, SubProcess &p)
+{
+    ssize_t cnt;
+    struct pollfd *poll = &io[pollidx];
 
-    //
-    // setup pipes for stdin, stderr, stdout if we use buffers
-    //
-    int stdin_pipes[2] = { 0, 0 };
-    int stdout_pipes[2] = { 0, 0 };
-    int stderr_pipes[2] = { 0, 0 };
-    if (m_stdinBuffer) {
-        int ret = pipe(stdin_pipes);
-        if (ret < 0)
-            throw KSystemError("Could not create stdin pipes.", errno);
-    }
-    if (m_stdoutBuffer) {
-        int ret = pipe(stdout_pipes);
-        if (ret < 0)
-            throw KSystemError("Could not create stdout pipes.", errno);
-    }
-    if (m_stderrBuffer) {
-        int ret = pipe(stderr_pipes);
-        if (ret < 0)
-            throw KSystemError("Could not create stderr pipes.", errno);
+    if (poll->revents & (POLLIN | POLLHUP)) {
+	if (poll->revents & POLLIN) {
+	    m_buffer->resize(m_pos + BUFSIZ);
+	    unsigned char *bufptr = m_buffer->data() + m_pos;
+	    size_t buflen = m_buffer->size() - m_pos;
+	    cnt = read(poll->fd, bufptr, buflen);
+	} else
+	    cnt = 0;
+
+	if (cnt < 0)
+	    throw KSystemError("Cannot get data from output pipe", errno);
+	m_pos += cnt;
+	m_buffer->resize(m_pos);
+
+	if (!cnt) {
+	    p.setPipeDirection(m_fd, SubProcess::None);
+	    io.deactivate(pollidx);
+	}
     }
+}
+
+//}}}
+//{{{ ProcessFilter ------------------------------------------------------------
+
+// -----------------------------------------------------------------------------
+ProcessFilter::~ProcessFilter()
+    throw ()
+{
+    std::map<int, IO*>::const_iterator it;
+    for (it = m_iomap.begin(); it != m_iomap.end(); ++it)
+	delete it->second;
+}
 
-    //
-    // execute the child
-    //
-
-    child = fork();
-    if (child > 0) { // parent code, just wait until the child has exited
-        int ret;
-
-        // close the ends of the pipes that are not necessary
-        if (stdin_pipes[0]) // read end of stdin
-            close(stdin_pipes[0]);
-        if (stdout_pipes[1]) // write end of stdout
-            close(stdout_pipes[1]);
-        if (stderr_pipes[1]) // write end of stderr
-            close(stderr_pipes[1]);
-
-        if (stdin_pipes[1] || stdout_pipes[0] || stderr_pipes[0]) {
-
-            loff_t write_pos = 0;
-
-            while (true) {
-                int maxfd = -1;
-                fd_set read_fds;
-                fd_set write_fds;
-                struct timeval tv = { 1, 0 };
-                int retval;
-
-                FD_ZERO(&read_fds);
-                FD_ZERO(&write_fds);
-
-                if (stdin_pipes[1]) {
-                    // check if fd is still valid
-                    if (dup2(stdin_pipes[1], stdin_pipes[1]) < 0) {
-                        close(stdin_pipes[1]);
-                        stdin_pipes[1] = 0;
-                    } else {
-                        FD_SET(stdin_pipes[1], &write_fds);
-                        maxfd = max(maxfd, stdin_pipes[1]);
-                    }
-                }
-                if (stdout_pipes[0]) {
-                    // check if fd is still valid
-                    if (dup2(stdout_pipes[0], stdout_pipes[0]) < 0) {
-                        close(stdout_pipes[0]);
-                        stdout_pipes[0] = 0;
-                    } else {
-                        FD_SET(stdout_pipes[0], &read_fds);
-                        maxfd = max(maxfd, stdout_pipes[0]);
-                    }
-                }
-                if (stderr_pipes[0]) {
-                    // check if fd is still valid
-                    if (dup2(stderr_pipes[0], stderr_pipes[0]) < 0) {
-                        close(stderr_pipes[0]);
-                        stderr_pipes[0] = 0;
-                    } else {
-                        FD_SET(stderr_pipes[0], &read_fds);
-                        maxfd = max(maxfd, stderr_pipes[0]);
-                    }
-                }
-
-                // one more
-                maxfd++;
-
-                // if we don't have valid FDs, process the read()/write() once
-                if (maxfd == 0)
-                    retval = 0;
-                else
-                    retval = select(maxfd, &read_fds, &write_fds, NULL, &tv);
-
-                if (retval < 0 && errno != EBADF) {
-                    closeFds();
-                    throw KSystemError("select() failed", errno);
-                } else if (retval > 0) {
-
-                    char buffer[BUFSIZ];
-
-                    if (FD_ISSET(stdin_pipes[1], &write_fds)) {
-
-                        loff_t end_pos = min((loff_t)m_stdinBuffer->size(),
-                                             write_pos+BUFSIZ);
-
-                        copy(m_stdinBuffer->begin() + write_pos,
-                             m_stdinBuffer->begin() + end_pos,
-                             buffer);
-
-                        retval = write(stdin_pipes[1], buffer, end_pos - write_pos);
-                        if (retval < 0) {
-                            closeFds();
-                            throw KSystemError("write() failed", errno);
-                        }
-
-                        write_pos += retval;
-
-                        if (write_pos >= (loff_t)m_stdinBuffer->size())
-                            close(stdin_pipes[1]);
-
-                    }
-                    if (FD_ISSET(stdout_pipes[0], &read_fds)) {
-
-                        retval = read(stdout_pipes[0], buffer, BUFSIZ);
-                        if (retval < 0) {
-                            closeFds();
-                            throw KSystemError("read() failed", errno);
-                        }
-
-                        if (retval == 0) {
-                            close(stdout_pipes[0]);
-                            stdout_pipes[0] = 0;
-                        } else {
-                            m_stdoutBuffer->insert(m_stdoutBuffer->end(),
-                                                   buffer, buffer+retval);
-                        }
-                    }
-                    if (FD_ISSET(stderr_pipes[0], &read_fds)) {
-
-                        retval = read(stderr_pipes[0], buffer, BUFSIZ);
-                        if (retval < 0) {
-                            closeFds();
-                            throw KSystemError("read() failed", errno);
-                        }
-
-                        if (retval == 0) {
-                            close(stderr_pipes[0]);
-                            stderr_pipes[0] = 0;
-                        } else {
-                            m_stderrBuffer->insert(m_stderrBuffer->end(),
-                                                   buffer, buffer+retval);
-                        }
-                    }
-                }
-
-                if (maxfd == 0) {
-                    // check if the process is still running
-                    ret = waitpid(child, &status, 0);
-                    if (ret != child) {
-                        throw KSystemError("ProcessFilter::execute() waits for "
-                        + Stringutil::number2string(child) + " but "
-                        + Stringutil::number2string(ret) + " exited.", errno);
-                    }
-                    break;
-                }
-            }
-
-        } else {
-            ret = waitpid(child, &status, 0);
-            if (ret != child) {
-                throw KSystemError("ProcessFilter::execute() waits for "
-                    + Stringutil::number2string(child) + " but "
-                    + Stringutil::number2string(ret) + " exited.", errno);
-            }
-        }
-    } else if (child == 0) { // child code, execute the process
-
-        // stdin
-        if (stdin_pipes[1]) {
-            close(stdin_pipes[1]); // write end of stdin
-            dup2(stdin_pipes[0], STDIN_FILENO); // read end of stdin
-        }
-
-        // stdout
-        if (stdout_pipes[0]) {
-            close(stdout_pipes[0]); // read end of stdout
-            dup2(stdout_pipes[1], STDOUT_FILENO); // write end of stdout
-        }
-
-        // stderr
-        if (stderr_pipes[0]) {
-            close(stderr_pipes[0]); // read end of stderr
-            dup2(stderr_pipes[1], STDERR_FILENO); // write end of stderr
-        }
-
-        executeProcess(name, args);
-
-    } else { // error in fork()
-        closeFds();
-        throw KSystemError("fork() failed in ProcessFilter::execute", errno);
+void ProcessFilter::setIO(IO *io)
+{
+    std::pair<std::map<int, IO*>::iterator, bool> ret;
+    ret = m_iomap.insert(std::make_pair(io->getFD(), io));
+    if (ret.second == false) {
+	delete ret.first->second;
+	ret.first->second = io;
     }
+}
 
-    closeFds();
+// -----------------------------------------------------------------------------
+void ProcessFilter::setStdinBuffer(ByteVector *buffer)
+    throw ()
+{
+    setIO(new IBuffer(STDIN_FILENO, buffer));
+}
 
-    return WEXITSTATUS(status);
+// -----------------------------------------------------------------------------
+void ProcessFilter::setStdoutBuffer(ByteVector *buffer)
+    throw ()
+{
+    setIO(new OBuffer(STDOUT_FILENO, buffer));
 }
 
 // -----------------------------------------------------------------------------
-void ProcessFilter::closeFds()
+void ProcessFilter::setStderrBuffer(ByteVector *buffer)
     throw ()
 {
-    Debug::debug()->trace("ProcessFilter::closeFds()");
+    setIO(new OBuffer(STDERR_FILENO, buffer));
 }
 
 // -----------------------------------------------------------------------------
-void ProcessFilter::executeProcess(const string &name, const StringVector &args)
+uint8_t ProcessFilter::execute(const string &name, const StringVector &args)
     throw (KError)
 {
-    StringVector fullV = args;
-    fullV.insert(fullV.begin(), name);
-    char **vector = Stringutil::stringv2charv(fullV);
+    Debug::debug()->trace("ProcessFilter::execute(%s, %s)",
+        name.c_str(), Stringutil::vector2string(args, ":").c_str());
+
+    std::map<int, IO*>::const_iterator it;
 
-    int ret = execvp(name.c_str(), vector);
-    Util::freev(vector);
+    SubProcess p;
+    for (it = m_iomap.begin(); it != m_iomap.end(); ++it)
+	it->second->setupSubProcess(p);
+    p.spawn(name, args);
+
+    // initialize multiplex IO
+    MultiplexIO io;
+    for (it = m_iomap.begin(); it != m_iomap.end(); ++it)
+	it->second->setupIO(io, p);
+
+    while (io.active() > 0) {
+	io.monitor();
+	for (it = m_iomap.begin(); it != m_iomap.end(); ++it)
+	    it->second->handleEvents(io, p);
+    }
 
-    if (ret < 0)
-        throw KSystemError("Execution of '" + name + "' failed.", errno);
+    int status = p.wait();
+    return WEXITSTATUS(status);
 }
 
 //}}}
--- a/kdumptool/process.h
+++ b/kdumptool/process.h
@@ -290,15 +290,10 @@ class ProcessFilter {
 	class IO;
 
         /**
-         * Creates a new process filter.
+         * Destructor.
          */
-        ProcessFilter()
-        throw ();
-
-        /**
-         * Desctructor.
-         */
-        virtual ~ProcessFilter () {}
+        ~ProcessFilter ()
+	throw ();
 
         /**
          * Executes a process.
@@ -314,6 +309,17 @@ class ProcessFilter {
         uint8_t execute(const std::string &name, const StringVector &args)
         throw (KError);
 
+	/**
+	 * Set Input/Output handling.
+	 *
+	 * @param[in] io pointer to a class ProcessFilter::IO instance;
+	 *               the instance is deallocated by class ProcessFilter
+	 *
+	 * Note: The implementation uses io->getFD() to know which file
+	 * descriptor is modified.
+	 */
+	void setIO(IO *io);
+
         /**
          * Sets the buffer where the bytes from stdin are read from to
          * @p buffer. This function overwrites the ProcessFilter::setStdinFd()
@@ -344,17 +350,9 @@ class ProcessFilter {
         void setStderrBuffer(ByteVector *buffer)
         throw ();
 
-    protected:
-        void executeProcess(const std::string &name, const StringVector &args)
-        throw (KError);
-
-        void closeFds()
-        throw ();
-
     private:
-        ByteVector *m_stdinBuffer;
-        ByteVector *m_stdoutBuffer;
-        ByteVector *m_stderrBuffer;
+
+	std::map<int, IO*> m_iomap;
 };
 
 //}}}
