Core Threading Concepts

Threading represents the execution flow within a computational unit, serving as an abstract concept that defines the fundamental execution entity of a CPU. Understanding the distinction between processes and threads is crucial:

• Processes functon as resource containers, holding all necessary runtime resources, while threads act as execution pipelines responsible for code execution
• Every process contains at least one thread created automatically by the operating system, known as the main thread
• Processes can accommodate any number of additional threads
• Thread creation requires significantly fewer resources compared to process creation
• Threads within the same process share data space
• Threads maintain equal status without hierarchical relationships; threads in the same process share identical identifiers
• Thread creation code can appear anywhere in the application, not exclusively within main functions

The primary motivation for implementing threads lies in enhancing program execution efficiency.

Two Approaches to Thread Creation

Similar to process creation but more flexible regarding placement within the codebase:

# Method one: Direct Thread instantiation
from threading import Thread

def worker_function():
    print("background thread executing....")

t_instance = Thread(target=worker_function)
t_instance.start()
print('main thread completed....')

# Method two: Class inheritance approach
class CustomThread(Thread):
    def run(self):
        print("background thread executing....")

Main and Child Thread Relationships

Key behavioral characteristics:

• After main thread tasks complete, it waits for all child threads to finish before terminating
• All threads within the same process maintain equal status without parent-child hierarchies

# Demonstrating main thread waiting behavior
import random
import time
import threading
from threading import Thread
def worker_task(name):
    print(f"{name} is executing...")
    time.sleep(random.randint(1, 3))
    print(threading.enumerate())
    print(f"{name} has finished.....")

thread_instance = Thread(target=worker_task, args=('worker_a',))
thread_instance.start()

print('main completed....')

Verifying Thread vs Process Distinctions

from threading import Thread
import time

def worker_operation():
    global counter
    time.sleep(1)
    counter -= 1
counter = 10
thread_obj = Thread(target=worker_operation,)
thread_obj.start()
thread_obj.join()
print(counter)

Data sharing occurs within the same process among threads.

from multiprocessing import Process
from threading import Thread
import time
def worker_operation():
    pass
def measure_performance(cls):
    """Performance measurement for thread or process creation"""
    instances = []
    start_time = time.time()
    for i in range(50):
        p = cls(target=worker_operation,)
        p.start()
        instances.append(p)
    for p in instances:
        p.join()
    return time.time() - start_time

Thread creation demands substantially fewer resources than process creation.

Thread Safety Challenges

Mutual Exclusion Locks

Shared data inevitably creates competition scenarios, leading to data corruption issues.

Solution: Implement mutual exclusion locks similar to process synchronization.

from threading import Thread, Lock
import time

shared_counter = 10
def worker_with_lock(mutex):
    global shared_counter
    mutex.acquire()
    temp_value = shared_counter
    time.sleep(0.5)
    shared_counter = temp_value - 1
    mutex.release()

threads_list = []
main_lock = Lock()
for i in range(10):
    thread = Thread(target=worker_with_lock, args=(main_lock,))
    thread.start()
    threads_list.append(thread)
for thread in threads_list:
    thread.join()
print(shared_counter)

Mutual exclusion ensures data integrity.

Deadlock Scenarios

Deadlock represents a locking state rather then a specific lock type.

Common deadlock situations:

1. Multiple acquisitions on the same lock (resolved using RLock instead of Lock)
2. Two or more processes/threads competing for resources, creating circular wait conditions (solution: minimize lock usage or use single locks)

from threading import Thread, Lock
import time
def operation_one(name, lock_x, lock_y):
    lock_x.acquire()
    print(f"{name} acquired lock X")
    time.sleep(0.3)
    lock_y.acquire()
    print(f'{name} acquired lock Y')
    lock_y.release()
    lock_x.release()
def operation_two(name, lock_x, lock_y):
    lock_y.acquire()
    print(f"{name} acquired lock Y")
    time.sleep(0.3)
    lock_x.acquire()
    print(f'{name} acquired lock X')
    lock_x.release()
    lock_y.release()

lock_x = Lock()
lock_y = Lock()
thread_1 = Thread(target=operation_one, args=('thread_1', lock_x, lock_y))
thread_2 = Thread(target=operation_two, args=('thread_2', lock_x, lock_y))
thread_1.start()
thread_2.start()

Example demonstrating the second deadlock scenario.

Reentrent Locks

Address multiple acquire calls within the same thread.

Other threads can access the lock only after all acquire calls from the current thread are released.

Can still result in deadlock conditions.

from threading import Thread, RLock

reentrant_lock = RLock()
reentrant_lock.acquire()
reentrant_lock.acquire()
reentrant_lock.acquire()
reentrant_lock.acquire()

print("completed")
reentrant_lock = RLock()

def first_worker():
    reentrant_lock.acquire()
    print('first worker')
def second_worker():
    reentrant_lock.acquire()
    print('second worker')

Thread(target=first_worker).start()
Thread(target=second_worker).start()

Semaphores

Another locking mechanism controlling concurrent access by limiting the number of threads active simultaneously, not addressing thread safety issues.

from threading import Semaphore, Thread
import time
semaphore_lock = Semaphore(3)

def worker_function():
    semaphore_lock.acquire()
    time.sleep(1)
    print("executing.....")
    semaphore_lock.release()

for i in range(20):
    thread = Thread(target=worker_function)
    thread.start()

Daemon Threads

Daemon threads terminate when all non-daemon threads complete.

import threading
from threading import Thread
import time
def first_worker():
    print('worker-1 executing...')
    time.sleep(3)
    print('worker-1 finished....')
def second_worker():
    print('worker-2 executing...')
    time.sleep(1)
    print('worker-2 finished....')

if __name__ == '__main__':
    worker_1 = Thread(target=first_worker,)
    worker_2 = Thread(target=second_worker,)
    worker_1.setDaemon(True)
    worker_1.start()
    worker_2.start()
    print(worker_1.ident)
    print(threading.enumerate())
    print("main completed...")

Global Interpreter Lock (GIL)

The Global Interpreter Lock functions as a mutex existing only in CPython implementations.

Necessity: Since each python.exe process contains only one interpreter instance, multiple threads competing for execution access could create conflicts.

Benefits: Ensures data safety when multiple threads access the interpreter simultaneously.

Drawbacks: Only one thread can access the interpreter at any given time, preventing true concurrency in multithreading.

Origin: Initially implemented as a simple solution since single-threaded processes didn't present issues, but garbage collection threads introduced complexity.

GIL locking/unlocking timing:

• Lock: Acquired immediately when calling the interpreter
• Unlock: Released when the current thread encounters I/O operations or exceeds execution time limits

Solutions: Utilize multiprocessing or alternative Python interpreters.

Thread Pools and Process Pools

Containers essentially representing lists storing threads or processes.

Rationale: Servers cannot indefinitely create threads or processes, necessitating control over their quantities. Thread pools handle creation, destruction, and task distribution.

Characteristics:

• Thread pools don't activate threads during creation
• When submitting tasks without available threads and existing count remains below maximum, new threads start
• Once activated, threads remain active until the entire process terminates

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
# pool = ProcessPoolExecutor(maxsize) creates process pool with maxsize maximum processes
# res = pool.submit(task, 'a') submits tasks
# res.result(timeout) receives return values with timeout parameter, blocking until completion
# pool.shutdown(wait) blocks until all tasks complete
# wait=True waits for all tasks before releasing resources
# wait=False returns immediately without waiting

from concurrent.futures import ThreadPoolExecutor
import time
def worker_task(number):
    time.sleep(0.5)
    print(f"{number} is executing.....")
    return number ** 2

pool = ThreadPoolExecutor()
results = []
for i in range(10):
    result = pool.submit(worker_task, i)
    results.append(result)
pool.shutdown(wait=False)
for result in results:
     print(result.result())
print('completed')

Synchronous vs Asynchronous, Blocking vs Non-blocking

Blocking and non-blocking describe program execution states

Blocking: Program execution halts when encountering I/O operations, unable to continue

Non-blocking: Program execution continues without interruption from I/O operations through various mechanisms

Synchronous and asynchronous refer to task submission methods

Synchronous: After initiating tasks, execution must wait for completion before proceeding

Asynchronous: Task initiation doesn't require waiting; other operations can execute immediately

Asynchronous operations offer higher efficiency than synchronous ones. Asynchronous task implementation includes multithreading and multiprocessing.

Difference between synchronous and blocking: Blocking implies CPU context switching, while synchronous operations may wait without CPU migration, continuing execution within the current process.

Asynchronous Callbacks

Essentially callback functions bound to asynchronous tasks, automatically invoked upon task completion.

Advantages: Eliminates waiting time while providing immediate access to results.

Thread pools and process pools use add_done_callback() method for callbacks, requiring exactly one parameter (the future object itself).

Thread pool callbacks execute within child threads.

Process pool callbacks execute within the main process.

import requests
from concurrent.futures import ThreadPoolExecutor
import threading
def fetch_data(url):
    response = requests.get(url)
    return response.text, url
def process_result(future):
    result = future.result()
    print(len(result[0]), result[1], "current thread", threading.current_thread())
if __name__ == '__main__':
    urls = ['http://www.baidu.com','https://www.cnblogs.com/ywsun/', 'https://www.processon.com/']
    pool = ThreadPoolExecutor()
    for url in urls:
        future = pool.submit(fetch_data, url)
        future.add_done_callback(process_result)

Thread pool asynchronous invocation.

import requests
from concurrent.futures import ProcessPoolExecutor
import os
def fetch_data(url):
    response = requests.get(url)
    return response.text, url
def process_result(future):
    result = future.result()
    print(len(result[0]), result[1], ", callback pid", os.getpid())
if __name__ == '__main__':
    urls = ['http://www.baidu.com','https://www.cnblogs.com/ywsun/', 'https://www.processon.com/']
    pool = ProcessPoolExecutor()
    print('main process', os.getpid())
    for url in urls:
        future = pool.submit(fetch_data, url)
        future.add_done_callback(process_result)

Process pool asynchronous invocation.

Thread Queues

The queue module provides common data containers, purely as containers without shared data characteristics.

Queue: First In, First Out

LifoQueue: Last In, First Out

PriorityQueue: Priority-based queues accepting tuples where the first element determines priority; lower numeric values indicate higher priority

import queue
priority_queue = queue.PriorityQueue()
# Insert tuple with first element as priority (typically numbers, also applicable for character comparison)
# Lower numbers represent higher priority
priority_queue.put((2,'a'))
priority_queue.put((1,'b'))
priority_queue.put((3,'c'))
print(priority_queue.get())
print(priority_queue.get())
print(priority_queue.get())
# For characters, sorting follows ASCII table
priority_queue.put(('a', "sfsja"))
priority_queue.put(('b', "sdfsdf"))
priority_queue.put(('A', "sdfsdf"))
priority_queue.put(('ae', "sdfsdf"))
priority_queue.put(('ab', "sdfsdf"))
print(priority_queue.get())
print(priority_queue.get())
print(priority_queue.get())
print(priority_queue.get())
print(priority_queue.get())

PriorityQueue example.

Events

Facilitate thread communication; threads inherently share data, making events optional for basic functionality.

Threads execute independently; sometimes monitoring another thread's status becomes necessary for coordinated actions.

Methods:

• event.isSet(): Returns current event state
• event.wait(): Blocks if event.isSet() returns False
• event.set(): Sets event state to True, activating all blocked threads to ready state for OS scheduling
• event.clear(): Resets event state to False

from threading import Thread, Event
import time
import random
startup_event = Event()
def server_process():
    print('starting server......')
    time.sleep(random.randint(1,3))
    print('server running.......')
    startup_event.set()
def connection_process():
    print('attempting connection')
    startup_event.wait()
    print('connection successful')
t1 = Thread(target=server_process)
t1.start()
t2 = Thread(target=connection_process)
t2.start()