Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2024-12-18 20:34:26 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Major bugfixes and testing changes for ft.

Browse Source 
It is now almost ready for integration with rs.

Added ftserver3test.cc which successfully tests:
	* Hashing of a directory.
	* Transfer of Cache Files.
	* Local Searching.
	* Remote Searching.
	* Download from multiple sources.




git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@773 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
drbob 2008-10-29 20:58:23 +00:00
parent 889825b77e
commit 88ef2e0e34
17 changed files with 1010 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
libretroshare/src/ft/Makefile
View File

@ -12,10 +12,10 @@ RSOBJ = ftdata.o ftfileprovider.o ftfilecreator.o ftextralist.o \
ftcontroller.o pqitestor.o
TESTOBJ = ftfileprovidertest.o ftfilecreatortest.o ftextralisttest.o ftdataplextest.o ftserver1test.o ftserver2test.o fttransfermoduletest.o
TESTOBJ = ftfileprovidertest.o ftfilecreatortest.o ftextralisttest.o ftdataplextest.o ftserver1test.o ftserver2test.o fttransfermoduletest.o ftserver3test.o
TESTS = ftfileprovidertest ftfilecreatortest ftextralisttest ftdataplextest ftserver1test ftserver2test fttransfermoduletest
TESTS = ftfileprovidertest ftfilecreatortest ftextralisttest ftdataplextest ftserver1test ftserver2test fttransfermoduletest ftserver3test
all: librs tests
@ -40,6 +40,9 @@ ftserver1test : ftserver1test.o
ftserver2test : ftserver2test.o
$(CC) $(CFLAGS) -o ftserver2test ftserver2test.o $(LIBS)
ftserver3test : ftserver3test.o
$(CC) $(CFLAGS) -o ftserver3test ftserver3test.o $(LIBS)
###############################################################
include $(RS_TOP_DIR)/scripts/rules.mk
###############################################################

179
libretroshare/src/ft/ftcontroller.cc
View File

@ -60,12 +60,16 @@ ftFileControl::ftFileControl()
return;
}
ftFileControl::ftFileControl(std::string fname, uint64_t size,
std::string hash, uint32_t flags,
ftFileCreator *fc, ftTransferModule *tm)
:mTransfer(tm), mCreator(fc), mState(0), mHash(hash),
mName(fname), mSize(size), mFlags(0)
ftFileControl::ftFileControl(std::string fname,
std::string tmppath, std::string dest,
uint64_t size, std::string hash, uint32_t flags,
ftFileCreator *fc, ftTransferModule *tm, uint32_t cb)
:mName(fname), mCurrentPath(tmppath), mDestination(dest),
mTransfer(tm), mCreator(fc), mState(0), mHash(hash),
mSize(size), mFlags(0), mDoCallback(false), mCallbackCode(cb)
{
if (cb)
mDoCallback = true;
return;
}
@ -96,6 +100,9 @@ void ftController::run()
std::cerr << std::endl;
/* tick the transferModules */
std::list<std::string> done;
std::list<std::string>::iterator it;
{
RsStackMutex stack(ctrlMutex); /******* LOCKED ********/
std::map<std::string, ftFileControl>::iterator it;
@ -104,9 +111,19 @@ void ftController::run()
std::cerr << "\tTicking: " << it->first;
std::cerr << std::endl;
if (it->second.mTransfer)
(it->second.mTransfer)->tick();
}
}
RsStackMutex stack2(doneMutex);
for(it = mDone.begin(); it != mDone.end(); it++)
{
completeFile(*it);
}
mDone.clear();
}
}
@ -119,49 +136,111 @@ void ftController::checkDownloadQueue()
}
bool ftController::FlagFileComplete(std::string hash)
{
RsStackMutex stack2(doneMutex);
mDone.push_back(hash);
std::cerr << "ftController:FlagFileComplete(" << hash << ")";
std::cerr << std::endl;
return true;
}
bool ftController::completeFile(std::string hash)
{
#if 1
RsStackMutex stack(ctrlMutex); /******* LOCKED ********/
std::cerr << "ftController:completeFile(" << hash << ")";
std::cerr << std::endl;
std::map<std::string, ftFileControl>::iterator it;
it = mDownloads.find(hash);
if (it == mDownloads.end())
{
std::cerr << "ftController:completeFile(" << hash << ")";
std::cerr << " Not Found!";
std::cerr << std::endl;
return false;
}
/* check if finished */
//if (!(it->second).mCreator->finished())
if (!(it->second).mCreator->finished())
{
/* not done! */
//return false;
std::cerr << "ftController:completeFile(" << hash << ")";
std::cerr << " Transfer Not Done";
std::cerr << std::endl;
std::cerr << "FileSize: ";
std::cerr << (it->second).mCreator->getFileSize();
std::cerr << " and Recvd: ";
std::cerr << (it->second).mCreator->getRecvd();
return false;
}
ftFileControl *fc = &(it->second);
/* done - cleanup */
//fc->mTransfer->done();
mDataplex->removeTransferModule(fc->mTransfer->hash());
if (fc->mTransfer)
{
delete fc->mTransfer;
fc->mTransfer = NULL;
}
if (fc->mCreator)
{
delete fc->mCreator;
fc->mCreator = NULL;
}
fc->mState = ftFileControl::COMPLETED;
/* Move to Correct Location */
if (0 == rename(fc->mCurrentPath.c_str(), fc->mDestination.c_str()))
{
/* correct the file_name */
fc->mCurrentPath = fc->mDestination;
}
else
{
fc->mState = ftFileControl::ERROR_COMPLETION;
}
/* If it has a callback - do it now */
if (fc->mDoCallback)
{
switch (fc->mCallbackCode)
{
case CB_CODE_CACHE:
/* callback */
if (fc->mState == ftFileControl::COMPLETED)
{
CompletedCache(fc->mHash);
}
else
{
FailedCache(fc->mHash);
}
break;
case CB_CODE_MEDIA:
break;
}
}
/* switch map */
mCompleted[fc->mHash] = *fc;
mDownloads.erase(it);
return true;
#endif
}
/***************************************************************/
@ -189,10 +268,20 @@ bool ftController::FileRequest(std::string fname, std::string hash,
std::cerr << std::endl;
#endif
if (flags | RS_FILE_HINTS_NO_SEARCH)
bool doCallback = false;
uint32_t callbackCode = 0;
if (flags & RS_FILE_HINTS_NO_SEARCH)
{
#ifdef CONTROL_DEBUG
std::cerr << "ftController::FileRequest() Flags for NO_SEARCH ";
std::cerr << std::endl;
#endif
/* no search */
if (flags & RS_FILE_HINTS_CACHE)
{
doCallback = true;
callbackCode = CB_CODE_CACHE;
}
}
else
{
@ -203,6 +292,12 @@ bool ftController::FileRequest(std::string fname, std::string hash,
{
/* have it already */
/* add in as completed transfer */
#ifdef CONTROL_DEBUG
std::cerr << "ftController::FileRequest() Matches Local File";
std::cerr << std::endl;
std::cerr << "\tNo need to download";
std::cerr << std::endl;
#endif
return true;
}
@ -212,19 +307,54 @@ bool ftController::FileRequest(std::string fname, std::string hash,
RS_FILE_HINTS_SPEC_ONLY, info))
{
/* do something with results */
#ifdef CONTROL_DEBUG
std::cerr << "ftController::FileRequest() Found Other Sources";
std::cerr << std::endl;
#endif
/* if the sources don't exist already - add in */
for(it = info.peerIds.begin(); it != info.peerIds.end(); it++)
{
std::cerr << "\tSource: " << *it;
std::cerr << std::endl;
if (srcIds.end() == std::find(
srcIds.begin(), srcIds.end(), *it))
{
srcIds.push_back(*it);
std::cerr << "\tAdding in: " << *it;
std::cerr << std::endl;
}
}
}
std::map<std::string, ftTransferModule *> mTransfers;
std::map<std::string, ftFileCreator *> mFileCreators;
if (flags & RS_FILE_HINTS_MEDIA)
{
doCallback = true;
callbackCode = CB_CODE_MEDIA;
}
}
//std::map<std::string, ftTransferModule *> mTransfers;
//std::map<std::string, ftFileCreator *> mFileCreators;
/* add in new item for download */
std::string savepath = mDownloadPath + "/" + fname;
std::string savepath = mPartialsPath + "/" + hash;
std::string destination = dest + "/" + fname;
/* if no destpath - send to download directory */
if (dest == "")
{
destination = mDownloadPath + "/" + fname;
}
ftFileCreator *fc = new ftFileCreator(savepath, size, hash, 0);
ftTransferModule *tm = new ftTransferModule(fc, mDataplex,this);
/* add into maps */
ftFileControl ftfc(fname, size, hash, flags, fc, tm);
ftFileControl ftfc(fname, savepath, destination,
size, hash, flags, fc, tm, callbackCode);
/* add to ClientModule */
mDataplex->addTransferModule(tm, fc);
@ -275,6 +405,7 @@ bool ftController::FileRequest(std::string fname, std::string hash,
mDownloads[hash] = ftfc;
mSlowQueue.push_back(hash);
return true;
}
@ -381,7 +512,6 @@ bool ftController::setDownloadDirectory(std::string path)
bool ftController::setPartialsDirectory(std::string path)
{
#if 0 /*** FIX ME !!!**************/
/* check it is not a subdir of download / shared directories (BAD) - TODO */
@ -391,19 +521,20 @@ bool ftController::setPartialsDirectory(std::string path)
{
RsStackMutex stack(ctrlMutex); /******* LOCKED ********/
mPartialPath = path;
mPartialsPath = path;
#if 0 /*** FIX ME !!!**************/
/* move all existing files! */
std::map<std::string, ftFileControl>::iterator it;
for(it = mDownloads.begin(); it != mDownloads.end(); it++)
{
(it->second).mCreator->changePartialDirectory(mPartialPath);
}
#endif
return true;
}
return false;
#endif
}
std::string ftController::getDownloadDirectory()
@ -417,7 +548,7 @@ std::string ftController::getPartialsDirectory()
{
RsStackMutex stack(ctrlMutex); /******* LOCKED ********/
return mPartialPath;
return mPartialsPath;
}
bool ftController::FileDetails(std::string hash, FileInfo &info)
@ -525,6 +656,8 @@ bool ftController::RequestCacheFile(RsPeerId id, std::string path, std::string h
FileRequest(hash, hash, size, path,
RS_FILE_HINTS_CACHE | RS_FILE_HINTS_NO_SEARCH, ids);
return true;
}
@ -536,7 +669,7 @@ bool ftController::CancelCacheFile(RsPeerId id, std::string path, std::string ha
std::cerr << std::endl;
#endif
return true;
}

29
libretroshare/src/ft/ftcontroller.h
View File

@ -53,23 +53,34 @@ class ftDataMultiplex;
#include <map>
const uint32_t CB_CODE_CACHE = 0x0001;
const uint32_t CB_CODE_MEDIA = 0x0002;
const uint32_t FC_TRANSFER_COMPLETE = 0x0001;
class ftFileControl
{
public:
enum {DOWNLOADING,COMPLETED};
enum {DOWNLOADING,COMPLETED,ERROR_COMPLETION};
ftFileControl();
ftFileControl(std::string fname, uint64_t size, std::string hash,
uint32_t flags, ftFileCreator *fc, ftTransferModule *tm);
ftFileControl(std::string fname, std::string tmppath, std::string dest,
uint64_t size, std::string hash, uint32_t flags,
ftFileCreator *fc, ftTransferModule *tm, uint32_t cb_flags);
std::string mName;
std::string mCurrentPath; /* current full path (including name) */
std::string mDestination; /* final full path (including name) */
ftTransferModule * mTransfer;
ftFileCreator * mCreator;
uint32_t mState;
std::string mHash;
std::string mName;
uint64_t mSize;
uint32_t mFlags;
bool mDoCallback;
uint32_t mCallbackCode;
};
@ -95,7 +106,7 @@ bool FileRequest(std::string fname, std::string hash,
bool FileCancel(std::string hash);
bool FileControl(std::string hash, uint32_t flags);
bool FileClearCompleted();
bool completeFile(std::string hash);
bool FlagFileComplete(std::string hash);
/* get Details of File Transfers */
bool FileDownloads(std::list<std::string> &hashs);
@ -135,7 +146,7 @@ virtual bool loadList(std::list<RsItem *> load);
/* RunTime Functions */
void checkDownloadQueue();
//bool completeFile(std::string hash);
bool completeFile(std::string hash);
/* pointers to other components */
@ -155,12 +166,16 @@ void checkDownloadQueue();
std::string mConfigPath;
std::string mDownloadPath;
std::string mPartialPath;
std::string mPartialsPath;
/**** SPEED QUEUES ****/
std::list<std::string> mSlowQueue;
std::list<std::string> mStreamQueue;
std::list<std::string> mFastQueue;
/* callback list (for File Completion) */
RsMutex doneMutex;
std::list<std::string> mDone;
};
#endif

21
libretroshare/src/ft/ftdatamultiplex.cc
View File

@ -40,7 +40,9 @@ const uint32_t DMULTIPLEX_MIN = 10; /* 1ms sleep */
const uint32_t DMULTIPLEX_MAX = 1000; /* 1 sec sleep */
const double DMULTIPLEX_RELAX = 0.5; /* ??? */
#define MPLEX_DEBUG 1
/******
* #define MPLEX_DEBUG 1
*****/
ftClient::ftClient(ftTransferModule *module, ftFileCreator *creator)
:mModule(module), mCreator(creator)
@ -386,12 +388,29 @@ bool ftDataMultiplex::locked_handleServerRequest(ftFileProvider *provider,
uint64_t offset, uint32_t chunksize)
{
void *data = malloc(size);
#ifdef MPLEX_DEBUG
std::cerr << "ftDataMultiplex::locked_handleServerRequest()";
std::cerr << "\t peer: " << peerId << " hash: " << hash;
std::cerr << " size: " << size;
std::cerr << std::endl;
std::cerr << "\t offset: " << offset;
std::cerr << " chunksize: " << chunksize << " data: " << data;
std::cerr << std::endl;
#endif
if (provider->getFileData(offset, chunksize, data))
{
/* send data out */
sendData(peerId, hash, size, offset, chunksize, data);
return true;
}
#ifdef MPLEX_DEBUG
std::cerr << "ftDataMultiplex::locked_handleServerRequest()";
std::cerr << " FAILED";
std::cerr << std::endl;
#endif
return false;
}

85
libretroshare/src/ft/ftdbase.cc
View File

@ -26,6 +26,7 @@
#include "ft/ftdbase.h"
#include "util/rsdir.h"
#define DB_DEBUG 1
ftFiStore::ftFiStore(CacheStrapper *cs, CacheTransfer *cft, NotifyBase *cb_in,
RsPeerId ownid, std::string cachedir)
@ -36,20 +37,71 @@ ftFiStore::ftFiStore(CacheStrapper *cs, CacheTransfer *cft, NotifyBase *cb_in,
bool ftFiStore::search(std::string hash, uint64_t size, uint32_t hintflags, FileInfo &info) const
{
/* could use hintflags to specify which bits of fileinfo to use additionally.
eg. hintflags & FT_SEARCH_PEER_ID, then only return matching peers + hash.
eg. hintflags & FT_SEARCH_NAME, then only return matching name + hash.
*
* Still to see if concept is worthwhle
*/
#ifdef DB_DEBUG
std::cerr << "ftFiStore::search(" << hash << "," << size << "," << hintflags;
std::cerr << ")";
std::cerr << std::endl;
#endif
std::list<FileDetail> results;
std::list<FileDetail>::iterator it;
if (SearchHash(hash, results))
{
bool first = true;
for(it = results.begin(); it != results.end(); it++)
{
if (it->size == size)
#ifdef DB_DEBUG
std::cerr << "ftFiStore::search() found: ";
std::cerr << it->name << " (" << it->size;
std::cerr << ") @ " << it->id << " = " << hash;
std::cerr << std::endl;
#endif
bool fullmatch = true;
if (it->size != size)
fullmatch = false;
#if 0
if (hintflags & FT_SEARCH_PEER_ID)
{
/*
pit = std::find(info.srcIds.begin(),
info.srcId.end(). it->id);
if (pit == info.srcIds.end())
{
fullmatch = false;
}
}
#endif
if (fullmatch)
{
if (first)
{
first = false;
info.fname = it->name;
info.size = it->size;
info.hash = it->hash;
}
info.peerIds.push_back(it->id);
}
}
/* if the first flag is cleared, we've definitely
* had a full match!.
*/
}
}
if (!first)
return true;
}
return false;
}
@ -66,9 +118,21 @@ bool ftFiMonitor::search(std::string hash, uint64_t size, uint32_t hintflags, Fi
uint64_t fsize;
std::string path;
#ifdef DB_DEBUG
std::cerr << "ftFiMonitor::search(" << hash << "," << size << "," << hintflags;
std::cerr << ")";
std::cerr << std::endl;
#endif
if (findLocalFile(hash, path, fsize))
{
/* fill in details */
#ifdef DB_DEBUG
std::cerr << "ftFiMonitor::search() found: ";
std::cerr << path;
std::cerr << " = " << hash << "," << fsize;
std::cerr << std::endl;
#endif
info.size = fsize;
info.fname = RsDirUtil::getTopDir(path);
@ -89,9 +153,22 @@ ftCacheStrapper::ftCacheStrapper(p3AuthMgr *am, p3ConnectMgr *cm)
/* overloaded search function */
bool ftCacheStrapper::search(std::string hash, uint64_t size, uint32_t hintflags, FileInfo &info) const
{
#ifdef DB_DEBUG
std::cerr << "ftCacheStrapper::search(" << hash << "," << size << "," << hintflags;
std::cerr << ")";
std::cerr << std::endl;
#endif
CacheData data;
if (findCache(hash, data))
{
#ifdef DB_DEBUG
std::cerr << "ftCacheStrapper::search() found: ";
std::cerr << data.path << "/" << data.name;
std::cerr << " = " << data.hash << "," << data.size;
std::cerr << std::endl;
#endif
/* ... */
info.size = data.size;
info.fname = data.name;

52
libretroshare/src/ft/ftfilecreator.cc
View File

@ -1,6 +1,9 @@
#include "ftfilecreator.h"
#include <errno.h>
#define FILE_DEBUG 1
/*******
* #define FILE_DEBUG 1
******/
#define CHUNK_MAX_AGE 30
@ -97,9 +100,15 @@ bool ftFileCreator::addFileData(uint64_t offset, uint32_t chunk_size, void *data
/*
* add the data
*/
//void *data2 = malloc(chunk_size);
//std::cerr << "data2: " << data2 << std::endl;
//if (1 != fwrite(data2, chunk_size, 1, this->fd))
if (1 != fwrite(data, chunk_size, 1, this->fd))
{
std::cerr << "ftFileCreator::addFileData() Bad fwrite" << std::endl;
std::cerr << "ERRNO: " << errno << std::endl;
return 0;
}
@ -127,6 +136,7 @@ int ftFileCreator::initializeFileAttrs()
{
std::cerr << "ftFileCreator::initializeFileAttrs() Filename: ";
std::cerr << file_name;
std::cerr << std::endl;
/*
* check if the file exists
@ -143,8 +153,8 @@ int ftFileCreator::initializeFileAttrs()
return 1;
{
std::cout <<
"ftFileCreator::initializeFileAttrs() Filename: " << file_name;
std::cerr << "ftFileCreator::initializeFileAttrs() opening w+b";
std::cerr << std::endl;
}
/*
@ -154,8 +164,8 @@ int ftFileCreator::initializeFileAttrs()
fd = fopen(file_name.c_str(), "w+b");
if (!fd)
{
std::cout <<
"ftFileCreator::initializeFileAttrs() Failed to open (w+b): "<< file_name << std::endl;
std::cerr << "ftFileCreator::initializeFileAttrs()";
std::cerr << " Failed to open (w+b): "<< file_name << std::endl;
return 0;
}
@ -192,10 +202,15 @@ int ftFileCreator::notifyReceived(uint64_t offset, uint32_t chunk_size)
/* find the chunk */
std::map<uint64_t, ftChunk>::iterator it;
it = mChunks.find(offset);
bool isFirst = false;
if (it == mChunks.end())
{
return 0; /* ignoring */
}
else if (it == mChunks.begin())
{
isFirst = true;
}
ftChunk chunk = it->second;
mChunks.erase(it);
@ -208,21 +223,26 @@ int ftFileCreator::notifyReceived(uint64_t offset, uint32_t chunk_size)
mChunks[chunk.offset] = chunk;
}
/* if we've cleaned up chunks, must update counters */
it = mChunks.find(chunk.offset);
if (it == mChunks.end())
/* update how much has been completed */
if (isFirst)
{
mStart = offset + chunk_size;
}
if (mChunks.size() == 0)
{
mStart = mEnd;
}
else if (it == mChunks.begin())
{
mStart += it->second.offset;
}
/* otherwise there is another earlier block to go
*/
return 1;
}
/* Returns true if more to get
* But can return size = 0, if we are still waiting for the data.
*/
bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
{
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
@ -271,8 +291,12 @@ bool ftFileCreator::getMissingChunk(uint64_t &offset, uint32_t &chunk)
offset = mEnd;
mEnd += chunk;
if (chunk > 0)
{
mChunks[offset] = ftChunk(offset, chunk, ts);
return true;
}
return true; /* cos more data to get */
}
/***********************************************************

1
libretroshare/src/ft/ftfilecreator.h
View File

@ -46,6 +46,7 @@ public:
/* overloaded from FileProvider */
virtual bool getFileData(uint64_t offset, uint32_t chunk_size, void *data);
bool finished() { return getRecvd() == getFileSize(); }
uint64_t getRecvd();

31
libretroshare/src/ft/ftfilecreatortest.cc
View File

@ -36,7 +36,7 @@ int test_timeout(ftFileCreator *creator)
creator->getMissingChunk(offset, chunk);
std::cerr << "Allocated Offset: " << offset << " chunk: " << chunk << std::endl;
CHECK(offset < max_offset);
CHECK(offset <= max_offset);
sleep(1);
}
@ -45,11 +45,11 @@ int test_timeout(ftFileCreator *creator)
for(i = 0; i < max_timeout; i++)
{
sleep(1);
creator->getMissingChunk(offset, chunk);
std::cerr << "Allocated Offset: " << offset << " chunk: " << chunk << std::endl;
CHECK(offset < max_offset);
sleep(1);
CHECK(offset <= max_offset);
}
REPORT("Chunk Queue");
@ -69,10 +69,33 @@ int test_fill(ftFileCreator *creator)
while(creator->getMissingChunk(offset, chunk))
{
/* give it too them */
if (chunk == 0)
{
std::cerr << "Missing Data already Alloced... wait";
std::cerr << std::endl;
sleep(1);
chunk = 1000;
continue;
}
/* give it to them */
void *data = malloc(chunk);
/* fill with ascending numbers */
for(int i = 0; i < chunk; i++)
{
((uint8_t *) data)[i] = 'a' + i % 27;
if (i % 27 == 26)
{
((uint8_t *) data)[i] = '\n';
}
}
creator->addFileData(offset, chunk, data);
free(data);
usleep(250000); /* 1/4 of sec */
chunk = 1000; /* reset chunk size */
}
uint64_t end_size = creator->getFileSize();

13
libretroshare/src/ft/ftfileprovider.cc
View File

@ -79,7 +79,8 @@ bool ftFileProvider::getFileData(uint64_t offset, uint32_t chunk_size, void *dat
*/
fseek(fd, base_loc, SEEK_SET);
void *data = malloc(chunk_size);
// Data space allocated by caller.
//void *data = malloc(chunk_size);
/*
* read the data
@ -117,7 +118,7 @@ int ftFileProvider::initializeFileAttrs()
{
std::cerr << "ftFileProvider::initializeFileAttrs() Filename: ";
std::cerr << file_name;
std::cerr << std::endl;
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/
if (fd)
@ -128,8 +129,8 @@ int ftFileProvider::initializeFileAttrs()
*/
{
std::cout <<
"ftFileProvider::initializeFileAttrs() Filename: " << file_name;
std::cerr << "ftFileProvider::initializeFileAttrs() trying (r+b) ";
std::cerr << std::endl;
}
/*
@ -139,8 +140,8 @@ int ftFileProvider::initializeFileAttrs()
fd = fopen(file_name.c_str(), "r+b");
if (!fd)
{
std::cout <<
"ftFileProvider::initializeFileAttrs() Failed to open (r+b): "<< file_name << std::endl;
std::cerr << "ftFileProvider::initializeFileAttrs() Failed to open (r+b): ";
std::cerr << file_name << std::endl;
return 0;
}

19
libretroshare/src/ft/ftfileprovidertest.cc
View File

@ -21,6 +21,14 @@ int main()
/* use creator to make it */
void *data = malloc(max_chunk);
for(int i = 0; i < max_chunk; i++)
{
((uint8_t *) data)[i] = 'a' + i % 27;
if (i % 27 == 26)
{
((uint8_t *) data)[i] = '\n';
}
}
ftFileCreator *creator = new ftFileCreator(filename, size, "hash", 0);
for(offset = 0; offset != size; offset += chunk)
@ -45,6 +53,17 @@ int main()
while(creator->getMissingChunk(offset, chunk))
{
if (chunk == 0)
{
std::cerr << "All currently allocated .... waiting";
std::cerr << std::endl;
sleep(1);
/* reset chunk size */
chunk = (uint64_t) max_chunk * (rand() / (1.0 + RAND_MAX));
std::cerr << "ChunkSize = " << chunk << std::endl;
continue;
}
if (!provider->getFileData(offset, chunk, data))
{
FAILED("Read from Test Data File");

38
libretroshare/src/ft/ftserver.cc
View File

@ -87,6 +87,14 @@ void ftServer::addConfigComponents(p3ConfigMgr *mgr)
/* NOT SURE ABOUT THIS ONE */
}
std::string ftServer::OwnId()
{
std::string ownId;
if (mConnMgr)
ownId = mConnMgr->getOwnId();
return ownId;
}
/* Final Setup (once everything is assigned) */
void ftServer::SetupFtServer(NotifyBase *cb)
{
@ -189,7 +197,7 @@ void ftServer::run()
/********************** Controller Access **********************/
/***************************************************************/
bool ftServer::FileRequest(std::string fname, std::string hash, uint32_t size,
bool ftServer::FileRequest(std::string fname, std::string hash, uint64_t size,
std::string dest, uint32_t flags, std::list<std::string> srcIds)
{
return mFtController->FileRequest(fname, hash, size,
@ -271,7 +279,7 @@ bool ftServer::FileDetails(std::string hash, uint32_t hintflags, FileInfo &info)
/******************* ExtraFileList Access **********************/
/***************************************************************/
bool ftServer::ExtraFileAdd(std::string fname, std::string hash, uint32_t size,
bool ftServer::ExtraFileAdd(std::string fname, std::string hash, uint64_t size,
uint32_t period, uint32_t flags)
{
return mFtExtra->addExtraFile(fname, hash, size, period, flags);
@ -456,6 +464,14 @@ bool ftServer::sendData(std::string peerId, std::string hash, uint64_t size,
uint64_t offset = 0;
uint32_t chunk;
#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 << " data: " << data;
std::cerr << std::endl;
#endif
while(tosend > 0)
{
@ -487,6 +503,17 @@ bool ftServer::sendData(std::string peerId, std::string hash, uint64_t size,
rfd->fd.binData.setBinData(
&(((uint8_t *) data)[offset]), chunk);
/* print the data pointer */
#ifdef SERVER_DEBUG
std::cerr << "ftServer::sendData() Packet: " << std::endl;
std::cerr << " offset: " << rfd->fd.file_offset;
std::cerr << " chunk: " << chunk;
std::cerr << " len: " << rfd->fd.binData.bin_len;
std::cerr << " data: " << rfd->fd.binData.bin_data;
std::cerr << std::endl;
#endif
mP3iface->SendFileData(rfd);
offset += chunk;
@ -679,6 +706,11 @@ FileInfo(ffr);
{
#ifdef SERVER_DEBUG
std::cerr << "ftServer::handleFileData() Recvd ftFiler Data" << std::endl;
std::cerr << "hash: " << fd->fd.file.hash;
std::cerr << " length: " << fd->fd.binData.bin_len;
std::cerr << " data: " << fd->fd.binData.bin_data;
std::cerr << std::endl;
std::ostringstream out;
if (i == i_init)
{
@ -698,7 +730,7 @@ FileInfo(ffr);
/* we've stolen the data part -> so blank before delete
*/
fd->fd.TlvShallowClear();
fd->fd.binData.TlvShallowClear();
delete fd;
}

5
libretroshare/src/ft/ftserver.h
View File

@ -89,6 +89,7 @@ void addConfigComponents(p3ConfigMgr *mgr);
CacheStrapper *getCacheStrapper();
CacheTransfer *getCacheTransfer();
std::string OwnId();
/* Final Setup (once everything is assigned) */
//void SetupFtServer();
@ -107,7 +108,7 @@ virtual void run();
/***
* Control of Downloads
***/
virtual bool FileRequest(std::string fname, std::string hash, uint32_t size,
virtual bool FileRequest(std::string fname, std::string hash, uint64_t size,
std::string dest, uint32_t flags, std::list<std::string> srcIds);
virtual bool FileCancel(std::string hash);
virtual bool FileControl(std::string hash, uint32_t flags);
@ -123,7 +124,7 @@ virtual bool FileDetails(std::string hash, uint32_t hintflags, FileInfo &info);
/***
* Extra List Access
***/
virtual bool ExtraFileAdd(std::string fname, std::string hash, uint32_t size,
virtual bool ExtraFileAdd(std::string fname, std::string hash, uint64_t size,
uint32_t period, uint32_t flags);
virtual bool ExtraFileRemove(std::string hash, uint32_t flags);
virtual bool ExtraFileHash(std::string localpath,

10
libretroshare/src/ft/ftserver1test.cc
View File

@ -214,6 +214,12 @@ int main(int argc, char **argv)
std::string cachepath = configpath + "/cache";
RsDirUtil::checkCreateDirectory(cachepath);
std::string partialspath = configpath + "/partials";
RsDirUtil::checkCreateDirectory(partialspath);
std::string downloadpath = configpath + "/downloads";
RsDirUtil::checkCreateDirectory(downloadpath);
std::string localpath = cachepath + "/local";
RsDirUtil::checkCreateDirectory(localpath);
@ -222,7 +228,7 @@ int main(int argc, char **argv)
server->setConfigDirectory(configpath);
sleep(60);
//sleep(60);
NotifyBase *base = NULL;
server->SetupFtServer(base);
@ -231,6 +237,8 @@ int main(int argc, char **argv)
server->StartupThreads();
/* setup any extra bits */
server->setPartialsDirectory(partialspath);
server->setDownloadDirectory(downloadpath);
server->setSharedDirectories(fileList);
}

11
libretroshare/src/ft/ftserver2test.cc
View File

@ -25,6 +25,7 @@
/*
* ftServer2Test - Demonstrates how to check for test stuff.
* This tests hashing of files using extraList.
*/
#ifdef WIN32
@ -239,6 +240,12 @@ int main(int argc, char **argv)
std::string cachepath = configpath + "/cache";
RsDirUtil::checkCreateDirectory(cachepath);
std::string partialspath = configpath + "/partials";
RsDirUtil::checkCreateDirectory(partialspath);
std::string downloadpath = configpath + "/downloads";
RsDirUtil::checkCreateDirectory(downloadpath);
std::string localpath = cachepath + "/local";
RsDirUtil::checkCreateDirectory(localpath);
@ -254,6 +261,10 @@ int main(int argc, char **argv)
server->StartupThreads();
/* setup any extra bits */
/* everyone gets download directories */
server->setPartialsDirectory(partialspath);
server->setDownloadDirectory(downloadpath);
if (loadAll)
{
server->setSharedDirectories(fileList);

515
libretroshare/src/ft/ftserver3test.cc Normal file
View File

@ -0,0 +1,515 @@
/*
* libretroshare/src/ft: ftserver3test.cc
*
* File Transfer for RetroShare.
*
* Copyright 2008 by Robert Fernie.
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License Version 2 as published by the Free Software Foundation.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
* USA.
*
* Please report all bugs and problems to "retroshare@lunamutt.com".
*
*/
/*
* ftServer3Test - Test of the file transfer from a server level.
* Steps:
* 1) load shared directories into others, and let them be
transferred between clients.
* 2) search for local item on others.
* 3) request item on load server.
should transfer from all others simultaneously.
*/
#ifdef WIN32
#include "util/rswin.h"
#endif
#include "rsiface/rsexpr.h"
#include "ft/ftserver.h"
#include "ft/ftextralist.h"
#include "ft/ftdatamultiplex.h"
#include "ft/ftfilesearch.h"
#include "pqi/p3authmgr.h"
#include "pqi/p3connmgr.h"
#include "util/rsdebug.h"
#include "ft/pqitestor.h"
#include "util/rsdir.h"
#include "util/utest.h"
#include <sstream>
class TestData
{
public:
ftServer *loadServer;
std::list<ftServer *> otherServers;
std::list<std::string> extraList;
};
extern "C" void *do_server_test_thread(void *p);
void usage(char *name)
{
std::cerr << "Usage: " << name << " [-soa] [-p <peerId>] [-d <debugLvl>] [-e <extrafile>] [<path> [<path2> ... ]] ";
std::cerr << std::endl;
}
int main(int argc, char **argv)
{
int c;
uint32_t period = 1;
uint32_t debugLevel = 5;
bool debugStderr = true;
bool loadAll = false;
bool loadOthers = false;
std::list<std::string> fileList;
std::list<std::string> extraList;
std::list<std::string> peerIds;
std::map<std::string, ftServer *> mFtServers;
std::map<std::string, p3ConnectMgr *> mConnMgrs;
ftServer *mLoadServer = NULL;
std::list<ftServer *> mOtherServers;
std::list<std::string>::iterator eit;
#ifdef PTW32_STATIC_LIB
pthread_win32_process_attach_np();
#endif
#ifdef WIN32
// Windows Networking Init.
WORD wVerReq = MAKEWORD(2,2);
WSADATA wsaData;
if (0 != WSAStartup(wVerReq, &wsaData))
{
std::cerr << "Failed to Startup Windows Networking";
std::cerr << std::endl;
}
else
{
std::cerr << "Started Windows Networking";
std::cerr << std::endl;
}
#endif
while(-1 != (c = getopt(argc, argv, "aosd:p:e:")))
{
switch (c)
{
case 'p':
peerIds.push_back(optarg);
break;
case 'd':
debugLevel = atoi(optarg);
break;
case 's':
debugStderr = true;
break;
case 'e':
extraList.push_back(optarg);
break;
case 'a':
loadAll = true;
break;
case 'o':
loadOthers = true;
break;
default:
usage(argv[0]);
break;
}
}
/* do logging */
setOutputLevel(debugLevel);
if (optind >= argc)
{
std::cerr << "Missing Shared Directories" << std::endl;
usage(argv[0]);
}
for(; optind < argc; optind++)
{
std::cerr << "Adding: " << argv[optind] << std::endl;
fileList.push_back(std::string(argv[optind]));
}
/* We need to setup a series 2 - 4 different ftServers....
*
* Each one needs:
*
*
* A List of peerIds...
*/
std::list<std::string>::const_iterator it, jit;
std::list<pqiAuthDetails> baseFriendList, friendList;
std::list<pqiAuthDetails>::iterator fit;
P3Hub *testHub = new P3Hub();
testHub->start();
/* Setup Base Friend Info */
for(it = peerIds.begin(); it != peerIds.end(); it++)
{
pqiAuthDetails pad;
pad.id = *it;
pad.name = *it;
pad.trustLvl = 5;
pad.ownsign = true;
pad.trusted = false;
baseFriendList.push_back(pad);
std::cerr << "ftserver1test::setup peer: " << *it;
std::cerr << std::endl;
}
std::ostringstream pname;
pname << "/tmp/rstst-" << time(NULL);
std::string basepath = pname.str();
RsDirUtil::checkCreateDirectory(basepath);
for(it = peerIds.begin(); it != peerIds.end(); it++)
{
friendList = baseFriendList;
/* remove current one */
for(fit = friendList.begin(); fit != friendList.end(); fit++)
{
if (fit->id == *it)
{
friendList.erase(fit);
break;
}
}
p3AuthMgr *authMgr = new p3DummyAuthMgr(*it, friendList);
p3ConnectMgr *connMgr = new p3ConnectMgr(authMgr);
mConnMgrs[*it] = connMgr;
for(fit = friendList.begin(); fit != friendList.end(); fit++)
{
/* add as peer to authMgr */
connMgr->addFriend(fit->id);
}
P3Pipe *pipe = new P3Pipe(); //(*it);
/* add server */
ftServer *server;
server = new ftServer(authMgr, connMgr);
mFtServers[*it] = server;
bool isOther;
if (!mLoadServer)
{
mLoadServer = server;
isOther = false;
}
else
{
mOtherServers.push_back(server);
isOther = true;
}
server->setP3Interface(pipe);
std::string configpath = basepath + "/" + *it;
RsDirUtil::checkCreateDirectory(configpath);
std::string cachepath = configpath + "/cache";
RsDirUtil::checkCreateDirectory(cachepath);
std::string partialspath = configpath + "/partials";
RsDirUtil::checkCreateDirectory(partialspath);
std::string downloadpath = configpath + "/downloads";
RsDirUtil::checkCreateDirectory(downloadpath);
std::string localpath = cachepath + "/local";
RsDirUtil::checkCreateDirectory(localpath);
std::string remotepath = cachepath + "/remote";
RsDirUtil::checkCreateDirectory(remotepath);
server->setConfigDirectory(configpath);
NotifyBase *base = NULL;
server->SetupFtServer(base);
testHub->addP3Pipe(*it, pipe, connMgr);
server->StartupThreads();
/* setup any extra bits */
server->setPartialsDirectory(partialspath);
server->setDownloadDirectory(downloadpath);
if ((loadAll) || (isOther && loadOthers))
{
server->setSharedDirectories(fileList);
for(eit = extraList.begin(); eit != extraList.end(); eit++)
{
server->ExtraFileHash(*eit, 3600, 0);
}
}
}
if ((mLoadServer) && (!loadOthers))
{
mLoadServer->setSharedDirectories(fileList);
for(eit = extraList.begin(); eit != extraList.end(); eit++)
{
mLoadServer->ExtraFileHash(*eit, 3600, 0);
}
}
/* stick your real test here */
std::map<std::string, ftServer *>::iterator sit;
std::map<std::string, p3ConnectMgr *>::iterator cit;
/* Start up test thread */
pthread_t tid;
TestData *mFt = new TestData;
/* set data */
mFt->loadServer = mLoadServer;
mFt->otherServers = mOtherServers;
mFt->extraList = extraList;
void *data = (void *) mFt;
pthread_create(&tid, 0, &do_server_test_thread, data);
pthread_detach(tid); /* so memory is reclaimed in linux */
while(1)
{
std::cerr << "ftserver2test::sleep()";
std::cerr << std::endl;
sleep(1);
/* tick the connmgrs */
for(sit = mFtServers.begin(); sit != mFtServers.end(); sit++)
{
/* update */
(sit->second)->tick();
}
for(cit = mConnMgrs.begin(); cit != mConnMgrs.end(); cit++)
{
/* update */
(cit->second)->tick();
}
}
}
/* So our actual test can run here.....
*
*/
INITTEST();
void *do_server_test_thread(void *data)
{
TestData *mFt = (TestData *) data;
std::cerr << "do_server_test_thread() running";
std::cerr << std::endl;
/* search Others for a suitable file
* (Tests GUI search functionality)
*/
if (mFt->otherServers.size() < 1)
{
std::cerr << "no Other Servers to search on";
std::cerr << std::endl;
exit(1);
return NULL;
}
for(int i = 0; i < 60; i++)
{
std::cerr << "Waiting 60 seconds to share caches";
std::cerr << std::endl;
sleep(1);
}
ftServer *oServer = *(mFt->otherServers.begin());
std::string oId = oServer->OwnId();
/* create Expression */
uint64_t minFileSize = 10000;
//SizeExpression se(Greater, minFileSize);
SizeExpression se(Smaller, minFileSize);
Expression *expr = &se;
std::list<FileDetail> results;
std::list<FileDetail>::iterator it;
oServer->SearchBoolExp(expr, results);
if (results.size() < 1)
{
std::cerr << "no Shared Files > " << minFileSize;
std::cerr << std::endl;
exit(1);
return NULL;
}
/* find the first remote entry */
FileDetail sFile;
bool foundFile = false;
for(it = results.begin();
(it != results.end()); it++)
{
std::cerr << "Shared File: " << it->name;
std::cerr << std::endl;
if (!foundFile)
{
if (it->id != mFt->loadServer->OwnId())
{
std::cerr << "Selected: " << it->name;
std::cerr << std::endl;
foundFile = true;
sFile = *it;
}
else
{
std::cerr << "LoadId: ";
std::cerr << mFt->loadServer->OwnId();
std::cerr << "FileId: ";
std::cerr << it->id;
std::cerr << std::endl;
}
}
}
if (!foundFile)
{
std::cerr << "Not Found Suitable File";
std::cerr << std::endl;
}
/*** Now Download it! ***/
std::list<std::string> srcIds;
//srcIds.push_back(sFile.id);
srcIds.push_back(oId);
if (foundFile)
{
mFt->loadServer->FileRequest(sFile.name, sFile.hash,
sFile.size, "", 0, srcIds);
}
/* Give it a while to transfer */
for(int i = 0; i < 10; i++)
{
std::cerr << "Waiting 10 seconds to transfer";
std::cerr << std::endl;
sleep(10);
}
#if 0
bool
while(!mFt->loadServer->ExtraFileStatus(*eit, info))
{
/* max of 30 seconds */
now = time(NULL);
if (now - start > 30)
{
/* FAIL */
REPORT2( false, "Extra File Hashing");
}
sleep(1);
}
/* Got ExtraFileStatus */
REPORT("Successfully Found ExtraFile");
/* now we can try a search (should succeed) */
uint32_t hintflags = 0;
if (mFt->loadServer->FileDetails(info.hash, hintflags, info2))
{
CHECK(info2.hash == info.hash);
CHECK(info2.size == info.size);
CHECK(info2.fname == info.fname);
}
else
{
REPORT2( false, "Search for Extra File (Basic)");
}
/* search with flags (should succeed) */
hintflags = RS_FILE_HINTS_EXTRA;
if (mFt->loadServer->FileDetails(info.hash, hintflags, info2))
{
CHECK(info2.hash == info.hash);
CHECK(info2.size == info.size);
CHECK(info2.fname == info.fname);
}
else
{
REPORT2( false, "Search for Extra File (Extra Flag)");
}
/* search with other flags (should fail) */
hintflags = RS_FILE_HINTS_REMOTE | RS_FILE_HINTS_SPEC_ONLY;
if (mFt->loadServer->FileDetails(info.hash, hintflags, info2))
{
REPORT2( false, "Search for Extra File (Fail Flags)");
}
else
{
REPORT("Search for Extra File (Fail Flags)");
}
/* if we try to download it ... should just find existing one*/
REPORT("Testing with Extra File");
}
#endif
FINALREPORT("Shared Directories, Bool Search, multi-source transfers");
exit(1);
}

48
libretroshare/src/ft/fttransfermodule.cc
View File

@ -23,7 +23,9 @@
*
*/
#define FT_DEBUG 1
/*******
* #define FT_DEBUG 1
******/
#include "fttransfermodule.h"
@ -133,6 +135,7 @@ bool ftTransferModule::recvFileData(std::string peerId, uint64_t offset,
std::cerr << " peerId: " << peerId;
std::cerr << " offset: " << offset;
std::cerr << " chunksize: " << chunk_size;
std::cerr << " data: " << data;
std::cerr << std::endl;
#endif
@ -271,33 +274,21 @@ bool ftTransferModule::queryInactive()
{
//Peer side has change from online to offline during transfer
case PQIPEER_NOT_ONLINE:
/*
if (ts - (pInfo->lastTS) > PQIPEER_OFFLINE_CHECK)
{//start to request data
req_size = TRANSFER_START_MIN;
if (getChunk(req_offset,req_size))
{
pInfo->offset = req_offset;
pInfo->chunkSize = req_size;
pInfo->lastTS = ts;
pInfo->state = PQIPEER_DOWNLOADING;
requestData(peerId, req_offset,req_size);
}
else mFlag = 1;
}
*/
break;
//file request has been sent to peer side, but no response received yet
case PQIPEER_DOWNLOADING:
if (ts - (pInfo->lastTS) > PQIPEER_DOWNLOAD_CHECK)
requestData(peerId, pInfo->offset,pInfo->chunkSize); //give a push
if (ts - (pInfo->lastTS) < PQIPEER_DOWNLOAD_CHECK)
{
/* if not timed out yet.... ignore */
actualRate += pInfo->actualRate;
break;
}
//file response has been received or peer side is just ready for download
/* otherwise fall through to request it again (with getChunk);
*/
//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)
@ -310,13 +301,22 @@ bool ftTransferModule::queryInactive()
}
if (getChunk(req_offset,req_size))
{
if (req_size > 0)
{
pInfo->offset = req_offset;
pInfo->chunkSize = req_size;
pInfo->lastTS = ts;
pInfo->state = PQIPEER_DOWNLOADING;
pInfo->receivedSize = 0;
requestData(peerId,req_offset,req_size);
}
else
{
std::cerr << "transfermodule::Waiting for data to be available";
std::cerr << std::endl;
}
}
else mFlag = 1;
actualRate += pInfo->actualRate;
@ -371,7 +371,11 @@ bool ftTransferModule::cancelTransfer()
bool ftTransferModule::completeFileTransfer()
{
//mFtController->completeFile(mHash);
std::cerr << "ftTransferModule::completeFileTransfer()";
std::cerr << std::endl;
if (mFtController)
mFtController->FlagFileComplete(mHash);
return true;
}

22
libretroshare/src/ft/pqitestor.cc
View File

@ -26,6 +26,9 @@
#include "ft/pqitestor.h"
#include "pqi/p3connmgr.h"
/******
*#define HUB_DEBUG
*****/
P3Hub::P3Hub()
{
@ -46,35 +49,45 @@ void P3Hub::addP3Pipe(std::string id, P3Pipe *pqi, p3ConnectMgr *mgr)
mPeers[id] = item;
/* tell all the other peers we are connected */
#ifdef HUB_DEBUG
std::cerr << "P3Hub::addPQIPipe()";
std::cerr << std::endl;
#endif
}
void P3Hub::run()
{
#ifdef HUB_DEBUG
std::cerr << "P3Hub::run()";
std::cerr << std::endl;
#endif
RsItem *item;
std::list<std::pair<std::string, RsItem *> > recvdQ;
std::list<std::pair<std::string, RsItem *> >::iterator lit;
while(1)
{
#ifdef HUB_DEBUG
std::cerr << "P3Hub::run()";
std::cerr << std::endl;
#endif
std::map<std::string, hubItem>::iterator it;
for(it = mPeers.begin(); it != mPeers.end(); it++)
{
while (NULL != (item = it->second.mPQI->PopSentItem()))
{
#ifdef HUB_DEBUG
std::cerr << "P3Hub::run() recvd msg from: ";
std::cerr << it->first;
std::cerr << " for " << item->PeerId();
std::cerr << std::endl;
item->print(std::cerr, 10);
std::cerr << std::endl;
#endif
recvdQ.push_back(make_pair(it->first, item));
@ -88,10 +101,12 @@ void P3Hub::run()
std::string destId = (lit->second)->PeerId();
if (mPeers.end() == (it = mPeers.find(destId)))
{
#ifdef HUB_DEBUG
std::cerr << "Failed to Find destination: " << destId;
std::cerr << std::endl;
std::cerr << "Deleting Packet";
std::cerr << std::endl;
#endif
delete (lit->second);
@ -100,11 +115,14 @@ void P3Hub::run()
{
/* now we have dest, set source Id */
(lit->second)->PeerId(srcId);
std::cerr << "P3Hub::run() sending msg to: ";
std::cerr << it->first;
#ifdef HUB_DEBUG
std::cerr << "P3Hub::run() sending msg from: ";
std::cerr << srcId << "to: ";
std::cerr << destId;
std::cerr << std::endl;
(lit->second)->print(std::cerr, 10);
std::cerr << std::endl;
#endif
(it->second).mPQI->PushRecvdItem(lit->second);
}