An atomic, thread-safe incrementing counter for Python

atomic_counter.py

"""An atomic, thread-safe incrementing counter."""

import threading


class AtomicCounter:
    """An atomic, thread-safe incrementing counter.

    >>> counter = AtomicCounter()
    >>> counter.increment()
    1
    >>> counter.increment(4)
    5

    >>> counter = AtomicCounter(42.5)
    >>> counter.value
    42.5
    >>> counter.increment(0.5)
    43.0

    >>> counter = AtomicCounter()
    >>> def incrementor():
    ...     for i in range(100000):
    ...         counter.increment()
    >>> threads = []
    >>> for i in range(4):
    ...     thread = threading.Thread(target=incrementor)
    ...     thread.start()
    ...     threads.append(thread)
    >>> for thread in threads:
    ...     thread.join()
    >>> counter.value
    400000
    """
    def __init__(self, initial=0):
        """Initialize a new atomic counter to given initial value (default 0)."""
        self.value = initial
        self._lock = threading.Lock()

    def increment(self, num=1):
        """Atomically increment the counter by num (default 1) and return the
        new value.
        """
        with self._lock:
            self.value += num
            return self.value


if __name__ == '__main__':
    import doctest
    doctest.testmod()

This doesn't perfectly fit your original doc tests because it's only an int counter, but this gist is currently the #3 result on google for "python atomic increment", so I figured it might help some people.

Here is a lockless (CPython GIL-dependent/itertools-written-in-C-dependent) way to do atomic integer increments and decrements:

import itertools

class Counter:
    def __init__(self):
        self._incs = itertools.count()
        self._decs = itertools.count()

    def inc(self):
        next(self._incs)

    def dec(self):
        next(self._decs)

    def count(self):
        return next(self._incs) - next(self._decs)

If you really need to jump more than 1 at a time and don't mind linear time to do that, you can add something like:

    def update(self, num):
        if num > 0:
            op = self.inc
        else:
            op = self.dec

        for _ in range(abs(num)):
            op()

I haven't timed it, but I'd guess for a smallish num, this could still be faster than explicit locking, especially under high contention.

@PhilMarsh You sure this is alright? The docs don't say anything about itertools.count() being thread-safe.

@PhilMarsh You sure this is alright? The docs don't say anything about itertools.count() being thread-safe.

The lockless approach does assume a count() implementation that is thread-safe. In CPython, the entire itertools module is written in C for maximum performance. So a call to itertools.count() is a single python expression, and thus atomic under the GIL. This isn't guaranteed by any standard, but it is a known implementation detail.

In other flavors of python, YMMV.

If you're writing a portable library, stick with the explicit locking in benhoyt's version.
But if you're writing code only for yourself or your company, and you control the runtime, then leaning on the GIL to avoid lock contention can be useful.

---

Also note that the lockless Counter.count() is not atomic. An extra inc() or dec() can sneak in between the two next() calls. So Counter.count() isn't guaranteed to be strictly monotonically increasing when there are many parallel writes happening. But, no information is ever lost, so the count will always settle on the correct value, when the system is "at rest" (ie: no new writes). And in a high-throughput multithreaded use-case, an occasional +/-1 shouldn't be a problem.¹

---

¹ The count will be out-of-date by the time it is read anyway, unless there is locking at a higher level to block writers while the count is being read and used to make some decision. And if there is such locking, then it would mean the system is "at rest" for the read, so the count has settled and is correct for that moment.

Thank you!