C++ магистрали потоков и древо алгоритма

Классная штука! Какая у неё производительность на поток и overhead? Я обычно не пишу на плюсах когда можно тратить больше микросекунды на message/event/data-row или когда надо память много динамически аллокировать, так как AKKA actors/streams могут пару сотен тысяч сообщений в секунду обрабатывать, CompletableFuture - того же порядка и GC очень хорошо работает в новых JVM (JRE11+).

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

В библиотеке эта проблема решена. В следующем примере используется std::weak_ptr<SelfProtectedTask> в качестве защиты от запуска лямбды если целевой объект уже разрушился при выходе из скоупа

Позволю себе позанудствовать и высказать свои опасения в легальности примененного вами решения. А именно:

struct SharedHolder : public std::enable_shared_from_this<SharedHolder>
{
	alignas(T) unsigned char buffer_[sizeof(T)];
	T * ptr_{nullptr};
	std::shared_ptr<T> construct(Args &&... args)
	{
		auto ptr = std::launder(reinterpret_cast<T *>(buffer_)); // (1)
		auto alias = std::shared_ptr<T>(this->shared_from_this(), ptr); // (2)
		::new (ptr) T(alias, std::forward<Args>(args)...); // (3)
		ptr_ = ptr;
		return alias;
	}

Здесь в точке (1) вы "отмываете" указатель и получаете якобы легальный указатель на объект типа T. Оставим вопрос о том, насколько легально делать это, если по факту в buffer_ на момент вызова launder не было корректно сформированного экземпляра типа T. Допустим, что ptr валиден и передача его в shared_ptr в точке (2) допустима.

Но в точке (3) вы переписываете содержимое buffer_, тем самым делая невалидными все предыдущие указатели, которые указывали на T внутри buffer_. Поэтому и сам ptr после вызова placement new перестает быть валидным. Чтобы он остался таковым вам нужно было бы написать, например, так:

ptr = ::new(buffer_) T(alias, std::forward<Args>(args)...);

Или же сделать еще один std::launder после завершения placement new:

::new (ptr) T(alias, std::forward<Args>(args)...); // (3)
ptr_ = std::launder(reinterpret_cast<T*>(buffer_));

Без этих действий в ptr_ окажется указатель, который является невалидным. По крайней мере в C++17.

Но даже если вы корректно "отмываете" значение для ptr_, то остается вопрос с тем, что в alias вы передали указатель, который перестал быть валидным после вызова placement new.

А это UB.

По крайней мере на уровне моего понимания того, как вся эта механика работает в C++17 (см. пояснения здесь). В C++20 вроде бы стало попроще, но в C++20 у меня познаний нет.

PS. Есть ощущение, что SharedHolder можно было бы сделать проще за счет использования std::optional<T> вместо buffer_ и ptr_.

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

Момент первый. Сочетание терминов "канал" и "издатель/подписчик". Я, пожалуй, в первый раз сталкиваюсь с ситуацией, когда эти термины используются совместно. Обычно "канал" -- это всего лишь один писатель и всего лишь один читатель. Никаких подписок, просто в канал кто-то пишет, кто-то читает (причем читает когда ему это удобно). Тогда как с "издатель/подписчик" обычно используется термин "топик". Издатель публикует данные в топик, подписчики подписываются на топик. Понятно, что вы, как автор библиотеки, можете выбирать любые термины, которые вам кажутся наиболее уместными. Но для стороннего читателя создается дополнительная когнитивная нагрузка для того, чтобы привыкнуть, что "канал" здесь, это не то же самое, к чему читатель привык.

Момент второй. Вы в описании достаточно часто начинаете говорить о мутексах или weak_ptr. Для вас, как для автора, достаточно просто понимать что к чему и почему упоминания мутексов/weak_ptr важны. Но для того, кто впервые знакомиться с библиотекой, такие погружения в дебри картину происходящего только усложняют. Ведь сначала хочется понять что это и зачем, что может, как это выглядит. А уже когда общее понимание достигнуто, тогда можно и про какие-то потроха поговорить. Однако, у вас такого последовательного погружения (на мой взгляд) не получилось и вы начинаете нагружать читателя техническими подробностями слишком рано. Из-за этого после первого прочтения я вообще не понял зачем о мутексах и weak_ptr здесь нужно было говорить. Вроде как библиотека нужна чтобы освободить пользователя от всех этих низкоуровневых деталей, но нет, эти самые детали чуть ли не подчеркиваются специально. ИМХО, вполне можно было бы обойтись и без этого.

Если позволите, маленький совет.

Сейчас у вас обработчики получают два параметра связанных с информацией о контексте исполнения:

hi::SubscriptionCallback<std::string>::create(
		[&](std::string publication,
        const std::atomic<std::uint32_t> & global_run_id,
        const std::uint32_t your_run_id)
		{

Я бы предложил объединить оба этих параметра в одну структуру и передавать ссылку на нее (или отдавать ее по значению). Что-то вроде:

struct ExecContextInfo {
  std::reference_wrapper<const std::atomic<std::uint32_t>> global_run_id_;
  std::uint32_t your_run_id_;
  ...
};

hi::SubscriptionCallback<std::string>::create(
		[&](std::string publication,
        ExecContextInfo exec_ctx)
		{

В этом случае прототипы обработчиков бы упростились и пользователям пришлось бы иметь дело с меньшим количеством бойлерплейта.

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

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

std::string str{"worker"};
str.append(std::to_string(id));

std::string{str}.append(std::to_string(i))

Тогда как привычнее видеть код вроде вот такого:

std::string str{"worker" + std::to_string(id)};

str + std::to_string(i)

Вроде как разницы по производительности между двумя этими способами быть не должно. Есть ли какие-то невидимые мне преимущества у той формы, которую вы используете?