Procedura: utilizzare l'oversubscription per compensare la latenza
L'oversubscription può migliorare l'efficienza complessiva di alcune applicazioni che contengono attività che hanno un'eccessiva quantità di latenza. In questo argomento viene illustrato come utilizzare l'oversubscription per compensare la latenza causata dalla lettura dei dati da una connessione di rete.
Esempio
In questo esempio viene utilizzata la Libreria di agenti asincroni per scaricare file dai server HTTP. La classe http_reader deriva da Concurrency::agent e utilizza il passaggio dei messaggi per leggere in modo asincrono i nomi URL da scaricare.
La classe http_reader utilizza la classe Concurrency::task_group per leggere simultaneamente ciascun file. Ogni attività chiama il metodo Concurrency::Context::Oversubscribe con il parametro _BeginOversubscription impostato su true per abilitare l'oversubscription nel contesto corrente. Ogni attività utilizza quindi le classi MFC (Microsoft Foundation Classes) CInternetSession e CHttpFile per scaricare il file. Infine ogni attività chiama Context::Oversubscribe con il parametro _BeginOversubscription impostato a false per disabilitare l'oversubscription.
Quando l'oversubscription è abilitato, il runtime crea un thread aggiuntivo in cui eseguire le attività. Ognuno di questi thread può inoltre abilitare l'oversubscription nel contesto corrente creando quindi thread aggiuntivi. La classe http_reader utilizza un oggetto Concurrency::unbounded_buffer per limitare il numero di thread utilizzati dall'applicazione. L'agente inizializza il buffer con un numero fisso di valori token. Per ogni operazione di download, l'agente legge un valore token dal buffer prima dell'avvio dell'operazione, quindi scrive il valore di nuovo nel buffer dopo il completamento dell'operazione. Quando il buffer è vuoto, l'agente attende che una delle operazioni di download scriva un valore di nuovo nel buffer.
Nell'esempio seguente il numero di attività simultanee viene limitato a due volte il numero di thread di hardware disponibili. Questo valore rappresenta un buon punto di partenza da utilizzare con l'oversubscription. È possibile utilizzare un valore appropriato per uno specifico ambiente di elaborazione o modificare dinamicamente questo valore per rispondere al carico di lavoro effettivo.
// 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(urls.begin(), urls.end(), [&](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;
}
L'output di esempio seguente è relativo a un computer con quattro processori.
Downloading http://www.adatum.com/...
Downloading https://www.adventure-works.com/...
Downloading http://www.alpineskihouse.com/...
Downloading http://www.cpandl.com/...
Downloading http://www.cohovineyard.com/...
Downloading http://www.cohowinery.com/...
Downloading http://www.cohovineyardandwinery.com/...
Downloading https://www.contoso.com/...
Downloading http://www.consolidatedmessenger.com/...
Downloading http://www.fabrikam.com/...
Downloading https://www.fourthcoffee.com/...
Downloading http://www.graphicdesigninstitute.com/...
Downloading http://www.humongousinsurance.com/...
Downloading http://www.litwareinc.com/...
Downloading http://www.lucernepublishing.com/...
Downloading http://www.margiestravel.com/...
Downloading http://www.northwindtraders.com/...
Downloading https://www.proseware.com/...
Downloading http://www.fineartschool.net...
Downloading http://www.tailspintoys.com/...
Downloaded 1801040 bytes in 3276 ms.
L'esempio può essere eseguito più velocemente quando è abilitato l'oversubscription poiché le attività aggiuntive vengono eseguite mentre le altre attività attendono il completamento di un'operazione latente.
Compilazione del codice
Copiare il codice di esempio e incollarlo in un progetto di Visual Studio o incollarlo in un file denominato download-oversubscription.cpp, quindi eseguire uno dei comandi seguenti in una finestra del prompt dei comandi di Visual Studio 2010.
cl.exe /EHsc /MD /D "_AFXDLL" download-oversubscription.cpp
cl.exe /EHsc /MT download-oversubscription.cpp
Programmazione efficiente
Disabilitare sempre l'oversubscription quando non è più richiesta. Considerare una funzione che non gestisce un'eccezione generata da un'altra funzione. Se non si disabilita l'oversubscription prima della restituzione della funzione, l'oversubscription verrà inoltre abilitato per qualsiasi lavoro parallelo aggiuntivo nel contesto corrente.
È possibile utilizzare il modello RAII (Resource Acquisition Is Initialization) per limitare l'oversubscription a un ambito specificato. In base al modello RAII, una struttura dei dati viene allocata sullo stack. La struttura dei dati inizializza o acquisisce una risorsa quando viene creata ed elimina o rilascia tale risorsa quando la struttura dei dati viene eliminata. Il modello RAII garantisce che il distruttore venga chiamato prima della chiusura dell'ambito che lo contiene. Pertanto, la risorsa viene gestita correttamente quando viene generata un'eccezione o quando una funzione contiene più istruzioni return.
Nell'esempio seguente viene definita una struttura denominata scoped_blocking_signal. Il costruttore della struttura scoped_blocking_signal abilita l'oversubscription mentre il distruttore la disabilita.
struct scoped_blocking_signal
{
scoped_blocking_signal()
{
Concurrency::Context::Oversubscribe(true);
}
~scoped_blocking_signal()
{
Concurrency::Context::Oversubscribe(false);
}
};
Nell'esempio seguente viene modificato il corpo del metodo download in modo da utilizzare il modello RAII per assicurarsi che l'oversubscription venga disabilitata prima del termine della funzione. Questa tecnica garantisce che il metodo download venga eseguito correttamente indipendentemente dalle eccezioni.
// 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()));
}