is nonsense, e.g. asking for replica #3 in a 2-Node system.
175986
int D( int repl ) {
    int hsz = H2O.CLOUD.size();
  
    // See if this is a specifically homed Key
    if( !user_allowed() && repl < _kb[1] ) { // Asking for a replica# from the homed list?
      assert _kb[0] != Key.DVEC;
      H2ONode h2o = H2ONode.intern(_kb,2+repl*(4+2/*serialized bytesize of H2OKey*/));
      // Reverse the home to the index
      int idx = h2o.index();
      if( idx >= 0 ) return idx;
      // Else homed to a node which is no longer in the cloud!
      // Fall back to the normal home mode
    }
  
    // Distribution of Fluid Vectors is a special case.
    // Fluid Vectors are grouped into vector groups, each of which must have
    // the same distribution of chunks so that MRTask2 run over group of
    // vectors will keep data-locality.  The fluid vecs from the same group
    // share the same key pattern + each has 4 bytes identifying particular
    // vector in the group.  Since we need the same chunks end up on the same
    // node in the group, we need to skip the 4 bytes containing vec# from the
    // hash.  Apart from that, we keep the previous mode of operation, so that
    // ByteVec would have first 64MB distributed around cloud randomly and then
    // go round-robin in 64MB chunks.
    if( _kb[0] == DVEC ) {
      // Homed Chunk?
      if( _kb[1] != -1 ) throw H2O.unimpl();
      // For round-robin on Chunks in the following pattern:
      // 1 Chunk-per-node, until all nodes have 1 chunk (max parallelism).
      // Then 2 chunks-per-node, once around, then 4, then 8, then 16.
      // Getting several chunks-in-a-row on a single Node means that stencil
      // calculations that step off the end of one chunk into the next won't
      // force a chunk local - replicating the data.  If all chunks round robin
      // exactly, then any stencil calc will double the cached volume of data
      // (every node will have it's own chunk, plus a cached next-chunk).
      // Above 16-chunks-in-a-row we hit diminishing returns.
      int cidx = UnsafeUtils.get4(_kb, 1 + 1 + 4); // Chunk index
      int x = cidx/hsz; // Multiples of cluster size
      // 0 -> 1st trip around the cluster;            nidx= (cidx- 0*hsz)>>0
      // 1,2 -> 2nd & 3rd trip; allocate in pairs:    nidx= (cidx- 1*hsz)>>1
      // 3,4,5,6 -> next 4 rounds; allocate in quads: nidx= (cidx- 3*hsz)>>2
      // 7-14 -> next 8 rounds in octets:             nidx= (cidx- 7*hsz)>>3
      // 15+ -> remaining rounds in groups of 16:     nidx= (cidx-15*hsz)>>4
      int z = x==0 ? 0 : (x<=2 ? 1 : (x<=6 ? 2 : (x<=14 ? 3 : 4)));
      int nidx = (cidx-((1<<z)-1)*hsz)>>z;
      return ((nidx+repl)&0x7FFFFFFF) % hsz;
    }
  
    // Easy Cheesy Stupid:
    return ((_hash+repl)&0x7FFFFFFF) % hsz;
  }
174204
function executeWriteOperation(args, options, callback) {
  if (typeof options === 'function') (callback = options), (options = {});
  options = options || {};

  // TODO: once we drop Node 4, use destructuring either here or in arguments.
  const self = args.self;
  const op = args.op;
  const ns = args.ns;
  const ops = args.ops;

  if (self.state === DESTROYED) {
    return callback(new MongoError(f('topology was destroyed')));
  }

  const willRetryWrite =
    !args.retrying &&
    !!options.retryWrites &&
    options.session &&
    isRetryableWritesSupported(self) &&
    !options.session.inTransaction();

  if (!self.s.replicaSetState.hasPrimary()) {
    if (self.s.disconnectHandler) {
      // Not connected but we have a disconnecthandler
      return self.s.disconnectHandler.add(op, ns, ops, options, callback);
    } else if (!willRetryWrite) {
      // No server returned we had an error
      return callback(new MongoError('no primary server found'));
    }
  }

  const handler = (err, result) => {
    if (!err) return callback(null, result);
    if (!isRetryableError(err)) {
      return callback(err);
    }

    if (willRetryWrite) {
      const newArgs = Object.assign({}, args, { retrying: true });
      return executeWriteOperation(newArgs, options, callback);
    }

    // Per SDAM, remove primary from replicaset
    if (self.s.replicaSetState.primary) {
      self.s.replicaSetState.remove(self.s.replicaSetState.primary, { force: true });
    }

    return callback(err);
  };

  if (callback.operationId) {
    handler.operationId = callback.operationId;
  }

  // increment and assign txnNumber
  if (willRetryWrite) {
    options.session.incrementTransactionNumber();
    options.willRetryWrite = willRetryWrite;
  }

  self.s.replicaSetState.primary[op](ns, ops, options, handler);
}
1004131
def _maybe_connect(self, node_id):
        """"""
        with self._lock:
            conn = self._conns.get(node_id)

            if conn is None:
                broker = self.cluster.broker_metadata(node_id)
                assert broker, 'Broker id %s not in current metadata' % (node_id,)

                log.debug("Initiating connection to node %s at %s:%s",
                          node_id, broker.host, broker.port)
                host, port, afi = get_ip_port_afi(broker.host)
                cb = WeakMethod(self._conn_state_change)
                conn = BrokerConnection(host, broker.port, afi,
                                        state_change_callback=cb,
                                        node_id=node_id,
                                        **self.config)
                self._conns[node_id] = conn

            # Check if existing connection should be recreated because host/port changed
            elif self._should_recycle_connection(conn):
                self._conns.pop(node_id)
                return False

            elif conn.connected():
                return True

            conn.connect()
            return conn.connected()
131890
def get_min_isr(zk, topic):
    """"""
    ISR_CONF_NAME = 'min.insync.replicas'
    try:
        config = zk.get_topic_config(topic)
    except NoNodeError:
        return None
    if ISR_CONF_NAME in config['config']:
        return int(config['config'][ISR_CONF_NAME])
    else:
        return None
1088397
private void processOverReplicatedBlock(Block block, short replication,
                                          DatanodeDescriptor addedNode, DatanodeDescriptor delNodeHint) {

    List<DatanodeID> excessReplicateTmp = new ArrayList<DatanodeID>();
    List<DatanodeID> originalDatanodes = new ArrayList<DatanodeID>();

    // find all replicas that can possibly be deleted.
    // The results are returned in excessReplicateTmp.
    findOverReplicatedReplicas(block, replication,
      addedNode, delNodeHint,
      excessReplicateTmp, originalDatanodes);
    if (excessReplicateTmp.size() <= 0) {
      return;
    }

    writeLock();                     // acquire write lock,
    try {
      BlockInfo storedBlock = blocksMap.getBlockInfo(block);
      INodeFile inode = (storedBlock == null) ? null : storedBlock.getINode(); 
      if (inode == null) {
        return; // file has been deleted already, nothing to do.
      }

      //
      // if the state of replicas of this block has changed since the time
      // when we released and reacquired the lock, then all the decisions
      // that we have made so far might not be correct. Do not delete excess
      // replicas in this case.

      int live = 0;
      Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(block);
      for (Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(block);
           it.hasNext();) {
        DatanodeDescriptor node = it.next();
        if (((nodesCorrupt != null) && (nodesCorrupt.contains(node))) ||
          node.isDecommissionInProgress() || node.isDecommissioned()) {
          // do nothing
        } else {
          live++;            // number of live nodes
          originalDatanodes.remove(node);
        }
      }

      if (originalDatanodes.size() > 0) {
        NameNode.stateChangeLog.info("Unable to delete excess replicas for block " +
          block +
          " because the state of the original replicas have changed." +
          " Will retry later.");
        overReplicatedBlocks.add(block);
        return;
      }
      
      short blockReplication = inode.getBlockReplication(storedBlock);

      // loop through datanodes that have excess-replicas of this block
      for (ListIterator<DatanodeID> iter = excessReplicateTmp.listIterator();
           iter.hasNext();) {
        DatanodeID datanodeId = iter.next();

        // re-check that block still has excess replicas.
        // If not, then there is nothing more to do.
        if (live <= blockReplication) {
          break;
        }

        // find the DatanodeDescriptor for this datanode
        DatanodeDescriptor datanode = null;
        try {
          datanode = getDatanode(datanodeId);
        } catch (IOException e) {
        }
        if (datanode == null) {
          NameNode.stateChangeLog.info("No datanode found while processing " +
            "overreplicated block " + block);
          continue;              // dead datanode?
        }

        // insert into excessReplicateMap
        LightWeightHashSet<Block> excessBlocks = excessReplicateMap.get(datanodeId.getStorageID());
        if (excessBlocks == null) {
          excessBlocks = new LightWeightHashSet<Block>();
          excessReplicateMap.put(datanodeId.getStorageID(), excessBlocks);
        }

        if (excessBlocks.add(block)) {
          excessBlocksCount++;
          if (NameNode.stateChangeLog.isDebugEnabled()) {
            NameNode.stateChangeLog.debug("BLOCK* NameSystem.chooseExcessReplicates: "
              + "(" + datanodeId.getName() + ", " + block
              + ") is added to excessReplicateMap");
          }
        }
        //
        // The 'excessblocks' tracks blocks until we get confirmation
        // that the datanode has deleted them; the only way we remove them
        // is when we get a "removeBlock" message.
        //
        // The 'invalidate' list is used to inform the datanode the block
        // should be deleted.  Items are removed from the invalidate list
        // upon giving instructions to the namenode.
        //
        addToInvalidatesNoLog(block, datanode, true);
        live--;
        if (NameNode.stateChangeLog.isDebugEnabled()) {
          NameNode.stateChangeLog.debug("BLOCK* NameSystem.chooseExcessReplicates: "
            + "(" + datanode.getName() + ", " + block + ") is added to recentInvalidateSets");
        }
      }
    } finally {
      writeUnlock();
    }
  }
1020349
def _request_pre_prepare(self, three_pc_key: Tuple[int, int],
                             stash_data: Optional[Tuple[str, str, str]] = None) -> bool:
        
        recipients = self.primaryName
        return self._request_three_phase_msg(three_pc_key,
                                             self.requested_pre_prepares,
                                             PREPREPARE,
                                             recipients,
                                             stash_data)
1184625
boolean isStale(long[] newVersions, int replicaIndex) {
        int index = replicaIndex - 1;
        long currentVersion = versions[index];
        long newVersion = newVersions[index];
        return currentVersion > newVersion;
    }
1079360
def select(max_staleness, selection):
    """"""
    if max_staleness == -1:
        return selection

    # Server Selection Spec: If the TopologyType is ReplicaSetWithPrimary or
    # ReplicaSetNoPrimary, a client MUST raise an error if maxStaleness <
    # heartbeatFrequency + IDLE_WRITE_PERIOD, or if maxStaleness < 90.
    _validate_max_staleness(max_staleness, selection.heartbeat_frequency)

    if selection.primary:
        return _with_primary(max_staleness, selection)
    else:
        return _no_primary(max_staleness, selection)
1045824
def do_numberrep(self, line):
        """"""
        n_replicas = self._split_args(line, 1, 0)[0]
        self._command_processor.get_session().get_replication_policy().set_number_of_replicas(
            n_replicas
        )
        self._print_info_if_verbose("Set number of replicas to {}".format(n_replicas))
689710
private void verifyReplication(String src,
                                 short replication,
                                 String clientName
  ) throws IOException {
    String text = "file " + src
      + ((clientName != null) ? " on client " + clientName : "")
      + ".\n"
      + "Requested replication " + replication;

    if (replication > maxReplication) {
      throw new IOException(text + " exceeds maximum " + maxReplication);
    }

    if (replication < minReplication) {
      throw new IOException(
        text + " is less than the required minimum " + minReplication);
    }
  }
803671
boolean setReplica( H2ONode h2o ) {
    assert _key.home(); // Only the HOME node for a key tracks replicas
    assert h2o != H2O.SELF;     // Do not track self as a replica
    while( true ) {     // Repeat, in case racing GETs are bumping the counter
      int old = _rwlock.get();
      if( old == -1 ) return false; // Write-locked; no new replications.  Read fails to read *this* value
      assert old >= 0;              // Not negative
      if( RW_CAS(old,old+1,"rlock+") ) break;
    }
    // Narrow non-race here.  Here is a time window where the rwlock count went
    // up, but the replica list does not account for the new replica.  However,
    // the rwlock cannot go down until an ACKACK is received, and the ACK
    // (hence ACKACK) doesn't go out until after this function returns.
    _replicas.add(h2o._unique_idx);
    // Both rwlock taken, and replica count is up now.
    return true;
  }
687230
public void forwardPendingTaskToRejoinNode(long[] replicasAdded, long snapshotSpHandle) {
        if (tmLog.isDebugEnabled()) {
            tmLog.debug("Forward pending tasks in backlog to rejoin node: " + Arrays.toString(replicasAdded));
        }
        if (replicasAdded.length == 0) {
            return;
        }
        boolean sentAny = false;
        for (Map.Entry<DuplicateCounterKey, DuplicateCounter> entry : m_duplicateCounters.entrySet()) {
            if (snapshotSpHandle < entry.getKey().m_spHandle) {
                if (!sentAny) {
                    sentAny = true;
                    if (tmLog.isDebugEnabled()) {
                        tmLog.debug("Start forwarding pending tasks to rejoin node.");
                    }
                }

                // Then forward any message after the MP txn, I expect them are all Iv2InitiateMessages
                if (tmLog.isDebugEnabled()) {
                    tmLog.debug(entry.getValue().getOpenMessage().getMessageInfo());
                }
                m_mailbox.send(replicasAdded, entry.getValue().getOpenMessage());
            }
        }
        if (sentAny && tmLog.isDebugEnabled()) {
            tmLog.debug("Finish forwarding pending tasks to rejoin node.");
        }
    }
653469
private boolean isGoodReplica(DatanodeDescriptor node, Block block) {
    Collection<Block> excessBlocks =
      excessReplicateMap.get(node.getStorageID());
    Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(block);
    return (nodesCorrupt == null || !nodesCorrupt.contains(node)) // not corrupt
        // not over scheduling for replication
        && (node.getNumberOfBlocksToBeReplicated() < maxReplicationStreams)
        // not alredy scheduled for removal
        && (excessBlocks == null || !excessBlocks.contains(block))
        && !node.isDecommissioned(); // not decommissioned
  }
807554
def request_propagates(self, req_keys):
        
        i = 0
        for digest in req_keys:
            if digest not in self.requested_propagates_for:
                if digest not in self.requests:
                    # Request from all nodes
                    self.request_msg(PROPAGATE, {f.DIGEST.nm: digest})
                else:
                    # Request from nodes that didn't send
                    send_to = [conn for conn in self.nodestack.connecteds if
                               conn not in self.requests[digest].propagates.keys()]
                    self.request_msg(PROPAGATE, {f.DIGEST.nm: digest}, frm=send_to)

                self._add_to_recently_requested(digest)
                i += 1
            else:
                logger.debug('{} already requested PROPAGATE recently for {}'.
                             format(self, digest))
        return i
830421
boolean setReplica( H2ONode h2o ) {
    assert _key.home(); // Only the HOME node for a key tracks replicas
    assert h2o != H2O.SELF;     // Do not track self as a replica
    if( !read_lock() ) return false; // Write-locked; no new replications.  Read fails to read *this* value
    // Narrow non-race here.  Here is a time window where the rwlock count went
    // up, but the replica list does not account for the new replica.  However,
    // the rwlock cannot go down until an ACKACK is received, and the ACK
    // (hence ACKACK) doesn't go out until after this function returns.
    replicas()[h2o._unique_idx] = 1;
    // Both rwlock taken, and replica count is up now.
    return true;
  }
1230141
def _error_message(self, selector):
        
        is_replica_set = self._description.topology_type in (
            TOPOLOGY_TYPE.ReplicaSetWithPrimary,
            TOPOLOGY_TYPE.ReplicaSetNoPrimary)

        if is_replica_set:
            server_plural = 'replica set members'
        elif self._description.topology_type == TOPOLOGY_TYPE.Sharded:
            server_plural = 'mongoses'
        else:
            server_plural = 'servers'

        if self._description.known_servers:
            # We've connected, but no servers match the selector.
            if selector is writable_server_selector:
                if is_replica_set:
                    return 'No primary available for writes'
                else:
                    return 'No %s available for writes' % server_plural
            else:
                return 'No %s match selector "%s"' % (server_plural, selector)
        else:
            addresses = list(self._description.server_descriptions())
            servers = list(self._description.server_descriptions().values())
            if not servers:
                if is_replica_set:
                    # We removed all servers because of the wrong setName?
                    return 'No %s available for replica set name "%s"' % (
                        server_plural, self._settings.replica_set_name)
                else:
                    return 'No %s available' % server_plural

            # 1 or more servers, all Unknown. Are they unknown for one reason?
            error = servers[0].error
            same = all(server.error == error for server in servers[1:])
            if same:
                if error is None:
                    # We're still discovering.
                    return 'No %s found yet' % server_plural

                if (is_replica_set and not
                        set(addresses).intersection(self._seed_addresses)):
                    # We replaced our seeds with new hosts but can't reach any.
                    return (
                        'Could not reach any servers in %s. Replica set is'
                        ' configured with internal hostnames or IPs?' %
                        addresses)

                return str(error)
            else:
                return ','.join(str(server.error) for server in servers
                                if server.error)
67798
def do_GET(self, write_status_code_only=False):
        """"""

        path = '/master' if self.path == '/' else self.path
        response = self.get_postgresql_status()

        patroni = self.server.patroni
        cluster = patroni.dcs.cluster

        if not cluster and patroni.ha.is_paused():
            primary_status_code = 200 if response['role'] == 'master' else 503
        else:
            primary_status_code = 200 if patroni.ha.is_leader() else 503

        replica_status_code = 200 if not patroni.noloadbalance and response.get('role') == 'replica' else 503
        status_code = 503

        if patroni.ha.is_standby_cluster() and ('standby_leader' in path or 'standby-leader' in path):
            status_code = 200 if patroni.ha.is_leader() else 503
        elif 'master' in path or 'leader' in path or 'primary' in path or 'read-write' in path:
            status_code = primary_status_code
        elif 'replica' in path:
            status_code = replica_status_code
        elif 'read-only' in path:
            status_code = 200 if primary_status_code == 200 else replica_status_code
        elif cluster:  # dcs is available
            is_synchronous = cluster.is_synchronous_mode() and cluster.sync \
                    and cluster.sync.sync_standby == patroni.postgresql.name
            if path in ('/sync', '/synchronous') and is_synchronous:
                status_code = replica_status_code
            elif path in ('/async', '/asynchronous') and not is_synchronous:
                status_code = replica_status_code

        if write_status_code_only:  # when haproxy sends OPTIONS request it reads only status code and nothing more
            message = self.responses[status_code][0]
            self.wfile.write('{0} {1} {2}\r\n'.format(self.protocol_version, status_code, message).encode('utf-8'))
        else:
            self._write_status_response(status_code, response)