Начало
Учебные материалыПодразделы
Другие разделы
| Библиотека для графических интерфейсов nuklear |
| Графическая библиотека |
| Консольный ввод-вывод coniow |
| Консольный ввод-вывод curses |
| Передача данных по сети |
| Управляющие последовательности |
| Файловая система |
Дата и время
22/12/2024 08:07:17
Распараллеливание программ
Потоки и библиотеки для распараллеливания OpenMP Стандартная библиотека С++ Стандартная библиотека С
Стандартная библиотека С
Тип ``thrd_t`` используется для хранения идентификатора потока. В С имеются следующие функции для работы с потоками.
``int thrd_create(thrd_t *tid, thrd_start_t func, void *arg);``
Выполняет вызов функции ``func(arg)`` в параллельном потоке. Возвращается ``thrd_success``, если создание потока прошло успешно, а идентификатор потока записывается в ``*tid``.
``int thrd_join(thrd_t tid, int *res);``
Ожидает завершения потока с идентификатором ``tid``. Результат, возвращаемый ``func``, помещается в ``*res``.
``thrd_t thrd_current(void);``
Возвращает идентификатор текущего потока.
``void thrd_exit(int res);``
Завершает выполнение текущего потока с результатом ``res``. Также можно завершить с помощью ``return res;``
``void thrd_yield(void);``
Переключение на следующий готовый к выполнению поток.
``int thrd_sleep(const struct timespec* duration, struct timespec* remaining);``
Ожидать указанное количество времени (в поле ``tv_sec`` нужно указать количество секунд, ``tv_nsec`` -- наносекунд). Второй параметр можно указать ``NULL``, в общем случае в ``remaining`` помещается оставшееся время, если ожидание было прервано досрочно.
> ``struct timespec dt={1,5e8}; // 1.5 секунды``
> ``thrd_sleep(&dt,NULL);``
Пример программы
```c
#include <threads.h>
#include <stdio.h>
char p[100000001];
int maxn=100000000;
int primes(void *arg) // определение простых чисел с помощью решета
{ for(int i=2;i<=maxn;++i)
p[i]=1;
for(int i=2;i*i<=maxn;++i)
if(p[i])
for(int j=i*i;j<=maxn;j+=i)
p[j]=0;
return 0;
}
int main()
{ thrd_t ptid;
int a,b,k=0,res;
// пока выполняется подсчет, делаем ввод
thrd_create(&ptid,primes,NULL);
printf("Введите диапазон чисел: ");
scanf("%d%d",&a,&b);
thrd_join(ptid,&res); // ожидаем завершения расчета
k=0;
for(int i=a;i<=b;++i)
if(p[i]) ++k;
printf("В этом диапазоне %d простых чисел\n", k);
}
```
Тип ``mtx_t`` используется для хранения мьютекса. В С имеются следующие функции для работы с мьютексами.
``int mtx_init(mtx_t* mutex, int type);``
Создание мьютекса. В ``type`` можно указать ``mtx_plain`` для создания обычного мьютекса, ``mtx_timed`` -- мьютекса с возможностью ожидания по времени, ``mtx_plain | mtx_recursive`` -- мьютекса с подсчетом количества блокировок (возможно повторная блокировка тем же потоком, мьютекс освобождается, когда количество блокировок уменьшается до 0) и ``mtx_timed | mtx_recursive``.
``void mtx_destroy(mtx_t* mutex);``
Уничтожение мьютекса.
``int mtx_lock(mtx_t* mutex);``
Блокировка мьютекса. Выполнение кода будет продолжено только после успешной блокировки.
``int mtx_unlock(mtx_t* mutex);``
Разблокировать мьютекс.
``int mtx_trylock(mtx_t* mutex);``
Попытка блокировки мьютекса. При успешной блокировке возвращается ``thrd_success``, иначе ``thrd_busy``.
``int mtx_timedlock(mtx_t* mutex, const struct timespec* time_point);``
Попытка блокировки мьютекса до успеха или пока не наступит время ``time_point``. При успешной блокировке возвращается ``thrd_success``, иначе ``thrd_timedout``.
```c
struct timespec ts;
timespec_get(&ts, TIME_UTC);
ts.tv_sec+=1;
if(mtx_timedlock(&mutex,&ts)==thrd_success)) {
// ...заблокирован
}
```
Пример блокировки одновременного вывода на консоль:
```c
#include <threads.h>
#include <stdio.h>
typedef struct thrd_info {
char name[100]; // информация о потоке
int a, b;
} thrd_info;
mtx_t m_console; // мьютекс для консоли
void print(const char name[], long long n) {
mtx_lock(&m_console);
printf("Thread %s: ", name); // printf по стандарту потокобезопасен,
printf("%lld",n); // поэтому вывод разбит на несколько частей,
printf("\n"); // чтобы была возможность другому потоку прервать процесс
mtx_unlock(&m_console);
}
int func(void *arg)
{ thrd_info *info=(thrd_info *)arg;
for(int i=info->a;i<=info->b;++i)
{ long long s=0; // сложное вычисление
for(int j=1;j<100000*i;++j)
s+=j;
print(info->name,s); // печать
}
return 0;
}
int main()
{
thrd_t t1, t2;
thrd_info tinfo1={"A",1,15}, tinfo2={"B",50,57};
mtx_init(&m_console,mtx_plain);
thrd_create(&t1,func,&tinfo1);
thrd_create(&t2,func,&tinfo2);
int res;
thrd_join(t1,&res);
thrd_join(t2,&res);
mtx_destroy(&m_console);
}
```
Тип ``atomic_flag`` из ``<stdatomic.h>`` позволяет создать простейший вариант семафора "спинлок" (spinlock):
```c
atomic_flag flag_console = ATOMIC_FLAG_INIT;
// установить флаг в 1 и получить предыдущее значение флага
while(atomic_flag_test_and_set(&flag_console))
thrd_yield(); // ждать
// вывод на консоль
atomic_flag_clear(&flag_console);
```
Недостаток спинлока в том, что поток всегда активен, в отличие от мьютекса, в котором ожидающий поток ставится в очередь и становится активным только после разблокирования мьютекса.
Если попытаться реализовать передачу информации из одного потока в другой с помощью очереди, то получим аналогичный цикл ожидания. Для решения этой проблемы в библиотеке есть переменные для блокировки по условию ``cnd_t`` и следующие функции для работы с ними.
``int cnd_init(cnd_t* cond);``
Создание условной переменной.
``int cnd_destroy(cnd_t* cond);``
Уничтожение условной переменной.
``int cnd_wait(cnd_t* cond, mtx_t* mutex);``
Ожидание выполнения условия ``cond``, ``mutex`` должен быть заблокирован перед выполнением этой функции, при возврате также остается заблокирован, но на время выполнения мьютекс разблокируется, чтобы другие потоки могли получить доступ к защищаемой структуре данных.
``int cnd_timedwait(cnd_t* cond, mtx_t* mutex, const struct timespec* time_point);``
Аналогично, но ожидание не превышает указанного времени.
``int cnd_signal(cnd_t *cond);``
Сообщить о выполнении условия ``cond`` первому ожидающему потоку, если он есть.
``int cnd_broadcast(cnd_t *cond);``
Сообщить о выполнении условия ``cond`` всем ожидающим потокам.
Пример взаимодействия потоков через очередь и использования условий.
```c
#include <threads.h>
#include <stdio.h>
typedef struct ts_queue {
mtx_t m; // мьютекс для доступа к структуре
cnd_t notempty, notfull;
int first, size, done;
long data[100];
} ts_queue;
void queue_init(ts_queue* q) // инициализация очереди
{ q->first=q->size=q->done=0;
mtx_init(&q->m,mtx_plain);
cnd_init(&q->notempty);
cnd_init(&q->notfull);
}
void queue_destroy(ts_queue* q) // уничтожение очереди
{ mtx_destroy(&q->m);
cnd_destroy(&q->notempty);
cnd_destroy(&q->notfull);
}
void queue_add(ts_queue* q, long v) // добавить значение в очередь
{ mtx_lock(&q->m);
if(q->size==100)
cnd_wait(&q->notfull,&q->m);
q->data[(q->first+q->size++)%100]=v;
cnd_signal(&q->notempty);
mtx_unlock(&q->m);
}
void queue_done(ts_queue* q) // данные закончились
{ mtx_lock(&q->m);
q->done=1;
cnd_broadcast(&q->notempty);
mtx_unlock(&q->m);
}
int queue_get(ts_queue* q, long *v) // взять первое значение
{ int r=0;
mtx_lock(&q->m);
if(q->size==0 && !q->done)
cnd_wait(&q->notempty,&q->m);
if(q->size>0)
{
*v=q->data[q->first];
q->size--;
q->first=(q->first+1)%100;
r=1;
cnd_signal(&q->notfull);
}
mtx_unlock(&q->m);
return r;
}
ts_queue q1;
int producer(void *arg)
{ long n;
while(scanf("%ld",&n)==1)
queue_add(&q1,n);
queue_done(&q1);
return 0;
}
int consumer(void *arg)
{
long n;
while(queue_get(&q1,&n))
{ int k=0;
for(int i=1; i<=n; ++i)
if(n%i==0)
++k;
printf("Количество делителей %ld %d\n",n,k);
}
return 0;
}
int main()
{
thrd_t t1, t2;
queue_init(&q1);
thrd_create(&t1,producer,NULL);
thrd_create(&t2,consumer,NULL);
int res;
thrd_join(t1,&res);
thrd_join(t2,&res);
queue_destroy(&q1);
}
```
