github
/
aria2
mirror of https://github.com/aria2/aria2
1
0
Fork 0
Code Issues Releases Wiki Activity

2009-07-22 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 

Rewritten stream error handling
	* src/DHTAutoSaveCommand.cc
	* src/DHTRoutingTableDeserializer.cc
	* src/DHTRoutingTableSerializer.cc
	* src/DHTSetup.cc
pull/1/head
Tatsuhiro Tsujikawa 2009-07-22 12:54:35 +00:00
parent 0d101d47c9
commit 65906c07fe
5 changed files with 213 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
ChangeLog
View File

@ -1,3 +1,11 @@
2009-07-22 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Rewritten stream error handling
* src/DHTAutoSaveCommand.cc
* src/DHTRoutingTableDeserializer.cc
* src/DHTRoutingTableSerializer.cc
* src/DHTSetup.cc
2009-07-22 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 2009-07-22 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Check stream status after file is opened. Check stream status after file is opened.

21
src/DHTAutoSaveCommand.cc
View File

@ -53,6 +53,8 @@
#include "Logger.h" #include "Logger.h"
#include "a2functional.h" #include "a2functional.h"
#include "FileEntry.h" #include "FileEntry.h"
#include "DlAbortEx.h"
#include "StringFormat.h"
namespace aria2 { namespace aria2 {
@ -112,21 +114,22 @@ void DHTAutoSaveCommand::save()
serializer.setNodes(nodes); serializer.setNodes(nodes);
try { try {
std::ofstream o(tempFile.c_str(), std::ios::out|std::ios::binary); {
o.exceptions(std::ios::failbit); std::ofstream o(tempFile.c_str(), std::ios::out|std::ios::binary);
if(!o) {
serializer.serialize(o); throw DL_ABORT_EX
(StringFormat("Failed to save DHT routing table to %s. cause:%s",
dhtFile.c_str(), strerror(errno)).str());
}
serializer.serialize(o);
}
if(!File(tempFile).renameTo(dhtFile)) { if(!File(tempFile).renameTo(dhtFile)) {
logger->error("Cannot move file from %s to %s.", logger->error("Cannot move file from %s to %s.",
tempFile.c_str(), dhtFile.c_str()); tempFile.c_str(), dhtFile.c_str());
} }
} catch(std::ios::failure const& e) {
logger->error("Failed to save DHT routing table to %s. cause:%s",
tempFile.c_str(), strerror(errno));
} catch(RecoverableException& e) { } catch(RecoverableException& e) {
logger->error("Exception caught while saving DHT routing table to %s", logger->error("Exception caught while saving DHT routing table to %s",
e, tempFile.c_str()); e, dhtFile.c_str());
} }
} }

257
src/DHTRoutingTableDeserializer.cc
View File

@ -56,136 +56,161 @@ DHTRoutingTableDeserializer::DHTRoutingTableDeserializer() {}
DHTRoutingTableDeserializer::~DHTRoutingTableDeserializer() {} DHTRoutingTableDeserializer::~DHTRoutingTableDeserializer() {}
static std::istream& readBytes(unsigned char* buf, size_t buflen, static void readBytes(unsigned char* buf, size_t buflen,
std::istream& in, size_t readlen) std::istream& in, size_t readlen)
{ {
assert(readlen <= buflen); assert(readlen <= buflen);
return in.read(reinterpret_cast<char*>(buf), readlen); in.read(reinterpret_cast<char*>(buf), readlen);
} }
#define CHECK_STREAM(in, length) \
if(in.gcount() != length) { \
throw DL_ABORT_EX \
(StringFormat("Failed to load DHT routing table. cause:%s", \
"Unexpected EOF").str()); \
} \
if(!in) { \
throw DL_ABORT_EX \
(StringFormat("Failed to load DHT routing table. cause:%s", \
strerror(errno)).str()); \
}
void DHTRoutingTableDeserializer::deserialize(std::istream& in) void DHTRoutingTableDeserializer::deserialize(std::istream& in)
{ {
try { char header[8];
char header[8]; memset(header, 0, sizeof(header));
memset(header, 0, sizeof(header)); // magic
// magic header[0] = 0xa1;
header[0] = 0xa1; header[1] = 0xa2;
header[1] = 0xa2; // format ID
// format ID header[2] = 0x02;
header[2] = 0x02; // version
// version header[6] = 0;
header[6] = 0; header[7] = 0x03;
header[7] = 0x03;
char headerCompat[8]; char headerCompat[8];
memset(headerCompat, 0, sizeof(headerCompat)); memset(headerCompat, 0, sizeof(headerCompat));
// magic // magic
headerCompat[0] = 0xa1; headerCompat[0] = 0xa1;
headerCompat[1] = 0xa2; headerCompat[1] = 0xa2;
// format ID // format ID
headerCompat[2] = 0x02; headerCompat[2] = 0x02;
// version // version
headerCompat[6] = 0; headerCompat[6] = 0;
headerCompat[7] = 0x02; headerCompat[7] = 0x02;
char zero[8]; char zero[8];
memset(zero, 0, sizeof(zero)); memset(zero, 0, sizeof(zero));
int version; int version;
// If you change the code to read more than the size of buf, then // If you change the code to read more than the size of buf, then
// expand the buf size here. // expand the buf size here.
array_wrapper<unsigned char, 255> buf; array_wrapper<unsigned char, 255> buf;
// header // header
readBytes(buf, buf.size(), in, 8); readBytes(buf, buf.size(), in, 8);
if(memcmp(header, buf, 8) == 0) { CHECK_STREAM(in, 8);
version = 3; if(memcmp(header, buf, 8) == 0) {
} else if(memcmp(headerCompat, buf, 8) == 0) { version = 3;
version = 2; } else if(memcmp(headerCompat, buf, 8) == 0) {
} else { version = 2;
throw DL_ABORT_EX } else {
(StringFormat("Failed to load DHT routing table. cause:%s",
"bad header").str());
}
uint32_t temp32;
uint64_t temp64;
// time
if(version == 2) {
in.read(reinterpret_cast<char*>(&temp32), sizeof(temp32));
_serializedTime.setTimeInSec(ntohl(temp32));
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
} else {
in.read(reinterpret_cast<char*>(&temp64), sizeof(temp64));
_serializedTime.setTimeInSec(ntoh64(temp64));
}
// localnode
// 8bytes reserved
readBytes(buf, buf.size(), in, 8);
// localnode ID
readBytes(buf, buf.size(), in, DHT_ID_LENGTH);
SharedHandle<DHTNode> localNode(new DHTNode(buf));
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
// number of nodes
in.read(reinterpret_cast<char*>(&temp32), sizeof(temp32));
uint32_t numNodes = ntohl(temp32);
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
// nodes
for(size_t i = 0; i < numNodes; ++i) {
// Currently, only IPv4 addresses are supported.
// 1byte compact peer info length
uint8_t peerInfoLen;
in >> peerInfoLen;
if(peerInfoLen != 6) {
// skip this entry
readBytes(buf, buf.size(), in, 42+7+6);
continue;
}
// 7bytes reserved
readBytes(buf, buf.size(), in, 7);
// 6bytes compact peer info
readBytes(buf, buf.size(), in, 6);
if(memcmp(zero, buf, 6) == 0) {
// skip this entry
readBytes(buf, buf.size(), in, 42);
continue;
}
std::pair<std::string, uint16_t> peer =
PeerMessageUtil::unpackcompact(buf);
if(peer.first.empty()) {
// skip this entry
readBytes(buf, buf.size(), in, 42);
continue;
}
// 2bytes reserved
readBytes(buf, buf.size(), in, 2);
// 16byte reserved
readBytes(buf, buf.size(), in, 16);
// localnode ID
readBytes(buf, buf.size(), in, DHT_ID_LENGTH);
SharedHandle<DHTNode> node(new DHTNode(buf));
node->setIPAddress(peer.first);
node->setPort(peer.second);
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
_nodes.push_back(node);
}
_localNode = localNode;
} catch(std::ios::failure const& exception) {
_nodes.clear();
throw DL_ABORT_EX throw DL_ABORT_EX
(StringFormat("Failed to load DHT routing table. cause:%s", (StringFormat("Failed to load DHT routing table. cause:%s",
strerror(errno)).str()); "bad header").str());
} }
uint32_t temp32;
uint64_t temp64;
// time
if(version == 2) {
in.read(reinterpret_cast<char*>(&temp32), sizeof(temp32));
CHECK_STREAM(in, sizeof(temp32));
_serializedTime.setTimeInSec(ntohl(temp32));
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
CHECK_STREAM(in, 4);
} else {
in.read(reinterpret_cast<char*>(&temp64), sizeof(temp64));
CHECK_STREAM(in, sizeof(temp64));
_serializedTime.setTimeInSec(ntoh64(temp64));
}
// localnode
// 8bytes reserved
readBytes(buf, buf.size(), in, 8);
CHECK_STREAM(in, 8);
// localnode ID
readBytes(buf, buf.size(), in, DHT_ID_LENGTH);
CHECK_STREAM(in, DHT_ID_LENGTH);
SharedHandle<DHTNode> localNode(new DHTNode(buf));
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
CHECK_STREAM(in, 4);
// number of nodes
in.read(reinterpret_cast<char*>(&temp32), sizeof(temp32));
CHECK_STREAM(in, sizeof(temp32));
uint32_t numNodes = ntohl(temp32);
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
CHECK_STREAM(in, 4);
std::deque<SharedHandle<DHTNode> > nodes;
// nodes
for(size_t i = 0; i < numNodes; ++i) {
// Currently, only IPv4 addresses are supported.
// 1byte compact peer info length
uint8_t peerInfoLen;
in >> peerInfoLen;
if(peerInfoLen != 6) {
// skip this entry
readBytes(buf, buf.size(), in, 42+7+6);
CHECK_STREAM(in, 42+7+6);
continue;
}
// 7bytes reserved
readBytes(buf, buf.size(), in, 7);
CHECK_STREAM(in, 7);
// 6bytes compact peer info
readBytes(buf, buf.size(), in, 6);
CHECK_STREAM(in, 6);
if(memcmp(zero, buf, 6) == 0) {
// skip this entry
readBytes(buf, buf.size(), in, 42);
CHECK_STREAM(in, 42);
continue;
}
std::pair<std::string, uint16_t> peer =
PeerMessageUtil::unpackcompact(buf);
if(peer.first.empty()) {
// skip this entry
readBytes(buf, buf.size(), in, 42);
CHECK_STREAM(in, 42);
continue;
}
// 2bytes reserved
readBytes(buf, buf.size(), in, 2);
CHECK_STREAM(in, 2);
// 16byte reserved
readBytes(buf, buf.size(), in, 16);
CHECK_STREAM(in, 16);
// localnode ID
readBytes(buf, buf.size(), in, DHT_ID_LENGTH);
CHECK_STREAM(in, DHT_ID_LENGTH);
SharedHandle<DHTNode> node(new DHTNode(buf));
node->setIPAddress(peer.first);
node->setPort(peer.second);
// 4bytes reserved
readBytes(buf, buf.size(), in, 4);
CHECK_STREAM(in, 4);
nodes.push_back(node);
}
_localNode = localNode;
_nodes = nodes;
} }
} // namespace aria2 } // namespace aria2

84
src/DHTRoutingTableSerializer.cc
View File

@ -76,51 +76,53 @@ void DHTRoutingTableSerializer::serialize(std::ostream& o)
char zero[16]; char zero[16];
memset(zero, 0, sizeof(zero)); memset(zero, 0, sizeof(zero));
try {
o.write(header, 8);
// write save date
uint64_t ntime = hton64(Time().getTime());
o.write(reinterpret_cast<const char*>(&ntime), sizeof(ntime));
// localnode o.write(header, 8);
// 8bytes reserved // write save date
o.write(zero, 8); uint64_t ntime = hton64(Time().getTime());
// 20bytes localnode ID o.write(reinterpret_cast<const char*>(&ntime), sizeof(ntime));
o.write(reinterpret_cast<const char*>(_localNode->getID()), DHT_ID_LENGTH);
// 4bytes reserved
o.write(zero, 4);
// number of nodes // localnode
uint32_t numNodes = htonl(_nodes.size()); // 8bytes reserved
o.write(reinterpret_cast<const char*>(&numNodes), sizeof(uint32_t)); o.write(zero, 8);
// 4bytes reserved // 20bytes localnode ID
o.write(zero, 4); o.write(reinterpret_cast<const char*>(_localNode->getID()), DHT_ID_LENGTH);
// 4bytes reserved
o.write(zero, 4);
// nodes // number of nodes
for(std::deque<SharedHandle<DHTNode> >::const_iterator i = _nodes.begin(); i != _nodes.end(); ++i) { uint32_t numNodes = htonl(_nodes.size());
const SharedHandle<DHTNode>& node = *i; o.write(reinterpret_cast<const char*>(&numNodes), sizeof(uint32_t));
// Currently, only IPv4 address and IPv4-mapped address are saved. // 4bytes reserved
// 6bytes: write IP address + port in Compact IP-address/port info form. o.write(zero, 4);
unsigned char compactPeer[6];
if(!PeerMessageUtil::createcompact(compactPeer, node->getIPAddress(), node->getPort())) { // nodes
memset(compactPeer, 0, 6); for(std::deque<SharedHandle<DHTNode> >::const_iterator i = _nodes.begin(); i != _nodes.end(); ++i) {
} const SharedHandle<DHTNode>& node = *i;
// 1byte compact peer format length // Currently, only IPv4 address and IPv4-mapped address are saved.
o << static_cast<uint8_t>(sizeof(compactPeer)); // 6bytes: write IP address + port in Compact IP-address/port info form.
// 7bytes reserved unsigned char compactPeer[6];
o.write(zero, 7); if(!PeerMessageUtil::createcompact(compactPeer, node->getIPAddress(), node->getPort())) {
// 6 bytes compact peer memset(compactPeer, 0, 6);
o.write(reinterpret_cast<const char*>(compactPeer), 6);
// 2bytes reserved
o.write(zero, 2);
// 16bytes reserved
o.write(zero, 16);
// 20bytes: node ID
o.write(reinterpret_cast<const char*>(node->getID()), DHT_ID_LENGTH);
// 4bytes reserved
o.write(zero, 4);
} }
} catch(std::ios::failure const& exception) { // 1byte compact peer format length
o << static_cast<uint8_t>(sizeof(compactPeer));
// 7bytes reserved
o.write(zero, 7);
// 6 bytes compact peer
o.write(reinterpret_cast<const char*>(compactPeer), 6);
// 2bytes reserved
o.write(zero, 2);
// 16bytes reserved
o.write(zero, 16);
// 20bytes: node ID
o.write(reinterpret_cast<const char*>(node->getID()), DHT_ID_LENGTH);
// 4bytes reserved
o.write(zero, 4);
}
o.flush();
if(!o) {
throw DL_ABORT_EX throw DL_ABORT_EX
(StringFormat("Failed to save DHT routing table. cause:%s", (StringFormat("Failed to save DHT routing table. cause:%s",
strerror(errno)).str()); strerror(errno)).str());

18
src/DHTSetup.cc
View File

@ -94,16 +94,16 @@ void DHTSetup::setup(std::deque<Command*>& commands,
DHTRoutingTableDeserializer deserializer; DHTRoutingTableDeserializer deserializer;
std::string dhtFile = option->get(PREF_DHT_FILE_PATH); std::string dhtFile = option->get(PREF_DHT_FILE_PATH);
if(File(dhtFile).isFile()) { try {
try { std::ifstream in(dhtFile.c_str(), std::ios::binary);
std::ifstream in(dhtFile.c_str(), std::ios::binary); if(!in) {
in.exceptions(std::ios::failbit); throw DL_ABORT_EX("Could not open file");
deserializer.deserialize(in);
localNode = deserializer.getLocalNode();
} catch(RecoverableException& e) {
_logger->error("Exception caught while loading DHT routing table from %s",
e, dhtFile.c_str());
} }
deserializer.deserialize(in);
localNode = deserializer.getLocalNode();
} catch(RecoverableException& e) {
_logger->error("Exception caught while loading DHT routing table from %s",
e, dhtFile.c_str());
} }
if(localNode.isNull()) { if(localNode.isNull()) {
localNode.reset(new DHTNode()); localNode.reset(new DHTNode());