std.concurrency

Это низкоуровневый API обмена сообщениями, на основе которого могут быть построены более структурированные или ограниченные API. Основная идея заключается в том, что каждый объект обмена сообщениями представлен общим типом дескриптора, называемым Tid, который позволяет отправлять сообщения логическим потокам, выполняющимся как в текущем процессе, так и во внешних процессах, используя один и тот же интерфейс. Это важный аспект масштабируемости, поскольку позволяет компонентам программы распределяться по доступным ресурсам с минимальными или отсутствующими изменениями в фактической реализации.

Логический поток — это контекст выполнения, имеющий свою собственную стек и выполняющийся асинхронно по отношению к другим логическим потокам. Это могут быть прерывисто планируемые ядра потоков, волокна (кооперативные потоки пространства пользователя) или какой-либо другой концепции с аналогичным поведением.

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

Лицензия:: Boost License 1.0.

Авторы:: Шон Келлли, Алекс Ронне Петерсен, Мартин Новак

Источник: std/concurrency.d

Примеры:

__gshared string received;
static void spawnedFunc(Tid ownerTid)
{
    import std.conv : text;
    // Receive a message from the owner thread.
    receive((int i){
        received = text("Received the number ", i);

        // Send a message back to the owner thread
        // indicating success.
        send(ownerTid, true);
    });
}

// Start spawnedFunc in a new thread.
auto childTid = spawn(&spawnedFunc, thisTid);

// Send the number 42 to this new thread.
send(childTid, 42);

// Receive the result code.
auto wasSuccessful = receiveOnly!(bool);
assert(wasSuccessful);
writeln(received); // "Received the number 42"

class MessageMismatch: object.Exception;

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

pure nothrow @nogc @safe this(string msg = "Unexpected message type");

class OwnerTerminated: object.Exception;

Выбрасывается при вызовах receive , если поток, запустивший получающий поток, завершился, и больше сообщений не существует.

pure nothrow @nogc @safe this(Tid t, string msg = "Owner terminated");

class LinkTerminated: object.Exception;

Выбрасывается, если связанный поток завершился.

pure nothrow @nogc @safe this(Tid t, string msg = "Link terminated");

class PriorityMessageException: object.Exception;

Выбрасывается, если сообщение было отправлено потоку через std.concurrency.prioritySend, и получатель не имеет обработчика для сообщения такого типа.

this(Variant vals);
Variant message;: Отправленное сообщение.

class MailboxFull: object.Exception;

Выбрасывается при переполнении почтового ящика, если почтовый ящик настроен с OnCrowding.throwException.

pure nothrow @nogc @safe this(Tid t, string msg = "Mailbox full");

class TidMissingException: object.Exception;

Выбрасывается, когда Tid отсутствует, например, когда ownerTid не находит владеющий поток.

struct Tid;

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

void toString(scope void delegate(const(char)[]) sink);: Генерирует удобную строку для идентификации этого Tid. Это полезно только для проверки, являются ли выполняемые Tid одинаковыми или разными, например, для ведения журнала и отладки. Теоретически возможно, что Tid, выполненный в будущем, будет иметь такое же значение toString(), как другой Tid, который уже завершился.

@property @safe Tid thisTid();

Возвращает:: Tid вызывающего потока.

@property Tid ownerTid();

Возвращает Tid потока, который запустил вызывающий поток.

Выбрасывает:: Исключение TidMissingException , если владеющего потока нет.

Tid spawn(F, T...)(F fn, T args)
Constraints: if (isSpawnable!(F, T));

Запускает fn(args) в новом логическом потоке.

Выполняет предоставленную функцию в новом логическом потоке, представленном Tid. Вызывающий поток обозначается как владелец нового потока. При завершении владеющего потока новому потоку будет отправлено сообщение OwnerTerminated, вызвав исключение OwnerTerminated при receive().

Параметры:

F `fn`	Функция для выполнения.
T `args`	Аргументы функции.

Возвращает:: Tid, представляющий новый логический поток.

Примечания: args не должно иметь неразделяемой алиасии. Другими словами, все аргументы к fn должны быть либо shared, либо immutable, или не иметь указателей. Это необходимо для обеспечения изоляции между потоками.

Примеры:

static void f(string msg)
{
    writeln(msg); // "Hello World"
}

auto tid = spawn(&f, "Hello World");

Примеры:

Неудачно: char[] имеет изменяемую алиасию.

string msg = "Hello, World!";

static void f1(string msg) {}
static assert(!__traits(compiles, spawn(&f1, msg.dup)));
static assert( __traits(compiles, spawn(&f1, msg.idup)));

static void f2(char[] msg) {}
static assert(!__traits(compiles, spawn(&f2, msg.dup)));
static assert(!__traits(compiles, spawn(&f2, msg.idup)));

Примеры:

Новый поток с анонимной функцией

spawn({
    ownerTid.send("This is so great!");
});
writeln(receiveOnly!string); // "This is so great!"

Tid spawnLinked(F, T...)(F fn, T args)
Constraints: if (isSpawnable!(F, T));

Запускает fn(args) в логическом потоке и получит сообщение LinkTerminated при завершении операции.

Выполняет предоставленную функцию в новом логическом потоке, представленном Tid. Этот новый поток связан с вызывающим потоком, так что если любой из них или вызывающий поток завершится, другому будет отправлено сообщение LinkTerminated, вызвав исключение LinkTerminated при получении. Связь потоков из spawn() также сохраняется, поэтому, если связь между потоками разорвана, завершение владельца все равно приведет к исключению OwnerTerminated при получении.

Параметры:

F `fn`	Функция для выполнения.
T `args`	Аргументы функции.

Возвращает:: Tid, представляющий новый поток.

void send(T...)(Tid tid, T vals);

Размещает значения в качестве сообщения в конце очереди сообщений tid.

Отправляет указанное значение в поток, представленный tid. Как и в std.concurrency.spawn, T не должно иметь неразделяемой алиасии.

void prioritySend(T...)(Tid tid, T vals);

Размещает значения в качестве сообщения в начале очереди сообщений tid.

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

void receive(T...)(T ops);

Получает сообщение от другого потока.

Получает сообщение от другого потока или блокируется, если нет сообщений указанных типов. Эта функция работает путем сопоставления шаблона сообщения с набором делегатов и выполнения первого найденного совпадения.

Если делегат, принимающий std.variant.Variant, включен в качестве последнего аргумента к receive, он будет сопоставлять любое сообщение, которое не было сопоставлено предыдущим делегатом. Если отправлено более одного аргумента, Variant будет содержать std.typecons.Tuple всех отправленных значений.

Параметры:

T ops Список функций-указателей и делегатов. Элементы в этом списке не должны перекрывать последующие элементы.

Выбрасывает:: OwnerTerminated при завершении потока-отправителя.

Примеры:

import std.variant : Variant;

auto process = ()
{
    receive(
        (int i) { ownerTid.send(1); },
        (double f) { ownerTid.send(2); },
        (Variant v) { ownerTid.send(3); }
    );
};

{
    auto tid = spawn(process);
    send(tid, 42);
    writeln(receiveOnly!int); // 1
}

{
    auto tid = spawn(process);
    send(tid, 3.14);
    writeln(receiveOnly!int); // 2
}

{
    auto tid = spawn(process);
    send(tid, "something else");
    writeln(receiveOnly!int); // 3
}

receiveOnlyRet!T receiveOnly(T...)();

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

Параметры:

T	Список типов, которые нужно получить.

Возвращает:: Полученное сообщение. Если T содержит более одного элемента, сообщение будет упаковано в std.typecons.Tuple.

Выбрасывает:: MessageMismatch если получено сообщение с типами, отличными от T, OwnerTerminated при завершении потока-отправителя.

Примеры:

auto tid = spawn(
{
    assert(receiveOnly!int == 42);
});
send(tid, 42);

Примеры:

auto tid = spawn(
{
    assert(receiveOnly!string == "text");
});
send(tid, "text");

Примеры:

struct Record { string name; int age; }

auto tid = spawn(
{
    auto msg = receiveOnly!(double, Record);
    assert(msg[0] == 0.5);
    assert(msg[1].name == "Alice");
    assert(msg[1].age == 31);
});

send(tid, 0.5, Record("Alice", 31));

bool receiveTimeout(T...)(Duration duration, T ops);

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

Получает сообщение от другого потока или блокируется до тех пор, пока duration не превысит, если нет сообщений указанных типов. Эта функция работает путем сопоставления шаблона сообщения с набором делегатов и выполнения первого найденного совпадения.

Если делегат, принимающий std.variant.Variant, включен в качестве последнего аргумента, он будет сопоставлять любое сообщение, которое не было сопоставлено предыдущим делегатом. Если отправлено более одного аргумента, Variant будет содержать std.typecons.Tuple всех отправленных значений.

Параметры:

Duration `duration`	Срок ожидания. Если `duration` отрицательно, ожидание не происходит.
T `ops`	Список функций-указателей и делегатов. Элементы в этом списке не должны перекрывать последующие элементы.

Возвращает:: true , если получено сообщение, и false , если ожидание закончилось.

Выбрасывает:: OwnerTerminated при завершении потока-отправителя.

enum OnCrowding: int;

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

block: Ожидать освобождения места.
throwException: Выбросить исключение MailboxFull.
ignore: Прервать отправку и вернуть значение.

pure @safe void setMaxMailboxSize(Tid tid, size_t messages, OnCrowding doThis);

Устанавливает максимальный размер почтового ящика.

Устанавливает ограничение на максимальное количество сообщений пользователей, разрешенных в почтовом ящике. Если это ограничение достигнуто, вызывающий поток, пытающийся добавить новое сообщение, выполнит поведение, указанное в doThis. Если messages равно нулю, почтовый ящик не ограничен.

Параметры:

Tid `tid`	Tid потока, для которого должно быть установлено это ограничение.
size_t `messages`	Максимальное количество сообщений или ноль, если нет ограничения.
OnCrowding `doThis`	Поведение, выполняемое при отправке сообщения в полный почтовый ящик.

void setMaxMailboxSize(Tid tid, size_t messages, bool function(Tid) onCrowdingDoThis);

Устанавливает максимальный размер почтового ящика.

Устанавливает ограничение на максимальное количество сообщений пользователей, разрешенных в почтовом ящике. Если это ограничение достигнуто, вызывающий поток, пытающийся добавить новое сообщение, выполнит onCrowdingDoThis. Если messages равно нулю, почтовый ящик не ограничен.

Параметры:

Tid `tid`	Tid потока, для которого должно быть установлено это ограничение.
size_t `messages`	Максимальное количество сообщений или ноль, если нет ограничения.
bool function(Tid) `onCrowdingDoThis`	Процедура, вызываемая при отправке сообщения в полный почтовый ящик.

bool register(string name, Tid tid);

Связывает имя с tid.

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

Параметры:

string `name`	Имя, которое нужно связать с tid.
Tid `tid`	Tid, зарегистрированный по имени.

Возвращает:: true, если имя доступно и tid не известно как представляющее завершившийся поток.

bool unregister(string name);

Удаляет зарегистрированное имя, связанное с tid.

Параметры:

string name Имя, которое нужно аннулировать.

Возвращает:: true, если имя зарегистрировано, false — если нет.

Tid locate(string name);

Получает Tid, связанное с именем.

Параметры:

string name Имя для поиска в реестре.

Возвращает:: Связанное Tid или Tid.init, если имя не зарегистрировано.

struct ThreadInfo;

Инкапсулирует все данные на уровне реализации, необходимые для планирования.

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

static nothrow @property ref auto thisInfo();

Получает локальную для потока экземпляр ThreadInfo.

Получает локальную для потока экземпляр ThreadInfo, который следует использовать в качестве стандартной экземпляра, когда info запрашивается для потока, не созданного Scheduler.

void cleanup();

Очищает этот ThreadInfo.

Это необходимо вызывать при завершении запланированного потока. Она разрывает систему обмена сообщениями для потока и уведомляет заинтересованные стороны о завершении работы потока.

interface Scheduler;

Scheduler управляет тем, как выполняется многопоточность с помощью spawn.

Реализация Scheduler позволяет настроить механизм конкурентности, используемый этим модулем, в соответствии с различными потребностями. По умолчанию вызов spawn создаст новый поток ядра, который выполнит предоставленную функцию и завершится по завершении. Но можно создать Scheduler, которые повторно используют потоки, объединяют волокна (корутины) в одном потоке или любое другое подходы. Сделав выбор Scheduler опцией на уровне пользователя, std.concurrency можно использовать для гораздо большего числа типов приложений, чем если бы это поведение было предварительно определено.

Пример

import std.concurrency;
import std.stdio;

void main()
{
    scheduler = new FiberScheduler;
    scheduler.start(
    {
        writeln("the rest of main goes here");
    });
}

Некоторые schedulers имеют цикл диспетчеризации, который должен работать, если они должны работать должным образом, поэтому во избежание неполадок, при использовании планировщика, start() необходимо вызвать внутри main(). Это передаёт управление планировщику и гарантирует, что все запущенные потоки будут выполнены ожидаемым образом.

abstract void start(void delegate() op);

Запускает предоставленный op и запускает Scheduler.

Это предназначено для вызова в начале программы, чтобы передать всё планирование активному экземпляру Scheduler. Это необходимо для планировщиков, которые явно отправляют потоки, а не просто полагаются на операционную систему, и поэтому start всегда должен вызываться внутри main(), чтобы начать обычное выполнение программы.

Параметры:

void delegate() op Обёртка для того, что основной поток сделал бы в отсутствие пользовательского планировщика. Он будет автоматически выполнен путём вызова spawn планировщиком.

abstract void spawn(void delegate() op);

Назначает логический поток для выполнения предоставленного op.

Эта процедура вызывается spawn. Ожидается, что она создаст новый логический поток и выполнит заданную операцию. Этот поток должен вызвать thisInfo.cleanup(), когда поток завершится, если запланированный поток не является потоком ядра — все потоки ядра будут иметь ThreadInfo, очищенный автоматически локальным для потока деструктором.

Параметры:

void delegate() op Функция для выполнения. Это может быть фактическая функция, переданная пользователем в spawn, или может быть обёрткой функции.

abstract nothrow void yield();

Передаёт выполнение другому логическому потоку.

Эта процедура вызывается в различных точках в API, ориентированных на конкурентность, чтобы дать планировщику возможность передать выполнение, когда используется какая-то модель кооперативного многопоточного доступа. Если это не подходит, например, когда каждый логический поток поддерживается выделенным потоком ядра, эта процедура может быть пустой операцией.

abstract nothrow @property ref ThreadInfo thisInfo();

Возвращает соответствующий экземпляр ThreadInfo.

Возвращает экземпляр ThreadInfo, специфичный для логического потока, вызывающего эту процедуру, или, если вызывающий поток не был создан этим планировщиком, возвращает ThreadInfo.thisInfo.

abstract nothrow Condition newCondition(Mutex m);

Создаёт аналог переменной Condition для сигнализации.

Создаёт новый аналог переменной Condition, который используется для проверки и сигнализации добавления сообщений в очередь сообщений потока. Как и yield, некоторые планировщики могут потребовать определения пользовательского поведения, чтобы вызовы Condition.wait() передавали выполнение другому потоку, когда новые сообщения недоступны, вместо блокировки.

Параметры:

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

class ThreadScheduler: std.concurrency.Scheduler;

Пример Scheduler, использующий потоки ядра.

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

void start(void delegate() op);: Этот код просто выполняет op напрямую, поскольку в этом подходе нет реального планирования.
void spawn(void delegate() op);: Создаёт новый поток ядра и назначает его для выполнения предоставленного op.
nothrow void yield();: Этот планировщик не выполняет явного множественного доступа, поэтому это пустая операция.
nothrow @property ref ThreadInfo thisInfo();: Возвращает ThreadInfo.thisInfo, поскольку это экземпляр ThreadInfo, локальный для потока, что является правильным поведением для этого планировщика.
nothrow Condition newCondition(Mutex m);: Создаёт новую переменную Condition. Здесь не требуется пользовательское поведение.

class FiberScheduler: std.concurrency.Scheduler;

Пример планировщика с использованием волокон.

Это пример планировщика, который создаёт новое волокно при каждом вызове spawn и мультиплексирует выполнение всех волокон в главном потоке.

void start(void delegate() op);

Это создаёт новое волокно для переданного op и затем запускает диспетчер.

nothrow void spawn(void delegate() op);

Это создаёт новое волокно для переданного op и добавляет его в список диспетчеризации.

nothrow void yield();

Если вызывающее волокно запланировано, это уступает выполнение другому запланированному волокну.

nothrow @property ref ThreadInfo thisInfo();

Возвращает соответствующий экземпляр ThreadInfo.

Возвращает экземпляр ThreadInfo, специфичный для вызывающего волокна, если волокно было создано этим диспетчером, в противном случае возвращает ThreadInfo.thisInfo.

nothrow Condition newCondition(Mutex m);

Возвращает аналог Condition, который уступает выполнение при вызове wait или notify.

Ошибка: Для реализации по умолчанию, notifyAll будет вести себя как notify.

Параметры:

Mutex m A Mutex для блокировки, если к условию требуется обращение ожидания или уведомления из нескольких Thread. Если null, никакого Mutex не будет использоваться, и предполагается, что к Condition обращаются только для ожидания/уведомления из одного Thread.

protected nothrow void create(void delegate() op);

Создаёт новое волокно, которое вызывает указанный делегат.

Параметры:

void delegate() op Делегат, который должно вызвать волокно

class InfoFiber: core.thread.fiber.Fiber;

Волокно, которое встраивает ThreadInfo

Scheduler scheduler;

Устанавливает поведение планировщика в программе.

Эта переменная устанавливает поведение планировщика в этой программе. Обычно, при установке планировщика, scheduler.start() должен быть вызван в main. Эта процедура не вернётся, пока выполнение программы не завершится.

nothrow void yield();

Если вызывающее элемент — волокно и не генератор, эта функция вызовет scheduler.yield() или Fiber.yield(), в зависимости от ситуации.

class Generator(T): Fiber, IsGenerator, InputRange!T;

Генератор — это волокно, которое периодически возвращает значения типа T вызывающему элементу через yield. Это представлено как InputRange.

Примеры:

auto tid = spawn({
    int i;
    while (i < 9)
        i = receiveOnly!int;

    ownerTid.send(i * 2);
});

auto r = new Generator!int({
    foreach (i; 1 .. 10)
        yield(i);
});

foreach (e; r)
    tid.send(e);

writeln(receiveOnly!int); // 18

this(void function() fn);

Инициализирует объект генератора, связанный со статической функцией D. Функция будет вызвана один раз для подготовки диапазона для итерации.

Параметры:

void function() fn Функция волокна.

В: fn не должен быть null.

this(void function() fn, size_t sz);

Параметры:

void function() `fn`	Функция волокна.
size_t `sz`	Размер стека для этого волокна.

В: fn не должен быть null.

this(void function() fn, size_t sz, size_t guardPageSize);

Параметры:

void function() `fn`	Функция волокна.
size_t `sz`	Размер стека для этого волокна.
size_t `guardPageSize`	Размер страницы стража для перехвата переполнений стека волокна. См. документацию `core.thread.Fiber` для получения более подробной информации.

В: fn не должен быть null.

this(void delegate() dg);

Инициализирует объект генератора, связанный с динамической функцией D. Функция будет вызвана один раз для подготовки диапазона для итерации.

Параметры:

void delegate() dg Функция волокна.

В: dg не должен быть null.

this(void delegate() dg, size_t sz);

Параметры:

void delegate() `dg`	Функция волокна.
size_t `sz`	Размер стека для этого волокна.

В: dg не должен быть null.

this(void delegate() dg, size_t sz, size_t guardPageSize);

Параметры:

void delegate() `dg`	Функция волокна.
size_t `sz`	Размер стека для этого волокна.
size_t `guardPageSize`	Размер страницы стража для перехвата переполнений стека волокна. См. документацию `core.thread.Fiber` для получения более подробной информации.

В: dg не должен быть null.

final @property bool empty();

Возвращает true, если генератор пустой.

final void popFront();

Получает следующее значение из подлежащей функции.

final @property T front();

Возвращает последнее сгенерированное значение с помощью копирования по значению.

final T moveFront();

Возвращает последнее сгенерированное значение без выполнения конструктора копирования. Не будет компилироваться для типов элементов, определяющих postblit, потому что Generator не возвращает по ссылке.

void yield(T)(ref T value);

void yield(T)(T value);

Уступает значение типа T вызывающему элементу текущего выполняющегося генератора.

Параметры:

T value Значение для уступки.

Примеры:

import std.range;

InputRange!int myIota = iota(10).inputRangeObject;

myIota.popFront();
myIota.popFront();
writeln(myIota.moveFront); // 2
writeln(myIota.front); // 2
myIota.popFront();
writeln(myIota.front); // 3

//can be assigned to std.range.interfaces.InputRange directly
myIota = new Generator!int(
{
    foreach (i; 0 .. 10) yield(i);
});

myIota.popFront();
myIota.popFront();
writeln(myIota.moveFront); // 2
writeln(myIota.front); // 2
myIota.popFront();
writeln(myIota.front); // 3

size_t[2] counter = [0, 0];
foreach (i, unused; myIota) counter[] += [1, i];

assert(myIota.empty);
writeln(counter); // [7, 21]

ref auto initOnce(alias var)(lazy typeof(var) init);

Инициализирует var значением init в потокобезопасном режиме.

Реализация гарантирует, что все потоки, одновременно вызывающие initOnce с тем же аргументом var, будут блокироваться, пока var не будет полностью инициализирован. Все побочные эффекты init будут видны глобально после этого.

Параметры:

var	Переменная для инициализации
typeof(var) `init`	Ленивое значение инициализатора

Возвращает:: Ссылка на инициализированную переменную

Примеры:

Типичный случай использования — ленивая, но потокобезопасная инициализация.

static class MySingleton
{
    static MySingleton instance()
    {
        __gshared MySingleton inst;
        return initOnce!inst(new MySingleton);
    }
}

assert(MySingleton.instance !is null);

ref auto initOnce(alias var)(lazy typeof(var) init, shared Mutex mutex);

ref auto initOnce(alias var)(lazy typeof(var) init, Mutex mutex);

То же самое, но принимает отдельный mutex вместо совместного использования одного mutex среди всех экземпляров initOnce.

Это необходимо для предотвращения тупиковых ситуаций, когда выражение init ждёт результата другого потока, который также может вызвать initOnce. Используйте с осторожностью.

Параметры:

var	Переменная для инициализации
typeof(var) `init`	Ленивое значение инициализатора
Mutex `mutex`	Mutex для предотвращения гонок

Возвращает:: Ссылка на инициализированную переменную

Примеры:

Используйте отдельный mutex, когда init ожидает другого потока, который также может вызвать initOnce.

import core.sync.mutex : Mutex;

static shared bool varA, varB;
static shared Mutex m;
m = new shared Mutex;

spawn({
    // use a different mutex for varB to avoid a dead-lock
    initOnce!varB(true, m);
    ownerTid.send(true);
});
// init depends on the result of the spawned thread
initOnce!varA(receiveOnly!bool);
writeln(varA); // true
writeln(varB); // true

© 1999–2021 The D Language Foundation
Licensed under the Boost License 1.0.
https://dlang.org/phobos/std_concurrency.html