performs redis operations by yielding a passed in block, waiting for a new master to
 show up on the network if the operation throws an exception. if a new master doesn't
 appear after the configured timeout interval, we raise an exception.
557969
def with_failover #:nodoc:
      end_time = Time.now.to_i + @config.redis_failover_timeout.to_i
      begin
        yield
      rescue Exception => e
        Beetle::reraise_expectation_errors!
        logger.error "Beetle: redis connection error #{e} #{@config.redis_server} (#{e.backtrace[0]})"
        if Time.now.to_i < end_time
          sleep 1
          logger.info "Beetle: retrying redis operation"
          retry
        else
          raise NoRedisMaster.new(e.to_s)
        end
      end
    end
279358
def join(self, timeout=None):
        
        remaining = timeout

        while self._cb_poll and (remaining is None or remaining > 0):
            now = time.time()
            rv = self._cb_poll.poll(remaining)
            if remaining is not None:
                remaining -= (time.time() - now)

            for command_buffer, event in rv:
                # This command buffer still has pending requests which
                # means we have to send them out first before we can read
                # all the data from it.
                if command_buffer.has_pending_requests:
                    if event == 'close':
                        self._try_reconnect(command_buffer)
                    elif event == 'write':
                        self._send_or_reconnect(command_buffer)

                # The general assumption is that all response is available
                # or this might block.  On reading we do not use async
                # receiving.  This generally works because latency in the
                # network is low and redis is super quick in sending.  It
                # does not make a lot of sense to complicate things here.
                elif event in ('read', 'close'):
                    try:
                        command_buffer.wait_for_responses(self)
                    finally:
                        self._release_command_buffer(command_buffer)

        if self._cb_poll and timeout is not None:
            raise TimeoutError('Did not receive all data in time.')
58809
function() {
      return Promise.retry(function() {
        log.debug(format('Connecting to redis on %s:%s', REDIS_IP, REDIS_PORT));
        return redis.createClient(REDIS_PORT, REDIS_IP);
      });
    }
467556
def method_missing(method, *args, &block)
      if redis_operation?(method)
        dispatch(method, *args, &block)
      else
        super
      end
    end
786349
public boolean await(long timeout, TimeUnit unit) {
        try {
            return latch.await(timeout, unit);
        } catch (InterruptedException e) {
            throw new RedisCommandInterruptedException(e);
        }
    }
647937
private void checkTimeouts() {
    long waitTime = getMs(PropertyKey.MASTER_WORKER_CONNECT_WAIT_TIME);
    long retryInterval = getMs(PropertyKey.USER_RPC_RETRY_MAX_SLEEP_MS);
    if (waitTime < retryInterval) {
      LOG.warn("{}={}ms is smaller than {}={}ms. Workers might not have enough time to register. "
              + "Consider either increasing {} or decreasing {}",
          PropertyKey.Name.MASTER_WORKER_CONNECT_WAIT_TIME, waitTime,
          PropertyKey.Name.USER_RPC_RETRY_MAX_SLEEP_MS, retryInterval,
          PropertyKey.Name.MASTER_WORKER_CONNECT_WAIT_TIME,
          PropertyKey.Name.USER_RPC_RETRY_MAX_SLEEP_MS);
    }
    checkHeartbeatTimeout(PropertyKey.MASTER_MASTER_HEARTBEAT_INTERVAL,
        PropertyKey.MASTER_HEARTBEAT_TIMEOUT);
    // Skip checking block worker heartbeat config because the timeout is master-side while the
    // heartbeat interval is worker-side.
  }
450373
def wait_for_service(service_name, timeout_sec)
      begin
        timeout(timeout_sec) do
          while @is_ok
            if @master.lookup_service(service_name)
              return true
            end
            sleep(0.1)
          end
        end
      rescue Timeout::Error
        puts "time out for wait service #{service_name}"
        return nil
      rescue
        raise "connection with master failed. master = #{@master_uri}"
      end
    end
1171783
public final void retry(final ConnectionState state, final RetryHandler retryHandler, final ConnectionFailHandler failHandler) throws RetryCancelled {
    if (state.retryCount == 0) {
      state.msBeforeRetry = this.startRetryTime;
    }

    state.retryCount++;
    state.uriIndex = state.retryCount % state.uris.length;

    if (this.shouldRetry(state)) {
      if (logger.isDebugEnabled()) {
        logger.debug("Retry {} to send command", state.retryCount);
      }

      if(state.msBeforeRetry > 0) {
        try {
          state.loop.schedule(new Runnable() {
            @Override
            public void run() {
              try {
                retryHandler.doRetry(state);
              } catch (IOException e) {
                failHandler.catchException(e);
              }
            }
          }, state.msBeforeRetry, TimeUnit.MILLISECONDS);
        } catch (IllegalStateException e) {
          failHandler.catchException(
            new IOException(new CancellationException("Etcd client was closed"))
          );
        }
      } else {
        try {
          retryHandler.doRetry(state);
        } catch (IOException e) {
          failHandler.catchException(e);
        }
      }
    } else {
      throw new RetryCancelled();
    }
  }
186518
public synchronized void waitForEnd()
	throws ServerException,
	       ClientException,
	       IOException {
        try {
            while(!isDone() && !hasError()) {
                wait();
            }
        } catch(InterruptedException e) {
            // break
        }
	    checkError();
    }
1200191
def _wait(self, generator, method, timeout=None, *args, **kwargs):
        """"""
        if self.debug:
            print("waiting for %s to pause" % generator)

        original_timeout = timeout
        while timeout is None or timeout > 0:
            last_time = time.time()
            if self._lock.acquire(False):  # timeout param was added in 3.2
                try:
                    if self.can_resume():
                        return method(generator, *args, **kwargs)
                    elif self.has_terminated():
                        raise RuntimeError("%s has already terminated" % generator)
                finally:
                    self._lock.release()

            if timeout is not None:
                timeout -= time.time() - last_time

        msg = "%s did not pause after %ss" % (generator, original_timeout)
        if self.debug:
            print(msg)
        raise WaitTimeoutError(msg)
952695
def acquire(self, blocking=True):
        
        sleep = self.sleep
        timeout = self.timeout
        while 1:
            unixtime = time.time()
            if timeout:
                timeout_at = unixtime + timeout
            else:
                timeout_at = Lock.LOCK_FOREVER
            timeout_at = float(timeout_at)
            if self.redis.setnx(self.name, timeout_at):
                self.acquired_until = timeout_at
                return True
            # We want blocking, but didn't acquire the lock
            # check to see if the current lock is expired
            try:
                existing = float(self.redis.get(self.name) or 1)
            except ValueError:
                existing = Lock.LOCK_FOREVER

            if existing < unixtime:
                # the previous lock is expired, attempt to overwrite it
                try:
                    existing = float(self.redis.getset(self.name, timeout_at) or 1)
                except ValueError:
                    existing = Lock.LOCK_FOREVER
                if existing < unixtime:
                    # we successfully acquired the lock
                    self.acquired_until = timeout_at
                    return True
            if not blocking:
                return False
            time.sleep(sleep)
1318894
public void waitForLock(T jedis){
        while(!lock(jedis)){
            try {
                logger.debug("Waiting for Redis lock.");
                Thread.sleep(randomInt(75, 125));
            } catch (InterruptedException e) {
                logger.error("Interrupted while waiting for lock.", e);
            }
        }
    }
1356561
void handleFailure(IOException ex, int failures)
      throws IOException {
    // Check if the exception was thrown by the network stack
    if (failoverClient.isShuttingdown() || !shouldHandleException(ex)) {
      throw ex;
    }
    if (failures > FAILURE_RETRY) {
      throw ex;
    }
    try {
      // This might've happened because we are failing over
      if (!watchZK) {
        LOG.debug("Not watching ZK, so checking explicitly");
        // Check with zookeeper
        fsLock.readLock().unlock();
        InjectionHandler.processEvent(InjectionEvent.DAFS_CHECK_FAILOVER);
        fsLock.writeLock().lock();
        boolean failover = false;
        try {
          failover = zkCheckFailover(ex);
        } finally {
          fsLock.writeLock().unlock();
          fsLock.readLock().lock();
        }
        if (failover) {
          return;
        }
      }
      Thread.sleep(1000);
    } catch (InterruptedException iex) {
      LOG.error("Interrupted while waiting for a failover", iex);
      Thread.currentThread().interrupt();
    }
  }
218210
def handle(self):
        ''''''
        try:
            yield self
        except:
            # Requeue the message and raise the original exception
            typ, value, trace = sys.exc_info()
            if not self.processed:
                try:
                    self.req(self.delay())
                except socket.error:
                    self.connection.close()
            raise typ, value, trace
        else:
            if not self.processed:
                try:
                    self.fin()
                except socket.error:
                    self.connection.close()
57124
private <T> T execute(RedisCallBack<T> redisCallBack, List<RedisClient> clients, Object key, boolean isRead) {
        for (int i = 0; i < getRetryTimes(); i++) {
            boolean result = redisCallBack.doInRedis(clients, isRead, key);
            if (result) {
                return redisCallBack.getResult();
            }
        }

        Throwable e = redisCallBack.getException();
        if (e != null) {
            logger.error("Return null because exception occurs: " + e.getMessage(), e);
        }
        return null;
    }
821334
private synchronized void initiateRetry() {
    retryCount += 1;
    currentListener = new RetryingBlockFetchListener();

    logger.info("Retrying fetch ({}/{}) for {} outstanding blocks after {} ms",
      retryCount, maxRetries, outstandingBlocksIds.size(), retryWaitTime);

    executorService.submit(() -> {
      Uninterruptibles.sleepUninterruptibly(retryWaitTime, TimeUnit.MILLISECONDS);
      fetchAllOutstanding();
    });
  }
1200266
public void sync() throws Exception {
        try {
            JedisClusterPipeline pipeline = new JedisClusterPipeline(clusterInfoCache);
            execute(pipeline);
            pipeline.sync();
        } catch (JedisMovedDataException jre) {
            // if MOVED redirection occurred, rebuilds cluster's slot cache,
            // recommended by Redis cluster specification
            connectionHandler.renewSlotCache();
            if (maxAttempts > 0) {
                maxAttempts--;
                sync();
                return;
            }
            throw jre;
        } catch (Exception e) {
            throw e;
        }
    }