os: new Block::Request_stream API

Issue #3092

repos/os/include/block/request.h

@@ -0,0 +1,43 @@
+/*
+ * \brief  Block request
+ * \author Norman Feske
+ * \date   2018-12-17
+ */
+
+/*
+ * Copyright (C) 2018 Genode Labs GmbH
+ *
+ * This file is part of the Genode OS framework, which is distributed
+ * under the terms of the GNU Affero General Public License version 3.
+ */
+
+#ifndef _INCLUDE__BLOCK__REQUEST_H_
+#define _INCLUDE__BLOCK__REQUEST_H_
+
+/* Genode includes */
+#include <base/stdint.h>
+
+namespace Block { struct Request; }
+
+
+struct Block::Request
+{
+	enum class Operation : Genode::uint32_t { INVALID, READ, WRITE, SYNC };
+	enum class Success   : Genode::uint32_t { FALSE, TRUE };
+
+	Operation         operation;
+	Success           success;
+	Genode::uint64_t  block_number;
+	Genode::uint64_t  offset;
+	Genode::uint32_t  count;
+
+	bool operation_defined() const
+	{
+		return operation == Operation::READ
+		    || operation == Operation::WRITE
+		    || operation == Operation::SYNC;
+	}
+
+} __attribute__((packed));
+
+#endif /* _INCLUDE__BLOCK__REQUEST_H_ */

repos/os/include/block/request_stream.h

@@ -0,0 +1,317 @@
+/*
+ * \brief  Stream of block-operation requests
+ * \author Norman Feske
+ * \date   2018-12-17
+ */
+
+/*
+ * Copyright (C) 2018 Genode Labs GmbH
+ *
+ * This file is part of the Genode OS framework, which is distributed
+ * under the terms of the GNU Affero General Public License version 3.
+ */
+
+#ifndef _INCLUDE__BLOCK__REQUEST_STREAM_H_
+#define _INCLUDE__BLOCK__REQUEST_STREAM_H_
+
+/* Genode includes */
+#include <block_session/block_session.h>
+#include <packet_stream_tx/rpc_object.h>
+#include <block/request.h>
+
+namespace Block { struct Request_stream; }
+
+
+class Block::Request_stream : Genode::Noncopyable
+{
+	public:
+
+		/**
+		 * Interface for accessing the content of a 'Request'
+		 *
+		 * The 'Payload' is separate from 'Request_stream' to allow its use
+		 * as argument without exposing the entirely of the 'Request_stream'
+		 * to the called code.
+		 */
+		class Payload
+		{
+			private:
+
+				friend class Request_stream;
+
+				Genode::addr_t   const _base;
+				Genode::size_t   const _size;
+				Genode::uint32_t const _block_size;
+
+				/**
+				 * Return pointer to the first byte of the request content
+				 */
+				void *_request_ptr(Block::Request request) const
+				{
+					return (void *)(_base + request.offset);
+				}
+
+				/**
+				 * Return request size in bytes
+				 */
+				Genode::size_t _request_size(Block::Request const &request) const
+				{
+					return request.count * _block_size;
+				}
+
+				bool _valid_range(Block::Request const &request) const
+				{
+					/* local address of the last byte of the request */
+					Genode::addr_t const request_end = _base + request.offset
+					                                 + _request_size(request) - 1;
+
+					/* check for integer overflow and zero-sized request */
+					if (request_end <= _base)
+						return false;
+
+					/* check for upper bound */
+					if (request_end > _base + _size - 1)
+						return false;
+
+					return true;
+				}
+
+				Payload(Genode::addr_t base, Genode::size_t size,
+				        Genode::uint32_t block_size)
+				:
+					_base(base), _size(size), _block_size(block_size)
+				{ }
+
+			public:
+
+				/**
+				 * Call functor 'fn' with the pointer and size of the 'request'
+				 * content as arguments
+				 *
+				 * If the request does not carry any payload, 'fn' is not
+				 * called.
+				 */
+				template <typename FN>
+				void with_content(Block::Request request, FN const &fn) const
+				{
+					if (_valid_range(request))
+						fn(_request_ptr(request), _request_size(request));
+				}
+		};
+
+	private:
+
+		Packet_stream_tx::Rpc_object<Block::Session::Tx> _tx;
+
+		typedef Genode::Packet_stream_sink<Block::Session::Tx_policy> Tx_sink;
+
+		Payload const _payload;
+
+	public:
+
+		Request_stream(Genode::Region_map               &rm,
+		               Genode::Dataspace_capability      ds,
+		               Genode::Entrypoint               &ep,
+		               Genode::Signal_context_capability sigh,
+		               Genode::uint32_t                  block_size)
+		:
+			_tx(ds, rm, ep.rpc_ep()),
+			_payload(_tx.sink()->ds_local_base(), _tx.sink()->ds_size(), block_size)
+		{
+			_tx.sigh_ready_to_ack(sigh);
+			_tx.sigh_packet_avail(sigh);
+		}
+
+		virtual ~Request_stream()
+		{
+			_tx.sigh_ready_to_ack(Genode::Signal_context_capability());
+			_tx.sigh_packet_avail(Genode::Signal_context_capability());
+		}
+
+		Genode::Capability<Block::Session::Tx> tx_cap() { return _tx.cap(); }
+
+		/**
+		 * Call functor 'fn' with 'Payload' interface as argument
+		 *
+		 * The 'Payload' interface allows the functor to access the content
+		 * of a request by calling 'Payload::with_content'.
+		 */
+		template <typename FN>
+		void with_payload(FN const &fn) const { fn(_payload); }
+
+		/**
+		 * Call functor 'fn' with the pointer and size to the 'request' content
+		 *
+		 * This is a wrapper for 'Payload::with_content'. It is convenient
+		 * in situations where the 'Payload' interface does not need to be
+		 * propagated as argument.
+		 */
+		template <typename FN>
+		void with_content(Request const &request, FN const &fn) const
+		{
+			_payload.with_content(request, fn);
+		}
+
+		enum class Response { ACCEPTED, REJECTED, RETRY };
+
+		/**
+		 * Call functor 'fn' for each pending request, with its packet as argument
+		 *
+		 * The boolean return value of the functor expresses whether the request
+		 * was accepted or not. If it was accepted, the request is removed from the
+		 * packet stream. If the request could not be accepted, the iteration
+		 * aborts and the request packet stays in the packet stream.
+		 */
+		template <typename FN>
+		void with_requests(FN const &fn)
+		{
+			Tx_sink &tx_sink = *_tx.sink();
+
+			using namespace Genode;
+
+			for (;;) {
+
+				if (!tx_sink.packet_avail())
+					return;
+
+				typedef Block::Packet_descriptor Packet_descriptor;
+
+				Packet_descriptor const packet = tx_sink.peek_packet();
+
+				auto operation = [] (Packet_descriptor::Opcode op)
+				{
+					switch (op) {
+					case Packet_descriptor::READ:  return Request::Operation::READ;
+					case Packet_descriptor::WRITE: return Request::Operation::WRITE;
+					case Packet_descriptor::END:   return Request::Operation::INVALID;
+					};
+					return Request::Operation::INVALID;
+				};
+
+				bool const packet_valid = (tx_sink.packet_valid(packet)
+				                       && (packet.offset() >= 0)
+				                       && (packet.size() <= (size_t)((uint32_t)~0UL)));
+
+				Request request { .operation    = operation(packet.operation()),
+				                  .success      = Request::Success::FALSE,
+				                  .block_number = (uint64_t)packet.block_number(),
+				                  .offset       = (uint64_t)packet.offset(),
+				                  .count        = (uint32_t)packet.block_count() };
+
+				Response const response = packet_valid
+				                        ? fn(request)
+				                        : Response::REJECTED;
+				bool progress = false;
+
+				switch (response) {
+
+				case Response::REJECTED:
+
+					/*
+					 * Acknowledge rejected packet if there is enough room in
+					 * the acknowledgement queue. Otherwise, the rejected
+					 * packet stays in the request queue and is evaluated
+					 * again.
+					 */
+					if (tx_sink.ack_slots_free()) {
+						(void)tx_sink.try_get_packet();
+						tx_sink.try_ack_packet(packet);
+						progress = true;
+					}
+					break;
+
+				case Response::ACCEPTED:
+					(void)tx_sink.try_get_packet();
+					progress = true;
+					break;
+
+				case Response::RETRY:
+					break;
+				}
+
+				/*
+				 * Stop iterating of no request-queue elements can be consumed.
+				 */
+				if (!progress)
+					break;
+			}
+		}
+
+		class Ack
+		{
+			private:
+
+				friend class Request_stream;
+
+				Tx_sink &_tx_sink;
+
+				bool _submitted = false;
+
+				Genode::uint32_t const _block_size;
+
+				Ack(Tx_sink &tx_sink, Genode::uint32_t block_size)
+				: _tx_sink(tx_sink), _block_size(block_size) { }
+
+			public:
+
+				void submit(Block::Request request)
+				{
+					if (_submitted) {
+						Genode::warning("attempt to ack the same packet twice");
+						return;
+					}
+
+					typedef Block::Packet_descriptor Packet_descriptor;
+					Packet_descriptor packet { (Genode::off_t)request.offset,
+					                           request.count * _block_size };
+
+					auto opcode = [] (Request::Operation operation)
+					{
+						switch (operation) {
+						case Request::Operation::READ:    return Packet_descriptor::READ;
+						case Request::Operation::WRITE:   return Packet_descriptor::WRITE;
+						case Request::Operation::SYNC:    return Packet_descriptor::END;
+						case Request::Operation::INVALID: return Packet_descriptor::END;
+						};
+						return Packet_descriptor::END;
+					};
+
+					packet = Packet_descriptor(packet, opcode(request.operation),
+					                           request.block_number, request.count);
+
+					packet.succeeded(request.success == Request::Success::TRUE);
+
+					_tx_sink.try_ack_packet(packet);
+					_submitted = true;
+				}
+		};
+
+		/**
+		 * Try to submit acknowledgement packets
+		 *
+		 * The method repeatedly calls the functor 'fn' with an 'Ack' reference,
+		 * which provides an interface to 'submit' one acknowledgement. The
+		 * iteration stops when the acknowledgement queue is fully populated or if
+		 * the functor does not call 'Ack::submit'.
+		 */
+		template <typename FN>
+		void try_acknowledge(FN const &fn)
+		{
+			Tx_sink &tx_sink = *_tx.sink();
+
+			while (tx_sink.ack_slots_free()) {
+
+				Ack ack(tx_sink, _payload._block_size);
+
+				fn(ack);
+
+				if (!ack._submitted)
+					break;
+			}
+		}
+
+		void wakeup_client() { _tx.sink()->wakeup(); }
+};
+
+
+#endif /* _INCLUDE__BLOCK__REQUEST_STREAM_H_ */

repos/os/run/block_request_stream.run

@@ -0,0 +1,62 @@
+build { core init timer test/block_request_stream app/block_tester }
+
+create_boot_directory
+
+install_config {
+<config>
+	<parent-provides>
+		<service name="ROM"/>
+		<service name="RAM"/>
+		<service name="IRQ"/>
+		<service name="IO_MEM"/>
+		<service name="IO_PORT"/>
+		<service name="PD"/>
+		<service name="RM"/>
+		<service name="CPU"/>
+		<service name="LOG"/>
+	</parent-provides>
+
+	<default-route>
+		<any-service> <parent/> <any-child/> </any-service>
+	</default-route>
+
+	<default caps="100"/>
+	<start name="timer">
+		<resource name="RAM" quantum="1M"/>
+		<provides><service name="Timer"/></provides>
+	</start>
+
+	<start name="test-block_request_stream">
+		<resource name="RAM" quantum="1M"/>
+		<provides><service name="Block"/></provides>
+		<config/>
+		<route> <any-service> <parent/> </any-service> </route>
+	</start>
+
+	<start name="block_tester">
+		<resource name="RAM" quantum="32M"/>
+		<config verbose="yes" report="no" log="yes" stop_on_error="no">
+			<tests>
+				<sequential length="1M" size="4K" synchronous="yes"/>
+				<sequential length="1M" size="8K" synchronous="yes"/>
+				<sequential length="1M" size="16K"/>
+				<sequential length="1M" size="64K"/>
+				<sequential length="1M" size="128K"/>
+				<sequential length="1M" size="4K" synchronous="yes" write="yes"/>
+				<sequential length="1M" size="64K" write="yes" synchronous="yes"/>
+			</tests>
+		</config>
+		<route>
+			<service name="Block"> <child name="test-block_request_stream"/> </service>
+			<any-service> <parent/> <any-child/> </any-service>
+		</route>
+	</start>
+</config>}
+
+
+build_boot_image { core init timer block_tester test-block_request_stream ld.lib.so }
+
+
+append qemu_args " -nographic -m 512 "
+
+run_genode_until {.*child "block_tester" exited with exit value 0.*\n} 360

repos/os/src/test/block_request_stream/main.cc

@@ -0,0 +1,247 @@
+/*
+ * \brief  Example block service
+ * \author Norman Feske
+ * \date   2018-12-06
+ */
+
+/*
+ * Copyright (C) 2018 Genode Labs GmbH
+ *
+ * This file is part of the Genode OS framework, which is distributed
+ * under the terms of the GNU Affero General Public License version 3.
+ */
+
+/* Genode includes */
+#include <block/request_stream.h>
+#include <base/component.h>
+#include <base/attached_ram_dataspace.h>
+#include <root/root.h>
+
+namespace Test {
+
+	struct Block_session_component;
+	template <unsigned> struct Jobs;
+
+	struct Main;
+
+	using namespace Genode;
+}
+
+
+struct Test::Block_session_component : Rpc_object<Block::Session>,
+                                       Block::Request_stream
+{
+	Entrypoint &_ep;
+
+	static constexpr size_t BLOCK_SIZE = 4096;
+	static constexpr size_t NUM_BLOCKS = 16;
+
+	Block_session_component(Region_map               &rm,
+	                        Dataspace_capability      ds,
+	                        Entrypoint               &ep,
+	                        Signal_context_capability sigh)
+	:
+		Request_stream(rm, ds, ep, sigh, BLOCK_SIZE), _ep(ep)
+	{
+		_ep.manage(*this);
+	}
+
+	~Block_session_component() { _ep.dissolve(*this); }
+
+	void info(Block::sector_t *count, size_t *block_size, Operations *ops) override
+	{
+		*count      = NUM_BLOCKS;
+		*block_size = BLOCK_SIZE;
+		*ops        = Operations();
+
+		ops->set_operation(Block::Packet_descriptor::Opcode::READ);
+		ops->set_operation(Block::Packet_descriptor::Opcode::WRITE);
+	}
+
+	void sync() override { }
+
+	Capability<Tx> tx_cap() override { return Request_stream::tx_cap(); }
+};
+
+
+template <unsigned N>
+struct Test::Jobs : Noncopyable
+{
+	struct Entry
+	{
+		Block::Request request;
+
+		enum State { UNUSED, IN_PROGRESS, COMPLETE } state;
+	};
+
+	Entry _entries[N] { };
+
+	bool acceptable(Block::Request) const
+	{
+		for (unsigned i = 0; i < N; i++)
+			if (_entries[i].state == Entry::UNUSED)
+				return true;
+
+		return false;
+	}
+
+	void submit(Block::Request request)
+	{
+		for (unsigned i = 0; i < N; i++) {
+			if (_entries[i].state == Entry::UNUSED) {
+				_entries[i] = Entry { .request = request,
+				                      .state   = Entry::IN_PROGRESS };
+				return;
+			}
+		}
+
+		error("failed to accept request");
+	}
+
+	bool execute()
+	{
+		bool progress = false;
+		for (unsigned i = 0; i < N; i++) {
+			if (_entries[i].state == Entry::IN_PROGRESS) {
+				_entries[i].state = Entry::COMPLETE;
+				_entries[i].request.success = Block::Request::Success::TRUE;
+				progress = true;
+			}
+		}
+
+		return progress;
+	}
+
+	void completed_job(Block::Request &out)
+	{
+		out = Block::Request { };
+
+		for (unsigned i = 0; i < N; i++) {
+			if (_entries[i].state == Entry::COMPLETE) {
+				out = _entries[i].request;
+				_entries[i].state = Entry::UNUSED;
+				return;
+			}
+		}
+	}
+
+	/**
+	 * Apply 'fn' with completed job, reset job
+	 */
+	template <typename FN>
+	void with_any_completed_job(FN const &fn)
+	{
+		Block::Request request { };
+
+		completed_job(request);
+
+		if (request.operation_defined())
+			fn(request);
+	}
+};
+
+
+struct Test::Main : Rpc_object<Typed_root<Block::Session> >
+{
+	Env &_env;
+
+	Constructible<Attached_ram_dataspace> _block_ds { };
+
+	Constructible<Block_session_component> _block_session { };
+
+	Signal_handler<Main> _request_handler { _env.ep(), *this, &Main::_handle_requests };
+
+	Jobs<10> _jobs { };
+
+	void _handle_requests()
+	{
+		if (!_block_session.constructed())
+			return;
+
+		Block_session_component &block_session = *_block_session;
+
+		for (;;) {
+
+			bool progress = false;
+
+			/* import new requests */
+			block_session.with_requests([&] (Block::Request request) {
+
+				if (!_jobs.acceptable(request))
+					return Block_session_component::Response::RETRY;
+
+				/* access content of the request */
+				block_session.with_content(request, [&] (void *ptr, size_t size) {
+					(void)ptr;
+					(void)size;
+				});
+
+				_jobs.submit(request);
+
+				progress = true;
+
+				return Block_session_component::Response::ACCEPTED;
+			});
+
+			/* process I/O */
+			progress |= _jobs.execute();
+
+			/* acknowledge finished jobs */
+			block_session.try_acknowledge([&] (Block_session_component::Ack &ack) {
+
+				_jobs.with_any_completed_job([&] (Block::Request request) {
+					progress |= true;
+					ack.submit(request);
+				});
+			});
+
+			if (!progress)
+				break;
+		}
+
+		block_session.wakeup_client();
+	}
+
+
+	/*
+	 * Root interface
+	 */
+
+	Capability<Session> session(Root::Session_args const &args,
+	                            Affinity const &) override
+	{
+		log("new block session: ", args.string());
+
+		size_t const ds_size =
+			Arg_string::find_arg(args.string(), "tx_buf_size").ulong_value(0);
+
+		Ram_quota const ram_quota = ram_quota_from_args(args.string());
+
+		if (ds_size >= ram_quota.value) {
+			warning("communication buffer size exceeds session quota");
+			throw Insufficient_ram_quota();
+		}
+
+		_block_ds.construct(_env.ram(), _env.rm(), ds_size);
+		_block_session.construct(_env.rm(), _block_ds->cap(), _env.ep(),
+		                         _request_handler);
+
+		return _block_session->cap();
+	}
+
+	void upgrade(Capability<Session>, Root::Upgrade_args const &) override { }
+
+	void close(Capability<Session>) override
+	{
+		_block_session.destruct();
+		_block_ds.destruct();
+	}
+
+	Main(Env &env) : _env(env)
+	{
+		_env.parent().announce(_env.ep().manage(*this));
+	}
+};
+
+
+void Component::construct(Genode::Env &env) { static Test::Main inst(env); }

repos/os/src/test/block_request_stream/target.mk

@@ -0,0 +1,3 @@
+TARGET := test-block_request_stream
+SRC_CC := main.cc
+LIBS   += base