Unexpected end of file while reading mpstat output with BOOST ASIO async_read_until

Question

I try to read the output of mpstat command (collecting cpu information every second, ie: "mptstat -P ALL 1") in order to get information about cpu and core usage. On a multicore cpu, I get an unexpected "end of file" status just after having read the first measurements.

It appears that mpstat formats its output in such a way that measurements for all cores are separated by an empty line.

I have used async_read_until with a delimiter equal to '\n'.

Please find below a small reproducer. With this reproducer, I get the following:

Enter handle_read
handle_read got data: --Linux 4.13.0-46-generic (pierre)    26/08/2018  _x86_64_    (4 CPU)
--Enter handle_read
handle_read got data: --
11:39:11     CPU    %usr   %nice    %sys %iowait    %irq   %soft  %steal  %guest  %gnice   %idle
11:39:11     all--Enter handle_read
handle_read got data: --    5,69    0,06    2,90    0,20    0,00    0,02    0,00    0,00    0,00   91,13
11:39:11       0    5,95    0,05--Enter handle_read
handle_read got data: --    2,80    0,13    0,00    0,01    0,00    0,00    0,00   91,06
11:39:11       1    5,24    0,01    2,50    0,14    0,00    0,02    0,00    0,00    0,00   92,09
11:39:11       2    6,30    0,17--Enter handle_read
handle_read got data: --    2,29    0,36    0,00    0,04    0,00    0,00    0,00   90,85
11:39:11       3    5,28    0,01    4,01    0,17    0,00    0,00    0,00    0,00    0,00   90,52
--Enter handle_read
Problem while trying to read mpstat data: End of file

Basically, I am able to receive the first measurement but I immediatly get an "end of file" just after. Looks like the empty line issued by mpstat is why I get the "end of line" indication... but I can't understand why...

Could somebody provide some help? Many thanks in advance.

#include <boost/asio.hpp>

#include <unistd.h>
#include <iostream>

#define PIPE_READ 0
#define PIPE_WRITE 1

#define ENDOFLINE "\n"

static boost::asio::streambuf data;
static std::shared_ptr<boost::asio::posix::stream_descriptor> cpuLoadDataStream;

static void handle_read(const boost::system::error_code& error, std::size_t bytes_transferred) {
  printf("Enter handle_read\n");
  if (error == boost::system::errc::success) {
    if (data.size() > 0) {
      std::string dataReceived((std::istreambuf_iterator<char>(&data)), std::istreambuf_iterator<char>());
      std::cout << "handle_read got data: " << "--" << dataReceived << "--";
    }
    boost::asio::async_read_until(*cpuLoadDataStream, data, ENDOFLINE, handle_read);
  } else {
      std::cout << "Problem while trying to read mpstat data: " << error.message() << std::endl;
  }
}

int main() {
  int pipeFd[2];
  boost::asio::io_service ioService;

  if (pipe(pipeFd) < 0) {
    std::cout << "pipe error" << std::endl;
    exit(EXIT_FAILURE);
  }

  int pid;

  if ((pid = fork()) == 0) {
    // son
    close(pipeFd[PIPE_READ]);
    if (dup2(pipeFd[PIPE_WRITE], STDOUT_FILENO) == -1) {
      std::cout << "dup2 error" << std::endl;
      exit(EXIT_FAILURE);
    }
    close(pipeFd[PIPE_WRITE]);

    if (execlp("mpstat", "1", "-P", "ALL", 0) == -1) {
      std::cout << "execlp error" << std::endl;
      exit(EXIT_FAILURE);
    }
  } else {
    // parent
    if (pid == -1) { 
      std::cout << "fork error" << std::endl;
      exit(EXIT_FAILURE);
    } else {
      close(pipeFd[PIPE_WRITE]);
      cpuLoadDataStream = std::make_shared<boost::asio::posix::stream_descriptor>(ioService, ::dup(pipeFd[PIPE_READ]));
      boost::asio::async_read_until(*cpuLoadDataStream, data, ENDOFLINE, handle_read);
    }
  }

  ioService.run();

  return 1;
}

sehe · Accepted Answer · 2018-08-26T13:34:37.663

First off: if you want "1" to mean that mpstat runs repeatedly at 1 second interval, you must make it the first argument, not the process name:

if (execlp("mpstat", "mpstat", "1", "-P", "ALL", 0) == -1) {

See the documentation:

The first argument, by convention, should point to the filename associated with the file being executed.

async_read_until reads until the input buffer at least contains the delimiter. So when you read the buffer, you must take into account the bytes_tranferred which will be excluding the delimiter.

Make sure to also consume the delimiter to avoid getting stuck in an infinite loop (because the stopping condition is already met):

void handle_read(const boost::system::error_code &error, std::size_t bytes_transferred) {
    std::cout << "Enter handle_read (" << error.message() << ")\n";

    if (!error) {
        if (bytes_transferred > 0) {
            std::copy_n(std::istreambuf_iterator<char>(&data), bytes_transferred, std::ostreambuf_iterator<char>(std::cout << "handle_read got data: '"));
            std::cout << "' --\n";
            data.consume(delimiter.size());
        }
        do_read_loop();
    }
}

Or simpler:

void handle_read(const boost::system::error_code &error, std::size_t bytes_transferred) {
    std::cout << "Enter handle_read (" << error.message() << ")\n";

    if (!error) {
        std::string line;
        if (getline(std::istream(&data), line)) 
            std::cout << "handle_read got data: '" << line << "'\n";

        do_read_loop();
    }
}

I'd prefer the first one because it is more explicit and generally applicable.

Side Notes:

there was a problem with the way you used global data (that would only get destroyed after the corresponding io_servce was already out of scope). That was UB - using asan/ubsan allows you to spot these bugs.

Using io_service across forks is not supported without support from the service implementations, see https://www.boost.org/doc/libs/1_68_0/doc/html/boost_asio/reference/io_context/notify_fork.html

Fixed/Simplified Asio

Here's with some simplifications, and cutting the globals:

Live On Coliru

#include <boost/asio.hpp>
#include <boost/bind.hpp>

#include <iostream>
#include <unistd.h>

namespace ba = boost::asio;

struct Reader : std::enable_shared_from_this<Reader> {
    ba::streambuf data;
    ba::posix::stream_descriptor cpuLoadDataStream;
    std::string const delimiter = "\n";

    Reader(ba::io_service& svc, int fd)
        : cpuLoadDataStream(svc, fd) {}

    void do_read_loop() {
        async_read_until(cpuLoadDataStream, data, delimiter, boost::bind(&Reader::handle_read, shared_from_this(), _1, _2));
    }

    void handle_read(const boost::system::error_code &error, std::size_t bytes_transferred) {
        std::cout << "Enter handle_read (" << error.message() << ")\n";

        if (!error) {
            if (bytes_transferred > 0) {
                std::copy_n(std::istreambuf_iterator<char>(&data), bytes_transferred, std::ostreambuf_iterator<char>(std::cout << "handle_read got data: '"));
                std::cout << "' --\n";
                data.consume(delimiter.size());
            }
            do_read_loop();
        }
    }
};

int main() {
    int fds[2];
    int& readFd = fds[0];
    int& writeFd = fds[1];

    if (pipe(fds) == -1) {
        std::cout << "pipe error" << std::endl;
        return 1;
    }

    if (int pid = fork()) {
        ba::io_service ioService;

        // parent
        if (pid == -1) {
            std::cerr << "fork error" << std::endl;
            return 2;
        } else {
            close(writeFd);
            std::make_shared<Reader>(ioService, ::dup(readFd))->do_read_loop();
        }

        ioService.run();
    } else {
        ba::io_service ioService;

        // child
        if (dup2(writeFd, STDOUT_FILENO) == -1) {
            std::cerr << "dup2 error" << std::endl;
            return 3;
        }
        close(readFd);
        close(writeFd);

        if (execlp("mpstat", "mpstat", "-P", "ALL", 0) == -1) {
            std::cerr << "execlp error" << std::endl;
            return 4;
        }

        ioService.run();
    }
}

Note it's not continuous for obvious reasons on Coliru

Much Simpler: Boost Process

Why not use Boost Process instead?

#include <boost/process.hpp>
#include <iostream>
#include <iomanip>

namespace bp = boost::process;

int main() {
    bp::pstream output;
    bp::system("mpstat -P ALL", bp::std_out > output);

    for (std::string line; std::getline(output, line);) {
        std::cout << "Got: " << std::quoted(line) << "\n";
    }
}

Or to make it async again:

Live On Coliru

#include <boost/asio.hpp>
#include <boost/process.hpp>
#include <iostream>
#include <iomanip>

namespace bp = boost::process;
using Args = std::vector<std::string>;

int main() {
    boost::asio::io_service io;
    bp::async_pipe output(io);
    bp::child mpstat(bp::search_path("mpstat"),
            //Args { "1", "-P", "ALL" },
            Args { "1", "3" }, // limited to 3 iterations on Coliru
            bp::std_out > output, io);

    boost::asio::streambuf data;
    std::function<void(boost::system::error_code, size_t)> handle = [&](boost::system::error_code ec, size_t /*bytes_transferred*/) {
        if (ec) {
            std::cout << "Good bye (" << ec.message() << ")\n";
        } else {
            std::string line;
            std::getline(std::istream(&data), line);
            std::cout << "Got: " << std::quoted(line) << "\n";

            async_read_until(output, data, "\n", handle);
        }
    };

    async_read_until(output, data, "\n", handle);

    io.run();
}

Prints

Got: "Linux 4.4.0-57-generic (stacked-crooked)  08/26/18    _x86_64_    (4 CPU)"
Got: ""
Got: "13:23:20     CPU    %usr   %nice    %sys %iowait    %irq   %soft  %steal  %guest  %gnice   %idle"
Got: "13:23:21     all    0.00    0.00    0.00    0.00    0.00    0.00    0.00    0.00    0.00  100.00"
Got: "13:23:22     all    0.00    0.00    0.00    0.25    0.00    0.00    0.00    0.00    0.00   99.75"
Got: "13:23:23     all    0.00    0.00    0.00    0.00    0.00    0.00    0.00    0.00    0.00  100.00"
Got: "Average:     all    0.00    0.00    0.00    0.08    0.00    0.00    0.00    0.00    0.00   99.92"
Good bye (End of file)

[Fixed](https://stackoverflow.com/posts/52026676/revisions) errors (copy/paste and confusing typo) — sehe, Aug 26 '18 at 13:39

Unexpected end of file while reading mpstat output with BOOST ASIO async_read_until

1 Answers1

Fixed/Simplified Asio

Much Simpler: Boost Process

Linked