hackbunny fucked around with this message at 17:12 on Nov 29, 2016

hackbunny fucked around with this message at 17:19 on Nov 29, 2016

// this is not an emptyquoted block of code: rather, this is my post, in the
// form of comments to A Code. a little like literate programming if you think
// about it, except bad. yosprogramming (c) hackbunny 2016 do not steal!!!

#include <cassert>
#include <cstdlib>

#include <deque>
#include <mutex>
#include <condition_variable>
#include <type_traits>

template<class T, class AllocatorT = std::allocator<T>>
class object_pipe
{
public:
	using value_type = T;
	using allocator_type = AllocatorT;

private:
	// PSA: using a mutex around a deque is a terrible way to do what I'm
	// doing. boost's lock-free SPSC queue uses a circular buffer backed by
	// a fixed-size array, and I believe Java's equivalent uses multiple
	// chunks, independently locked (and if it's not the queue that does
	// this, there's another thread-safe container that does). this solution
	// is already massively overengineered for the amount of data it deals
	// with, so we'll stay with the dumb monkey implementation
	std::deque<std::unique_ptr<value_type>, allocator_type> m_queue;

	// like I promised, this is now mutable so that const instances of
	// object_pipe can use the lock too
	std::mutex mutable m_queue_mutex;

	std::condition_variable m_queue_changed;

	// I realized that object_pipe can grow without bounds, while the
	// behavior of a UNIX pipe is that writers will block if the pipe is
	// "full". let's add a maximum size parameter, that defaults to the
	// maximum capacity of our backing queue
	std::size_t const m_max_size = m_queue.max_size();

	// I promised I would split the eof flag and I did. the two ends of the
	// pipe can now be closed independently, but each end can only be closed
	// once
	bool m_closed_write;
	bool m_closed_read;

private:
	// the eof flag is now the closed flag, because "eof" proper is when the
	// pipe is closed *and* empty. named xxx_unlocked after getc_unlocked,
	// indicating that the lock must be acquired beforehand
	bool is_closed_unlocked() const
	{
		return m_closed_write || m_closed_read;
	}

	// this flag may be useful to poll the read end without blocking
	bool is_empty_unlocked() const
	{
		return m_queue.empty();
	}

	// finally, meet the new and improved eof flag
	bool is_eof_unlocked() const
	{
		return is_closed_unlocked() && is_empty_unlocked();
	}

	std::size_t max_size_unlocked() const
	{
		assert(m_max_size <= m_queue.max_size());
		return m_max_size;
	}

	bool is_full_unlocked() const
	{
		return m_queue.size() >= max_size_unlocked();
	}

public:
	void write(value_type&& value)
	{
		// TODO we should use our allocator here, ugh. I don't know how,
		// yet. why doesn't unique_ptr natively support allocators?!
		write(std::make_unique<value_type>(std::move(value)));
	}

	// and I guess this unique_ptr should be specialized for our allocator
	// too? and we should write a make_unique static method for this class
	// that returns appropriately specialized allocators??? or should it be
	// a member method??? can I copy the allocator or should all unique_ptrs
	// share the same allocator instance through... what, a shared_ptr? and
	// what will allocate the shared_ptr??? jesus christ I hate allocators
	void write(std::unique_ptr<value_type>&& p)
	{
		if (!p) {
			throw std::exception();
		}

		{
			// now that write can block too (if the queue is full),
			// write, too, will need to wait on the condition
			// variable, which means unique_lock instead of
			// lock_guard
			std::unique_lock<std::mutex> lock(m_queue_mutex);

			// if the queue is full, wait until it's no longer full.
			// exit early if the queue is closed
			while (!is_closed_unlocked() && is_full_unlocked()) {
				// aren't you glad that we made read signal the
				// condition variable too, instead of chickening
				// out and using notify_all? we can safely block
				// here, secure in the knowledge that, if a
				// reader came along and freed space in the
				// queue, we would awaken
				m_queue_changed.wait(lock);
			}

			if (is_closed_unlocked()) {
				throw std::exception();
			}

			assert(!is_full_unlocked());
			m_queue.emplace_back(std::move(p));
		}

		m_queue_changed.notify_one();
	}

	// non-blocking and timed variants of write raise lots of questions!
	// like if the write would block (or times out), what to do with the
	// object that wasn't written? as always, my answer is "gently caress it"

	template<class... Args>
	void emplace(Args&&... args)
	{
		write(std::make_unique<value_type>(std::forward<Args>(args)...));
	}

	void close_write()
	{
		{
			std::lock_guard<std::mutex> lock(m_queue_mutex);

			if (m_closed_write) {
				throw std::exception();
			}

			m_closed_write = true;
			assert(is_closed_unlocked());
		}

		m_queue_changed.notify_one();
	}

	std::unique_ptr<value_type> read()
	{
		std::unique_ptr<value_type> p;

		{
			std::unique_lock<std::mutex> lock(m_queue_mutex);

			while (is_empty_unlocked() && !is_closed_unlocked()) {
				m_queue_changed.wait(lock);
			}

			if (!is_empty_unlocked()) {
				assert(!is_eof_unlocked());
				p = std::move(m_queue.front());
				assert(p);
				m_queue.pop_front();
			}
			else if (is_closed_unlocked()) {
				assert(is_eof_unlocked());
				throw std::exception();
			}
			else {
				// assert(0) is a semi-standard way of saying
				// that a branch is unreachable
				assert(0);
				std::abort();
			}
		}

		m_queue_changed.notify_one();

		assert(p);
		return p;
	}

	// read_value died on its way back to its home planet

	// inspired by boost's SPSC queue, I'll add a method to receive the next
	// object without messing with smart pointers or move constructors that
	// may throw etc.
	template<class Consumer>
	void consume(Consumer consumer)
	{
		consumer(*read());
	}

	void close_read()
	{
		{
			std::lock_guard<std::mutex> lock(m_queue_mutex);

			if (m_closed_read) {
				throw std::exception();
			}

			m_closed_read = true;
			assert(is_closed_unlocked());

			m_queue.clear();
			assert(is_eof_unlocked());
		}

		m_queue_changed.notify_one();
	}

	// thanks to the new, mutable mutex, this method can now be const
	bool is_eof() const
	{
		std::lock_guard<std::mutex> lock(m_queue_mutex);
		return is_eof_unlocked();
	}


};

hackbunny fucked around with this message at 18:06 on Nov 29, 2016

Sweeper fucked around with this message at 19:24 on Nov 29, 2016