Changeset 100d307


Ignore:
Timestamp:
Dec 2, 2017 11:12:19 PM (3 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
89d7a59
Parents:
3d0e15a
Message:

i2ptunnel: Propagate resets from streaming to Socket and vice versa (ticket #2071)

Files:
6 edited

Legend:

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

    r3d0e15a r100d307  
    2525
    2626import net.i2p.client.streaming.I2PSocket;
     27import net.i2p.client.streaming.I2PSocketException;
    2728import net.i2p.I2PAppContext;
    2829import net.i2p.data.Base32;
     
    598599            InputStream browserin = null;
    599600            InputStream serverin = null;
     601            Sender s = null;
     602            IOException ioex = null;
    600603            try {
    601604                serverout = _webserver.getOutputStream();
     
    643646                compressedOut.write(DataHelper.getUTF8(modifiedHeaders));
    644647
    645                 Sender s = new Sender(compressedOut, serverin, "server: server to browser", _log);
     648                s = new Sender(compressedOut, serverin, "server: server to browser", _log);
    646649                if (_log.shouldLog(Log.INFO))
    647650                    _log.info("Before pumping the compressed response");
     
    659662                if (_log.shouldLog(Log.WARN))
    660663                    _log.warn("error compressing", ioe);
     664                 ioex = ioe;
    661665            } finally {
     666                if (ioex == null && s != null)
     667                    ioex = s.getFailure();
     668                if (ioex != null) {
     669                    // Reset propagation, simplified from I2PTunnelRunner
     670                    boolean i2pReset = false;
     671                    if (ioex instanceof I2PSocketException) {
     672                        I2PSocketException ise = (I2PSocketException) ioex;
     673                        int status = ise.getStatus();
     674                        i2pReset = status == I2PSocketException.STATUS_CONNECTION_RESET;
     675                        if (i2pReset) {
     676                            if (_log.shouldWarn())
     677                                _log.warn("Server got I2P reset, resetting socket", ioex);
     678                            try {
     679                                _webserver.setSoLinger(true, 0);
     680                            } catch (IOException ioe) {}
     681                            try {
     682                                _webserver.close();
     683                            } catch (IOException ioe) {}
     684                            try {
     685                                _browser.close();
     686                            } catch (IOException ioe) {}
     687                        }
     688                    }
     689                    if (!i2pReset && ioex instanceof SocketException) {
     690                        String msg = ioex.getMessage();
     691                        boolean sockReset = msg != null && msg.contains("reset");
     692                        if (sockReset) {
     693                            if (_log.shouldWarn())
     694                                _log.warn("Server got socket reset, resetting I2P socket");
     695                            try {
     696                                _browser.reset();
     697                            } catch (IOException ioe) {}
     698                            try {
     699                                _webserver.close();
     700                            } catch (IOException ioe) {}
     701                        }
     702                    }
     703                }
    662704                if (browserout != null) try { browserout.close(); } catch (IOException ioe) {}
    663705                if (serverout != null) try { serverout.close(); } catch (IOException ioe) {}
     
    674716        // shadows _log in super()
    675717        private final Log _log;
     718        private IOException _failure;
    676719
    677720        public Sender(OutputStream out, InputStream in, String name, Log log) {
     
    692735            } catch (IOException ioe) {
    693736                if (_log.shouldLog(Log.DEBUG))
    694                     _log.debug("Error sending", ioe);
     737                    _log.debug(_name + " Error sending", ioe);
     738                synchronized(this) {
     739                    _failure = ioe;
     740                }
    695741            } finally {
    696742                if (_out != null) try { _out.close(); } catch (IOException ioe) {}
    697743                if (_in != null) try { _in.close(); } catch (IOException ioe) {}
    698744            }
     745        }
     746
     747        /**
     748         *  @since 0.9.33
     749         */
     750        public synchronized IOException getFailure() {
     751            return _failure;
    699752        }
    700753    }
  • apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelRunner.java

    r3d0e15a r100d307  
    1818import net.i2p.I2PAppContext;
    1919import net.i2p.client.streaming.I2PSocket;
     20import net.i2p.client.streaming.I2PSocketException;
    2021import net.i2p.data.ByteArray;
    2122import net.i2p.data.DataHelper;
     
    288289                _log.debug("At least one forwarder completed, closing and joining");
    289290           
     291            boolean i2pReset = false;
     292            boolean sockReset = false;
    290293            // this task is useful for the httpclient
    291             if (onTimeout != null || _onFail != null) {
     294            if ((onTimeout != null || _onFail != null) && totalReceived <= 0) {
    292295                if (_log.shouldLog(Log.DEBUG))
    293296                    _log.debug("runner has a timeout job, totalReceived = " + totalReceived
     
    297300                // responsibility to know that and decide whether or not to write to the socket.
    298301                // HTTPClient never sets initialSocketData.
    299                 if (totalReceived <= 0) {
    300                     if (_onFail != null) {
    301                         Exception e = fromI2P.getFailure();
    302                         if (e == null)
    303                             e = toI2P.getFailure();
    304                         _onFail.onFail(e);
    305                     } else {
    306                         onTimeout.run();
    307                     }
    308                 }
    309             }
    310            
    311             // now one connection is dead - kill the other as well, after making sure we flush
    312             close(out, in, i2pout, i2pin, s, i2ps, toI2P, fromI2P);
     302                if (_onFail != null) {
     303                    Exception e = fromI2P.getFailure();
     304                    if (e == null)
     305                        e = toI2P.getFailure();
     306                    _onFail.onFail(e);
     307                } else {
     308                    onTimeout.run();
     309                }
     310            } else {
     311                // Detect a reset on one side, and propagate to the other
     312                Exception e1 = fromI2P.getFailure();
     313                Exception e2 = toI2P.getFailure();
     314                Throwable c1 = e1 != null ? e1.getCause() : null;
     315                Throwable c2 = e2 != null ? e2.getCause() : null;
     316                if (c1 != null && c1 instanceof I2PSocketException) {
     317                    I2PSocketException ise = (I2PSocketException) c1;
     318                    int status = ise.getStatus();
     319                    i2pReset = status == I2PSocketException.STATUS_CONNECTION_RESET;
     320                }
     321                if (!i2pReset && c2 != null && c2 instanceof I2PSocketException) {
     322                    I2PSocketException ise = (I2PSocketException) c2;
     323                    int status = ise.getStatus();
     324                    i2pReset = status == I2PSocketException.STATUS_CONNECTION_RESET;
     325                }
     326                if (!i2pReset && e1 != null && e1 instanceof SocketException) {
     327                        String msg = e1.getMessage();
     328                        sockReset = msg != null && msg.contains("reset");
     329                }
     330                if (!sockReset && e2 != null && e2 instanceof SocketException) {
     331                        String msg = e2.getMessage();
     332                        sockReset = msg != null && msg.contains("reset");
     333                }
     334            }
     335
     336            if (i2pReset) {
     337                if (_log.shouldWarn())
     338                    _log.warn("Got I2P reset, resetting socket");
     339                try {
     340                    s.setSoLinger(true, 0);
     341                } catch (IOException ioe) {}
     342                try {
     343                    s.close();
     344                } catch (IOException ioe) {}
     345                try {
     346                    i2ps.close();
     347                } catch (IOException ioe) {}
     348            } else if (sockReset) {
     349                if (_log.shouldWarn())
     350                    _log.warn("Got socket reset, resetting I2P socket");
     351                try {
     352                    i2ps.reset();
     353                } catch (IOException ioe) {}
     354                try {
     355                    s.close();
     356                } catch (IOException ioe) {}
     357            } else {
     358                // now one connection is dead - kill the other as well, after making sure we flush
     359                close(out, in, i2pout, i2pin, s, i2ps, toI2P, fromI2P);
     360            }
    313361        } catch (InterruptedException ex) {
    314362            if (_log.shouldLog(Log.ERROR))
     
    362410        try {
    363411            out.flush();
    364         } catch (IOException ioe) {
    365             // ignore
    366         }
     412        } catch (IOException ioe) {}
    367413        try {
    368414            i2pout.flush();
    369         } catch (IOException ioe) {
    370             // ignore
    371         }
     415        } catch (IOException ioe) {}
    372416        try {
    373417            in.close();
    374         } catch (IOException ioe) {
    375             // ignore
    376         }
     418        } catch (IOException ioe) {}
    377419        try {
    378420            i2pin.close();
    379         } catch (IOException ioe) {
    380             // ignore
    381         }
     421        } catch (IOException ioe) {}
    382422        // ok, yeah, there's a race here in theory, if data comes in after flushing and before
    383423        // closing, but its better than before...
    384424        try {
    385425            s.close();
    386         } catch (IOException ioe) {
    387             // ignore
    388         }
     426        } catch (IOException ioe) {}
    389427        try {
    390428            i2ps.close();
    391         } catch (IOException ioe) {
    392             // ignore
    393         }
     429        } catch (IOException ioe) {}
    394430        t1.join(30*1000);
    395431        // t2 = fromI2P now run inline
     
    470506                        //if (_log.shouldLog(Log.DEBUG))
    471507                        //    _log.debug("Flushing after sending " + len + " bytes through");
    472                         if (_log.shouldLog(Log.DEBUG))
    473                             _log.debug(direction + ": " + len + " bytes flushed through " + (_toI2P ? "to " : "from ")
    474                                        + to);
     508                        //if (_log.shouldLog(Log.DEBUG))
     509                        //    _log.debug(direction + ": " + len + " bytes flushed through " + (_toI2P ? "to " : "from ")
     510                        //               + to);
    475511                        if (_toI2P) {
    476512                            try {
     
    487523                    }
    488524                }
     525                //if (_log.shouldDebug())
     526                //    _log.debug(direction + ": Normal EOF on read");
    489527                //out.flush(); // close() flushes
    490528            } catch (SocketException ex) {
    491529                // this *will* occur when the other threads closes the socket
    492                 synchronized (finishLock) {
    493                     if (!finished) {
    494                         if (_log.shouldLog(Log.DEBUG))
    495                             _log.debug(direction + ": Socket closed - error reading and writing",
    496                                        ex);
    497                     }
    498                 }
    499                 _failure = ex;
    500             } catch (InterruptedIOException ex) {
    501                 if (_log.shouldLog(Log.WARN))
    502                     _log.warn(direction + ": Closing connection due to timeout (error: \""
    503                               + ex.getMessage() + "\")");
     530                if (_log.shouldDebug()) {
     531                    boolean fnshd;
     532                    synchronized (finishLock) {
     533                        fnshd = finished;
     534                    }
     535                    if (!fnshd) {
     536                        _log.debug(direction + ": IOE - error forwarding", ex);
     537                    } else {
     538                        _log.debug(direction + ": IOE caused by other direction", ex);
     539                    }
     540                }
    504541                _failure = ex;
    505542            } catch (IOException ex) {
    506                 if (!finished) {
    507                     if (_log.shouldLog(Log.WARN))
    508                         _log.warn(direction + ": Error forwarding", ex);
    509                 }
    510                 //else
    511                 //    _log.warn("You may ignore this", ex);
     543                if (_log.shouldWarn()) {
     544                    boolean fnshd;
     545                    synchronized (finishLock) {
     546                        fnshd = finished;
     547                    }
     548                    if (!fnshd) {
     549                        _log.warn(direction + ": IOE - error forwarding", ex);
     550                    } else if (_log.shouldDebug()) {
     551                        _log.debug(direction + ": IOE caused by other direction", ex);
     552                    }
     553                }
    512554                _failure = ex;
    513555            } finally {
  • apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java

    r3d0e15a r100d307  
    201201                // packets through
    202202                // Incorrect assumption, the constructor defaults _connected to true --Sponge
    203                 if (!_connected.get())
     203                if (!_connected.get()) {
     204                    if (getResetReceived())
     205                        throw new I2PSocketException(I2PSocketException.STATUS_CONNECTION_RESET);
    204206                    throw new IOException("disconnected");
     207                }
    205208                if (_outputStream.getClosed())
    206209                    throw new IOException("output stream closed");
  • apps/streaming/java/src/net/i2p/client/streaming/impl/PacketLocal.java

    r3d0e15a r100d307  
    1212import net.i2p.data.SessionTag;
    1313import net.i2p.data.SigningPrivateKey;
     14import net.i2p.client.streaming.I2PSocketException;
    1415import net.i2p.util.Log;
    1516import net.i2p.util.SimpleTimer2;
     
    322323                synchronized (this) {
    323324                    if (_ackOn > 0) break;
    324                     if (!_connection.getIsConnected())
     325                    if (!_connection.getIsConnected()) {
     326                        if (_connection.getResetReceived())
     327                            throw new I2PSocketException(I2PSocketException.STATUS_CONNECTION_RESET);
    325328                        throw new IOException("disconnected");
     329                    }
    326330                    if (_cancelledOn > 0)
    327331                        throw new IOException("cancelled");
  • history.txt

    r3d0e15a r100d307  
     12017-12-02 zzz
     2 * i2ptunnel: Propagate resets from streaming to Socket
     3   and vice versa (ticket #2071)
     4 * Streaming: Send reset when receiving more data after locally closed,
     5   rather than acking (ticket #2071)
     6 * Tests: Fix up deprecation warnings
     7
    182017-12-01 zzz
    29 * Build: Split net.i2p.router.web into two packages
    310 * Console: Move /configkeyring HTML to console, fix deletion,
    411   don't truncate hashes, better form errors, tag for translation (ticket #2108)
     12 * LeaseSet: Better error messages when decode fails (ticket #2108)
    513 * Streaming: Double the RTO on congestion (ticket #1939)
    614
  • router/java/src/net/i2p/router/RouterVersion.java

    r3d0e15a r100d307  
    1919    public final static String ID = "Monotone";
    2020    public final static String VERSION = CoreVersion.VERSION;
    21     public final static long BUILD = 10;
     21    public final static long BUILD = 11;
    2222
    2323    /** for example "-test" */
Note: See TracChangeset for help on using the changeset viewer.