I2P_Developers/i2p.itoopie
3
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

big ol' memory, cpu usage, and shutdown handling update. main changes include:

Browse Source 
* rather than have all jobs created hooked into the clock for offset updates, have the jobQueue stay hooked up and update any active jobs accordingly (killing a memory leak of a JobTiming objects - one per job)
* dont go totally insane during shutdown and log like mad (though the clientApp things still log like mad, since they don't know the router is going down)
* adjust memory buffer sizes based on real world values so we don't have to expand/contract a lot
* dont display things that are completely useless (who cares what the first 32 bytes of a public key are?)
* reduce temporary object creation
* use more efficient collections at times
* on shutdown, log some state information (ready/timed jobs, pending messages, etc)
* explicit GC every 10 jobs.  yeah, not efficient, but just for now we'll keep 'er in there
* only reread the router config file if it changes (duh)
This commit is contained in:
jrandom
2004-05-16 04:54:50 +00:00
committed by zzz
parent 8c6bf5a1cc
commit ff0023a889
32 changed files with 195 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
router/java/src/net/i2p/data/i2np/TunnelMessage.java
View File

@ -89,7 +89,7 @@ public class TunnelMessage extends I2NPMessageImpl {
if ( (_tunnelId == null) || (_data == null) || (_data.length <= 0) )
throw new I2NPMessageException("Not enough data to write out");
ByteArrayOutputStream os = new ByteArrayOutputStream(32);
ByteArrayOutputStream os = new ByteArrayOutputStream(4096);
try {
_tunnelId.writeBytes(os);
if (_log.shouldLog(Log.DEBUG))

4
router/java/src/net/i2p/router/JobImpl.java
View File

@ -9,6 +9,7 @@ package net.i2p.router;
*/
import net.i2p.util.Clock;
import net.i2p.util.Log;
/**
* Base implementation of a Job
*/
@ -39,7 +40,8 @@ public abstract class JobImpl implements Job {
}
void addedToQueue() {
_addedBy = new Exception();
if (_context.logManager().getLog(JobImpl.class).shouldLog(Log.DEBUG))
_addedBy = new Exception();
}
public Exception getAddedBy() { return _addedBy; }

56
router/java/src/net/i2p/router/JobQueue.java
View File

@ -90,9 +90,6 @@ public class JobQueue {
private final static int DEFAULT_MAX_WAITING_JOBS = 20;
private final static String PROP_MAX_WAITING_JOBS = "router.maxWaitingJobs";
static {
}
/**
* queue runners wait on this whenever they're not doing anything, and
* this gets notified *once* whenever there are ready jobs
@ -229,7 +226,17 @@ public class JobQueue {
}
public void allowParallelOperation() { _allowParallelOperation = true; }
void shutdown() { _alive = false; }
void shutdown() {
_alive = false;
StringBuffer buf = new StringBuffer(1024);
buf.append("jobs: \nready jobs: ").append(_readyJobs.size()).append("\n\t");
for (int i = 0; i < _readyJobs.size(); i++)
buf.append(_readyJobs.get(i).toString()).append("\n\t");
buf.append("\n\ntimed jobs: ").append(_timedJobs.size()).append("\n\t");
for (int i = 0; i < _timedJobs.size(); i++)
buf.append(_timedJobs.get(i).toString()).append("\n\t");
_log.log(Log.CRIT, buf.toString());
}
boolean isAlive() { return _alive; }
/**
@ -276,7 +283,7 @@ public class JobQueue {
private int checkJobTimings() {
boolean newJobsReady = false;
long now = _context.clock().now();
ArrayList toAdd = new ArrayList(4);
ArrayList toAdd = null;
synchronized (_timedJobs) {
for (int i = 0; i < _timedJobs.size(); i++) {
Job j = (Job)_timedJobs.get(i);
@ -285,6 +292,7 @@ public class JobQueue {
if (j instanceof JobImpl)
((JobImpl)j).madeReady();
if (toAdd == null) toAdd = new ArrayList(4);
toAdd.add(j);
_timedJobs.remove(i);
i--; // so the index stays consistent
@ -294,7 +302,15 @@ public class JobQueue {
int ready = 0;
synchronized (_readyJobs) {
_readyJobs.addAll(toAdd);
if (toAdd != null) {
// rather than addAll, which allocs a byte array rv before adding,
// we iterate, since toAdd is usually going to only be 1 or 2 entries
// and since readyJobs will often have the space, we can avoid the
// extra alloc. (no, i'm not just being insane - i'm updating this based
// on some profiling data ;)
for (int i = 0; i < toAdd.size(); i++)
_readyJobs.add(toAdd.get(i));
}
ready = _readyJobs.size();
}
@ -399,10 +415,38 @@ public class JobQueue {
public void offsetChanged(long delta) {
if (_lastLimitUpdated > 0)
_lastLimitUpdated += delta;
updateJobTimings(delta);
}
}
/**
* Update the clock data for all jobs in process or scheduled for
* completion.
*/
private void updateJobTimings(long delta) {
synchronized (_timedJobs) {
for (int i = 0; i < _timedJobs.size(); i++) {
Job j = (Job)_timedJobs.get(i);
j.getTiming().offsetChanged(delta);
}
}
synchronized (_readyJobs) {
for (int i = 0; i < _readyJobs.size(); i++) {
Job j = (Job)_readyJobs.get(i);
j.getTiming().offsetChanged(delta);
}
}
synchronized (_runnerLock) {
for (Iterator iter = _queueRunners.values().iterator(); iter.hasNext(); ) {
JobQueueRunner runner = (JobQueueRunner)iter.next();
Job job = runner.getCurrentJob();
if (job != null)
job.getTiming().offsetChanged(delta);
}
}
}
/**
* calculate and update the job timings
* if it was lagged too much or took too long to run, spit out

16
router/java/src/net/i2p/router/JobQueueRunner.java
View File

@ -34,12 +34,14 @@ class JobQueueRunner implements Runnable {
public void stopRunning() { _keepRunning = false; }
public void run() {
long lastActive = _context.clock().now();
long jobNum = 0;
while ( (_keepRunning) && (_context.jobQueue().isAlive()) ) {
try {
Job job = _context.jobQueue().getNext();
if (job == null) {
if (_log.shouldLog(Log.ERROR))
_log.error("getNext returned null - dead?");
if (_context.router().isAlive())
if (_log.shouldLog(Log.ERROR))
_log.error("getNext returned null - dead?");
continue;
}
long now = _context.clock().now();
@ -85,13 +87,18 @@ class JobQueueRunner implements Runnable {
lastActive = _context.clock().now();
_lastJob = _currentJob;
_currentJob = null;
jobNum++;
if ( (jobNum % 10) == 0)
System.gc();
} catch (Throwable t) {
if (_log.shouldLog(Log.CRIT))
_log.log(Log.CRIT, "WTF, error running?", t);
}
}
if (_log.shouldLog(Log.CRIT))
_log.log(Log.CRIT, "Queue runner " + _id + " exiting");
if (_context.router().isAlive())
if (_log.shouldLog(Log.CRIT))
_log.log(Log.CRIT, "Queue runner " + _id + " exiting");
_context.jobQueue().removeRunner(_id);
}
@ -102,6 +109,7 @@ class JobQueueRunner implements Runnable {
try {
if (_log.shouldLog(Log.CRIT))
_log.log(Log.CRIT, "Router ran out of memory, shutting down", oom);
_log.log(Log.CRIT, _currentJob.getClass().getName());
_context.router().shutdown();
} catch (Throwable t) {
System.err.println("***Router ran out of memory, shutting down hard");

2
router/java/src/net/i2p/router/JobTiming.java
View File

@ -25,7 +25,7 @@ public class JobTiming implements Clock.ClockUpdateListener {
_start = context.clock().now();
_actualStart = 0;
_actualEnd = 0;
context.clock().addUpdateListener(this);
//context.clock().addUpdateListener(this);
}
/**

10
router/java/src/net/i2p/router/MessageHistory.java
View File

@ -5,7 +5,7 @@ import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.ArrayList;
import java.util.List;
import java.util.TimeZone;
@ -94,7 +94,7 @@ public class MessageHistory {
_doLog = DEFAULT_KEEP_MESSAGE_HISTORY;
_historyFile = filename;
_localIdent = getName(_context.routerHash());
_unwrittenEntries = new LinkedList();
_unwrittenEntries = new ArrayList(64);
updateSettings();
addEntry(getPrefix() + "** Router initialized (started up or changed identities)");
_context.jobQueue().addJob(_writeJob);
@ -338,7 +338,7 @@ public class MessageHistory {
*/
public void sendMessage(String messageType, long messageId, Date expiration, Hash peer, boolean sentOk) {
if (!_doLog) return;
StringBuffer buf = new StringBuffer(128);
StringBuffer buf = new StringBuffer(256);
buf.append(getPrefix());
buf.append("send [").append(messageType).append("] message [").append(messageId).append("] ");
buf.append("to [").append(getName(peer)).append("] ");
@ -363,7 +363,7 @@ public class MessageHistory {
*/
public void receiveMessage(String messageType, long messageId, Date expiration, Hash from, boolean isValid) {
if (!_doLog) return;
StringBuffer buf = new StringBuffer(128);
StringBuffer buf = new StringBuffer(256);
buf.append(getPrefix());
buf.append("receive [").append(messageType).append("] with id [").append(messageId).append("] ");
if (from != null)
@ -473,7 +473,7 @@ public class MessageHistory {
if (_doPause) return;
List entries = null;
synchronized (_unwrittenEntries) {
entries = new LinkedList(_unwrittenEntries);
entries = new ArrayList(_unwrittenEntries);
_unwrittenEntries.clear();
}
writeEntries(entries);

8
router/java/src/net/i2p/router/MessageValidator.java
View File

@ -35,7 +35,7 @@ public class MessageValidator {
public MessageValidator(RouterContext context) {
_log = context.logManager().getLog(MessageValidator.class);
_receivedIdExpirations = new TreeMap();
_receivedIds = new HashSet(1024);
_receivedIds = new HashSet(32*1024);
_receivedIdLock = new Object();
_context = context;
}
@ -130,4 +130,10 @@ public class MessageValidator {
if (_log.shouldLog(Log.INFO))
_log.info("Cleaned out " + toRemoveDates.size() + " expired messageIds, leaving " + _receivedIds.size() + " remaining");
}
void shutdown() {
StringBuffer buf = new StringBuffer(1024);
buf.append("Validated messages: ").append(_receivedIds.size());
_log.log(Log.CRIT, buf.toString());
}
}

4
router/java/src/net/i2p/router/OutNetMessage.java
View File

@ -115,7 +115,7 @@ public class OutNetMessage {
public long getMessageSize() {
if (_messageSize <= 0) {
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream(4096); // large enough to hold most messages
ByteArrayOutputStream baos = new ByteArrayOutputStream(2048); // large enough to hold most messages
_message.writeBytes(baos);
long sz = baos.size();
baos.reset();
@ -136,7 +136,7 @@ public class OutNetMessage {
ByteArrayOutputStream baos = new ByteArrayOutputStream(4096); // large enough to hold most messages
_message.writeBytes(baos);
byte data[] = baos.toByteArray();
baos.reset();
_messageSize = data.length;
return data;
} catch (DataFormatException dfe) {
_log.error("Error serializing the I2NPMessage for the OutNetMessage", dfe);

4
router/java/src/net/i2p/router/Router.java
View File

@ -392,6 +392,8 @@ public class Router {
try { _context.netDb().shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the networkDb", t); }
try { _context.commSystem().shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the comm system", t); }
try { _context.peerManager().shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the peer manager", t); }
try { _context.messageRegistry().shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the message registry", t); }
try { _context.messageValidator().shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the message validator", t); }
try { _sessionKeyPersistenceHelper.shutdown(); } catch (Throwable t) { _log.log(Log.CRIT, "Error shutting down the session key manager", t); }
dumpStats();
_log.log(Log.CRIT, "Shutdown complete", new Exception("Shutdown"));
@ -413,7 +415,7 @@ public class Router {
private class ShutdownHook extends Thread {
public void run() {
_log.log(Log.CRIT, "Shutting down the router...", new Exception("Shutting down"));
_log.log(Log.CRIT, "Shutting down the router...");
shutdown();
}
}

2
router/java/src/net/i2p/router/StatisticsManager.java
View File

@ -162,7 +162,7 @@ public class StatisticsManager implements Service {
}
private static String renderRate(Rate rate, boolean fudgeQuantity) {
StringBuffer buf = new StringBuffer(255);
StringBuffer buf = new StringBuffer(128);
buf.append(num(rate.getAverageValue())).append(';');
buf.append(num(rate.getExtremeAverageValue())).append(';');
buf.append(pct(rate.getPercentageOfLifetimeValue())).append(';');

5
router/java/src/net/i2p/router/client/ClientConnectionRunner.java
View File

@ -109,8 +109,9 @@ public class ClientConnectionRunner {
/** die a horrible death */
void stopRunning() {
if (_dead) return;
_log.error("Stop the I2CP connection! current leaseSet: "
+ _currentLeaseSet, new Exception("Stop client connection"));
if (_context.router().isAlive())
_log.error("Stop the I2CP connection! current leaseSet: "
+ _currentLeaseSet, new Exception("Stop client connection"));
_dead = true;
// we need these keys to unpublish the leaseSet
if (_reader != null) _reader.stopReading();

16
router/java/src/net/i2p/router/client/ClientListenerRunner.java
View File

@ -73,28 +73,34 @@ public class ClientListenerRunner implements Runnable {
socket.close();
}
} catch (IOException ioe) {
_log.error("Server error accepting", ioe);
if (_context.router().isAlive())
_log.error("Server error accepting", ioe);
} catch (Throwable t) {
_log.error("Fatal error running client listener - killing the thread!", t);
if (_context.router().isAlive())
_log.error("Fatal error running client listener - killing the thread!", t);
return;
}
}
} catch (IOException ioe) {
_log.error("Error listening on port " + _port, ioe);
if (_context.router().isAlive())
_log.error("Error listening on port " + _port, ioe);
}
if (_socket != null) {
try { _socket.close(); } catch (IOException ioe) {}
_socket = null;
}
if (!_context.router().isAlive()) break;
_log.error("Error listening, waiting " + _nextFailDelay + "ms before we try again");
try { Thread.sleep(_nextFailDelay); } catch (InterruptedException ie) {}
curDelay += _nextFailDelay;
_nextFailDelay *= 5;
}
_log.error("CANCELING I2CP LISTEN. delay = " + curDelay, new Exception("I2CP Listen cancelled!!!"));
if (_context.router().isAlive())
_log.error("CANCELING I2CP LISTEN. delay = " + curDelay, new Exception("I2CP Listen cancelled!!!"));
_running = false;
}

14
router/java/src/net/i2p/router/peermanager/ProfileOrganizer.java
View File

@ -75,11 +75,11 @@ public class ProfileOrganizer {
public ProfileOrganizer(RouterContext context) {
_context = context;
_log = context.logManager().getLog(ProfileOrganizer.class);
_fastAndReliablePeers = new HashMap(64);
_reliablePeers = new HashMap(512);
_wellIntegratedPeers = new HashMap(256);
_notFailingPeers = new HashMap(1024);
_failingPeers = new HashMap(4096);
_fastAndReliablePeers = new HashMap(16);
_reliablePeers = new HashMap(16);
_wellIntegratedPeers = new HashMap(16);
_notFailingPeers = new HashMap(16);
_failingPeers = new HashMap(16);
_strictReliabilityOrder = new TreeSet(new InverseReliabilityComparator());
_thresholdSpeedValue = 0.0d;
_thresholdReliabilityValue = 0.0d;
@ -466,11 +466,9 @@ public class ProfileOrganizer {
all.remove(_us);
howMany -= matches.size();
Collections.shuffle(all, _random);
Set rv = new HashSet(howMany);
for (int i = 0; i < howMany && i < all.size(); i++) {
rv.add(all.get(i));
matches.add(all.get(i));
}
matches.addAll(rv);
}
/**

15
router/java/src/net/i2p/router/startup/ReadConfigJob.java
View File

@ -26,6 +26,7 @@ import net.i2p.router.RouterContext;
*/
public class ReadConfigJob extends JobImpl {
private final static long DELAY = 30*1000; // reread every 30 seconds
private long _lastRead = -1;
public ReadConfigJob(RouterContext ctx) {
super(ctx);
@ -33,11 +34,23 @@ public class ReadConfigJob extends JobImpl {
public String getName() { return "Read Router Configuration"; }
public void runJob() {
doRead(_context);
if (shouldReread()) {
doRead(_context);
_lastRead = _context.clock().now();
}
getTiming().setStartAfter(_context.clock().now() + DELAY);
_context.jobQueue().addJob(this);
}
private boolean shouldReread() {
File configFile = new File(_context.router().getConfigFilename());
if (!configFile.exists()) return false;
if (configFile.lastModified() > _lastRead)
return true;
else
return false;
}
public static void doRead(RouterContext ctx) {
Router r = ctx.router();
String f = r.getConfigFilename();

9
router/java/src/net/i2p/router/transport/OutboundMessageRegistry.java
View File

@ -41,6 +41,15 @@ public class OutboundMessageRegistry {
_context.jobQueue().addJob(new CleanupPendingMessagesJob());
}
public void shutdown() {
StringBuffer buf = new StringBuffer(1024);
buf.append("Pending messages: ").append(_pendingMessages.size()).append("\n");
for (Iterator iter = _pendingMessages.values().iterator(); iter.hasNext(); ) {
buf.append(iter.next().toString()).append("\n\t");
}
_log.log(Log.CRIT, buf.toString());
}
public List getOriginalMessages(I2NPMessage message) {
HashSet matches = new HashSet(4);
long beforeSync = _context.clock().now();

5
router/java/src/net/i2p/router/transport/VMCommSystem.java
View File

@ -61,8 +61,9 @@ public class VMCommSystem extends CommSystemFacade {
} else {
_context.jobQueue().addJob(msg.getOnSendJob());
_context.profileManager().messageSent(msg.getTarget().getIdentity().getHash(), "vm", sendTime, msg.getMessageSize());
_context.statManager().addRateData("transport.sendMessageSize", msg.getMessageSize(), sendTime);
peerSys.receive(msg.getMessage().toByteArray(), _context.routerHash());
byte data[] = msg.getMessageData();
_context.statManager().addRateData("transport.sendMessageSize", data.length, sendTime);
peerSys.receive(data, _context.routerHash());
//_context.jobQueue().addJob(new SendJob(peerSys, msg.getMessage(), _context));
sendSuccessful = true;
}

3
router/java/src/net/i2p/router/tunnelmanager/TunnelPool.java
View File

@ -568,7 +568,8 @@ class TunnelPool {
public void shutdown() {
if (_log.shouldLog(Log.INFO)) _log.info("Shutting down tunnel pool");
_persistenceHelper.writePool(this);
if (_persistenceHelper != null)
_persistenceHelper.writePool(this);
_isLive = false; // the subjobs [should] check getIsLive() on each run
_outboundTunnels = null;
_freeInboundTunnels = null;