Threading

Compare the Difference

1. Consider the following code, which takes about 20 seconds to run:

   from icmplib import ping
   
   def main():
       hosts = ['dns.google',
               'family.cloudflare-dns.com',
               'moodle.ncnu.edu.tw',
               'sis.ncnu.edu.tw',
               ]
       for host in hosts:
           h = ping(host, privileged=False)
           result = f"{host} ({h.address}) is\t"
           if h.is_alive:
               result += 'alive'
           else:
               result += 'unreachable'
           print(result)
   
   main()

2. The major delay occurs when a host does not respond to the ICMP Echo Request. Your program must wait for the possible delayed packets. Four pings will take about 11 seconds.
3. On the contrary, 4 successful pings will only take 3 seconds.
4. The command "time python icmp.py" shows that the user does not consume much CPU time, nor does system function calls. The process is mostly idle, waiting for ICMP Echo Reply.

   real    0m20.171s
   user    0m0.072s
   sys     0m0.036s

5. Now imagine you have to scan all 254 hosts in 10.20.128.0/24. Even in best case, it will take longer than 10 minutes!
6. For I/O bound tasks, multi-threading is a perfect skill to speed up these tasks. (On the contrary, for CPU bound tasks, multi-tasking does not help much. You should offload some tasks to other machines.)
7. Try the following code:

   from icmplib import ping
   import threading
   
   def probe(host):
       h = ping(host, privileged=False)
       result = f"{host} ({h.address}) is\t"
       if h.is_alive:
           result += 'alive'
       else:
           result += 'unreachable'
       print(result)
   
   def main():
       hosts = ['dns.google',
               'family.cloudflare-dns.com',
               'moodle.ncnu.edu.tw',
               'sis.ncnu.edu.tw',
               ]
       for host in hosts:
           t = threading.Thread(target=probe, args=(host,))
           t.start()
   
   main()

8. This significantly reduces the idle time because 4 threading are running concurrently.

Creating Threads

1. In Python, creating threads are easy. You simply import the threading module, and use it to create threads to run functions concurrently.

   import threading
   
   def worker():
       """thread worker function"""
       print('Worker')
   
   
   for i in range(5):
       t = threading.Thread(target=worker)
       t.start()
   

2. Once a thread object is created, its activity must be started by calling the thread's start() method.
3. Most of the time, you may want to pass arguments to the invoked function. Use the args parameter to specify the arguments.

   import threading
   
   def worker(num):
       """thread worker function"""
       print(f'Worker: {num}')
   
   for i in range(5):
       t = threading.Thread(target=worker, args=(i,))
       t.start()

4. Q: Can we remove the comma and pass "(i)"?
5. A: (1) is only an integer, (1,) is a tuple.
6. If you pass the argument as a list, you don't need the comma, but the overhead of a list is higher than a tuple.
7. Exercise: Modify your ICMP exercise and scan the subnet 10.20.128.0/24. Report which among 10.20.128.1~10.20.128.254 are alive, which are unavailable. Totally this should be done in less than 12 seconds.
8. Exercise: Modify your ICMP exercise so that the output is sorted by IP addresses. The totally time should still be less than 12 seconds.

Thread Names & Logging

1. You may assign a name to a thread when it is created. For those threads without assigned names, the default is starting from "Thread-1", "Thread-2", etc.
2. The name of the main thread is "MainThread".

   import threading
   import time
   def worker():
       print(threading.current_thread().name, 'Starting')
       time.sleep(2)
       print(threading.current_thread().name, 'Exiting')
   
   def manager():
       print(threading.current_thread().name, 'Starting')
       time.sleep(3)
       print(threading.current_thread().name, 'Exiting')
   
   m = threading.Thread(name='Manager', target=manager)
   w = threading.Thread(name='Worker', target=worker)
   w1 = threading.Thread(target=worker)  # use default name
   
   w.start()
   w1.start()
   m.start()
   print(threading.current_thread().name, 'Exiting')

3. As we learned two weeks ago, most programs do not use print to debug.

   import logging, threading, time
   
   def worker():
       logging.info('Starting')
       time.sleep(2)
       logging.debug('Exiting')
   
   def manager():
       logging.info('Starting')
       time.sleep(3)
       logging.debug('Exiting')
   
   logging.basicConfig(
       level=logging.DEBUG,
       format='[%(levelname)s] (%(threadName)-10s) %(message)s',)
   
   m = threading.Thread(name='Manager', target=manager)
   w = threading.Thread(name='Worker', target=worker)
   w1 = threading.Thread(target=worker)  # use default name
   
   w.start()
   w1.start()
   m.start()
   logging.debug('Exiting')

4. Note that you only need to specify the message "Exiting". The levelname and threadName are automatically shown.
5. Logging levels:
   • CIRTICAL 50
   • ERROR 40
   • WARNING 30
   • INFO 20
   • DEBUG 10
   • NOTSET 0
6. By default, the main thread will wait for all working threads to finish before exiting the program.

Daemon Threads

1. More precisely, the main thread will exit after all non-daemon threads have completed.
2. The default is to create threads as non-daemons.

   import threading, time, logging
   
   def daemon():
       logging.info('Starting')
       time.sleep(2)
       logging.debug('Exiting')
   
   def non_daemon():
       logging.info('Starting')
       logging.debug('Exiting')
   
   logging.basicConfig(
       level=logging.DEBUG,
       format='[%(levelname)s] (%(threadName)-10s) %(message)s', )
   
   d = threading.Thread(name='daemon', target=daemon, daemon=True)
   n = threading.Thread(name='non-daemon', target=non_daemon)
   
   d.start()
   n.start() 

3. Change "daemon=True" to "daemon=False" to see the difference.

Synchronize

Barrier

1. When you have multiple threads running concurrently, sometimes you want to coordinate their execution.
   

   import threading, logging, random, time
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   def worker():
       i = random.randint(2, 10)
       logging.info(f"Waiting {i} seconds")
       time.sleep(i)
       # b.wait()
       logging.info("Exiting")
   
   b = threading.Barrier(5)
   for i in range(5):
       t = threading.Thread(target=worker)
       t.start()
   

2. Without synchronization, some threads may exit earlier; some threads exit later.
3. Now remove the remark, and you'll see all threads exit at the same time.
4. Barriers can be re-used.

   import threading, logging, random, time
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   def worker():
       i = random.randint(2, 10)
       logging.info(f"Waiting {i} seconds")
       time.sleep(i)
       b.wait()
       logging.info("First barrier passed")
       time.sleep(i)
       b.wait()
       logging.info("Exiting")
   
   b = threading.Barrier(5)
   for i in range(5):
       t = threading.Thread(target=worker)
       t.start()
   

5. The return value of barrier.wait() can be used to select a thread to do some special housekeeping.

   import threading, logging, time
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   def worker():
       n = b.wait()
       logging.info(f"Still {n} threads haven't passed the barrier.")
       if n == 0:
           time.sleep(0.1)
           logging.info("I am the last one.")
   
   b = threading.Barrier(5)
   for i in range(5):
       t = threading.Thread(target=worker)
       t.start()

6. enumerate() returns a list of active Thread instances. You don't need to maintain a list by yourself.

   import threading ,time ,logging
   
   def worker():
       time.sleep(5)
   
   logging.basicConfig(
       level=logging.DEBUG,
       format='(%(threadName)-10s) %(message)s',
   )
   
   for i in range(3):
       t = threading.Thread(target=worker)
       t.start()
   
   main_thread = threading.main_thread()
   logging.debug(main_thread.name)
   
   for t in threading.enumerate():
       if t is not main_thread:            # The list includes the current thread (main thread)
           logging.debug(t.name)

7. Exercise:Consider the following code.

   import threading ,time ,logging, random
   N = 3
   status = [0]*N
   
   def worker(num):
       i = random.randint(2, 9)
       logging.debug(f'Waiting {i} seconds.')
       time.sleep(i)
       status[num] = i
   
   logging.basicConfig(
       level=logging.DEBUG,
       format='(%(threadName)-10s) %(message)s',
   )
   
   for i in range(N):
       t = threading.Thread(target=worker, args=(i,) )
       t.start()
   
   for i in range(N):
       logging.debug(f'status[{i}] = {status[i]}')

8. Each worker is expected to generate a non-zero integer and store it in the global list status. However, the result shows that it is not running as we expected.
9. Is it because the array was not declared as a global variable or something else? Please try to find the bug and correct it.

Event

1. Python supports flexible synchronization mechanisms. In addition to barriers, which all threads in a group wait together, threads can also wait for the signal from another thread.
   

   import threading, time, logging, random
   
   def worker():
       n = random.randint(3, 8)
       logging.info(f"I need {n} seconds.")
       time.sleep(n)
       logging.info("I am ready")
       ready.set() # All threads waiting for it are awakened.
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   ready = threading.Event()
   logging.info("Tell me when you are ready.")
   t = threading.Thread(target=worker)
   t.start()
   ready.wait()
   logging.info("OK")
   

2. Q: If one thread await the event later than the event is set, what will happen?

   import threading, time, logging, random
   
   def worker(i):
       if i % 2 == 1:
           time.sleep(5)
       ready.wait()
       logging.info("I am ready")
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   ready = threading.Event()
   for i in range(4):
       t = threading.Thread(target=worker, name=f'Thread-{i}', args=(i,))
       t.start()
   ready.set()

3. When an event is set, it remains to be set, until you clear() it.

   import threading, time, logging, random
   
   def worker(i):
       if i % 2 == 1:
           time.sleep(5)
       ready.wait()
       logging.info("I am ready")
   
   logging.basicConfig(
       level=logging.INFO,
       format='(%(threadName)-10s) %(message)s',
   )
   
   ready = threading.Event()
   for i in range(4):
       t = threading.Thread(target=worker, args=(i,), daemon=True)
       t.start()
   ready.set()
   ready.clear()
   time.sleep(7)
   # ready.set()
   # time.sleep(1) # If you don't wait, the main thread will exit before daemon threads show responses.

Signaling Between Threads

1. Consider the following code.

   import logging, threading, time
   
   def server():
       print("I'll sleep 30 seconds.")
       INTERVAL=5
       for i in range(6):
           time.sleep(INTERVAL)
           print("Have slept %d seconds" % ((i+1)*INTERVAL))
       print("Ending server")
   
   def client():
       for i in range(3):
           n = eval(input("? "))
           if n == 0:
               break
       print("Ending client")
   
   c = threading.Thread(target=client, daemon=True)
   s = threading.Thread(target=server)
   c.start()
   s.start()

2. If there is no mechanism for the two threads to communicate with each other, the server has to wait for 30 seconds even if the client has gotten inputs from the user earlier.
3. Let's create an event so that the client can notify the server that its task is done.

   import logging, threading, time
   
   def server():
       print("I'll sleep 30 seconds.")
       INTERVAL=5
       for i in range(6):
           if flagToExit.is_set():         # Initial value of the flag is False.
               break
           time.sleep(INTERVAL)
           print("Have slept %d seconds" % ((i+1)*INTERVAL))
       print("Ending server")
   
   def client():
       for i in range(3):
           n = eval(input("? "))   # Input "0" to notify the server to terminate.
           if n == 0:
               break
       flagToExit.set()
       print("Ending client")
   
   flagToExit = threading.Event()
   c = threading.Thread(target=client, daemon=True)
   s = threading.Thread(target=server)
   c.start()
   s.start()

4. This is good, because the server can end earlier when it detects the client has finished its task. However, the "polling interval" of the server is 5 seconds, which implies that there will always be some delay (in average, 2.5s).

   import logging, threading, time
   
   def server():
       print("I'll sleep 30 seconds.")
       INTERVAL=5
       for i in range(6):
           if flagToExit.wait(INTERVAL): 
               break
           time.sleep(INTERVAL)
           print("Have slept %d seconds" % ((i+1)*INTERVAL))
       print("Ending server")
   
   def client():
       for i in range(3):
           n = eval(input("? "))   # Input "0" to notify the server to terminate.
           if n == 0:
               break
       flagToExit.set()
       print("Ending client")
   
   flagToExit = threading.Event()
   c = threading.Thread(target=client, daemon=True)
   s = threading.Thread(target=server)
   c.start()
   s.start()

Timer

1. threading.Timer(interval, function, args=None, kwargs=None)
2. Timer is a subclass of Thread.
   • After it is started, the specified function will be run after interval seconds.
   • Before the action is executed, the timer can be cancelled.

   from threading import Timer
   
   def hello():
       print("hello, world")
   
   t = Timer(10.0, hello)
   t.start()  # after 10 seconds, "hello, world" will be printed
   print("======")
   print("Now trying to type some sentences.")
   s = input()
   print(s)
   

3. Exercise: input_with_timeout.py

Race Condition

1. [YouTube] https://www.youtube.com/watch?v=7aF0q7NfwfA (4:37)
2. Consider the following code which is easy to cause Race Conditions:

   # ch07_thread_race1.py
   import threading
   import os.path
   
   FN = "a.txt"
   counter = 0     # Shared resource (counter)
   
   def worker():
       global counter
       for _ in range(1000):
           with open(FN, "r") as f:
               s = f.readline()
           if s == '':
               counter = 0
           else:
               counter = int(s)
           counter += 1
           with open(FN, 'w') as f:
               print(counter, file=f)
   
   # Reset FN
   with open(FN, 'w') as f:
       print(0, file=f)
   
   # Create and start two worker threads
   thread1 = threading.Thread(target=worker)
   thread2 = threading.Thread(target=worker)
   
   thread1.start()
   thread2.start()
   
   # Wait for both threads to finish
   thread1.join()
   thread2.join()
   
   print(f"Final counter value: {counter}")

3. If there is only 1 thread, no race condition will occur.
4. We need a mechanism so that we can enjoy the benefit of multi-threading, but prevent race conditions from occurring.
5. Lock - Only One Thread Can Run
   • threading.Lock
     • A lock is in either "locked" or "unlocked" state.
     • It is created in the unlock state.
     • Once a thread has acquired a lock, subsequent attempts to acquire it will block, until it is released.
   • acquire()
     • When a thread acquires a lock,
       • When the state is unlocked, acquire() changes the state to locked, and returns immediately.
       • When the state is locked, acquire() blocks, until the lock is released (by the thread which acquired it).
   • release()
     • The release() method should only be called in the locked state; it changes the state to unlocked and returns immediately.
     • If an attempt is made to release an unlocked lock, a RuntimeError will be raised.
       • When more than one thread is blocked in waiting for a lock, only one thread proceeds when a release() call resets the state to unlocked; which one of the waiting threads proceeds is not defined, and may vary across implementations.
6. Using Locks to Create a Critical Section

   # ch07_thread_race2.py
   import threading
   import os.path
   
   FN = "a.txt"
   counter = 0     # Shared resource (counter)
   lock = threading.Lock()
   
   def worker():
       global counter
       lock.acquire()
       for _ in range(1000):
           with open(FN, "r") as f:
               s = f.readline()
           if s == '':
               counter = 0
           else:
               counter = int(s)
           counter += 1
           with open(FN, 'w') as f:
               print(counter, file=f)
       lock.release()
   
   # Reset FN
   with open(FN, 'w') as f:
       print(0, file=f)
   
   # Create and start two worker threads
   thread1 = threading.Thread(target=worker)
   thread2 = threading.Thread(target=worker)
   
   thread1.start()
   thread2.start()
   
   # Wait for both threads to finish
   thread1.join()
   thread2.join()
   
   print(f"Final counter value: {counter}")

7. Lock vs. Event
   • Q: Locks and Events seem to be very similar. When should I use a lock and when should I use an event?
   • A:
     • When a lock is released, only a thread will be awaken.
     • When an event is set, all threads waiting for that event are waken.