mirror of
https://github.com/RetroShare/RetroShare.git
synced 2025-06-06 21:58:57 -04:00
More Improvements to FileTransfer:
* Added TlvShallowClear() to serialisers * Added RsQueueThread for periodic queue processes. * Completed ftDataMultiplex which replaces ftServer/ClientModules. * Added Server Queue / Thread to ftDataMultiplex. * Added ftdataplextest to exercise ftDataMultiplex. * Generalised ftFileSearch to handle an array of ftSearch classes. * Tweaked rsfiles.h #defines to match new ftFileSearch scheme. * Added Generic ftData Interfaces for Testing. * added ftDataSend/Recv Interfaces to ftServer + ftDataMultiplex respectively. * Completed much of ftServer (External Interface), but not yet done. * Extra debugging and small changes to ftExtraList * Integrated new ftTransferModule with minor interface changes. git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@660 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
drbob
parent
d584516859
commit
5c6e558942
22 changed files with 1355 additions and 326 deletions
|
|
@ -35,6 +35,11 @@
|
|||
#include "ft/ftfileprovider.h"
|
||||
#include "ft/ftsearch.h"
|
||||
|
||||
/* For Thread Behaviour */
|
||||
const uint32_t DMULTIPLEX_MIN = 10; /* 1ms sleep */
|
||||
const uint32_t DMULTIPLEX_MAX = 1000; /* 1 sec sleep */
|
||||
const double DMULTIPLEX_RELAX = 0.5; /* ??? */
|
||||
|
||||
ftClient::ftClient(ftTransferModule *module, ftFileCreator *creator)
|
||||
:mModule(module), mCreator(creator)
|
||||
{
|
||||
|
|
@ -44,15 +49,16 @@ ftClient::ftClient(ftTransferModule *module, ftFileCreator *creator)
|
|||
const uint32_t FT_DATA = 0x0001;
|
||||
const uint32_t FT_DATA_REQ = 0x0002;
|
||||
|
||||
ftRequest::ftRequest(uint32_t type, std::string peerId, std::string hash, uint64_t offset, uint32_t chunk, void *data)
|
||||
:mType(type), mPeerId(peerId), mHash(hash),
|
||||
ftRequest::ftRequest(uint32_t type, std::string peerId, std::string hash, uint64_t size, uint64_t offset, uint32_t chunk, void *data)
|
||||
:mType(type), mPeerId(peerId), mHash(hash), mSize(size),
|
||||
mOffset(offset), mChunk(chunk), mData(data)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
ftDataMultiplex::ftDataMultiplex(ftDataSend *server, ftSearch *search)
|
||||
:mDataSend(server), mSearch(search)
|
||||
:RsQueueThread(DMULTIPLEX_MIN, DMULTIPLEX_MAX, DMULTIPLEX_RELAX),
|
||||
mDataSend(server), mSearch(search)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
|
@ -90,49 +96,135 @@ bool ftDataMultiplex::removeTransferModule(ftTransferModule *mod, ftFileCreator
|
|||
/*************** SEND INTERFACE (calls ftDataSend) *******************/
|
||||
|
||||
/* Client Send */
|
||||
bool ftDataMultiplex::sendDataRequest(std::string peerId, std::string hash, uint64_t offset, uint32_t size)
|
||||
bool ftDataMultiplex::sendDataRequest(std::string peerId,
|
||||
std::string hash, uint64_t size, uint64_t offset, uint32_t chunksize)
|
||||
{
|
||||
return mDataSend->sendDataRequest(peerId, hash, offset, size);
|
||||
return mDataSend->sendDataRequest(peerId,hash,size,offset,chunksize);
|
||||
}
|
||||
|
||||
/* Server Send */
|
||||
bool ftDataMultiplex::sendData(std::string peerId, std::string hash, uint64_t offset, uint32_t size, void *data)
|
||||
bool ftDataMultiplex::sendData(std::string peerId,
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize, void *data)
|
||||
{
|
||||
return mDataSend->sendData(peerId, hash, offset, size, data);
|
||||
return mDataSend->sendData(peerId,hash,size,offset,chunksize,data);
|
||||
}
|
||||
|
||||
|
||||
/*************** RECV INTERFACE (provides ftDataRecv) ****************/
|
||||
|
||||
/* Client Recv */
|
||||
bool ftDataMultiplex::recvData(std::string peerId, std::string hash, uint64_t offset, uint32_t size, void *data)
|
||||
bool ftDataMultiplex::recvData(std::string peerId,
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize, void *data)
|
||||
{
|
||||
/* Store in Queue */
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
mRequestQueue.push_back(
|
||||
ftRequest(FT_DATA, peerId, hash, offset, size, data));
|
||||
ftRequest(FT_DATA,peerId,hash,size,offset,chunksize,data));
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
/* Server Recv */
|
||||
bool ftDataMultiplex::recvDataRequest(std::string peerId, std::string hash, uint64_t offset, uint32_t size)
|
||||
bool ftDataMultiplex::recvDataRequest(std::string peerId,
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize)
|
||||
{
|
||||
/* Store in Queue */
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
mRequestQueue.push_back(
|
||||
ftRequest(FT_DATA_REQ, peerId, hash, offset, size, NULL));
|
||||
ftRequest(FT_DATA_REQ,peerId,hash,size,offset,chunksize,NULL));
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
/*********** BACKGROUND THREAD OPERATIONS ***********/
|
||||
bool ftDataMultiplex::workQueued()
|
||||
{
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
if (mRequestQueue.size() > 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (mSearchQueue.size() > 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
bool ftDataMultiplex::doWork()
|
||||
{
|
||||
bool doRequests = true;
|
||||
|
||||
/* Handle All the current Requests */
|
||||
while(doRequests)
|
||||
{
|
||||
ftRequest req;
|
||||
|
||||
{
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
if (mRequestQueue.size() == 0)
|
||||
{
|
||||
doRequests = false;
|
||||
continue;
|
||||
}
|
||||
|
||||
req = mRequestQueue.front();
|
||||
mRequestQueue.pop_front();
|
||||
}
|
||||
|
||||
/* MUTEX FREE */
|
||||
|
||||
switch(req.mType)
|
||||
{
|
||||
case FT_DATA:
|
||||
handleRecvData(req.mPeerId, req.mHash, req.mSize,
|
||||
req.mOffset, req.mChunk, req.mData);
|
||||
break;
|
||||
|
||||
case FT_DATA_REQ:
|
||||
handleRecvDataRequest(req.mPeerId, req.mHash,
|
||||
req.mSize, req.mOffset, req.mChunk);
|
||||
break;
|
||||
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* Only Handle One Search Per Period....
|
||||
* Lower Priority
|
||||
*/
|
||||
ftRequest req;
|
||||
|
||||
{
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
if (mSearchQueue.size() == 0)
|
||||
{
|
||||
/* Finished */
|
||||
return true;
|
||||
}
|
||||
|
||||
req = mSearchQueue.front();
|
||||
mSearchQueue.pop_front();
|
||||
}
|
||||
|
||||
handleSearchRequest(req.mPeerId, req.mHash, req.mSize,
|
||||
req.mOffset, req.mChunk);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool ftDataMultiplex::handleRecvData(std::string peerId,
|
||||
std::string hash, uint64_t offset, uint32_t size, void *data)
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize, void *data)
|
||||
{
|
||||
RsStackMutex stack(dataMtx); /******* LOCK MUTEX ******/
|
||||
std::map<std::string, ftClient>::iterator it;
|
||||
|
|
@ -142,7 +234,7 @@ bool ftDataMultiplex::handleRecvData(std::string peerId,
|
|||
return false;
|
||||
}
|
||||
|
||||
(it->second).mModule->recvFileData(peerId, offset, size, data);
|
||||
(it->second).mModule->recvFileData(peerId, offset, chunksize, data);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
|
@ -150,7 +242,8 @@ bool ftDataMultiplex::handleRecvData(std::string peerId,
|
|||
|
||||
/* called by ftTransferModule */
|
||||
bool ftDataMultiplex::handleRecvDataRequest(std::string peerId,
|
||||
std::string hash, uint64_t offset, uint32_t size)
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize)
|
||||
{
|
||||
/**** Find Files *****/
|
||||
|
||||
|
|
@ -159,7 +252,7 @@ bool ftDataMultiplex::handleRecvDataRequest(std::string peerId,
|
|||
if (mClients.end() != (cit = mClients.find(hash)))
|
||||
{
|
||||
locked_handleServerRequest((cit->second).mCreator,
|
||||
peerId, hash, offset, size);
|
||||
peerId, hash, size, offset, chunksize);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -167,26 +260,28 @@ bool ftDataMultiplex::handleRecvDataRequest(std::string peerId,
|
|||
if (mServers.end() != (sit = mServers.find(hash)))
|
||||
{
|
||||
locked_handleServerRequest(sit->second,
|
||||
peerId, hash, offset, size);
|
||||
peerId, hash, size, offset, chunksize);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
/* Add to Search Queue */
|
||||
mSearchQueue.push_back(
|
||||
ftRequest(FT_DATA_REQ, peerId, hash, offset, size, NULL));
|
||||
ftRequest(FT_DATA_REQ, peerId, hash,
|
||||
size, offset, chunksize, NULL));
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool ftDataMultiplex::locked_handleServerRequest(ftFileProvider *provider,
|
||||
std::string peerId, std::string hash, uint64_t offset, uint32_t size)
|
||||
std::string peerId, std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize)
|
||||
{
|
||||
void *data = malloc(size);
|
||||
if (provider->getFileData(offset, size, data))
|
||||
if (provider->getFileData(offset, chunksize, data))
|
||||
{
|
||||
/* send data out */
|
||||
sendData(peerId, hash, offset, size, data);
|
||||
sendData(peerId, hash, size, offset, chunksize, data);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
|
|
@ -194,7 +289,8 @@ bool ftDataMultiplex::locked_handleServerRequest(ftFileProvider *provider,
|
|||
|
||||
|
||||
bool ftDataMultiplex::handleSearchRequest(std::string peerId,
|
||||
std::string hash, uint64_t offset, uint32_t size)
|
||||
std::string hash, uint64_t size,
|
||||
uint64_t offset, uint32_t chunksize)
|
||||
{
|
||||
|
||||
{
|
||||
|
|
@ -235,7 +331,7 @@ bool ftDataMultiplex::handleSearchRequest(std::string peerId,
|
|||
|
||||
/* handle request finally */
|
||||
locked_handleServerRequest(provider,
|
||||
peerId, hash, offset, size);
|
||||
peerId, hash, size, offset, chunksize);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue