zim-wiki team mailing list archive

[João Santos <jmcs@xxxxxxxxxx>]

I've made a simple lock mechanism. I believe this lock mechanism should
work both on Windows and Linux (I've only tested on Linux and only on a
local filesystem). I'll test it in windows and shared filesystems later.


2012/3/1 Jaap Karssenberg <jaap.karssenberg@xxxxxxxxx>

> On Thu, Mar 1, 2012 at 11:01 AM, João Santos <jmcs@xxxxxxxxxx> wrote:
>
>> Wouldn't this module: http://packages.python.org/lockfile/lockfile.htmldo the job?
>>
>
> Afraid not. This module will do locking between multiple processes on the
> same system, but it is not suitable for .e.g. multiple systems accessing
> the same share drive.
>
> I think the most important use case for this multi-user sharing is each
> user having his or her own computer, but sharing a share drive.
>
> So locking needs to be platform independent, and can only use the (shared)
> file system.
>
> Regards,
>
> Jaap
>
>

import os
import uuid
import shutil


class LockFailedError(Exception):
    pass


class NotMyLockError(Exception):
    pass


class CrossLock(object):

    def __init__(self, name):
        """Represents a lockfile with a name"""
        self.lock = ".{0}.lock".format(name)
        #We will have a unique id to assure that the lock
        #belongs to the instance
        self.id = uuid.uuid1()
        self.filename = "{0}/{1}".format(self.lock, self.id)

    def _create_lockfile(self):
        """Creates the lock dir and file"""
        #mkdir is atomic on windows and linux
        #and will fail if dir already exists
        os.mkdir(self.lock)
        os.mknod(self.filename)

    def acquire(self):
        """Tries to acquire lock."""
        try:
            self._create_lockfile()
        except:
            raise LockFailedError

    def break_lock(self):
        """Removes the lock directory"""
        shutil.rmtree(self.lock)

    def is_locked(self):
        """Tests if the file is locked"""
        return os.access(self.lock, os.F_OK)

    def is_my_lock(self):
        """Tests if the lock belongs to the instance"""
        return os.access(self.filename, os.F_OK)

    def release(self):
        """Removes the lock but only if owned by the instance"""
        if not self.is_my_lock():
            raise NotMyLockError
        else:
            self.break_lock()

if __name__ == '__main__':
    lock = CrossLock("teste")
    lock2 = CrossLock("teste")

    print "Trying to get get lock"
    try:
        lock.acquire()
        print "Correct (Succeeded)"
    except:
        print "Incorrect (Failed)"

    print "Trying to get a lock again"
    try:
        lock.acquire()
        print "Incorrect (Succeeded)"
    except:
        print "Correct (Failed)"

    print "Trying to release the lock with the wrong instance"
    try:
        lock2.release()
        print "Incorrect (Succeeded)"
    except:
        print "Correct (Failed)"

    print "Tryng to release the lock with the correct instance"
    try:
        lock.release()
        print "Correct (Succeeded)"
    except:
        print "Incorrect (Failed)"
        raise

    print "Testing if it's locked"
    if lock.is_locked():
        print "Incorrect (True)"
    else:
        print "Correct (False)"

    print "Trying to get get lock again"
    try:
        lock.acquire()
        print "Correct (Succeeded)"
    except:
        print "Incorrect (Failed)"

    print "Testing if it's locked"
    if lock.is_locked():
        print "Correct (True)"
    else:
        print "Incorrect (False)"

    print "Trying to break the lock"
    try:
        lock2.break_lock()
        print "Correct (Succeeded)"
    except:
        print "Incorrect (Failed)"