code: def test_locking_thread():
      import traceback
  
      with tempfile.TemporaryDirectory() as dir_name:
  
          def test_fn_1(arg):
              try:
                  with lock_ctx(dir_name):
                      # Notify the outside that we've obtained the lock
                      arg.append(False)
                      while True not in arg:
                          time.sleep(0.5)
              except Exception:
                  # Notify the outside that we done
                  arg.append(False)
                  # If something unexpected happened, we want to know what it was
                  traceback.print_exc()
  
          def test_fn_2(arg):
              try:
                  with lock_ctx(dir_name, timeout=0.1):
                      # If this can get the lock, then our file lock has failed
                      raise AssertionError()
              except filelock.Timeout:
                  # It timed out, which means that the lock was still held by the
                  # first thread
                  arg.append(True)
              except Exception:
                  # If something unexpected happened, we want to know what it was
                  traceback.print_exc()
  
          res = []
          thread_1 = threading.Thread(target=test_fn_1, args=(res,))
          thread_2 = threading.Thread(target=test_fn_2, args=(res,))
  
          thread_1.start()
  
          # Make sure the first thread has obtained the lock
          while False not in res:
              time.sleep(0.5)
  
          thread_2.start()
  
          # The second thread should raise `filelock.Timeout`
          thread_2.join()
          assert True in res
  
          thread_1.join()
          assert not thread_1.is_alive()
          assert not thread_2.is_alive()
  
file_path: pytensor/tests/compile/test_compilelock.py
imports: import multiprocessing
  import os
  import sys
  import tempfile
  import threading
  import time
  import filelock
  import pytest
  from pytensor.compile.compilelock import force_unlock, local_mem, lock_ctx
label_desc: Use image node_id 5 for calling a global function with example usage: test_locking_thread() without return types