mirror of
https://github.com/RetroShare/RetroShare.git
synced 2025-06-12 08:33:05 -04:00
Enabled new File Transfer Code in libretroshare.
NB: This is not yet working. so SVN is now officially broken :( Hope to have the bugs ironed out soon. git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@782 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
drbob
parent
9f68fc5393
commit
0d96b43d34
29 changed files with 567 additions and 117 deletions
|
|
@ -96,8 +96,10 @@ void ftController::run()
|
|||
sleep(1);
|
||||
#endif
|
||||
|
||||
std::cerr << "ftController::run()";
|
||||
std::cerr << std::endl;
|
||||
#ifdef CONTROL_DEBUG
|
||||
//std::cerr << "ftController::run()";
|
||||
//std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
/* tick the transferModules */
|
||||
std::list<std::string> done;
|
||||
|
|
@ -254,6 +256,7 @@ bool ftController::FileRequest(std::string fname, std::string hash,
|
|||
/* check if we have the file */
|
||||
FileInfo info;
|
||||
std::list<std::string>::iterator it;
|
||||
std::list<TransferInfo>::iterator pit;
|
||||
|
||||
#ifdef CONTROL_DEBUG
|
||||
std::cerr << "ftController::FileRequest(" << fname << ",";
|
||||
|
|
@ -313,17 +316,17 @@ bool ftController::FileRequest(std::string fname, std::string hash,
|
|||
#endif
|
||||
|
||||
/* if the sources don't exist already - add in */
|
||||
for(it = info.peerIds.begin(); it != info.peerIds.end(); it++)
|
||||
for(pit = info.peers.begin(); pit != info.peers.end(); pit++)
|
||||
{
|
||||
std::cerr << "\tSource: " << *it;
|
||||
std::cerr << "\tSource: " << pit->peerId;
|
||||
std::cerr << std::endl;
|
||||
|
||||
if (srcIds.end() == std::find(
|
||||
srcIds.begin(), srcIds.end(), *it))
|
||||
srcIds.begin(), srcIds.end(), pit->peerId))
|
||||
{
|
||||
srcIds.push_back(*it);
|
||||
srcIds.push_back(pit->peerId);
|
||||
|
||||
std::cerr << "\tAdding in: " << *it;
|
||||
std::cerr << "\tAdding in: " << pit->peerId;
|
||||
std::cerr << std::endl;
|
||||
}
|
||||
}
|
||||
|
|
@ -563,6 +566,75 @@ bool ftController::FileDetails(std::string hash, FileInfo &info)
|
|||
}
|
||||
|
||||
/* extract details */
|
||||
info.hash = hash;
|
||||
info.fname = it->second.mName;
|
||||
|
||||
/* get list of sources from transferModule */
|
||||
std::list<std::string> peerIds;
|
||||
std::list<std::string>::iterator pit;
|
||||
|
||||
it->second.mTransfer->getFileSources(peerIds);
|
||||
|
||||
double totalRate;
|
||||
uint32_t tfRate;
|
||||
uint32_t state;
|
||||
|
||||
bool isDownloading = false;
|
||||
bool isSuspended = false;
|
||||
|
||||
for(pit = peerIds.begin(); pit != peerIds.end(); pit++)
|
||||
{
|
||||
if (it->second.mTransfer->getPeerState(*pit, state, tfRate))
|
||||
{
|
||||
TransferInfo ti;
|
||||
switch(state)
|
||||
{
|
||||
case PQIPEER_INIT:
|
||||
ti.status = FT_STATE_OKAY;
|
||||
break;
|
||||
case PQIPEER_NOT_ONLINE:
|
||||
ti.status = FT_STATE_WAITING;
|
||||
break;
|
||||
case PQIPEER_DOWNLOADING:
|
||||
isDownloading = true;
|
||||
ti.status = FT_STATE_DOWNLOADING;
|
||||
break;
|
||||
case PQIPEER_IDLE:
|
||||
ti.status = FT_STATE_OKAY;
|
||||
break;
|
||||
default:
|
||||
case PQIPEER_SUSPEND:
|
||||
isSuspended = true;
|
||||
ti.status = FT_STATE_FAILED;
|
||||
break;
|
||||
}
|
||||
|
||||
ti.tfRate = tfRate / 1024.0;
|
||||
ti.peerId = *pit;
|
||||
info.peers.push_back(ti);
|
||||
totalRate += tfRate / 1024.0;
|
||||
}
|
||||
}
|
||||
|
||||
if ((it->second).mCreator->finished())
|
||||
{
|
||||
info.downloadStatus = FT_STATE_COMPLETE;
|
||||
}
|
||||
else if (isDownloading)
|
||||
{
|
||||
info.downloadStatus = FT_STATE_DOWNLOADING;
|
||||
}
|
||||
else if (isSuspended)
|
||||
{
|
||||
info.downloadStatus = FT_STATE_FAILED;
|
||||
}
|
||||
else
|
||||
{
|
||||
info.downloadStatus = FT_STATE_WAITING;
|
||||
}
|
||||
info.tfRate = totalRate;
|
||||
info.size = (it->second).mSize;
|
||||
info.transfered = (it->second).mCreator->getRecvd();
|
||||
|
||||
return true;
|
||||
|
||||
|
|
|
|||
|
|
@ -91,11 +91,21 @@ bool ftFiStore::search(std::string hash, uint64_t size, uint32_t hintflags, File
|
|||
info.fname = it->name;
|
||||
info.size = it->size;
|
||||
info.hash = it->hash;
|
||||
|
||||
}
|
||||
info.peerIds.push_back(it->id);
|
||||
|
||||
TransferInfo ti;
|
||||
ti.peerId = it->id;
|
||||
ti.name = it->name;
|
||||
ti.tfRate = 0;
|
||||
info.peers.push_back(ti);
|
||||
}
|
||||
}
|
||||
|
||||
/**** DEPENDS ON SOURCES!
|
||||
info.downloadStatus = FT_STATE_COMPLETE:
|
||||
****/
|
||||
|
||||
/* if the first flag is cleared, we've definitely
|
||||
* had a full match!.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -44,8 +44,8 @@ void ftExtraList::run()
|
|||
while (1)
|
||||
{
|
||||
#ifdef DEBUG_ELIST
|
||||
std::cerr << "ftExtraList::run() Iteration";
|
||||
std::cerr << std::endl;
|
||||
//std::cerr << "ftExtraList::run() Iteration";
|
||||
//std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
now = time(NULL);
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@
|
|||
* #define FILE_DEBUG 1
|
||||
******/
|
||||
|
||||
#define FILE_DEBUG 1
|
||||
|
||||
#define CHUNK_MAX_AGE 30
|
||||
|
||||
|
||||
|
|
@ -42,12 +44,15 @@ hash, uint64_t recvd): ftFileProvider(path,size,hash)
|
|||
bool ftFileCreator::getFileData(uint64_t offset,
|
||||
uint32_t chunk_size, void *data)
|
||||
{
|
||||
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
|
||||
if (offset + chunk_size > mStart)
|
||||
{
|
||||
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
|
||||
if (offset + chunk_size > mStart)
|
||||
{
|
||||
/* don't have the data */
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ftFileProvider::getFileData(offset, chunk_size, data);
|
||||
}
|
||||
|
||||
|
|
@ -64,6 +69,7 @@ bool ftFileCreator::addFileData(uint64_t offset, uint32_t chunk_size, void *data
|
|||
std::cerr << offset;
|
||||
std::cerr << ", " << chunk_size;
|
||||
std::cerr << ", " << data << ")";
|
||||
std::cerr << " this: " << this;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
/* dodgey checking outside of mutex...
|
||||
|
|
@ -123,7 +129,7 @@ bool ftFileCreator::addFileData(uint64_t offset, uint32_t chunk_size, void *data
|
|||
/*
|
||||
* Notify ftFileChunker about chunks received
|
||||
*/
|
||||
notifyReceived(offset,chunk_size);
|
||||
locked_notifyReceived(offset,chunk_size);
|
||||
|
||||
/*
|
||||
* FIXME HANDLE COMPLETION HERE - Any better way?
|
||||
|
|
@ -136,6 +142,7 @@ int ftFileCreator::initializeFileAttrs()
|
|||
{
|
||||
std::cerr << "ftFileCreator::initializeFileAttrs() Filename: ";
|
||||
std::cerr << file_name;
|
||||
std::cerr << " this: " << this;
|
||||
std::cerr << std::endl;
|
||||
|
||||
/*
|
||||
|
|
@ -195,9 +202,15 @@ ftFileCreator::~ftFileCreator()
|
|||
}
|
||||
|
||||
|
||||
int ftFileCreator::notifyReceived(uint64_t offset, uint32_t chunk_size)
|
||||
int ftFileCreator::locked_notifyReceived(uint64_t offset, uint32_t chunk_size)
|
||||
{
|
||||
/* ALREADY LOCKED */
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ftFileCreator::locked_notifyReceived( " << offset;
|
||||
std::cerr << ", " << chunk_size << " )";
|
||||
std::cerr << " this: " << this;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
/* find the chunk */
|
||||
std::map<uint64_t, ftChunk>::iterator it;
|
||||
|
|
@ -205,6 +218,13 @@ int ftFileCreator::notifyReceived(uint64_t offset, uint32_t chunk_size)
|
|||
bool isFirst = false;
|
||||
if (it == mChunks.end())
|
||||
{
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ftFileCreator::locked_notifyReceived() ";
|
||||
std::cerr << " Failed to match to existing chunk - ignoring";
|
||||
std::cerr << std::endl;
|
||||
|
||||
locked_printChunkMap();
|
||||
#endif
|
||||
return 0; /* ignoring */
|
||||
}
|
||||
else if (it == mChunks.begin())
|
||||
|
|
@ -246,14 +266,21 @@ int ftFileCreator::notifyReceived(uint64_t offset, uint32_t chunk_size)
|
|||
bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
|
||||
{
|
||||
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
|
||||
std::cerr << "ffc::getMissingChunk(...,"<< chunk << ")"<< std::endl;
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ffc::getMissingChunk(...,"<< chunk << ")";
|
||||
std::cerr << " this: " << this;
|
||||
std::cerr << std::endl;
|
||||
locked_printChunkMap();
|
||||
#endif
|
||||
|
||||
/* check start point */
|
||||
|
||||
if (mStart == mSize)
|
||||
{
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ffc::getMissingChunk() File Done";
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -267,8 +294,10 @@ bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
|
|||
/* very simple algorithm */
|
||||
if (it->second.ts < old)
|
||||
{
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ffc::getMissingChunk() ReAlloc";
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
/* retry this one */
|
||||
it->second.ts = ts;
|
||||
|
|
@ -279,10 +308,12 @@ bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
|
|||
}
|
||||
}
|
||||
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ffc::getMissingChunk() new Alloc";
|
||||
std::cerr << " mStart: " << mStart << " mEnd: " << mEnd;
|
||||
std::cerr << "mSize: " << mSize;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
/* else allocate a new chunk */
|
||||
if (mSize - mEnd < chunk)
|
||||
|
|
@ -293,12 +324,48 @@ bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
|
|||
|
||||
if (chunk > 0)
|
||||
{
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ffc::getMissingChunk() Allocated " << chunk;
|
||||
std::cerr << " offset: " << offset;
|
||||
std::cerr << std::endl;
|
||||
std::cerr << " mStart: " << mStart << " mEnd: " << mEnd;
|
||||
std::cerr << "mSize: " << mSize;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
mChunks[offset] = ftChunk(offset, chunk, ts);
|
||||
}
|
||||
|
||||
return true; /* cos more data to get */
|
||||
}
|
||||
|
||||
|
||||
bool ftFileCreator::locked_printChunkMap()
|
||||
{
|
||||
#ifdef FILE_DEBUG
|
||||
std::cerr << "ftFileCreator::locked_printChunkMap()";
|
||||
std::cerr << " this: " << this;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
/* check start point */
|
||||
std::cerr << "Size: " << mSize << " Start: " << mStart << " End: " << mEnd;
|
||||
std::cerr << std::endl;
|
||||
std::cerr << "\tOutstanding Chunks (in the middle)";
|
||||
std::cerr << std::endl;
|
||||
|
||||
std::map<uint64_t, ftChunk>::iterator it;
|
||||
time_t ts = time(NULL);
|
||||
for(it = mChunks.begin(); it != mChunks.end(); it++)
|
||||
{
|
||||
std::cerr << "\tChunk [" << it->second.offset << "] size: ";
|
||||
std::cerr << it->second.chunk;
|
||||
std::cerr << " Age: " << ts - it->second.ts;
|
||||
std::cerr << std::endl;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/***********************************************************
|
||||
*
|
||||
* ftChunk methods
|
||||
|
|
|
|||
|
|
@ -63,7 +63,8 @@ virtual int initializeFileAttrs();
|
|||
|
||||
private:
|
||||
|
||||
int notifyReceived(uint64_t offset, uint32_t chunk_size);
|
||||
bool locked_printChunkMap();
|
||||
int locked_notifyReceived(uint64_t offset, uint32_t chunk_size);
|
||||
/*
|
||||
* structure to track missing chunks
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -452,7 +452,9 @@ bool ftServer::sendDataRequest(std::string peerId, std::string hash,
|
|||
return true;
|
||||
}
|
||||
|
||||
const uint32_t MAX_FT_CHUNK = 32 * 1024; /* 32K */
|
||||
//const uint32_t MAX_FT_CHUNK = 32 * 1024; /* 32K */
|
||||
//const uint32_t MAX_FT_CHUNK = 16 * 1024; /* 16K */
|
||||
const uint32_t MAX_FT_CHUNK = 8 * 1024; /* 16K */
|
||||
|
||||
/* Server Send */
|
||||
bool ftServer::sendData(std::string peerId, std::string hash, uint64_t size,
|
||||
|
|
@ -467,8 +469,8 @@ bool ftServer::sendData(std::string peerId, std::string hash, uint64_t size,
|
|||
#ifdef SERVER_DEBUG
|
||||
std::cerr << "ftServer::sendData() to " << peerId << std::endl;
|
||||
std::cerr << "hash: " << hash;
|
||||
std::cerr << " offset: " << offset;
|
||||
std::cerr << " chunk: " << chunk;
|
||||
std::cerr << " offset: " << baseoffset;
|
||||
std::cerr << " chunk: " << chunksize;
|
||||
std::cerr << " data: " << data;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
|
@ -527,6 +529,10 @@ bool ftServer::sendData(std::string peerId, std::string hash, uint64_t size,
|
|||
}
|
||||
|
||||
|
||||
/* NB: The rsCore lock must be activated before calling this.
|
||||
* This Lock should be moved lower into the system...
|
||||
* most likely destination is in ftServer.
|
||||
*/
|
||||
int ftServer::tick()
|
||||
{
|
||||
rslog(RSL_DEBUG_BASIC, ftserverzone,
|
||||
|
|
@ -586,7 +592,7 @@ bool ftServer::handleCacheData()
|
|||
int i_init = 0;
|
||||
|
||||
#ifdef SERVER_DEBUG
|
||||
std::cerr << "ftServer::handleCacheData()" << std::endl;
|
||||
//std::cerr << "ftServer::handleCacheData()" << std::endl;
|
||||
#endif
|
||||
while((ci = mP3iface -> GetSearchResult()) != NULL)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -145,7 +145,7 @@ int main(int argc, char **argv)
|
|||
std::list<pqiAuthDetails>::iterator fit;
|
||||
|
||||
std::cerr << "Point 3" << std::endl;
|
||||
P3Hub *testHub = new P3Hub();
|
||||
P3Hub *testHub = new P3Hub(0, NULL);
|
||||
testHub->start();
|
||||
std::cerr << "Point 4" << std::endl;
|
||||
|
||||
|
|
|
|||
|
|
@ -165,7 +165,16 @@ int main(int argc, char **argv)
|
|||
std::list<pqiAuthDetails> baseFriendList, friendList;
|
||||
std::list<pqiAuthDetails>::iterator fit;
|
||||
|
||||
P3Hub *testHub = new P3Hub();
|
||||
|
||||
/* Add in serialiser */
|
||||
|
||||
RsSerialiser *rss = new RsSerialiser();
|
||||
rss->addSerialType(new RsFileItemSerialiser());
|
||||
rss->addSerialType(new RsCacheItemSerialiser());
|
||||
rss->addSerialType(new RsServiceSerialiser());
|
||||
|
||||
|
||||
P3Hub *testHub = new P3Hub(0, rss);
|
||||
testHub->start();
|
||||
|
||||
/* Setup Base Friend Info */
|
||||
|
|
|
|||
|
|
@ -175,7 +175,14 @@ int main(int argc, char **argv)
|
|||
std::list<pqiAuthDetails> baseFriendList, friendList;
|
||||
std::list<pqiAuthDetails>::iterator fit;
|
||||
|
||||
P3Hub *testHub = new P3Hub();
|
||||
/* Add in Serialiser Test
|
||||
*/
|
||||
RsSerialiser *rss = new RsSerialiser();
|
||||
rss->addSerialType(new RsFileItemSerialiser());
|
||||
rss->addSerialType(new RsCacheItemSerialiser());
|
||||
rss->addSerialType(new RsServiceSerialiser());
|
||||
|
||||
P3Hub *testHub = new P3Hub(0, rss);
|
||||
testHub->start();
|
||||
|
||||
/* Setup Base Friend Info */
|
||||
|
|
@ -317,8 +324,8 @@ int main(int argc, char **argv)
|
|||
|
||||
while(1)
|
||||
{
|
||||
std::cerr << "ftserver2test::sleep()";
|
||||
std::cerr << std::endl;
|
||||
//std::cerr << "ftserver3test::sleep()";
|
||||
//std::cerr << std::endl;
|
||||
sleep(1);
|
||||
|
||||
/* tick the connmgrs */
|
||||
|
|
@ -372,7 +379,7 @@ void *do_server_test_thread(void *data)
|
|||
std::string oId = oServer->OwnId();
|
||||
|
||||
/* create Expression */
|
||||
uint64_t minFileSize = 10000;
|
||||
uint64_t minFileSize = 100000;
|
||||
//SizeExpression se(Greater, minFileSize);
|
||||
SizeExpression se(Smaller, minFileSize);
|
||||
Expression *expr = &se;
|
||||
|
|
@ -437,9 +444,9 @@ void *do_server_test_thread(void *data)
|
|||
}
|
||||
|
||||
/* Give it a while to transfer */
|
||||
for(int i = 0; i < 10; i++)
|
||||
for(int i = 0; i < 300; i++)
|
||||
{
|
||||
std::cerr << "Waiting 10 seconds to transfer";
|
||||
std::cerr << "Waited " << i * 10 << " seconds for transfer";
|
||||
std::cerr << std::endl;
|
||||
sleep(10);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,11 +27,15 @@
|
|||
* #define FT_DEBUG 1
|
||||
******/
|
||||
|
||||
#define FT_DEBUG 1
|
||||
|
||||
#include "fttransfermodule.h"
|
||||
|
||||
ftTransferModule::ftTransferModule(ftFileCreator *fc, ftDataMultiplex *dm, ftController *c)
|
||||
:mFileCreator(fc), mMultiplexor(dm), mFtController(c), mFlag(0)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
mHash = mFileCreator->getHash();
|
||||
mSize = mFileCreator->getFileSize();
|
||||
mFileStatus.hash = mHash;
|
||||
|
|
@ -45,8 +49,11 @@ ftTransferModule::ftTransferModule(ftFileCreator *fc, ftDataMultiplex *dm, ftCon
|
|||
ftTransferModule::~ftTransferModule()
|
||||
{}
|
||||
|
||||
|
||||
bool ftTransferModule::setFileSources(std::list<std::string> peerIds)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
mFileSources.clear();
|
||||
|
||||
#ifdef FT_DEBUG
|
||||
|
|
@ -73,8 +80,20 @@ bool ftTransferModule::setFileSources(std::list<std::string> peerIds)
|
|||
return true;
|
||||
}
|
||||
|
||||
bool ftTransferModule::getFileSources(std::list<std::string> &peerIds)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
std::map<std::string,peerInfo>::iterator it;
|
||||
for(it = mFileSources.begin(); it != mFileSources.end(); it++)
|
||||
{
|
||||
peerIds.push_back(it->first);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool ftTransferModule::setPeerState(std::string peerId,uint32_t state,uint32_t maxRate)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::setPeerState()";
|
||||
std::cerr << " peerId: " << peerId;
|
||||
|
|
@ -108,8 +127,30 @@ bool ftTransferModule::setPeerState(std::string peerId,uint32_t state,uint32_t m
|
|||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool ftTransferModule::getPeerState(std::string peerId,uint32_t &state,uint32_t &tfRate)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
mit = mFileSources.find(peerId);
|
||||
|
||||
if (mit == mFileSources.end()) return false;
|
||||
|
||||
state = (mit->second).state;
|
||||
tfRate = (uint32_t) (mit->second).actualRate;
|
||||
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::getPeerState()";
|
||||
std::cerr << " peerId: " << peerId;
|
||||
std::cerr << " state: " << state;
|
||||
std::cerr << " tfRate: " << tfRate << std::endl;
|
||||
#endif
|
||||
return true;
|
||||
}
|
||||
|
||||
uint32_t ftTransferModule::getDataRate(std::string peerId)
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
mit = mFileSources.find(peerId);
|
||||
if (mit == mFileSources.end())
|
||||
|
|
@ -139,6 +180,9 @@ bool ftTransferModule::recvFileData(std::string peerId, uint64_t offset,
|
|||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
mit = mFileSources.find(peerId);
|
||||
|
||||
|
|
@ -176,7 +220,7 @@ bool ftTransferModule::recvFileData(std::string peerId, uint64_t offset,
|
|||
|
||||
(mit->second).receivedSize += chunk_size;
|
||||
(mit->second).state = PQIPEER_IDLE;
|
||||
|
||||
} /***** STACK MUTEX END ****/
|
||||
return storeData(offset, chunk_size, data);
|
||||
}
|
||||
|
||||
|
|
@ -246,6 +290,10 @@ bool ftTransferModule::storeData(uint64_t offset, uint32_t chunk_size,void *data
|
|||
|
||||
bool ftTransferModule::queryInactive()
|
||||
{
|
||||
/* NB: Not sure about this lock... might cause deadlock.
|
||||
*/
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::queryInactive()" << std::endl;
|
||||
#endif
|
||||
|
|
@ -291,12 +339,20 @@ bool ftTransferModule::queryInactive()
|
|||
//file response received or peer side is just ready for download
|
||||
case PQIPEER_IDLE:
|
||||
pInfo->actualRate = pInfo->chunkSize/(ts-(pInfo->lastTS));
|
||||
if (pInfo->actualRate < pInfo->desiredRate/2)
|
||||
{
|
||||
req_size = pInfo->chunkSize * 2 ;
|
||||
|
||||
if (pInfo->actualRate < pInfo->desiredRate)
|
||||
{
|
||||
if (pInfo->actualRate < pInfo->desiredRate/2)
|
||||
{
|
||||
req_size = pInfo->chunkSize * 2 ;
|
||||
}
|
||||
else
|
||||
{
|
||||
req_size = (uint32_t ) (pInfo->chunkSize * 1.1) ;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
else
|
||||
{
|
||||
req_size = (uint32_t ) (pInfo->chunkSize * 0.9) ;
|
||||
}
|
||||
|
||||
|
|
@ -336,6 +392,8 @@ bool ftTransferModule::queryInactive()
|
|||
|
||||
bool ftTransferModule::pauseTransfer()
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
/*
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
for(mit = mOnlinePeers.begin(); mit != mOnlinePeers.end(); mit++)
|
||||
|
|
@ -350,6 +408,7 @@ bool ftTransferModule::pauseTransfer()
|
|||
|
||||
bool ftTransferModule::resumeTransfer()
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
/*
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
for(mit = mOnlinePeers.begin(); mit != mOnlinePeers.end(); mit++)
|
||||
|
|
@ -364,6 +423,7 @@ bool ftTransferModule::resumeTransfer()
|
|||
|
||||
bool ftTransferModule::cancelTransfer()
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
mFileStatus.stat=ftFileStatus::PQIFILE_FAIL_CANCEL;
|
||||
|
||||
return 1;
|
||||
|
|
@ -388,7 +448,14 @@ int ftTransferModule::tick()
|
|||
#endif
|
||||
|
||||
queryInactive();
|
||||
switch (mFlag)
|
||||
|
||||
uint32_t flags = 0;
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
flags = mFlag;
|
||||
}
|
||||
|
||||
switch (flags)
|
||||
{
|
||||
case 0:
|
||||
adjustSpeed();
|
||||
|
|
@ -409,12 +476,30 @@ int ftTransferModule::tick()
|
|||
|
||||
void ftTransferModule::adjustSpeed()
|
||||
{
|
||||
RsStackMutex stack(tfMtx); /******* STACK LOCKED ******/
|
||||
|
||||
std::map<std::string,peerInfo>::iterator mit;
|
||||
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::adjustSpeed()";
|
||||
std::cerr << " Initial Desired Rate: " << desiredRate << " Actual Rate: " << actualRate;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
|
||||
|
||||
for(mit = mFileSources.begin(); mit != mFileSources.end(); mit++)
|
||||
{
|
||||
if (((mit->second).state == PQIPEER_DOWNLOADING)
|
||||
|| ((mit->second).state == PQIPEER_IDLE))
|
||||
{
|
||||
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::adjustSpeed()";
|
||||
std::cerr << "\t" << mit->first << " Desired Rate: " << desiredRate << " Actual Rate: " << actualRate;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
if ((actualRate < desiredRate) && ((mit->second).actualRate >= (mit->second).desiredRate))
|
||||
{
|
||||
(mit->second).desiredRate *= 1.1;
|
||||
|
|
@ -426,6 +511,11 @@ void ftTransferModule::adjustSpeed()
|
|||
}
|
||||
}
|
||||
}
|
||||
#ifdef FT_DEBUG
|
||||
std::cerr << "ftTransferModule::adjustSpeed()";
|
||||
std::cerr << " Initial Desired Rate: " << desiredRate << " Actual Rate: " << actualRate;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -132,6 +132,8 @@ public:
|
|||
//interface to download controller
|
||||
bool setFileSources(std::list<std::string> peerIds);
|
||||
bool setPeerState(std::string peerId,uint32_t state,uint32_t maxRate); //state = ONLINE/OFFLINE
|
||||
bool getFileSources(std::list<std::string> &peerIds);
|
||||
bool getPeerState(std::string peerId,uint32_t &state,uint32_t &tfRate);
|
||||
uint32_t getDataRate(std::string peerId);
|
||||
bool pauseTransfer();
|
||||
bool resumeTransfer();
|
||||
|
|
|
|||
|
|
@ -27,11 +27,18 @@
|
|||
#include "pqi/p3connmgr.h"
|
||||
|
||||
/******
|
||||
*#define HUB_DEBUG
|
||||
*#define HUB_DEBUG 1
|
||||
*****/
|
||||
|
||||
P3Hub::P3Hub()
|
||||
#define HUB_DEBUG 1
|
||||
|
||||
P3Hub::P3Hub(uint32_t flags, RsSerialiser *rss)
|
||||
:mSerialiser(rss), mUseSerialiser(false)
|
||||
{
|
||||
if (rss)
|
||||
{
|
||||
mUseSerialiser = true;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -57,6 +64,54 @@ void P3Hub::addP3Pipe(std::string id, P3Pipe *pqi, p3ConnectMgr *mgr)
|
|||
|
||||
}
|
||||
|
||||
RsItem *P3Hub::SerialiserPass(RsItem *inItem)
|
||||
{
|
||||
/* pass through serialiser */
|
||||
|
||||
RsItem *item = NULL;
|
||||
|
||||
uint32_t pktsize = mSerialiser->size(inItem);
|
||||
void *ptr = malloc(pktsize);
|
||||
|
||||
#ifdef HUB_DEBUG
|
||||
std::cerr << "P3Hub::SerialiserPass() Expected Size: " << pktsize;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
if (!mSerialiser->serialise(inItem, ptr, &pktsize))
|
||||
{
|
||||
|
||||
#ifdef HUB_DEBUG
|
||||
std::cerr << "P3Hub::SerialiserPass() serialise Failed";
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
}
|
||||
else
|
||||
{
|
||||
|
||||
#ifdef HUB_DEBUG
|
||||
std::cerr << "P3Hub::SerialiserPass() serialise success, size: " << pktsize;
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
item = mSerialiser->deserialise(ptr, &pktsize);
|
||||
item->PeerId(inItem->PeerId());
|
||||
|
||||
if (!item)
|
||||
{
|
||||
#ifdef HUB_DEBUG
|
||||
std::cerr << "P3Hub::SerialiserPass() deSerialise Failed";
|
||||
std::cerr << std::endl;
|
||||
#endif
|
||||
}
|
||||
}
|
||||
|
||||
delete inItem;
|
||||
free(ptr);
|
||||
return item;
|
||||
}
|
||||
|
||||
|
||||
void P3Hub::run()
|
||||
{
|
||||
|
|
@ -89,8 +144,16 @@ void P3Hub::run()
|
|||
std::cerr << std::endl;
|
||||
#endif
|
||||
|
||||
|
||||
recvdQ.push_back(make_pair(it->first, item));
|
||||
if (mUseSerialiser)
|
||||
{
|
||||
item = SerialiserPass(item);
|
||||
}
|
||||
|
||||
/* serialiser might hav munched it. */
|
||||
if (item)
|
||||
{
|
||||
recvdQ.push_back(make_pair(it->first, item));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -67,13 +67,18 @@ class P3Hub: public RsThread
|
|||
{
|
||||
public:
|
||||
|
||||
P3Hub();
|
||||
P3Hub(uint32_t flags, RsSerialiser *rss);
|
||||
void addP3Pipe(std::string id, P3Pipe *, p3ConnectMgr *mgr);
|
||||
|
||||
virtual void run();
|
||||
|
||||
private:
|
||||
|
||||
RsItem* SerialiserPass(RsItem *inItem);
|
||||
|
||||
std::map<std::string, hubItem> mPeers;
|
||||
RsSerialiser *mSerialiser;
|
||||
bool mUseSerialiser;
|
||||
};
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue