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.

• Loki Astari: Note: Thanks for the explicit permission. But just as a heads up for future posts; by posting on this site you are making the code available for all to use under a creative commons license. See the bottom of this page for details. user contributions licensed under cc by-sa 3.0 with attribution required -> link

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.

• KaareZ: What do you mean by not initializing my thread pointers? And thanks for your review.
• forsvarir: @KaareZ I would expect them to be assigned to nullptr at instantiation, that way you can check for nullptr in your initialize and terminate methods.
• KaareZ: Like this? boost::thread* Console::input_thread = nullptr
• forsvarir: @KaareZ yes, like that.