Aquire a file lock in a thread-safe manner that also reaps stale locks
    possibly left behind by processes that crashed hard.
6495
def lock(path, poll_interval=5, max_age=60):
    
    if max_age < 30:
        raise ValueError("`max_age` must be at least 30 seconds")
    if poll_interval < 1:
        raise ValueError("`poll_interval` must be at least 1 second")
    if poll_interval >= max_age:
        raise ValueError("`poll_interval` must be less than `max_age`")
    proc = '{0:d}@{1}'.format(os.getpid(), platform.node())
    lock = LockFile(path)
    log.debug("{0} attempting to lock {1}".format(proc, path))
    while not lock.i_am_locking():
        if lock.is_locked():
            # Protect against race condition
            try:
                # Check age of the lock file
                age = time.time() - os.stat(lock.lock_file)[stat.ST_MTIME]
                # Break the lock if too old (considered stale)
                if age > max_age:
                    lock.break_lock()
                    # What if lock was released and reacquired in the meantime?
                    # We don't want to break a fresh lock!
                    # If a lock is stale then we may have many threads
                    # attempting to break it here at the "same time".
                    # Avoid the possibility of some thread trying to break the
                    # lock after it has already been broken and after the first
                    # other thread attempting to acquire the lock by sleeping
                    # for 0.5 seconds below.
                    log.warning(
                        "{0} broke lock on {1} "
                        "that is {2:d} seconds old".format(
                            proc, path, int(age)))
            except OSError:
                # Lock was released just now
                # os.path.exists(lock.lock_file) is False
                # OSError may be raised by os.stat() or lock.break_lock() above
                pass
        time.sleep(0.5)
        try:
            log.debug(
                "{0} waiting for {1:d} seconds "
                "for lock on {2} to be released".format(
                    proc, poll_interval, path))
            # Use float() here since acquire sleeps for timeout/10
            lock.acquire(timeout=float(poll_interval))
        except LockTimeout:
            pass
    log.debug("{0} locked {1}".format(proc, path))
    yield lock
    lock.release()
    log.debug("{0} released lock on {1}".format(proc, path))
575080
def lock_file(f, block=False):
    
    try:
        flags = fcntl.LOCK_EX
        if not block:
            flags |= fcntl.LOCK_NB
        fcntl.flock(f.fileno(), flags)
    except IOError as e:
        if e.errno in (errno.EACCES, errno.EAGAIN):
            raise SystemExit("ERROR: %s is locked by another process." %
                             f.name)
        raise
1078651
private boolean tryLockCacheFile()
  {
    int cnt = 0;
    boolean locked = false;
    while (cnt < 100 && !(locked = lockCacheFile()))
    {
      try
      {
        Thread.sleep(100);
      }
      catch (InterruptedException ex)
      {
        // doesn't matter
      }
      ++cnt;
    }
    if (!locked)
    {
      LOGGER.debug("Failed to lock the cache file.");
    }
    return locked;
  }
1335611
private final void contendedLock() {
            boolean waiting = false;
            for (int s;;) {
                if (((s = lockState) & ~WAITER) == 0) {
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
                        if (waiting)
                            waiter = null;
                        return;
                    }
                }
                else if ((s & WAITER) == 0) {
                    if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
                        waiting = true;
                        waiter = Thread.currentThread();
                    }
                }
                else if (waiting)
                    LockSupport.park(this);
            }
        }
202427
def locked?
      return false unless File.exists?(lockfile)
      result = false
      open(lockfile, 'w') do |f|
        # exclusive non-blocking lock
        result = !lock(f, File::LOCK_EX | File::LOCK_NB)
      end
      result
    end
1015403
@Override
    public FileLock tryLock(long position, long size, boolean shared) throws IOException
    {
        return new FileLockImpl(this, position, size, shared);
    }
793184
def lock(self):
        '''
        
        '''

        # Open file
        self.__fd = open(self.__lockfile, "w")

        # Get it locked
        if self.__locktype == "wait":
            # Try to get it locked until ready
            fcntl.flock(self.__fd.fileno(), fcntl.LOCK_EX)
        elif self.__locktype == "lock":
            # Try to get the locker if can not raise an exception
            try:
                fcntl.flock(self.__fd.fileno(), fcntl.LOCK_EX|fcntl.LOCK_NB)
            except IOError:
                raise AlreadyLocked("File is already locked")
1245306
private void blockIfNecessary() {
        if (diskPermits == null || diskLimit == Long.MAX_VALUE)
            return;
        try {
            diskPermits.acquire();
        } catch (InterruptedException e) {
            // don't want to re-interrupt so it won't cause SDK stream to be
            // closed in case the thread is reused for a different request
            throw new AbortedException(e);
        }
    }
372515
def raise_or_lock(self, key, timeout):
        
        lock_path = self._get_lock_path(key)
        try:
            # Create lock file, raise exception if it exists
            fd = os.open(lock_path, os.O_CREAT | os.O_EXCL)
        except OSError as error:
            if error.errno == errno.EEXIST:
                # File already exists, check its modification time
                mtime = os.path.getmtime(lock_path)
                ttl = mtime + timeout - time.time()
                if ttl > 0:
                    raise AlreadyQueued(ttl)
                else:
                    # Update modification time if timeout happens
                    os.utime(lock_path, None)
                    return
            else:
                # Re-raise unexpected OSError
                raise
        else:
            os.close(fd)
375405
void unlockProcessWait(ServerRequest.PROCESS_WAIT_LOCK lock) {
        synchronized (reqQueueLockObject) {
            for (ServerRequest req : queue) {
                if (req != null) {
                    req.removeProcessWaitLock(lock);
                }
            }
        }
    }
1177261
def acquire_try_once(self):
        
        
        if self.is_locked_by_me():
            return True
        else:
            try:
                self.fd = os.open(self.lock_filename, os.O_CREAT | os.O_RDWR | os.O_EXCL)
            except FileExistsError:
                logger.debug('I tried to get the lock {}, but another process was faster.'.format(self.lock_filename))
                return False
            else:
                os.write(self.fd, str(self.pid).encode('utf-8'))
                logger.debug('I got the lock {}.'.format(self.lock_filename))
                return True
643730
private <T> T doWithLock(LockCallback<T> callback) throws JobPersistenceException {
        return doWithLock(callback, null);
    }
436027
def interprocess_locked(path):
    """"""

    lock = InterProcessLock(path)

    def decorator(f):

        @six.wraps(f)
        def wrapper(*args, **kwargs):
            with lock:
                return f(*args, **kwargs)

        return wrapper

    return decorator
1156733
def is_locked_by_me(self):
        
        
        ## get pid of lock
        lock_pid = self.lock_pid()

        ## not locked
        if lock_pid is None:
            logger.debug('Lock {} is not aquired.'.format(self.lock_filename))
            return False
        
        ## locked by me
        if self.pid == lock_pid:
            logger.debug('Lock {} is aquired by me (pid: {}).'.format(self.lock_filename, self.pid))
            return True

        ## locked by alive process
        try:
            os.kill(lock_pid, 0)
            logger.debug('Lock {} is aquired by another (alive) process (pid: {}).'.format(self.lock_filename, lock_pid))
        
        ## locked by dead process
        except OSError:
            logger.debug('Lock {} is aquired by a dead process (pid: {}).'.format(self.lock_filename, lock_pid))
            try:
                lock_fd = os.open(self.lock_filename, os.O_RDWR)
            except FileNotFoundError:
                logger.debug('The lock is now removed by another process.')
            else:
                try:
                    fcntl.lockf(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                except BlockingIOError:
                    logger.debug('The lock will be removed by another process.')
                else:
                    if self.lock_pid() == lock_pid:
                        os.remove(self.lock_filename)
                        fcntl.lockf(lock_fd, fcntl.LOCK_UN)
                        logger.debug('The lock is removed by me.')
                    else:
                        logger.debug('The lock is substituted by another lock.')
                os.close(lock_fd)
        return False
608735
private function acquireLock()
    {
        if (file_exists($this->lockFile)) {
            $stale_after = filemtime($this->lockFile) + $this->maxLockAge;
            if (time() > $stale_after) {
                // The lockfile is stale, delete it and reacquire a lock
                @rmdir($this->lockFile);
            } else {
                // The lockfile is valid, WURFL is probably being reloaded in another thread
                return false;
            }
        }
        
        // Using mkdir instead of touch since mkdir is atomic
        $this->isLocked = @mkdir($this->lockFile, 0775);
        return $this->isLocked;
    }
1199035
private FileLock getOldestExpiredLock() throws IOException {
    // 1 - acquire lock on dir
    DirLock dirlock = DirLock.tryLock(hdfs, lockDirPath);
    if (dirlock == null) {
      dirlock = DirLock.takeOwnershipIfStale(hdfs, lockDirPath, lockTimeoutSec);
      if (dirlock == null) {
        LOG.debug("Spout {} could not take over ownership of DirLock for {}", spoutId, lockDirPath);
        return null;
      }
      LOG.debug("Spout {} now took over ownership of abandoned DirLock for {}", spoutId, lockDirPath);
    } else {
      LOG.debug("Spout {} now owns DirLock for {}", spoutId, lockDirPath);
    }

    try {
      // 2 - if clocks are in sync then simply take ownership of the oldest expired lock
      if (clocksInSync) {
        return FileLock.acquireOldestExpiredLock(hdfs, lockDirPath, lockTimeoutSec, spoutId);
      }

      // 3 - if clocks are not in sync ..
      if( lastExpiredLock == null ) {
        // just make a note of the oldest expired lock now and check if its still unmodified after lockTimeoutSec
        lastExpiredLock = FileLock.locateOldestExpiredLock(hdfs, lockDirPath, lockTimeoutSec);
        lastExpiredLockTime = System.currentTimeMillis();
        return null;
      }
      // see if lockTimeoutSec time has elapsed since we last selected the lock file
      if( hasExpired(lastExpiredLockTime) ) {
        return null;
      }

      // If lock file has expired, then own it
      FileLock.LogEntry lastEntry = FileLock.getLastEntry(hdfs, lastExpiredLock.getKey());
      if( lastEntry.equals(lastExpiredLock.getValue()) ) {
        FileLock result = FileLock.takeOwnership(hdfs, lastExpiredLock.getKey(), lastEntry, spoutId);
        lastExpiredLock = null;
        return  result;
      } else {
        // if lock file has been updated since last time, then leave this lock file alone
        lastExpiredLock = null;
        return null;
      }
    } finally {
      dirlock.release();
      LOG.debug("Released DirLock {}, SpoutID {} ", dirlock.getLockFile(), spoutId);
    }
  }
902256
public void lockBlocks(long fileId, List<Long> blockIds) throws IOException {
    Map<Long, Long> blockIdToLockId = new HashMap<>();
    List<Throwable> errors = new ArrayList<>();
    synchronized (mLock) {
      if (mPersistingInProgressFiles.containsKey(fileId)) {
        throw new IOException("the file " + fileId + " is already being persisted");
      }
    }
    try {
      // lock all the blocks to prevent any eviction
      for (long blockId : blockIds) {
        long lockId = mBlockWorker.lockBlock(Sessions.CHECKPOINT_SESSION_ID, blockId);
        blockIdToLockId.put(blockId, lockId);
      }
    } catch (BlockDoesNotExistException e) {
      errors.add(e);
      // make sure all the locks are released
      for (long lockId : blockIdToLockId.values()) {
        try {
          mBlockWorker.unlockBlock(lockId);
        } catch (BlockDoesNotExistException bdnee) {
          errors.add(bdnee);
        }
      }

      if (!errors.isEmpty()) {
        StringBuilder errorStr = new StringBuilder();
        errorStr.append("failed to lock all blocks of file ").append(fileId).append("\n");
        for (Throwable error : errors) {
          errorStr.append(error).append('\n');
        }
        throw new IOException(errorStr.toString());
      }
    }
    synchronized (mLock) {
      mPersistingInProgressFiles.put(fileId, blockIdToLockId);
    }
  }