mirror of
https://github.com/RetroShare/RetroShare.git
synced 2024-10-01 02:35:48 -04:00
3a82fffd0e
- removed unused RsFileTransferCacheRequestItem and all related functions - fixed bug in turtle router calling members of a deleted item - removed some debug info git-svn-id: http://svn.code.sf.net/p/retroshare/code/branches/v0.6-initdev@6789 b45a01b8-16f6-495d-af2f-9b41ad6348cc
1022 lines
26 KiB
C++
1022 lines
26 KiB
C++
/*
|
|
* "$Id: pqistreamer.cc,v 1.19 2007-02-18 21:46:50 rmf24 Exp $"
|
|
*
|
|
* 3P/PQI network interface for RetroShare.
|
|
*
|
|
* Copyright 2004-2006 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".
|
|
*
|
|
*/
|
|
|
|
|
|
#include <iostream>
|
|
#include <fstream>
|
|
#include "util/rsdebug.h"
|
|
#include "util/rsstring.h"
|
|
|
|
#include "pqi/pqistreamer.h"
|
|
#include "pqi/pqinotify.h"
|
|
|
|
#include "serialiser/rsserial.h"
|
|
|
|
const int pqistreamerzone = 8221;
|
|
|
|
const int PQISTREAM_ABS_MAX = 100000000; /* 100 MB/sec (actually per loop) */
|
|
|
|
/* This removes the print statements (which hammer pqidebug) */
|
|
/***
|
|
#define RSITEM_DEBUG 1
|
|
#define DEBUG_TRANSFERS 1
|
|
#define DEBUG_PQISTREAMER 1
|
|
***/
|
|
|
|
#ifdef DEBUG_TRANSFERS
|
|
#include "util/rsprint.h"
|
|
#endif
|
|
|
|
|
|
pqistreamer::pqistreamer(RsSerialiser *rss, std::string id, BinInterface *bio_in, int bio_flags_in)
|
|
:PQInterface(id), mStreamerMtx("pqistreamer"),
|
|
mBio(bio_in), mBio_flags(bio_flags_in), mRsSerialiser(rss),
|
|
mPkt_wpending(NULL),
|
|
mTotalRead(0), mTotalSent(0),
|
|
mCurrRead(0), mCurrSent(0),
|
|
mAvgReadCount(0), mAvgSentCount(0)
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
mAvgLastUpdate = mCurrReadTS = mCurrSentTS = time(NULL);
|
|
|
|
/* allocated once */
|
|
mPkt_rpend_size = getRsPktMaxSize();
|
|
mPkt_rpending = malloc(mPkt_rpend_size);
|
|
mReading_state = reading_state_initial ;
|
|
|
|
// avoid uninitialized (and random) memory read.
|
|
memset(mPkt_rpending,0,mPkt_rpend_size) ;
|
|
|
|
// 100 B/s (minimal)
|
|
setMaxRate(true, 0.1);
|
|
setMaxRate(false, 0.1);
|
|
setRate(true, 0);
|
|
setRate(false, 0);
|
|
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::pqistreamer() Initialisation!");
|
|
|
|
if (!bio_in)
|
|
{
|
|
pqioutput(PQL_ALERT, pqistreamerzone, "pqistreamer::pqistreamer() NULL bio, FATAL ERROR!");
|
|
exit(1);
|
|
}
|
|
|
|
mFailed_read_attempts = 0; // reset failed read, as no packet is still read.
|
|
|
|
return;
|
|
}
|
|
|
|
pqistreamer::~pqistreamer()
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::~pqistreamer() Destruction!");
|
|
|
|
if (mBio_flags & BIN_FLAGS_NO_CLOSE)
|
|
{
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::~pqistreamer() Not Closing BinInterface!");
|
|
}
|
|
else if (mBio)
|
|
{
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::~pqistreamer() Deleting BinInterface!");
|
|
|
|
delete mBio;
|
|
}
|
|
|
|
/* clean up serialiser */
|
|
if (mRsSerialiser)
|
|
delete mRsSerialiser;
|
|
|
|
// clean up outgoing. (cntrl packets)
|
|
locked_clear_out_queue() ;
|
|
|
|
if (mPkt_wpending)
|
|
{
|
|
free(mPkt_wpending);
|
|
mPkt_wpending = NULL;
|
|
}
|
|
|
|
free(mPkt_rpending);
|
|
|
|
// clean up incoming.
|
|
while(mIncoming.size() > 0)
|
|
{
|
|
RsItem *i = mIncoming.front();
|
|
mIncoming.pop_front();
|
|
delete i;
|
|
}
|
|
return;
|
|
}
|
|
|
|
|
|
// Get/Send Items.
|
|
int pqistreamer::SendItem(RsItem *si,uint32_t& out_size)
|
|
{
|
|
#ifdef RSITEM_DEBUG
|
|
{
|
|
std::string out = "pqistreamer::SendItem():\n";
|
|
si -> print_string(out);
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
std::cerr << out;
|
|
}
|
|
#endif
|
|
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
return queue_outpqi_locked(si,out_size);
|
|
}
|
|
|
|
RsItem *pqistreamer::GetItem()
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::GetItem()");
|
|
#endif
|
|
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
if(mIncoming.empty())
|
|
return NULL;
|
|
|
|
RsItem *osr = mIncoming.front() ;
|
|
mIncoming.pop_front() ;
|
|
|
|
return osr;
|
|
}
|
|
|
|
// // PQInterface
|
|
int pqistreamer::tick()
|
|
{
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
std::string out = "pqistreamer::tick()\n" + PeerId();
|
|
rs_sprintf_append(out, ": currRead/Sent: %d/%d", mCurrRead, mCurrSent);
|
|
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
if (!tick_bio())
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
tick_recv(0);
|
|
tick_send(0);
|
|
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
#ifdef DEBUG_PQISTREAMER
|
|
/* give details of the packets */
|
|
{
|
|
std::list<void *>::iterator it;
|
|
|
|
std::string out = "pqistreamer::tick() Queued Data: for " + PeerId();
|
|
|
|
if (mBio->isactive())
|
|
{
|
|
out += " (active)";
|
|
}
|
|
else
|
|
{
|
|
out += " (waiting)";
|
|
}
|
|
out += "\n";
|
|
|
|
{
|
|
int total = locked_compute_out_pkt_size() ;
|
|
|
|
rs_sprintf_append(out, "\t Out Packets [%d] => %d bytes\n", locked_out_queue_size(), total);
|
|
rs_sprintf_append(out, "\t Incoming [%d]\n", mIncoming.size());
|
|
}
|
|
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
/* if there is more stuff in the queues */
|
|
if ((mIncoming.size() > 0) || (locked_out_queue_size() > 0))
|
|
{
|
|
return 1;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int pqistreamer::tick_bio()
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
mBio->tick();
|
|
|
|
/* short circuit everything is bio isn't active */
|
|
if (!(mBio->isactive()))
|
|
{
|
|
return 0;
|
|
}
|
|
return 1;
|
|
}
|
|
|
|
|
|
int pqistreamer::tick_recv(uint32_t timeout)
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
if (mBio->moretoread(timeout))
|
|
{
|
|
handleincoming_locked();
|
|
}
|
|
return 1;
|
|
}
|
|
|
|
|
|
int pqistreamer::tick_send(uint32_t timeout)
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
/* short circuit everything is bio isn't active */
|
|
if (!(mBio->isactive()))
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
if (mBio->cansend(timeout))
|
|
{
|
|
handleoutgoing_locked();
|
|
}
|
|
return 1;
|
|
}
|
|
|
|
int pqistreamer::status()
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::status()");
|
|
|
|
if (mBio->isactive())
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "Data in:%d out:%d", mTotalRead, mTotalSent);
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
return 0;
|
|
}
|
|
|
|
void pqistreamer::locked_storeInOutputQueue(void *ptr,int)
|
|
{
|
|
mOutPkts.push_back(ptr);
|
|
}
|
|
//
|
|
/**************** HANDLE OUTGOING TRANSLATION + TRANSMISSION ******/
|
|
|
|
int pqistreamer::queue_outpqi_locked(RsItem *pqi,uint32_t& pktsize)
|
|
{
|
|
pktsize = 0 ;
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "pqistreamer::queue_outpqi() called." << std::endl;
|
|
#endif
|
|
|
|
|
|
/* decide which type of packet it is */
|
|
|
|
pktsize = mRsSerialiser->size(pqi);
|
|
void *ptr = malloc(pktsize);
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "pqistreamer::queue_outpqi() serializing packet with packet size : " << pktsize << std::endl;
|
|
#endif
|
|
if (mRsSerialiser->serialise(pqi, ptr, &pktsize))
|
|
{
|
|
locked_storeInOutputQueue(ptr,pqi->priority_level()) ;
|
|
|
|
if (!(mBio_flags & BIN_FLAGS_NO_DELETE))
|
|
{
|
|
delete pqi;
|
|
}
|
|
return 1;
|
|
}
|
|
else
|
|
{
|
|
/* cleanup serialiser */
|
|
free(ptr);
|
|
}
|
|
|
|
std::string out = "pqistreamer::queue_outpqi() Null Pkt generated!\nCaused By:\n";
|
|
pqi -> print_string(out);
|
|
pqioutput(PQL_ALERT, pqistreamerzone, out);
|
|
|
|
if (!(mBio_flags & BIN_FLAGS_NO_DELETE))
|
|
{
|
|
delete pqi;
|
|
}
|
|
return 1; // keep error internal.
|
|
}
|
|
|
|
int pqistreamer::handleincomingitem_locked(RsItem *pqi)
|
|
{
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincomingitem_locked()");
|
|
#endif
|
|
// timestamp last received packet.
|
|
mLastIncomingTs = time(NULL);
|
|
|
|
// Use overloaded Contact function
|
|
pqi -> PeerId(PeerId());
|
|
mIncoming.push_back(pqi);
|
|
return 1;
|
|
}
|
|
|
|
time_t pqistreamer::getLastIncomingTS()
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
return mLastIncomingTs;
|
|
}
|
|
|
|
int pqistreamer::handleoutgoing_locked()
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleoutgoing_locked()");
|
|
#endif
|
|
|
|
int maxbytes = outAllowedBytes_locked();
|
|
int sentbytes = 0;
|
|
int len;
|
|
int ss;
|
|
// std::cerr << "pqistreamer: maxbytes=" << maxbytes<< std::endl ;
|
|
|
|
std::list<void *>::iterator it;
|
|
|
|
// if not connection, or cannot send anything... pause.
|
|
if (!(mBio->isactive()))
|
|
{
|
|
/* if we are not active - clear anything in the queues. */
|
|
locked_clear_out_queue() ;
|
|
|
|
/* also remove the pending packets */
|
|
if (mPkt_wpending)
|
|
{
|
|
free(mPkt_wpending);
|
|
mPkt_wpending = NULL;
|
|
}
|
|
|
|
outSentBytes_locked(sentbytes);
|
|
return 0;
|
|
}
|
|
|
|
// a very simple round robin
|
|
|
|
bool sent = true;
|
|
while(sent) // catch if all items sent.
|
|
{
|
|
sent = false;
|
|
|
|
if ((!(mBio->cansend(0))) || (maxbytes < sentbytes))
|
|
{
|
|
|
|
#ifdef DEBUG_TRANSFERS
|
|
if (maxbytes < sentbytes)
|
|
{
|
|
std::cerr << "pqistreamer::handleoutgoing_locked() Stopped sending sentbytes > maxbytes. Sent " << sentbytes << " bytes ";
|
|
std::cerr << std::endl;
|
|
}
|
|
else
|
|
{
|
|
std::cerr << "pqistreamer::handleoutgoing_locked() Stopped sending at cansend() is false";
|
|
std::cerr << std::endl;
|
|
}
|
|
#endif
|
|
|
|
outSentBytes_locked(sentbytes);
|
|
return 0;
|
|
}
|
|
|
|
// send a out_pkt., else send out_data. unless
|
|
// there is a pending packet.
|
|
if (!mPkt_wpending)
|
|
mPkt_wpending = locked_pop_out_data() ;
|
|
|
|
if (mPkt_wpending)
|
|
{
|
|
// write packet.
|
|
len = getRsItemSize(mPkt_wpending);
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cout << "Sending Out Pkt of size " << len << " !" << std::endl;
|
|
#endif
|
|
|
|
if (len != (ss = mBio->senddata(mPkt_wpending, len)))
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::string out;
|
|
rs_sprintf(out, "Problems with Send Data! (only %d bytes sent, total pkt size=%d)", ss, len);
|
|
// std::cerr << out << std::endl ;
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, out);
|
|
#endif
|
|
|
|
outSentBytes_locked(sentbytes);
|
|
// pkt_wpending will kept til next time.
|
|
// ensuring exactly the same data is written (openSSL requirement).
|
|
return -1;
|
|
}
|
|
|
|
#ifdef DEBUG_TRANSFERS
|
|
std::cerr << "pqistreamer::handleoutgoing_locked() Sent Packet len: " << len << " @ " << RsUtil::AccurateTimeString();
|
|
std::cerr << std::endl;
|
|
#endif
|
|
|
|
free(mPkt_wpending);
|
|
mPkt_wpending = NULL;
|
|
|
|
sentbytes += len;
|
|
sent = true;
|
|
}
|
|
}
|
|
outSentBytes_locked(sentbytes);
|
|
return 1;
|
|
}
|
|
|
|
|
|
/* Handles reading from input stream.
|
|
*/
|
|
int pqistreamer::handleincoming_locked()
|
|
{
|
|
int readbytes = 0;
|
|
static const int max_failed_read_attempts = 2000 ;
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincoming_locked()");
|
|
#endif
|
|
|
|
if(!(mBio->isactive()))
|
|
{
|
|
mReading_state = reading_state_initial ;
|
|
inReadBytes_locked(readbytes);
|
|
return 0;
|
|
}
|
|
|
|
// enough space to read any packet.
|
|
int maxlen = mPkt_rpend_size;
|
|
void *block = mPkt_rpending;
|
|
|
|
// initial read size: basic packet.
|
|
int blen = getRsPktBaseSize();
|
|
|
|
int maxin = inAllowedBytes_locked();
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "reading state = " << mReading_state << std::endl ;
|
|
#endif
|
|
switch(mReading_state)
|
|
{
|
|
case reading_state_initial: /*std::cerr << "jumping to start" << std::endl; */ goto start_packet_read ;
|
|
case reading_state_packet_started: /*std::cerr << "jumping to middle" << std::endl;*/ goto continue_packet ;
|
|
}
|
|
|
|
start_packet_read:
|
|
{ // scope to ensure variable visibility
|
|
// read the basic block (minimum packet size)
|
|
int tmplen;
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "starting packet" << std::endl ;
|
|
#endif
|
|
memset(block,0,blen) ; // reset the block, to avoid uninitialized memory reads.
|
|
|
|
if (blen != (tmplen = mBio->readdata(block, blen)))
|
|
{
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, "pqistreamer::handleincoming() Didn't read BasePkt!");
|
|
|
|
inReadBytes_locked(readbytes);
|
|
|
|
// error.... (either blocked or failure)
|
|
if (tmplen == 0)
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
// most likely blocked!
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, "pqistreamer::handleincoming() read blocked");
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "given up 1" << std::endl ;
|
|
#endif
|
|
return 0;
|
|
}
|
|
else if (tmplen < 0)
|
|
{
|
|
// Most likely it is that the packet is pending but could not be read by pqissl because of stream flow.
|
|
// So we return without an error, and leave the machine state in 'start_read'.
|
|
//
|
|
//pqioutput(PQL_WARNING, pqistreamerzone, "pqistreamer::handleincoming() Error in bio read");
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "given up 2, state = " << mReading_state << std::endl ;
|
|
#endif
|
|
return 0;
|
|
}
|
|
else // tmplen > 0
|
|
{
|
|
// strange case....This should never happen as partial reads are handled by pqissl below.
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::string out = "pqistreamer::handleincoming() Incomplete ";
|
|
rs_sprintf_append(out, "(Strange) read of %d bytes", tmplen);
|
|
pqioutput(PQL_ALERT, pqistreamerzone, out);
|
|
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "given up 3" << std::endl ;
|
|
#endif
|
|
return -1;
|
|
}
|
|
}
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "block 0 : " << (int)(((unsigned char*)block)[0]) << " " << (int)(((unsigned char*)block)[1]) << " " << (int)(((unsigned char*)block)[2]) << " "
|
|
<< (int)(((unsigned char*)block)[3]) << " "
|
|
<< (int)(((unsigned char*)block)[4]) << " "
|
|
<< (int)(((unsigned char*)block)[5]) << " "
|
|
<< (int)(((unsigned char*)block)[6]) << " "
|
|
<< (int)(((unsigned char*)block)[7]) << " " << std::endl ;
|
|
#endif
|
|
|
|
readbytes += blen;
|
|
mReading_state = reading_state_packet_started ;
|
|
mFailed_read_attempts = 0 ; // reset failed read, as the packet has been totally read.
|
|
}
|
|
continue_packet:
|
|
{
|
|
// workout how much more to read.
|
|
int extralen = getRsItemSize(block) - blen;
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "continuing packet getRsItemSize(block) = " << getRsItemSize(block) << std::endl ;
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "continuing packet extralen = " << extralen << std::endl ;
|
|
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "continuing packet state=" << mReading_state << std::endl ;
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "block 1 : " << (int)(((unsigned char*)block)[0]) << " " << (int)(((unsigned char*)block)[1]) << " " << (int)(((unsigned char*)block)[2]) << " " << (int)(((unsigned char*)block)[3]) << " "
|
|
<< (int)(((unsigned char*)block)[4]) << " "
|
|
<< (int)(((unsigned char*)block)[5]) << " "
|
|
<< (int)(((unsigned char*)block)[6]) << " "
|
|
<< (int)(((unsigned char*)block)[7]) << " " << std::endl ;
|
|
#endif
|
|
if (extralen > maxlen - blen)
|
|
{
|
|
pqioutput(PQL_ALERT, pqistreamerzone, "ERROR: Read Packet too Big!");
|
|
|
|
pqiNotify *notify = getPqiNotify();
|
|
if (notify)
|
|
{
|
|
std::string title =
|
|
"Warning: Bad Packet Read";
|
|
|
|
std::string msg;
|
|
msg = " **** WARNING **** \n";
|
|
msg += "Retroshare has caught a BAD Packet Read";
|
|
msg += "\n";
|
|
msg += "This is normally caused by connecting to an";
|
|
msg += " OLD version of Retroshare";
|
|
msg += "\n";
|
|
rs_sprintf_append(msg, "(M:%d B:%d E:%d)\n", maxlen, blen, extralen);
|
|
msg += "\n";
|
|
rs_sprintf_append(msg, "block = %d %d %d %d %d %d %d %d\n",
|
|
(int)(((unsigned char*)block)[0]),
|
|
(int)(((unsigned char*)block)[1]),
|
|
(int)(((unsigned char*)block)[2]),
|
|
(int)(((unsigned char*)block)[3]),
|
|
(int)(((unsigned char*)block)[4]),
|
|
(int)(((unsigned char*)block)[5]),
|
|
(int)(((unsigned char*)block)[6]),
|
|
(int)(((unsigned char*)block)[7])) ;
|
|
msg += "\n";
|
|
msg += "Please get your friends to upgrade to the latest version";
|
|
msg += "\n";
|
|
msg += "\n";
|
|
msg += "If you are sure the error was not caused by an old version";
|
|
msg += "\n";
|
|
msg += "Please report the problem to Retroshare's developers";
|
|
msg += "\n";
|
|
|
|
notify->AddLogMessage(0, RS_SYS_WARNING, title, msg);
|
|
|
|
std::cerr << "pqistreamer::handle_incoming() ERROR: Read Packet too Big" << std::endl;
|
|
std::cerr << msg;
|
|
std::cerr << std::endl;
|
|
|
|
}
|
|
mBio->close();
|
|
mReading_state = reading_state_initial ; // restart at state 1.
|
|
mFailed_read_attempts = 0 ;
|
|
return -1;
|
|
|
|
// Used to exit now! exit(1);
|
|
}
|
|
|
|
if (extralen > 0)
|
|
{
|
|
void *extradata = (void *) (((char *) block) + blen);
|
|
int tmplen ;
|
|
|
|
// Don't reset the block now! If pqissl is in the middle of a multiple-chunk
|
|
// packet (larger than 16384 bytes), and pqistreamer jumped directly yo
|
|
// continue_packet:, then readdata is going to write after the beginning of
|
|
// extradata, yet not exactly at start -> the start of the packet would be wiped out.
|
|
//
|
|
// so, don't do that:
|
|
// memset( extradata,0,extralen ) ;
|
|
|
|
if (extralen != (tmplen = mBio->readdata(extradata, extralen)))
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
if(tmplen > 0)
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "Incomplete packet read ! This is a real problem ;-)" << std::endl ;
|
|
#endif
|
|
|
|
if(++mFailed_read_attempts > max_failed_read_attempts)
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "Error Completing Read (read %d/%d)", tmplen, extralen);
|
|
std::cerr << out << std::endl ;
|
|
pqioutput(PQL_ALERT, pqistreamerzone, out);
|
|
|
|
pqiNotify *notify = getPqiNotify();
|
|
if (notify)
|
|
{
|
|
std::string title = "Warning: Error Completing Read";
|
|
|
|
std::string msgout;
|
|
msgout = " **** WARNING **** \n";
|
|
msgout += "Retroshare has experienced an unexpected Read ERROR";
|
|
msgout += "\n";
|
|
rs_sprintf_append(msgout, "(M:%d B:%d E:%d R:%d)\n", maxlen, blen, extralen, tmplen);
|
|
msgout += "\n";
|
|
msgout += "Note: this error might as well happen (rarely) when a peer disconnects in between a transmission of a large packet.\n";
|
|
msgout += "If it happens manny time, please contact the developers, and send them these numbers:";
|
|
msgout += "\n";
|
|
|
|
rs_sprintf_append(msgout, "block = %d %d %d %d %d %d %d %d\n",
|
|
(int)(((unsigned char*)block)[0]),
|
|
(int)(((unsigned char*)block)[1]),
|
|
(int)(((unsigned char*)block)[2]),
|
|
(int)(((unsigned char*)block)[3]),
|
|
(int)(((unsigned char*)block)[4]),
|
|
(int)(((unsigned char*)block)[5]),
|
|
(int)(((unsigned char*)block)[6]),
|
|
(int)(((unsigned char*)block)[7]));
|
|
|
|
//notify->AddSysMessage(0, RS_SYS_WARNING, title, msgout.str());
|
|
|
|
std::cerr << msgout << std::endl;
|
|
}
|
|
|
|
mBio->close();
|
|
mReading_state = reading_state_initial ; // restart at state 1.
|
|
mFailed_read_attempts = 0 ;
|
|
return -1;
|
|
}
|
|
else
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "given up 5, state = " << mReading_state << std::endl ;
|
|
#endif
|
|
return 0 ; // this is just a SSL_WANT_READ error. Don't panic, we'll re-try the read soon.
|
|
// we assume readdata() returned either -1 or the complete read size.
|
|
}
|
|
}
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "continuing packet state=" << mReading_state << std::endl ;
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "block 2 : " << (int)(((unsigned char*)extradata)[0]) << " " << (int)(((unsigned char*)extradata)[1]) << " " << (int)(((unsigned char*)extradata)[2]) << " " << (int)(((unsigned char*)extradata)[3]) << " "
|
|
<< (int)(((unsigned char*)extradata)[4]) << " "
|
|
<< (int)(((unsigned char*)extradata)[5]) << " "
|
|
<< (int)(((unsigned char*)extradata)[6]) << " "
|
|
<< (int)(((unsigned char*)extradata)[7]) << " " << std::endl ;
|
|
#endif
|
|
|
|
mFailed_read_attempts = 0 ;
|
|
readbytes += extralen;
|
|
}
|
|
|
|
// create packet, based on header.
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "Read Data Block -> Incoming Pkt(%d)", blen + extralen);
|
|
//std::cerr << out ;
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
// std::cerr << "Deserializing packet of size " << pktlen <<std::endl ;
|
|
|
|
uint32_t pktlen = blen+extralen ;
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::cerr << "[" << (void*)pthread_self() << "] " << "deserializing. Size=" << pktlen << std::endl ;
|
|
#endif
|
|
|
|
// if(pktlen == 17306)
|
|
// {
|
|
// FILE *f = RsDirUtil::rs_fopen("dbug.packet.bin","w");
|
|
// fwrite(block,pktlen,1,f) ;
|
|
// fclose(f) ;
|
|
// exit(-1) ;
|
|
// }
|
|
RsItem *pkt = mRsSerialiser->deserialise(block, &pktlen);
|
|
|
|
if ((pkt != NULL) && (0 < handleincomingitem_locked(pkt)))
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, "Successfully Read a Packet!");
|
|
#endif
|
|
}
|
|
else
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_ALERT, pqistreamerzone, "Failed to handle Packet!");
|
|
#endif
|
|
}
|
|
|
|
mReading_state = reading_state_initial ; // restart at state 1.
|
|
mFailed_read_attempts = 0 ; // reset failed read, as the packet has been totally read.
|
|
}
|
|
|
|
if(maxin > readbytes && mBio->moretoread(0))
|
|
goto start_packet_read ;
|
|
|
|
#ifdef DEBUG_TRANSFERS
|
|
if (readbytes >= maxin)
|
|
{
|
|
std::cerr << "pqistreamer::handleincoming() Stopped reading as readbytes >= maxin. Read " << readbytes << " bytes ";
|
|
std::cerr << std::endl;
|
|
}
|
|
#endif
|
|
|
|
inReadBytes_locked(readbytes);
|
|
return 0;
|
|
}
|
|
|
|
|
|
/* BandWidth Management Assistance */
|
|
|
|
float pqistreamer::outTimeSlice_locked()
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::outTimeSlice()");
|
|
#endif
|
|
|
|
//fixme("pqistreamer::outTimeSlice()", 1);
|
|
return 1;
|
|
}
|
|
|
|
// very simple.....
|
|
int pqistreamer::outAllowedBytes_locked()
|
|
{
|
|
int t = time(NULL); // get current timestep.
|
|
|
|
/* allow a lot if not bandwidthLimited */
|
|
if (!mBio->bandwidthLimited())
|
|
{
|
|
mCurrSent = 0;
|
|
mCurrSentTS = t;
|
|
return PQISTREAM_ABS_MAX;
|
|
}
|
|
|
|
int dt = t - mCurrSentTS;
|
|
// limiter -> for when currSentTs -> 0.
|
|
if (dt > 5)
|
|
dt = 5;
|
|
|
|
int maxout = (int) (getMaxRate(false) * 1000.0);
|
|
mCurrSent -= dt * maxout;
|
|
if (mCurrSent < 0)
|
|
{
|
|
mCurrSent = 0;
|
|
}
|
|
|
|
mCurrSentTS = t;
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "pqistreamer::outAllowedBytes() is %d/%d", maxout - mCurrSent, maxout);
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
|
|
return maxout - mCurrSent;
|
|
}
|
|
|
|
int pqistreamer::inAllowedBytes_locked()
|
|
{
|
|
int t = time(NULL); // get current timestep.
|
|
|
|
/* allow a lot if not bandwidthLimited */
|
|
if (!mBio->bandwidthLimited())
|
|
{
|
|
mCurrReadTS = t;
|
|
mCurrRead = 0;
|
|
return PQISTREAM_ABS_MAX;
|
|
}
|
|
|
|
int dt = t - mCurrReadTS;
|
|
// limiter -> for when currReadTs -> 0.
|
|
if (dt > 5)
|
|
dt = 5;
|
|
|
|
int maxin = (int) (getMaxRate(true) * 1000.0);
|
|
mCurrRead -= dt * maxin;
|
|
if (mCurrRead < 0)
|
|
{
|
|
mCurrRead = 0;
|
|
}
|
|
|
|
mCurrReadTS = t;
|
|
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "pqistreamer::inAllowedBytes() is %d/%d", maxin - mCurrRead, maxin);
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
|
|
return maxin - mCurrRead;
|
|
}
|
|
|
|
|
|
static const float AVG_PERIOD = 5; // sec
|
|
static const float AVG_FRAC = 0.8; // for low pass filter.
|
|
|
|
void pqistreamer::outSentBytes_locked(int outb)
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "pqistreamer::outSentBytes(): %d@%gkB/s", outb, getRate(false));
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
/*** One theory for the massive delays - is that the queue here is filling up ****/
|
|
//#define DEBUG_LAG 1
|
|
#ifdef DEBUG_LAG
|
|
|
|
#define MIN_PKTS_FOR_MSG 100
|
|
if (out_queue_size() > MIN_PKTS_FOR_MSG)
|
|
{
|
|
std::cerr << "pqistreamer::outSentBytes() for: " << PeerId();
|
|
std::cerr << " End of Write and still " << out_queue_size() << " pkts left";
|
|
std::cerr << std::endl;
|
|
}
|
|
|
|
#endif
|
|
|
|
|
|
|
|
|
|
|
|
|
|
mTotalSent += outb;
|
|
mCurrSent += outb;
|
|
mAvgSentCount += outb;
|
|
|
|
int t = time(NULL); // get current timestep.
|
|
if (t - mAvgLastUpdate > AVG_PERIOD)
|
|
{
|
|
float avgReadpSec = getRate(true);
|
|
float avgSentpSec = getRate(false);
|
|
|
|
avgReadpSec *= AVG_FRAC;
|
|
avgReadpSec += (1.0 - AVG_FRAC) * mAvgReadCount /
|
|
(1000.0 * (t - mAvgLastUpdate));
|
|
|
|
avgSentpSec *= AVG_FRAC;
|
|
avgSentpSec += (1.0 - AVG_FRAC) * mAvgSentCount /
|
|
(1000.0 * (t - mAvgLastUpdate));
|
|
|
|
|
|
/* pretend our rate is zero if we are
|
|
* not bandwidthLimited().
|
|
*/
|
|
if (mBio->bandwidthLimited())
|
|
{
|
|
setRate(true, avgReadpSec);
|
|
setRate(false, avgSentpSec);
|
|
}
|
|
else
|
|
{
|
|
setRate(true, 0);
|
|
setRate(false, 0);
|
|
}
|
|
|
|
|
|
mAvgLastUpdate = t;
|
|
mAvgReadCount = 0;
|
|
mAvgSentCount = 0;
|
|
}
|
|
return;
|
|
}
|
|
|
|
void pqistreamer::inReadBytes_locked(int inb)
|
|
{
|
|
#ifdef DEBUG_PQISTREAMER
|
|
{
|
|
std::string out;
|
|
rs_sprintf(out, "pqistreamer::inReadBytes(): %d@%gkB/s", inb, getRate(true));
|
|
pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out);
|
|
}
|
|
#endif
|
|
|
|
mTotalRead += inb;
|
|
mCurrRead += inb;
|
|
mAvgReadCount += inb;
|
|
|
|
return;
|
|
}
|
|
|
|
int pqistreamer::getQueueSize(bool in)
|
|
{
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
if (in)
|
|
return mIncoming.size();
|
|
return locked_out_queue_size();
|
|
}
|
|
|
|
void pqistreamer::getRates(RsBwRates &rates)
|
|
{
|
|
RateInterface::getRates(rates);
|
|
|
|
RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
|
|
|
|
rates.mQueueIn = mIncoming.size();
|
|
rates.mQueueOut = locked_out_queue_size();
|
|
}
|
|
|
|
int pqistreamer::locked_out_queue_size() const
|
|
{
|
|
// Warning: because out_pkt is a list, calling size
|
|
// is O(n) ! Makign algorithms pretty inefficient. We should record how many
|
|
// items get stored and discarded to have a proper size value at any time
|
|
//
|
|
return mOutPkts.size() ;
|
|
}
|
|
|
|
void pqistreamer::locked_clear_out_queue()
|
|
{
|
|
for(std::list<void*>::iterator it = mOutPkts.begin(); it != mOutPkts.end(); )
|
|
{
|
|
free(*it);
|
|
it = mOutPkts.erase(it);
|
|
#ifdef DEBUG_PQISTREAMER
|
|
std::string out = "pqistreamer::locked_clear_out_queue() Not active -> Clearing Pkt!";
|
|
std::cerr << out << std::endl;
|
|
pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, out);
|
|
#endif
|
|
}
|
|
}
|
|
|
|
int pqistreamer::locked_compute_out_pkt_size() const
|
|
{
|
|
int total = 0 ;
|
|
|
|
for(std::list<void*>::const_iterator it = mOutPkts.begin(); it != mOutPkts.end(); ++it)
|
|
total += getRsItemSize(*it);
|
|
|
|
return total ;
|
|
}
|
|
|
|
void *pqistreamer::locked_pop_out_data()
|
|
{
|
|
void *res = NULL ;
|
|
|
|
if (!mOutPkts.empty())
|
|
{
|
|
res = *(mOutPkts.begin());
|
|
mOutPkts.pop_front();
|
|
#ifdef DEBUG_TRANSFERS
|
|
std::cerr << "pqistreamer::locked_pop_out_data() getting next pkt from mOutPkts queue";
|
|
std::cerr << std::endl;
|
|
#endif
|
|
}
|
|
return res ;
|
|
}
|