Como: use a sobresubscrição para deslocar a latência
A sobresubscrição pode melhorar a eficiência total de alguns aplicativos que contêm tarefas que têm uma quantidade de latência dependendo de alta.Este tópico mostra como usar a sobresubscrição para deslocar a latência que é causado ler dados de uma conexão de rede.
Exemplo
Este exemplo usa Biblioteca de agentes assíncrono para baixar arquivos de servidores HTTP.A classe deriva de http_reader de concurrency::agent e usa a mensagem que passa de forma assíncrona para ler os nomes de URL para baixar.
A classe de http_reader usa a classe de concurrency::task_group para ler simultaneamente cada arquivo.Cada tarefa chama o método de concurrency::Context::Oversubscribe com o parâmetro de _BeginOversubscription definido como true para ativar a sobresubscrição no contexto atual.Cada tarefa usa nas classes (MFC) CInternetSession da Microsoft foundation classes e de CHttpFile para baixar o arquivo.Finalmente, cada tarefa chama Context::Oversubscribe com o parâmetro de _BeginOversubscription definido como false para desativar a sobresubscrição.
Quando a sobresubscrição estiver ativada, o tempo de execução cria um segmento adicional para executar tarefas.Cada um desses segmentos enlata também o oversubscribe o contexto atual e cria segmentos adicionais dessa maneira.A classe de http_reader usa um objeto de concurrency::unbounded_buffer para limitar o número de segmentos que o aplicativo usa.O agente inicializa o buffer com um número fixo dos valores um.Para cada operação de download, o agente ler um valor simbólico de buffer antes que inicia e então grava o valor da operação de volta para o buffer após a conclusão da operação.Quando o buffer estiver vazia, o agente espera uma das operações de download para gravar um valor de volta para o buffer.
O exemplo limitar o número de tarefas simultâneas a duas vezes que o número de hardware disponível segmento.Esse valor é um bom ponto de partida para usar quando você experimenta com a sobresubscrição.Você pode usar um valor que preenche um determinado ambiente de processamento ou alterar dinamicamente esse valor para responder à carga de trabalho real.
// download-oversubscription.cpp
// compile with: /EHsc /MD /D "_AFXDLL"
#define _WIN32_WINNT 0x0501
#include <afxinet.h>
#include <concrtrm.h>
#include <agents.h>
#include <ppl.h>
#include <sstream>
#include <iostream>
#include <array>
using namespace concurrency;
using namespace std;
// Calls the provided work function and returns the number of milliseconds
// that it takes to call that function.
template <class Function>
__int64 time_call(Function&& f)
{
__int64 begin = GetTickCount();
f();
return GetTickCount() - begin;
}
// Downloads the file at the given URL.
CString GetHttpFile(CInternetSession& session, const CString& strUrl);
// Reads files from HTTP servers.
class http_reader : public agent
{
public:
explicit http_reader(CInternetSession& session,
ISource<string>& source,
unsigned int& total_bytes,
unsigned int max_concurrent_reads)
: _session(session)
, _source(source)
, _total_bytes(total_bytes)
{
// Add one token to the available tasks buffer for each
// possible concurrent read operation. The value of each token
// is not important, but can be useful for debugging.
for (unsigned int i = 0; i < max_concurrent_reads; ++i)
send(_available_tasks, i);
}
// Signals to the agent that there are no more items to download.
static const string input_sentinel;
protected:
void run()
{
// A task group. Each task in the group downloads one file.
task_group tasks;
// Holds the total number of bytes downloaded.
combinable<unsigned int> total_bytes;
// Read from the source buffer until the application
// sends the sentinel value.
string url;
while ((url = receive(_source)) != input_sentinel)
{
// Wait for a task to release an available slot.
unsigned int token = receive(_available_tasks);
// Create a task to download the file.
tasks.run([&, token, url] {
// Print a message.
wstringstream ss;
ss << L"Downloading " << url.c_str() << L"..." << endl;
wcout << ss.str();
// Download the file.
string content = download(url);
// Update the total number of bytes downloaded.
total_bytes.local() += content.size();
// Release the slot for another task.
send(_available_tasks, token);
});
}
// Wait for all tasks to finish.
tasks.wait();
// Compute the total number of bytes download on all threads.
_total_bytes = total_bytes.combine(plus<unsigned int>());
// Set the status of the agent to agent_done.
done();
}
// Downloads the file at the given URL.
string download(const string& url)
{
// Enable oversubscription.
Context::Oversubscribe(true);
// Download the file.
string content = GetHttpFile(_session, url.c_str());
// Disable oversubscription.
Context::Oversubscribe(false);
return content;
}
private:
// Manages the network connection.
CInternetSession& _session;
// A message buffer that holds the URL names to download.
ISource<string>& _source;
// The total number of bytes downloaded
unsigned int& _total_bytes;
// Limits the agent to a given number of simultaneous tasks.
unbounded_buffer<unsigned int> _available_tasks;
};
const string http_reader::input_sentinel("");
int wmain()
{
// Create an array of URL names to download.
// A real-world application might read the names from user input.
array<string, 21> urls = {
"http://www.adatum.com/",
"https://www.adventure-works.com/",
"http://www.alpineskihouse.com/",
"http://www.cpandl.com/",
"http://www.cohovineyard.com/",
"http://www.cohowinery.com/",
"http://www.cohovineyardandwinery.com/",
"https://www.contoso.com/",
"http://www.consolidatedmessenger.com/",
"http://www.fabrikam.com/",
"https://www.fourthcoffee.com/",
"http://www.graphicdesigninstitute.com/",
"http://www.humongousinsurance.com/",
"http://www.litwareinc.com/",
"http://www.lucernepublishing.com/",
"http://www.margiestravel.com/",
"http://www.northwindtraders.com/",
"https://www.proseware.com/",
"http://www.fineartschool.net",
"http://www.tailspintoys.com/",
http_reader::input_sentinel,
};
// Manages the network connection.
CInternetSession session("Microsoft Internet Browser");
// A message buffer that enables the application to send URL names to the
// agent.
unbounded_buffer<string> source_urls;
// The total number of bytes that the agent has downloaded.
unsigned int total_bytes = 0u;
// Create an http_reader object that can oversubscribe each processor by one.
http_reader reader(session, source_urls, total_bytes, 2*GetProcessorCount());
// Compute the amount of time that it takes for the agent to download all files.
__int64 elapsed = time_call([&] {
// Start the agent.
reader.start();
// Use the message buffer to send each URL name to the agent.
for_each(begin(urls), end(urls), [&](const string& url) {
send(source_urls, url);
});
// Wait for the agent to finish downloading.
agent::wait(&reader);
});
// Print the results.
wcout << L"Downloaded " << total_bytes
<< L" bytes in " << elapsed << " ms." << endl;
}
// Downloads the file at the given URL and returns the size of that file.
CString GetHttpFile(CInternetSession& session, const CString& strUrl)
{
CString strResult;
// Reads data from an HTTP server.
CHttpFile* pHttpFile = NULL;
try
{
// Open URL.
pHttpFile = (CHttpFile*)session.OpenURL(strUrl, 1,
INTERNET_FLAG_TRANSFER_ASCII |
INTERNET_FLAG_RELOAD | INTERNET_FLAG_DONT_CACHE);
// Read the file.
if(pHttpFile != NULL)
{
UINT uiBytesRead;
do
{
char chBuffer[10000];
uiBytesRead = pHttpFile->Read(chBuffer, sizeof(chBuffer));
strResult += chBuffer;
}
while (uiBytesRead > 0);
}
}
catch (CInternetException)
{
// TODO: Handle exception
}
// Clean up and return.
delete pHttpFile;
return strResult;
}
Este exemplo produz a seguinte saída em um computador que tem quatro processadores:
O exemplo pode ser executado mais rapidamente quando a sobresubscrição está ativado porque as tarefas adicionais executadas quando outras tarefas esperarem uma operação de para.
Compilando o código
Copie o código de exemplo e cole-o em um projeto do Visual Studio, ou cole em um arquivo denominado download-oversubscription.cpp e então executado dos seguintes comandos em uma janela de prompt de comando do Visual Studio.
cl.exe /EHsc /MD /D "_AFXDLL" download-oversubscription.cpp
cl.exe /EHsc /MT download-oversubscription.cpp
Programação robusta
Sempre sobresubscrição desative depois que você não requer a.Considere uma função que não trata uma exceção que é lançada por outra função.Se você não desativar a sobresubscrição antes que a função retorna, qualquer trabalho adicional paralelo também oversubscribe o contexto atual.
Você pode usar o padrão Aquisição de recurso é inicialização (RAII) para limitar a sobresubscrição a um determinado escopo.No padrão de RAII, uma estrutura de dados é atribuída na pilha.A estrutura de dados ou inicializa obtém um recurso quando é criada e o for ou versões esse recurso quando a estrutura de dados seja destruída.O padrão de RAII garante que o destrutor é chamado antes que o escopo delimitador sair.Como consequência, o recurso é gerenciado corretamente quando uma exceção é lançada ou quando uma função contiver várias instruções de return .
O exemplo a seguir define uma estrutura que é chamada scoped_blocking_signal.O construtor de estrutura de scoped_blocking_signal permite a sobresubscrição e o destrutor desativa a sobresubscrição.
struct scoped_blocking_signal
{
scoped_blocking_signal()
{
concurrency::Context::Oversubscribe(true);
}
~scoped_blocking_signal()
{
concurrency::Context::Oversubscribe(false);
}
};
O exemplo a seguir altera o corpo do método de download para usar RAII para garantir que a sobresubscrição está desativada antes que a função retorna.Essa técnica garante que o método seja manipulação de download seguro.
// Downloads the file at the given URL.
string download(const string& url)
{
scoped_blocking_signal signal;
// Download the file.
return string(GetHttpFile(_session, url.c_str()));
}
Consulte também
Referência
Método de Context::Oversubscribe