Changeset 346faa3


Ignore:
Timestamp:
Aug 24, 2005 10:55:25 PM (15 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
0473e08
Parents:
5ec6dca6
git-author:
jrandom <jrandom> (08/24/05 22:55:25)
git-committer:
zzz <zzz@…> (08/24/05 22:55:25)
Message:

2005-08-24 jrandom

  • Catch errors with corrupt tunnel messages more gracefully (no need to kill the thread and cause an OOM…)
  • Don't skip shitlisted peers for netDb store messages, as they aren't necessarily shitlisted by other people (though they probably are).
  • Adjust the netDb store per-peer timeout based on each particular peer's profile (timeout = 4x their average netDb store response time)
  • Don't republish leaseSets to *failed* peers - send them to peers who replied but just didn't know the value.
  • Set a 5 second timeout on the I2PTunnelHTTPServer reading the client's HTTP headers, rather than blocking indefinitely. HTTP headers should be sent entirely within the first streaming packet anyway, so this won't be a problem.
  • Don't use the I2PTunnel*Server handler thread pool by default, as it may prevent any clients from accessing the server if the handlers get blocked by the streaming lib or other issues.
  • Don't overwrite a known status (OK/ERR-Reject/ERR-SymmetricNAT) with Unknown.
Files:
14 edited

Legend:

Unmodified
Added
Removed
  • apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelHTTPServer.java

    r5ec6dca6 r346faa3  
    6161        //threads.
    6262        try {
    63             socket.setReadTimeout(readTimeout);
     63            // give them 5 seconds to send in the HTTP request
     64            socket.setReadTimeout(5*1000);
    6465            String modifiedHeader = getModifiedHeader(socket);
    6566            if (_log.shouldLog(Log.DEBUG))
    6667                _log.debug("Modified header: [" + modifiedHeader + "]");
    6768
     69            socket.setReadTimeout(readTimeout);
    6870            Socket s = new Socket(remoteHost, remotePort);
    6971            afterSocket = getTunnel().getContext().clock().now();
  • apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelServer.java

    r5ec6dca6 r346faa3  
    1212import java.net.Socket;
    1313import java.net.SocketException;
     14import java.net.ConnectException;
    1415import java.util.Iterator;
    1516import java.util.Properties;
     
    4041    protected InetAddress remoteHost;
    4142    protected int remotePort;
     43    private boolean _usePool;
    4244
    4345    private Logging l;
     
    4749    protected long readTimeout = DEFAULT_READ_TIMEOUT;
    4850
     51    private static final boolean DEFAULT_USE_POOL = false;
     52   
    4953    public I2PTunnelServer(InetAddress host, int port, String privData, Logging l, EventDispatcher notifyThis, I2PTunnel tunnel) {
    5054        super(host + ":" + port + " <- " + privData, notifyThis, tunnel);
    5155        ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decode(privData));
     56        String usePool = tunnel.getClientOptions().getProperty("i2ptunnel.usePool");
     57        if (usePool != null)
     58            _usePool = "true".equalsIgnoreCase(usePool);
     59        else
     60            _usePool = DEFAULT_USE_POOL;
    5261        init(host, port, bais, privData, l);
    5362    }
     
    5665                           EventDispatcher notifyThis, I2PTunnel tunnel) {
    5766        super(host + ":" + port + " <- " + privkeyname, notifyThis, tunnel);
     67        String usePool = tunnel.getClientOptions().getProperty("i2ptunnel.usePool");
     68        if (usePool != null)
     69            _usePool = "true".equalsIgnoreCase(usePool);
     70        else
     71            _usePool = DEFAULT_USE_POOL;
    5872        try {
    5973            init(host, port, new FileInputStream(privkey), privkeyname, l);
     
    6680    public I2PTunnelServer(InetAddress host, int port, InputStream privData, String privkeyname, Logging l,  EventDispatcher notifyThis, I2PTunnel tunnel) {
    6781        super(host + ":" + port + " <- " + privkeyname, notifyThis, tunnel);
     82        String usePool = tunnel.getClientOptions().getProperty("i2ptunnel.usePool");
     83        if (usePool != null)
     84            _usePool = "true".equalsIgnoreCase(usePool);
     85        else
     86            _usePool = DEFAULT_USE_POOL;
    6887        init(host, port, privData, privkeyname, l);
    6988    }
     
    179198   
    180199    public void run() {
     200        if (shouldUsePool()) {
    181201            I2PServerSocket i2pss = sockMgr.getServerSocket();
    182202            int handlers = getHandlerCount();
     
    185205                handler.start();
    186206            }
    187             /*
     207        } else {
     208            I2PServerSocket i2pss = sockMgr.getServerSocket();
    188209            while (true) {
    189                 I2PSocket i2ps = i2pss.accept();
    190                 if (i2ps == null) throw new I2PException("I2PServerSocket closed");
    191                 I2PThread t = new I2PThread(new Handler(i2ps));
    192                 t.start();
    193             }
    194              */
    195     }
    196    
     210                try {
     211                    final I2PSocket i2ps = i2pss.accept();
     212                    if (i2ps == null) throw new I2PException("I2PServerSocket closed");
     213                    new I2PThread(new Runnable() { public void run() { blockingHandle(i2ps); } }).start();
     214                } catch (I2PException ipe) {
     215                    if (_log.shouldLog(Log.ERROR))
     216                        _log.error("Error accepting - KILLING THE TUNNEL SERVER", ipe);
     217                    return;
     218                } catch (ConnectException ce) {
     219                    if (_log.shouldLog(Log.ERROR))
     220                        _log.error("Error accepting", ce);
     221                    // not killing the server..
     222                }
     223            }
     224        }
     225    }
     226   
     227    public boolean shouldUsePool() { return _usePool; }
    197228   
    198229    /**
  • core/java/src/net/i2p/data/RouterInfo.java

    r5ec6dca6 r346faa3  
    5151
    5252    public static final String PROP_NETWORK_ID = "netId";
     53    public static final String PROP_CAPABILITIES = "caps";
    5354   
    5455    public RouterInfo() {
     
    299300        return -1;
    300301    }
     302
     303    /**
     304     * what special capabilities this router offers
     305     *
     306     */
     307    public String getCapabilities() {
     308        if (_options == null) return "";
     309        String capabilities = null;
     310        synchronized (_options) {
     311            capabilities = _options.getProperty(PROP_CAPABILITIES);
     312        }
     313        if (capabilities != null)
     314            return capabilities;
     315        else
     316            return "";
     317    }
    301318       
    302319    /**
  • core/java/src/net/i2p/stat/RateStat.java

    r5ec6dca6 r346faa3  
    6565            rv[i] = _rates[i].getPeriod();
    6666        return rv;
     67    }
     68
     69    public double getLifetimeAverageValue() {
     70        if ( (_rates == null) || (_rates.length <= 0) ) return 0;
     71        return _rates[0].getLifetimeAverageValue();
     72    }
     73    public double getLifetimeEventCount() {
     74        if ( (_rates == null) || (_rates.length <= 0) ) return 0;
     75        return _rates[0].getLifetimeEventCount();
    6776    }
    6877
  • history.txt

    r5ec6dca6 r346faa3  
    1 $Id: history.txt,v 1.228 2005/08/21 13:39:06 jrandom Exp $
     1$Id: history.txt,v 1.229 2005/08/23 16:25:49 jrandom Exp $
     2
     32005-08-24  jrandom
     4    * Catch errors with corrupt tunnel messages more gracefully (no need to
     5      kill the thread and cause an OOM...)
     6    * Don't skip shitlisted peers for netDb store messages, as they aren't
     7      necessarily shitlisted by other people (though they probably are).
     8    * Adjust the netDb store per-peer timeout based on each particular peer's
     9      profile (timeout = 4x their average netDb store response time)   
     10    * Don't republish leaseSets to *failed* peers - send them to peers who
     11      replied but just didn't know the value.
     12    * Set a 5 second timeout on the I2PTunnelHTTPServer reading the client's
     13      HTTP headers, rather than blocking indefinitely.  HTTP headers should be
     14      sent entirely within the first streaming packet anyway, so this won't be
     15      a problem.
     16    * Don't use the I2PTunnel*Server handler thread pool by default, as it may
     17      prevent any clients from accessing the server if the handlers get
     18      blocked by the streaming lib or other issues.
     19    * Don't overwrite a known status (OK/ERR-Reject/ERR-SymmetricNAT) with
     20      Unknown.
    221
    3222005-08-23  jrandom
  • router/java/src/net/i2p/router/RouterVersion.java

    r5ec6dca6 r346faa3  
    1616 */
    1717public class RouterVersion {
    18     public final static String ID = "$Revision: 1.217 $ $Date: 2005/08/21 13:39:05 $";
     18    public final static String ID = "$Revision: 1.218 $ $Date: 2005/08/23 16:25:49 $";
    1919    public final static String VERSION = "0.6.0.3";
    20     public final static long BUILD = 1;
     20    public final static long BUILD = 2;
    2121    public static void main(String args[]) {
    2222        System.out.println("I2P Router version: " + VERSION);
  • router/java/src/net/i2p/router/networkdb/kademlia/KademliaNetworkDatabaseFacade.java

    r5ec6dca6 r346faa3  
    3939import net.i2p.router.networkdb.DatabaseStoreMessageHandler;
    4040import net.i2p.router.networkdb.PublishLocalRouterInfoJob;
     41import net.i2p.router.peermanager.PeerProfile;
    4142import net.i2p.util.Log;
    4243
     
    805806        return routers;
    806807    }
     808
     809    /** smallest allowed period */
     810    private static final int MIN_PER_PEER_TIMEOUT = 1*1000;
     811    private static final int MAX_PER_PEER_TIMEOUT = 5*1000;
     812   
     813    public int getPeerTimeout(Hash peer) {
     814        PeerProfile prof = _context.profileOrganizer().getProfile(peer);
     815        double responseTime = prof.getDbResponseTime().getLifetimeAverageValue();
     816        if (responseTime < MIN_PER_PEER_TIMEOUT)
     817            responseTime = MIN_PER_PEER_TIMEOUT;
     818        else if (responseTime > MAX_PER_PEER_TIMEOUT)
     819            responseTime = MAX_PER_PEER_TIMEOUT;
     820        return 4 * (int)responseTime;  // give it up to 4x the average response time
     821    }
    807822   
    808823    public void renderStatusHTML(Writer out) throws IOException {
  • router/java/src/net/i2p/router/networkdb/kademlia/SearchJob.java

    r5ec6dca6 r346faa3  
    136136        int rv = -1;
    137137        RateStat rs = getContext().statManager().getRate("netDb.successTime");
    138         if (rs != null) {
    139             Rate r = rs.getRate(rs.getPeriods()[0]);
    140             rv = (int)r.getLifetimeAverageValue();
    141         }
     138        if (rs != null)
     139            rv = (int)rs.getLifetimeAverageValue();
    142140       
    143141        rv <<= 1; // double it to give some leeway.  (bah, too lazy to record stdev)
     
    345343        //}
    346344       
    347         long expiration = getContext().clock().now() + getPerPeerTimeoutMs();
     345        int timeout = _facade.getPeerTimeout(router.getIdentity().getHash());
     346        long expiration = getContext().clock().now() + timeout;
    348347
    349348        DatabaseLookupMessage msg = buildMessage(inTunnelId, inTunnel.getPeer(0), expiration); 
     
    367366        SearchUpdateReplyFoundJob reply = new SearchUpdateReplyFoundJob(getContext(), router, _state, _facade, this);
    368367       
    369         getContext().messageRegistry().registerPending(sel, reply, new FailedJob(getContext(), router), getPerPeerTimeoutMs());
     368        getContext().messageRegistry().registerPending(sel, reply, new FailedJob(getContext(), router), timeout);
    370369        getContext().tunnelDispatcher().dispatchOutbound(msg, outTunnelId, router.getIdentity().getHash());
    371370    }
     
    373372    /** we're searching for a router, so we can just send direct */
    374373    protected void sendRouterSearch(RouterInfo router) {
    375         long expiration = getContext().clock().now() + getPerPeerTimeoutMs();
     374        int timeout = _facade.getPeerTimeout(router.getIdentity().getHash());
     375        long expiration = getContext().clock().now() + timeout;
    376376
    377377        DatabaseLookupMessage msg = buildMessage(expiration);
     
    384384        SearchUpdateReplyFoundJob reply = new SearchUpdateReplyFoundJob(getContext(), router, _state, _facade, this);
    385385        SendMessageDirectJob j = new SendMessageDirectJob(getContext(), msg, router.getIdentity().getHash(),
    386                                                           reply, new FailedJob(getContext(), router), sel, getPerPeerTimeoutMs(), SEARCH_PRIORITY);
     386                                                          reply, new FailedJob(getContext(), router), sel, timeout, SEARCH_PRIORITY);
    387387        getContext().jobQueue().addJob(j);
    388388    }
     
    663663            }
    664664        } else {
    665             Set sendTo = _state.getFailed();
     665            Set sendTo = _state.getRepliedPeers(); // _state.getFailed();
    666666            sendTo.addAll(_state.getPending());
    667667            int numSent = 0;
  • router/java/src/net/i2p/router/networkdb/kademlia/SearchState.java

    r5ec6dca6 r346faa3  
    2323    private HashSet _failedPeers;
    2424    private HashSet _successfulPeers;
     25    private HashSet _repliedPeers;
    2526    private Hash _searchKey;
    2627    private volatile long _completed;
     
    3536        _successfulPeers = new HashSet(16);
    3637        _pendingPeerTimes = new HashMap(16);
     38        _repliedPeers = new HashSet(16);
    3739        _completed = -1;
    3840        _started = _context.clock().now();
     
    121123    /** how long did it take to get the reply, or -1 if we dont know */
    122124    public long replyFound(Hash peer) {
     125        synchronized (_repliedPeers) {
     126            _repliedPeers.add(peer);
     127        }
    123128        synchronized (_pendingPeers) {
    124129            _pendingPeers.remove(peer);
     
    130135        }
    131136    }
     137   
     138    public Set getRepliedPeers() { synchronized (_repliedPeers) { return (Set)_repliedPeers.clone(); } }
    132139   
    133140    public void replyTimeout(Hash peer) {
  • router/java/src/net/i2p/router/networkdb/kademlia/StoreJob.java

    r5ec6dca6 r346faa3  
    2525import net.i2p.router.RouterContext;
    2626import net.i2p.router.TunnelInfo;
     27import net.i2p.router.peermanager.PeerProfile;
    2728import net.i2p.stat.Rate;
    2829import net.i2p.stat.RateStat;
     
    3940    private PeerSelector _peerSelector;
    4041
    41     private final static int PARALLELIZATION = 6; // how many sent at a time
     42    private final static int PARALLELIZATION = 3; // how many sent at a time
    4243    private final static int REDUNDANCY = 6; // we want the data sent to 6 peers
    4344    /**
     
    5354    private final static int STORE_PRIORITY = 100;
    5455   
    55     /** default period we allow for an ACK to take after a store */
    56     private final static int PER_PEER_TIMEOUT = 5*1000;
    57     /** smallest allowed period */
    58     private static final int MIN_PER_PEER_TIMEOUT = 1*1000;
    59 
    6056    /**
    6157     * Create a new search for the routingKey specified
     
    158154                    _state.addSkipped(peer);
    159155                } else {
    160                     if (getContext().shitlist().isShitlisted(((RouterInfo)ds).getIdentity().calculateHash())) {
    161                         _state.addSkipped(peer);
    162                     } else {
     156                    int peerTimeout = _facade.getPeerTimeout(peer);
     157                    //RateStat failing = prof.getDBHistory().getFailedLookupRate();
     158                    //Rate failed = failing.getRate(60*60*1000);
     159                    //if (failed.getCurrentEventCount() + failed.getLastEventCount() > avg) {
     160                    //    _state.addSkipped(peer);
     161                    //}
     162                   
     163                    // we don't want to filter out peers based on our local shitlist, as that opens an avenue for
     164                    // manipulation (since a peer can get us to shitlist them by, well, being shitty, and that
     165                    // in turn would let them assume that a netDb store received didn't come from us)
     166                    //if (getContext().shitlist().isShitlisted(((RouterInfo)ds).getIdentity().calculateHash())) {
     167                    //    _state.addSkipped(peer);
     168                    //} else {
    163169                        _state.addPending(peer);
    164                         sendStore((RouterInfo)ds);
    165                     }
     170                        sendStore((RouterInfo)ds, peerTimeout);
     171                    //}
    166172                }
    167173            }
     
    190196     *
    191197     */
    192     private void sendStore(RouterInfo router) {
     198    private void sendStore(RouterInfo router, int responseTime) {
    193199        DatabaseStoreMessage msg = new DatabaseStoreMessage(getContext());
    194200        msg.setKey(_state.getTarget());
     
    211217        }
    212218
    213         sendStore(msg, router, getContext().clock().now() + getPerPeerTimeoutMs());
     219        sendStore(msg, router, getContext().clock().now() + responseTime);
    214220    }
    215221   
     
    258264            if (_log.shouldLog(Log.DEBUG))
    259265                _log.debug("sending store to " + peer.getIdentity().getHash() + " through " + outTunnel + ": " + msg);
    260             getContext().messageRegistry().registerPending(selector, onReply, onFail, getPerPeerTimeoutMs());
     266            getContext().messageRegistry().registerPending(selector, onReply, onFail, (int)(expiration - getContext().clock().now()));
    261267            getContext().tunnelDispatcher().dispatchOutbound(msg, outTunnel.getSendTunnelId(0), null, peer.getIdentity().getHash());
    262268        } else {
     
    361367        getContext().statManager().addRateData("netDb.storeFailedPeers", _state.getAttempted().size(), _state.getWhenCompleted()-_state.getWhenStarted());
    362368    }
    363    
    364     /**
    365      * Let each peer take up to the average successful search RTT
    366      *
    367      */
    368     private int getPerPeerTimeoutMs() {
    369         int rv = -1;
    370         RateStat rs = getContext().statManager().getRate("netDb.ackTime");
    371         if (rs != null) {
    372             Rate r = rs.getRate(rs.getPeriods()[0]);
    373             rv = (int)r.getLifetimeAverageValue();
    374         }
    375        
    376         rv <<= 1; // double it to give some leeway.  (bah, too lazy to record stdev)
    377         if (rv <= 0)
    378             return PER_PEER_TIMEOUT;
    379         else if (rv < MIN_PER_PEER_TIMEOUT)
    380             return MIN_PER_PEER_TIMEOUT;
    381         else if (rv > PER_PEER_TIMEOUT)
    382             return PER_PEER_TIMEOUT;
    383         else
    384             return rv;
    385     }
    386369}
  • router/java/src/net/i2p/router/transport/udp/PeerTestManager.java

    r5ec6dca6 r346faa3  
    353353            PeerState bob = _transport.getPeerState(from);
    354354            if (bob == null) {
    355                 if (_log.shouldLog(Log.ERROR))
    356                     _log.error("Received from bob (" + from + ") who hasn't established a session with us, refusing to help him test " + aliceIP +":" + alicePort);
     355                if (_log.shouldLog(Log.WARN))
     356                    _log.warn("Received from bob (" + from + ") who hasn't established a session with us, refusing to help him test " + aliceIP +":" + alicePort);
    357357                return;
    358358            } else {
  • router/java/src/net/i2p/router/transport/udp/UDPTransport.java

    r5ec6dca6 r346faa3  
    10811081            default:
    10821082                _context.statManager().addRateData("udp.statusUnknown", 1, 0);
    1083                 if (now - _reachabilityStatusLastUpdated < STATUS_GRACE_PERIOD) {
    1084                     _testEvent.forceRun();
    1085                     SimpleTimer.getInstance().addEvent(_testEvent, 5*1000);
    1086                 } else {
    1087                     _reachabilityStatus = status;
    1088                     _reachabilityStatusLastUpdated = now;
    1089                 }
     1083                //if (now - _reachabilityStatusLastUpdated < STATUS_GRACE_PERIOD) {
     1084                //    _testEvent.forceRun();
     1085                //    SimpleTimer.getInstance().addEvent(_testEvent, 5*1000);
     1086                //} else {
     1087                //    _reachabilityStatus = status;
     1088                //    _reachabilityStatusLastUpdated = now;
     1089                //}
    10901090                break;
    10911091        }
  • router/java/src/net/i2p/router/tunnel/FragmentHandler.java

    r5ec6dca6 r346faa3  
    8383        }
    8484        try {
    85             while (offset < length)
    86                 offset = receiveFragment(preprocessed, offset, length);
     85            while (offset < length) {
     86                int off = receiveFragment(preprocessed, offset, length);
     87                if (off < 0) {
     88                    _context.statManager().addRateData("tunnel.corruptMessage", 1, 1);
     89                    return;
     90                }
     91                offset = off;
     92            }
    8793        } catch (RuntimeException e) {
    8894            if (_log.shouldLog(Log.ERROR))
     
    254260        }
    255261       
    256         msg.receive(messageId, preprocessed, offset, size, !fragmented, router, tunnelId);
     262        boolean ok = msg.receive(messageId, preprocessed, offset, size, !fragmented, router, tunnelId);
     263        if (!ok) return -1;
    257264        if (msg.isComplete()) {
    258265            if (fragmented) {
     
    316323        }
    317324       
    318         msg.receive(messageId, fragmentNum, preprocessed, offset, size, isLast);
     325        boolean ok = msg.receive(messageId, fragmentNum, preprocessed, offset, size, isLast);
     326        if (!ok) return -1;
    319327       
    320328        if (msg.isComplete()) {
  • router/java/src/net/i2p/router/tunnel/FragmentedMessage.java

    r5ec6dca6 r346faa3  
    6666     * @param isLast is this the last fragment in the message?
    6767     */
    68     public void receive(long messageId, int fragmentNum, byte payload[], int offset, int length, boolean isLast) {
    69         if (fragmentNum < 0) throw new RuntimeException("Fragment # == " + fragmentNum + " for messageId " + messageId);
    70         if (payload == null) throw new RuntimeException("Payload is null for messageId " + messageId);
    71         if (length <= 0) throw new RuntimeException("Length is impossible (" + length + ") for messageId " + messageId);
    72         if (offset + length > payload.length) throw new RuntimeException("Length is impossible (" + length + "/" + offset + " out of " + payload.length + ") for messageId " + messageId);
     68    public boolean receive(long messageId, int fragmentNum, byte payload[], int offset, int length, boolean isLast) {
     69        if (fragmentNum < 0) {
     70            if (_log.shouldLog(Log.ERROR))
     71                _log.error("Fragment # == " + fragmentNum + " for messageId " + messageId);
     72            return false;
     73        }
     74        if (payload == null) {
     75            if (_log.shouldLog(Log.ERROR))
     76                _log.error("Payload is null for messageId " + messageId);
     77            return false;
     78        }
     79        if (length <= 0) {
     80            if (_log.shouldLog(Log.ERROR))
     81                _log.error("Length is impossible (" + length + ") for messageId " + messageId);
     82            return false;
     83        }
     84        if (offset + length > payload.length) {
     85            if (_log.shouldLog(Log.ERROR))
     86                _log.error("Length is impossible (" + length + "/" + offset + " out of " + payload.length + ") for messageId " + messageId);
     87            return false;
     88        }
    7389        if (_log.shouldLog(Log.DEBUG))
    7490            _log.debug("Receive message " + messageId + " fragment " + fragmentNum + " with " + length + " bytes (last? " + isLast + ") offset = " + offset);
     
    88104        if (fragmentNum > _highFragmentNum)
    89105            _highFragmentNum = fragmentNum;
    90         if (isLast && fragmentNum <= 0)
    91             throw new RuntimeException("hmm, isLast and fragmentNum=" + fragmentNum + " for message " + messageId);
     106        if (isLast && fragmentNum <= 0) {
     107            if (_log.shouldLog(Log.ERROR))
     108                _log.error("hmm, isLast and fragmentNum=" + fragmentNum + " for message " + messageId);
     109            return false;
     110        }
     111        return true;
    92112    }
    93113   
     
    104124     * @param toTunnel what tunnel is this destined for (may be null)
    105125     */
    106     public void receive(long messageId, byte payload[], int offset, int length, boolean isLast, Hash toRouter, TunnelId toTunnel) {
    107         if (payload == null) throw new RuntimeException("Payload is null for messageId " + messageId);
    108         if (length <= 0) throw new RuntimeException("Length is impossible (" + length + ") for messageId " + messageId);
    109         if (offset + length > payload.length) throw new RuntimeException("Length is impossible (" + length + "/" + offset + " out of " + payload.length + ") for messageId " + messageId);
     126    public boolean receive(long messageId, byte payload[], int offset, int length, boolean isLast, Hash toRouter, TunnelId toTunnel) {
     127        if (payload == null) {
     128            if (_log.shouldLog(Log.ERROR))
     129                _log.error("Payload is null for messageId " + messageId);
     130            return false;
     131        }
     132        if (length <= 0) {
     133            if (_log.shouldLog(Log.ERROR))
     134                _log.error("Length is impossible (" + length + ") for messageId " + messageId);
     135            return false;
     136        }
     137        if (offset + length > payload.length) {
     138            if (_log.shouldLog(Log.ERROR))
     139                _log.error("Length is impossible (" + length + "/" + offset + " out of " + payload.length + ") for messageId " + messageId);
     140            return false;
     141        }
    110142        if (_log.shouldLog(Log.DEBUG))
    111143            _log.debug("Receive message " + messageId + " with " + length + " bytes (last? " + isLast + ") targetting " + toRouter + " / " + toTunnel + " offset=" + offset);
     
    125157        if (_highFragmentNum < 0)
    126158            _highFragmentNum = 0;
     159        return true;
    127160    }
    128161   
Note: See TracChangeset for help on using the changeset viewer.