cppcoro
1.0.0
A biblioteca 'cppcoro' fornece um grande conjunto de primitivas de uso geral para fazer uso da proposta de coroutinas TS descrita no N4680.
Estes incluem:
task<T>shared_task<T>generator<T>recursive_generator<T>async_generator<T>single_consumer_eventsingle_consumer_async_auto_reset_eventasync_mutexasync_manual_reset_eventasync_auto_reset_eventasync_latchsequence_barriermulti_producer_sequencersingle_producer_sequencersync_wait()when_all()when_all_ready()fmap()schedule_on()resume_on()cancellation_tokencancellation_sourcecancellation_registrationstatic_thread_poolio_service e io_work_scopefile , readable_file , writable_fileread_only_file , write_only_file , read_write_filesocketip_address , ipv4_address , ipv6_addressip_endpoint , ipv4_endpoint , ipv6_endpointis_awaitable<T>awaitable_traits<T>Awaitable<T>Awaiter<T>SchedulerDelayedSchedulerEsta biblioteca é uma biblioteca experimental que está explorando o espaço de abstrações de programação assíncrona de alto desempenho e escalonável que podem ser construídas sobre a proposta de coroutinas C ++.
Foi de origem aberta na esperança de que outros achem útil e que a comunidade C ++ possa fornecer feedback sobre ela e maneiras de melhorá-lo.
Requer um compilador que suporta as coroutinas TS:
A versão Linux é funcional, exceto para as classes relacionadas io_context e E/S de arquivo que ainda não foram implementadas para o Linux (consulte a edição 15 para obter mais informações).
task<T>Uma tarefa representa um cálculo assíncrono que é executado preguiçosamente, pois a execução da coroutina não começa até que a tarefa seja aguardada.
Exemplo:
# include < cppcoro/read_only_file.hpp >
# include < cppcoro/task.hpp >
cppcoro::task< int > count_lines (std::string path)
{
auto file = co_await cppcoro::read_only_file::open (path);
int lineCount = 0 ;
char buffer[ 1024 ];
size_t bytesRead;
std:: uint64_t offset = 0 ;
do
{
bytesRead = co_await file. read (offset, buffer, sizeof (buffer));
lineCount += std::count (buffer, buffer + bytesRead, ' n ' );
offset += bytesRead;
} while (bytesRead > 0 );
co_return lineCount;
}
cppcoro::task<> usage_example ()
{
// Calling function creates a new task but doesn't start
// executing the coroutine yet.
cppcoro::task< int > countTask = count_lines ( " foo.txt " );
// ...
// Coroutine is only started when we later co_await the task.
int lineCount = co_await countTask;
std::cout << " line count = " << lineCount << std::endl;
}Visão geral da API:
// <cppcoro/task.hpp>
namespace cppcoro
{
template < typename T>
class task
{
public:
using promise_type = <unspecified>;
using value_type = T;
task () noexcept ;
task (task&& other) noexcept ;
task& operator =(task&& other);
// task is a move-only type.
task ( const task& other) = delete ;
task& operator =( const task& other) = delete ;
// Query if the task result is ready.
bool is_ready () const noexcept ;
// Wait for the task to complete and return the result or rethrow the
// exception if the operation completed with an unhandled exception.
//
// If the task is not yet ready then the awaiting coroutine will be
// suspended until the task completes. If the the task is_ready() then
// this operation will return the result synchronously without suspending.
Awaiter<T&> operator co_await () const & noexcept ;
Awaiter<T&&> operator co_await () const && noexcept ;
// Returns an awaitable that can be co_await'ed to suspend the current
// coroutine until the task completes.
//
// The 'co_await t.when_ready()' expression differs from 'co_await t' in
// that when_ready() only performs synchronization, it does not return
// the result or rethrow the exception.
//
// This can be useful if you want to synchronize with the task without
// the possibility of it throwing an exception.
Awaitable< void > when_ready () const noexcept ;
};
template < typename T>
void swap (task<T>& a, task<T>& b);
// Creates a task that yields the result of co_await'ing the specified awaitable.
//
// This can be used as a form of type-erasure of the concrete awaitable, allowing
// different awaitables that return the same await-result type to be stored in
// the same task<RESULT> type.
template <
typename AWAITABLE,
typename RESULT = typename awaitable_traits<AWAITABLE>:: await_result_t >
task<RESULT> make_task (AWAITABLE awaitable);
} Você pode criar um objeto task<T> , chamando uma função de coroutina que retorna uma task<T> .
A coroutina deve conter um uso de co_await ou co_return . Observe que uma task<T> coroutine não pode usar a palavra -chave co_yield .
Quando uma coroutina que retorna uma task<T> é chamada, um quadro de coroutina é alocado, se necessário, e os parâmetros são capturados no quadro da coroutina. A coroutina é suspensa no início do corpo da coroutina e a execução é retornada ao chamador e um valor task<T> que representa o cálculo assíncrono é retornado da chamada de função.
O corpo da coroutina começará a executar quando o valor task<T> for co_await ed. Isso suspenderá a coroutina aguardando e iniciará a execução da coroutina associada ao valor da task<T> aguardada.
A coroutina aguardando será posteriormente retomada no thread que concluir a execução da corota da task<T> aguardada. ou seja. O encadeamento que executa o co_return ou que lança uma exceção não atendida que encerra a execução da coroutina.
Se a tarefa já tiver executado até a conclusão, aguardá-la novamente, obterá o resultado já computado sem suspender a coroutina aguardando.
Se o objeto task for destruído antes de ser aguardado, a coroutina nunca é executada e o destruidor simplesmente destruir os parâmetros capturados e libera qualquer memória usada pelo quadro da coroutina.
shared_task<T> A classe shared_task<T> é um tipo de corotação que produz um único valor de forma assíncrona.
É 'preguiçoso' nessa execução da tarefa não começa até que seja aguardada por alguma corootina.
É 'compartilhado', pois o valor da tarefa pode ser copiado, permitindo várias referências ao resultado da tarefa a ser criada. Ele também permite que várias coroutinas aguardem simultaneamente o resultado.
A tarefa começará a executar no thread que primeiro co_await é a tarefa. Os aguardadores subsequentes serão suspensos e são filmados para retomada quando a tarefa for concluída ou continuará de maneira síncrona se a tarefa já tiver executado até a conclusão.
Se um aguardo for suspenso enquanto aguarda a conclusão da tarefa, ela será retomada no thread que concluir a execução da tarefa. ou seja. O encadeamento que executa o co_return ou que lança a exceção não atendida que encerra a execução da coroutina.
Resumo da API
namespace cppcoro
{
template < typename T = void >
class shared_task
{
public:
using promise_type = <unspecified>;
using value_type = T;
shared_task () noexcept ;
shared_task ( const shared_task& other) noexcept ;
shared_task (shared_task&& other) noexcept ;
shared_task& operator =( const shared_task& other) noexcept ;
shared_task& operator =(shared_task&& other) noexcept ;
void swap (shared_task& other) noexcept ;
// Query if the task has completed and the result is ready.
bool is_ready () const noexcept ;
// Returns an operation that when awaited will suspend the
// current coroutine until the task completes and the result
// is available.
//
// The type of the result of the 'co_await someTask' expression
// is an l-value reference to the task's result value (unless T
// is void in which case the expression has type 'void').
// If the task completed with an unhandled exception then the
// exception will be rethrown by the co_await expression.
Awaiter<T&> operator co_await () const noexcept ;
// Returns an operation that when awaited will suspend the
// calling coroutine until the task completes and the result
// is available.
//
// The result is not returned from the co_await expression.
// This can be used to synchronize with the task without the
// possibility of the co_await expression throwing an exception.
Awaiter< void > when_ready () const noexcept ;
};
template < typename T>
bool operator ==( const shared_task<T>& a, const shared_task<T>& b) noexcept ;
template < typename T>
bool operator !=( const shared_task<T>& a, const shared_task<T>& b) noexcept ;
template < typename T>
void swap (shared_task<T>& a, shared_task<T>& b) noexcept ;
// Wrap an awaitable value in a shared_task to allow multiple coroutines
// to concurrently await the result.
template <
typename AWAITABLE,
typename RESULT = typename awaitable_traits<AWAITABLE>:: await_result_t >
shared_task<RESULT> make_shared_task (AWAITABLE awaitable);
} Todos os métodos const const em shared_task<T> são seguros de ligar simultaneamente com outros métodos const na mesma instância de vários threads. Não é seguro chamar métodos que não sejam consconst de shared_task<T> simultaneamente com qualquer outro método na mesma instância de um shared_task<T> .
task<T> A classe shared_task<T> é semelhante à task<T> , pois a tarefa não inicia a execução imediatamente da função coroutina que está sendo chamada. A tarefa começa a executar apenas quando é aguardada pela primeira vez.
Difere da task<T> em que o objeto de tarefa resultante pode ser copiado, permitindo que vários objetos de tarefa referenciem o mesmo resultado assíncrono. Ele também suporta várias coroutinas aguardando simultaneamente o resultado da tarefa.
O trade-off é que o resultado é sempre uma referência de valor L ao resultado, nunca uma referência de valor R (já que o resultado pode ser compartilhado) que pode limitar a capacidade de se mover o resultado em uma variável local. Ele também tem um custo de tempo de execução um pouco mais alto devido à necessidade de manter uma contagem de referência e apoiar vários aguardadores.
generator<T> Um generator representa um tipo de coroutina que produz uma sequência de valores do tipo, T , onde os valores são produzidos preguiçosamente e de síncrona.
O corpo da coroutina é capaz de produzir valores do tipo T usando a palavra -chave co_yield . Observe, no entanto, que o corpo da coroutina não pode usar a palavra -chave co_await ; Os valores devem ser produzidos de maneira síncrona.
Por exemplo:
cppcoro::generator< const std:: uint64_t > fibonacci ()
{
std:: uint64_t a = 0 , b = 1 ;
while ( true )
{
co_yield b;
auto tmp = a;
a = b;
b += tmp;
}
}
void usage ()
{
for ( auto i : fibonacci ())
{
if (i > 1'000'000 ) break ;
std::cout << i << std::endl;
}
} Quando uma função de coroutina retornando um generator<T> é chamada de coroutina é criada inicialmente suspensa. A execução da coroutina entra no corpo da coroutina quando o método generator<T>::begin() é chamado e continua até que a primeira instrução co_yield seja atingida ou a coroartina é concluída.
Se o iterador retornado não for igual ao iterador end() a desreferência do iterador retornará uma referência ao valor passado para a instrução co_yield .
operator++() no iterador retomará a execução da coroutina e continuará até que o próximo ponto co_yield seja atingido ou a coroutina execute para conclusão ().
Quaisquer exceções não tratadas lançadas pela coroutina se propagam das chamadas begin() ou operator++() para o chamador.
Resumo da API:
namespace cppcoro
{
template < typename T>
class generator
{
public:
using promise_type = <unspecified>;
class iterator
{
public:
using iterator_category = std::input_iterator_tag;
using value_type = std:: remove_reference_t <T>;
using reference = value_type&;
using pointer = value_type*;
using difference_type = std:: size_t ;
iterator ( const iterator& other) noexcept ;
iterator& operator =( const iterator& other) noexcept ;
// If the generator coroutine throws an unhandled exception before producing
// the next element then the exception will propagate out of this call.
iterator& operator ++();
reference operator *() const noexcept ;
pointer operator ->() const noexcept ;
bool operator ==( const iterator& other) const noexcept ;
bool operator !=( const iterator& other) const noexcept ;
};
// Constructs to the empty sequence.
generator () noexcept ;
generator (generator&& other) noexcept ;
generator& operator =(generator&& other) noexcept ;
generator ( const generator& other) = delete ;
generator& operator =( const generator&) = delete ;
~generator ();
// Starts executing the generator coroutine which runs until either a value is yielded
// or the coroutine runs to completion or an unhandled exception propagates out of the
// the coroutine.
iterator begin ();
iterator end () noexcept ;
// Swap the contents of two generators.
void swap (generator& other) noexcept ;
};
template < typename T>
void swap (generator<T>& a, generator<T>& b) noexcept ;
// Apply function, func, lazily to each element of the source generator
// and yield a sequence of the results of calls to func().
template < typename FUNC, typename T>
generator<std:: invoke_result_t <FUNC, T&>> fmap (FUNC func, generator<T> source);
}recursive_generator<T> Um recursive_generator é semelhante a um generator , exceto que ele foi projetado para suportar com mais eficiência, produzindo os elementos de uma sequência aninhada como elementos de uma sequência externa.
Além de poder co_yield um valor do tipo T você também pode co_yield um valor do tipo recursive_generator<T> .
Quando você co_yield um recursive_generator<T> Valor, todos os elementos do gerador produzido são produzidos como elementos do gerador atual. A coroutina atual é suspensa até que o consumidor termine de consumir todos os elementos do gerador aninhado, após o que a execução do ponto da coroutina atual retomará a execução para produzir o próximo elemento.
O benefício do recursive_generator<T> sobre generator<T> para iterar sobre estruturas de dados recursivas é que o iterator::operator++() é capaz de retomar diretamente as corotas da folha para produzir o próximo elemento, em vez de ter que retomar/suspender o (profundidade) coroares para cada elemento. O lado baixo é que há uma sobrecarga adicional
Por exemplo:
// Lists the immediate contents of a directory.
cppcoro::generator<dir_entry> list_directory (std::filesystem::path path);
cppcoro::recursive_generator<dir_entry> list_directory_recursive (std::filesystem::path path)
{
for ( auto & entry : list_directory (path))
{
co_yield entry;
if (entry. is_directory ())
{
co_yield list_directory_recursive (entry. path ());
}
}
} Observe que a aplicação do operador fmap() a um recursive_generator<T> produzirá um tipo de generator<U> em vez de um recursive_generator<U> . Isso ocorre porque os usos do fmap geralmente não são usados em contextos recursivos e tentamos evitar a sobrecarga extra incorrida pelo recursive_generator .
async_generator<T> Um async_generator representa um tipo de coroutina que produz uma sequência de valores do tipo, T , onde os valores são produzidos preguiçosamente e os valores podem ser produzidos de forma assíncrona.
O corpo da coroutina é capaz de usar expressões co_await e co_yield .
Os consumidores do gerador podem usar um loop for-loop baseado em gama for co_await para consumir os valores.
Exemplo
cppcoro::async_generator< int > ticker ( int count, threadpool& tp)
{
for ( int i = 0 ; i < count; ++i)
{
co_await tp. delay ( std::chrono::seconds ( 1 ));
co_yield i;
}
}
cppcoro::task<> consumer (threadpool& tp)
{
auto sequence = ticker ( 10 , tp);
for co_await (std:: uint32_t i : sequence)
{
std::cout << " Tick " << i << std::endl;
}
}Resumo da API
// <cppcoro/async_generator.hpp>
namespace cppcoro
{
template < typename T>
class async_generator
{
public:
class iterator
{
public:
using iterator_tag = std::forward_iterator_tag;
using difference_type = std:: size_t ;
using value_type = std:: remove_reference_t <T>;
using reference = value_type&;
using pointer = value_type*;
iterator ( const iterator& other) noexcept ;
iterator& operator =( const iterator& other) noexcept ;
// Resumes the generator coroutine if suspended
// Returns an operation object that must be awaited to wait
// for the increment operation to complete.
// If the coroutine runs to completion then the iterator
// will subsequently become equal to the end() iterator.
// If the coroutine completes with an unhandled exception then
// that exception will be rethrown from the co_await expression.
Awaitable<iterator&> operator ++() noexcept ;
// Dereference the iterator.
pointer operator ->() const noexcept ;
reference operator *() const noexcept ;
bool operator ==( const iterator& other) const noexcept ;
bool operator !=( const iterator& other) const noexcept ;
};
// Construct to the empty sequence.
async_generator () noexcept ;
async_generator ( const async_generator&) = delete ;
async_generator (async_generator&& other) noexcept ;
~async_generator ();
async_generator& operator =( const async_generator&) = delete ;
async_generator& operator =(async_generator&& other) noexcept ;
void swap (async_generator& other) noexcept ;
// Starts execution of the coroutine and returns an operation object
// that must be awaited to wait for the first value to become available.
// The result of co_await'ing the returned object is an iterator that
// can be used to advance to subsequent elements of the sequence.
//
// This method is not valid to be called once the coroutine has
// run to completion.
Awaitable<iterator> begin () noexcept ;
iterator end () noexcept ;
};
template < typename T>
void swap (async_generator<T>& a, async_generator<T>& b);
// Apply 'func' to each element of the source generator, yielding a sequence of
// the results of calling 'func' on the source elements.
template < typename FUNC, typename T>
async_generator<std:: invoke_result_t <FUNC, T&>> fmap (FUNC func, async_generator<T> source);
} Quando o objeto async_generator é destruído, solicita o cancelamento da coroutina subjacente. Se a coroutina já foi concluída ou está atualmente suspensa em uma expressão co_yield , a coroutina será destruída imediatamente. Caso contrário, a coroutina continuará a execução até que seja executada até a conclusão ou atinja a próxima expressão co_yield .
Quando o quadro da coroutina é destruído, os destruidores de todas as variáveis no escopo nesse ponto serão executados para garantir que os recursos do gerador sejam limpos.
Observe que o chamador deve garantir que o objeto async_generator não deve ser destruído enquanto uma coroutina de consumidor estiver executando uma expressão co_await esperando o próximo item ser produzido.
single_consumer_eventEste é um tipo simples de evento manual-reset que suporta apenas uma única coroutina aguardando-a de cada vez. Isso pode ser usado para
Resumo da API:
// <cppcoro/single_consumer_event.hpp>
namespace cppcoro
{
class single_consumer_event
{
public:
single_consumer_event ( bool initiallySet = false ) noexcept ;
bool is_set () const noexcept ;
void set ();
void reset () noexcept ;
Awaiter< void > operator co_await () const noexcept ;
};
}Exemplo:
# include < cppcoro/single_consumer_event.hpp >
cppcoro::single_consumer_event event;
std::string value;
cppcoro::task<> consumer ()
{
// Coroutine will suspend here until some thread calls event.set()
// eg. inside the producer() function below.
co_await event;
std::cout << value << std::endl;
}
void producer ()
{
value = " foo " ;
// This will resume the consumer() coroutine inside the call to set()
// if it is currently suspended.
event. set ();
}single_consumer_async_auto_reset_event Esta classe fornece uma primitiva de sincronização assíncrona que permite que uma única coroutina aguarde até que o evento seja sinalizado por uma chamada para o método set() .
Uma vez que a coroutina aguardando o evento é lançada por uma chamada anterior ou subsequente para set() o evento é redefinido automaticamente de volta ao estado 'não definido'.
Esta classe é uma versão mais eficiente do async_auto_reset_event que pode ser usado nos casos em que apenas uma única coroutina aguarda o evento por vez. Se você precisar oferecer suporte a várias coroutinas simultâneas que aguardam o evento, use a classe async_auto_reset_event .
Resumo da API:
// <cppcoro/single_consumer_async_auto_reset_event.hpp>
namespace cppcoro
{
class single_consumer_async_auto_reset_event
{
public:
single_consumer_async_auto_reset_event (
bool initiallySet = false ) noexcept ;
// Change the event to the 'set' state. If a coroutine is awaiting the
// event then the event is immediately transitioned back to the 'not set'
// state and the coroutine is resumed.
void set () noexcept ;
// Returns an Awaitable type that can be awaited to wait until
// the event becomes 'set' via a call to the .set() method. If
// the event is already in the 'set' state then the coroutine
// continues without suspending.
// The event is automatically reset back to the 'not set' state
// before resuming the coroutine.
Awaiter< void > operator co_await () const noexcept ;
};
}Exemplo de uso:
std::atomic< int > value;
cppcoro::single_consumer_async_auto_reset_event valueDecreasedEvent;
cppcoro::task<> wait_until_value_is_below ( int limit)
{
while (value. load (std::memory_order_relaxed) >= limit)
{
// Wait until there has been some change that we're interested in.
co_await valueDecreasedEvent;
}
}
void change_value ( int delta)
{
value. fetch_add (delta, std::memory_order_relaxed);
// Notify the waiter if there has been some change.
if (delta < 0 ) valueDecreasedEvent. set ();
}async_mutexFornece uma abstração de exclusão mútua simples que permite ao chamador 'co_await' o mutex de dentro de uma coroutina para suspender a coroutina até que o bloqueio mutex seja adquirido.
A implementação é livre de bloqueio, na medida em que uma coroutina que aguarda o mutex não bloqueará o thread, mas suspenderá a coroutina e posteriormente a retomará dentro da chamada para unlock() pelo seguidor anterior.
Resumo da API:
// <cppcoro/async_mutex.hpp>
namespace cppcoro
{
class async_mutex_lock ;
class async_mutex_lock_operation ;
class async_mutex_scoped_lock_operation ;
class async_mutex
{
public:
async_mutex () noexcept ;
~async_mutex ();
async_mutex ( const async_mutex&) = delete ;
async_mutex& operator ( const async_mutex&) = delete;
bool try_lock () noexcept ;
async_mutex_lock_operation lock_async () noexcept ;
async_mutex_scoped_lock_operation scoped_lock_async () noexcept ;
void unlock ();
};
class async_mutex_lock_operation
{
public:
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter) noexcept ;
void await_resume () const noexcept ;
};
class async_mutex_scoped_lock_operation
{
public:
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter) noexcept ;
[[nodiscard]] async_mutex_lock await_resume () const noexcept ;
};
class async_mutex_lock
{
public:
// Takes ownership of the lock.
async_mutex_lock (async_mutex& mutex, std:: adopt_lock_t ) noexcept ;
// Transfer ownership of the lock.
async_mutex_lock (async_mutex_lock&& other) noexcept ;
async_mutex_lock ( const async_mutex_lock&) = delete ;
async_mutex_lock& operator =( const async_mutex_lock&) = delete ;
// Releases the lock by calling unlock() on the mutex.
~async_mutex_lock ();
};
}Exemplo de uso:
# include < cppcoro/async_mutex.hpp >
# include < cppcoro/task.hpp >
# include < set >
# include < string >
cppcoro::async_mutex mutex;
std::set<std::string> values;
cppcoro::task<> add_item (std::string value)
{
cppcoro::async_mutex_lock lock = co_await mutex. scoped_lock_async ();
values. insert ( std::move (value));
}async_manual_reset_event Um evento de reset manual é um primitivo de coroutina/sincronização de threads que permite que um ou mais threads aguardem até que o evento seja sinalizado por um thread que chama set() .
O evento está em um dos dois estados; 'Set' e 'Not Set' .
Se o evento estiver no estado de 'conjunto' quando uma coroutina aguarda o evento, a coroutina continua sem suspender. No entanto, se a coroutina estiver no estado 'não definido' , a coroutina será suspensa até que algum thread chama posteriormente o método set() .
Quaisquer threads suspensos enquanto aguardam o evento se tornarem 'conjunto' será retomado dentro da próxima chamada para set() por algum thread.
Observe que você deve garantir que nenhuma coroutina esteja aguardando um evento 'não definido' quando o evento for destruído, pois não serão retomados.
Exemplo:
cppcoro::async_manual_reset_event event;
std::string value;
void producer ()
{
value = get_some_string_value ();
// Publish a value by setting the event.
event. set ();
}
// Can be called many times to create many tasks.
// All consumer tasks will wait until value has been published.
cppcoro::task<> consumer ()
{
// Wait until value has been published by awaiting event.
co_await event;
consume_value (value);
}Resumo da API:
namespace cppcoro
{
class async_manual_reset_event_operation ;
class async_manual_reset_event
{
public:
async_manual_reset_event ( bool initiallySet = false ) noexcept ;
~async_manual_reset_event ();
async_manual_reset_event ( const async_manual_reset_event&) = delete ;
async_manual_reset_event (async_manual_reset_event&&) = delete ;
async_manual_reset_event& operator =( const async_manual_reset_event&) = delete ;
async_manual_reset_event& operator =(async_manual_reset_event&&) = delete ;
// Wait until the event becomes set.
async_manual_reset_event_operation operator co_await () const noexcept ;
bool is_set () const noexcept ;
void set () noexcept ;
void reset () noexcept ;
};
class async_manual_reset_event_operation
{
public:
async_manual_reset_event_operation (async_manual_reset_event& event) noexcept ;
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter) noexcept ;
void await_resume () const noexcept ;
};
}async_auto_reset_event Um evento de retenção automática é uma primitiva de coroutina/sincronização de threads que permite que um ou mais threads esperem até que o evento seja sinalizado por um thread, chamando set() .
Uma vez que uma coroutina aguardando o evento é lançada por uma chamada anterior ou subsequente para set() o evento é redefinido automaticamente de volta ao estado 'não definido'.
Resumo da API:
// <cppcoro/async_auto_reset_event.hpp>
namespace cppcoro
{
class async_auto_reset_event_operation ;
class async_auto_reset_event
{
public:
async_auto_reset_event ( bool initiallySet = false ) noexcept ;
~async_auto_reset_event ();
async_auto_reset_event ( const async_auto_reset_event&) = delete ;
async_auto_reset_event (async_auto_reset_event&&) = delete ;
async_auto_reset_event& operator =( const async_auto_reset_event&) = delete ;
async_auto_reset_event& operator =(async_auto_reset_event&&) = delete ;
// Wait for the event to enter the 'set' state.
//
// If the event is already 'set' then the event is set to the 'not set'
// state and the awaiting coroutine continues without suspending.
// Otherwise, the coroutine is suspended and later resumed when some
// thread calls 'set()'.
//
// Note that the coroutine may be resumed inside a call to 'set()'
// or inside another thread's call to 'operator co_await()'.
async_auto_reset_event_operation operator co_await () const noexcept ;
// Set the state of the event to 'set'.
//
// If there are pending coroutines awaiting the event then one
// pending coroutine is resumed and the state is immediately
// set back to the 'not set' state.
//
// This operation is a no-op if the event was already 'set'.
void set () noexcept ;
// Set the state of the event to 'not-set'.
//
// This is a no-op if the state was already 'not set'.
void reset () noexcept ;
};
class async_auto_reset_event_operation
{
public:
explicit async_auto_reset_event_operation (async_auto_reset_event& event) noexcept ;
async_auto_reset_event_operation ( const async_auto_reset_event_operation& other) noexcept ;
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter) noexcept ;
void await_resume () const noexcept ;
};
}async_latchUma trava assíncrona é uma primitiva de sincronização que permite que os coroutinas esperem assíncronos até que um contador seja diminuído a zero.
A trava é um objeto de uso único. Quando o balcão atingir zero, a trava ficará 'pronta' e permanecerá pronta até que a trava seja destruída.
Resumo da API:
// <cppcoro/async_latch.hpp>
namespace cppcoro
{
class async_latch
{
public:
// Initialise the latch with the specified count.
async_latch (std:: ptrdiff_t initialCount) noexcept ;
// Query if the count has reached zero yet.
bool is_ready () const noexcept ;
// Decrement the count by n.
// This will resume any waiting coroutines if the count reaches zero
// as a result of this call.
// It is undefined behaviour to decrement the count below zero.
void count_down (std:: ptrdiff_t n = 1 ) noexcept ;
// Wait until the latch becomes ready.
// If the latch count is not yet zero then the awaiting coroutine will
// be suspended and later resumed by a call to count_down() that decrements
// the count to zero. If the latch count was already zero then the coroutine
// continues without suspending.
Awaiter< void > operator co_await () const noexcept ;
};
}sequence_barrier Um sequence_barrier é um primitivo de sincronização que permite que um produtor único e vários consumidores coordenem em relação a um número de sequência que cresce monotonicamente.
Um único produtor avança o número da sequência publicando novos números de sequência em uma ordem monotonicamente crescente. Um ou mais consumidores podem consultar o último número de sequência publicado e podem esperar até que um número de sequência específico seja publicado.
Uma barreira de sequência pode ser usada para representar um cursor em um produtor/buffer de anel de consumo seguro para roscas
Veja o padrão LMAX Disruptor para mais fundo: https://lmax-exchange.github.io/disruptor/files/disruptor-1.0.pdf
Sinopse da API:
namespace cppcoro
{
template < typename SEQUENCE = std:: size_t ,
typename TRAITS = sequence_traits<SEQUENCE>>
class sequence_barrier
{
public:
sequence_barrier (SEQUENCE initialSequence = TRAITS::initial_sequence) noexcept ;
~sequence_barrier ();
SEQUENCE last_published () const noexcept ;
// Wait until the specified targetSequence number has been published.
//
// If the operation does not complete synchronously then the awaiting
// coroutine is resumed on the specified scheduler. Otherwise, the
// coroutine continues without suspending.
//
// The co_await expression resumes with the updated last_published()
// value, which is guaranteed to be at least 'targetSequence'.
template < typename SCHEDULER>
[[nodiscard]]
Awaitable<SEQUENCE> wait_until_published (SEQUENCE targetSequence,
SCHEDULER& scheduler) const noexcept ;
void publish (SEQUENCE sequence) noexcept ;
};
}single_producer_sequencer Um single_producer_sequencer é um primitivo de sincronização que pode ser usado para coordenar o acesso a um buffer de anel para um único produtor e um ou mais consumidores.
Um produtor primeiro adquire um ou mais slots em um buffer de anel, grava nos elementos de buffer de anel correspondentes a esses slots e, finalmente, publica os valores gravados nesses slots. Um produtor nunca pode produzir mais do que 'buffersize' elementos antes de onde o consumidor consumiu.
Um consumidor aguarda que certos elementos sejam publicados, processa os itens e notifica o produtor quando terminar o processamento de itens publicando o número da sequência que terminou de consumir em um objeto sequence_barrier .
Sinopse da API:
// <cppcoro/single_producer_sequencer.hpp>
namespace cppcoro
{
template <
typename SEQUENCE = std:: size_t ,
typename TRAITS = sequence_traits<SEQUENCE>>
class single_producer_sequencer
{
public:
using size_type = typename sequence_range<SEQUENCE, TRAITS>::size_type;
single_producer_sequencer (
const sequence_barrier<SEQUENCE, TRAITS>& consumerBarrier,
std:: size_t bufferSize,
SEQUENCE initialSequence = TRAITS::initial_sequence) noexcept ;
// Publisher API:
template < typename SCHEDULER>
[[nodiscard]]
Awaitable<SEQUENCE> claim_one (SCHEDULER& scheduler) noexcept ;
template < typename SCHEDULER>
[[nodiscard]]
Awaitable<sequence_range<SEQUENCE>> claim_up_to (
std:: size_t count,
SCHEDULER& scheduler) noexcept ;
void publish (SEQUENCE sequence) noexcept ;
// Consumer API:
SEQUENCE last_published () const noexcept ;
template < typename SCHEDULER>
[[nodiscard]]
Awaitable<SEQUENCE> wait_until_published (
SEQUENCE targetSequence,
SCHEDULER& scheduler) const noexcept ;
};
}Exemplo de uso:
using namespace cppcoro ;
using namespace std ::chrono ;
struct message
{
int id;
steady_clock::time_point timestamp;
float data;
};
constexpr size_t bufferSize = 16384 ; // Must be power-of-two
constexpr size_t indexMask = bufferSize - 1 ;
message buffer[bufferSize];
task< void > producer (
io_service& ioSvc,
single_producer_sequencer< size_t >& sequencer)
{
auto start = steady_clock::now ();
for ( int i = 0 ; i < 1'000'000 ; ++i)
{
// Wait until a slot is free in the buffer.
size_t seq = co_await sequencer. claim_one (ioSvc);
// Populate the message.
auto & msg = buffer[seq & indexMask];
msg. id = i;
msg. timestamp = steady_clock::now ();
msg. data = 123 ;
// Publish the message.
sequencer. publish (seq);
}
// Publish a sentinel
auto seq = co_await sequencer. claim_one (ioSvc);
auto & msg = buffer[seq & indexMask];
msg. id = - 1 ;
sequencer. publish (seq);
}
task< void > consumer (
static_thread_pool& threadPool,
const single_producer_sequencer< size_t >& sequencer,
sequence_barrier< size_t >& consumerBarrier)
{
size_t nextToRead = 0 ;
while ( true )
{
// Wait until the next message is available
// There may be more than one available.
const size_t available = co_await sequencer. wait_until_published (nextToRead, threadPool);
do {
auto & msg = buffer[nextToRead & indexMask];
if (msg. id == - 1 )
{
consumerBarrier. publish (nextToRead);
co_return ;
}
processMessage (msg);
} while (nextToRead++ != available);
// Notify the producer that we've finished processing
// up to 'nextToRead - 1'.
consumerBarrier. publish (available);
}
}
task< void > example (io_service& ioSvc, static_thread_pool& threadPool)
{
sequence_barrier< size_t > barrier;
single_producer_sequencer< size_t > sequencer{barrier, bufferSize};
co_await when_all (
producer (tp, sequencer),
consumer (tp, sequencer, barrier));
}multi_producer_sequencer A classe multi_producer_sequencer é uma primitiva de sincronização que coordena o acesso a um buffer de anel para vários produtores e um ou mais consumidores.
Para uma variante de produtora única, consulte a classe single_producer_sequencer .
Observe que o buffer de anel deve ter um tamanho que seja um poder de dois. Isso ocorre porque a implementação usa máscaras de bits em vez de divisão/módulo inteiro para calcular o deslocamento no buffer. Além disso, isso permite que o número da sequência envolva com segurança o valor de 32 bits/64 bits.
Resumo da API:
// <cppcoro/multi_producer_sequencer.hpp>
namespace cppcoro
{
template < typename SEQUENCE = std:: size_t ,
typename TRAITS = sequence_traits<SEQUENCE>>
class multi_producer_sequencer
{
public:
multi_producer_sequencer (
const sequence_barrier<SEQUENCE, TRAITS>& consumerBarrier,
SEQUENCE initialSequence = TRAITS::initial_sequence);
std:: size_t buffer_size () const noexcept ;
// Consumer interface
//
// Each consumer keeps track of their own 'lastKnownPublished' value
// and must pass this to the methods that query for an updated last-known
// published sequence number.
SEQUENCE last_published_after (SEQUENCE lastKnownPublished) const noexcept ;
template < typename SCHEDULER>
Awaitable<SEQUENCE> wait_until_published (
SEQUENCE targetSequence,
SEQUENCE lastKnownPublished,
SCHEDULER& scheduler) const noexcept ;
// Producer interface
// Query whether any slots available for claiming (approx.)
bool any_available () const noexcept ;
template < typename SCHEDULER>
Awaitable<SEQUENCE> claim_one (SCHEDULER& scheduler) noexcept ;
template < typename SCHEDULER>
Awaitable<sequence_range<SEQUENCE, TRAITS>> claim_up_to (
std:: size_t count,
SCHEDULER& scheduler) noexcept ;
// Mark the specified sequence number as published.
void publish (SEQUENCE sequence) noexcept ;
// Mark all sequence numbers in the specified range as published.
void publish ( const sequence_range<SEQUENCE, TRAITS>& range) noexcept ;
};
} Um cancellation_token é um valor que pode ser passado para uma função que permite ao chamador comunicar posteriormente uma solicitação para cancelar a operação dessa função.
Para obter um cancellation_token capaz de ser cancelado, você deve primeiro criar um objeto cancellation_source . O método cancellation_source::token() pode ser usado para fabricar novos valores de cancellation_token que estão vinculados a esse objeto cancellation_source .
Quando você deseja solicitar mais tarde o cancelamento de uma operação, você passou por um cancellation_token , pode chamar cancellation_source::request_cancellation() em um objeto cancellation_source associado.
As funções podem responder a um pedido de cancelamento de uma de duas maneiras:
cancellation_token::is_cancellation_requested() ou cancellation_token::throw_if_cancellation_requested() .cancellation_registration .Resumo da API:
namespace cppcoro
{
class cancellation_source
{
public:
// Construct a new, independently cancellable cancellation source.
cancellation_source ();
// Construct a new reference to the same cancellation state.
cancellation_source ( const cancellation_source& other) noexcept ;
cancellation_source (cancellation_source&& other) noexcept ;
~cancellation_source ();
cancellation_source& operator =( const cancellation_source& other) noexcept ;
cancellation_source& operator =(cancellation_source&& other) noexcept ;
bool is_cancellation_requested () const noexcept ;
bool can_be_cancelled () const noexcept ;
void request_cancellation ();
cancellation_token token () const noexcept ;
};
class cancellation_token
{
public:
// Construct a token that can't be cancelled.
cancellation_token () noexcept ;
cancellation_token ( const cancellation_token& other) noexcept ;
cancellation_token (cancellation_token&& other) noexcept ;
~cancellation_token ();
cancellation_token& operator =( const cancellation_token& other) noexcept ;
cancellation_token& operator =(cancellation_token&& other) noexcept ;
bool is_cancellation_requested () const noexcept ;
void throw_if_cancellation_requested () const ;
// Query if this token can ever have cancellation requested.
// Code can use this to take a more efficient code-path in cases
// that the operation does not need to handle cancellation.
bool can_be_cancelled () const noexcept ;
};
// RAII class for registering a callback to be executed if cancellation
// is requested on a particular cancellation token.
class cancellation_registration
{
public:
// Register a callback to be executed if cancellation is requested.
// Callback will be called with no arguments on the thread that calls
// request_cancellation() if cancellation is not yet requested, or
// called immediately if cancellation has already been requested.
// Callback must not throw an unhandled exception when called.
template < typename CALLBACK>
cancellation_registration (cancellation_token token, CALLBACK&& callback);
cancellation_registration ( const cancellation_registration& other) = delete ;
~cancellation_registration ();
};
class operation_cancelled : public std :: exception
{
public:
operation_cancelled ();
const char * what () const override ;
};
}Exemplo: abordagem de votação
cppcoro::task<> do_something_async (cppcoro::cancellation_token token)
{
// Explicitly define cancellation points within the function
// by calling throw_if_cancellation_requested().
token. throw_if_cancellation_requested ();
co_await do_step_1 ();
token. throw_if_cancellation_requested ();
do_step_2 ();
// Alternatively, you can query if cancellation has been
// requested to allow yourself to do some cleanup before
// returning.
if (token. is_cancellation_requested ())
{
display_message_to_user ( " Cancelling operation... " );
do_cleanup ();
throw cppcoro::operation_cancelled{};
}
do_final_step ();
}Exemplo: abordagem de retorno de chamada
// Say we already have a timer abstraction that supports being
// cancelled but it doesn't support cancellation_tokens natively.
// You can use a cancellation_registration to register a callback
// that calls the existing cancellation API. e.g.
cppcoro::task<> cancellable_timer_wait (cppcoro::cancellation_token token)
{
auto timer = create_timer (10s);
cppcoro::cancellation_registration registration (token, [&]
{
// Call existing timer cancellation API.
timer. cancel ();
});
co_await timer;
}static_thread_pool A classe static_thread_pool fornece uma abstração que permite agendar o trabalho em um conjunto de threads de tamanho fixo.
Esta classe implementa o conceito de agendador (veja abaixo).
Você pode envolver o trabalho no thread-pool executando co_await threadPool.schedule() . Esta operação suspenderá a coroutina atual, a prenderá para execução na piscina de threads e o pool de threads retomará a coroutina quando um fio na piscina de thread estiver livre para executar a coroutina. Esta operação é garantida para não jogar e, no caso comum, não alocará nenhuma memória .
Esta classe utiliza um algoritmo de roubo de trabalho para o trabalho de bala de carga em vários threads. O trabalho previsto para a piscina de thread a partir de um thread thread será agendado para execução no mesmo thread em uma fila LIFO. Trabalho envolvido na piscina de um fio remoto será inserido em uma fila global do FIFO. Quando um tópico de trabalhador fica sem trabalho a partir de sua fila local, ele primeiro tenta desacreditar o trabalho da fila global. Se essa fila estiver vazia, ela tentará roubar trabalhos da parte de trás das filas dos outros threads de trabalhadores.
Resumo da API:
namespace cppcoro
{
class static_thread_pool
{
public:
// Initialise the thread-pool with a number of threads equal to
// std::thread::hardware_concurrency().
static_thread_pool ();
// Initialise the thread pool with the specified number of threads.
explicit static_thread_pool (std:: uint32_t threadCount);
std:: uint32_t thread_count () const noexcept ;
class schedule_operation
{
public:
schedule_operation (static_thread_pool* tp) noexcept ;
bool await_ready () noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> h) noexcept ;
bool await_resume () noexcept ;
private:
// unspecified
};
// Return an operation that can be awaited by a coroutine.
//
//
[[nodiscard]]
schedule_operation schedule () noexcept ;
private:
// Unspecified
};
}Exemplo de uso: simples
cppcoro::task<std::string> do_something_on_threadpool (cppcoro::static_thread_pool& tp)
{
// First schedule the coroutine onto the threadpool.
co_await tp. schedule ();
// When it resumes, this coroutine is now running on the threadpool.
do_something ();
} Exemplo de uso: fazer as coisas em paralelo - usando o operador schedule_on() com static_thread_pool .
cppcoro::task< double > dot_product (static_thread_pool& tp, double a[], double b[], size_t count)
{
if (count > 1000 )
{
// Subdivide the work recursively into two equal tasks
// The first half is scheduled to the thread pool so it can run concurrently
// with the second half which continues on this thread.
size_t halfCount = count / 2 ;
auto [first, second] = co_await when_all (
schedule_on (tp, dot_product (tp, a, b, halfCount),
dot_product (tp, a + halfCount, b + halfCount, count - halfCount));
co_return first + second;
}
else
{
double sum = 0.0 ;
for ( size_t i = 0 ; i < count; ++i)
{
sum += a[i] * b[i];
}
co_return sum;
}
}io_service e io_work_scope A classe io_service fornece uma abstração para o processamento de eventos de conclusão de E/S a partir de operações de E/S assíncronas.
Quando uma operação de E/S assíncrona é concluída, a coroutina que aguardava essa operação será retomada em um encadeamento de E/S dentro de uma chamada para um dos métodos de processamento de eventos: process_events() , process_pending_events() , process_one_event() ou process_one_pending_event() .
A classe io_service não gerencia nenhum threads de E/S. Você deve garantir que algumas chamadas de threads um dos métodos de processamento de eventos para coroutinas que aguardam eventos de conclusão de E/S a serem despachadas. Este pode ser um thread dedicado que chama process_events() ou misturado com algum outro loop de eventos (por exemplo, um loop de evento da interface do usuário) pesquisando periodicamente para novos eventos por meio de uma chamada para process_pending_events() ou process_one_pending_event() .
Isso permite a integração do loop de eventos io_service com outros loops de eventos, como um loop de eventos de interface do usuário.
Você pode processamento multiplex de eventos em vários threads, com vários threads chamam process_events() . Você pode especificar uma dica quanto ao número máximo de threads para processar ativamente os eventos por meio de um parâmetro opcional de construtor io_service .
No Windows, a implementação utiliza as instalações da porta de conclusão de E/S do Windows para despachar eventos para threads de E/S de maneira escalável.
Resumo da API:
namespace cppcoro
{
class io_service
{
public:
class schedule_operation ;
class timed_schedule_operation ;
io_service ();
io_service (std:: uint32_t concurrencyHint);
io_service (io_service&&) = delete ;
io_service ( const io_service&) = delete ;
io_service& operator =(io_service&&) = delete ;
io_service& operator =( const io_service&) = delete ;
~io_service ();
// Scheduler methods
[[nodiscard]]
schedule_operation schedule () noexcept ;
template < typename REP, typename RATIO>
[[nodiscard]]
timed_schedule_operation schedule_after (
std::chrono::duration<REP, RATIO> delay,
cppcoro::cancellation_token cancellationToken = {}) noexcept ;
// Event-loop methods
//
// I/O threads must call these to process I/O events and execute
// scheduled coroutines.
std:: uint64_t process_events ();
std:: uint64_t process_pending_events ();
std:: uint64_t process_one_event ();
std:: uint64_t process_one_pending_event ();
// Request that all threads processing events exit their event loops.
void stop () noexcept ;
// Query if some thread has called stop()
bool is_stop_requested () const noexcept ;
// Reset the event-loop after a call to stop() so that threads can
// start processing events again.
void reset ();
// Reference-counting methods for tracking outstanding references
// to the io_service.
//
// The io_service::stop() method will be called when the last work
// reference is decremented.
//
// Use the io_work_scope RAII class to manage calling these methods on
// entry-to and exit-from a scope.
void notify_work_started () noexcept ;
void notify_work_finished () noexcept ;
};
class io_service ::schedule_operation
{
public:
schedule_operation ( const schedule_operation&) noexcept ;
schedule_operation& operator =( const schedule_operation&) noexcept ;
bool await_ready () const noexcept ;
void await_suspend (std::experimental::coroutine_handle<> awaiter) noexcept ;
void await_resume () noexcept ;
};
class io_service ::timed_schedule_operation
{
public:
timed_schedule_operation (timed_schedule_operation&&) noexcept ;
timed_schedule_operation ( const timed_schedule_operation&) = delete ;
timed_schedule_operation& operator =( const timed_schedule_operation&) = delete ;
timed_schedule_operation& operator =(timed_schedule_operation&&) = delete ;
bool await_ready () const noexcept ;
void await_suspend (std::experimental::coroutine_handle<> awaiter);
void await_resume ();
};
class io_work_scope
{
public:
io_work_scope (io_service& ioService) noexcept ;
io_work_scope ( const io_work_scope& other) noexcept ;
io_work_scope (io_work_scope&& other) noexcept ;
~io_work_scope ();
io_work_scope& operator =( const io_work_scope& other) noexcept ;
io_work_scope& operator =(io_work_scope&& other) noexcept ;
io_service& service () const noexcept ;
};
}Exemplo:
# include < cppcoro/task.hpp >
# include < cppcoro/task.hpp >
# include < cppcoro/io_service.hpp >
# include < cppcoro/read_only_file.hpp >
# include < experimental/filesystem >
# include < memory >
# include < algorithm >
# include < iostream >
namespace fs = std::experimental::filesystem;
cppcoro::task<std:: uint64_t > count_lines (cppcoro::io_service& ioService, fs::path path)
{
auto file = cppcoro::read_only_file::open (ioService, path);
constexpr size_t bufferSize = 4096 ;
auto buffer = std::make_unique<std:: uint8_t []>(bufferSize);
std:: uint64_t newlineCount = 0 ;
for (std:: uint64_t offset = 0 , fileSize = file. size (); offset < fileSize;)
{
const auto bytesToRead = static_cast < size_t >(
std::min<std:: uint64_t >(bufferSize, fileSize - offset));
const auto bytesRead = co_await file. read (offset, buffer. get (), bytesToRead);
newlineCount += std::count (buffer. get (), buffer. get () + bytesRead, ' n ' );
offset += bytesRead;
}
co_return newlineCount;
}
cppcoro::task<> run (cppcoro::io_service& ioService)
{
cppcoro::io_work_scope ioScope (ioService);
auto lineCount = co_await count_lines (ioService, fs::path{ " foo.txt " });
std::cout << " foo.txt has " << lineCount << " lines. " << std::endl;;
}
cppcoro::task<> process_events (cppcoro::io_service& ioService)
{
// Process events until the io_service is stopped.
// ie. when the last io_work_scope goes out of scope.
ioService. process_events ();
co_return ;
}
int main ()
{
cppcoro::io_service ioService;
cppcoro::sync_wait ( cppcoro::when_all_ready (
run (ioService),
process_events (ioService)));
return 0 ;
}io_service como um agendador Uma classe io_service implementa as interfaces para os conceitos Scheduler e DelayedScheduler .
Isso permite que uma coroutina suspenda a execução do encadeamento atual e agende -se para retomada em um encadeamento de E/S associado a um objeto io_service específico.
Exemplo:
cppcoro::task<> do_something (cppcoro::io_service& ioService)
{
// Coroutine starts execution on the thread of the task awaiter.
// A coroutine can transfer execution to an I/O thread by awaiting the
// result of io_service::schedule().
co_await ioService. schedule ();
// At this point, the coroutine is now executing on an I/O thread
// inside a call to one of the io_service event processing methods.
// A coroutine can also perform a delayed-schedule that will suspend
// the coroutine for a specified duration of time before scheduling
// it for resumption on an I/O thread.
co_await ioService. schedule_after (100ms);
// At this point, the coroutine is executing on a potentially different I/O thread.
}file , readable_file , writable_fileEsses tipos são classes base abstratas para realizar E/S de arquivo de concreto.
Resumo da API:
namespace cppcoro
{
class file_read_operation ;
class file_write_operation ;
class file
{
public:
virtual ~file ();
std:: uint64_t size () const ;
protected:
file (file&& other) noexcept ;
};
class readable_file : public virtual file
{
public:
[[nodiscard]]
file_read_operation read (
std:: uint64_t offset,
void * buffer,
std:: size_t byteCount,
cancellation_token ct = {}) const noexcept ;
};
class writable_file : public virtual file
{
public:
void set_size (std:: uint64_t fileSize);
[[nodiscard]]
file_write_operation write (
std:: uint64_t offset,
const void * buffer,
std:: size_t byteCount,
cancellation_token ct = {}) noexcept ;
};
class file_read_operation
{
public:
file_read_operation (file_read_operation&& other) noexcept ;
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter);
std:: size_t await_resume ();
};
class file_write_operation
{
public:
file_write_operation (file_write_operation&& other) noexcept ;
bool await_ready () const noexcept ;
bool await_suspend (std::experimental::coroutine_handle<> awaiter);
std:: size_t await_resume ();
};
}read_only_file , write_only_file , read_write_fileEsses tipos representam classes de E/S de arquivo de concreto.
Resumo da API:
namespace cppcoro
{
class read_only_file : public readable_file
{
public:
[[nodiscard]]
static read_only_file open (
io_service& ioService,
const std::experimental::filesystem::path& path,
file_share_mode shareMode = file_share_mode::read,
file_buffering_mode bufferingMode = file_buffering_mode::default_);
};
class write_only_file : public writable_file
{
public:
[[nodiscard]]
static write_only_file open (
io_service& ioService,
const std::experimental::filesystem::path& path,
file_open_mode openMode = file_open_mode::create_or_open,
file_share_mode shareMode = file_share_mode::none,
file_buffering_mode bufferingMode = file_buffering_mode::default_);
};
class read_write_file : public readable_file , public writable_file
{
public:
[[nodiscard]]
static read_write_file open (
io_service& ioService,
const std::experimental::filesystem::path& path,
file_open_mode openMode = file_open_mode::create_or_open,
file_share_mode shareMode = file_share_mode::none,
file_buffering_mode bufferingMode = file_buffering_mode::default_);
};
} Todas as funções open() lançam std::system_error na falha.
Nota: Atualmente, as abstrações de rede são suportadas apenas na plataforma Windows. O suporte ao Linux estará chegando em breve.
socketA classe de soquete pode ser usada para enviar/receber dados sobre a rede de forma assíncrona.
Atualmente, suporta apenas TCP/IP, UDP/IP sobre IPv4 e IPv6.
Resumo da API:
// <cppcoro/net/socket.hpp>
namespace cppcoro ::net
{
class socket
{
public:
static socket create_tcpv4 (ip_service& ioSvc);
static socket create_tcpv6 (ip_service& ioSvc);
static socket create_updv4 (ip_service& ioSvc);
static socket create_udpv6 (ip_service& ioSvc);
socket (socket&& other) noexcept ;
~socket ();
socket& operator =(socket&& other) noexcept ;
// Return the native socket handle for the socket
<platform-specific> native_handle () noexcept ;
const ip_endpoint& local_endpoint () const noexcept ;
const ip_endpoint& remote_endpoint () const noexcept ;
void bind ( const ip_endpoint& localEndPoint);
void listen ();
[[nodiscard]]
Awaitable< void > connect ( const ip_endpoint& remoteEndPoint) noexcept ;
[[nodiscard]]
Awaitable< void > connect ( const ip_endpoint& remoteEndPoint,
cancellation_token ct) noexcept ;
[[nodiscard]]
Awaitable< void > accept (socket& acceptingSocket) noexcept ;
[[nodiscard]]
Awaitable< void > accept (socket& acceptingSocket,
cancellation_token ct) noexcept ;
[[nodiscard]]
Awaitable< void > disconnect () noexcept ;
[[nodiscard]]
Awaitable< void > disconnect (cancellation_token ct) noexcept ;
[[nodiscard]]
Awaitable<std:: size_t > send ( const void * buffer, std:: size_t size) noexcept ;
[[nodiscard]]
Awaitable<std:: size_t > send ( const void * buffer,
std:: size_t size,
cancellation_token ct) noexcept ;
[[nodiscard]]
Awaitable<std:: size_t > recv ( void * buffer, std:: size_t size) noexcept ;
[[nodiscard]]
Awaitable<std:: size_t > recv ( void * buffer,
std:: size_t size,
cancellation_token ct) noexcept ;
[[nodiscard]]
socket_recv_from_operation recv_from (
void * buffer,
std:: size_t size) noexcept ;
[[nodiscard]]
socket_recv_from_operation_cancellable recv_from (
void * buffer,
std:: size_t size,
cancellation_token ct) noexcept ;
[[nodiscard]]
socket_send_to_operation send_to (
const ip_endpoint& destination,
const void * buffer,
std:: size_t size) noexcept ;
[[nodiscard]]
socket_send_to_operation_cancellable send_to (
const ip_endpoint& destination,
const void * buffer,
std:: size_t size,
cancellation_token ct) noexcept ;
void close_send ();
void close_recv ();
};
}Exemplo: Echo Server
# include < cppcoro/net/socket.hpp >
# include < cppcoro/io_service.hpp >
# include < cppcoro/cancellation_source.hpp >
# include < cppcoro/async_scope.hpp >
# include < cppcoro/on_scope_exit.hpp >
# include < memory >
# include < iostream >
cppcoro::task< void > handle_connection (socket s)
{
try
{
const size_t bufferSize = 16384 ;
auto buffer = std::make_unique< unsigned char []>(bufferSize);
size_t bytesRead;
do {
// Read some bytes
bytesRead = co_await s. recv (buffer. get (), bufferSize);
// Write some bytes
size_t bytesWritten = 0 ;
while (bytesWritten < bytesRead) {
bytesWritten += co_await s. send (
buffer. get () + bytesWritten,
bytesRead - bytesWritten);
}
} while (bytesRead != 0 );
s. close_send ();
co_await s. disconnect ();
}
catch (...)
{
std::cout << " connection failed " << std::
}
}
cppcoro::task< void > echo_server (
cppcoro::net::ipv4_endpoint endpoint,
cppcoro::io_service& ioSvc,
cancellation_token ct)
{
cppcoro::async_scope scope;
std::exception_ptr ex;
try
{
auto listeningSocket = cppcoro::net::socket::create_tcpv4 (ioSvc);
listeningSocket. bind (endpoint);
listeningSocket. listen ();
while ( true ) {
auto connection = cppcoro::net::socket::create_tcpv4 (ioSvc);
co_await listeningSocket. accept (connection, ct);
scope. spawn ( handle_connection ( std::move (connection)));
}
}
catch (cppcoro::operation_cancelled)
{
}
catch (...)
{
ex = std::current_exception ();
}
// Wait until all handle_connection tasks have finished.
co_await scope. join ();
if (ex) std::rethrow_exception (ex);
}
int main ( int argc, const char * argv[])
{
cppcoro::io_service ioSvc;
if (argc != 2 ) return - 1 ;
auto endpoint = cppcoro::ipv4_endpoint::from_string (argv[ 1 ]);
if (!endpoint) return - 1 ;
( void ) cppcoro::sync_wait ( cppcoro::when_all (
[&]() -> task<>
{
// Shutdown the event loop once finished.
auto stopOnExit = cppcoro::on_scope_exit ([&] { ioSvc. stop (); });
cppcoro::cancellation_source canceller;
co_await cppcoro::when_all (
[&]() -> task<>
{
// Run for 30s then stop accepting new connections.
co_await ioSvc. schedule_after ( std::chrono::seconds ( 30 ));
canceller. request_cancellation ();
}(),
echo_server (*endpoint, ioSvc, canceller. token ()));
}(),
[&]() -> task<>
{
ioSvc. process_events ();
}()));
return 0 ;
}ip_address , ipv4_address , ipv6_addressClasses Helper para representar um endereço IP.
Sinopse da API:
namespace cppcoro ::net
{
class ipv4_address
{
using bytes_t = std:: uint8_t [ 4 ];
public:
constexpr ipv4_address ();
explicit constexpr ipv4_address (std:: uint32_t integer);
explicit constexpr ipv4_address ( const std::uint8_t (&bytes)[4]);
explicit constexpr ipv4_address (std:: uint8_t b0,
std:: uint8_t b1,
std:: uint8_t b2,
std:: uint8_t b3);
constexpr const bytes_t & bytes () const ;
constexpr std:: uint32_t to_integer () const ;
static constexpr ipv4_address loopback ();
constexpr bool is_loopback () const ;
constexpr bool is_private_network () const ;
constexpr bool operator ==(ipv4_address other) const ;
constexpr bool operator !=(ipv4_address other) const ;
constexpr bool operator <(ipv4_address other) const ;
constexpr bool operator >(ipv4_address other) const ;
constexpr bool operator <=(ipv4_address other) const ;
constexpr bool operator >=(ipv4_address other) const ;
std::string to_string ();
static std::optional<ipv4_address> from_string (std::string_view string) noexcept ;
};
class ipv6_address
{
using bytes_t = std:: uint8_t [ 16 ];
public:
constexpr ipv6_address ();
explicit constexpr ipv6_address (
std:: uint64_t subnetPrefix,
std:: uint64_t interfaceIdentifier);
constexpr ipv6_address (
std:: uint16_t part0,
std:: uint16_t part1,
std:: uint16_t part2,
std:: uint16_t part3,
std:: uint16_t part4,
std:: uint16_t part5,
std:: uint16_t part6,
std:: uint16_t part7);
explicit constexpr ipv6_address (
const std::uint16_t (&parts)[8]);
explicit constexpr ipv6_address (
const std::uint8_t (bytes)[16]);
constexpr const bytes_t & bytes () const ;
constexpr std:: uint64_t subnet_prefix () const ;
constexpr std:: uint64_t interface_identifier () const ;
static constexpr ipv6_address unspecified ();
static constexpr ipv6_address loopback ();
static std::optional<ipv6_address> from_string (std::string_view string) noexcept ;
std::string to_string () const ;
constexpr bool operator ==( const ipv6_address& other) const ;
constexpr bool operator !=( const ipv6_address& other) const ;
constexpr bool operator <( const ipv6_address& other) const ;
constexpr bool operator >( const ipv6_address& other) const ;
constexpr bool operator <=( const ipv6_address& other) const ;
constexpr bool operator >=( const ipv6_address& other) const ;
};
class ip_address
{
public:
// Constructs to IPv4 address 0.0.0.0
ip_address () noexcept ;
ip_address (ipv4_address address) noexcept ;
ip_address (ipv6_address address) noexcept ;
bool is_ipv4 () const noexcept ;
bool is_ipv6 () const noexcept ;
const ipv4_address& to_ipv4 () const ;
const ipv6_address& to_ipv6 () const ;
const std:: uint8_t * bytes () const noexcept ;
std::string to_string () const ;
static std::optional<ip_address> from_string (std::string_view string) noexcept ;
bool operator ==( const ip_address& rhs) const noexcept ;
bool operator !=( const ip_address& rhs) const noexcept ;
// ipv4_address sorts less than ipv6_address
bool operator <( const ip_address& rhs) const noexcept ;
bool operator >( const ip_address& rhs) const noexcept ;
bool operator <=( const ip_address& rhs) const noexcept ;
bool operator >=( const ip_address& rhs) const noexcept ;
};
}ip_endpoint , ipv4_endpoint ipv6_endpointClasses Helper para representar um endereço IP e um número de porta.
Sinopse da API:
namespace cppcoro ::net
{
class ipv4_endpoint
{
public:
ipv4_endpoint () noexcept ;
explicit ipv4_endpoint (ipv4_address address, std:: uint16_t port = 0 ) noexcept ;
const ipv4_address& address () const noexcept ;
std:: uint16_t port () const noexcept ;
std::string to_string () const ;
static std::optional<ipv4_endpoint> from_string (std::string_view string) noexcept ;
};
bool operator ==( const ipv4_endpoint& a, const ipv4_endpoint& b);
bool operator !=( const ipv4_endpoint& a, const ipv4_endpoint& b);
bool operator <( const ipv4_endpoint& a, const ipv4_endpoint& b);
bool operator >( const ipv4_endpoint& a, const ipv4_endpoint& b);
bool operator <=( const ipv4_endpoint& a, const ipv4_endpoint& b);
bool operator >=( const ipv4_endpoint& a, const ipv4_endpoint& b);
class ipv6_endpoint
{
public:
ipv6_endpoint () noexcept ;
explicit ipv6_endpoint (ipv6_address address, std:: uint16_t port = 0 ) noexcept ;
const ipv6_address& address () const noexcept ;
std:: uint16_t port () const noexcept ;
std::string to_string () const ;
static std::optional<ipv6_endpoint> from_string (std::string_view string) noexcept ;
};
bool operator ==( const ipv6_endpoint& a, const ipv6_endpoint& b);
bool operator !=( const ipv6_endpoint& a, const ipv6_endpoint& b);
bool operator <( const ipv6_endpoint& a, const ipv6_endpoint& b);
bool operator >( const ipv6_endpoint& a, const ipv6_endpoint& b);
bool operator <=( const ipv6_endpoint& a, const ipv6_endpoint& b);
bool operator >=( const ipv6_endpoint& a, const ipv6_endpoint& b);
class ip_endpoint
{
public:
// Constructs to IPv4 end-point 0.0.0.0:0
ip_endpoint () noexcept ;
ip_endpoint (ipv4_endpoint endpoint) noexcept ;
ip_endpoint (ipv6_endpoint endpoint) noexcept ;
bool is_ipv4 () const noexcept ;
bool is_ipv6 () const noexcept ;
const ipv4_endpoint& to_ipv4 () const ;
const ipv6_endpoint& to_ipv6 () const ;
ip_address address () const noexcept ;
std:: uint16_t port () const noexcept ;
std::string to_string () const ;
static std::optional<ip_endpoint> from_string (std::string_view string) noexcept ;
bool operator ==( const ip_endpoint& rhs) const noexcept ;
bool operator !=( const ip_endpoint& rhs) const noexcept ;
// ipv4_endpoint sorts less than ipv6_endpoint
bool operator <( const ip_endpoint& rhs) const noexcept ;
bool operator >( const ip_endpoint& rhs) const noexcept ;
bool operator <=( const ip_endpoint& rhs) const noexcept ;
bool operator >=( const ip_endpoint& rhs) const noexcept ;
};
}sync_wait() A função sync_wait() pode ser usada para aguardar síncrona até que o especificado awaitable seja concluído.
O especificado aguardável será co_await ed no encadeamento atual dentro de uma coroutina recém -criada.
A chamada sync_wait() será bloqueada até que a operação seja concluída e retorne o resultado da expressão co_await ou repensará a exceção se a expressão co_await concluída com uma exceção não tratada.
A função sync_wait() é principalmente útil para iniciar uma tarefa de nível superior de dentro de main() e aguardar até que a tarefa termine, na prática é a única maneira de iniciar a primeira task primeiro/de nível superior.
Resumo da API:
// <cppcoro/sync_wait.hpp>
namespace cppcoro
{
template < typename AWAITABLE>
auto sync_wait (AWAITABLE&& awaitable)
-> typename awaitable_traits<AWAITABLE&&>::await_result_t;
}Exemplos:
void example_task ()
{
auto makeTask = []() -> task<std::string>
{
co_return " foo " ;
};
auto task = makeTask ();
// start the lazy task and wait until it completes
sync_wait (task); // -> "foo"
sync_wait ( makeTask ()); // -> "foo"
}
void example_shared_task ()
{
auto makeTask = []() -> shared_task<std::string>
{
co_return " foo " ;
};
auto task = makeTask ();
// start the shared task and wait until it completes
sync_wait (task) == " foo " ;
sync_wait ( makeTask ()) == " foo " ;
}when_all_ready() A função when_all_ready() pode ser usada para criar um novo aguardável que seja concluído quando todas as entradas aguardam concluídas.
As tarefas de entrada podem ser qualquer tipo de espera.
Quando o retorno aguardável é co_await ed, ele co_await cada uma das entradas aguarda, por sua vez, o encadeamento aguardando na ordem em que eles são passados para a função when_all_ready() . Se essas tarefas não concluirem de maneira síncrona, eles serão executados simultaneamente.
Depois que todas as expressões co_await sobre entrada aguardam a conclusão, o retorno aguardável será concluído e retomará a coroutina aguardando. A coroutina aguardando será retomada no segmento da entrada aguardável que seja a última a ser concluída.
O devolvido aguardável é garantido para não lançar uma exceção quando co_await ed, mesmo que algumas das contribuições aguardáveis falhem com uma exceção não atendida.
Observe, no entanto, que a chamada when_all_ready() pode lançar std::bad_alloc se não conseguir alocar memória para os quadros de coroutina necessários para aguardar cada uma das entradas aguardando. Também pode lançar uma exceção se algum dos objetos de entrada aguardam de seus construtores de cópia/movimentação.
O resultado de co_await do retorno aguardável é uma std::tuple ou std::vector de when_all_task<RESULT> objetos. Esses objetos permitem que você obtenha o resultado (ou exceção) de cada entrada aguardável separadamente, chamando o método when_all_task<RESULT>::result() da tarefa de saída correspondente. Isso permite que o chamador aguarde simultaneamente múltiplos aguardam e sincronize sua conclusão, mantendo a capacidade de inspecionar posteriormente os resultados de cada uma das operações co_await para obter sucesso/falha.
Isso difere de when_all() , onde a falha de qualquer operação individual co_await faz com que a operação geral falhe com uma exceção. Isso significa que você não pode determinar qual das operações do componente co_await falhou e também impede que você obtenha os resultados das outras operações co_await .
Resumo da API:
// <cppcoro/when_all_ready.hpp>
namespace cppcoro
{
// Concurrently await multiple awaitables.
//
// Returns an awaitable object that, when co_await'ed, will co_await each of the input
// awaitable objects and will resume the awaiting coroutine only when all of the
// component co_await operations complete.
//
// Result of co_await'ing the returned awaitable is a std::tuple of detail::when_all_task<T>,
// one for each input awaitable and where T is the result-type of the co_await expression
// on the corresponding awaitable.
//
// AWAITABLES must be awaitable types and must be movable (if passed as rvalue) or copyable
// (if passed as lvalue). The co_await expression will be executed on an rvalue of the
// copied awaitable.
template < typename ... AWAITABLES>
auto when_all_ready (AWAITABLES&&... awaitables)
-> Awaitable<std::tuple<detail::when_all_task<typename awaitable_traits<AWAITABLES>::await_result_t>...>>;
// Concurrently await each awaitable in a vector of input awaitables.
template <
typename AWAITABLE,
typename RESULT = typename awaitable_traits<AWAITABLE>:: await_result_t >
auto when_all_ready (std::vector<AWAITABLE> awaitables)
-> Awaitable<std::vector<detail::when_all_task<RESULT>>>;
}Exemplo de uso:
task<std::string> get_record ( int id);
task<> example1 ()
{
// Run 3 get_record() operations concurrently and wait until they're all ready.
// Returns a std::tuple of tasks that can be unpacked using structured bindings.
auto [task1, task2, task3] = co_await when_all_ready (
get_record ( 123 ),
get_record ( 456 ),
get_record ( 789 ));
// Unpack the result of each task
std::string& record1 = task1. result ();
std::string& record2 = task2. result ();
std::string& record3 = task3. result ();
// Use records....
}
task<> example2 ()
{
// Create the input tasks. They don't start executing yet.
std::vector<task<std::string>> tasks;
for ( int i = 0 ; i < 1000 ; ++i)
{
tasks. emplace_back ( get_record (i));
}
// Execute all tasks concurrently.
std::vector<detail::when_all_task<std::string>> resultTasks =
co_await when_all_ready ( std::move (tasks));
// Unpack and handle each result individually once they're all complete.
for ( int i = 0 ; i < 1000 ; ++i)
{
try
{
std::string& record = tasks[i]. result ();
std::cout << i << " = " << record << std::endl;
}
catch ( const std:: exception & ex)
{
std::cout << i << " : " << ex. what () << std::endl;
}
}
}when_all() A função when_all() pode ser usada para criar um novo aguardável que, quando co_await ed co_await , cada uma das entradas aguarda simultaneamente e retornará um agregado de seus resultados individuais.
Quando o retorno aguardável é aguardado, ele co_await cada uma das entradas aguarda no thread atual. Uma vez que o primeiro aguardável suspense, a segunda tarefa será iniciada e assim por diante. As operações são executadas simultaneamente até que todas sejam executadas até a conclusão.
Depois que todas as operações de componentes co_await foram executadas até a conclusão, um agregado dos resultados é construído a partir de cada resultado individual. Se uma exceção for lançada por qualquer uma das tarefas de entrada ou se a construção do resultado agregado lançar uma exceção, a exceção se propagará da co_await do retorno aguardável.
Se várias operações co_await falharem com uma exceção, uma das exceções se propagará do co_await when_all() a expressão de outras exceções será ignorada silenciosamente. Não é especificado qual exceção da operação será escolhida.
Se for importante saber qual componente co_await a operação falhou ou reter a capacidade de obter resultados de outras operações, mesmo que alguns deles falhem, você deve usar when_all_ready() .
Resumo da API:
// <cppcoro/when_all.hpp>
namespace cppcoro
{
// Variadic version.
//
// Note that if the result of `co_await awaitable` yields a void-type
// for some awaitables then the corresponding component for that awaitable
// in the tuple will be an empty struct of type detail::void_value.
template < typename ... AWAITABLES>
auto when_all (AWAITABLES&&... awaitables)
-> Awaitable<std::tuple<typename awaitable_traits<AWAITABLES>::await_result_t...>>;
// Overload for vector<Awaitable<void>>.
template <
typename AWAITABLE,
typename RESULT = typename awaitable_traits<AWAITABLE>:: await_result_t ,
std:: enable_if_t <std::is_void_v<RESULT>, int > = 0 >
auto when_all (std::vector<AWAITABLE> awaitables)
-> Awaitable<void>;
// Overload for vector<Awaitable<NonVoid>> that yield a value when awaited.
template <
typename AWAITABLE,
typename RESULT = typename awaitable_traits<AWAITABLE>:: await_result_t ,
std:: enable_if_t <!std::is_void_v<RESULT>, int > = 0 >
auto when_all (std::vector<AWAITABLE> awaitables)
-> Awaitable<std::vector<std::conditional_t<
std::is_lvalue_reference_v<RESULT>,
std::reference_wrapper<std::remove_reference_t<RESULT>>,
std::remove_reference_t<RESULT>>>>;
}Exemplos:
task<A> get_a ();
task<B> get_b ();
task<> example1 ()
{
// Run get_a() and get_b() concurrently.
// Task yields a std::tuple<A, B> which can be unpacked using structured bindings.
auto [a, b] = co_await when_all ( get_a (), get_b ());
// use a, b
}
task<std::string> get_record ( int id);
task<> example2 ()
{
std::vector<task<std::string>> tasks;
for ( int i = 0 ; i < 1000 ; ++i)
{
tasks. emplace_back ( get_record (i));
}
// Concurrently execute all get_record() tasks.
// If any of them fail with an exception then the exception will propagate
// out of the co_await expression once they have all completed.
std::vector<std::string> records = co_await when_all ( std::move (tasks));
// Process results
for ( int i = 0 ; i < 1000 ; ++i)
{
std::cout << i << " = " << records[i] << std::endl;
}
}fmap() A função fmap() pode ser usada para aplicar uma função chamada ao (s) valor (s) contido em um tipo de contêiner, retornando um novo tipo de contêiner dos resultados da aplicação da função do (s) valor (s) contido (s).
A função fmap() pode aplicar uma função aos valores do generator<T> , recursive_generator<T> e async_generator<T> bem como qualquer valor que suporta o conceito Awaitable (por exemplo, task<T> ).
Cada um desses tipos fornece uma sobrecarga para fmap() que leva dois argumentos; uma função a ser aplicada e o valor do contêiner. Consulte a documentação para cada tipo para as sobrecargas fmap() suportadas.
Por exemplo, a função fmap() pode ser usada para aplicar uma função ao resultado eventual de uma task<T> , produzindo uma nova task<U> que será concluída com o valor de retorno da função.
// Given a function you want to apply that converts
// a value of type A to value of type B.
B a_to_b (A value);
// And a task that yields a value of type A
cppcoro::task<A> get_an_a ();
// We can apply the function to the result of the task using fmap()
// and obtain a new task yielding the result.
cppcoro::task<B> bTask = fmap(a_to_b, get_an_a());
// An alternative syntax is to use the pipe notation.
cppcoro::task<B> bTask = get_an_a() | cppcoro::fmap(a_to_b);Resumo da API:
// <cppcoro/fmap.hpp>
namespace cppcoro
{
template < typename FUNC>
struct fmap_transform
{
fmap_transform (FUNC&& func) noexcept (std::is_nothrow_move_constructible_v<FUNC>);
FUNC func;
};
// Type-deducing constructor for fmap_transform object that can be used
// in conjunction with operator|.
template < typename FUNC>
fmap_transform<FUNC> fmap (FUNC&& func);
// operator| overloads for providing pipe-based syntactic sugar for fmap()
// such that the expression:
// <value-expr> | cppcoro::fmap(<func-expr>)
// is equivalent to:
// fmap(<func-expr>, <value-expr>)
template < typename T, typename FUNC>
decltype ( auto ) operator |(T&& value, fmap_transform<FUNC>&& transform);
template < typename T, typename FUNC>
decltype ( auto ) operator |(T&& value, fmap_transform<FUNC>& transform);
template < typename T, typename FUNC>
decltype ( auto ) operator |(T&& value, const fmap_transform<FUNC>& transform);
// Generic overload for all awaitable types.
//
// Returns an awaitable that when co_awaited, co_awaits the specified awaitable
// and applies the specified func to the result of the 'co_await awaitable'
// expression as if by 'std::invoke(func, co_await awaitable)'.
//
// If the type of 'co_await awaitable' expression is 'void' then co_awaiting the
// returned awaitable is equivalent to 'co_await awaitable, func()'.
template <
typename FUNC,
typename AWAITABLE,
std:: enable_if_t <is_awaitable_v<AWAITABLE>, int > = 0 >
auto fmap (FUNC&& func, AWAITABLE&& awaitable)
-> Awaitable<std::invoke_result_t<FUNC, typename awaitable_traits<AWAITABLE>::await_result_t>>;
} A função fmap() foi projetada para procurar a sobrecarga correta por pesquisa dependente de argumento (ADL), para que geralmente deve ser chamada sem o prefixo cppcoro:: .
resume_on() A função resume_on() pode ser usada para controlar o contexto de execução em que um aguardável retomará a coroutina aguardando quando aguardada. Quando aplicado a um async_generator ele controla qual contexto de execução o co_await g.begin() e co_await ++it retomam as corotas aguardando.
Normalmente, a coroutina aguardando de uma tarefa aguardável (por exemplo, uma task ) ou async_generator retomará a execução em qualquer tópico que a operação seja concluída. Em alguns casos, esse pode não ser o tópico em que você deseja continuar executando. Nesses casos, você pode usar a função resume_on() para criar um novo ou gerador aguardável que retome a execução em um thread associado a um agendador especificado.
A função resume_on() pode ser usada como uma função normal retornando um novo aguardável/gerador. Ou pode ser usado em um síntax de pipeline.
Exemplo:
task<record> load_record ( int id);
ui_thread_scheduler uiThreadScheduler;
task<> example ()
{
// This will start load_record() on the current thread.
// Then when load_record() completes (probably on an I/O thread)
// it will reschedule execution onto thread pool and call to_json
// Once to_json completes it will transfer execution onto the
// ui thread before resuming this coroutine and returning the json text.
task<std::string> jsonTask =
load_record ( 123 )
| cppcoro::resume_on ( threadpool::default ())
| cppcoro::fmap (to_json)
| cppcoro::resume_on (uiThreadScheduler);
// At this point, all we've done is create a pipeline of tasks.
// The tasks haven't started executing yet.
// Await the result. Starts the pipeline of tasks.
std::string jsonText = co_await jsonTask;
// Guaranteed to be executing on ui thread here.
someUiControl. set_text (jsonText);
}Resumo da API:
// <cppcoro/resume_on.hpp>
namespace cppcoro
{
template < typename SCHEDULER, typename AWAITABLE>
auto resume_on (SCHEDULER& scheduler, AWAITABLE awaitable)
-> Awaitable<typename awaitable_traits<AWAITABLE>::await_traits_t>;
template < typename SCHEDULER, typename T>
async_generator<T> resume_on (SCHEDULER& scheduler, async_generator<T> source);
template < typename SCHEDULER>
struct resume_on_transform
{
explicit resume_on_transform (SCHEDULER& scheduler) noexcept ;
SCHEDULER& scheduler;
};
// Construct a transform/operation that can be applied to a source object
// using "pipe" notation (ie. operator|).
template < typename SCHEDULER>
resume_on_transform<SCHEDULER> resume_on (SCHEDULER& scheduler) noexcept ;
// Equivalent to 'resume_on(transform.scheduler, std::forward<T>(value))'
template < typename T, typename SCHEDULER>
decltype ( auto ) operator |(T&& value, resume_on_transform<SCHEDULER> transform)
{
return resume_on (transform. scheduler , std::forward<T>(value));
}
}schedule_on() A função schedule_on() pode ser usada para alterar o contexto de execução que um determinado aguardável ou async_generator começa a executar.
Quando aplicado a um async_generator , ele também afeta qual contexto de execução ele retoma a instrução After co_yield .
Observe que a transformação schedule_on não especifica o encadeamento em que o aguardável ou async_generator preencherá ou produzirá resultados, que depende da implementação do aguardável ou gerador.
Consulte o operador resume_on() para obter uma transformação que controla o encadeamento que a operação é concluída.
Por exemplo:
task< int > get_value ();
io_service ioSvc;
task<> example ()
{
// Starts executing get_value() on the current thread.
int a = co_await get_value ();
// Starts executing get_value() on a thread associated with ioSvc.
int b = co_await schedule_on (ioSvc, get_value ());
}Resumo da API:
// <cppcoro/schedule_on.hpp>
namespace cppcoro
{
// Return a task that yields the same result as 't' but that
// ensures that 't' is co_await'ed on a thread associated with
// the specified scheduler. Resulting task will complete on
// whatever thread 't' would normally complete on.
template < typename SCHEDULER, typename AWAITABLE>
auto schedule_on (SCHEDULER& scheduler, AWAITABLE awaitable)
-> Awaitable<typename awaitable_traits<AWAITABLE>::await_result_t>;
// Return a generator that yields the same sequence of results as
// 'source' but that ensures that execution of the coroutine starts
// execution on a thread associated with 'scheduler' and resumes
// after a 'co_yield' on a thread associated with 'scheduler'.
template < typename SCHEDULER, typename T>
async_generator<T> schedule_on (SCHEDULER& scheduler, async_generator<T> source);
template < typename SCHEDULER>
struct schedule_on_transform
{
explicit schedule_on_transform (SCHEDULER& scheduler) noexcept ;
SCHEDULER& scheduler;
};
template < typename SCHEDULER>
schedule_on_transform<SCHEDULER> schedule_on (SCHEDULER& scheduler) noexcept ;
template < typename T, typename SCHEDULER>
decltype ( auto ) operator |(T&& value, schedule_on_transform<SCHEDULER> transform);
}awaitable_traits<T> Esta metafunção de modelo pode ser usada para determinar qual será o tipo resultante de uma expressão de co_await se aplicado a uma expressão do tipo T .
Observe que isso pressupõe que o valor do Tipo T esteja sendo aguardado em um contexto em que não seja afetado por qualquer await_transform aplicado pelo objeto Promise da Coroutine. Os resultados podem diferir se um valor do tipo T for aguardado nesse contexto.
O modelo awaitable_traits<T> Metafunction não define o awaiter_t ou await_result_t typeDefs aninhados se o tipo, T , não é aguardável. Isso permite seu uso em contextos Sfinae que desativa as sobrecargas quando T não é aguardável.
Resumo da API:
// <cppcoro/awaitable_traits.hpp>
namespace cppcoro
{
template < typename T>
struct awaitable_traits
{
// The type that results from applying `operator co_await()` to a value
// of type T, if T supports an `operator co_await()`, otherwise is type `T&&`.
typename awaiter_t = <unspecified>;
// The type of the result of co_await'ing a value of type T.
typename await_result_t = <unspecified>;
};
}is_awaitable<T> O modelo is_awaitable<T> metafunção permite consultar se um determinado tipo pode ou não ser co_await ed ou não dentro de uma coroutina.
Resumo da API:
// <cppcoro/is_awaitable.hpp>
namespace cppcoro
{
template < typename T>
struct is_awaitable : std::bool_constant<...>
{};
template < typename T>
constexpr bool is_awaitable_v = is_awaitable<T>::value;
}Awaitable<T> Um Awaitable<T> é um conceito que indica que um tipo pode ser co_await ed em um contexto de coroartina que não tem sobrecarga await_transform e que o resultado da expressão co_await tem tipo, T .
Por exemplo, a task<T> implementa o conceito Awaitable<T&&> enquanto a task<T>& implementa o conceito Awaitable<T&> .
Awaiter<T> Conceito Um Awaiter<T> é um conceito que indica que um tipo contém os métodos await_ready , await_suspend e await_resume os métodos necessários para implementar o protocolo para suspender/retomar uma coroutina aguardando.
Um tipo que satisfaz Awaiter<T> deve ter, para uma instância do tipo, awaiter :
awaiter.await_ready() -> boolawaiter.await_suspend(std::experimental::coroutine_handle<void>{}) -> void ou bool ou std::experimental::coroutine_handle<P> Para alguns P .awaiter.await_resume() -> T Qualquer tipo que implementa o conceito Awaiter<T> também implementa o conceito Awaitable<T> .
Scheduler Um Scheduler é um conceito que permite agendar a execução de coroutinas em algum contexto de execução.
concept Scheduler
{
Awaitable< void > schedule ();
} Dado um tipo, S , que implementa o conceito Scheduler e uma instância, s , do tipo S :
s.schedule() retorna um tipo aguardável, de modo que co_await s.schedule() suspenda incondicionalmente a coroutina atual e a agende para retomada no contexto de execução associado ao agendador, s .co_await s.schedule() tem tipo void . cppcoro::task<> f (Scheduler& scheduler)
{
// Execution of the coroutine is initially on the caller's execution context.
// Suspends execution of the coroutine and schedules it for resumption on
// the scheduler's execution context.
co_await scheduler. schedule ();
// At this point the coroutine is now executing on the scheduler's
// execution context.
}DelayedScheduler Um DelayedScheduler é um conceito que permite que uma coroutina se agenda para execução no contexto de execução do agendador após a decorrência de uma duração especificada.
concept DelayedScheduler : Scheduler
{
template < typename REP, typename RATIO>
Awaitable< void > schedule_after (std::chrono::duration<REP, RATIO> delay);
template < typename REP, typename RATIO>
Awaitable< void > schedule_after (
std::chrono::duration<REP, RATIO> delay,
cppcoro::cancellation_token cancellationToken);
} Dado um tipo, S , que implementa o DelayedScheduler e uma instância, s do tipo S :
s.schedule_after(delay) retorna um objeto que pode ser aguardado de modo que co_await s.schedule_after(delay) suspenda a corota atual por um delay antes de agendar a coroutina para retomada no contexto de execução associado ao s .co_await s.schedule_after(delay) tem tipo void .A biblioteca CPPCORO suporta a construção do Windows com o Visual Studio 2017 e o Linux com o Clang 5.0+.
Esta biblioteca utiliza o sistema de compilação de bolo (não, não o C# um).
O sistema de compilação de bolo é verificado automaticamente como um submódulo Git, para que você não precise baixar ou instalá -lo separadamente.
Atualmente, esta biblioteca requer o Visual Studio 2017 ou posterior e o Windows 10 SDK.
O suporte para Clang (#3) e Linux (#15) está planejado.
O sistema de construção do bolo é implementado no Python e requer que o Python 2.7 seja instalado.
Verifique se o intérprete do Python 2.7 está no seu caminho e está disponível como 'Python'.
Verifique se a atualização 3 do Visual Studio 2017 está instalada ou posterior. Observe que existem alguns problemas conhecidos com as coroutinas na atualização 2 ou anterior que foram corrigidas na atualização 3.
Você também pode usar uma versão experimental do Visual Studio Compiler, baixando um pacote NUGET em https://vcppdogfooding.azurewebsites.net/ e descompactando o arquivo .nuget em um diretório. Basta atualizar o arquivo config.cake para apontar para o local desconfiado modificando e descomando a seguinte linha:
nugetPath = None # r'C:PathToVisualCppTools.14.0.25224-Pre'Verifique se você tem o Windows 10 SDK instalado. Ele usará a versão mais recente do Windows 10 SDK e Universal C Runtime por padrão.
O repositório CPPCORO faz uso de submódulos Git para puxar a fonte do sistema de compilação de bolo.
Isso significa que você precisa passar o sinalizador --recursive para o comando git clone . por exemplo.
c:Code> git clone --recursive https://github.com/lewissbaker/cppcoro.git
Se você já clonou o CPPCORO, atualize os submódulos após a retirada das alterações.
c:Codecppcoro> git submodule update --init --recursive
Para construir a partir da linha de comando, basta executar 'bolo.bat' na raiz do espaço de trabalho.
por exemplo.
C:cppcoro> cake.bat
Building with C:cppcoroconfig.cake - Variant(release='debug', platform='windows', architecture='x86', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='optimised', platform='windows', architecture='x64', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='debug', platform='windows', architecture='x64', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='optimised', platform='windows', architecture='x86', compilerFamily='msvc', compiler='msvc14.10')
Compiling testmain.cpp
Compiling testmain.cpp
Compiling testmain.cpp
Compiling testmain.cpp
...
Linking buildwindows_x86_msvc14.10_debugtestrun.exe
Linking buildwindows_x64_msvc14.10_optimisedtestrun.exe
Linking buildwindows_x86_msvc14.10_optimisedtestrun.exe
Linking buildwindows_x64_msvc14.10_debugtestrun.exe
Generating code
Finished generating code
Generating code
Finished generating code
Build succeeded.
Build took 0:00:02.419.
Por padrão, a execução cake sem argumentos criará todos os projetos com todas as variantes de construção e executará os testes de unidade. Você pode restringir o que é construído passando argumentos adicionais da linha de comando. por exemplo.
c:cppcoro> cake.bat release=debug architecture=x64 lib/build.cake
Building with C:UsersLewisCodecppcoroconfig.cake - Variant(release='debug', platform='windows', architecture='x64', compilerFamily='msvc', compiler='msvc14.10')
Archiving buildwindows_x64_msvc14.10_debuglibcppcoro.lib
Build succeeded.
Build took 0:00:00.321.
Você pode executar cake --help para listar as opções de linha de comando disponíveis.
Para se desenvolver no Visual Studio, você pode criar arquivos .vcproj/.sln executando cake.bat -p .
por exemplo.
c:cppcoro> cake.bat -p
Building with C:cppcoroconfig.cake - Variant(release='debug', platform='windows', architecture='x86', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='optimised', platform='windows', architecture='x64', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='debug', platform='windows', architecture='x64', compilerFamily='msvc', compiler='msvc14.10')
Building with C:cppcoroconfig.cake - Variant(release='optimised', platform='windows', architecture='x86', compilerFamily='msvc', compiler='msvc14.10')
Generating Solution build/project/cppcoro.sln
Generating Project build/project/cppcoro_tests.vcxproj
Generating Filters build/project/cppcoro_tests.vcxproj.filters
Generating Project build/project/cppcoro.vcxproj
Generating Filters build/project/cppcoro.vcxproj.filters
Build succeeded.
Build took 0:00:00.247.
Quando você constrói esses projetos no Visual Studio, ele chama o bolo para realizar a compilação.
O projeto CPPCORO também pode ser construído no Linux usando o CLANG+ LIBC ++ 5.0 ou posterior.
A construção do CPPCORO foi testada no Ubuntu 17.04.
Certifique -se de ter os seguintes pacotes instalados:
Isso supõe que você tenha CLANG e LIBC ++ construídos e instalados.
Se você ainda não possui Clang, consulte as seções a seguir para obter detalhes sobre a configuração de Clang para a construção do CPPCORO.
Checkout cppcoro e seus submódulos:
git clone --recursive https://github.com/lewissbaker/cppcoro.git cppcoro
Execute init.sh para configurar a função cake Bash:
cd cppcoro
source init.sh
Em seguida, você pode executar cake da raiz do espaço de trabalho para construir o CPPCORO e executar testes:
$ cake
Você pode especificar argumentos adicionais da linha de comando para personalizar a compilação:
--help imprimirá ajuda para argumentos de linha de comando--debug=run mostrará as linhas de comando de compilação sendo executadasrelease=debug ou release=optimised limitará a variante de compilação a depurar ou otimizada (por padrão, ele criará os dois).lib/build.cake apenas criará a biblioteca CPPCORO e não os testes.test/build.cake@task_tests.cpp apenas compilará um arquivo de origem específicotest/build.cake@testresult construirá e executará os testesPor exemplo:
$ cake --debug=run release=debug lib/build.cake
Se o seu compilador de clang não estiver localizado em /usr/bin/clang , você poderá especificar um local alternativo usando uma ou mais das seguintes opções de linha de comando para cake :
--clang-executable=<name> -Especifique o nome executável do CLANG para usar em vez de clang . por exemplo. Para forçar o uso de Clang 8.0 Pass --clang-executable=clang-8--clang-executable=<abspath> -Especifique o caminho completo para o CLANG executável. O sistema Build também procurará outros executáveis no mesmo diretório. Se esse caminho tiver o formulário <prefix>/bin/<name> , isso também definirá o prefixo Clang-Install padrão como <prefix> .--clang-install-prefix=<path> -Especifique o caminho onde o CLANG foi instalado. Isso fará com que o sistema de construção procure o Clang Under <path>/bin (a menos que seja substituído por --clang-executable ).--libcxx-install-prefix=<path> -Especifique o caminho em que o libc ++ foi instalado. Por padrão, o sistema de construção procurará LIBC ++ no mesmo local que o CLANG. Use esta opção de linha de comando se estiver instalada em um local diferente.Exemplo: use uma versão específica do Clang instalado no local padrão
$ cake --clang-executable=clang-8
Exemplo: use a versão padrão do CLANG de um local personalizado
$ cake --clang-install-prefix=/path/to/clang-install
Exemplo: use uma versão específica do CLANG, em um local personalizado, com libc ++ de um local diferente
$ cake --clang-executable=/path/to/clang-install/bin/clang-8 --libcxx-install-prefix=/path/to/libcxx-install
If your Linux distribution does not have a version of Clang 5.0 or later available, you can install a snapshot build from the LLVM project.
Follow instructions at http://apt.llvm.org/ to setup your package manager to support pulling from the LLVM package manager.
For example, for Ubuntu 17.04 Zesty:
Edit /etc/apt/sources.list and add the following lines:
deb http://apt.llvm.org/zesty/ llvm-toolchain-zesty main
deb-src http://apt.llvm.org/zesty/ llvm-toolchain-zesty main
Install the PGP key for those packages:
$ wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key | sudo apt-key add -
Install Clang and LLD:
$ sudo apt-get install clang-6.0 lld-6.0
The LLVM snapshot builds do not include libc++ versions so you'll need to build that yourself. Veja abaixo.
You can also use the bleeding-edge Clang version by building Clang from source yourself.
See instructions here:
To do this you will need to install the following pre-requisites:
$ sudo apt-get install git cmake ninja-build clang lld
Note that we are using your distribution's version of clang to build clang from source. GCC could also be used here instead.
Checkout LLVM + Clang + LLD + libc++ repositories:
mkdir llvm
cd llvm
git clone --depth=1 https://github.com/llvm-mirror/llvm.git llvm
git clone --depth=1 https://github.com/llvm-mirror/clang.git llvm/tools/clang
git clone --depth=1 https://github.com/llvm-mirror/lld.git llvm/tools/lld
git clone --depth=1 https://github.com/llvm-mirror/libcxx.git llvm/projects/libcxx
ln -s llvm/tools/clang clang
ln -s llvm/tools/lld lld
ln -s llvm/projects/libcxx libcxx
Configure and build Clang:
mkdir clang-build
cd clang-build
cmake -GNinja
-DCMAKE_CXX_COMPILER=/usr/bin/clang++
-DCMAKE_C_COMPILER=/usr/bin/clang
-DCMAKE_BUILD_TYPE=MinSizeRel
-DCMAKE_INSTALL_PREFIX="/path/to/clang/install"
-DCMAKE_BUILD_WITH_INSTALL_RPATH="yes"
-DLLVM_TARGETS_TO_BUILD=X86
-DLLVM_ENABLE_PROJECTS="lld;clang"
../llvm
ninja install-clang
install-clang-headers
install-llvm-ar
install-lld
The cppcoro project requires libc++ as it contains the <experimental/coroutine> header required to use C++ coroutines under Clang.
Checkout libc++ + llvm :
mkdir llvm
cd llvm
git clone --depth=1 https://github.com/llvm-mirror/llvm.git llvm
git clone --depth=1 https://github.com/llvm-mirror/libcxx.git llvm/projects/libcxx
ln -s llvm/projects/libcxx libcxx
Build libc++ :
mkdir libcxx-build
cd libcxx-build
cmake -GNinja
-DCMAKE_CXX_COMPILER="/path/to/clang/install/bin/clang++"
-DCMAKE_C_COMPILER="/path/to/clang/install/bin/clang"
-DCMAKE_BUILD_TYPE=Release
-DCMAKE_INSTALL_PREFIX="/path/to/clang/install"
-DLLVM_PATH="../llvm"
-DLIBCXX_CXX_ABI=libstdc++
-DLIBCXX_CXX_ABI_INCLUDE_PATHS="/usr/include/c++/6.3.0/;/usr/include/x86_64-linux-gnu/c++/6.3.0/"
../libcxx
ninja cxx
ninja install
This will build and install libc++ into the same install directory where you have clang installed.
The cppcoro port in vcpkg is kept up to date by Microsoft team members and community contributors. The url of vcpkg is: https://github.com/Microsoft/vcpkg . You can download and install cppcoro using the vcpkg dependency manager:
git clone https://github.com/Microsoft/vcpkg.git
cd vcpkg
./bootstrap-vcpkg.sh # ./bootstrap-vcpkg.bat for Windows
./vcpkg integrate install
./vcpkg install cppcoroSe a versão estiver desatualizada, crie uma solicitação de problema ou puxe no repositório VCPKG.
GitHub issues are the primary mechanism for support, bug reports and feature requests.
Contributions are welcome and pull-requests will be happily reviewed. I only ask that you agree to license any contributions that you make under the MIT license.
If you have general questions about C++ coroutines, you can generally find someone to help in the #coroutines channel on Cpplang Slack group.
