Python threading and multithreading

Многозадачность

На уровне процессора нет таких понятий, как поток и процесс: всё, что мы называем многопоточностью, предоставляет операционная система. А , на самом деле, не может заставить процессор ждать. Эта команда переводит его в режим пониженного энергопотребления (lower energy state).

И один процессор делает одно и то же, за исключением Hyper-threading, который способен выполнять другие команды.

Для процессора существует только понятие задачи (). Вместо потоков есть сегмент состояния задачи, который позволяет сменить таск. Состояние процессора означает состояние доступа к регистрам и маппинга памяти.

Можно использовать на примере очень простой реализации .

Весь код (финальная версия):

Класс Node (внутри класса ) хранит значение и содержит ссылку на следующий элемент.

Давайте сначала посмотрим на неблокирующую реализацию стека:

Посмотрим на неблокирующую реализацию стека, здесь мы не используем ключевое слово lock и wait-операции:

Зачем здесь нужен условный оператор? Может случиться так, что два потока пытаются запушить новое значение в стек. Эти два потока видят одно и то же поле . Первый поток начал вставлять новое значение до того, как это начал делать второй поток. После того как первый поток вставил значение, актуальная ссылка на есть только у этого потока. У второго потока будет ссылка на элемент, который теперь считается следующим после , то есть

Такая ситуация показывает, насколько важно бывает иметь такие атомарные операции, как

Этот способ решения задачи основан на неблокирующей структуре данных. В методе мы используем тот же прием:

Берем head и заменяем её следующим узлом, если, конечно, она уже не была изменена.

В время теста участвовало два ядра. Одно ядро выполняло операцию Push(), другое — операцию Pop(), ожидание отсутствовало в течение 6 миллионов операций по обмену сообщениями между двумя ядрами.

У этой структуры данных есть два недостатка:

1. необходимость очистки
2. необходимость выделения памяти для элементов коллекции

Другая структура данных лишена этих недостатков. Эта структура данных называется «кольцевой буфер». В кольцевом буфере вам не нужно каждый раз выделять память. Есть один большой участок памяти, который перезаписывается при необходимости.

В результате все работает гораздо быстрее: 9 миллионов транзакций в секунду.

Project details

Meta

License: Apache Software License (Apache License, Version 2.0)

Author: Alexey Stepanov

Maintainer: Alexey Stepanov <penguinolog@gmail.com>, Antonio Esposito <esposito.cloud@gmail.com>, Dennis Dmitriev <dis-xcom@gmail.com>

Requires: Python >=3.6.0

Classifiers

• Development Status

  5 — Production/Stable

• Intended Audience

  Developers

• License

  OSI Approved :: Apache Software License

• Programming Language

  • Python :: 3

  • Python :: 3.6

  • Python :: 3.7

  • Python :: 3.8

  • Python :: 3.9

  • Python :: Implementation :: CPython

  • Python :: Implementation :: PyPy

• Topic

  Software Development :: Libraries :: Python Modules

What is a Thread?

A thread is a separate flow of execution. Your program will be doing two things at once (not at the exact same time).

Because many processors these days have multiple cores, you can run a different process on each one. While they may be running on different processors, they are not running at the same time.

To use threading in Python, import the threading module. You can create a thread with this code:

my_thread = threading.Thread(target=thread_function, args = (1,))

The thread will be the execution of a function named thread_function() (it can have any name). This can be a simple function like:

def thread_function(name):    logging.info("Thread %s: started", name)    time.sleep(2)    logging.info("Thread %s: ended", name)

If you want a thread to continue for a long time, you can set a time limit or put it inside a loop.

Threads don’t start themself, the above code only created the thread but didn’t start it. You can start a thread by calling the .start() method.

my_thread.start()

In the Python code below, a thrread is started from the main program. The module logging is used to output some data.

import loggingimport threadingimport timedef thread_function(name):    logging.info("Thread %s: started", name)    time.sleep(2)    logging.info("Thread %s: ended", name)if __name__ == "__main__":    format = "%(asctime)s: %(message)s"    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")    logging.info("Main, before thread created")    my_thread = threading.Thread(target=thread_function, args=(1,))    logging.info("Main, before thread started")    my_thread.start()    logging.info("Main, wait for the thread to finish")    logging.info("Ending thread and program")

If you run the program, you’ll see it starts both the main thread and the thread my_thread, and then waits for both to exit.

➜  ~ python3 example.py15:18:27: Main, before thread created15:18:27: Main, before thread started15:18:27: Thread 1: started15:18:27: Main, wait for the thread to finish15:18:27: Ending thread and program15:18:29: Thread 1: ended

17.1.4. RLock Objects¶

A reentrant lock is a synchronization primitive that may be acquired multiple
times by the same thread. Internally, it uses the concepts of “owning thread”
and “recursion level” in addition to the locked/unlocked state used by primitive
locks. In the locked state, some thread owns the lock; in the unlocked state,
no thread owns it.

To lock the lock, a thread calls its method; this
returns once the thread owns the lock. To unlock the lock, a thread calls
its method. /
call pairs may be nested; only the final (the
of the outermost pair) resets the lock to unlocked and
allows another thread blocked in to proceed.

Reentrant locks also support the .

Timer Threads¶

One example of a reason to subclass Thread is provided by
Timer, also included in . A Timer
starts its work after a delay, and can be canceled at any point within
that delay time period.

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-10s) %(message)s',
                    )

def delayed():
    logging.debug('worker running')
    return

t1 = threading.Timer(3, delayed)
t1.setName('t1')
t2 = threading.Timer(3, delayed)
t2.setName('t2')

logging.debug('starting timers')
t1.start()
t2.start()

logging.debug('waiting before canceling %s', t2.getName())
time.sleep(2)
logging.debug('canceling %s', t2.getName())
t2.cancel()
logging.debug('done')

Notice that the second timer is never run, and the first timer appears
to run after the rest of the main program is done. Since it is not a
daemon thread, it is joined implicitly when the main thread is done.

Структуры

17.1.3. Lock Objects¶

A primitive lock is a synchronization primitive that is not owned by a
particular thread when locked. In Python, it is currently the lowest level
synchronization primitive available, implemented directly by the
extension module.

A primitive lock is in one of two states, “locked” or “unlocked”. It is created
in the unlocked state. It has two basic methods, and
. When the state is unlocked,
changes the state to locked and returns immediately. When the state is locked,
blocks until a call to in another
thread changes it to unlocked, then the call resets it
to locked and returns. The 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
will be raised.

Locks also support the .

When more than one thread is blocked in waiting for the
state to turn to unlocked, only one thread proceeds when a
call resets the state to unlocked; which one of the waiting threads proceeds
is not defined, and may vary across implementations.

All methods are executed atomically.

Starting a New Thread

To spawn another thread, you need to call the following method available in the thread module −

_thread.start_new_thread ( function, args )

This method call enables a fast and efficient way to create new threads in both Linux and Windows.

The method call returns immediately and the child thread starts and calls function with the passed list of args. When the function returns, the thread terminates.

Here, args is a tuple of arguments; use an empty tuple to call function without passing any arguments. kwargs is an optional dictionary of keyword arguments.

Example

#!/usr/bin/python3

import _thread
import time

# Define a function for the thread
def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print ("%s: %s" % ( threadName, time.ctime(time.time()) ))

# Create two threads as follows
try:
   _thread.start_new_thread( print_time, ("Thread-1", 2, ) )
   _thread.start_new_thread( print_time, ("Thread-2", 4, ) )
except:
   print ("Error: unable to start thread")

while 1:
   pass

Output

When the above code is executed, it produces the following result −

Thread-1: Fri Feb 19 09:41:39 2016
Thread-2: Fri Feb 19 09:41:41 2016
Thread-1: Fri Feb 19 09:41:41 2016
Thread-1: Fri Feb 19 09:41:43 2016
Thread-2: Fri Feb 19 09:41:45 2016
Thread-1: Fri Feb 19 09:41:45 2016
Thread-1: Fri Feb 19 09:41:47 2016
Thread-2: Fri Feb 19 09:41:49 2016
Thread-2: Fri Feb 19 09:41:53 2016

Program goes in an infinite loop. You will have to press ctrl-c to stop

Although it is very effective for low-level threading, the thread module is very limited compared to the newer threading module.

Subclassing Thread

It is also possible to start a thread by subclassing threading.Thread. Depending on the design of your application, you may prefer this approach. Here, you extend threading.Thread and provide the implementation of your task in the run() method.

import threading
import random, time

class MyTask(threading.Thread):
    def __init__(self, sleepFor):
        self.secs = sleepFor
        threading.Thread.__init__(self)

    def run(self):
        print self, 'begin sleep(', self.secs, ')'
        time.sleep(self.secs)
        print self, 'end sleep(', self.secs, ')'

And here is the usage of the class defined above.

tasks = []
for x in xrange(0, 5):
    t = MyTask(random.randint(1, 10))
    tasks.append(t)
    t.start()

print 'joining ..'
while threading.active_count() > 1:
    for t in tasks:
        t.join()
        print t, 'is done.'
print 'all done.'

Semaphore Objects¶

This is one of the oldest synchronization primitives in the history of computer
science, invented by the early Dutch computer scientist Edsger W. Dijkstra (he
used the names and instead of and
).

A semaphore manages an internal counter which is decremented by each
call and incremented by each
call. The counter can never go below zero; when
finds that it is zero, it blocks, waiting until some other thread calls
.

Semaphores also support the .

class (value=1)

This class implements semaphore objects. A semaphore manages an atomic
counter representing the number of calls minus the number of
calls, plus an initial value. The method
blocks if necessary until it can return without making the counter negative.
If not given, value defaults to 1.

The optional argument gives the initial value for the internal counter; it
defaults to . If the value given is less than 0, is
raised.

Changed in version 3.3: changed from a factory function to a class.

(blocking=True, timeout=None)

Acquire a semaphore.

When invoked without arguments:

• If the internal counter is larger than zero on entry, decrement it by
  one and return immediately.

• If the internal counter is zero on entry, block until awoken by a call to
  . Once awoken (and the counter is greater
  than 0), decrement the counter by 1 and return . Exactly one
  thread will be awoken by each call to . The
  order in which threads are awoken should not be relied on.

When invoked with blocking set to false, do not block. If a call
without an argument would block, return immediately; otherwise, do
the same thing as when called without arguments, and return .

When invoked with a timeout other than , it will block for at
most timeout seconds. If acquire does not complete successfully in
that interval, return . Return otherwise.

Changed in version 3.2: The timeout parameter is new.

(n=1)

Release a semaphore, incrementing the internal counter by n. When it
was zero on entry and other threads are waiting for it to become larger
than zero again, wake up n of those threads.

Changed in version 3.9: Added the n parameter to release multiple waiting threads at once.

class (value=1)

Class implementing bounded semaphore objects. A bounded semaphore checks to
make sure its current value doesn’t exceed its initial value. If it does,
is raised. In most situations semaphores are used to guard
resources with limited capacity. If the semaphore is released too many times
it’s a sign of a bug. If not given, value defaults to 1.

Changed in version 3.3: changed from a factory function to a class.

Поля

Структура SpinWait

Представьте, что у вас есть у вас есть два процессора, и на нулевом вы выполняете присваивание A = 1, потом устанавливаете барьер памяти, а затем снова выполняете присваивание B = 1.

На первом процессоре вы получаете А, равное 1, а затем вы ждете, пока B не присвоят значение 1. Казалось бы, такие операции должны быстро выполниться и в кэше L1, и даже в кэше L3. Здесь загвоздка в том, что выполнение нулевым процессором может прерваться из-за вытесняющего прерывания, и между операциями из строк 1 и 3 может пройти несколько миллисекунд. способен решать такие проблемы.

Под капотом работает :

1. Сначала происходит вызов (не происходит переключение контекста потока). Этот шаг будет пропущен для одноядерных процессоров:
2. Затем будет вызван (может быть прерван потоками такого же приоритета)
3. Затем, — (может быть прерван потоками любого приоритета)
4. Если операция все еще не завершена, то будет вызван (поток засыпает на 1мс)

Без цикл мог бы выполняться бесконечно, потому что если нулевой процессор запустится как фоновый поток, то первый процессор заблокирует этот фоновый поток, до тех пор пока поток на первом процессоре не будет прерван.

Ограничение одновременного доступа к ресурсам

Как разрешить доступ к ресурсу нескольким worker одновременно, но при этом ограничить их количество. Например, пул соединений может поддерживать фиксированное число одновременных подключений, или сетевое приложение может поддерживать фиксированное количество одновременных загрузок. Semaphore является одним из способов управления соединениями.

import logging
import random
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s (%(threadName)-2s) %(message)s',
                    )

class ActivePool(object):
    def __init__(self):
        super(ActivePool, self).__init__()
        self.active = []
        self.lock = threading.Lock()
    def makeActive(self, name):
        with self.lock:
            self.active.append(name)
            logging.debug('Running: %s', self.active)
    def makeInactive(self, name):
        with self.lock:
            self.active.remove(name)
            logging.debug('Running: %s', self.active)

def worker(s, pool):
    logging.debug('Waiting to join the pool')
    with s:
        name = threading.currentThread().getName()
        pool.makeActive(name)
        time.sleep(0.1)
        pool.makeInactive(name)

pool = ActivePool()
s = threading.Semaphore(2)
for i in range(4):
    t = threading.Thread(target=worker, name=str(i), args=(s, pool))
    t.start()

В этом примере класс ActivePool является удобным способом отслеживания того, какие потоки могут запускаться в данный момент. Реальный пул ресурсов будет выделять соединение для нового потока и восстанавливать значение, когда поток завершен. В данном случае он используется для хранения имен активных потоков, чтобы показать, что только пять из них работают одновременно.

$ python threading_semaphore.py

2013-02-21 06:37:53,629 (0 ) Waiting to join the pool
2013-02-21 06:37:53,629 (1 ) Waiting to join the pool
2013-02-21 06:37:53,629 (0 ) Running: 
2013-02-21 06:37:53,629 (2 ) Waiting to join the pool
2013-02-21 06:37:53,630 (3 ) Waiting to join the pool
2013-02-21 06:37:53,630 (1 ) Running: 
2013-02-21 06:37:53,730 (0 ) Running: 
2013-02-21 06:37:53,731 (2 ) Running: 
2013-02-21 06:37:53,731 (1 ) Running: 
2013-02-21 06:37:53,732 (3 ) Running: 
2013-02-21 06:37:53,831 (2 ) Running: 
2013-02-21 06:37:53,833 (3 ) Running: []

Делегаты

Performance¶

This section discusses the performance of the provided concrete I/O
implementations.

Binary I/O

By reading and writing only large chunks of data even when the user asks for a
single byte, buffered I/O hides any inefficiency in calling and executing the
operating system’s unbuffered I/O routines. The gain depends on the OS and the
kind of I/O which is performed. For example, on some modern OSes such as Linux,
unbuffered disk I/O can be as fast as buffered I/O. The bottom line, however,
is that buffered I/O offers predictable performance regardless of the platform
and the backing device. Therefore, it is almost always preferable to use
buffered I/O rather than unbuffered I/O for binary data.

Text I/O

Text I/O over a binary storage (such as a file) is significantly slower than
binary I/O over the same storage, because it requires conversions between
unicode and binary data using a character codec. This can become noticeable
handling huge amounts of text data like large log files. Also,
and are both quite slow
due to the reconstruction algorithm used.

, however, is a native in-memory unicode container and will
exhibit similar speed to .

Multi-threading

objects are thread-safe to the extent that the operating system
calls (such as under Unix) they wrap are thread-safe too.

Binary buffered objects (instances of ,
, and )
protect their internal structures using a lock; it is therefore safe to call
them from multiple threads at once.

objects are not thread-safe.

Reentrancy

Binary buffered objects (instances of ,
, and )
are not reentrant. While reentrant calls will not happen in normal situations,
they can arise from doing I/O in a handler. If a thread tries to
re-enter a buffered object which it is already accessing, a
is raised. Note this doesn’t prohibit a different thread from entering the
buffered object.

The above implicitly extends to text files, since the function
will wrap a buffered object inside a . This includes
standard streams and therefore affects the built-in function as
well.

Creating Thread Using Threading Module

To implement a new thread using the threading module, you have to do the following −

• Define a new subclass of the Thread class.

• Override the __init__(self ) method to add additional arguments.

• Then, override the run(self ) method to implement what the thread should do when started.

Once you have created the new Thread subclass, you can create an instance of it and then start a new thread by invoking the start(), which in turn calls the run() method.

Example

#!/usr/bin/python3

import threading
import time

exitFlag = 0

class myThread (threading.Thread):
   def __init__(self, threadID, name, counter):
      threading.Thread.__init__(self)
      self.threadID = threadID
      self.name = name
      self.counter = counter
   def run(self):
      print ("Starting " + self.name)
      print_time(self.name, self.counter, 5)
      print ("Exiting " + self.name)

def print_time(threadName, delay, counter):
   while counter:
      if exitFlag:
         threadName.exit()
      time.sleep(delay)
      print ("%s: %s" % (threadName, time.ctime(time.time())))
      counter -= 1

# Create new threads
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# Start new Threads
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print ("Exiting Main Thread")

Result

When we run the above program, it produces the following result −

Starting Thread-1
Starting Thread-2
Thread-1: Fri Feb 19 10:00:21 2016
Thread-2: Fri Feb 19 10:00:22 2016
Thread-1: Fri Feb 19 10:00:22 2016
Thread-1: Fri Feb 19 10:00:23 2016
Thread-2: Fri Feb 19 10:00:24 2016
Thread-1: Fri Feb 19 10:00:24 2016
Thread-1: Fri Feb 19 10:00:25 2016
Exiting Thread-1
Thread-2: Fri Feb 19 10:00:26 2016
Thread-2: Fri Feb 19 10:00:28 2016
Thread-2: Fri Feb 19 10:00:30 2016
Exiting Thread-2
Exiting Main Thread

Threads using queue

• In this example, I have imported modules called queue and threading. The function employee is used as a def employee().
• Infinite loop(while True) is called to make threads ready to accept all the tasks.
• Then define queue as project = q.get() .
• task_done() tells the queue that the processing on task is completed. When the project is put in the queue task_done is called.
• threading.Thread(target=employee, daemon=True).start() is used to start the employee thread.
• for a time in range(5)means 5 tasks are sent to the employee.
• q.join blocks till all the tasks are completed.

Example:

You can refer the below screenshot to see output of all the 5 tasks.

Threads using queue

Метод after() — Погружение в сон для Tkinter

tkinter является частью стандартной библиотеки Python. В случае, если вы используете заранее установленную версию Python на Linux или Mac, он может быть вам недоступен. При получении ошибки стоит самостоятельно добавить его в систему. В том случае, если вы ранее установили Python сами, должен быть доступен.

Начнем с разбора примера, где используется . Запустите следующий код и посмотрите, что произойдет при неправильном добавлении вызова в Python:

Python

import tkinter
import time

class MyApp:
def __init__(self, parent):
self.root = parent
self.root.geometry(«400×400″)
self.frame = tkinter.Frame(parent)
self.frame.pack()
b = tkinter.Button(text=»click me», command=self.delayed)
b.pack()

def delayed(self):
time.sleep(3)

if __name__ == «__main__»:
root = tkinter.Tk()
app = MyApp(root)
root.mainloop()

После запуска кода нажмите кнопку в GUI. Кнопка не будет реагировать три секунды, ожидая завершения . Если в приложении есть другие кнопки, на них тоже нельзя будет нажать. Закрыть приложение во время сна нельзя, так как оно не будет откликаться на событие закрытия.

Для должного погружения в сон потребуется использовать :

Python

import tkinter

class MyApp:
def __init__(self, parent):
self.root = parent
self.root.geometry(«400×400»)
self.frame = tkinter.Frame(parent)
self.frame.pack()
self.root.after(3000, self.delayed)

def delayed(self):
print(‘Я задержался’)

if __name__ == «__main__»:
root = tkinter.Tk()
app = MyApp(root)
root.mainloop()

Здесь создается приложение, высота которого 400 пикселей, и ширина также 400 пикселей. На нем нет виджетов. Оно только показывает фрейм. Затем вызывается , где является отсылкой к объекту . принимает два аргумента:

1. Количество миллисекунд для сна;
2. Метод который вызовется после завершения сна.

В данном случае приложение выведет строку в стандартный поток вывода (stdout) через 3 секунды. Можно рассматривать как Tkinter-версию того же , только он добавляет способность вызова функции после завершения сна.

Данную функциональность можно использовать для улучшения работы пользователя. Добавив в Python вызов , можно ускорить процесс загрузки приложения, после чего начать какой-то длительный процесс. В таком случае пользователю не придется ждать открытия приложения.