Refactor queue (217356ec) · Commits · olcf / container-builder

BuilderQueue/include/BuilderQueue.h

+2 −2

Original line number	Diff line number	Diff line
		@@ -9,7 +9,7 @@

		class BuilderQueue {
		public:
		explicit BuilderQueue(asio::io_service &io_service) : io_service(io_service),
		explicit BuilderQueue(asio::io_context &io_context) : io_context(io_context),
		max_builders(20),
		max_available_builders(5),
		pending_requests(0) {}
		@@ -21,7 +21,7 @@ public:
		void tick(asio::yield_context yield);

		private:
		asio::io_service &io_service;
		asio::io_context &io_context;

		// Hold reservations that are to be fulfilled
		std::list<Reservation> reservations;

BuilderQueue/include/Connection.h

+14 −15

Original line number	Diff line number	Diff line
		@@ -9,30 +9,29 @@ using asio::ip::tcp;

		class Connection : public std::enable_shared_from_this<Connection> {
		public:
		explicit Connection(BuilderQueue &queue) : queue(queue) {}
		explicit Connection(BuilderQueue &queue, Messenger messenger) : queue(queue),
		messenger(std::move(messenger)) {}

		~Connection() {
		logger::write("Ending connection");
		}

		void start(tcp::socket socket) {
		void start(asio::io_context& io_context) {
		try {
		auto self(shared_from_this());

		logger::write(socket, "Established connection");
		logger::write("Established connection");

		asio::spawn(socket.get_io_service(),
		[this, self, socket=std::move(socket)](asio::yield_context yield) mutable {

		Messenger client(std::move(socket), yield);

		auto request = client.async_receive(MessageType::string);
		if (client.error) {
		logger::write(client.socket, "Request failure" + client.error.message());
		asio::spawn(io_context,
		[this, self](asio::yield_context yield) {
		boost::system::error_code error;
		auto request = messenger.async_read_string(yield, error);
		if (error) {
		logger::write("Request failure" + error.message());
		} else if (request == "checkout_builder_request") {
		checkout_builder(client);
		checkout_builder(yield, error);
		} else {
		logger::write(client.socket, "Invalid request message received: " + request);
		logger::write("Invalid request message received: " + request);
		}

		});
		@@ -45,7 +44,7 @@ public:

		private:
		BuilderQueue &queue;
		Messenger messenger;

		// Checkout a builder
		void checkout_builder(Messenger& client);
		void checkout_builder(asio::yield_context yield, boost::system::error_code& error);
		};
		No newline at end of file

BuilderQueue/include/OpenStackBuilder.h

+5 −4

Original line number	Diff line number	Diff line
		#pragma once

		#include "boost/asio/spawn.hpp"
		#include "Builder.h"
		#include "boost/asio/io_context.hpp"
		#include "BuilderData.h"
		#include <set>

		namespace asio = boost::asio;

		namespace OpenStackBuilder {
		std::set<BuilderData> get_builders(asio::io_service &io_service, asio::yield_context yield);
		void request_create(asio::io_service& io_service, asio::yield_context yield);
		void destroy(BuilderData builder, asio::io_service& io_service, asio::yield_context yield);
		std::set<BuilderData> get_builders(asio::io_context &io_context, asio::yield_context yield, boost::system::error_code &error);
		void request_create(asio::io_context& io_context, asio::yield_context yield, boost::system::error_code &error);
		void destroy(BuilderData builder, asio::io_context& io_context, asio::yield_context yield, boost::system::error_code &error);
		};
		No newline at end of file

BuilderQueue/include/Reservation.h

+6 −6

Original line number	Diff line number	Diff line
		#pragma once

		#include <boost/asio/io_service.hpp>
		#include <boost/asio/io_context.hpp>
		#include <boost/asio/spawn.hpp>
		#include <boost/asio/deadline_timer.hpp>
		#include <boost/optional.hpp>
		#include "Builder.h"
		#include "BuilderData.h"
		#include "Messenger.h"

		namespace asio = boost::asio;

		@@ -19,16 +20,16 @@ enum class ReservationStatus {
		// Reservations are handled by the queue and assigned builders as available
		class Reservation {
		public:
		explicit Reservation(asio::io_service& io_service) :
		explicit Reservation(asio::io_context& io_context) :
		status(ReservationStatus::pending),
		ready_timer(io_service) {}
		ready_timer(io_context) {}

		~Reservation() {
		status = ReservationStatus::request_complete;
		}

		// Create an infinite timer that will be cancelled by the queue when the job is ready
		void async_wait(Messenger& client);
		void async_wait(asio::yield_context yield, boost::system::error_code &error);

		// Callback used by BuilderQueue to cancel the timer which signals our reservation is ready
		void ready(BuilderData acquired_builder);
		@@ -64,7 +65,6 @@ public:

		boost::optional<BuilderData> builder;
		ReservationStatus status;
		boost::system::error_code error;
		private:
		asio::deadline_timer ready_timer;
		};
		No newline at end of file

BuilderQueue/include/ReservationRequest.h

+3 −5

Original line number	Diff line number	Diff line
		#pragma once

		#include "BuilderQueue.h"
		#include "Builder.h"
		#include "BuilderData.h"

		// ReservationRequest breaks circular dependence between Reservation and BuilderQueue
		// We must remove a reservation from the queue if it is destructed but we don't want
		@@ -9,9 +9,8 @@
		class ReservationRequest {

		public:
		explicit ReservationRequest(BuilderQueue &queue, Messenger& client) :
		explicit ReservationRequest(BuilderQueue &queue) :
		queue(queue),
		client(client),
		reservation(queue.enter())
		{}

		@@ -19,10 +18,9 @@ public:
		reservation.set_request_complete();
		}

		BuilderData async_wait();
		BuilderData async_wait(asio::yield_context yield, boost::system::error_code &error);

		private:
		BuilderQueue &queue;
		Messenger& client;
		Reservation &reservation;
		};
		No newline at end of file