core: demangle stdout

When using print() to debug on smp, it is very annoying to get interleaved
output.

Fix by wrapping stdout with a fake stream that has a line buffer for each
thread.

configure.py

@@ -141,6 +141,7 @@ def debug_flag(compiler):
     'core/posix.cc',
     'core/memory.cc',
     'core/resource.cc',
+    'core/stdio.cc',
     'core/scollectd.cc',
     'core/app-template.cc',
     'core/dpdk_rte.cc',

core/app-template.cc

@@ -6,6 +6,7 @@
 #include "core/reactor.hh"
 #include "core/scollectd.hh"
 #include "core/print.hh"
+#include "stdio.hh"
 #include <boost/program_options.hpp>
 #include <boost/make_shared.hpp>
 #include <fstream>
@@ -70,6 +71,8 @@ app_template::run(int ac, char ** av, std::function<void ()>&& func) {
         std::cout << _opts << "\n";
         return 1;
     }
+    // intentional leak
+    stdout = smp_synchronize_lines(stdout);
     smp::configure(configuration);
     _configuration = {std::move(configuration)};
     engine().when_started().then([this] {

core/stdio.cc

@@ -0,0 +1,76 @@
+/*
+ * Copyright (C) 2015 Cloudius Systems, Ltd.
+ */
+
+
+#include "stdio.hh"
+#include <memory>
+#include <pthread.h>
+#include <mutex>
+#include <unordered_map>
+#include <boost/thread/tss.hpp>
+#include <vector>
+#include <algorithm>
+
+class spinlock {
+    pthread_spinlock_t _l;
+public:
+    spinlock() { pthread_spin_init(&_l, PTHREAD_PROCESS_PRIVATE); }
+    ~spinlock() { pthread_spin_destroy(&_l); }
+    void lock() { pthread_spin_lock(&_l); }
+    void unlock() { pthread_spin_unlock(&_l); }
+};
+
+class smp_file {
+    FILE* _out;
+    spinlock _lock;
+    boost::thread_specific_ptr<std::vector<char>> _buffer;
+    static constexpr size_t max = 2000;
+public:
+    smp_file(FILE* out) : _out(out) {}
+    ssize_t write(const char* buffer, size_t size) {
+        auto& b = *_buffer;
+        b.insert(b.end(), buffer, buffer + size);
+        size_t now = 0;
+        if (b.size() >= max) {
+            now = b.size();
+        } else {
+            auto lf = std::find(b.rbegin(), b.rend(), '\n');
+            if (lf != b.rend()) {
+                auto remain = lf - b.rbegin();
+                now = b.size() - remain;
+            }
+        }
+        if (now) {
+            auto ret = fwrite(b.data(), 1, now, _out);
+            b.erase(b.begin(), b.begin() + now);
+            return ret;
+        } else {
+            return 0;
+        }
+    }
+};
+
+static smp_file* to_smp_file(void* cookie) {
+    return static_cast<smp_file*>(cookie);
+}
+
+FILE*
+smp_synchronize_lines(FILE* out) {
+    auto p = std::make_unique<smp_file>(out);
+    cookie_io_functions_t vtable = {};
+    vtable.write = [] (void* ck, const char* b, size_t s) { return to_smp_file(ck)->write(b, s); };
+    vtable.close = [] (void* ck) { delete to_smp_file(ck); return 0; };
+    auto ret = fopencookie(p.get(), "w", vtable);
+    if (!ret) {
+        return ret;
+    }
+    // disable buffering so that writes to ret don't get mixed
+    // up but are sent to smp_file immediately instead.
+    setvbuf(ret, nullptr, _IONBF, 0);
+    p.release();
+    return ret;
+}
+
+
+

core/stdio.hh

@@ -0,0 +1,12 @@
+/*
+ * Copyright (C) 2015 Cloudius Systems, Ltd.
+ */
+
+#pragma once
+
+#include <stdio.h>
+
+// Returns a FILE* that writes all its output to @out, but attempts
+// not to mix lines if multiple threads write to it simultaenously.
+FILE* smp_synchronize_lines(FILE* out);
+