Syncrhonous replication of Redis as described here: http://antirez.com/news/66 Since Java
Object class has implemented "wait" method, we cannot use it, so I had to change the name of
the method. Sorry :S
182970
@Override
  public Long waitReplicas(final int replicas, final long timeout) {
    checkIsInMultiOrPipeline();
    client.waitReplicas(replicas, timeout);
    return client.getIntegerReply();
  }
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);
      });
    }
732882
def start
      puts "== Connected to #{redis.client.id}"
      puts "== Waiting for commands from `#{options[:queue]}`"
      
      if options[:daemon]
        puts "== Daemonizing..."
        Daemons.daemonize
      end
      
      loop do
        begin 
          cid = redis.blpop(options[:queue], 10)[1]
          if cmd = commands[cid.to_s]
            puts "\e[33m[#{Time.now}]\e[0m Reacting for `#{cid}` command"
            threads.add(Thread.new { system(cmd) })
          end
        rescue Interrupt, SystemExit
          puts "\nCleaning up..."
          return 0
        rescue => ex
          puts "ERROR: #{ex}"
        end
      end
    end
1248243
@SuppressWarnings("unchecked")
	@Override
	public T retry(final int count, final int waitTime) {
		context().setMaxRetryCount(count).setRetryWaitTime(waitTime);
		return (T) this;
	}
943340
def replicate(ctx, args):
    
    slave = ClusterNode.from_uri(args.node)
    master = ClusterNode.from_uri(args.master)
    if not master.is_master():
        ctx.abort("Node {!r} is not a master.".format(args.master))

    try:
        slave.replicate(master.name)
    except redis.ResponseError as e:
        ctx.abort(str(e))

    Cluster.from_node(master).wait()
1052694
@Override
    public boolean flushPuts(long waitTime, TimeUnit timeUnit) throws InterruptedException
    {
        long    msWaitRemaining = TimeUnit.MILLISECONDS.convert(waitTime, timeUnit);
        synchronized(putCount)
        {
            while ( putCount.get() > 0 )
            {
                if ( msWaitRemaining <= 0 )
                {
                    return false;
                }

                long        startMs = System.currentTimeMillis();

                putCount.wait(msWaitRemaining);

                long        elapsedMs = System.currentTimeMillis() - startMs;
                msWaitRemaining -= elapsedMs;
            }
        }
        return true;
    }
703061
function sync() {
    syncDom.setAttribute('data-sync-state', 'syncing');
    var opts = {continuous: true, complete: syncError};
    db.replicate.to(remoteCouch, opts);
    db.replicate.from(remoteCouch, opts);
  }
1189683
async def waitAccessAsync(self):
    
    async with self.lock:
      while True:
        last_access_ts = self.__getLastAccess()
        if last_access_ts is not None:
          now = time.time()
          last_access_ts = last_access_ts[0]
          time_since_last_access = now - last_access_ts
          if time_since_last_access < self.min_delay_between_accesses:
            time_to_wait = self.min_delay_between_accesses - time_since_last_access
            if self.jitter_range_ms is not None:
              time_to_wait += random.randint(*self.jitter_range_ms) / 1000
            self.logger.debug("Sleeping for %.2fms because of rate limit for domain %s" % (time_to_wait * 1000,
                                                                                           self.domain))
            await asyncio.sleep(time_to_wait)

        access_time = time.time()
        self.__access(access_time)

        # now we should be good... except if another process did the same query at the same time
        # the database serves as an atomic lock, query again to be sure the last row is the one
        # we just inserted
        last_access_ts = self.__getLastAccess()
        if last_access_ts[0] == access_time:
          break
771911
function sync (state, docsOrIds) {
  var syncedObjects = []
  var errors = state.db.constructor.Errors

  return Promise.resolve(state.remote)

  .then(function (remote) {
    return new Promise(function (resolve, reject) {
      if (Array.isArray(docsOrIds)) {
        docsOrIds = docsOrIds.map(toId)
      } else {
        docsOrIds = docsOrIds && [toId(docsOrIds)]
      }

      if (docsOrIds && docsOrIds.filter(Boolean).length !== docsOrIds.length) {
        return Promise.reject(errors.NOT_AN_OBJECT)
      }

      var replication = state.db.sync(remote, {
        doc_ids: docsOrIds,
        include_docs: true
      })

      /* istanbul ignore next */
      replication.catch(function () {
        // handled trough 'error' event
      })

      replication.on('complete', function () {
        resolve(syncedObjects)
      })
      replication.on('error', reject)

      replication.on('change', function (change) {
        syncedObjects = syncedObjects.concat(change.change.docs)

        for (var i = 0; i < change.change.docs.length; i++) {
          state.emitter.emit(change.direction, change.change.docs[i])
        }
      })
    })
  })
}
860469
function request(client, path, callback) {
  var key = SyncLock.generateKey(client.username);
  var id = client.id;

  // Try to set this key/value pair, but fail if the path for the key already exists.
  redis.hsetnx(key, path, id, function(err, reply) {
    if(err) {
      log.error({err: err, client: client}, 'Error trying to set redis key with hsetnx');
      return callback(err);
    }

    if(reply === 1) {
      // Success, we have the lock (path for the key was set). Return a new SyncLock instance
      var lock = new SyncLock(key, id, path);
      log.debug({client: client, syncLock: lock}, 'Lock acquired.');
      return callback(null, lock);
    }

    // Path for key was not set (held by another client). See if the lock owner would be
    // willing to let us take it. We'll wait a bit for a reply, and if
    // we don't get one, assume the client holding the lock, or its server,
    // has crashed, and the lock is OK to take.

    // Act if we don't hear back from the lock owner in a reasonable
    // amount of time, and set the lock ourselves.
    var waitTimer = setTimeout(function() {
      redis.removeListener('lock-response', client._handleLockResponseFn);
      client._handleLockResponseFn = null;

      redis.hset(key, path, id, function(err) {
        if(err) {
          log.error({err: err, client: client}, 'Error setting redis lock key.');
          return callback(err);
        }

        var lock = new SyncLock(key, id, path);
        log.debug({client: client, syncLock: lock}, 'Lock request timeout, setting lock manually.');
        callback(null, lock);
      });
    }, CLIENT_TIMEOUT_MS);
    waitTimer.unref();

    // Listen for a response from the client holding the lock
    client._handleLockResponseFn = function(message) {
      handleLockResponse(message, key, path, client, waitTimer, callback);
    };
    redis.on('lock-response', client._handleLockResponseFn);

    // Ask the client holding the lock to give it to us
    log.debug({client: client}, 'Requesting lock override for ' + path);
    redis.publish(Constants.server.lockRequestChannel, JSON.stringify({key: key, id: id, path: path}));
  });
}
164434
synchronized void sync()
    {
        try
        {
            // check we have more work to do
            if (allocatePosition.get() <= lastSyncedOffset + SYNC_MARKER_SIZE)
                return;

            // allocate a new sync marker; this is both necessary in itself, but also serves to demarcate
            // the point at which we can safely consider records to have been completely written to
            int nextMarker;
            nextMarker = allocate(SYNC_MARKER_SIZE);
            boolean close = false;
            if (nextMarker < 0)
            {
                // ensure no more of this CLS is writeable, and mark ourselves for closing
                discardUnusedTail();
                close = true;

                // wait for modifications guards both discardedTailFrom, and any outstanding appends
                waitForModifications();

                if (discardedTailFrom < buffer.capacity() - SYNC_MARKER_SIZE)
                {
                    // if there's room in the discard section to write an empty header, use that as the nextMarker
                    nextMarker = discardedTailFrom;
                }
                else
                {
                    // not enough space left in the buffer, so mark the next sync marker as the EOF position
                    nextMarker = buffer.capacity();
                }
            }
            else
            {
                waitForModifications();
            }

            assert nextMarker > lastSyncedOffset;

            // write previous sync marker to point to next sync marker
            // we don't chain the crcs here to ensure this method is idempotent if it fails
            int offset = lastSyncedOffset;
            final PureJavaCrc32 crc = new PureJavaCrc32();
            crc.updateInt((int) (id & 0xFFFFFFFFL));
            crc.updateInt((int) (id >>> 32));
            crc.updateInt(offset);
            buffer.putInt(offset, nextMarker);
            buffer.putInt(offset + 4, crc.getCrc());

            // zero out the next sync marker so replayer can cleanly exit
            if (nextMarker < buffer.capacity())
            {
                buffer.putInt(nextMarker, 0);
                buffer.putInt(nextMarker + 4, 0);
            }

            // actually perform the sync and signal those waiting for it
            buffer.force();

            if (close)
                nextMarker = buffer.capacity();

            lastSyncedOffset = nextMarker;
            syncComplete.signalAll();

            CLibrary.trySkipCache(fd, offset, nextMarker);
            if (close)
                internalClose();
        }
        catch (Exception e) // MappedByteBuffer.force() does not declare IOException but can actually throw it
        {
            throw new FSWriteError(e, getPath());
        }
    }
863770
def wait(self):
        
        if self._pool is None:
            return
        self._pool.close()
        self._pool.join()
1061643
@Override
    public void sync(
            final ObjectClass objClass,
            final SyncToken token,
            final SyncResultsHandler handler,
            final OperationOptions options) {

        // check objectclass
        if (objClass == null || (!objClass.equals(ObjectClass.ACCOUNT))) {
            throw new IllegalArgumentException("Invalid objectclass");
        }

        // check objectclass
        if (handler == null) {
            throw new IllegalArgumentException("Invalid handler");
        }

        // get web service client
        final Provisioning provisioning = connection.getProvisioning();
        if (provisioning == null) {
            throw new IllegalStateException("Web Service client not found");
        }

        try {
            final List<WSChange> changes = provisioning.sync();

            SyncDeltaBuilder sdb = null;

            boolean handle = true;
            for (final Iterator<WSChange> i = changes.iterator(); i.hasNext() && handle;) {
                final WSChange change = i.next();

                if (LOG.isOk()) {
                    LOG.ok("Found change {0}", change);
                }

                try {
                    sdb = buildSyncDelta(change);
                    handle = handler.handle(sdb.build());
                } catch (Exception e) {
                    LOG.error(e, "Error building connector object for change {0}", change);
                }
            }
        } catch (ProvisioningException e) {
            LOG.error("Synchronization failed");

            throw new IllegalStateException(e);
        }
    }
1230304
def wait(self):
        
        try:
            SpawningProxy(self.containers, abort_on_error=True).wait()
        except Exception:
            # If a single container failed, stop its peers and re-raise the
            # exception
            self.stop()
            raise
1235144
def persist
      future = nil
      set_id if new_record?
      callback = new_record? ? :create : :update

      self.class.transaction do |redis|
        run_callbacks :save do
          run_callbacks callback do
            redis.pipelined do
              attrs = coerced_attributes
              key = self.class.sti_child? ? self.class.sti_base_key_for(id) : self.class.key_for(id)
              future = attrs.any? ? redis.hmset(key, attrs) : :unchanged

              if new_record?
                if self.class.all_index_enabled?
                  redis.sadd(self.class.key_for(:all), id)
                  redis.sadd(self.class.sti_base_key_for(:all), id) if self.class.sti_child?
                end
                add_to_indexes(redis)
              else
                update_indexes(redis)
              end
            end
          end
        end
      end

      future
    end
242254
def wait_for_lock(self, lockname, locktime=60, auto_renewal=False):
        '''  '''
        pid = os.getpid()
        caller = inspect.stack()[0][3]
        try:
            # rl = redlock.Redlock([{"host": settings.REDIS_SERVERS['std_redis']['host'], "port": settings.REDIS_SERVERS['std_redis']['port'], "db": settings.REDIS_SERVERS['std_redis']['db']}, ])
            rl = redis_lock.Lock(self, lockname, expire=locktime, auto_renewal=auto_renewal)
        except AssertionError:
            if self.logger:
                self.logger.error('Process {0} ({1}) could not get lock {2}. Going ahead without locking!!! {3}'.format(pid, caller, lockname, traceback.format_exc()))
            return False
        cont = 1
        t0 = time.time()
        lock = None
        while not lock:
            time.sleep(.05)
            cont += 1
            if cont % 20 == 0:
                if self.logger:
                    self.logger.debug('Process {0} ({1}) waiting for lock {2}. {3} seconds elapsed.'.format(pid, caller, lockname, time.time() - t0))
            # lock = rl.lock(lockname, locktime_ms)
            try:
                lock = rl.acquire()
            except RedisError:
                pass
        if self.logger:
            self.logger.debug('Process {0} ({1}) got lock {2} for {3} seconds'.format(pid, caller, lockname, locktime))
        return rl
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)