- implemented a visualisation for currently handled chunks, availability maps, and transfer info

- implemented transfer protocol for chunk availability maps between peers (not enabled yet though)
- suppressed rsiface directory from retroshare-gui
- moved notifyqt.{h,cpp} to gui/

next moves: 
	- send availability maps to clients; 
	- connect turtle search to unfinished files;
	- test multisource download with unfinished files.



git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@1939 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
csoler 2009-12-28 21:11:00 +00:00
parent 1a86871556
commit f4a2eaecce
19 changed files with 622 additions and 747 deletions

View file

@ -212,16 +212,28 @@ bool ChunkMap::getDataChunk(const std::string& peer_id,uint32_t size_hint,ftChun
return true ;
}
#ifdef A_FAIRE
void setPeerAvailabilityMap(const std::string& peer_id,const std::vector<uint32_t>& peer_map)
void ChunkMap::setPeerAvailabilityMap(const std::string& peer_id,uint32_t chunk_size,uint32_t nb_chunks,const std::vector<uint32_t>& compressed_peer_map)
{
#ifdef DEBUG_FTCHUNK
std::cout << "Receiving new availability map for peer " << peer_id << std::endl ;
std::cout << "ChunkMap::Receiving new availability map for peer " << peer_id << std::endl ;
#endif
// Check that the parameters are the same. Otherwise we should convert the info into the local format.
// If all RS instances have the same policy for deciding the sizes of chunks, this should not happen.
_peers_chunks_availability[peer_id] = peer_map ;
}
if(chunk_size != _chunk_size || nb_chunks != _map.size())
{
std::cerr << "ChunkMap::setPeerAvailabilityMap: chunk size / number of chunks is not correct. Dropping the info." << std::endl ;
return ;
}
// sets the map.
//
_peers_chunks_availability[peer_id] = compressed_peer_map ;
#ifdef DEBUG_FTCHUNK
std::cerr << "ChunkMap::setPeerAvailabilityMap: Setting chunk availability info for peer " << peer_id << std::endl ;
#endif
}
uint32_t ChunkMap::sizeOfChunk(uint32_t cid) const
{
@ -231,11 +243,10 @@ uint32_t ChunkMap::sizeOfChunk(uint32_t cid) const
return _chunk_size ;
}
uint32_t ChunkMap::getAvailableChunk(uint32_t start_location,const std::string& peer_id) const
uint32_t ChunkMap::getAvailableChunk(uint32_t start_location,const std::string& peer_id)
{
// Very bold algorithm: checks for 1st availabe chunk for this peer starting
// from the given start location.
#ifdef A_FAIRE
std::map<std::string,std::vector<uint32_t> >::const_iterator it(_peers_chunks_availability.find(peer_id)) ;
// Do we have records for this file source ?
@ -243,18 +254,21 @@ uint32_t ChunkMap::getAvailableChunk(uint32_t start_location,const std::string&
if(it == _peers_chunks_availability.end())
{
#ifdef DEBUG_FTCHUNK
std::cout << "No chunk map for peer " << peer_id << ": returning false" << endl ;
std::cout << "No chunk map for peer " << peer_id << ": supposing full data." << std::endl ;
#endif
return _map.size() ;
std::vector<uint32_t>& pchunks(_peers_chunks_availability[peer_id]) ;
pchunks.resize( (_map.size() >> 5)+!!(_map.size() & 0x11111),~(uint32_t)0 ) ;
it = _peers_chunks_availability.find(peer_id) ;
}
const std::vector<uint32_t> peer_chunks(it->second) ;
#endif
const std::vector<uint32_t>& peer_chunks(it->second) ;
for(unsigned int i=0;i<_map.size();++i)
{
uint32_t j = (start_location+i)%_map.size() ;
uint32_t j = (start_location+i)%(int)_map.size() ; // index of the chunk
if(_map[j] == FileChunksInfo::CHUNK_OUTSTANDING /*&& peers_chunks[j] == CHUNK_DONE*/)
if(_map[j] == FileChunksInfo::CHUNK_OUTSTANDING && COMPRESSED_MAP_READ(peer_chunks,j))
{
#ifdef DEBUG_FTCHUNK
std::cerr << "ChunkMap::getAvailableChunk: returning chunk " << j << " for peer " << peer_id << std::endl;
@ -274,6 +288,16 @@ void ChunkMap::getChunksInfo(FileChunksInfo& info) const
info.file_size = _file_size ;
info.chunk_size = _chunk_size ;
info.chunks = _map ;
info.active_chunks.clear() ;
for(std::map<ChunkNumber,ChunkDownloadInfo>::const_iterator it(_slices_to_download.begin());it!=_slices_to_download.end();++it)
info.active_chunks.push_back(std::pair<uint32_t,uint32_t>(it->first,it->second._remains)) ;
info.compressed_peer_availability_maps.clear() ;
for(std::map<std::string,std::vector<uint32_t> >::const_iterator it(_peers_chunks_availability.begin());it!= _peers_chunks_availability.end();++it)
info.compressed_peer_availability_maps.push_back(std::pair<std::string,std::vector<uint32_t> >(it->first,it->second)) ;
}
void ChunkMap::buildAvailabilityMap(std::vector<uint32_t>& map,uint32_t& chunk_size,uint32_t& chunk_number,FileChunksInfo::ChunkStrategy& strategy) const

View file

@ -24,7 +24,8 @@
// corresponding acive slice. When asked a slice, ftChunkMap cuts out a slice from the remaining part of the chunk
// to download, sends the slice's coordinates and gives a unique slice id (such as the slice offset).
// This class handles a slice of a chunk, at the level of ftFileCreator
// This class handles a slice of a chunk of arbitrary uint32_t size, at the level of ftFileCreator
class ftChunk
{
@ -41,10 +42,10 @@ class ftChunk
time_t ts;
};
// This class handles a single chunk. Although each chunk is requested at once,
// This class handles a single fixed-sized chunk. Although each chunk is requested at once,
// it may be sent back into sub-chunks because of file transfer rate constraints.
// So the dataReceived function should be called to progressively complete the chunk,
// and the getChunk method should ask for a sub0chunk of a given size.
// and the getChunk method should ask for a sub-chunk of a given size.
//
class Chunk
{
@ -118,11 +119,9 @@ class ChunkMap
void buildAvailabilityMap(std::vector<uint32_t>& map,uint32_t& chunk_size,uint32_t& chunk_number,FileChunksInfo::ChunkStrategy& s) const ;
void loadAvailabilityMap(const std::vector<uint32_t>& map,uint32_t chunk_size,uint32_t chunk_number,FileChunksInfo::ChunkStrategy s) ;
#ifdef TO_DO
// Updates the peer's availablility map
//
void setPeerAvailabilityMap(const std::string& peer_id,const std::vector<uint32_t>& peer_map) ;
#endif
void setPeerAvailabilityMap(const std::string& peer_id,uint32_t chunk_size,uint32_t nb_chunks,const std::vector<uint32_t>& peer_map) ;
// Returns the total size of downloaded data in the file.
uint64_t getTotalReceived() const { return _total_downloaded ; }
@ -134,7 +133,7 @@ class ChunkMap
// Returns the first chunk available starting from start_location for this peer_id.
//
uint32_t getAvailableChunk(uint32_t start_location,const std::string& peer_id) const ;
uint32_t getAvailableChunk(uint32_t start_location,const std::string& peer_id) ;
private:
uint64_t _file_size ; // total size of the file in bytes.
@ -145,7 +144,7 @@ class ChunkMap
std::vector<FileChunksInfo::ChunkState> _map ; // vector of chunk state over the whole file
std::map<std::string,std::vector<uint32_t> > _peers_chunks_availability ; // what does each source peer have.
std::map<std::string,std::vector<uint32_t> > _peers_chunks_availability ; // what does each source peer have, stored in compressed format.
uint64_t _total_downloaded ;
};

View file

@ -599,6 +599,8 @@ bool ftController::handleAPendingRequest()
mPendingChunkMaps.erase(it) ;
}
}
return true ;
}
@ -750,7 +752,7 @@ bool ftController::FileRequest(std::string fname, std::string hash,
/* do a source search - for any extra sources */
if (mSearch->search(hash, size,
RS_FILE_HINTS_REMOTE |
RS_FILE_HINTS_TURTLE |
// RS_FILE_HINTS_TURTLE |
RS_FILE_HINTS_SPEC_ONLY, info))
{
/* do something with results */

View file

@ -328,6 +328,31 @@ bool ftDataMultiplex::doWork()
return true;
}
bool ftDataMultiplex::recvFileMap(const std::string& peerId, const std::string& hash, uint32_t chunk_size, uint32_t nb_chunks, const std::vector<uint32_t>& compressed_map)
{
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
std::map<std::string, ftClient>::iterator it;
if (mClients.end() == (it = mClients.find(hash)))
{
#ifdef MPLEX_DEBUG
std::cerr << "ftDataMultiplex::handleRecvMap() ERROR: No matching Client!";
std::cerr << std::endl;
#endif
/* error */
return false;
}
#ifdef MPLEX_DEBUG
std::cerr << "ftDataMultiplex::handleRecvMap() Passing map to FT Module";
std::cerr << std::endl;
#endif
(it->second).mCreator->setSourceMap(peerId, chunk_size, nb_chunks,compressed_map);
return true;
}
bool ftDataMultiplex::handleRecvData(std::string peerId,
std::string hash, uint64_t size,

View file

@ -84,81 +84,72 @@ class ftDataMultiplex: public ftDataRecv, public RsQueueThread
public:
ftDataMultiplex(std::string ownId, ftDataSend *server, ftSearch *search);
ftDataMultiplex(std::string ownId, ftDataSend *server, ftSearch *search);
/* ftController Interface */
bool addTransferModule(ftTransferModule *mod, ftFileCreator *f);
bool removeTransferModule(std::string hash);
/* ftController Interface */
bool addTransferModule(ftTransferModule *mod, ftFileCreator *f);
bool removeTransferModule(std::string hash);
/* data interface */
/* get Details of File Transfers */
bool FileUploads(std::list<std::string> &hashs);
bool FileDownloads(std::list<std::string> &hashs);
bool FileDetails(std::string hash, uint32_t hintsflag, FileInfo &info);
/* data interface */
/* get Details of File Transfers */
bool FileUploads(std::list<std::string> &hashs);
bool FileDownloads(std::list<std::string> &hashs);
bool FileDetails(std::string hash, uint32_t hintsflag, FileInfo &info);
void deleteServers(const std::list<std::string>& serv_hash) ;
void deleteServers(const std::list<std::string>& serv_hash) ;
/*************** SEND INTERFACE (calls ftDataSend) *******************/
/*************** SEND INTERFACE (calls ftDataSend) *******************/
/* Client Send */
bool sendDataRequest(std::string peerId, std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize);
/* Client Send */
bool sendDataRequest(std::string peerId, std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize);
/* Server Send */
bool sendData(std::string peerId, std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize, void *data);
/* Server Send */
bool sendData(std::string peerId, std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize, void *data);
/*************** RECV INTERFACE (provides ftDataRecv) ****************/
/*************** RECV INTERFACE (provides ftDataRecv) ****************/
/* Client Recv */
virtual bool recvData(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize, void *data);
/* Client Recv */
virtual bool recvData(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize, void *data);
virtual bool recvFileMap(const std::string& peerId, const std::string& hash, uint32_t chunk_size, uint32_t nb_chunks, const std::vector<uint32_t>& compressed_map) ;
/* Server Recv */
virtual bool recvDataRequest(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize);
/* Server Recv */
virtual bool recvDataRequest(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize);
protected:
/* Overloaded from RsQueueThread */
virtual bool workQueued();
virtual bool doWork();
/* Overloaded from RsQueueThread */
virtual bool workQueued();
virtual bool doWork();
private:
/* Handling Job Queues */
bool handleRecvData(std::string peerId,
std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize, void *data);
/* Handling Job Queues */
bool handleRecvData(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize, void *data);
bool handleRecvDataRequest(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize);
bool handleSearchRequest(std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize);
bool handleRecvDataRequest(std::string peerId,
std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize);
/* We end up doing the actual server job here */
bool locked_handleServerRequest(ftFileProvider *provider, std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize);
bool handleSearchRequest(std::string peerId,
std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize);
RsMutex dataMtx;
/* We end up doing the actual server job here */
bool locked_handleServerRequest(ftFileProvider *provider,
std::string peerId, std::string hash, uint64_t size,
uint64_t offset, uint32_t chunksize);
std::map<std::string, ftClient> mClients;
std::map<std::string, ftFileProvider *> mServers;
RsMutex dataMtx;
std::list<ftRequest> mRequestQueue;
std::list<ftRequest> mSearchQueue;
std::map<std::string, time_t> mUnknownHashs;
std::map<std::string, ftClient> mClients;
std::map<std::string, ftFileProvider *> mServers;
ftDataSend *mDataSend;
ftSearch *mSearch;
std::string mOwnId;
std::list<ftRequest> mRequestQueue;
std::list<ftRequest> mSearchQueue;
std::map<std::string, time_t> mUnknownHashs;
ftDataSend *mDataSend;
ftSearch *mSearch;
std::string mOwnId;
friend class ftServer;
friend class ftServer;
};
#endif

View file

@ -409,6 +409,7 @@ void ftFileCreator::loadAvailabilityMap(const std::vector<uint32_t>& map,uint32_
chunkMap = ChunkMap(mSize,map,chunk_size,chunk_number,FileChunksInfo::ChunkStrategy(strategy)) ;
}
void ftFileCreator::storeAvailabilityMap(std::vector<uint32_t>& map,uint32_t& chunk_size,uint32_t& chunk_number,uint32_t& strategy)
{
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
@ -418,4 +419,11 @@ void ftFileCreator::storeAvailabilityMap(std::vector<uint32_t>& map,uint32_t& ch
strategy = (uint32_t)strat ;
}
void ftFileCreator::setSourceMap(const std::string& peer_id,uint32_t chunk_size,uint32_t nb_chunks,const std::vector<uint32_t>& compressed_map)
{
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
chunkMap.setPeerAvailabilityMap(peer_id,chunk_size,nb_chunks,compressed_map) ;
}

View file

@ -38,48 +38,64 @@
class ftFileCreator: public ftFileProvider
{
public:
public:
ftFileCreator(std::string savepath, uint64_t size, std::string hash, uint64_t recvd);
~ftFileCreator();
ftFileCreator(std::string savepath, uint64_t size, std::string hash, uint64_t recvd);
/* overloaded from FileProvider */
virtual bool getFileData(uint64_t offset, uint32_t &chunk_size, void *data);
bool finished() { return getRecvd() == getFileSize(); }
uint64_t getRecvd();
void getChunkMap(FileChunksInfo& info) ;
void setChunkStrategy(FileChunksInfo::ChunkStrategy s) ;
~ftFileCreator();
/*
* creation functions for FileCreator
*/
/* overloaded from FileProvider */
virtual bool getFileData(uint64_t offset, uint32_t &chunk_size, void *data);
bool finished() { return getRecvd() == getFileSize(); }
uint64_t getRecvd();
bool getMissingChunk(const std::string& peer_id,uint32_t size_hint,uint64_t& offset, uint32_t& size);
bool addFileData(uint64_t offset, uint32_t chunk_size, void *data);
void getChunkMap(FileChunksInfo& info) ;
void setChunkStrategy(FileChunksInfo::ChunkStrategy s) ;
void loadAvailabilityMap(const std::vector<uint32_t>& map,uint32_t chunk_size,uint32_t chunk_number,uint32_t chunk_strategy) ;
void storeAvailabilityMap(std::vector<uint32_t>& map,uint32_t& chunk_size,uint32_t& chunk_number,uint32_t& chunk_strategy) ;
/*
* creation functions for FileCreator
*/
protected:
// Gets a new variable-sized chunk of size "size_hint" from the given peer id. The returned size, "size" is
// at most equal to size_hint.
//
bool getMissingChunk(const std::string& peer_id,uint32_t size_hint,uint64_t& offset, uint32_t& size);
virtual int initializeFileAttrs();
// actually store data in the file, and update chunks info
//
bool addFileData(uint64_t offset, uint32_t chunk_size, void *data);
private:
// Load/save the availability map for the file being downloaded, in a compact/compressed form.
// This is used for
// - loading and saving info about the current transfers
// - getting info about current chunks for the GUI
// - sending availability info to the peers for which we also are a source
//
void loadAvailabilityMap(const std::vector<uint32_t>& map,uint32_t chunk_size,uint32_t chunk_number,uint32_t chunk_strategy) ;
void storeAvailabilityMap(std::vector<uint32_t>& map,uint32_t& chunk_size,uint32_t& chunk_number,uint32_t& chunk_strategy) ;
bool locked_printChunkMap();
int locked_notifyReceived(uint64_t offset, uint32_t chunk_size);
/*
* structure to track missing chunks
*/
uint64_t mStart;
uint64_t mEnd;
// This is called when receiving the availability map from a source peer, for the file being handled.
//
void setSourceMap(const std::string& peer_id,uint32_t chunk_size,uint32_t nb_chunks,const std::vector<uint32_t>& map) ;
std::map<uint64_t, ftChunk> mChunks;
protected:
ChunkMap chunkMap ;
virtual int initializeFileAttrs();
private:
bool locked_printChunkMap();
int locked_notifyReceived(uint64_t offset, uint32_t chunk_size);
/*
* structure to track missing chunks
*/
uint64_t mStart;
uint64_t mEnd;
std::map<uint64_t, ftChunk> mChunks;
ChunkMap chunkMap ;
};
#endif // FT_FILE_CREATOR_HEADER

View file

@ -156,7 +156,7 @@ void ftServer::connectToTurtleRouter(p3turtle *fts)
{
mTurtleRouter = fts ;
mFtSearch->addSearchMode(fts, RS_FILE_HINTS_TURTLE);
// mFtSearch->addSearchMode(fts, RS_FILE_HINTS_TURTLE);
mFtController->setTurtleRouter(fts) ;
}