/*************************************************************************/ /* file_access_network.cpp */ /*************************************************************************/ /* This file is part of: */ /* GODOT ENGINE */ /* http://www.godotengine.org */ /*************************************************************************/ /* Copyright (c) 2007-2017 Juan Linietsky, Ariel Manzur. */ /* */ /* Permission is hereby granted, free of charge, to any person obtaining */ /* a copy of this software and associated documentation files (the */ /* "Software"), to deal in the Software without restriction, including */ /* without limitation the rights to use, copy, modify, merge, publish, */ /* distribute, sublicense, and/or sell copies of the Software, and to */ /* permit persons to whom the Software is furnished to do so, subject to */ /* the following conditions: */ /* */ /* The above copyright notice and this permission notice shall be */ /* included in all copies or substantial portions of the Software. */ /* */ /* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, */ /* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF */ /* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.*/ /* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY */ /* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, */ /* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE */ /* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ /*************************************************************************/ #include "file_access_network.h" #include "marshalls.h" #include "globals.h" #include "os/os.h" #include "io/ip.h" //#define DEBUG_PRINT(m_p) print_line(m_p) //#define DEBUG_TIME(m_what) printf("MS: %s - %lli\n",m_what,OS::get_singleton()->get_ticks_usec()); #define DEBUG_PRINT(m_p) #define DEBUG_TIME(m_what) void FileAccessNetworkClient::lock_mutex() { mutex->lock(); lockcount++; } void FileAccessNetworkClient::unlock_mutex() { lockcount--; mutex->unlock(); } void FileAccessNetworkClient::put_32(int p_32) { uint8_t buf[4]; encode_uint32(p_32,buf); client->put_data(buf,4); DEBUG_PRINT("put32: "+itos(p_32)); } void FileAccessNetworkClient::put_64(int64_t p_64) { uint8_t buf[8]; encode_uint64(p_64,buf); client->put_data(buf,8); DEBUG_PRINT("put64: "+itos(p_64)); } int FileAccessNetworkClient::get_32() { uint8_t buf[4]; client->get_data(buf,4); return decode_uint32(buf); } int64_t FileAccessNetworkClient::get_64() { uint8_t buf[8]; client->get_data(buf,8); return decode_uint64(buf); } void FileAccessNetworkClient::_thread_func() { client->set_nodelay(true); while(!quit) { DEBUG_PRINT("SEM WAIT - "+itos(sem->get())); Error err = sem->wait(); DEBUG_TIME("sem_unlock"); //DEBUG_PRINT("semwait returned "+itos(werr)); DEBUG_PRINT("MUTEX LOCK "+itos(lockcount)); DEBUG_PRINT("POPO"); DEBUG_PRINT("PEPE"); lock_mutex(); DEBUG_PRINT("MUTEX PASS"); blockrequest_mutex->lock(); while(block_requests.size()) { put_32(block_requests.front()->get().id); put_32(FileAccessNetwork::COMMAND_READ_BLOCK); put_64(block_requests.front()->get().offset); put_32(block_requests.front()->get().size); block_requests.pop_front(); } blockrequest_mutex->unlock(); DEBUG_PRINT("THREAD ITER"); DEBUG_TIME("sem_read"); int id = get_32(); int response = get_32(); DEBUG_PRINT("GET RESPONSE: "+itos(response)); FileAccessNetwork *fa=NULL; if (response!=FileAccessNetwork::RESPONSE_DATA) { ERR_FAIL_COND(!accesses.has(id)); } if (accesses.has(id)) fa=accesses[id]; switch(response) { case FileAccessNetwork::RESPONSE_OPEN: { DEBUG_TIME("sem_open"); int status = get_32(); if (status!=OK) { fa->_respond(0,Error(status)); } else { uint64_t len = get_64(); fa->_respond(len,Error(status)); } fa->sem->post(); } break; case FileAccessNetwork::RESPONSE_DATA: { int64_t offset = get_64(); uint32_t len = get_32(); Vector block; block.resize(len); client->get_data(block.ptr(),len); if (fa) //may have been queued fa->_set_block(offset,block); } break; case FileAccessNetwork::RESPONSE_FILE_EXISTS: { int status = get_32(); fa->exists_modtime=status!=0; fa->sem->post(); } break; case FileAccessNetwork::RESPONSE_GET_MODTIME: { uint64_t status = get_64(); fa->exists_modtime=status; fa->sem->post(); } break; } unlock_mutex(); } } void FileAccessNetworkClient::_thread_func(void *s) { FileAccessNetworkClient *self =(FileAccessNetworkClient*)s; self->_thread_func(); } Error FileAccessNetworkClient::connect(const String& p_host,int p_port,const String& p_password) { IP_Address ip; if (p_host.is_valid_ip_address()) { ip=p_host; } else { ip=IP::get_singleton()->resolve_hostname(p_host); } DEBUG_PRINT("IP: "+String(ip)+" port "+itos(p_port)); Error err = client->connect_to_host(ip,p_port); ERR_FAIL_COND_V(err,err); while(client->get_status()==StreamPeerTCP::STATUS_CONNECTING) { //DEBUG_PRINT("trying to connect...."); OS::get_singleton()->delay_usec(1000); } if (client->get_status()!=StreamPeerTCP::STATUS_CONNECTED) { return ERR_CANT_CONNECT; } CharString cs = p_password.utf8(); put_32(cs.length()); client->put_data((const uint8_t*)cs.ptr(),cs.length()); int e = get_32(); if (e!=OK) { return ERR_INVALID_PARAMETER; } thread = Thread::create(_thread_func,this); return OK; } FileAccessNetworkClient *FileAccessNetworkClient::singleton=NULL; FileAccessNetworkClient::FileAccessNetworkClient() { thread=NULL; mutex = Mutex::create(); blockrequest_mutex = Mutex::create(); quit=false; singleton=this; last_id=0; client = Ref( StreamPeerTCP::create_ref() ); sem=Semaphore::create(); lockcount=0; } FileAccessNetworkClient::~FileAccessNetworkClient() { if (thread) { quit=true; sem->post(); Thread::wait_to_finish(thread); memdelete(thread); } memdelete(blockrequest_mutex); memdelete(mutex); memdelete(sem); } void FileAccessNetwork::_set_block(size_t p_offset,const Vector& p_block) { int page = p_offset/page_size; ERR_FAIL_INDEX(page,pages.size()); if (pagelock(); pages[page].buffer=p_block; pages[page].queued=false; buffer_mutex->unlock(); if (waiting_on_page==page) { waiting_on_page=-1; page_sem->post(); } } void FileAccessNetwork::_respond(size_t p_len,Error p_status) { DEBUG_PRINT("GOT RESPONSE - len: "+itos(p_len)+" status: "+itos(p_status)); response=p_status; if (response!=OK) return; opened=true; total_size=p_len; int pc = ((total_size-1)/page_size)+1; pages.resize(pc); } Error FileAccessNetwork::_open(const String& p_path, int p_mode_flags) { ERR_FAIL_COND_V(p_mode_flags!=READ,ERR_UNAVAILABLE); if (opened) close(); FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; DEBUG_PRINT("open: "+p_path); DEBUG_TIME("open_begin"); nc->lock_mutex(); nc->put_32(id); nc->accesses[id]=this; nc->put_32(COMMAND_OPEN_FILE); CharString cs =p_path.utf8(); nc->put_32(cs.length()); nc->client->put_data((const uint8_t*)cs.ptr(),cs.length()); pos=0; eof_flag=false; last_page=-1; last_page_buff=NULL; //buffers.clear(); nc->unlock_mutex(); DEBUG_PRINT("OPEN POST"); DEBUG_TIME("open_post"); nc->sem->post(); //awaiting answer DEBUG_PRINT("WAIT..."); sem->wait(); DEBUG_TIME("open_end"); DEBUG_PRINT("WAIT ENDED..."); return response; } void FileAccessNetwork::close(){ if (!opened) return; FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; DEBUG_PRINT("CLOSE"); nc->lock_mutex(); nc->put_32(id); nc->put_32(COMMAND_CLOSE); pages.clear(); opened=false; nc->unlock_mutex(); } bool FileAccessNetwork::is_open() const{ return opened; } void FileAccessNetwork::seek(size_t p_position){ ERR_FAIL_COND(!opened); eof_flag=p_position>total_size; if (p_position>=total_size) { p_position=total_size; } pos=p_position; } void FileAccessNetwork::seek_end(int64_t p_position){ seek(total_size+p_position); } size_t FileAccessNetwork::get_pos() const{ ERR_FAIL_COND_V(!opened,0); return pos; } size_t FileAccessNetwork::get_len() const{ ERR_FAIL_COND_V(!opened,0); return total_size; } bool FileAccessNetwork::eof_reached() const{ ERR_FAIL_COND_V(!opened,false); return eof_flag; } uint8_t FileAccessNetwork::get_8() const{ uint8_t v; get_buffer(&v,1); return v; } void FileAccessNetwork::_queue_page(int p_page) const { if (p_page>=pages.size()) return; if (pages[p_page].buffer.empty() && !pages[p_page].queued) { FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; nc->blockrequest_mutex->lock(); FileAccessNetworkClient::BlockRequest br; br.id=id; br.offset=size_t(p_page)*page_size; br.size=page_size; nc->block_requests.push_back(br); pages[p_page].queued=true; nc->blockrequest_mutex->unlock(); DEBUG_PRINT("QUEUE PAGE POST"); nc->sem->post(); DEBUG_PRINT("queued "+itos(p_page)); } } int FileAccessNetwork::get_buffer(uint8_t *p_dst, int p_length) const{ //bool eof=false; if (pos+p_length>total_size) { eof_flag=true; } if (pos+p_length>=total_size) { p_length=total_size-pos; } //FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; uint8_t *buff=last_page_buff; for(int i=0;ilock(); if (pages[page].buffer.empty()) { //fuck waiting_on_page=page; for(int j=0;junlock(); DEBUG_PRINT("wait"); page_sem->wait(); DEBUG_PRINT("done"); } else { for(int j=0;junlock(); } buff=pages[page].buffer.ptr(); last_page_buff=buff; last_page=page; } p_dst[i]=buff[pos-uint64_t(page)*page_size]; pos++; } return p_length; } Error FileAccessNetwork::get_error() const{ return pos==total_size?ERR_FILE_EOF:OK; } void FileAccessNetwork::store_8(uint8_t p_dest) { ERR_FAIL(); } bool FileAccessNetwork::file_exists(const String& p_path){ FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; nc->lock_mutex(); nc->put_32(id); nc->put_32(COMMAND_FILE_EXISTS); CharString cs=p_path.utf8(); nc->put_32(cs.length()); nc->client->put_data((const uint8_t*)cs.ptr(),cs.length()); nc->unlock_mutex(); DEBUG_PRINT("FILE EXISTS POST"); nc->sem->post(); sem->wait(); return exists_modtime!=0; } uint64_t FileAccessNetwork::_get_modified_time(const String& p_file){ FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; nc->lock_mutex(); nc->put_32(id); nc->put_32(COMMAND_GET_MODTIME); CharString cs=p_file.utf8(); nc->put_32(cs.length()); nc->client->put_data((const uint8_t*)cs.ptr(),cs.length()); nc->unlock_mutex(); DEBUG_PRINT("MODTIME POST"); nc->sem->post(); sem->wait(); return exists_modtime; } void FileAccessNetwork::configure() { GLOBAL_DEF("network/remote_fs/page_size",65536); GLOBAL_DEF("network/remote_fs/page_read_ahead",4); GLOBAL_DEF("network/remote_fs/max_pages",20); } FileAccessNetwork::FileAccessNetwork() { eof_flag=false; opened=false; pos=0; sem=Semaphore::create(); page_sem=Semaphore::create(); buffer_mutex=Mutex::create(); FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; nc->lock_mutex(); id=nc->last_id++; nc->accesses[id]=this; nc->unlock_mutex(); page_size = GLOBAL_GET("network/remote_fs/page_size"); read_ahead = GLOBAL_GET("network/remote_fs/page_read_ahead"); max_pages = GLOBAL_GET("network/remote_fs/max_pages"); last_activity_val=0; waiting_on_page=-1; last_page=-1; } FileAccessNetwork::~FileAccessNetwork() { close(); memdelete(sem); memdelete(page_sem); memdelete(buffer_mutex); FileAccessNetworkClient *nc = FileAccessNetworkClient::singleton; nc->lock_mutex(); id=nc->last_id++; nc->accesses.erase(id); nc->unlock_mutex(); }