filelock 3.11 breaks our multi-threaded parallel processing use case

[slanzmich]

Hi,

In #219 @csm10495 wondered

I'm curious though: Is there ever a case where this would break someone? I guess if someone expected the object to be the same across threads.. but then why would they be using the lock to begin with?

I guess we have a use case that breaks:

We have a multi-threaded processing pipeline that is set up roughly like this:

1. Initialize and create a SoftFileLock for the output file.
2. Prepare some data for step 3.
3. Generate the output file with the data from step 2.
4. Clean up and release the lock acquired in 1.

These steps are executed for multiple inputs, each generating a different output file. Step 2 for one input can run in parallel to step 3 for another input. However, step 3 for one input must not run in parallel to step 3 for another input, as it uses resources that shall not be shared.

The SoftFileLock is required because there may be another process that is executed on an overlapping set of input files (thus generating an overlapping set of output files). With filelock 3.10, we could pass the lock from the thread executing step 1 to the thread executing step 4, but this broke with #219.

For this use case, I don't see anything wrong with sharing the lock between threads, because the lock is intended to prevent concurrent access between different processes. What do you think?

[gaborbernat]

I'm not sure what would be a workaround here, clearly 99% of case makes sense to have it thread local, but how could we allow usage for this 1% 🤔 cc @csm10495

[csm10495]

Any chance we can have a small code sample to play with?

Is it one lock per thread or are the same locks ever used across threads?

Or does the initial process just generate locks for threads to eventually use?

[csm10495]

I tried via this sample (and the latest released filelock):

import filelock
import tempfile
import pathlib
import threading

LOCK = filelock.SoftFileLock(tempfile.NamedTemporaryFile().name)
TEST_FILE = pathlib.Path(tempfile.NamedTemporaryFile().name)

def in_thread():
    with LOCK:
        # the lol attribute is getting set inside the thread (so with the thread-local copy)
        # ... so since that happens after this check, we never print HAS LOL
        if hasattr(LOCK, 'lol'):
            print("HAS LOL")
        LOCK.lol = 1
        TEST_FILE.write_text(threading.current_thread().name)

if __name__ == '__main__':
    threads = []
    for i in range(10):
        threads.append(threading.Thread(target=in_thread))
        threads[-1].start()

    for t in threads:
        t.join()

    print(TEST_FILE.read_text())

It seems to work properly

[gaborbernat]

@slanzmich can you confirm please

[rgeronimi]

Same issue there - we use the lock to prevent concurrent access between batch processes (not threads). Our code is asynchronous (asyncio) and we lock and unlock through asynchronous calls to asyncio.to_thread (which allocates different threads depending on the state of its internal asyncio thread pool), these threads calling filelock in turn in a synchronous fashion. This used to work very well until now. Following #219 this makes the latest version of filelock unusable in our architecture.

More generally, using asyncio.to_thread to wrap a synchronous function (like filelock methods) into asynchronous code is a pattern becoming more and more common as asynchronous code is becoming more popular in python. This use case prevents any synchronous code to assume that it is running in a particular thread (it has to be thread-agnostic).

[csm10495]

@dmoklaf please can you provide a minimal code sample to reproduce? I tried via:

import filelock
import tempfile
import pathlib
import asyncio
import uuid


def lock_write_verify(lock: filelock.FileLock, file_path: pathlib.Path):
    data = str(uuid.uuid4())
    with lock:
        file_path.write_text(data)
        assert file_path.read_text() == data

async def main():
    file_path = pathlib.Path(tempfile.mktemp())
    lock_path = pathlib.Path(tempfile.mktemp())

    lock = filelock.FileLock(lock_path)

    await asyncio.gather(
        asyncio.to_thread(lock_write_verify, lock, file_path),
        asyncio.to_thread(lock_write_verify, lock, file_path),
        asyncio.to_thread(lock_write_verify, lock, file_path),
        asyncio.to_thread(lock_write_verify, lock, file_path),
        asyncio.to_thread(lock_write_verify, lock, file_path),
    )

if __name__ == '__main__':
    asyncio.run(main())

But it works fine for me.

[csm10495]

Maybe its something like this:

import filelock
import tempfile
import pathlib
import asyncio
import uuid


def write_verify(file_path: pathlib.Path):
    data = str(uuid.uuid4())
    file_path.write_text(data)
    assert file_path.read_text() == data

async def main():
    file_path = pathlib.Path(tempfile.mktemp())
    lock_path = pathlib.Path(tempfile.mktemp())

    lock = filelock.FileLock(lock_path)

    await asyncio.gather(
        asyncio.to_thread(lock.acquire),
        asyncio.to_thread(write_verify, file_path),
        asyncio.to_thread(lock.release),
        asyncio.to_thread(lock.acquire),
        asyncio.to_thread(write_verify, file_path),
        asyncio.to_thread(lock.release),
    )

if __name__ == '__main__':
    for i in range(100000):
        try:
            asyncio.run(main())
        except AssertionError:
            continue
        else:
            print("PASS")
            break

The assertion fails very often on both 3.10.7 and 3.11.0 .. though often on 3.11.0 it hangs.

The possible hang in 3.11.0 is because the order of the gather'd tasks isn't guaranteed by asyncio. So it can wind up doing an acquire while already locked leading to deadlock.. The same hang doesn't happen in 3.10.7 because the lock object is re-entrant and allows you to acquire more than once without calling release.

The assertion in either case shows that this type of usage isn't really safe (in either case) due to the ordering of the threads running.

I feel like this seems like a bit of a stretch though so please if either of you can provide a similarly length failing case, i'd be happy evaluate further.

[rgeronimi]

Here is some sample code that reproduces the issue. The symptom shall be that the programs freezes at some point - this requires to use filelock 3.11.0 (works perfectly with 3.10.7) AND to have some multi-threading going on (hence why I run parallel tasks below and pauses to force the asyncio runtime to allocate multiple threads in its internal pool).

import asyncio
import random

import filelock

async def acquire(l):
	await asyncio.to_thread(l.acquire)

async def pause():
	delay = random.uniform(0.5, 1.5)
	await asyncio.sleep(delay)

async def release(l):
	await asyncio.to_thread(l.release)

async def process_segment(segment):
	print(f"Processing segment {segment}...")
	l = filelock.FileLock(f"segment_{segment}.lock")
	await acquire(l)
	try:
		await pause()
	finally:
		await release(l)
	print(f"Processed segment {segment}")

async def process_segments():
	tasks = []
	for segment in range(40):
		task = asyncio.create_task(process_segment(segment))
		tasks.append(task)
	await asyncio.gather(*tasks)

async def process_all():
	for _ in range(20):
		await process_segments()

asyncio.run(process_all())

[csm10495]

Thanks for the example. Why is the lock needed if each process_segment task is running sequentially?

For example I pulled out the calls to acquire/release (removing the need for the lock) and tried:

import asyncio
import uuid
import random
import pathlib

import filelock

async def acquire(l):
    return
    await asyncio.to_thread(l.acquire)

async def mess_with(segment):
    r = str(uuid.uuid4())
    f = pathlib.Path(f"segment_{segment}.txt")
    f.write_text(r)
    assert f.read_text() == r
    f.unlink()


async def release(l):
    return
    await asyncio.to_thread(l.release)

async def process_segment(segment):
    print(f"Processing segment {segment}...")
    l = filelock.FileLock(f"segment_{segment}.lock")
    await acquire(l)
    try:
        await mess_with(segment)
    finally:
        await release(l)
    print(f"Processed segment {segment}")

async def process_segments():
    tasks = []
    for segment in range(40):
        task = asyncio.create_task(process_segment(segment))
        tasks.append(task)
    await asyncio.gather(*tasks)

async def process_all():
    for _ in range(20):
        await process_segments()

asyncio.run(process_all())

and it seemed to work because mess_with always runs at one point in time per segment.

[csm10495]

Here is a similar version that has a chance for overlapping a segment:

import asyncio
import uuid
import random
import pathlib

import filelock

async def acquire(l):
    await asyncio.to_thread(l.acquire)

def mess_with_thread(segment):
    r = str(uuid.uuid4())
    f = pathlib.Path(f"segment_{segment}.txt")
    f.write_text(r)
    assert f.read_text() == r
    f.unlink()

async def release(l):
    await asyncio.to_thread(l.release)

async def process_segment(segment, l):
    print(f"Processing segment {segment}...")
    await acquire(l)
    try:
        await asyncio.to_thread(mess_with_thread, segment)
    finally:
        await release(l)
    print(f"Processed segment {segment}")

async def process_segments():
    tasks = []
    segment_to_lock = {}
    for _ in range(2):
        for segment in range(40):
            if segment not in segment_to_lock:
                segment_to_lock[segment] = filelock.FileLock(f"segment_{segment}.lock")
            task = asyncio.create_task(process_segment(segment, segment_to_lock[segment]))
            tasks.append(task)
    await asyncio.gather(*tasks)

async def process_all():
    for _ in range(20):
        await process_segments()

asyncio.run(process_all())

This version on 3.10.7 is likely to hit an assertion error because of the re-entrant nature of the lock, which allows multiple threads to work with the segment at the same time. (On 3.11.0 it'll probably hang)

If you don't have overlapping segments (via removing the for _ in range(2)) line it works... but then you don't need the lock to begin with since there is no overlap and it goes back to the given example and the question as to why the lock is needed.

[rgeronimi]

Thanks for the example. Why is the lock needed if each process_segment task is running sequentially?

This is because we use filelock to prevent separate processes from stepping on each other toes while processing data segments - this is our only interest for filelock.

This example is an extreme simplification of our python processes, that I extracted only to reproduce the "freeze" bug as requested - it does not show the multi-processing part of course. But running several times in parallel this process through the bash shell could simulate the multi-processing part easily.

It works perfectly with filelock < 3.11. And it fails every time with 3.11.

[csm10495]

I tried this with 3.11.0:

import asyncio
import random
import os
from concurrent.futures import ProcessPoolExecutor
import filelock

def print_with_pid(*args):
    print(f"[{os.getpid()}]", *args)

async def acquire(l):
	await asyncio.to_thread(l.acquire)

async def pause():
	delay = random.uniform(0.5, 1.5)
	await asyncio.sleep(delay)

async def release(l):
	await asyncio.to_thread(l.release)

async def process_segment(segment):
	print_with_pid(f"Processing segment {segment}...")
	l = filelock.FileLock(f"segment_{segment}.lock")
	await acquire(l)
	try:
		await pause()
	finally:
		await release(l)
	print_with_pid(f"Processed segment {segment}")

async def process_segments():
	tasks = []
	for segment in range(40):
		task = asyncio.create_task(process_segment(segment))
		tasks.append(task)
	await asyncio.gather(*tasks)

async def process_all():
	for _ in range(20):
		await process_segments()

def run_via_asyncio():
    asyncio.run(process_all())

if __name__ == '__main__':

    with ProcessPoolExecutor(32) as executor:
        for i in range(32):
            executor.submit(run_via_asyncio)

        executor.shutdown(wait=True)

This does both multi process and asyncio, and didn't get a hang. What OS are you on? I'm on Windows.