Restarting or replacing a stopped thread in Python3?

In the following, a thread is started which soon stops because of a raised exception. How can the main section detect when the thread has failed and create a new replacement, while keeping pause() in effect?

#!/usr/bin/env python3                                                          

from threading import Thread
from time import sleep
from signal import pause

def pool():
    while True:
        print("Threading")
        sleep(4)
        raise OSError("Uff da")

sample = Thread(target=pool, daemon=True, name="working")
sample.start()
pause()

 signal.pause()
    Cause the process to sleep until a signal is received; the appropriate handler will then be called. Returns nothing.

So you need to replace pause and check the threads (if they are running or exited somehow) repeatedly, see here (for example): https://superfastpython.com/thread-wait-for-result/

Thanks. I'm not sure it would be feasible to replace pause() there. The above is a very simplified version of a more complicated script which is having the trouble. The pause() is a stand-in for mqtt.loop_forever() from paho.mqtt.client though maybe there is another approach using mqtt.loop_start() instead.

    def loop_forever(self, timeout=1.0, max_packets=1, retry_first_connection=False):
        """This function calls the network loop functions for you in an
        infinite blocking loop. It is useful for the case where you only want
        to run the MQTT client loop in your program.

You cannot use loop_forever in a multithreaded environment.

Thanks. I'll try with something similar to the following. The old script tended to fail one every week or two, so it'll be a while before it'll be seen if there is an improvement.

#!/usr/bin/env python3

from threading import Thread
from time import sleep
import signal

def handler(signum, frame):
    signame = signal.Signals(signum).name
    print(" OK",signame)
    exit(0)

def pool():
    while True:
        print("Threading")
        sleep(4)
        raise OSError("Uff da")

signal.signal(signal.SIGINT, handler)

# add mqtt loop_start() here

while True:
    process_thread = Thread(target=pool,
                        daemon=True,
                        name="working")
    process_thread.start()
    process_thread.join()

you have to dump all relevant parts into a logfile, including the full exception trace/message

Originally Posted by pan64 you have to dump all relevant parts into a logfile, including the full exception trace/message

I tracked that down already: There are some static system files which are, apparently, temporarily unreadable once every few weeks. So the real problem may be with the OS but since that is Raspberry Pi OS, fat chance of getting it fixed. Better to wait for the next edition instead and try some work-arounds until then.

The two work-arounds would be the above modification and to move the reading up in the script so that it happens once per session and not once per nested loop.

hm. I don't really understand that (how can a static system file disappear), but such problems [probably] can be solved with a simple retry.

Originally Posted by pan64 hm. I don't really understand that (how can a static system file disappear), but such problems [probably] can be solved with a simple retry.

Yes, a third work-around would be to wrap the relevant lines in a try: conditional, but I'll use the above two for now.

I don't know that the the system file actually disappears, just that every once in a very rare while trying to read it results in an OSError exception. I tried running strace on the whole script but since it kept running even after the thread raised an exception, I could not capture what had gone on precisely. Though if it happen again, I'll add some removable storage and capture the strace output to it.

that may mean concurrent [write] access to that file, like logs or similar, but without details hard to say more.

See this:

https://stackoverflow.com/questions/...er-or-a-signal


Another strategy is to create a thread which does nothing but sleep() and then signal the parent thread in an endless loop.

Thanks. I'll look into .Event() and see if I can both figure it out and if it would help.

I generally avoid stankoverflow posts since looking at the few topics I know about shows that it's a site full of wrong answers.

How can the main section detect when the thread has failed and create a new replacement, while keeping pause() in effect?

If I understand what you are wanting.

#!/usr/bin/python

#Lock thread example with Queue
#Lock thread and wait until all items are processed. No race condition. No other
#thread can write at the same time.

from threading import Thread, Lock, current_thread
from queue import Queue
from time import sleep

def worker(q, lock):
    #Inf loop, wait until available
    while True:
        value = q.get()
        with lock:
            print(f'In {current_thread().name} got Value {value}')
        q.task_done()
        sleep(1)
        
if __name__ == "__main__":
    
    q = Queue()
    lock = Lock()
    
    num_threads = 10
    
    for i in range(num_threads):
        thread = Thread(target=worker, args=(q, lock))
        thread.daemon=True
        thread.start()
    
    for i in range(1, 21):
        q.put(i)
        
    q.join()

So you just want one thread at a time, and each one starts when the previous one exits?

This works:

#!/usr/bin/env python3

import threading
import time

lock = threading.Lock()


def hello():
    lock.acquire()
    time.sleep(1)
    print("Hello world")
    lock.release()


for _ in range(5):
    t = threading.Thread(target=hello)
    t.start()

If something in the thread can throw an exception, you need to catch it.

#!/usr/bin/python

#Processes like threads don't live in the same memory space. They need shared memory objects.
#These processes may try to access the same shared variable at the same time

from multiprocessing import Process, Value, Array
import os
from time import sleep
 
def add_100(number):
    for i in range(100):
        sleep(.01)
        number.value += 1
        
if __name__ == "__main__":
    shared_number = Value('i', 0)
    print('Number at beginning is', shared_number.value) 
    
    p1 = Process(target=add_100, args=(shared_number,))
    p2 = Process(target=add_100, args=(shared_number,))
     
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    print('Number at end is', shared_number.value)
        
        
#Lock them
#These won't walk on each other.

from multiprocessing import Process, Value, Array, Lock
import os
from time import sleep 
        
def add_100(number, lock):
    for i in range(100):
        sleep(.01)
        lock.acquire()
        number.value += 1
        lock.release() 
        
if __name__ == "__main__":
    
    lock = Lock()
    shared_number = Value('i', 0)
    print('Number at beginning is', shared_number.value) 
    
    p1 = Process(target=add_100, args=(shared_number, lock))
    p2 = Process(target=add_100, args=(shared_number, lock))
     
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    print('Number at end is', shared_number.value)