Multi threaded console IO

Question

For many console applications it is handy to be able to receive input and output at the same time from multiple threads. This is supposed to allow receiving input commands from multiple threads, while simultaneously outputting.

Console.h

#pragma once

#include <iostream>
#include <string>
#include <queue>
#include <mutex>
#include <atomic>
#include <conio.h>
#include <boost/thread.hpp>

namespace kaarez {
    namespace util {
        class Console {
        public:
            static void initialize();
            static void terminate();
            static void write(std::string text);
            static bool read(std::string& text);

        private:
            static void input_loop();
            static void write_loop();
            static std::string              input_buffer; //Current input buffer
            static std::string              input_field; //Printed at the bottom of the console
            static std::queue<std::string>  writeQueue;
            static std::queue<std::string>  readQueue;
            static std::condition_variable  condition_variable;
            static std::mutex               write_mutex;
            static std::mutex               read_mutex;
            static boost::thread*           write_thread;
            static boost::thread*           input_thread;
            static std::atomic<bool>        stop_loops;
            static std::atomic<bool>        initialized;

        };

        boost::thread*          Console::input_thread;
        boost::thread*          Console::write_thread;
        std::string             Console::input_buffer;
        std::string             Console::input_field;
        std::queue<std::string> Console::writeQueue;
        std::condition_variable Console::condition_variable;
        std::mutex              Console::write_mutex;
        std::mutex              Console::read_mutex;
        std::queue<std::string> Console::readQueue;
        std::atomic<bool>       Console::stop_loops = false;
        std::atomic<bool>       Console::initialized = false;

        void Console::input_loop() {
            while (!stop_loops) {
                uint32_t c = getch();
                if (c == 8) { //Backspace
                    if (!input_buffer.empty()) {
                        input_buffer.pop_back();
                        input_field = input_buffer;
                        std::unique_lock<std::mutex> l(write_mutex);
                        condition_variable.notify_one();
                    }
                }
                else if (c == 13) { //Carriage return
                    if (!input_buffer.empty()) {
                        std::unique_lock<std::mutex> l(read_mutex);
                        readQueue.push(input_buffer);
                        write(input_buffer);
                        input_buffer.clear();
                        input_field = input_buffer;
                    }
                }
                else if (c >= 32) { //Accepted input characters
                    input_buffer.push_back(c);
                    input_field = input_buffer;
                    std::unique_lock<std::mutex> l(write_mutex);
                    condition_variable.notify_one();
                }
            }
        }

        void Console::write_loop() {
            while (!stop_loops) {
                std::unique_lock<std::mutex> l(write_mutex);
                condition_variable.wait(l);

                if (!writeQueue.empty()) { //Update input field and add a new line
                    std::string text = writeQueue.front();
                    writeQueue.pop();
                    std::cout << "\r" << text << "                                                                                " << std::endl;
                    std::cout << "> " << input_field;
                }
                else { //Update input field
                    std::cout << "\r> " << input_field << "                                                                                ";
                }
            }
        }

        void Console::write(std::string text) {
            std::unique_lock<std::mutex> l(write_mutex);
            writeQueue.push(text);
            condition_variable.notify_one();
        }

        bool Console::read(std::string& text) {
            std::unique_lock<std::mutex> l(read_mutex);
            if (!readQueue.empty()) {
                text = readQueue.front();
                readQueue.pop();
                return true;
            }
            return false;
        }

        void Console::initialize() {
            if (!initialized) {
                input_thread = new boost::thread(&input_loop);
                write_thread = new boost::thread(&write_loop);
                initialized = true;
            }
        }

        void Console::terminate() {
            stop_loops = true;
            condition_variable.notify_one();
            input_thread->interrupt();
            input_thread->join();
            write_thread->join();
        }
    }
}

Is there any flaws I have not noticed or some way I can improve performance? It's important that there's as little wait time when calling write().

Also, you are free to use the code for your own project if you want.

Answer 1

A couple of things stand out...

Initialise your pointers

You don't initialise your thread pointers, which means there's no way for you to tell if they're pointing at valid objects or not:

boost::thread*          Console::input_thread;
boost::thread*          Console::write_thread;

Initialise isn't protected

When you start introducing threads to your code, you need to think about them being used in a multi-threaded environment. If the client application calls initialize from two different threads, it is possible for your class to have multiple initialisations. The second thread can enter the if statement while the first is still between the marked sections below:

void Console::initialize() {
    if (!initialized) {
/* Between HERE <------------------------------------------*/
        input_thread = new boost::thread(&input_loop);
        write_thread = new boost::thread(&write_loop);
/* And HERE <---------------------------                   */
        initialized = true;
    }
}

Terminate Vs Destroy

Terminate shutsdown the threads, but doesn't destroy the objects / reinitialize the pointers. This puts your class into an inconsistent state. I can't call terminate then call initialize (you don't provide a start method) again without causing a leak. This limits the usage of your class to a single time per application, which seems unnecessarily restrictive.

Terminate also doesn't check to make sure the system is actually running, so calling terminate twice, or before initialize is going to produce unexpected results. It feels like this should be catered for by the API.