Differential D36 Diff 86 intern/cycles/device/device_network.cpp

Changeset View

Standalone View

intern/cycles/device/device_network.cpp

Context not available.

	#include "util_foreach.h"	#include "util_foreach.h"

		#if defined(WITH_NETWORK)

	CCL_NAMESPACE_BEGIN	CCL_NAMESPACE_BEGIN

	#ifdef WITH_NETWORK	typedef map<device_ptr, device_ptr> PtrMap;
		typedef vector<uint8_t> DataVector;
		typedef map<device_ptr, DataVector> DataMap;

		/* tile list */
		typedef vector<RenderTile> TileList;

		/* search a list of tiles and find the one that matches the passed render tile */
		static TileList::iterator tile_list_find(TileList& tile_list, RenderTile& tile)
		{
		for(TileList::iterator it = tile_list.begin(); it != tile_list.end(); ++it)
		if(tile.x == it->x && tile.y == it->y && tile.start_sample == it->start_sample)
		return it;
		return tile_list.end();
		}

	class NetworkDevice : public Device	class NetworkDevice : public Device
	{	{
Context not available.
	device_ptr mem_counter;	device_ptr mem_counter;
	DeviceTask the_task; /* todo: handle multiple tasks */	DeviceTask the_task; /* todo: handle multiple tasks */

	NetworkDevice(Stats &stats, const char *address)	thread_mutex rpc_lock;
	: Device(stats), socket(io_service)
		NetworkDevice(DeviceInfo& info, Stats &stats, const char *address)
		: Device(info, stats, true)
		, socket(io_service)
	{	{
	stringstream portstr;	stringstream portstr;
	portstr << SERVER_PORT;	portstr << SERVER_PORT;
Context not available.

	void mem_alloc(device_memory& mem, MemoryType type)	void mem_alloc(device_memory& mem, MemoryType type)
	{	{
		thread_scoped_lock lock(rpc_lock);

	mem.device_pointer = ++mem_counter;	mem.device_pointer = ++mem_counter;

	RPCSend snd(socket, "mem_alloc");	RPCSend snd(socket, "mem_alloc");
Context not available.

	void mem_copy_to(device_memory& mem)	void mem_copy_to(device_memory& mem)
	{	{
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "mem_copy_to");	RPCSend snd(socket, "mem_copy_to");

	snd.add(mem);	snd.add(mem);
Context not available.

	void mem_copy_from(device_memory& mem, int y, int w, int h, int elem)	void mem_copy_from(device_memory& mem, int y, int w, int h, int elem)
	{	{
		thread_scoped_lock lock(rpc_lock);

		size_t data_size = mem.memory_size();

	RPCSend snd(socket, "mem_copy_from");	RPCSend snd(socket, "mem_copy_from");

	snd.add(mem);	snd.add(mem);
Context not available.
	snd.write();	snd.write();

	RPCReceive rcv(socket);	RPCReceive rcv(socket);
	rcv.read_buffer((void*)mem.data_pointer, mem.memory_size());	rcv.read_buffer((void*)mem.data_pointer, data_size);
	}	}

	void mem_zero(device_memory& mem)	void mem_zero(device_memory& mem)
	{	{
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "mem_zero");	RPCSend snd(socket, "mem_zero");

	snd.add(mem);	snd.add(mem);
Context not available.
	void mem_free(device_memory& mem)	void mem_free(device_memory& mem)
	{	{
	if(mem.device_pointer) {	if(mem.device_pointer) {
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "mem_free");	RPCSend snd(socket, "mem_free");

	snd.add(mem);	snd.add(mem);
Context not available.

	void const_copy_to(const char name, void host, size_t size)	void const_copy_to(const char name, void host, size_t size)
	{	{
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "const_copy_to");	RPCSend snd(socket, "const_copy_to");

	string name_string(name);	string name_string(name);
Context not available.

	void tex_alloc(const char *name, device_memory& mem, bool interpolation, bool periodic)	void tex_alloc(const char *name, device_memory& mem, bool interpolation, bool periodic)
	{	{
		thread_scoped_lock lock(rpc_lock);

	mem.device_pointer = ++mem_counter;	mem.device_pointer = ++mem_counter;

	RPCSend snd(socket, "tex_alloc");	RPCSend snd(socket, "tex_alloc");
Context not available.
	void tex_free(device_memory& mem)	void tex_free(device_memory& mem)
	{	{
	if(mem.device_pointer) {	if(mem.device_pointer) {
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "tex_free");	RPCSend snd(socket, "tex_free");

	snd.add(mem);	snd.add(mem);
Context not available.
	}	}
	}	}

		bool load_kernels(bool experimental)
		{
		thread_scoped_lock lock(rpc_lock);

		RPCSend snd(socket, "load_kernels");
		snd.add(experimental);
		snd.write();

		bool result;
		RPCReceive rcv(socket);
		rcv.read(result);

		return result;
		}

	void task_add(DeviceTask& task)	void task_add(DeviceTask& task)
	{	{
		thread_scoped_lock lock(rpc_lock);

	the_task = task;	the_task = task;

	RPCSend snd(socket, "task_add");	RPCSend snd(socket, "task_add");
Context not available.

	void task_wait()	void task_wait()
	{	{
		thread_scoped_lock lock(rpc_lock);

	RPCSend snd(socket, "task_wait");	RPCSend snd(socket, "task_wait");
	snd.write();	snd.write();

	list<RenderTile> the_tiles;	lock.unlock();

		TileList the_tiles;

	/* todo: run this threaded for connecting to multiple clients */	/* todo: run this threaded for connecting to multiple clients */
	for(;;) {	for(;;) {
	RPCReceive rcv(socket);
	RenderTile tile;	RenderTile tile;

		lock.lock();
		RPCReceive rcv(socket);

	if(rcv.name == "acquire_tile") {	if(rcv.name == "acquire_tile") {
		lock.unlock();

	/* todo: watch out for recursive calls! */	/* todo: watch out for recursive calls! */
	if(the_task.acquire_tile(this, tile)) { /* write return as bool */	if(the_task.acquire_tile(this, tile)) { /* write return as bool */
	the_tiles.push_back(tile);	the_tiles.push_back(tile);

		lock.lock();
	RPCSend snd(socket, "acquire_tile");	RPCSend snd(socket, "acquire_tile");
	snd.add(tile);	snd.add(tile);
	snd.write();	snd.write();
		lock.unlock();
	}	}
	else {	else {
		lock.lock();
	RPCSend snd(socket, "acquire_tile_none");	RPCSend snd(socket, "acquire_tile_none");
	snd.write();	snd.write();
		lock.unlock();
	}	}
	}	}
	else if(rcv.name == "release_tile") {	else if(rcv.name == "release_tile") {
	rcv.read(tile);	rcv.read(tile);
		lock.unlock();

	for(list<RenderTile>::iterator it = the_tiles.begin(); it != the_tiles.end(); it++) {	TileList::iterator it = tile_list_find(the_tiles, tile);
	if(tile.x == it->x && tile.y == it->y && tile.start_sample == it->start_sample) {	if (it != the_tiles.end()) {
	tile.buffers = it->buffers;	tile.buffers = it->buffers;
	the_tiles.erase(it);	the_tiles.erase(it);
	break;
	}
	}	}

	assert(tile.buffers != NULL);	assert(tile.buffers != NULL);

	the_task.release_tile(tile);	the_task.release_tile(tile);

		lock.lock();
	RPCSend snd(socket, "release_tile");	RPCSend snd(socket, "release_tile");
	snd.write();	snd.write();
		lock.unlock();
	}	}
	else if(rcv.name == "task_wait_done")	else if(rcv.name == "task_wait_done")
	break;	break;
		brechtUnsubmitted Not Done Inline Actions This case is missing a lock.unlock(); brecht: This case is missing a lock.unlock();
		else
		lock.unlock();
	}	}
	}	}

	void task_cancel()	void task_cancel()
	{	{
		thread_scoped_lock lock(rpc_lock);
	RPCSend snd(socket, "task_cancel");	RPCSend snd(socket, "task_cancel");
	snd.write();	snd.write();
	}	}
Context not available.

	Device device_network_create(DeviceInfo& info, Stats &stats, const char address)	Device device_network_create(DeviceInfo& info, Stats &stats, const char address)
	{	{
	return new NetworkDevice(stats, address);	return new NetworkDevice(info, stats, address);
	}	}

	void device_network_info(vector<DeviceInfo>& devices)	void device_network_info(vector<DeviceInfo>& devices)
Context not available.

	class DeviceServer {	class DeviceServer {
	public:	public:
		thread_mutex rpc_lock;

	DeviceServer(Device *device_, tcp::socket& socket_)	DeviceServer(Device *device_, tcp::socket& socket_)
	: device(device_), socket(socket_)	: device(device_), socket(socket_)
	{	{
	}	}

Context not available.
	{	{
	/* receive remote function calls */	/* receive remote function calls */
	for(;;) {	for(;;) {
		thread_scoped_lock lock(rpc_lock);
	RPCReceive rcv(socket);	RPCReceive rcv(socket);

	if(rcv.name == "stop")	if(rcv.name == "stop")
	break;	break;

	process(rcv);	process(rcv, lock);
	}	}
	}	}

	protected:	protected:
	void process(RPCReceive& rcv)	/* create a memory buffer for a device buffer and insert it into mem_data */
		DataVector &data_vector_insert(device_ptr client_pointer, size_t data_size)
		{
		/* create a new DataVector and insert it into mem_data */
		pair<DataMap::iterator,bool> data_ins = mem_data.insert(
		DataMap::value_type(client_pointer, DataVector()));

		/* make sure it was a unique insertion */
		assert(data_ins.second);

		/* get a reference to the inserted vector */
		DataVector &data_v = data_ins.first->second;

		/* size the vector */
		data_v.resize(data_size);

		return data_v;
		}

		DataVector &data_vector_find(device_ptr client_pointer)
		{
		DataMap::iterator i = mem_data.find(client_pointer);
		assert(i != mem_data.end());
		return i->second;
		}

		/* setup mapping and reverse mapping of client_pointer<->real_pointer */
		void pointer_mapping_insert(device_ptr client_pointer, device_ptr real_pointer)
		{
		pair<PtrMap::iterator,bool> mapins;

		/* insert mapping from client pointer to our real device pointer */
		mapins = ptr_map.insert(PtrMap::value_type(client_pointer, real_pointer));
		assert(mapins.second);

		/* insert reverse mapping from real our device pointer to client pointer */
		mapins = ptr_imap.insert(PtrMap::value_type(real_pointer, client_pointer));
		assert(mapins.second);
		}

		device_ptr device_ptr_from_client_pointer(device_ptr client_pointer)
		{
		PtrMap::iterator i = ptr_map.find(client_pointer);
		assert(i != ptr_map.end());
		return i->second;
		}

		device_ptr device_ptr_from_client_pointer_erase(device_ptr client_pointer)
	{	{
	// fprintf(stderr, "receive process %s\n", rcv.name.c_str());	PtrMap::iterator i = ptr_map.find(client_pointer);
		assert(i != ptr_map.end());

		device_ptr result = i->second;

		/* erase the mapping */
		ptr_map.erase(i);

		/* erase the reverse mapping */
		PtrMap::iterator irev = ptr_imap.find(result);
		assert(irev != ptr_imap.end());
		ptr_imap.erase(irev);

		/* erase the data vector */
		DataMap::iterator idata = mem_data.find(client_pointer);
		assert(idata != mem_data.end());
		mem_data.erase(idata);

		return result;
		}

		/* note that the lock must be already acquired upon entry.
		* This is necessary because the caller often peeks at
		* the header and delegates control to here when it doesn't
		* specifically handle the current RPC.
		* The lock must be unlocked before returning */
		void process(RPCReceive& rcv, thread_scoped_lock &lock)
		{
		fprintf(stderr, "receive process %s\n", rcv.name.c_str());

	if(rcv.name == "mem_alloc") {	if(rcv.name == "mem_alloc") {
	MemoryType type;	MemoryType type;
	network_device_memory mem;	network_device_memory mem;
	device_ptr remote_pointer;	device_ptr client_pointer;

	rcv.read(mem);	rcv.read(mem);
	rcv.read(type);	rcv.read(type);

	/* todo: CPU needs mem.data_pointer */	lock.unlock();

	remote_pointer = mem.device_pointer;	client_pointer = mem.device_pointer;

	mem_data[remote_pointer] = vector<uint8_t>();	/* create a memory buffer for the device buffer */
	mem_data[remote_pointer].resize(mem.memory_size());	size_t data_size = mem.memory_size();
	if(mem.memory_size())	DataVector &data_v = data_vector_insert(client_pointer, data_size);
	mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]);
		if(data_size)
		mem.data_pointer = (device_ptr)&(data_v[0]);
	else	else
	mem.data_pointer = 0;	mem.data_pointer = 0;

		/* perform the allocation on the actual device */
	device->mem_alloc(mem, type);	device->mem_alloc(mem, type);

	ptr_map[remote_pointer] = mem.device_pointer;	/* store a mapping to/from client_pointer and real device pointer */
	ptr_imap[mem.device_pointer] = remote_pointer;	pointer_mapping_insert(client_pointer, mem.device_pointer);
	}	}
	else if(rcv.name == "mem_copy_to") {	else if(rcv.name == "mem_copy_to") {
	network_device_memory mem;	network_device_memory mem;

	rcv.read(mem);	rcv.read(mem);
		lock.unlock();

		device_ptr client_pointer = mem.device_pointer;

		DataVector &data_v = data_vector_find(client_pointer);

	device_ptr remote_pointer = mem.device_pointer;	size_t data_size = mem.memory_size();
	mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]);

	rcv.read_buffer((uint8_t*)mem.data_pointer, mem.memory_size());	/* get pointer to memory buffer for device buffer */
		mem.data_pointer = (device_ptr)&data_v[0];

	mem.device_pointer = ptr_map[remote_pointer];	/* copy data from network into memory buffer */
		rcv.read_buffer((uint8_t*)mem.data_pointer, data_size);

		/* translate the client pointer to a real device pointer */
		mem.device_pointer = device_ptr_from_client_pointer(client_pointer);

		/* copy the data from the memory buffer to the device buffer */
	device->mem_copy_to(mem);	device->mem_copy_to(mem);
	}	}
	else if(rcv.name == "mem_copy_from") {	else if(rcv.name == "mem_copy_from") {
Context not available.
	rcv.read(h);	rcv.read(h);
	rcv.read(elem);	rcv.read(elem);

	device_ptr remote_pointer = mem.device_pointer;	device_ptr client_pointer = mem.device_pointer;
	mem.device_pointer = ptr_map[remote_pointer];	mem.device_pointer = device_ptr_from_client_pointer(client_pointer);
	mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]);
		DataVector &data_v = data_vector_find(client_pointer);

		mem.data_pointer = (device_ptr)&(data_v[0]);

	device->mem_copy_from(mem, y, w, h, elem);	device->mem_copy_from(mem, y, w, h, elem);

		size_t data_size = mem.memory_size();

	RPCSend snd(socket);	RPCSend snd(socket);
	snd.write();	snd.write();
	snd.write_buffer((uint8_t*)mem.data_pointer, mem.memory_size());	snd.write_buffer((uint8_t*)mem.data_pointer, data_size);
		lock.unlock();
	}	}
	else if(rcv.name == "mem_zero") {	else if(rcv.name == "mem_zero") {
	network_device_memory mem;	network_device_memory mem;

	rcv.read(mem);	rcv.read(mem);
	device_ptr remote_pointer = mem.device_pointer;	lock.unlock();
	mem.device_pointer = ptr_map[mem.device_pointer];
	mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]);	device_ptr client_pointer = mem.device_pointer;
		mem.device_pointer = device_ptr_from_client_pointer(client_pointer);

		DataVector &data_v = data_vector_find(client_pointer);

		mem.data_pointer = (device_ptr)&(data_v[0]);

	device->mem_zero(mem);	device->mem_zero(mem);
	}	}
	else if(rcv.name == "mem_free") {	else if(rcv.name == "mem_free") {
	network_device_memory mem;	network_device_memory mem;
	device_ptr remote_pointer;	device_ptr client_pointer;

	rcv.read(mem);	rcv.read(mem);
		lock.unlock();

		client_pointer = mem.device_pointer;

	remote_pointer = mem.device_pointer;	mem.device_pointer = device_ptr_from_client_pointer_erase(client_pointer);
	mem.device_pointer = ptr_map[mem.device_pointer];
	ptr_map.erase(remote_pointer);
	ptr_imap.erase(mem.device_pointer);
	mem_data.erase(remote_pointer);

	device->mem_free(mem);	device->mem_free(mem);
	}	}
Context not available.

	vector<char> host_vector(size);	vector<char> host_vector(size);
	rcv.read_buffer(&host_vector[0], size);	rcv.read_buffer(&host_vector[0], size);
		lock.unlock();

	device->const_copy_to(name_string.c_str(), &host_vector[0], size);	device->const_copy_to(name_string.c_str(), &host_vector[0], size);
	}	}
Context not available.
	string name;	string name;
	bool interpolation;	bool interpolation;
	bool periodic;	bool periodic;
	device_ptr remote_pointer;	device_ptr client_pointer;

	rcv.read(name);	rcv.read(name);
	rcv.read(mem);	rcv.read(mem);
	rcv.read(interpolation);	rcv.read(interpolation);
	rcv.read(periodic);	rcv.read(periodic);
		lock.unlock();

	remote_pointer = mem.device_pointer;	client_pointer = mem.device_pointer;

	mem_data[remote_pointer] = vector<uint8_t>();	size_t data_size = mem.memory_size();
	mem_data[remote_pointer].resize(mem.memory_size());
	if(mem.memory_size())	DataVector &data_v = data_vector_insert(client_pointer, data_size);
	mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]);
		if(data_size)
		mem.data_pointer = (device_ptr)&(data_v[0]);
	else	else
	mem.data_pointer = 0;	mem.data_pointer = 0;

	rcv.read_buffer((uint8_t*)mem.data_pointer, mem.memory_size());	rcv.read_buffer((uint8_t*)mem.data_pointer, data_size);

	device->tex_alloc(name.c_str(), mem, interpolation, periodic);	device->tex_alloc(name.c_str(), mem, interpolation, periodic);

	ptr_map[remote_pointer] = mem.device_pointer;	pointer_mapping_insert(client_pointer, mem.device_pointer);
	ptr_imap[mem.device_pointer] = remote_pointer;
	}	}
	else if(rcv.name == "tex_free") {	else if(rcv.name == "tex_free") {
	network_device_memory mem;	network_device_memory mem;
	device_ptr remote_pointer;	device_ptr client_pointer;

	rcv.read(mem);	rcv.read(mem);
		lock.unlock();

		client_pointer = mem.device_pointer;

	remote_pointer = mem.device_pointer;	mem.device_pointer = device_ptr_from_client_pointer_erase(client_pointer);
	mem.device_pointer = ptr_map[mem.device_pointer];
	ptr_map.erase(remote_pointer);
	ptr_map.erase(mem.device_pointer);
	mem_data.erase(remote_pointer);

	device->tex_free(mem);	device->tex_free(mem);
	}	}
		else if(rcv.name == "load_kernels") {
		bool experimental;
		rcv.read(experimental);

		bool result;
		result = device->load_kernels(experimental);
		RPCSend snd(socket);
		snd.add(result);
		snd.write();
		lock.unlock();
		}
	else if(rcv.name == "task_add") {	else if(rcv.name == "task_add") {
	DeviceTask task;	DeviceTask task;

	rcv.read(task);	rcv.read(task);
		lock.unlock();

		if(task.buffer)
		task.buffer = device_ptr_from_client_pointer(task.buffer);

		if(task.rgba_half)
		task.rgba_half = device_ptr_from_client_pointer(task.rgba_half);

		if(task.rgba_byte)
		task.rgba_byte = device_ptr_from_client_pointer(task.rgba_byte);

		if(task.shader_input)
		task.shader_input = device_ptr_from_client_pointer(task.shader_input);

		if(task.shader_output)
		task.shader_output = device_ptr_from_client_pointer(task.shader_output);

	if(task.buffer) task.buffer = ptr_map[task.buffer];
	if(task.rgba_byte) task.rgba_byte = ptr_map[task.rgba_byte];
	if(task.rgba_half) task.rgba_half = ptr_map[task.rgba_half];
	if(task.shader_input) task.shader_input = ptr_map[task.shader_input];
	if(task.shader_output) task.shader_output = ptr_map[task.shader_output];

	task.acquire_tile = function_bind(&DeviceServer::task_acquire_tile, this, _1, _2);	task.acquire_tile = function_bind(&DeviceServer::task_acquire_tile, this, _1, _2);
	task.release_tile = function_bind(&DeviceServer::task_release_tile, this, _1);	task.release_tile = function_bind(&DeviceServer::task_release_tile, this, _1);
Context not available.
	device->task_add(task);	device->task_add(task);
	}	}
	else if(rcv.name == "task_wait") {	else if(rcv.name == "task_wait") {
		lock.unlock();
	device->task_wait();	device->task_wait();

		lock.lock();
	RPCSend snd(socket, "task_wait_done");	RPCSend snd(socket, "task_wait_done");
	snd.write();	snd.write();
		lock.unlock();
	}	}
	else if(rcv.name == "task_cancel") {	else if(rcv.name == "task_cancel") {
		lock.unlock();
	device->task_cancel();	device->task_cancel();
	}	}
		else if(rcv.name == "acquire_tile") {
		RenderTile tile;
		rcv.read(tile);
		lock.unlock();
		//
		}
		else
		{
		raise(SIGTRAP);
		brechtUnsubmitted Not Done Inline Actions This case i missing lock.unlock(); brecht: This case i missing lock.unlock();
		}
	}	}

	bool task_acquire_tile(Device *device, RenderTile& tile)	bool task_acquire_tile(Device *device, RenderTile& tile)
Context not available.
	snd.write();	snd.write();

	while(1) {	while(1) {
		thread_scoped_lock lock(rpc_lock);
	RPCReceive rcv(socket);	RPCReceive rcv(socket);

	if(rcv.name == "acquire_tile") {	if(rcv.name == "acquire_tile") {
Context not available.
	else if(rcv.name == "acquire_tile_none")	else if(rcv.name == "acquire_tile_none")
	break;	break;
	else	else
	process(rcv);	process(rcv, lock);
	}	}

	return result;	return result;
Context not available.
	if(tile.buffer) tile.buffer = ptr_imap[tile.buffer];	if(tile.buffer) tile.buffer = ptr_imap[tile.buffer];
	if(tile.rng_state) tile.rng_state = ptr_imap[tile.rng_state];	if(tile.rng_state) tile.rng_state = ptr_imap[tile.rng_state];

		thread_scoped_lock lock(rpc_lock);
	RPCSend snd(socket, "release_tile");	RPCSend snd(socket, "release_tile");
	snd.add(tile);	snd.add(tile);
	snd.write();	snd.write();
		lock.unlock();

	while(1) {	while(1) {
		lock.lock();
	RPCReceive rcv(socket);	RPCReceive rcv(socket);

	if(rcv.name == "release_tile")	if(rcv.name == "release_tile")
	break;	break;
		brechtUnsubmitted Not Done Inline Actions This case is missing a lock.unlock(); I know the lock going out of scope probably takes care of it, but that kind of thing can get missed when adding extra code to this function. brecht: This case is missing a lock.unlock(); I know the lock going out of scope probably takes care…
	else	else
	process(rcv);	process(rcv, lock);
	}	}
	}	}

Context not available.
	tcp::socket& socket;	tcp::socket& socket;

	/* mapping of remote to local pointer */	/* mapping of remote to local pointer */
	map<device_ptr, device_ptr> ptr_map;	PtrMap ptr_map;
	map<device_ptr, device_ptr> ptr_imap;	PtrMap ptr_imap;
	map<device_ptr, vector<uint8_t> > mem_data;	DataMap mem_data;

	thread_mutex acquire_mutex;	thread_mutex acquire_mutex;

Context not available.
	}	}
	}	}

		CCL_NAMESPACE_END

	#endif	#endif

	CCL_NAMESPACE_END

Context not available.