Решаем проблемы с raii у std

this_thread Namespace

this_thread namespace from thread header offers possibilities to work with current thread. This namespace contains four useful functions:

1. id_get_id() – returns the id of current thread.

2. template void sleep_until (const chrono::time_point<Clock,Duration>& abs_time) – blocks current thread until abs_time is not reached.

3. template void sleep_for (const chrono::duration<Rep,Period>& rel_time); – thread is blocked during time span specified by rel_time.

4. void yield() – current thread allows implementation to reschedule the execution of thread. It used to avoid blocking.

This is an example of using these functions:

#include <iostream>
#include <iomanip> 
#include <thread> 
#include <chrono>
#include <ctime>

using namespace std;
using std::chrono::system_clock;
int main()
{
	cout << "The id of current thread is " << this_thread::get_id << endl;

	//sleep while next minute is not reached

	//get current time
	time_t timet = system_clock::to_time_t(system_clock::now());
	//convert it to tm struct
	struct tm * time = localtime(&timet);
	cout << "Current time: " << put_time(time, "%X") << '\n';
	std::cout << "Waiting for the next minute to begin...\n";
	time->tm_min++; time->tm_sec = 0;
	//sleep until next minute is not reached
	this_thread::sleep_until(system_clock::from_time_t(mktime(time)));
	cout << std::put_time(time, "%X") << " reached!\n";
	//sleep for 5 seconds
	this_thread::sleep_for(chrono::seconds(5));
	//get current time
	timet = system_clock::to_time_t(system_clock::now());
	//convert it to tm struct
	time = std::localtime(&timet);
	cout << "Current time: " << put_time(time, "%X") << '\n';
}

The id of current thread is 009717C6
Current time: 15:28:35
Waiting for the next minute to begin…
15:29:00 reached!
Current time: 15:29:05

Типы реализации потоков

1. Отсутствие прерывания по таймеру внутри одного процесса
2. При использовании блокирующего системного запроса для процесса все его потоки блокируются.
3. Сложность реализации

• Поток в пространстве ядра. Наряду с таблицей процессов в пространстве ядра имеется таблица потоков.
• «Волокна» (англ. fibers). Несколько потоков режима пользователя, исполняющихся в одном потоке режима ядра. Поток пространства ядра потребляет заметные ресурсы, в первую очередь физическую память и диапазон адресов режима ядра для стека режима ядра. Поэтому было введено понятие «волокна» — облегчённого потока, выполняемого исключительно в режиме пользователя. У каждого потока может быть несколько «волокон».

Как работает процессор

За единицу времени (она называется тик) процессор может выполнить только одну задачу — например, прочитать значение ячейки памяти. Поэтому на то, чтобы выполнить следующие операции, потребуется 4 тика:

1. Прочитать данные из двух ячеек.
2. Сложить их.
3. Записать результат в другую ячейку.

Это касается только атомарных операций, то есть самых маленьких и неделимых. Более сложные задачи могут состоять из нескольких атомарных. Например, чтобы провести умножение числа A на число B, нужно будет прибавить к числу A само себя B — 1 раз.

5 * 5 = 5 + 5 + 5 + 5 + 5

Ещё больше тиков потребуется на деление, а если операцию нужно провести над числами с плавающей запятой, то даже представить страшно, сколько их нужно.

Прямо сейчас у вас могут быть открыты десяток вкладок в браузере, плеер, мессенджер, редактор кода и ещё много всего. Поэтому кажется странным, что ничего из этого на самом деле не работает одновременно.

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

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

Количество тиков измеряется в герцах (Гц) — это единица измерения частоты протекания периодических процессов. Например, если мимо вашего дома раз в секунду проезжает гоночный болид, то его частота будет равна 1 Гц. Если болид проезжает два раза в секунду, то его частота — 2 Гц; если он проезжает трижды, то давно пора подумать о переезде.

Процессор так быстро выполняет процессы, что его частота измеряется в гигагерцах.

1 ГГц = 1 000 000 000 Гц

Understanding the basics

How is multithreading useful?

This multithreading model provides developers with a useful abstraction for concurrent execution. However, it really shines when applied to a single process: enabling parallel execution on a multiprocessor system.

What is multithreading?

Multithreading is a programming and execution model that allows multiple threads to exist within the context of a single process. These threads share the process’ resources but are able to execute independently.

What is a Qt application?

An application that is built with Qt framework. Qt applications are often built with C++ because the framework itself is built with C++. However, there are other language bindings such as Python-Qt.

Манипулирование сигнальной маской потока

int pthread_sigmask (int mode, sigset_t *set_p, sigset_t *old_p)

Изменяет сигнальную маску потока в соответствии с аргументом mode, который может принимать следующие значения:

• SIG_BLOCK — добавить сигналы из набора, указываемого set_p, в текущую сигнальную маску, описывающую блокируемые сигналы;
• SIG_UNBLOCK — удалить сигналы, содержащиеся в наборе, указываемом set_p, из текущей сигнальной маски;
• SIG_SETMASK — установить сигнальную маску, указываемую set_p, в качестве текущей.

Если значение аргумента old_p не равно NULL, то в область памяти, указываемую old_p, помещается предыдущее содержимое сигнальной маски.

Initializing thread with an object

You can initialize a thread not only with a function. You can use for this purpose function object (functor) or a member function of a class.

A functor is an object of a class that overloads operator () – function call operator.

If you want to initialize a thread with an object of a class, this class should overload operator(). It can be done in the following way:

class myFunctor
{
public:
	void operator()()
	{
		cout << "This is my function object" << endl;
	}
};

myFunctor

myFunctor myFunc;
thread functorTest(myFunc);
if (functorTest.joinable())
functorTest.join();

Add a public member function to myFunctor class:

void publicFunction()
{
	cout << "public function of myFunctor class is called" << endl;
}

publicFunction()myFunctor

myFunctor myFunc;
//initializing thread with member function of myFunctor class
thread functorTest(&myFunctor::publicFunction,myFunc);
if (functorTest.joinable())
	functorTest.join();

Joinable and not Joinable threads

After join() returns, thread becomes not joinable. A joinable thread is a thread that represents a thread of execution which has not yet been joined.

A thread is not joinable when it is default constructed or is moved/assigned to another thread or join() or detach() member function is called.

Not joinable thread can be destroyed safely.

You can check if a thread is joinable by using joinable() member function:

bool joinable()

//pass a function to thread
thread funcTest1(threadFunc);
//check if thread is joinable
if (funcTest1.joinable())
{
	//main is blocked until funcTest1 is not finished
	funcTest1.join();
}

Thread ID

Every thread has its unique identifier. Class thread has public member function that returns the ID of the thread:

id get_id()

Look on the following example:

//create 3 different threads
thread t1(showMessage);
thread t2(showMessage);
thread t3(showMessage);
//get id of all the threads
thread::id id1 = t1.get_id();
thread::id id2 = t2.get_id();
thread::id id3 = t3.get_id();
//join all the threads
if (t1.joinable())
{
	t1.join();
	cout << "Thread with id " << id1 << " is terminated" << endl;
}
if (t2.joinable())
{
	t2.join();
	cout << "Thread with id " << id2 << " is terminated" << endl;
}
if (t3.joinable())
{
	t3.join();
	cout << "Thread with id " << id3 << " is terminated" << endl;
}

Thread with id 8228 is terminated
Thread with id 10948 is terminated
Thread with id 9552 is terminated

Creating Threads

Threads are created by extending the Thread class. The extended Thread class then calls the Start() method to begin the child thread execution.

The following program demonstrates the concept −

using System;
using System.Threading;

namespace MultithreadingApplication {
   class ThreadCreationProgram {
      public static void CallToChildThread() {
         Console.WriteLine("Child thread starts");
      }
      static void Main(string[] args) {
         ThreadStart childref = new ThreadStart(CallToChildThread);
         Console.WriteLine("In Main: Creating the Child thread");
         Thread childThread = new Thread(childref);
         childThread.Start();
         Console.ReadKey();
      }
   }
}

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

In Main: Creating the Child thread
Child thread starts

Concurrent access to resources

Multithreading programming faces a problem with concurrent access to a shared resource. Simultaneous access to the same resource can leads to a lot of errors and chaos in the program.

Have a look at below example:

vector<int> vec;
void push()
{
	for (int i = 0; i != 10; ++i)
	{
		cout << "Push " << i << endl;
		_sleep(500);
		vec.push_back(i);
	}
}
void pop()
{
	for (int i = 0; i != 10; ++i)
	{
		if (vec.size() > 0)
		{
			int val = vec.back();
			vec.pop_back();
			cout << "Pop "<< val << endl;
		}
	_sleep(500);
	}
}
int main()
{
	//create two threads
	thread push(push);
	thread pop(pop);
	if (push.joinable())
		push.join();
	if (pop.joinable())
		pop.join();
}

vecpush pop

The access to the vector is not synchronized. Threads are accessing vector non-continuously. Because of simultaneous access to shared data many errors can appear.

Описание

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

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

К достоинствам многопоточной реализации той или иной системы перед многозадачной можно отнести следующее:

• Упрощение программы в некоторых случаях за счёт использования общего адресного пространства.
• Меньшие относительно процесса временны́е затраты на создание потока.

К достоинствам многопоточной реализации той или иной системы перед однопоточной можно отнести следующее:

• Упрощение программы в некоторых случаях, за счёт вынесения механизмов чередования выполнения различных слабо взаимосвязанных подзадач, требующих одновременного выполнения, в отдельную подсистему многопоточности.
• Повышение производительности процесса за счёт распараллеливания процессорных вычислений и операций ввода-вывода.

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

Многопотоковое программирование предложено в качестве средства разработки параллельных программ для многопроцессорных систем (систем с разделяемой памятью). При этом реальное разнесение потоков управления на разные процессоры — задача ОС. Фирма SUN Microsystems для поддержки потоков (нитей) управления реализовала легковесные процессы LWP (LightWeight Processes). Диспетчирование LWP — практически не управляемая пользователем процедура. Потоки характеризуются следующими атрибутами:

• идентификатор потока (уникален в рамках процесса);
• значение приоритета;
• сигнальная маска.

Предложены 2 API потокового программирования:

• фирмы SUN Microsystems (пионер в этом деле);
• комитета POSIX.1C по стандартизации.

Здесь рассматривается вариант POSIX (Portable Operating System Interface for Unix). Все функции этого варианта имеют в своих именах префикс pthread_ и объявлены в заголовочном файле pthread.h.

Initializing thread with a function

When you create a thread, you can pass a pointer of a function to its constructor. Once thread is created, this function starts its work in a separate thread. Look on an example:

#include <iostream>
#include <thread> 
using namespace std;
void threadFunc()
{
	cout << "Welcome to Multithreading" << endl;
}
int main()
{
	//pass a function to thread
	thread funcTest1(threadFunc);
}

As you can see, main thread creates new thread funcTest1 with a parameter threadFunc. Main thread does not wait for funcTest1 thread termination. It continues its work. The main thread finishes execution, but funcTest1 is still running. This causes error. All the threads must be terminated before main thread is terminated.

Running one task instance at a time

Imagine you need to ensure that only one task instance at a time can be executed and all pending requests to run the same task are waiting on a certain queue. This is often needed when a task is accessing an exclusive resource, such as writing to the same file or sending packets using TCP socket.

Let’s forget about computer science and producer-consumer pattern for a moment and consider something trivial; something that can be easily found in real projects.

A naïve solution to this problem could be using a . Inside the task function, you could simply acquire the mutex effectively serializing all threads attempting to run the task. This would guarantee that only one thread at a time could be running the function. However, this solution impacts the performance by introducing high contention problem because all those threads would be blocked (on the mutex) before they can proceed. If you have many threads actively using such a task and doing some useful job in between, then all these threads will be just sleeping most of the time.

To avoid contention, we need a queue and a worker that lives in its own thread and processing the queue. This is pretty much the classic producer-consumer pattern. The worker (consumer) would be picking requests from the queue one by one, and each producer can simply add its requests into the queue. Sounds simple at first and you may think of using and , but hold on and let’s see if we can achieve the goal without these primitives:

We can use QThreadPool as it has a queue of pending tasks

Or

We can use default QThread::run() because it has QEventLoop

The first option is to use . We can create a instance and use . Then we can be using to schedule requests:

This solution has one benefit: allows you to instantly cancel all pending requests, for example when your application needs to shut down quickly. However, there is also a significant drawback that is connected to thread-affinity: function will be likely executing in different threads from call to call. And we know Qt has some classes that require thread-affinity: , and possibly some others.

The better solution relies on using provided by . The idea is simple: we use a signal/slot mechanism to issue requests, and the event loop running inside the thread will serve as a queue allowing just one slot at a time to be executed.

Implementation of constructor and is straightforward and therefore not provided here. Now we need a service that will be managing the thread and the worker instance:

Let’s discuss how this code works:

• In the constructor, we create a thread and worker instance. Notice that the worker does not receive a parent, because it will be moved to the new thread. Because of this, Qt won’t be able to release the worker’s memory automatically, and therefore, we need to do this by connecting signal to slot. We also connect the proxy method to which will be using mode because of different threads.
• In the destructor, we put the event into the event loop’s queue. This event will be handled after all other events are handled. For example, if we have made hundreds of calls just prior to the destructor call, the logger will handle them all before it fetches the quit event. This takes time, of course, so we must until the event loop exits. It is worth mentioniong that all future logging requests posted after the quit event will never be processed.
• The logging itself () will always be done in the same thread, therefore this approach is working well for classes requiring thread-affinity. At the same time, constructor and destructor are executed in the main thread (specifically the thread is running in), and therefore, you need to be very careful about what code you are running there. Specifically, do not stop timers or use sockets in the worker’s destructor unless you could be running the destructor in the same thread!

Состязания за ресурсы

Состязание за ресурсы может возникать в случае, если два или более потоков получают доступ к одним и тем же объектам, а доступ к совместно используемому состоянию не синхронизируется.

Чтобы продемонстрировать состязание за ресурсы, ниже приведен пример, в котором определяется класс StateObject с полем int и методом ChangeState. В реализации ChangeState значение state проверяется на предмет равенства 5. Если это так, выполняется инкремент. Следующий оператор Trace.Assert немедленно проверяет, действительно ли state теперь имеет значение 6.

Кажется очевидным, что после инкремента переменной, имеющей значение 5, она должна быть равна 6. Однако это необязательно так. Например, если один поток только что выполнил оператор if (state == 5), планировщик может вытеснить его и запустить еще один поток. Второй поток попадет в тело if и, поскольку в переменной состояния по-прежнему содержится значение 5, оно будет инкрементировано до 6. После этого снова настанет черед выполнения первого потока, в результате чего в следующем операторе значение переменной состояния будет увеличено до 7. Именно здесь и возникает состязание за ресурсы с выводом соответствующего сообщения:

После запуска этой программы можно будет увидеть, как возникают состязания за ресурсы. То, сколько времени пройдет до возникновения первого состязания за ресурсы, зависит от используемой системы и компоновки программы — окончательной или отладочной. В случае если она компоновалась как окончательная версия, проблема будет возникать чаще, потому что код оптимизирован. Если в системе установлено несколько ЦП либо двух- или четырехядерные ЦП, на которых множество потоков могут выполняться одновременно, проблема тоже будет возникать чаще, чем в системе с одноядерным ЦП.

Из-за вытесняющей многозадачности в системе с одноядерным ЦП состязания также возникают, но не так часто. Ниже показан пример того, как может выглядеть выдаваемое программой сообщение. Здесь это сообщение информирует о том, что состязание за ресурсы возникло после 227 циклов. При каждом запуске приложения результаты будут выглядеть по-разному:

Избежать возникновения данной проблемы можно, заблокировав разделяемый объект. Это делается с помощью оператора lock.
Внутрь блока кода, отвечающего за блокировку объекта состояния, может попадать только один поток. Из-за того, что этот объект разделяется среди всех потоков, в случае, если какой-то один из потоков уже заблокировал его, другой поток при достижении отвечающего за блокировку блока кода должен остановиться и ожидать своей очереди.

При получении блокировки поток вступает во владение ею и снимает ее при достижении конца отвечающего за блокировку блока кода. Когда каждый поток, изменяющий объект, на который ссылается переменная состояния, использует блокировку, проблема с состязанием за ресурсы больше не возникает.

Слишком большое количество блокировок тоже может приводить к проблемам, например, к взаимоблокировке. Взаимоблокировкой (deadlock) называется ситуация, когда как минимум два потока останавливаются и ожидают друг от друга снятия блокировки. Поскольку оба потока ожидают друг от друга выполнения соответствующего действия, получается, что они блокируют друг друга, из-за чего их ожидание может длиться бесконечно.

Thread Life Cycle

The life cycle of a thread starts when an object of the System.Threading.Thread class is created and ends when the thread is terminated or completes execution.

Following are the various states in the life cycle of a thread −

• The Unstarted State − It is the situation when the instance of the thread is created but the Start method is not called.

• The Ready State − It is the situation when the thread is ready to run and waiting CPU cycle.

• The Not Runnable State − A thread is not executable, when

  • Sleep method has been called
  • Wait method has been called
  • Blocked by I/O operations

• The Dead State − It is the situation when the thread completes execution or is aborted.

Joining and Detaching Threads

There are following two routines which we can use to join or detach threads −

pthread_join (threadid, status) 
pthread_detach (threadid) 

The pthread_join() subroutine blocks the calling thread until the specified ‘threadid’ thread terminates. When a thread is created, one of its attributes defines whether it is joinable or detached. Only threads that are created as joinable can be joined. If a thread is created as detached, it can never be joined.

This example demonstrates how to wait for thread completions by using the Pthread join routine.

#include <iostream>
#include <cstdlib>
#include <pthread.h>
#include <unistd.h>

using namespace std;

#define NUM_THREADS 5

void *wait(void *t) {
   int i;
   long tid;

   tid = (long)t;

   sleep(1);
   cout << "Sleeping in thread " << endl;
   cout << "Thread with id : " << tid << "  ...exiting " << endl;
   pthread_exit(NULL);
}

int main () {
   int rc;
   int i;
   pthread_t threads;
   pthread_attr_t attr;
   void *status;

   // Initialize and set thread joinable
   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

   for( i = 0; i < NUM_THREADS; i++ ) {
      cout << "main() : creating thread, " << i << endl;
      rc = pthread_create(&threads, &attr, wait, (void *)i );
      if (rc) {
         cout << "Error:unable to create thread," << rc << endl;
         exit(-1);
      }
   }

   // free attribute and wait for the other threads
   pthread_attr_destroy(&attr);
   for( i = 0; i < NUM_THREADS; i++ ) {
      rc = pthread_join(threads, &status);
      if (rc) {
         cout << "Error:unable to join," << rc << endl;
         exit(-1);
      }
      cout << "Main: completed thread id :" << i ;
      cout << "  exiting with status :" << status << endl;
   }

   cout << "Main: program exiting." << endl;
   pthread_exit(NULL);
}

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

main() : creating thread, 0
main() : creating thread, 1
main() : creating thread, 2
main() : creating thread, 3
main() : creating thread, 4
Sleeping in thread
Thread with id : 0 .... exiting
Sleeping in thread
Thread with id : 1 .... exiting
Sleeping in thread
Thread with id : 2 .... exiting
Sleeping in thread
Thread with id : 3 .... exiting
Sleeping in thread
Thread with id : 4 .... exiting
Main: completed thread id :0  exiting with status :0
Main: completed thread id :1  exiting with status :0
Main: completed thread id :2  exiting with status :0
Main: completed thread id :3  exiting with status :0
Main: completed thread id :4  exiting with status :0
Main: program exiting.

Previous Page
Print Page

Next Page