<commit_before># -*- coding: utf-8 -*-

"""
Lock/semaphore implementation based on redis server.
The algorithm implemented is "Problem 2" in the following paper:
http://cs.nyu.edu/~lerner/spring10/MCP-S10-Read04-ReadersWriters.pdf

Using redis allows locks to be shared among processes.

Redis locks inspired by:
http://www.dr-josiah.com/2012/01/creating-lock-with-redis.html
http://redis.io/topics/distlock

Note that — in addition to a lock name — every acquire/release operation
requires an identifier. This guarantees that a lock can only be released by
the client (process/thread) that originally acquired it. Unless, of course,
the identifier is known to other clients as well (which is also a reasonable
use case).

For example a client may acquire a lock and be busy with an expensive
operation that takes longer than the lock's timeout. This causes the lock
to be automatically released. After that happened, another client may
acquire the same lock (with a different identifier). The different
identifier now prohibits the first client from releasing the lock, which
is good because the second client may be performing critical operations.
This reduces (but does not eliminate) potential damage. Clearly,
programmers should make sure that clients do not exceed lock timeouts.
"""

import os
import sys
import time
import contextlib
import uuid
import signal
from functools import wraps
from collections import defaultdict
import signal   # TODO remove

import redis

from .exithandler import handle_exit


# we make sure that redis connections do not time out
redis_conn = redis.StrictRedis(host='localhost', port=6379, db=0,
                               decode_responses=True)  # important for Python3


DEFAULT_TIMEOUT = 20  # seconds
ACQ_TIMEOUT = 15


# note that the process releasing the read/write lock may not be the
# same as the one that acquired it, so the identifier may have
# changed and the lock is never released!
# => we use an identifier unique to all readers/writers.
WRITELOCK_ID = 'id_reader'
READLOCK_ID = 'id_writer'


def reader(f):
    """
    Decorates methods reading an HDF5 file.
    """

    @wraps(f)
    def func_wrapper(self, *args, **kwargs):
        """
        Wraps reading functions.
        """

        # names of locks
        mutex3 = 'mutex3__{}'.format(self.file)
        mutex1 = 'mutex1__{}'.format(self.file)
        readcount = 'readcount__{}'.format(self.file)
        r = 'r__{}'.format(self.file)
        w = 'w__{}'.format(self.file)

        with handle_exit():
            # Note that try/finally must cover incrementing readcount as well
            # as acquiring w. Otherwise readcount/w cannot be
            # decremented/released if program execution ends, e.g., while
            # performing reading operation (e.g., because of a SIGTERM signal).
            readcount_val = None
            w_result = None
            try:
                with redis_lock(redis_conn, mutex3):
                    with redis_lock(redis_conn, r):
                        with redis_lock(redis_conn, mutex1):
                            readcount_val = redis_conn.incr(readcount, amount=1)

                            print("killing myself in 5 seconds...")
                            time.sleep(5)
                            os.kill(os.getpid(), signal.SIGTERM)

                            if readcount_val == 1:
                                # The first reader sets the write lock (if
                                # readcount_val > 1 it is already set).
                                # This locks out all writers.
                                w_result = acquire_lock(redis_conn, w, WRITELOCK_ID)
                                if not w_result:
                                    raise LockException("could not acquire write lock "
                                                        " {0}".format(w))
                result = f(self, *args, **kwargs)  # perform reading operation
                return result
            finally:
                # check if readcount has been incremented above. If so, we have
                # to decrement it.
                # Note that, if readcount was not incremented, then w
                # also was not acquired by this thread.
                print("@reader PID {0}, finally clause!!!".format(os.getpid()))<commit_msg>added comments, removed unused imports<commit_after># -*- coding: utf-8 -*-

"""
Lock/semaphore implementation based on redis server.
The algorithm implemented is "Problem 2" in the following paper:
http://cs.nyu.edu/~lerner/spring10/MCP-S10-Read04-ReadersWriters.pdf
Note that the proposed solution works for threads accessing a shared resource.
To get a working solution for process-based concurrency, one has to deal
with (unexpected) process termination, which makes our solution slightly
more involved.

Using redis allows locks to be shared among processes.
Redis locks inspired by:
http://www.dr-josiah.com/2012/01/creating-lock-with-redis.html
http://redis.io/topics/distlock

Note that — in addition to a lock name — every acquire/release operation
requires an identifier. This guarantees that a lock can only be released by
the client (process/thread) that acquired it. Unless, of course,
the identifier is known to other clients as well (which is also a reasonable
use case).
For example a client may acquire a lock and be busy with an expensive
operation that takes longer than the lock's timeout. This causes the lock
to be automatically released. After that happened, another client may
acquire the same lock (with a different identifier). The different
identifier now prohibits the first client from releasing the lock, which
is good because the second client may be performing critical operations.
This reduces (but does not eliminate) potential damage. Clearly,
programmers should make sure that clients do not exceed lock timeouts.
"""

import os
import time
import contextlib
import uuid
import signal
from functools import wraps
import signal   # TODO remove

import redis

from .exithandler import handle_exit


# we make sure that redis connections do not time out
redis_conn = redis.StrictRedis(host='localhost', port=6379, db=0,
                               decode_responses=True)  # important for Python3


DEFAULT_TIMEOUT = 20  # seconds
ACQ_TIMEOUT = 15


# note that the process releasing the read/write lock may not be the
# same as the one that acquired it, so the identifier may have
# changed and the lock is never released!
# => we use an identifier unique to all readers/writers.
WRITELOCK_ID = 'id_reader'
READLOCK_ID = 'id_writer'


def reader(f):
    """
    Decorates methods reading an HDF5 file.
    """

    @wraps(f)
    def func_wrapper(self, *args, **kwargs):
        """
        Wraps reading functions.
        """

        # names of locks
        mutex3 = 'mutex3__{}'.format(self.file)
        mutex1 = 'mutex1__{}'.format(self.file)
        readcount = 'readcount__{}'.format(self.file)
        r = 'r__{}'.format(self.file)
        w = 'w__{}'.format(self.file)

        with handle_exit():
            # Note that try/finally must cover incrementing readcount as well
            # as acquiring w. Otherwise readcount/w cannot be
            # decremented/released if program execution ends, e.g., while
            # performing reading operation (e.g., because of a SIGTERM signal).
            readcount_val = None
            w_result = None
            try:
                with redis_lock(redis_conn, mutex3):
                    with redis_lock(redis_conn, r):
                        # mutex1's purpose is to make readcount++ together with
                        # the readcount == 1 check atomic
                        with redis_lock(redis_conn, mutex1):
                            readcount_val = redis_conn.incr(readcount, amount=1)

                            print("killing myself in 5 seconds...")
                            time.sleep(5)
                            os.kill(os.getpid(), signal.SIGTERM)

                            # first reader sets the w lock to block writers
                            if readcount_val == 1:
                                w_result = acquire_lock(redis_conn, w, WRITELOCK_ID)
                                if not w_result:
                                    raise LockException("could not acquire write lock "
                                                        " {0}".format(w))
                result = f(self, *args, **kwargs)  # perform reading operation
                return result
            finally:
                # check if readcount has been incremented above. If so, we have
                # to decrement it.
                # Note that, if readcount was not incremented, then w
                # also was not acquired by this thread.
                print("@reader PID {0}, finally clause!!!".format(os.getpid()))
{'fraction_non_alphanumeric': 0.0728744939271255, 'fraction_numerical': 0.008659469185784975, 'mean_word_length': 2.1726721369960758, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3262972', 'n_tokens_mistral': 2356, 'n_tokens_neox': 2261, 'n_words': 916}