private void handleRead ( ByteBuffer in , Connection c ) throws IOException { // port is locked by VoltNetwork when in valid use. // assert(m_port.m_lock.tryLock() == true); long recvDests [ ] = null ; final long sourceHSId = in . getLong ( ) ; final int destCount = in . getInt ( ) ; if ( destCount == POISON_PILL ) { //This is a poison pill //Ignore poison pill during shutdown, in tests we receive crash messages from //leader appointer during shutdown if ( VoltDB . instance ( ) . getMode ( ) == OperationMode . SHUTTINGDOWN ) { return ; } byte messageBytes [ ] = new byte [ in . getInt ( ) ] ; in . get ( messageBytes ) ; String message = new String ( messageBytes , "UTF-8" ) ; message = String . format ( "Fatal error from id,hostname(%d,%s): %s" , m_hostId , hostnameAndIPAndPort ( ) , message ) ; //if poison pill with particular cause handle it. int cause = in . getInt ( ) ; if ( cause == ForeignHost . CRASH_ME ) { int hid = VoltDB . instance ( ) . getHostMessenger ( ) . getHostId ( ) ; hostLog . debug ( "Poison Pill with target me was sent.: " + hid ) ; //Killing myself. VoltDB . instance ( ) . halt ( ) ; } else if ( cause == ForeignHost . CRASH_ALL || cause == ForeignHost . CRASH_SPECIFIED ) { org . voltdb . VoltDB . crashLocalVoltDB ( message , false , null ) ; } else if ( cause == ForeignHost . PRINT_STACKTRACE ) { //collect thread dumps String dumpDir = new File ( VoltDB . instance ( ) . getVoltDBRootPath ( ) , "thread_dumps" ) . getAbsolutePath ( ) ; String fileName = m_hostMessenger . getHostname ( ) + "_host-" + m_hostId + "_" + System . currentTimeMillis ( ) + ".jstack" ; VoltDB . dumpThreadTraceToFile ( dumpDir , fileName ) ; } else { //Should never come here. hostLog . error ( "Invalid Cause in poison pill: " + cause ) ; } return ; } else if ( destCount == STOPNODE_NOTICE ) { int targetHostId = in . getInt ( ) ; hostLog . info ( "Receive StopNode notice for host " + targetHostId ) ; m_hostMessenger . addStopNodeNotice ( targetHostId ) ; return ; } recvDests = new long [ destCount ] ; for ( int i = 0 ; i < destCount ; i ++ ) { recvDests [ i ] = in . getLong ( ) ; } final VoltMessage message = m_hostMessenger . getMessageFactory ( ) . createMessageFromBuffer ( in , sourceHSId ) ; // ENG-1608. We sniff for SiteFailureMessage here so // that a node will participate in the failure resolution protocol // even if it hasn't directly witnessed a node fault. if ( message instanceof SiteFailureMessage && ! ( message instanceof SiteFailureForwardMessage ) ) { SiteFailureMessage sfm = ( SiteFailureMessage ) message ; for ( FaultMessage fm : sfm . asFaultMessages ( ) ) { m_hostMessenger . relayForeignHostFailed ( fm ) ; } } for ( int i = 0 ; i < destCount ; i ++ ) { deliverMessage ( recvDests [ i ] , message ) ; } //m_lastMessageMillis = System.currentTimeMillis(); m_lastMessageMillis . lazySet ( EstTime . currentTimeMillis ( ) ) ; }
Read data from the network . Runs in the context of PicoNetwork thread when data is available .