Alepha/Atomic/Dropbox.h

static_assert( __cplusplus > 2020'99 );

/*! 
 * @file Dropbox.h
 * @brief Class which abstracts a "dropbox" metaphor, for threaded programming
 *
 * Copyright (C) 2024 Alepha Library. All rights reserved. <BR>
 * @author ADAM David Alan Martin
 */

#pragma once

#include <Alepha/Alepha.h>

#include <vector>
#include <exception>

#include <Alepha/Thread.h>
#include <Alepha/Exception.h>
#include <Alepha/assertion.h>
#include <Alepha/error.h>

#include "DeadDrop.h"

namespace Alepha::Hydrogen::Atomic  ::detail::  Dropbox_m
{
	namespace C
	{
		const bool debug= false;
		const bool debugInterlock= false or C::debug;
		const bool debugPop= false or C::debug;
	}


	enum Dropbox_tag {};

	inline namespace exports
	{
		template< typename > class Dropbox;

		using DropboxFinishedException= synthetic_exception< Dropbox_tag, FinishedException >;
		using AnyTaggedDropboxFinishedException= AnyTagged< DropboxFinishedException >;
		template< typename tag > using TaggedDropboxFinishedException= Tagged< DropboxFinishedException, tag >;

		using DropboxFinishedCondition= synthetic_exception< Dropbox_tag, FinishedCondition, DropboxFinishedException >;
		using AnyTaggedDropboxFinishedCondition= AnyTagged< DropboxFinishedCondition >;
		template< typename tag > using TaggedDropboxFinishedCondition= Tagged< DropboxFinishedCondition, tag >;
	}

	template< typename T >
	std::size_t
	computeWeight( const T & )
	{
		return sizeof( T );
	}

	/*!
	 * @brief         The Dropbox class implements a Dropbox metaphor.  Just like anonymous dead-drop boxes,
	 *                the Dropbox class lets users add items to a box from a single producer thread
	 *                and in a single consumer thread consume the contents of the mailbox.
	 *
	 * @tparam Item   The Item type which is used in the dropbox containers.
	 *
	 * @note          Dropboxes only support one consumer and one producer.
	 *
	 * @note          Dropboxes only block on attempting to acquire the next `DeadDrop` to use after exhaustion.
	 *
	 * @invariant     Dropboxes give out their contents in the order placed in.
	 *
	 *                The Dropbox primitive is implemented as DeadDrop object which is rotated periodically.
	 *                When a producer or consumer exhausts the current DeadDrop port, it `release`es it.  When
	 *                it needs a new DeadDrop port to actually use, it will `acquire` it.  Only `acquire` blocks.
	 */
	template< typename Item >
	class exports::Dropbox
	{
		private:
			enum Finished_tag {};

		public:
			using FinishedException= synthetic_exception< Dropbox_tag, DropboxFinishedException >;
			//using AnyTaggedFinishedException= AnyTagged< FinishedException >;
			//template< typename tag > using TaggedFinishedException= Tagged< FinishedException, tag >;

			using FinishedCondition= synthetic_exception< Dropbox_tag, DropboxFinishedCondition, FinishedException >;
			//using AnyTaggedFinishedCondition= AnyTagged< FinishedCondition >;
			//template< typename tag > using TaggedFinishedCondition= Tagged< FinishedCondition, tag >;

		private:
			const std::size_t boxLimit;
			std::size_t weight= 0;

			struct Box
			{
				std::vector< Item > items;
				std::vector< Item >::iterator pos= begin( items );
				std::exception_ptr notification;
			};
			DeadDrop< Box > drops;

			void
			sendoff()
			{
				auto &producer= drops.producer.acquireBox();
				producer.pos= begin( producer.items );
				weight= 0;

				drops.producer.releaseBox();
			}

			void
			recycle()
			{
				auto &filled= drops.consumer.acquireBox();
				filled.notification= nullptr;
				filled.items.clear();
				drops.consumer.releaseBox();
			}

		public:
			explicit
			Dropbox( const size_t lim )
				: boxLimit( lim )
			{
				recycle();
			}

			Item
			pop()
			{
				auto &filled= drops.consumer.acquireBox();
				if( filled.pos == end( filled.items ) )
				{
					[[unlikely]];
					auto notification= std::move( filled.notification );
					recycle();
					std::rethrow_exception( std::move( notification ) );
				}
				assert( not filled.items.empty() );
				auto &rv= *filled.pos++;
				// If it was the last item and we're going to rotate/reccyle,
				// be sure that we don't reference the now-stale value.
				// This is a special case that will only come up at the end of each queue
				// cycle.
				//
				// I need to research if NRV applies in this circumstance -- if so, then the
				// `rv` variable can be an instance instead of a reference.
				if( filled.pos == end( filled.items ) and filled.notification == nullptr )
				{
					[[unlikely]];
					auto rvTemp= std::move( rv );
					recycle();
					return rvTemp;
				}
				return std::move( rv );
			}

			bool pushWouldBlock() const noexcept { return weight > boxLimit; }
				

			void
			push( Item item )
			{
				if( pushWouldBlock() ) [[unlikely]] sendoff();
				assertion( weight <= boxLimit );

				auto &preparing= drops.producer.acquireBox();

				weight+= computeWeight( item );
				preparing.items.push_back( std::move( item ) );
			}

			void
			notify( auto newNotification )
			{
				auto &preparing= drops.producer.acquireBox();
				preparing.notification= std::make_exception_ptr( std::move( newNotification ) );
				sendoff();
			}

			void
			finish( auto newNotification )
			{
				auto &preparing= drops.producer.acquireBox();
				preparing.notification= std::make_exception_ptr( std::move( newNotification ) );
				sendoff();
			}

			void
			finish()
			{
				finish( build_exception< FinishedCondition >( "" ) );
			}
	};
}

namespace Alepha::Hydrogen::Atomic::inline exports::inline Dropbox_m
{
	using namespace detail::Dropbox_m::exports;
}