zzz/i2p.i2p
1
0
Fork 0
forked from I2P_Developers/i2p.i2p
Code Issues Pull Requests Packages Projects Releases Wiki Activity

* SSU: Fix concurrency errors (ticket #536)

Browse Source
This commit is contained in:
zzz
2011-10-15 17:21:31 +00:00
parent e8712a3a11
commit 3dbefa8d01
1 changed files with 24 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java
View File

@@ -172,15 +172,20 @@ class EstablishmentManager {
if (_queuedOutbound.size() > MAX_QUEUED_OUTBOUND) { if (_queuedOutbound.size() > MAX_QUEUED_OUTBOUND) {
rejected = true; rejected = true;
} else { } else {
List<OutNetMessage> newQueued = new ArrayList(1); List<OutNetMessage> newQueued = new ArrayList(MAX_QUEUED_PER_PEER);
List<OutNetMessage> queued = _queuedOutbound.putIfAbsent(to, newQueued); List<OutNetMessage> queued = _queuedOutbound.putIfAbsent(to, newQueued);
if (queued == null) if (queued == null)
queued = newQueued; queued = newQueued;
queueCount = queued.size(); // this used to be inside a synchronized (_outboundStates) block,
if (queueCount < MAX_QUEUED_PER_PEER) // but that's now a CHM, so protect the ArrayList
queued.add(msg); // There are still races possible but this should prevent AIOOBE and NPE
synchronized (queued) {
queueCount = queued.size();
if (queueCount < MAX_QUEUED_PER_PEER)
queued.add(msg);
deferred = _queuedOutbound.size();
}
} }
deferred = _queuedOutbound.size();
} else { } else {
// must have a valid session key // must have a valid session key
byte[] keyBytes = addr.getIntroKey(); byte[] keyBytes = addr.getIntroKey();
@@ -212,9 +217,14 @@ class EstablishmentManager {
if (state != null) { if (state != null) {
state.addMessage(msg); state.addMessage(msg);
List<OutNetMessage> queued = _queuedOutbound.remove(to); List<OutNetMessage> queued = _queuedOutbound.remove(to);
if (queued != null) if (queued != null) {
for (int i = 0; i < queued.size(); i++) // see comments above
state.addMessage(queued.get(i)); synchronized (queued) {
for (OutNetMessage m : queued) {
state.addMessage(m);
}
}
}
} }
if (rejected) { if (rejected) {
@@ -399,8 +409,12 @@ class EstablishmentManager {
// there shouldn't have been queued messages for this active state, but just in case... // there shouldn't have been queued messages for this active state, but just in case...
List<OutNetMessage> queued = _queuedOutbound.remove(state.getRemoteHostId()); List<OutNetMessage> queued = _queuedOutbound.remove(state.getRemoteHostId());
if (queued != null) { if (queued != null) {
for (int i = 0; i < queued.size(); i++) // see comments above
state.addMessage(queued.get(i)); synchronized (queued) {
for (OutNetMessage m : queued) {
state.addMessage(m);
}
}
} }
//admitted = locked_admitQueued(); //admitted = locked_admitQueued();