server/src/main/java/com/orientechnologies/orient/server/distributed/ODistributedResponseManagerImpl.java
public boolean waitForSynchronousResponses() throws InterruptedException {
    final long beginTime = System.currentTimeMillis();
    try {

      boolean reachedTimeout = false;
      long currentTimeout = synchTimeout;
      while (currentTimeout > 0) {

        if (currentTimeout > 10000)
          // CUT THE TIMEOUT IN BLOCKS OF 10S EACH TO ALLOW CHECKING FOR ANY SERVER IF UNREACHABLE
          currentTimeout = 10000;

        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
              "Waiting max %dms for collecting all synchronous responses... (timeout=%d reqId=%s thread=%d)", currentTimeout,
              synchTimeout, request.getId(), Thread.currentThread().getId());

        // WAIT FOR THE RESPONSES
        if (synchronousResponsesArrived.await(currentTimeout, TimeUnit.MILLISECONDS)) {
          if (canceled.get())
            throw new ODistributedOperationException("Request has been canceled");

          // COMPLETED
          if (ODistributedServerLog.isDebugEnabled())
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "All synchronous responses collected in %dms (reqId=%s thread=%d)", (System.currentTimeMillis() - beginTime),
                request.getId(), Thread.currentThread().getId());

          return true;
        }

        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
              "All synchronous responses not collected in %dms, waiting again... (reqId=%s thread=%d)",
              (System.currentTimeMillis() - beginTime), request.getId(), Thread.currentThread().getId());

        if (Thread.currentThread().isInterrupted()) {
          // INTERRUPTED
          ODistributedServerLog
              .warn(this, dManager.getLocalNodeName(), null, DIRECTION.NONE, "Thread has been interrupted wait for request (%s)",
                  request);
          Thread.currentThread().interrupt();
          break;
        }

        synchronousResponsesLock.lock();
        try {

          final long now = System.currentTimeMillis();
          final long elapsed = now - beginTime;
          if (elapsed > synchTimeout)
            reachedTimeout = true;
          currentTimeout = synchTimeout - elapsed;

          // CHECK IF ANY NODE ARE UNREACHABLE IN THE MEANWHILE
          int synchronizingNodes = 0;
          int missingActiveNodes = 0;

          Map<String, ODistributedServerManager.DB_STATUS> missingResponseNodeStatuses = new HashMap<String, ODistributedServerManager.DB_STATUS>(
              responses.size());

          int missingResponses = 0;

          for (Iterator<Map.Entry<String, Object>> iter = responses.entrySet().iterator(); iter.hasNext(); ) {
            final Map.Entry<String, Object> curr = iter.next();

            if (curr.getValue() == NO_RESPONSE) {
              missingResponses++;

              // ANALYZE THE NODE WITHOUT A RESPONSE
              final ODistributedServerManager.DB_STATUS dbStatus = dManager.getDatabaseStatus(curr.getKey(), getDatabaseName());

              missingResponseNodeStatuses.put(curr.getKey(), dbStatus);

              switch (dbStatus) {
              case BACKUP:
              case SYNCHRONIZING:
                synchronizingNodes++;
                missingActiveNodes++;
                break;
              case ONLINE:
                missingActiveNodes++;
                break;
              }
            }
          }

          if (missingResponses == 0) {
            // ALL RESPONSE COLLECTED, BUT NO QUORUM REACHED
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "All responses collected %s, but no quorum reached (reqId=%s)", responses, request.getId());
            break;
          }

          request.getTask().checkIsValid(dManager);

          if (missingActiveNodes == 0) {
            // NO MORE ACTIVE NODES TO WAIT
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "No more active nodes to wait for request (%s): anticipate timeout (saved %d ms). Missing servers: %s", request,
                currentTimeout, missingResponseNodeStatuses);
            break;
          }

          final long lastClusterChange = dManager.getLastClusterChangeOn();
          if (lastClusterChange > 0 && now - lastClusterChange < (synchTimeout + ADDITIONAL_TIMEOUT_CLUSTER_SHAPE)) {
            // CHANGED CLUSTER SHAPE DURING WAIT: ENLARGE TIMEOUT
            currentTimeout = synchTimeout;
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "Cluster shape changed during request (%s): enlarge timeout +%dms, wait again for %dms", request, synchTimeout,
                currentTimeout);
            continue;
          } else if (synchronizingNodes > 0) {
            // SOME NODE IS SYNCHRONIZING: WAIT FOR THEM
            // currentTimeout = synchTimeout;
            // ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
            // "%d nodes are in synchronization mode during request (%s): enlarge timeout +%dms, wait again for %dms",
            // synchronizingNodes, request, synchTimeout, currentTimeout);
          }

        } finally {
          synchronousResponsesLock.unlock();
        }
      }

      if (canceled.get())
        throw new ODistributedOperationException("Request has been canceled");

      return isMinimumQuorumReached(reachedTimeout);

    } finally {
      Orient.instance().getProfiler()
          .stopChrono("distributed.synchResponses", "Time to collect all the synchronous responses from distributed nodes",
              beginTime);
    }
  }
public boolean waitForSynchronousResponses() throws InterruptedException {
    final long beginTime = System.currentTimeMillis();
    try {

      boolean reachedTimeout = false;
      long currentTimeout = synchTimeout;
      while (currentTimeout > 0) {

        if (currentTimeout > 10000)
          // CUT THE TIMEOUT IN BLOCKS OF 10S EACH TO ALLOW CHECKING FOR ANY SERVER IF UNREACHABLE
          currentTimeout = 10000;

        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
              "Waiting max %dms for collecting all synchronous responses... (timeout=%d reqId=%s thread=%d)", currentTimeout,
              synchTimeout, request.getId(), Thread.currentThread().getId());

        // WAIT FOR THE RESPONSES
        if (synchronousResponsesArrived.await(currentTimeout, TimeUnit.MILLISECONDS)) {
          if (canceled.get())
            throw new ODistributedOperationException("Request has been canceled");

          // COMPLETED
          if (ODistributedServerLog.isDebugEnabled())
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "All synchronous responses collected in %dms (reqId=%s thread=%d)", (System.currentTimeMillis() - beginTime),
                request.getId(), Thread.currentThread().getId());

          return true;
        }

        if (ODistributedServerLog.isDebugEnabled())
          ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
              "All synchronous responses not collected in %dms, waiting again... (reqId=%s thread=%d)",
              (System.currentTimeMillis() - beginTime), request.getId(), Thread.currentThread().getId());

        if (Thread.currentThread().isInterrupted()) {
          // INTERRUPTED
          ODistributedServerLog
              .warn(this, dManager.getLocalNodeName(), null, DIRECTION.NONE, "Thread has been interrupted wait for request (%s)",
                  request);
          Thread.currentThread().interrupt();
          break;
        }

        synchronousResponsesLock.lock();
        try {

          final long now = System.currentTimeMillis();
          final long elapsed = now - beginTime;
          if (elapsed > synchTimeout)
            reachedTimeout = true;
          currentTimeout = synchTimeout - elapsed;

          // CHECK IF ANY NODE ARE UNREACHABLE IN THE MEANWHILE
          int synchronizingNodes = 0;
          int missingActiveNodes = 0;

          Map<String, ODistributedServerManager.DB_STATUS> missingResponseNodeStatuses = new HashMap<String, ODistributedServerManager.DB_STATUS>(
              responses.size());

          int missingResponses = 0;

          for (Iterator<Map.Entry<String, Object>> iter = responses.entrySet().iterator(); iter.hasNext(); ) {
            final Map.Entry<String, Object> curr = iter.next();

            if (curr.getValue() == NO_RESPONSE) {
              missingResponses++;

              // ANALYZE THE NODE WITHOUT A RESPONSE
              final ODistributedServerManager.DB_STATUS dbStatus = dManager.getDatabaseStatus(curr.getKey(), getDatabaseName());

              missingResponseNodeStatuses.put(curr.getKey(), dbStatus);

              switch (dbStatus) {
              case BACKUP:
              case SYNCHRONIZING:
                synchronizingNodes++;
                missingActiveNodes++;
                break;
              case ONLINE:
                missingActiveNodes++;
                break;
              }
            }
          }

          if (missingResponses == 0) {
            // ALL RESPONSE COLLECTED, BUT NO QUORUM REACHED
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "All responses collected %s, but no quorum reached (reqId=%s)", responses, request.getId());
            break;
          }

          request.getTask().checkIsValid(dManager);

          if (missingActiveNodes == 0) {
            // NO MORE ACTIVE NODES TO WAIT
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "No more active nodes to wait for request (%s): anticipate timeout (saved %d ms). Missing servers: %s", request,
                currentTimeout, missingResponseNodeStatuses);
            break;
          }

          final long lastClusterChange = dManager.getLastClusterChangeOn();
          if (lastClusterChange > 0 && now - lastClusterChange < (synchTimeout + ADDITIONAL_TIMEOUT_CLUSTER_SHAPE)) {
            // CHANGED CLUSTER SHAPE DURING WAIT: ENLARGE TIMEOUT
            currentTimeout = synchTimeout;
            ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
                "Cluster shape changed during request (%s): enlarge timeout +%dms, wait again for %dms", request, synchTimeout,
                currentTimeout);
            continue;
          } else if (synchronizingNodes > 0) {
            // SOME NODE IS SYNCHRONIZING: WAIT FOR THEM
            // currentTimeout = synchTimeout;
            // ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE,
            // "%d nodes are in synchronization mode during request (%s): enlarge timeout +%dms, wait again for %dms",
            // synchronizingNodes, request, synchTimeout, currentTimeout);
          }

        } finally {
          synchronousResponsesLock.unlock();
        }
      }

      if (canceled.get())
        throw new ODistributedOperationException("Request has been canceled");

      return isMinimumQuorumReached(reachedTimeout);

    } finally {
      Orient.instance().getProfiler()
          .stopChrono("distributed.synchResponses", "Time to collect all the synchronous responses from distributed nodes",
              beginTime);
    }
  }
public boolean waitForSynchronousResponses ( ) throws InterruptedException { final long beginTime = System . currentTimeMillis ( ) ; try { boolean reachedTimeout = false ; long currentTimeout = synchTimeout ; while ( currentTimeout > 0 ) { if ( currentTimeout > 10000 ) // CUT THE TIMEOUT IN BLOCKS OF 10S EACH TO ALLOW CHECKING FOR ANY SERVER IF UNREACHABLE currentTimeout = 10000 ; if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "Waiting max %dms for collecting all synchronous responses... (timeout=%d reqId=%s thread=%d)" , currentTimeout , synchTimeout , request . getId ( ) , Thread . currentThread ( ) . getId ( ) ) ; // WAIT FOR THE RESPONSES if ( synchronousResponsesArrived . await ( currentTimeout , TimeUnit . MILLISECONDS ) ) { if ( canceled . get ( ) ) throw new ODistributedOperationException ( "Request has been canceled" ) ; // COMPLETED if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "All synchronous responses collected in %dms (reqId=%s thread=%d)" , ( System . currentTimeMillis ( ) - beginTime ) , request . getId ( ) , Thread . currentThread ( ) . getId ( ) ) ; return true ; } if ( ODistributedServerLog . isDebugEnabled ( ) ) ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "All synchronous responses not collected in %dms, waiting again... (reqId=%s thread=%d)" , ( System . currentTimeMillis ( ) - beginTime ) , request . getId ( ) , Thread . currentThread ( ) . getId ( ) ) ; if ( Thread . currentThread ( ) . isInterrupted ( ) ) { // INTERRUPTED ODistributedServerLog . warn ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "Thread has been interrupted wait for request (%s)" , request ) ; Thread . currentThread ( ) . interrupt ( ) ; break ; } synchronousResponsesLock . lock ( ) ; try { final long now = System . currentTimeMillis ( ) ; final long elapsed = now - beginTime ; if ( elapsed > synchTimeout ) reachedTimeout = true ; currentTimeout = synchTimeout - elapsed ; // CHECK IF ANY NODE ARE UNREACHABLE IN THE MEANWHILE int synchronizingNodes = 0 ; int missingActiveNodes = 0 ; Map < String , ODistributedServerManager . DB_STATUS > missingResponseNodeStatuses = new HashMap < String , ODistributedServerManager . DB_STATUS > ( responses . size ( ) ) ; int missingResponses = 0 ; for ( Iterator < Map . Entry < String , Object > > iter = responses . entrySet ( ) . iterator ( ) ; iter . hasNext ( ) ; ) { final Map . Entry < String , Object > curr = iter . next ( ) ; if ( curr . getValue ( ) == NO_RESPONSE ) { missingResponses ++ ; // ANALYZE THE NODE WITHOUT A RESPONSE final ODistributedServerManager . DB_STATUS dbStatus = dManager . getDatabaseStatus ( curr . getKey ( ) , getDatabaseName ( ) ) ; missingResponseNodeStatuses . put ( curr . getKey ( ) , dbStatus ) ; switch ( dbStatus ) { case BACKUP : case SYNCHRONIZING : synchronizingNodes ++ ; missingActiveNodes ++ ; break ; case ONLINE : missingActiveNodes ++ ; break ; } } } if ( missingResponses == 0 ) { // ALL RESPONSE COLLECTED, BUT NO QUORUM REACHED ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "All responses collected %s, but no quorum reached (reqId=%s)" , responses , request . getId ( ) ) ; break ; } request . getTask ( ) . checkIsValid ( dManager ) ; if ( missingActiveNodes == 0 ) { // NO MORE ACTIVE NODES TO WAIT ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "No more active nodes to wait for request (%s): anticipate timeout (saved %d ms). Missing servers: %s" , request , currentTimeout , missingResponseNodeStatuses ) ; break ; } final long lastClusterChange = dManager . getLastClusterChangeOn ( ) ; if ( lastClusterChange > 0 && now - lastClusterChange < ( synchTimeout + ADDITIONAL_TIMEOUT_CLUSTER_SHAPE ) ) { // CHANGED CLUSTER SHAPE DURING WAIT: ENLARGE TIMEOUT currentTimeout = synchTimeout ; ODistributedServerLog . debug ( this , dManager . getLocalNodeName ( ) , null , DIRECTION . NONE , "Cluster shape changed during request (%s): enlarge timeout +%dms, wait again for %dms" , request , synchTimeout , currentTimeout ) ; continue ; } else if ( synchronizingNodes > 0 ) { // SOME NODE IS SYNCHRONIZING: WAIT FOR THEM // currentTimeout = synchTimeout; // ODistributedServerLog.debug(this, dManager.getLocalNodeName(), null, DIRECTION.NONE, // "%d nodes are in synchronization mode during request (%s): enlarge timeout +%dms, wait again for %dms", // synchronizingNodes, request, synchTimeout, currentTimeout); } } finally { synchronousResponsesLock . unlock ( ) ; } } if ( canceled . get ( ) ) throw new ODistributedOperationException ( "Request has been canceled" ) ; return isMinimumQuorumReached ( reachedTimeout ) ; } finally { Orient . instance ( ) . getProfiler ( ) . stopChrono ( "distributed.synchResponses" , "Time to collect all the synchronous responses from distributed nodes" , beginTime ) ; } }
Waits until the minimum responses are collected or timeout occurs. If "waitForLocalNode" wait also for local node.

@return True if the received responses are major or equals then the expected synchronous responses, otherwise false

@throws InterruptedException
Waits until the minimum responses are collected or timeout occurs . If waitForLocalNode wait also for local node .