<commit_before>package water;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

import water.AutoBuffer.AutoBufferException;
import water.util.Log;

/**
 * The Thread that looks for TCP Cloud requests.
 *
 * This thread just spins on reading TCP requests from other Nodes.
 * @author <a href="mailto:<EMAIL>"></a>
 * @version 1.0
 */

public class TCPReceiverThread extends Thread {
  private ServerSocketChannel SOCK;
  private final boolean _udpLike; // true if this channel is for small async udp-like messages
  public TCPReceiverThread(ServerSocketChannel sock, boolean udpLike) { super("TCP-Accept"); SOCK = sock; _udpLike = udpLike; }

  // The Run Method.
  // Started by main() on a single thread, this code manages reading TCP requests
  @SuppressWarnings("resource")
  public void run() {
    Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
    ServerSocketChannel errsock = null;
    boolean saw_error = false;

    while( true ) {
      try {
        // Cleanup from any prior socket failures.  Rare unless we're really sick.
        if( errsock != null ) { // One time attempt a socket close
          final ServerSocketChannel tmp2 = errsock; errsock = null;
          tmp2.close();       // Could throw, but errsock cleared for next pass
        }
        if( saw_error ) Thread.sleep(100); // prevent deny-of-service endless socket-creates
        saw_error = false;

        // ---
        // More common-case setup of a ServerSocket
        if( SOCK == null ) {
          assert !_udpLike;
          SOCK = ServerSocketChannel.open();
          SOCK.socket().setReceiveBufferSize(AutoBuffer.BBP_BIG.size());
          SOCK.socket().bind(H2O.SELF._key);
        }
        // Block for TCP connection and setup to read from it.
        SocketChannel sock = SOCK.accept();
        // Pass off the TCP connection to a separate reader thread
        if(_udpLike)
          new TCP_UDP_ReaderThread(sock.socket().getInetAddress(),sock).start();
        else new TCPReaderThread(sock,new AutoBuffer(sock)).start();
      } catch( java.nio.channels.AsynchronousCloseException ex ) {
        break;                  // Socket closed for shutdown
      } catch( Exception e ) {
        // On any error from anybody, close all sockets & re-open
        Log.err("IO error on TCP port "+H2O.H2O_PORT+": ",e);
        saw_error = true;
        errsock = SOCK ;  SOCK = null; // Signal error recovery on the next loop
      }
    }
  }

  static class TCP_UDP_ReaderThread extends Thread {
    private final SocketChannel _chan;
    private final ByteBuffer _bb;
    private final InetAddress _ia;
    private H2ONode _h2o;

    public TCP_UDP_ReaderThread(InetAddress ia,SocketChannel chan) {
      super("TCP-UDP");
      _ia = ia;
      _chan = chan;
      _bb = ByteBuffer.allocateDirect(32*AutoBuffer.BBP_SML.size()).order(ByteOrder.nativeOrder());//AutoBuffer.BBP_BIG.make();
    }

    public String printBytes(ByteBuffer bb, int start, int sz) {
      StringBuilder sb = new StringBuilder();
      int idx = start + sz;
      try {
        for (int i = 5; i > 0; --i)
          sb.append("-" + i + ":" + (0xFF & bb.get(idx - i)) + " ");
        sb.append("0: " + (0xFF & bb.get(idx)) + " ");
        for (int i = 1; i <= 5; ++i)
          sb.append("+" + i + ":" + (0xFF & bb.get(idx + i)) + " ");
      } catch(Throwable t) {}
      return sb.toString();
    }
    private int read(int n)  throws IOException {
      if(_bb.remaining() < n)
        throw new IllegalStateException("Reading more bytes than available, reading " + n + " bytes, remaining = " + _bb.remaining());
      int sizeRead = 0;
      while(sizeRead < n) {
        int res = _chan.read(_bb);
        if( res == -1 )
          throw new EOFException("Reading "+n+" bytes, AB="+this);
        if( res ==  0 ) throw new RuntimeException("Reading zero bytes - so no progress?");
        sizeRead += res;
      }
      return sizeRead;
    }
    public void run(){
      int start = 0;
      boolean idle = true;
      try {
        while (true) {
          idle = true;
          if (_h2o != null)
            _h2o._last_heard_from = System.currentTimeMillis();
          if (start > _bb.position() - 2)
            read(start + 2 - _bb.position());
          idle = false;
          int sz = _bb.getShort(start); // message size in bytes
          assert sz < AutoBuffer.BBP_SML.size() : "Incoming message is too big, should've been sent by TCP-BIG, got " + sz + " bytes, start = " + start;
          int rem = _bb.position() - start;
          read(sz + 1 - rem);
          assert (0xFF & _bb.get(start + sz)) == 0xef:"Missing expected sentinel (0xef==239) at the end of the message, likely out of sync, start = " + start + ", size = " + sz + ", bytes = " + printBytes(_bb,start,sz);
          // extract the bytes
          byte[] ary = new byte[Math.max(sz, 18)];
          for (int i = 0; i < sz; ++i)
            ary[i] = _bb.get(start + i);
          if (_h2o == null)
            _h2o = H2ONode.intern(_ia, new AutoBuffer(null, ary).getPort());
          AutoBuffer ab = new AutoBuffer(_h2o, ary);
          int ctrl = ab.getCtrl();
          TimeLine.record_recv(ab, false, 0);
          H2O.submitTask(new FJPacket(ab, ctrl));
          start += sz + 1;
          if (_bb.remaining() < AutoBuffer.BBP_SML.size() + 1) { // + 2 bytes for size + 1 byte for 0xef sentinel
            _bb.limit(_bb.position());
            _bb.position(start);
            _bb.compact();
            start = 0;
          }
        }
      } catch (IOException ioe) {
        if(!idle) {
          Log.err("Got IO Error when reading small messages over TCP");
          Log.err(ioe);
        }
      } finally {
        AutoBuffer.BBP_BIG.free(_bb);
        if(_chan != null && _chan.isOpen())
          try { _chan.close();} catch (IOException e) {}
      }
    }<commit_msg>Updated TCP-UDP message handling to fail completely when getting out of sync (should never ever happen) instead of just throwing an exception and recovering via reliability protocol.<commit_after>package water;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

import water.AutoBuffer.AutoBufferException;
import water.util.Log;

/**
 * The Thread that looks for TCP Cloud requests.
 *
 * This thread just spins on reading TCP requests from other Nodes.
 * @author <a href="mailto:<EMAIL>"></a>
 * @version 1.0
 */

public class TCPReceiverThread extends Thread {
  private ServerSocketChannel SOCK;
  private final boolean _udpLike; // true if this channel is for small async udp-like messages
  public TCPReceiverThread(ServerSocketChannel sock, boolean udpLike) { super("TCP-Accept"); SOCK = sock; _udpLike = udpLike; }

  // The Run Method.
  // Started by main() on a single thread, this code manages reading TCP requests
  @SuppressWarnings("resource")
  public void run() {
    Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
    ServerSocketChannel errsock = null;
    boolean saw_error = false;

    while( true ) {
      try {
        // Cleanup from any prior socket failures.  Rare unless we're really sick.
        if( errsock != null ) { // One time attempt a socket close
          final ServerSocketChannel tmp2 = errsock; errsock = null;
          tmp2.close();       // Could throw, but errsock cleared for next pass
        }
        if( saw_error ) Thread.sleep(100); // prevent deny-of-service endless socket-creates
        saw_error = false;

        // ---
        // More common-case setup of a ServerSocket
        if( SOCK == null ) {
          assert !_udpLike;
          SOCK = ServerSocketChannel.open();
          SOCK.socket().setReceiveBufferSize(AutoBuffer.BBP_BIG.size());
          SOCK.socket().bind(H2O.SELF._key);
        }
        // Block for TCP connection and setup to read from it.
        SocketChannel sock = SOCK.accept();
        // Pass off the TCP connection to a separate reader thread
        if(_udpLike)
          new TCP_UDP_ReaderThread(sock.socket().getInetAddress(),sock).start();
        else new TCPReaderThread(sock,new AutoBuffer(sock)).start();
      } catch( java.nio.channels.AsynchronousCloseException ex ) {
        break;                  // Socket closed for shutdown
      } catch( Exception e ) {
        // On any error from anybody, close all sockets & re-open
        Log.err("IO error on TCP port "+H2O.H2O_PORT+": ",e);
        saw_error = true;
        errsock = SOCK ;  SOCK = null; // Signal error recovery on the next loop
      }
    }
  }

  static class TCP_UDP_ReaderThread extends Thread {
    private final SocketChannel _chan;
    private final ByteBuffer _bb;
    private final InetAddress _ia;
    private H2ONode _h2o;

    public TCP_UDP_ReaderThread(InetAddress ia,SocketChannel chan) {
      super("TCP-UDP");
      _ia = ia;
      _chan = chan;
      _bb = ByteBuffer.allocateDirect(32*AutoBuffer.BBP_SML.size()).order(ByteOrder.nativeOrder());//AutoBuffer.BBP_BIG.make();
    }

    public String printBytes(ByteBuffer bb, int start, int sz) {
      StringBuilder sb = new StringBuilder();
      int idx = start + sz;
      try {
        for (int i = 5; i > 0; --i)
          sb.append("-" + i + ":" + (0xFF & bb.get(idx - i)) + " ");
        sb.append("0: " + (0xFF & bb.get(idx)) + " ");
        for (int i = 1; i <= 5; ++i)
          sb.append("+" + i + ":" + (0xFF & bb.get(idx + i)) + " ");
      } catch(Throwable t) {}
      return sb.toString();
    }
    private int read(int n)  throws IOException {
      if(_bb.remaining() < n)
        throw new IllegalStateException("Reading more bytes than available, reading " + n + " bytes, remaining = " + _bb.remaining());
      int sizeRead = 0;
      while(sizeRead < n) {
        int res = _chan.read(_bb);
        if( res == -1 )
          throw new EOFException("Reading "+n+" bytes, AB="+this);
        if( res ==  0 ) throw new RuntimeException("Reading zero bytes - so no progress?");
        sizeRead += res;
      }
      return sizeRead;
    }
    public void run(){
      int start = 0;
      boolean idle = true;
      try {
        while (true) {
          idle = true;
          if (_h2o != null)
            _h2o._last_heard_from = System.currentTimeMillis();
          if (start > _bb.position() - 2)
            read(start + 2 - _bb.position());
          idle = false;
          int sz = _bb.getShort(start); // message size in bytes
          assert sz < AutoBuffer.BBP_SML.size() : "Incoming message is too big, should've been sent by TCP-BIG, got " + sz + " bytes, start = " + start;
          int rem = _bb.position() - start;
          read(sz + 1 - rem);
          if((0xFF & _bb.get(start + sz)) != 0xef)
            H2O.fail("Missing expected sentinel (0xef==239) at the end of the message, likely out of sync, start = " + start + ", size = " + sz + ", bytes = " + printBytes(_bb,start,sz));
          // extract the bytes
          byte[] ary = new byte[Math.max(sz, 18)];
          for (int i = 0; i < sz; ++i)
            ary[i] = _bb.get(start + i);
          if (_h2o == null)
            _h2o = H2ONode.intern(_ia, new AutoBuffer(null, ary).getPort());
          AutoBuffer ab = new AutoBuffer(_h2o, ary);
          int ctrl = ab.getCtrl();
          TimeLine.record_recv(ab, false, 0);
          H2O.submitTask(new FJPacket(ab, ctrl));
          start += sz + 1;
          if (_bb.remaining() < AutoBuffer.BBP_SML.size() + 1) { // + 2 bytes for size + 1 byte for 0xef sentinel
            _bb.limit(_bb.position());
            _bb.position(start);
            _bb.compact();
            start = 0;
          }
        }
      } catch (IOException ioe) {
        if(!idle) {
          Log.err("Got IO Error when reading small messages over TCP");
          Log.err(ioe);
        }
      } finally {
        AutoBuffer.BBP_BIG.free(_bb);
        if(_chan != null && _chan.isOpen())
          try { _chan.close();} catch (IOException e) {}
      }
    }
{'fraction_non_alphanumeric': 0.12324518445968005, 'fraction_numerical': 0.008570029382957884, 'mean_word_length': 2.265724946695096, 'pattern_counts': {'":': 6, '<': 21, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2762184', 'n_tokens_mistral': 3786, 'n_tokens_neox': 3676, 'n_words': 1096}