diff --git a/core/java/src/net/i2p/crypto/TransientSessionKeyManager.java b/core/java/src/net/i2p/crypto/TransientSessionKeyManager.java index 4f4dba1e7..4cd981860 100644 --- a/core/java/src/net/i2p/crypto/TransientSessionKeyManager.java +++ b/core/java/src/net/i2p/crypto/TransientSessionKeyManager.java @@ -318,8 +318,7 @@ public class TransientSessionKeyManager extends SessionKeyManager { private OutboundSession createAndReturnSession(PublicKey target, SessionKey key) { if (_log.shouldLog(Log.INFO)) _log.info("New OB session, sesskey: " + key + " target: " + toString(target)); - OutboundSession sess = new OutboundSession(_context, _log, target); - sess.setCurrentKey(key); + OutboundSession sess = new OutboundSession(_context, _log, target, key); addSession(sess); return sess; } @@ -476,6 +475,7 @@ public class TransientSessionKeyManager extends SessionKeyManager { /** * Mark these tags as acked, start to use them (if we haven't already) + * If the set was previously failed, it will be added back in. */ @Override public void tagsAcked(PublicKey target, SessionKey key, TagSetHandle ts) { @@ -822,11 +822,13 @@ public class TransientSessionKeyManager extends SessionKeyManager { * on the callers to call failTags() or ackTags() to remove them from this list. * Actually we now do a failsafe expire. * Synch on _tagSets to access this. + * No particular order. */ - private final List _unackedTagSets; + private final Set _unackedTagSets; /** * As tagsets are acked, they go here. * After the first ack, new tagsets go here (i.e. presumed acked) + * In order, earliest first. */ private final List _tagSets; /** @@ -844,20 +846,15 @@ public class TransientSessionKeyManager extends SessionKeyManager { private static final int MAX_FAILS = 2; - public OutboundSession(I2PAppContext ctx, Log log, PublicKey target) { - this(ctx, log, target, null, ctx.clock().now(), ctx.clock().now(), new ArrayList()); - } - - OutboundSession(I2PAppContext ctx, Log log, PublicKey target, SessionKey curKey, - long established, long lastUsed, List tagSets) { + public OutboundSession(I2PAppContext ctx, Log log, PublicKey target, SessionKey key) { _context = ctx; _log = log; _target = target; - _currentKey = curKey; - _established = established; - _lastUsed = lastUsed; - _unackedTagSets = tagSets; - _tagSets = new ArrayList(); + _currentKey = key; + _established = ctx.clock().now(); + _lastUsed = _established; + _unackedTagSets = new HashSet(4); + _tagSets = new ArrayList(6); } /** @@ -878,6 +875,7 @@ public class TransientSessionKeyManager extends SessionKeyManager { * got an ack for these tags * For tagsets delivered after the session was acked, this is a nop * because the tagset was originally placed directly on the acked list. + * If the set was previously failed, it will be added back in. */ void ackTags(TagSet set) { synchronized (_tagSets) { @@ -885,11 +883,14 @@ public class TransientSessionKeyManager extends SessionKeyManager { // we could perhaps use it even if not previuosly in unacked, // i.e. it was expired already, but _tagSets is a list not a set... _tagSets.add(set); - } else if (_log.shouldLog(Log.WARN)) { - if(!_tagSets.contains(set)) - _log.warn("Ack of unknown tagset: " + set); - else if (set.getAcked()) - _log.warn("Dup ack of tagset: " + set); + } else if (!_tagSets.contains(set)) { + // add back (sucess after fail) + _tagSets.add(set); + if (_log.shouldLog(Log.WARN)) + _log.warn("Ack of unknown (previously failed?) tagset: " + set); + } else if (set.getAcked()) { + if (_log.shouldLog(Log.WARN)) + _log.warn("Dup ack of tagset: " + set); } _acked = true; _consecutiveFailures = 0; @@ -1196,6 +1197,7 @@ public class TransientSessionKeyManager extends SessionKeyManager { buf.append("TagSet #").append(_id).append(" created: ").append(new Date(_date)); buf.append(" Session key: ").append(_key); buf.append(" Size: ").append(_sessionTags.size()); + buf.append('/').append(_origSize); buf.append(" Acked? ").append(_acked); return buf.toString(); } diff --git a/history.txt b/history.txt index 8d09534bd..b8dcd7c12 100644 --- a/history.txt +++ b/history.txt @@ -1,3 +1,11 @@ +2014-03-07 zzz + * Router: + - Look for DeliveryStatusMessages beyond the message expiration, + so we don't throw out a tagset that gets acked late + - Allow re-adding of a "failed" tagset to the SKM + - Extend max message age in MessageValidator + - Remove unused and confusing timeout param when registering a selector + 2014-03-06 zzz * Router: Encrypt DeliveryStatusMessages sent in garlics (ticket #1217) diff --git a/router/java/src/net/i2p/router/MessageValidator.java b/router/java/src/net/i2p/router/MessageValidator.java index 6a6a432d9..c86dff0a3 100644 --- a/router/java/src/net/i2p/router/MessageValidator.java +++ b/router/java/src/net/i2p/router/MessageValidator.java @@ -50,12 +50,13 @@ public class MessageValidator { return null; } } + /** * Only check the expiration for the message */ public String validateMessage(long expiration) { long now = _context.clock().now(); - if (now - Router.CLOCK_FUDGE_FACTOR >= expiration) { + if (now - (Router.CLOCK_FUDGE_FACTOR * 3 / 2) >= expiration) { if (_log.shouldLog(Log.INFO)) _log.info("Rejecting message because it expired " + (now-expiration) + "ms ago"); _context.statManager().addRateData("router.invalidMessageTime", (now-expiration), 0); diff --git a/router/java/src/net/i2p/router/OutNetMessage.java b/router/java/src/net/i2p/router/OutNetMessage.java index 6c1635a4d..633416640 100644 --- a/router/java/src/net/i2p/router/OutNetMessage.java +++ b/router/java/src/net/i2p/router/OutNetMessage.java @@ -81,11 +81,11 @@ public class OutNetMessage implements CDPQEntry { public static final int PRIORITY_LOWEST = 100; /** - * Null msg and target (used in OutboundMessageRegistry only) + * Null msg and target, zero expiration (used in OutboundMessageRegistry only) * @since 0.9.9 */ - public OutNetMessage(RouterContext context, long expiration) { - this(context, null, expiration, -1, null); + public OutNetMessage(RouterContext context) { + this(context, null, 0, -1, null); } /** @@ -182,13 +182,13 @@ public class OutNetMessage implements CDPQEntry { /** * Specifies the router to which the message should be delivered. - * + * Generally non-null but may be null in special cases. */ public RouterInfo getTarget() { return _target; } /** - * Specifies the message to be sent - * + * Specifies the message to be sent. + * Generally non-null but may be null in special cases. */ public I2NPMessage getMessage() { return _message; } diff --git a/router/java/src/net/i2p/router/RouterVersion.java b/router/java/src/net/i2p/router/RouterVersion.java index 31c35f8c3..ea17ad55f 100644 --- a/router/java/src/net/i2p/router/RouterVersion.java +++ b/router/java/src/net/i2p/router/RouterVersion.java @@ -18,7 +18,7 @@ public class RouterVersion { /** deprecated */ public final static String ID = "Monotone"; public final static String VERSION = CoreVersion.VERSION; - public final static long BUILD = 14; + public final static long BUILD = 15; /** for example "-test" */ public final static String EXTRA = ""; diff --git a/router/java/src/net/i2p/router/message/OutboundClientMessageOneShotJob.java b/router/java/src/net/i2p/router/message/OutboundClientMessageOneShotJob.java index cb4f6daf7..f41e808e7 100644 --- a/router/java/src/net/i2p/router/message/OutboundClientMessageOneShotJob.java +++ b/router/java/src/net/i2p/router/message/OutboundClientMessageOneShotJob.java @@ -2,15 +2,16 @@ package net.i2p.router.message; import java.util.ArrayList; import java.util.Collections; +import java.util.Date; import java.util.HashSet; import java.util.List; import java.util.Set; -import java.util.concurrent.atomic.AtomicBoolean; import net.i2p.client.SendMessageOptions; import net.i2p.crypto.SessionKeyManager; import net.i2p.crypto.TagSetHandle; import net.i2p.data.Certificate; +import net.i2p.data.DataHelper; import net.i2p.data.Destination; import net.i2p.data.Hash; import net.i2p.data.Lease; @@ -56,10 +57,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl { private LeaseSet _leaseSet; /** Actual lease the message is being routed through */ private Lease _lease; - private PayloadGarlicConfig _clove; - private long _cloveId; private final long _start; - private final AtomicBoolean _finished = new AtomicBoolean(); + /** note we can succeed after failure, but not vice versa */ + private enum Result {NONE, FAIL, SUCCESS} + private Result _finished = Result.NONE; private long _leaseSetLookupBegin; private TunnelInfo _outTunnel; private TunnelInfo _inTunnel; @@ -70,6 +71,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl { */ private final OutboundCache.HashPair _hashPair; + /** * final timeout (in milliseconds) that the outbound message will fail in. * This can be overridden in the router.config or the client's session config @@ -78,6 +80,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl { public final static String OVERALL_TIMEOUT_MS_PARAM = "clientMessageTimeout"; private final static long OVERALL_TIMEOUT_MS_DEFAULT = 60*1000; private final static long OVERALL_TIMEOUT_MS_MIN = 8*1000; + private final static long REPLY_TIMEOUT_MS_MIN = OVERALL_TIMEOUT_MS_DEFAULT - 5*1000; /** * NOTE: Changed as of 0.9.2. @@ -210,7 +213,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl { SendJob success = new SendJob(getContext()); _leaseSet = getContext().netDb().lookupLeaseSetLocally(key); if (_leaseSet != null) { - getContext().statManager().addRateData("client.leaseSetFoundLocally", 1, 0); + getContext().statManager().addRateData("client.leaseSetFoundLocally", 1); if (_log.shouldLog(Log.DEBUG)) _log.debug(getJobId() + ": Send outbound client message - leaseSet found locally for " + _toString); success.runJob(); @@ -252,16 +255,22 @@ public class OutboundClientMessageOneShotJob extends JobImpl { return newLS; } - /** send a message to a lease */ + /** + * Send a message to a lease. + * Note: This is generally run inline by runJob() above. + * It is only run on the job queue after a LS lookup. + */ private class SendJob extends JobImpl { public SendJob(RouterContext enclosingContext) { super(enclosingContext); } - public String getName() { return "Outbound client message send"; } + + public String getName() { return "Outbound client message delayed send"; } + public void runJob() { if (_leaseSetLookupBegin > 0) { long lookupTime = getContext().clock().now() - _leaseSetLookupBegin; - getContext().statManager().addRateData("client.leaseSetFoundRemoteTime", lookupTime, 0); + getContext().statManager().addRateData("client.leaseSetFoundRemoteTime", lookupTime); } _wantACK = false; boolean ok = getNextLease(); @@ -407,10 +416,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl { getContext().statManager().addRateData("client.leaseSetFailedRemoteTime", lookupTime, lookupTime); } - if (!_finished.get()) { + //if (_finished == Result.NONE) { if (_log.shouldLog(Log.WARN)) _log.warn("Unable to send to " + _toString + " because we couldn't find their leaseSet"); - } + //} dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_NO_LEASESET); } @@ -425,7 +434,14 @@ public class OutboundClientMessageOneShotJob extends JobImpl { * */ private void send() { - if (_finished.get()) return; + synchronized(this) { + if (_finished != Result.NONE) { + if (_log.shouldLog(Log.WARN)) + _log.warn(OutboundClientMessageOneShotJob.this.getJobId() + + ": SEND-AFTER-" + _finished); + return; + } + } long now = getContext().clock().now(); if (now >= _overallExpiration) { dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_EXPIRED); @@ -433,6 +449,14 @@ public class OutboundClientMessageOneShotJob extends JobImpl { } _outTunnel = selectOutboundTunnel(_to); + if (_outTunnel == null) { + if (_log.shouldLog(Log.WARN)) + _log.warn(getJobId() + ": Could not find any outbound tunnels to send the payload through... this might take a while"); + getContext().statManager().addRateData("client.dispatchNoTunnels", now - _start); + dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_NO_TUNNELS); + return; + } + // boolean wantACK = _wantACK || existingTags <= 30 || getContext().random().nextInt(100) < 5; // what's the point of 5% random? possible improvements or replacements: // DONE (getNextLease() is called before this): wantACK if we changed their inbound lease (getNextLease() sets _wantACK) @@ -451,10 +475,6 @@ public class OutboundClientMessageOneShotJob extends JobImpl { GarlicMessageBuilder.needsTags(getContext(), _leaseSet.getEncryptionKey(), _from.calculateHash(), tagsRequired); - PublicKey key = _leaseSet.getEncryptionKey(); - SessionKey sessKey = new SessionKey(); - Set tags = new HashSet(); - LeaseSet replyLeaseSet; // Per-message flag == false overrides session option which is default true String allow = _clientMessage.getSenderConfig().getOptions().getProperty(BUNDLE_REPLY_LEASESET); @@ -482,19 +502,24 @@ public class OutboundClientMessageOneShotJob extends JobImpl { token = -1; } - boolean ok = buildClove(); - if (!ok) { + PayloadGarlicConfig clove = buildClove(); + if (clove == null) { dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_UNSUPPORTED_ENCRYPTION); return; } //if (_log.shouldLog(Log.DEBUG)) // _log.debug(getJobId() + ": Clove built to " + _toString); + + PublicKey key = _leaseSet.getEncryptionKey(); + SessionKey sessKey = new SessionKey(); + Set tags = new HashSet(); + long msgExpiration = _overallExpiration; // getContext().clock().now() + OVERALL_TIMEOUT_MS_DEFAULT; // Per-message flag > 0 overrides per-session option int tagsToSend = SendMessageOptions.getTagsToSend(sendFlags); GarlicMessage msg = OutboundClientMessageJobHelper.createGarlicMessage(getContext(), token, msgExpiration, key, - _clove, _from.calculateHash(), + clove, _from.calculateHash(), _to, _inTunnel, tagsToSend, tagsRequired, sessKey, tags, wantACK, replyLeaseSet); @@ -504,7 +529,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl { // we dont receive the reply? hmm...) if (_log.shouldLog(Log.WARN)) _log.warn(getJobId() + ": Unable to create the garlic message (no tunnels left or too lagged) to " + _toString); - getContext().statManager().addRateData("client.dispatchNoTunnels", now - _start, 0); + getContext().statManager().addRateData("client.dispatchNoTunnels", now - _start); dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_NO_TUNNELS); return; } @@ -524,71 +549,95 @@ public class OutboundClientMessageOneShotJob extends JobImpl { } onReply = new SendSuccessJob(getContext(), sessKey, tsh); onFail = new SendTimeoutJob(getContext(), sessKey, tsh); - selector = new ReplySelector(token); + long expiration = Math.max(_overallExpiration, _start + REPLY_TIMEOUT_MS_MIN); + selector = new ReplySelector(token, expiration); } if (_log.shouldLog(Log.DEBUG)) - _log.debug(getJobId() + ": GarlicMessage in new tunnel msg for " - + _toString + " at " - + _lease.getTunnelId() + " on " - + _lease.getGateway()); - - if (_outTunnel != null) { - if (_log.shouldLog(Log.DEBUG)) - _log.debug(getJobId() + ": Sending msg out " + _outTunnel.getSendTunnelId(0) + " to " + _log.debug(getJobId() + ": Sending msg out " + _outTunnel.getSendTunnelId(0) + " to " + _toString + " at " + _lease.getTunnelId() + " on " + _lease.getGateway()); - DispatchJob dispatchJob = new DispatchJob(getContext(), msg, selector, onReply, onFail, (int)(_overallExpiration-getContext().clock().now())); - //if (false) // dispatch may take 100+ms, so toss it in its own job - // getContext().jobQueue().addJob(dispatchJob); - //else - dispatchJob.runJob(); - } else { - if (_log.shouldLog(Log.WARN)) - _log.warn(getJobId() + ": Could not find any outbound tunnels to send the payload through... this might take a while"); - getContext().statManager().addRateData("client.dispatchNoTunnels", now - _start, 0); - dieFatal(MessageStatusMessage.STATUS_SEND_FAILURE_NO_TUNNELS); - } - _clove = null; - getContext().statManager().addRateData("client.dispatchPrepareTime", now - _start, 0); + DispatchJob dispatchJob = new DispatchJob(getContext(), msg, selector, onReply, onFail); + //if (false) // dispatch may take 100+ms, so toss it in its own job + // getContext().jobQueue().addJob(dispatchJob); + //else + dispatchJob.runJob(); + getContext().statManager().addRateData("client.dispatchPrepareTime", now - _start); if (!wantACK) - getContext().statManager().addRateData("client.dispatchNoACK", 1, 0); + getContext().statManager().addRateData("client.dispatchNoACK", 1); } + /** + * Note: This is run inline by send(), not on the job queue. + * TODO replace with a method + */ private class DispatchJob extends JobImpl { private final GarlicMessage _msg; private final ReplySelector _selector; private final SendSuccessJob _replyFound; private final SendTimeoutJob _replyTimeout; - private final int _timeoutMs; - public DispatchJob(RouterContext ctx, GarlicMessage msg, ReplySelector sel, SendSuccessJob success, SendTimeoutJob timeout, int timeoutMs) { + /** + * @param sel may be null + * @param success non-null if sel non-null + * @param timeout non-null if sel non-null + */ + public DispatchJob(RouterContext ctx, GarlicMessage msg, ReplySelector sel, + SendSuccessJob success, SendTimeoutJob timeout) { super(ctx); _msg = msg; _selector = sel; _replyFound = success; _replyTimeout = timeout; - _timeoutMs = timeoutMs; } public String getName() { return "Outbound client message dispatch"; } public void runJob() { - if (_selector != null) - getContext().messageRegistry().registerPending(_selector, _replyFound, _replyTimeout, _timeoutMs); + if (_selector != null) { + if (_overallExpiration >= _selector.getExpiration()) { + // We use the Message Registry to call our timeout when the selector expires + // Either the success or timeout job will fire, never both. + getContext().messageRegistry().registerPending(_selector, _replyFound, _replyTimeout); + if (_log.shouldLog(Log.INFO)) + _log.info(OutboundClientMessageOneShotJob.this.getJobId() + + ": Reply selector expires " + + DataHelper.formatDuration(_overallExpiration - _selector.getExpiration()) + + " before message, using selector only"); + } else { + // We put our own timeout on the job queue before the selector expires, + // so we can keep waiting for the reply and restore the tags (success-after-failure) + // The timeout job will always fire, even after success. + // We don't bother cancelling the timeout job as JobQueue.removeJob() is a linear search + getContext().messageRegistry().registerPending(_selector, _replyFound, null); + _replyTimeout.getTiming().setStartAfter(_overallExpiration); + getContext().jobQueue().addJob(_replyTimeout); + if (_log.shouldLog(Log.INFO)) + _log.info(OutboundClientMessageOneShotJob.this.getJobId() + + ": Reply selector expires " + + DataHelper.formatDuration(_selector.getExpiration() - _overallExpiration) + + " after message, queueing separate timeout job"); + } + } if (_log.shouldLog(Log.INFO)) _log.info(OutboundClientMessageOneShotJob.this.getJobId() + ": Dispatching message to " + _toString + ": " + _msg); long before = getContext().clock().now(); + + // Note we do not have a first hop fail job, or a success job, here, + // as we do in e.g. build handler. + // Nor do we ever send a STATUS_SEND_BEST_EFFORT_SUCCESS (when no selector) getContext().tunnelDispatcher().dispatchOutbound(_msg, _outTunnel.getSendTunnelId(0), _lease.getTunnelId(), _lease.getGateway()); long dispatchSendTime = getContext().clock().now() - before; //if (_log.shouldLog(Log.INFO)) // _log.info(OutboundClientMessageOneShotJob.this.getJobId() + // ": Dispatching message to " + _toString + " complete"); - getContext().statManager().addRateData("client.dispatchTime", getContext().clock().now() - _start, 0); - getContext().statManager().addRateData("client.dispatchSendTime", dispatchSendTime, 0); + // avg. 6 ms on a 2005-era PC + getContext().statManager().addRateData("client.dispatchTime", getContext().clock().now() - _start); + // avg. 1 ms on a 2005-era PC + getContext().statManager().addRateData("client.dispatchSendTime", dispatchSendTime); } } @@ -690,10 +739,19 @@ public class OutboundClientMessageOneShotJob extends JobImpl { * give up the ghost, this message just aint going through. tell the client. * * this is safe to call multiple times (only tells the client once) + * We may still succeed later. */ private void dieFatal(int status) { - if (_finished.getAndSet(true)) - return; + // never fail twice or fail after success + synchronized(this) { + if (_finished != Result.NONE) { + if (_log.shouldLog(Log.WARN)) + _log.warn(OutboundClientMessageOneShotJob.this.getJobId() + + ": FAIL-AFTER-" + _finished); + return; + } + _finished = Result.FAIL; + } long sendTime = getContext().clock().now() - _start; if (_log.shouldLog(Log.WARN)) @@ -713,11 +771,15 @@ public class OutboundClientMessageOneShotJob extends JobImpl { getContext().messageHistory().sendPayloadMessage(_clientMessageId.getMessageId(), false, sendTime); getContext().clientManager().messageDeliveryStatusUpdate(_from, _clientMessageId, status); getContext().statManager().updateFrequency("client.sendMessageFailFrequency"); - _clove = null; } - /** build the payload clove that will be used for all of the messages, placing the clove in the status structure */ - private boolean buildClove() { + /** + * Build the payload clove that will be used for all of the messages, + * placing the clove in the status structure. + * + * @return null on failure + */ + private PayloadGarlicConfig buildClove() { PayloadGarlicConfig clove = new PayloadGarlicConfig(); DeliveryInstructions instructions = new DeliveryInstructions(); @@ -737,10 +799,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl { DataMessage msg = new DataMessage(getContext()); Payload p = _clientMessage.getPayload(); if (p == null) - return false; + return null; byte d[] = p.getEncryptedData(); if (d == null) - return false; + return null; msg.setData(d); msg.setMessageExpiration(clove.getExpiration()); @@ -749,12 +811,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl { //clove.setRecipientPublicKey(null); //clove.setRequestAck(false); - _clove = clove; - _cloveId = _clove.getId(); //if (_log.shouldLog(Log.DEBUG)) // _log.debug(getJobId() + ": Built payload clove with id " + clove.getId()); - return true; + return clove; } /** @@ -762,23 +822,20 @@ public class OutboundClientMessageOneShotJob extends JobImpl { * sent down the various tunnels to deliver this message * */ - private class ReplySelector implements MessageSelector { + private static class ReplySelector implements MessageSelector { private final long _pendingToken; + private final long _expiration; - public ReplySelector(long token) { + public ReplySelector(long token, long expiration) { _pendingToken = token; - //if (_log.shouldLog(Log.INFO)) - // _log.info(OutboundClientMessageOneShotJob.this.getJobId() - // + ": Reply selector for client message: token=" + token); + _expiration = expiration; } public boolean continueMatching() { - //if (_log.shouldLog(Log.DEBUG)) - // _log.debug(OutboundClientMessageOneShotJob.this.getJobId() - // + ": dont continue matching for token=" + _pendingToken); return false; } - public long getExpiration() { return _overallExpiration; } + + public long getExpiration() { return _expiration; } public boolean isMatch(I2NPMessage inMsg) { if (inMsg.getType() == DeliveryStatusMessage.MESSAGE_TYPE) { @@ -793,14 +850,13 @@ public class OutboundClientMessageOneShotJob extends JobImpl { @Override public String toString() { - return "sending " + _toString + " waiting for token " + _pendingToken - + " for cloveId " + _cloveId; + return "OCMOSJ.RS waiting for token " + _pendingToken + " until " + new Date(_expiration); } } /** - * Called after we get a confirmation that the message was delivered safely - * (hoo-ray!) + * Called after we get a confirmation that the message was delivered safely. + * This may be run after failure. * */ private class SendSuccessJob extends JobImpl implements ReplyJob { @@ -810,7 +866,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl { /** * Create a new success job that will be fired when the message encrypted with * the given session key and bearing the specified tags are confirmed delivered. + * This is only instantiated if we are expecting a reply. * + * @param key may be null + * @param tags may be null */ public SendSuccessJob(RouterContext enclosingContext, SessionKey key, TagSetHandle tags) { super(enclosingContext); @@ -820,24 +879,44 @@ public class OutboundClientMessageOneShotJob extends JobImpl { public String getName() { return "Outbound client message send success"; } + /** + * May be run after SendTimeoutJob, will re-add the tags. + */ public void runJob() { // do we leak tags here? - if (_finished.getAndSet(true)) - return; - long sendTime = getContext().clock().now() - _start; - if (_log.shouldLog(Log.INFO)) - _log.info(OutboundClientMessageOneShotJob.this.getJobId() - + ": SUCCESS! msg " + _clientMessageId - + " acked by DSM after " + sendTime + "ms"); - - if (_key != null && _tags != null && _leaseSet != null) { - SessionKeyManager skm = getContext().clientManager().getClientSessionKeyManager(_from.calculateHash()); - if (skm != null) - skm.tagsAcked(_leaseSet.getEncryptionKey(), _key, _tags); + Result old; + // never succeed twice but we can succeed after fail + synchronized(OutboundClientMessageOneShotJob.this) { + old = _finished; + if (old == Result.SUCCESS) { + if (_log.shouldLog(Log.WARN)) + _log.warn(OutboundClientMessageOneShotJob.this.getJobId() + + ": SUCCESS-AFTER-SUCCESS"); + return; + } + _finished = Result.SUCCESS; + // in sync block so we don't race with SendTimeoutJob + if (_key != null && _tags != null && _leaseSet != null) { + SessionKeyManager skm = getContext().clientManager().getClientSessionKeyManager(_from.calculateHash()); + if (skm != null) + skm.tagsAcked(_leaseSet.getEncryptionKey(), _key, _tags); + } } + + long sendTime = getContext().clock().now() - _start; + if (old == Result.FAIL) { + if (_log.shouldLog(Log.WARN)) + _log.warn(OutboundClientMessageOneShotJob.this.getJobId() + + ": SUCCESS-AFTER-TIMEOUT " + _clientMessageId + + " acked by DSM after " + sendTime + "ms"); + } else if (_log.shouldLog(Log.INFO)) { + _log.info(OutboundClientMessageOneShotJob.this.getJobId() + + ": SUCCESS " + _clientMessageId + + " acked by DSM after " + sendTime + "ms"); + } - long dataMsgId = _cloveId; - getContext().messageHistory().sendPayloadMessage(dataMsgId, true, sendTime); + //long dataMsgId = _cloveId; // fake ID 99999 + getContext().messageHistory().sendPayloadMessage(99999, true, sendTime); getContext().clientManager().messageDeliveryStatusUpdate(_from, _clientMessageId, MessageStatusMessage.STATUS_SEND_GUARANTEED_SUCCESS); // unused @@ -845,7 +924,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl { int size = _clientMessageSize; - getContext().statManager().addRateData("client.sendAckTime", sendTime, 0); + getContext().statManager().addRateData("client.sendAckTime", sendTime); getContext().statManager().addRateData("client.sendMessageSize", _clientMessageSize, sendTime); if (_outTunnel != null) { if (_outTunnel.getLength() > 0) @@ -874,6 +953,13 @@ public class OutboundClientMessageOneShotJob extends JobImpl { private final SessionKey _key; private final TagSetHandle _tags; + /** + * Create a new timeout job that will be fired when the reply is not received. + * This is only instantiated if we are expecting a reply. + * + * @param key may be null + * @param tags may be null + */ public SendTimeoutJob(RouterContext enclosingContext, SessionKey key, TagSetHandle tags) { super(enclosingContext); _key = key; @@ -882,19 +968,29 @@ public class OutboundClientMessageOneShotJob extends JobImpl { public String getName() { return "Outbound client message send timeout"; } + /** + * May be run after SendSuccessJob, will have no effect. + */ public void runJob() { - //if (_log.shouldLog(Log.INFO)) - // _log.info(OutboundClientMessageOneShotJob.this.getJobId() - // + ": Soft timeout through the lease " + _lease); - - // unused - //_lease.setNumFailure(_lease.getNumFailure()+1); - if (_key != null && _tags != null && _leaseSet != null) { - SessionKeyManager skm = getContext().clientManager().getClientSessionKeyManager(_from.calculateHash()); - if (skm != null) - skm.failTags(_leaseSet.getEncryptionKey(), _key, _tags); + Result old; + // never fail after success + synchronized(OutboundClientMessageOneShotJob.this) { + old = _finished; + if (old == Result.SUCCESS) { + if (_log.shouldLog(Log.INFO)) + _log.info(OutboundClientMessageOneShotJob.this.getJobId() + + ": TIMEOUT-AFTER-SUCCESS"); + return; + } + // in sync block so we don't fail after success + if (_key != null && _tags != null && _leaseSet != null) { + SessionKeyManager skm = getContext().clientManager().getClientSessionKeyManager(_from.calculateHash()); + if (skm != null) + skm.failTags(_leaseSet.getEncryptionKey(), _key, _tags); + } } - dieFatal(MessageStatusMessage.STATUS_SEND_BEST_EFFORT_FAILURE); + if (old == Result.NONE) + dieFatal(MessageStatusMessage.STATUS_SEND_BEST_EFFORT_FAILURE); } } } diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/FloodOnlySearchJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/FloodOnlySearchJob.java index 5eb970a24..9fcbbeb79 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/FloodOnlySearchJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/FloodOnlySearchJob.java @@ -103,7 +103,7 @@ class FloodOnlySearchJob extends FloodSearchJob { // This OutNetMessage is never used or sent (setMessage() is never called), it's only // so we can register a reply selector. - _out = getContext().messageRegistry().registerPending(_replySelector, _onReply, _onTimeout, _timeoutMs); + _out = getContext().messageRegistry().registerPending(_replySelector, _onReply, _onTimeout); /******** // We need to randomize our ff selection, else we stay with the same ones since diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/FloodfillVerifyStoreJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/FloodfillVerifyStoreJob.java index e98b86e8e..5e312e2d2 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/FloodfillVerifyStoreJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/FloodfillVerifyStoreJob.java @@ -157,7 +157,9 @@ class FloodfillVerifyStoreJob extends JobImpl { _log.info("Starting verify (stored " + _key + " to " + _sentTo + "), asking " + _target); _sendTime = getContext().clock().now(); _expiration = _sendTime + VERIFY_TIMEOUT; - getContext().messageRegistry().registerPending(new VerifyReplySelector(), new VerifyReplyJob(getContext()), new VerifyTimeoutJob(getContext()), VERIFY_TIMEOUT); + getContext().messageRegistry().registerPending(new VerifyReplySelector(), + new VerifyReplyJob(getContext()), + new VerifyTimeoutJob(getContext())); getContext().tunnelDispatcher().dispatchOutbound(sent, outTunnel.getSendTunnelId(0), _target); } diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/IterativeSearchJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/IterativeSearchJob.java index 23fd34b82..e01709936 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/IterativeSearchJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/IterativeSearchJob.java @@ -185,7 +185,7 @@ class IterativeSearchJob extends FloodSearchJob { MessageSelector replySelector = new IterativeLookupSelector(getContext(), this); ReplyJob onReply = new FloodOnlyLookupMatchJob(getContext(), this); Job onTimeout = new FloodOnlyLookupTimeoutJob(getContext(), this); - _out = getContext().messageRegistry().registerPending(replySelector, onReply, onTimeout, _timeoutMs); + _out = getContext().messageRegistry().registerPending(replySelector, onReply, onTimeout); if (_log.shouldLog(Log.INFO)) _log.info(getJobId() + ": New ISJ for " + (_isLease ? "LS " : "RI ") + diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/SearchJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/SearchJob.java index 136d3e90c..54a938319 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/SearchJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/SearchJob.java @@ -447,7 +447,7 @@ class SearchJob extends JobImpl { if (FloodfillNetworkDatabaseFacade.isFloodfill(router)) _floodfillSearchesOutstanding++; - getContext().messageRegistry().registerPending(sel, reply, new FailedJob(getContext(), router), timeout); + getContext().messageRegistry().registerPending(sel, reply, new FailedJob(getContext(), router)); // TODO pass a priority to the dispatcher getContext().tunnelDispatcher().dispatchOutbound(msg, outTunnelId, to); } diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/SingleSearchJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/SingleSearchJob.java index d90f10e67..1c7995089 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/SingleSearchJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/SingleSearchJob.java @@ -34,7 +34,7 @@ class SingleSearchJob extends FloodOnlySearchJob { @Override public void runJob() { - _onm = getContext().messageRegistry().registerPending(_replySelector, _onReply, _onTimeout, _timeoutMs); + _onm = getContext().messageRegistry().registerPending(_replySelector, _onReply, _onTimeout); DatabaseLookupMessage dlm = new DatabaseLookupMessage(getContext(), true); TunnelInfo replyTunnel = getContext().tunnelManager().selectInboundExploratoryTunnel(_to); TunnelInfo outTunnel = getContext().tunnelManager().selectOutboundExploratoryTunnel(_to); diff --git a/router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java b/router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java index ad2322c47..bd45f1924 100644 --- a/router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java +++ b/router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java @@ -376,7 +376,7 @@ class StoreJob extends JobImpl { if (_log.shouldLog(Log.DEBUG)) _log.debug("sending store to " + peer.getIdentity().getHash() + " through " + outTunnel + ": " + msg); - getContext().messageRegistry().registerPending(selector, onReply, onFail, (int)(expiration - getContext().clock().now())); + getContext().messageRegistry().registerPending(selector, onReply, onFail); getContext().tunnelDispatcher().dispatchOutbound(msg, outTunnel.getSendTunnelId(0), null, to); } else { if (_log.shouldLog(Log.WARN)) @@ -456,7 +456,7 @@ class StoreJob extends JobImpl { _log.debug("sending store to " + peer.getIdentity().getHash() + " through " + outTunnel + ": " + sent); //_log.debug("Expiration is " + new Date(sent.getMessageExpiration())); } - getContext().messageRegistry().registerPending(selector, onReply, onFail, (int)(expiration - getContext().clock().now())); + getContext().messageRegistry().registerPending(selector, onReply, onFail); getContext().tunnelDispatcher().dispatchOutbound(sent, outTunnel.getSendTunnelId(0), null, to); } else { if (_log.shouldLog(Log.WARN)) diff --git a/router/java/src/net/i2p/router/peermanager/PeerTestJob.java b/router/java/src/net/i2p/router/peermanager/PeerTestJob.java index a2096edcf..0919b5781 100644 --- a/router/java/src/net/i2p/router/peermanager/PeerTestJob.java +++ b/router/java/src/net/i2p/router/peermanager/PeerTestJob.java @@ -149,7 +149,7 @@ public class PeerTestJob extends JobImpl { PeerReplyFoundJob reply = new PeerReplyFoundJob(getContext(), peer, inTunnel, outTunnel); PeerReplyTimeoutJob timeoutJob = new PeerReplyTimeoutJob(getContext(), peer, inTunnel, outTunnel, sel); - getContext().messageRegistry().registerPending(sel, reply, timeoutJob, timeoutMs); + getContext().messageRegistry().registerPending(sel, reply, timeoutJob); getContext().tunnelDispatcher().dispatchOutbound(msg, outTunnelId, null, peer.getIdentity().getHash()); } diff --git a/router/java/src/net/i2p/router/transport/OutboundMessageRegistry.java b/router/java/src/net/i2p/router/transport/OutboundMessageRegistry.java index 9eabbd719..584da497f 100644 --- a/router/java/src/net/i2p/router/transport/OutboundMessageRegistry.java +++ b/router/java/src/net/i2p/router/transport/OutboundMessageRegistry.java @@ -121,7 +121,7 @@ public class OutboundMessageRegistry { } } - List rv = null; + List rv; if (matchedSelectors != null) { rv = new ArrayList(matchedSelectors.size()); for (MessageSelector sel : matchedSelectors) { @@ -162,19 +162,22 @@ public class OutboundMessageRegistry { /** * Registers a new, empty OutNetMessage, with the reply and timeout jobs specified. + * The onTimeout job is called at replySelector.getExpiration() (if no reply is received by then) * * @param replySelector non-null; The same selector may be used for more than one message. - * @param onReply may be null - * @param onTimeout Also called on failed send; may be null - * @return an ONM where getMessage() is null. Use it to call unregisterPending() later if desired. + * @param onReply non-null + * @param onTimeout may be null + * @return a dummy OutNetMessage where getMessage() is null. Use it to call unregisterPending() later if desired. */ - public OutNetMessage registerPending(MessageSelector replySelector, ReplyJob onReply, Job onTimeout, int timeoutMs) { - OutNetMessage msg = new OutNetMessage(_context, _context.clock().now() + timeoutMs); + public OutNetMessage registerPending(MessageSelector replySelector, ReplyJob onReply, Job onTimeout) { + OutNetMessage msg = new OutNetMessage(_context); msg.setOnFailedReplyJob(onTimeout); - msg.setOnFailedSendJob(onTimeout); msg.setOnReplyJob(onReply); msg.setReplySelector(replySelector); registerPending(msg, true); + if (_log.shouldLog(Log.DEBUG)) + _log.debug("Registered: " + replySelector + " with reply job " + onReply + + " and timeout job " + onTimeout); return msg; } @@ -279,6 +282,7 @@ public class OutboundMessageRegistry { } } } + boolean log = _log.shouldLog(Log.DEBUG); if (!removing.isEmpty()) { for (MessageSelector sel : removing) { OutNetMessage msg = null; @@ -297,17 +301,31 @@ public class OutboundMessageRegistry { Job fail = msg.getOnFailedReplyJob(); if (fail != null) _context.jobQueue().addJob(fail); + if (log) + _log.debug("Expired: " + sel + " with timeout job " + fail); } else if (msgs != null) { _activeMessages.removeAll(msgs); for (OutNetMessage m : msgs) { Job fail = m.getOnFailedReplyJob(); if (fail != null) _context.jobQueue().addJob(fail); + if (log) + _log.debug("Expired: " + sel + " with timeout job(s) " + fail); } + } else { + if (log) + _log.debug("Expired: " + sel + " with no known messages"); } } } + if (log) { + int e = removing.size(); + int r = _selectors.size(); + int a = _activeMessages.size(); + if (r > 0 || e > 0 || a > 0) + _log.debug("Expired: " + e + " remaining: " + r + " active: " + a); + } if (_nextExpire <= now) _nextExpire = now + 10*1000; schedule(_nextExpire - now); diff --git a/router/java/src/net/i2p/router/tunnel/pool/TestJob.java b/router/java/src/net/i2p/router/tunnel/pool/TestJob.java index 6b3f0daed..52923fa72 100644 --- a/router/java/src/net/i2p/router/tunnel/pool/TestJob.java +++ b/router/java/src/net/i2p/router/tunnel/pool/TestJob.java @@ -104,7 +104,7 @@ class TestJob extends JobImpl { ReplySelector sel = new ReplySelector(getContext(), m.getMessageId(), testExpiration); OnTestReply onReply = new OnTestReply(getContext()); OnTestTimeout onTimeout = new OnTestTimeout(getContext()); - OutNetMessage msg = getContext().messageRegistry().registerPending(sel, onReply, onTimeout, testPeriod); + OutNetMessage msg = getContext().messageRegistry().registerPending(sel, onReply, onTimeout); onReply.setSentMessage(msg); sendTest(m); }