diff options
author | Brecht Van Lommel <brechtvanlommel@pandora.be> | 2012-12-21 15:13:46 +0400 |
---|---|---|
committer | Brecht Van Lommel <brechtvanlommel@pandora.be> | 2012-12-21 15:13:46 +0400 |
commit | e5b457dbc9f009740fe3cca2c8cb44fa88f1f03a (patch) | |
tree | 63e1eeba718717d0c18db9f2a0542639c8f4e2ec /intern/cycles/device/device_network.cpp | |
parent | 5f4c7e5da48c488b807635dccee03e1476e2f0bc (diff) |
Cycles: merge some changes from a local branch to bring network rendering a bit
more up to date, still nowhere near working though, but might as well commit this
in case someone else is interested in working on it.
Diffstat (limited to 'intern/cycles/device/device_network.cpp')
-rw-r--r-- | intern/cycles/device/device_network.cpp | 544 |
1 files changed, 356 insertions, 188 deletions
diff --git a/intern/cycles/device/device_network.cpp b/intern/cycles/device/device_network.cpp index a5e0d39df73..f5545c3dfdf 100644 --- a/intern/cycles/device/device_network.cpp +++ b/intern/cycles/device/device_network.cpp @@ -31,6 +31,8 @@ class NetworkDevice : public Device public: boost::asio::io_service io_service; tcp::socket socket; + device_ptr mem_counter; + DeviceTask the_task; /* todo: handle multiple tasks */ NetworkDevice(Stats &stats, const char *address) : Device(stats), socket(io_service) @@ -49,75 +51,72 @@ public: socket.close(); socket.connect(*endpoint_iterator++, error); } + if(error) throw boost::system::system_error(error); + + mem_counter = 0; } ~NetworkDevice() { + RPCSend snd(socket, "stop"); + snd.write(); } void mem_alloc(device_memory& mem, MemoryType type) { -#if 0 + mem.device_pointer = ++mem_counter; + RPCSend snd(socket, "mem_alloc"); - snd.archive & size & type; + snd.add(mem); + snd.add(type); snd.write(); - - RPCReceive rcv(socket); - - device_ptr mem; - *rcv.archive & mem; - - return mem; -#endif } void mem_copy_to(device_memory& mem) { -#if 0 RPCSend snd(socket, "mem_copy_to"); - snd.archive & mem & size; + snd.add(mem); snd.write(); - snd.write_buffer(host, size); -#endif + snd.write_buffer((void*)mem.data_pointer, mem.memory_size()); } void mem_copy_from(device_memory& mem, int y, int w, int h, int elem) { -#if 0 RPCSend snd(socket, "mem_copy_from"); - snd.archive & mem & offset & size; + snd.add(mem); + snd.add(y); + snd.add(w); + snd.add(h); + snd.add(elem); snd.write(); RPCReceive rcv(socket); - rcv.read_buffer(host, size); -#endif + rcv.read_buffer((void*)mem.data_pointer, mem.memory_size()); } void mem_zero(device_memory& mem) { -#if 0 RPCSend snd(socket, "mem_zero"); - snd.archive & mem & size; + snd.add(mem); snd.write(); -#endif } void mem_free(device_memory& mem) { -#if 0 - if(mem) { + if(mem.device_pointer) { RPCSend snd(socket, "mem_free"); - snd.archive & mem; + snd.add(mem); snd.write(); + + mem.device_pointer = 0; } -#endif } void const_copy_to(const char *name, void *host, size_t size) @@ -126,79 +125,107 @@ public: string name_string(name); - snd.archive & name_string & size; + snd.add(name_string); + snd.add(size); snd.write(); snd.write_buffer(host, size); } void tex_alloc(const char *name, device_memory& mem, bool interpolation, bool periodic) { -#if 0 + mem.device_pointer = ++mem_counter; + RPCSend snd(socket, "tex_alloc"); string name_string(name); - snd.archive & name_string & width & height & datatype & components & interpolation; + snd.add(name_string); + snd.add(mem); + snd.add(interpolation); + snd.add(periodic); snd.write(); - - size_t size = width*height*components*datatype_size(datatype); - snd.write_buffer(host, size); - - RPCReceive rcv(socket); - - device_ptr mem; - *rcv.archive & mem; - - return mem; -#endif + snd.write_buffer((void*)mem.data_pointer, mem.memory_size()); } void tex_free(device_memory& mem) { -#if 0 - if(mem) { + if(mem.device_pointer) { RPCSend snd(socket, "tex_free"); - snd.archive & mem; + snd.add(mem); snd.write(); + + mem.device_pointer = 0; } -#endif } - void path_trace(int x, int y, int w, int h, device_ptr buffer, device_ptr rng_state, int sample) + void task_add(DeviceTask& task) { -#if 0 - RPCSend snd(socket, "path_trace"); + the_task = task; - snd.archive & x & y & w & h & buffer & rng_state & sample; + RPCSend snd(socket, "task_add"); + snd.add(task); snd.write(); -#endif } - void tonemap(int x, int y, int w, int h, device_ptr rgba, device_ptr buffer, int sample, int resolution) + void task_wait() { -#if 0 - RPCSend snd(socket, "tonemap"); - - snd.archive & x & y & w & h & rgba & buffer & sample & resolution; + RPCSend snd(socket, "task_wait"); snd.write(); -#endif - } - void task_add(DeviceTask& task) - { - if(task.type == DeviceTask::TONEMAP) - tonemap(task.x, task.y, task.w, task.h, task.rgba, task.buffer, task.sample, task.resolution); - else if(task.type == DeviceTask::PATH_TRACE) - path_trace(task.x, task.y, task.w, task.h, task.buffer, task.rng_state, task.sample); + list<RenderTile> the_tiles; + + /* todo: run this threaded for connecting to multiple clients */ + for(;;) { + RPCReceive rcv(socket); + RenderTile tile; + + if(rcv.name == "acquire_tile") { + /* todo: watch out for recursive calls! */ + if(the_task.acquire_tile(this, tile)) { /* write return as bool */ + the_tiles.push_back(tile); + + RPCSend snd(socket, "acquire_tile"); + snd.add(tile); + snd.write(); + } + else { + RPCSend snd(socket, "acquire_tile_none"); + snd.write(); + } + } + else if(rcv.name == "release_tile") { + rcv.read(tile); + + for(list<RenderTile>::iterator it = the_tiles.begin(); it != the_tiles.end(); it++) { + if(tile.x == it->x && tile.y == it->y && tile.start_sample == it->start_sample) { + tile.buffers = it->buffers; + the_tiles.erase(it); + break; + } + } + + assert(tile.buffers != NULL); + + the_task.release_tile(tile); + + RPCSend snd(socket, "release_tile"); + snd.write(); + } + else if(rcv.name == "task_wait_done") + break; + } } - void task_wait() + void task_cancel() { + RPCSend snd(socket, "task_cancel"); + snd.write(); } - void task_cancel() + bool support_advanced_shading() { + return true; /* todo: get this info from device */ } }; @@ -219,162 +246,303 @@ void device_network_info(vector<DeviceInfo>& devices) devices.push_back(info); } -void Device::server_run() -{ - try +class DeviceServer { +public: + DeviceServer(Device *device_, tcp::socket& socket_) + : device(device_), socket(socket_) { - /* starts thread that responds to discovery requests */ - ServerDiscovery discovery; + } - for(;;) - { + void listen() + { + /* receive remote function calls */ + for(;;) { + RPCReceive rcv(socket); - /* accept connection */ - boost::asio::io_service io_service; - tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), SERVER_PORT)); + if(rcv.name == "stop") + break; - tcp::socket socket(io_service); - acceptor.accept(socket); + process(rcv); + } + } - /* receive remote function calls */ - for(;;) { - RPCReceive rcv(socket); +protected: + void process(RPCReceive& rcv) + { + // fprintf(stderr, "receive process %s\n", rcv.name.c_str()); - if(rcv.name == "description") { - string desc = description(); + if(rcv.name == "mem_alloc") { + MemoryType type; + network_device_memory mem; + device_ptr remote_pointer; - RPCSend snd(socket); - snd.archive & desc; - snd.write(); - } - else if(rcv.name == "mem_alloc") { -#if 0 - MemoryType type; - size_t size; - device_ptr mem; + rcv.read(mem); + rcv.read(type); - *rcv.archive & size & type; - mem = mem_alloc(size, type); + /* todo: CPU needs mem.data_pointer */ - RPCSend snd(socket); - snd.archive & mem; - snd.write(); -#endif - } - else if(rcv.name == "mem_copy_to") { -#if 0 - device_ptr mem; - size_t size; + remote_pointer = mem.device_pointer; - *rcv.archive & mem & size; + mem_data[remote_pointer] = vector<uint8_t>(); + mem_data[remote_pointer].resize(mem.memory_size()); + if(mem.memory_size()) + mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]); + else + mem.data_pointer = 0; - vector<char> host_vector(size); - rcv.read_buffer(&host_vector[0], size); + device->mem_alloc(mem, type); - mem_copy_to(mem, &host_vector[0], size); -#endif - } - else if(rcv.name == "mem_copy_from") { -#if 0 - device_ptr mem; - size_t offset, size; + ptr_map[remote_pointer] = mem.device_pointer; + ptr_imap[mem.device_pointer] = remote_pointer; + } + else if(rcv.name == "mem_copy_to") { + network_device_memory mem; - *rcv.archive & mem & offset & size; + rcv.read(mem); - vector<char> host_vector(size); + device_ptr remote_pointer = mem.device_pointer; + mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]); - mem_copy_from(&host_vector[0], mem, offset, size); + rcv.read_buffer((uint8_t*)mem.data_pointer, mem.memory_size()); - RPCSend snd(socket); - snd.write(); - snd.write_buffer(&host_vector[0], size); -#endif - } - else if(rcv.name == "mem_zero") { -#if 0 - device_ptr mem; - size_t size; + mem.device_pointer = ptr_map[remote_pointer]; - *rcv.archive & mem & size; - mem_zero(mem, size); -#endif - } - else if(rcv.name == "mem_free") { -#if 0 - device_ptr mem; + device->mem_copy_to(mem); + } + else if(rcv.name == "mem_copy_from") { + network_device_memory mem; + int y, w, h, elem; - *rcv.archive & mem; - mem_free(mem); -#endif - } - else if(rcv.name == "const_copy_to") { - string name_string; - size_t size; + rcv.read(mem); + rcv.read(y); + rcv.read(w); + rcv.read(h); + rcv.read(elem); - *rcv.archive & name_string & size; + device_ptr remote_pointer = mem.device_pointer; + mem.device_pointer = ptr_map[remote_pointer]; + mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]); - vector<char> host_vector(size); - rcv.read_buffer(&host_vector[0], size); + device->mem_copy_from(mem, y, w, h, elem); - const_copy_to(name_string.c_str(), &host_vector[0], size); - } - else if(rcv.name == "tex_alloc") { -#if 0 - string name_string; - DataType datatype; - device_ptr mem; - size_t width, height; - int components; - bool interpolation; + RPCSend snd(socket); + snd.write(); + snd.write_buffer((uint8_t*)mem.data_pointer, mem.memory_size()); + } + else if(rcv.name == "mem_zero") { + network_device_memory mem; + + rcv.read(mem); + device_ptr remote_pointer = mem.device_pointer; + mem.device_pointer = ptr_map[mem.device_pointer]; + mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]); + + device->mem_zero(mem); + } + else if(rcv.name == "mem_free") { + network_device_memory mem; + device_ptr remote_pointer; - *rcv.archive & name_string & width & height & datatype & components & interpolation; + rcv.read(mem); - size_t size = width*height*components*datatype_size(datatype); + remote_pointer = mem.device_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); - vector<char> host_vector(size); - rcv.read_buffer(&host_vector[0], size); + device->mem_free(mem); + } + else if(rcv.name == "const_copy_to") { + string name_string; + size_t size; - mem = tex_alloc(name_string.c_str(), &host_vector[0], width, height, datatype, components, interpolation); + rcv.read(name_string); + rcv.read(size); - RPCSend snd(socket); - snd.archive & mem; - snd.write(); -#endif - } - else if(rcv.name == "tex_free") { -#if 0 - device_ptr mem; + vector<char> host_vector(size); + rcv.read_buffer(&host_vector[0], size); - *rcv.archive & mem; - tex_free(mem); -#endif - } - else if(rcv.name == "path_trace") { -#if 0 - device_ptr buffer, rng_state; - int x, y, w, h; - int sample; - - *rcv.archive & x & y & w & h & buffer & rng_state & sample; - path_trace(x, y, w, h, buffer, rng_state, sample); -#endif - } - else if(rcv.name == "tonemap") { -#if 0 - device_ptr rgba, buffer; - int x, y, w, h; - int sample, resolution; - - *rcv.archive & x & y & w & h & rgba & buffer & sample & resolution; - tonemap(x, y, w, h, rgba, buffer, sample, resolution); -#endif - } + device->const_copy_to(name_string.c_str(), &host_vector[0], size); + } + else if(rcv.name == "tex_alloc") { + network_device_memory mem; + string name; + bool interpolation; + bool periodic; + device_ptr remote_pointer; + + rcv.read(name); + rcv.read(mem); + rcv.read(interpolation); + rcv.read(periodic); + + remote_pointer = mem.device_pointer; + + mem_data[remote_pointer] = vector<uint8_t>(); + mem_data[remote_pointer].resize(mem.memory_size()); + if(mem.memory_size()) + mem.data_pointer = (device_ptr)&(mem_data[remote_pointer][0]); + else + mem.data_pointer = 0; + + rcv.read_buffer((uint8_t*)mem.data_pointer, mem.memory_size()); + + device->tex_alloc(name.c_str(), mem, interpolation, periodic); + + ptr_map[remote_pointer] = mem.device_pointer; + ptr_imap[mem.device_pointer] = remote_pointer; + } + else if(rcv.name == "tex_free") { + network_device_memory mem; + device_ptr remote_pointer; + + rcv.read(mem); + + remote_pointer = mem.device_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); + } + else if(rcv.name == "task_add") { + DeviceTask task; + + rcv.read(task); + + if(task.buffer) task.buffer = ptr_map[task.buffer]; + if(task.rgba) task.rgba = ptr_map[task.rgba]; + 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.release_tile = function_bind(&DeviceServer::task_release_tile, this, _1); + task.update_progress_sample = function_bind(&DeviceServer::task_update_progress_sample, this); + task.update_tile_sample = function_bind(&DeviceServer::task_update_tile_sample, this, _1); + task.get_cancel = function_bind(&DeviceServer::task_get_cancel, this); + + device->task_add(task); + } + else if(rcv.name == "task_wait") { + device->task_wait(); + + RPCSend snd(socket, "task_wait_done"); + snd.write(); + } + else if(rcv.name == "task_cancel") { + device->task_cancel(); + } + } + + bool task_acquire_tile(Device *device, RenderTile& tile) + { + thread_scoped_lock acquire_lock(acquire_mutex); + + bool result = false; + + RPCSend snd(socket, "acquire_tile"); + snd.write(); + + while(1) { + RPCReceive rcv(socket); + + if(rcv.name == "acquire_tile") { + rcv.read(tile); + + if(tile.buffer) tile.buffer = ptr_map[tile.buffer]; + if(tile.rng_state) tile.rng_state = ptr_map[tile.rng_state]; + if(tile.rgba) tile.rgba = ptr_map[tile.rgba]; + + result = true; + break; } + else if(rcv.name == "acquire_tile_none") + break; + else + process(rcv); } + + return result; } - catch(exception& e) + + void task_update_progress_sample() { - cerr << "Network server exception: " << e.what() << endl; + ; /* skip */ + } + + void task_update_tile_sample(RenderTile&) + { + ; /* skip */ + } + + void task_release_tile(RenderTile& tile) + { + thread_scoped_lock acquire_lock(acquire_mutex); + + if(tile.buffer) tile.buffer = ptr_imap[tile.buffer]; + if(tile.rng_state) tile.rng_state = ptr_imap[tile.rng_state]; + if(tile.rgba) tile.rgba = ptr_imap[tile.rgba]; + + RPCSend snd(socket, "release_tile"); + snd.add(tile); + snd.write(); + + while(1) { + RPCReceive rcv(socket); + + if(rcv.name == "release_tile") + break; + else + process(rcv); + } + } + + bool task_get_cancel() + { + return false; + } + + /* properties */ + Device *device; + tcp::socket& socket; + + /* mapping of remote to local pointer */ + map<device_ptr, device_ptr> ptr_map; + map<device_ptr, device_ptr> ptr_imap; + map<device_ptr, vector<uint8_t> > mem_data; + + thread_mutex acquire_mutex; + + /* todo: free memory and device (osl) on network error */ +}; + +void Device::server_run() +{ + try { + /* starts thread that responds to discovery requests */ + ServerDiscovery discovery; + + for(;;) { + /* accept connection */ + boost::asio::io_service io_service; + tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), SERVER_PORT)); + + tcp::socket socket(io_service); + acceptor.accept(socket); + + string remote_address = socket.remote_endpoint().address().to_string(); + printf("Connected to remote client at: %s\n", remote_address.c_str()); + + DeviceServer server(this, socket); + server.listen(); + + printf("Disconnected.\n"); + } + } + catch(exception& e) { + fprintf(stderr, "Network server exception: %s\n", e.what()); } } |