path: root/modules/remote-sycl/c++/remote.hpp

#pragma once

#include <forstio/io_codec/rpc.hpp>
#include <forstio/codec/data.hpp>
#include <forstio/codec/id_map.hpp>

#include <AdaptiveCpp/CL/sycl.hpp>

namespace saw {
namespace rmt {
struct Sycl {};
}

template<>
class remote<rmt::Sycl>;

/**
 * Remote data class for the Sycl backend.
 */
template<typename T, typename Encoding>
class remote_data<T, Encoding, rmt::Sycl> {
private:
	id<T> id_;
	id_map<T,Encoding>* map_;
public:
	/**
	 * Main constructor
	 */
	remote_data(const id<T>& id, id_map<T, Encoding>& map):
		id_{id},
		map_{&map}
	{}

	/**
	 * Request data asynchronously
	 */
	conveyor<data<T,Encoding>> on_receive(); /// Stopped here
};

/**
 * 
 */
template<typename T, uint64_t N>
class data<schema::Primitive<T,N>, encode::Native<rmt::Sycl>> {
public:
	using Schema = schema::Primitive<T,N>;
	using NativeType = typename native_data_type<Schema>::type;
private:
	/**
	 * 
	 */
	NativeType val_;
public:
	/**
	 *
	 */
	data(NativeType val__):
		val_{val__}
	{}

	NativeType get(){
		return val_;
	}
};

template<typename T, uint64_t D>
class data<schema::Array<T,D>, encode::Native<rmt::Sycl>> {
public:
	using Schema = schema::Array<T,D>;
private:
	uint64_t total_length_;
	typename native_data_type<T>::type* device_data_;
	// data<T>* device_data_;
	cl::sycl::queue* queue_;

	static_assert(is_primitive<T>::value, "Only supports primitives for now");
	static_assert(D==1u, "For now we only support 1D Arrays");
public:
	data(uint64_t size, cl::sycl::queue& q__):
		total_length_{size},
		device_data_{cl::sycl::malloc_device<typename native_data_type<T>::type>(size, q__)},
		queue_{&q__}
	{
		if(!device_data_){
			total_length_ = 0u;
		}
	}

	template<typename Encoding>
	data(const data<Schema, Encoding>& from, cl::sycl::queue& q__):
		total_length_{from.size()},
		device_data_{cl::sycl::malloc_device<typename native_data_type<T>::type>(from.size(), q__)},
		queue_{&q__}
	{
		if(!device_data_){
			total_length_ = 0u;
		}
	}

	~data(){
		// free data
		if(device_data_){
			/// SYCL FREE
			cl::sycl::free(device_data_, *queue_);
		}
	}

	data<T,encode::Native<rmt::Sycl>>& at(uint64_t i){
		return device_data_[i];
	}
};

namespace impl {

template<typename Iface, typename Encoding>
struct rpc_id_map_helper {
	static_assert(always_false<Iface, Encoding>, "Only support Interface schema types.");
};

template<typename... Members, typename Encoding>
struct rpc_id_map_helper<schema::Interface<Members...>, Encoding> {
	std::tuple<id_map<typename Members::ValueType::ResponseT, Encoding>...> maps;
};
}
/**
 * Rpc Client class for the Sycl backend.
 */
template<typename Iface, typename Encoding>
class rpc_client<Iface, Encoding, rmt::Sycl> {
public:
private:
	/**
	 * Server this client is tied to
	 */
	rpc_server<Iface, Encoding, rmt::Sycl>* srv_;
public:
	rpc_client(rpc_server<Iface, Encoding, rmt::Sycl>& srv):
		srv_{&srv}
	{}

	/**
	 * Rpc call
	 */
	template<string_literal Name, typename ClientEncoding>
	error_or<
		id<
			typename schema_member_type<Name, Iface>::type::ResponseT
		>
	> call(data_or_id<typename schema_member_type<Name, Iface>::type::RequestT, ClientEncoding> input){
		return make_error<err::not_implemented>("RpcClient side is not implemented");
	}

};

/**
 * Rpc Server class for the Sycl backend.
 */
template<typename Iface, typename Encoding>
class rpc_server<Iface, Encoding, rmt::Sycl> {
public:
	using InterfaceCtxT = cl::sycl::queue*;
	using InterfaceT = interface<Iface, Encoding, InterfaceCtxT>;
private:
	/**
	 * Command queue for the sycl backend
	 */
	cl::sycl::queue cmd_queue_;

	/**
	 * The interface including the relevant context class.
	 */
	interface<Iface, Encoding, InterfaceCtxT> cl_interface_;

	/**
	 * Basic storage for response data.
	 */
	impl::rpc_id_map_helper<Iface, Encoding> storage_;
public:
	rpc_server(interface<Iface, Encoding, InterfaceCtxT> cl_iface):
		cmd_queue_{},
		cl_interface_{std::move(cl_iface)},
		storage_{}
	{}

	template<typename IdT>
	remote_data<IdT, Encoding, rmt::Sycl> request_data(id<IdT> dat){
		return {dat, std::get<id_map<IdT, Encoding>>(storage_.maps)};
	}

	/**
	 * Rpc call
	 */
	template<string_literal Name, typename ClientAllocation>
	error_or<
		id<
			typename schema_member_type<Name, Iface>::type::ResponseT
		>
	> call(data_or_id<typename schema_member_type<Name, Iface>::type::RequestT, ClientAllocation> input){

		/**
		 * First check if it's data or an id.
		 * If it's an id, check if it's registered within the storage and retrieve it.
		 */
		auto eoinp = [&,this]() -> error_or<data<typename schema_member_type<Name, Iface>::type::RequestT, Encoding>* > {
			if(input.is_id()){
				// storage_.maps
				auto& inner_map = std::get<id_map<typename schema_member_type<Name, Iface>::type::RequestT, Encoding >> (storage_.maps);
				auto eov = inner_map.find(input.get_id());
				if(eov.is_error()){
					return std::move(eov.get_error());
				}
				return eov.get_value();
			} else {
				return &input.get_data();
			}
		}();
		if(eoinp.is_error()){
			return std::move(eoinp.get_error());
		}
		auto& inp = *(eoinp.get_value());

		auto eod = cl_interface_.template call<Name>(std::move(inp), &cmd_queue_);

		if(eod.is_error()){
			return std::move(eod.get_error());
		}

		/**
		 * Store returned data in rpc storage
		 */
		auto& val = eod.get_value();
		auto& inner_map = std::get<id_map<typename schema_member_type<Name, Iface>::type::RequestT, Encoding >> (storage_.maps);
		auto eoid = inner_map.insert(std::move(val));
		if(eoid.is_error()){
			return std::move(eoid.get_error());
		}
		return eoid.get_value();
	}
};


template<>
struct remote_address<rmt::Sycl> {
private:
	remote<rmt::Sycl>* ctx_;

	SAW_FORBID_COPY(remote_address);
	SAW_FORBID_MOVE(remote_address);
public:
	remote_address(remote<rmt::Sycl>& r_ctx):
		ctx_{&r_ctx}
	{}
};

template<>
class remote<rmt::Sycl> {
private:
	SAW_FORBID_COPY(remote);
	SAW_FORBID_MOVE(remote);

public:
	/**
	 * Default constructor
	 */
	remote(){}

	/**
	 * For now we don't need to specify the location since
	 * we just create a default.
	 */
	conveyor<own<remote_address<rmt::Sycl>>> resolve_address(){
		return heap<remote_address<rmt::Sycl>>(*this);
	}

	/**
	 * Spin up a rpc server
	 */
	template<typename Iface, typename Encoding>
	rpc_server<Iface, Encoding, rmt::Sycl> listen(const remote_address<rmt::Sycl>&, typename rpc_server<Iface, Encoding, rmt::Sycl>::InterfaceT iface){
		using RpcServerT = rpc_server<Iface, Encoding, rmt::Sycl>;
		using InterfaceT = typename RpcServerT::InterfaceT;
		return {std::move(iface)};
	}
};

}