¿Cómo limitar la cantidad de operaciones de E / S asíncronas concurrentes?

115
// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };

// now let's send HTTP requests to each of these URLs in parallel
urls.AsParallel().ForAll(async (url) => {
    var client = new HttpClient();
    var html = await client.GetStringAsync(url);
});

Aquí está el problema, inicia más de 1000 solicitudes web simultáneas. ¿Existe una manera fácil de limitar la cantidad simultánea de estas solicitudes http asincrónicas? Para que no se descarguen más de 20 páginas web en un momento dado. ¿Cómo hacerlo de la manera más eficiente?

Codificador de duelo
fuente
2
¿En qué se diferencia de su pregunta anterior ?
svick
1
stackoverflow.com/questions/9290498/… Con un parámetro ParallelOptions.
Chris Disley
4
@ChrisDisley, esto solo paralelizará el lanzamiento de las solicitudes.
gastador
@svick tiene razón, ¿en qué se diferencia? por cierto, me encanta la respuesta que hay stackoverflow.com/a/10802883/66372
eglasius
3
Además lo HttpClientes IDisposable, y debes desecharlo, especialmente cuando vas a usar más de 1000 de ellos. HttpClientse puede utilizar como singleton para múltiples solicitudes.
Shimmy Weitzhandler

Respuestas:

161

Definitivamente puede hacer esto en las últimas versiones de async para .NET, usando .NET 4.5 Beta. La publicación anterior de 'usr' apunta a un buen artículo escrito por Stephen Toub, pero la noticia menos anunciada es que el semáforo asíncrono en realidad llegó a la versión Beta de .NET 4.5

Si miras nuestra querida SemaphoreSlimclase (que deberías usar ya que es más eficiente que la original Semaphore), ahora cuenta con la WaitAsync(...)serie de sobrecargas, con todos los argumentos esperados: intervalos de tiempo de espera, tokens de cancelación, todos tus amigos de programación habituales: )

Stephen también escribió una publicación de blog más reciente sobre las nuevas ventajas de .NET 4.5 que salieron con la versión beta, consulte Novedades del paralelismo en .NET 4.5 Beta .

Por último, aquí hay un código de muestra sobre cómo usar SemaphoreSlim para la limitación del método asíncrono:

public async Task MyOuterMethod()
{
    // let's say there is a list of 1000+ URLs
    var urls = { "http://google.com", "http://yahoo.com", ... };

    // now let's send HTTP requests to each of these URLs in parallel
    var allTasks = new List<Task>();
    var throttler = new SemaphoreSlim(initialCount: 20);
    foreach (var url in urls)
    {
        // do an async wait until we can schedule again
        await throttler.WaitAsync();

        // using Task.Run(...) to run the lambda in its own parallel
        // flow on the threadpool
        allTasks.Add(
            Task.Run(async () =>
            {
                try
                {
                    var client = new HttpClient();
                    var html = await client.GetStringAsync(url);
                }
                finally
                {
                    throttler.Release();
                }
            }));
    }

    // won't get here until all urls have been put into tasks
    await Task.WhenAll(allTasks);

    // won't get here until all tasks have completed in some way
    // (either success or exception)
}

Por último, pero probablemente una mención digna de mención, es una solución que utiliza programación basada en TPL. Puede crear tareas vinculadas a delegados en el TPL que aún no se han iniciado y permitir que un programador de tareas personalizado limite la simultaneidad. De hecho, aquí hay una muestra de MSDN:

Consulte también TaskScheduler .

Theo Yaung
fuente
3
no es un paralelo. ¿Cada uno con un grado limitado de paralelismo es un enfoque mejor? msdn.microsoft.com/en-us/library/…
GreyCloud
2
¿Por qué no te HttpClient
deshaces de
4
@GreyCloud: Parallel.ForEachfunciona con código síncrono. Esto le permite llamar a código asincrónico.
Josh Noe
2
@TheMonarch estás equivocado . Además, siempre es un buen hábito envolver todos los IDisposablemensajes usingo try-finallydeclaraciones y asegurar su eliminación.
Shimmy Weitzhandler
29
Dado lo popular que es esta respuesta, vale la pena señalar que HttpClient puede y debe ser una única instancia común en lugar de una instancia por solicitud.
Rupert Rawnsley
15

Si tiene un IEnumerable (es decir, cadenas de URL) y desea realizar una operación de enlace de E / S con cada uno de estos (es decir, realizar una solicitud http asíncrona) al mismo tiempo Y, opcionalmente, también desea establecer el número máximo de Solicitudes de E / S en tiempo real, así es como puede hacerlo. De esta manera, no usa el grupo de subprocesos y otros, el método usa semaphoreslim para controlar el máximo de solicitudes de E / S concurrentes similares a un patrón de ventana deslizante que una solicitud completa, deja el semáforo y la siguiente ingresa.

uso: espera ForEachAsync (urlStrings, YourAsyncFunc, OptionalMaxDegreeOfConcurrency);

public static Task ForEachAsync<TIn>(
        IEnumerable<TIn> inputEnumerable,
        Func<TIn, Task> asyncProcessor,
        int? maxDegreeOfParallelism = null)
    {
        int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism;
        SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);

        IEnumerable<Task> tasks = inputEnumerable.Select(async input =>
        {
            await throttler.WaitAsync().ConfigureAwait(false);
            try
            {
                await asyncProcessor(input).ConfigureAwait(false);
            }
            finally
            {
                throttler.Release();
            }
        });

        return Task.WhenAll(tasks);
    }
Dogu Arslan
fuente
no, no debería necesitar eliminar explícitamente SemaphoreSlim en esta implementación y uso, ya que se usa internamente dentro del método y el método no accede a su propiedad AvailableWaitHandle, en cuyo caso habríamos necesitado eliminarlo o envolverlo dentro de un bloque using.
Dogu Arslan
1
Solo pensando en las mejores prácticas y lecciones que enseñamos a otras personas. Una usingsería bueno.
AgentFire
bueno, puedo seguir este ejemplo, pero tratando de averiguar cuál es la mejor manera de hacer esto, básicamente tengo un acelerador, pero mi Func devolvería una lista, que finalmente quiero en una lista final de todos los completados cuando termine ... que puede requiere bloqueado en la lista, ¿tiene alguna sugerencia?
Seabizkit
puedes actualizar ligeramente el método para que devuelva la lista de tareas reales y esperas Task.WhenAll desde el interior de tu código de llamada. Una vez que Task.WhenAll esté completo, puede enumerar cada tarea en la lista y agregar su lista a la lista final. Cambie la firma del método a 'public static IEnumerable <Task <TOut>> ForEachAsync <TIn, TOut> (IEnumerable <TIn> inputEnumerable, Func <TIn, Task <TOut>> asyncProcessor, int? MaxDegreeOfParallelism = null)'
Dogu Arslan
7

Desafortunadamente, .NET Framework carece de los combinadores más importantes para orquestar tareas asíncronas en paralelo. No hay tal cosa incorporada.

Mire la clase AsyncSemaphore construida por el más respetable Stephen Toub. Lo que quieres se llama semáforo y necesitas una versión asíncrona.

usr
fuente
12
Tenga en cuenta que "Desafortunadamente, a .NET Framework le faltan los combinadores más importantes para orquestar tareas asíncronas en paralelo. No existe tal cosa integrada". ya no es correcto a partir de .NET 4.5 Beta. SemaphoreSlim ahora ofrece la funcionalidad WaitAsync (...) :)
Theo Yaung
¿Debería preferirse SemaphoreSlim (con sus nuevos métodos asíncronos) sobre AsyncSemphore, o la implementación de Toub todavía tiene alguna ventaja?
Todd Menier
En mi opinión, se debe preferir el tipo integrado porque es probable que esté bien probado y bien diseñado.
usr
4
Stephen agregó un comentario en respuesta a una pregunta en la publicación de su blog que confirma que usar SemaphoreSlim para .NET 4.5 generalmente sería el camino a seguir.
jdasilva
7

Hay muchas trampas y el uso directo de un semáforo puede ser complicado en casos de error, por lo que sugeriría usar el paquete AsyncEnumerator NuGet en lugar de reinventar la rueda:

// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };

// now let's send HTTP requests to each of these URLs in parallel
await urls.ParallelForEachAsync(async (url) => {
    var client = new HttpClient();
    var html = await client.GetStringAsync(url);
}, maxDegreeOfParalellism: 20);
Serge Semenov
fuente
4

El ejemplo de Theo Yaung es bueno, pero hay una variante sin lista de tareas en espera.

 class SomeChecker
 {
    private const int ThreadCount=20;
    private CountdownEvent _countdownEvent;
    private SemaphoreSlim _throttler;

    public Task Check(IList<string> urls)
    {
        _countdownEvent = new CountdownEvent(urls.Count);
        _throttler = new SemaphoreSlim(ThreadCount); 

        return Task.Run( // prevent UI thread lock
            async  () =>{
                foreach (var url in urls)
                {
                    // do an async wait until we can schedule again
                    await _throttler.WaitAsync();
                    ProccessUrl(url); // NOT await
                }
                //instead of await Task.WhenAll(allTasks);
                _countdownEvent.Wait();
            });
    }

    private async Task ProccessUrl(string url)
    {
        try
        {
            var page = await new WebClient()
                       .DownloadStringTaskAsync(new Uri(url)); 
            ProccessResult(page);
        }
        finally
        {
            _throttler.Release();
            _countdownEvent.Signal();
        }
    }

    private void ProccessResult(string page){/*....*/}
}
vitidev
fuente
4
Tenga en cuenta que existe un peligro de utilizar este enfoque: cualquier excepción que ocurra en ProccessUrlo sus subfunciones será ignorada. Se capturarán en Tareas, pero no se filtrarán al llamador original de Check(...). Personalmente, es por eso que sigo usando Tasks y sus funciones de combinación como WhenAlly WhenAny- para obtener una mejor propagación de errores. :)
Theo Yaung
3

SemaphoreSlim puede ser muy útil aquí. Aquí está el método de extensión que he creado.

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxActionsToRunInParallel">Optional, max numbers of the actions to run in parallel,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxActionsToRunInParallel = null)
    {
        if (maxActionsToRunInParallel.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all of the provided tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

Uso de muestra:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
Jay Shah
fuente
0

Antigua pregunta, nueva respuesta. @vitidev tenía un bloque de código que se reutilizó casi intacto en un proyecto que revisé. Después de discutir con algunos colegas, uno preguntó "¿Por qué no usa los métodos TPL integrados?" ActionBlock parece el ganador allí. https://msdn.microsoft.com/en-us/library/hh194773(v=vs.110).aspx . Probablemente no terminará cambiando ningún código existente, pero definitivamente buscará adoptar este nuget y reutilizar las mejores prácticas del Sr. Softy para el paralelismo acelerado.

Sin reembolsos Sin devoluciones
fuente
0

Aquí hay una solución que aprovecha la naturaleza perezosa de LINQ. Es funcionalmente equivalente a la respuesta aceptada ), pero usa las tareas de los trabajadores en lugar de a SemaphoreSlim, reduciendo de esta manera la huella de memoria de toda la operación. Al principio, hagamos que funcione sin estrangulamiento. El primer paso es convertir nuestras URL en una enumeración de tareas.

string[] urls =
{
    "https://stackoverflow.com",
    "https://superuser.com",
    "https://serverfault.com",
    "https://meta.stackexchange.com",
    // ...
};
var httpClient = new HttpClient();
var tasks = urls.Select(async (url) =>
{
    return (Url: url, Html: await httpClient.GetStringAsync(url));
});

El segundo paso es realizar awaittodas las tareas al mismo tiempo utilizando el Task.WhenAllmétodo:

var results = await Task.WhenAll(tasks);
foreach (var result in results)
{
    Console.WriteLine($"Url: {result.Url}, {result.Html.Length:#,0} chars");
}

Salida:

Url: https://stackoverflow.com , 105.574 caracteres
Url: https://superuser.com , 126.953 caracteres
Url: https://serverfault.com , 125.963 caracteres
Url: https://meta.stackexchange.com , 185.276 caracteres
...

La implementación de Microsoft de Task.WhenAllmaterializa instantáneamente el enumerable proporcionado en una matriz, lo que hace que todas las tareas se inicien a la vez. No queremos eso, porque queremos limitar el número de operaciones asincrónicas concurrentes. Así que necesitaremos implementar una alternativa WhenAllque enumere nuestro enumerable de manera suave y lenta. Lo haremos creando una cantidad de tareas de trabajo (igual al nivel deseado de simultaneidad), y cada tarea de trabajador enumerará nuestra tarea enumerable una a la vez, usando un candado para asegurar que cada tarea de URL sea procesada por una sola tarea de trabajador. Luego, awaitpara que se completen todas las tareas de los trabajadores, y finalmente devolvemos los resultados. Aquí está la implementación:

public static async Task<T[]> WhenAll<T>(IEnumerable<Task<T>> tasks,
    int concurrencyLevel)
{
    if (tasks is ICollection<Task<T>>) throw new ArgumentException(
        "The enumerable should not be materialized.", nameof(tasks));
    var locker = new object();
    var results = new List<T>();
    var failed = false;
    using (var enumerator = tasks.GetEnumerator())
    {
        var workerTasks = Enumerable.Range(0, concurrencyLevel)
        .Select(async _ =>
        {
            try
            {
                while (true)
                {
                    Task<T> task;
                    int index;
                    lock (locker)
                    {
                        if (failed) break;
                        if (!enumerator.MoveNext()) break;
                        task = enumerator.Current;
                        index = results.Count;
                        results.Add(default); // Reserve space in the list
                    }
                    var result = await task.ConfigureAwait(false);
                    lock (locker) results[index] = result;
                }
            }
            catch (Exception)
            {
                lock (locker) failed = true;
                throw;
            }
        }).ToArray();
        await Task.WhenAll(workerTasks).ConfigureAwait(false);
    }
    lock (locker) return results.ToArray();
}

... y esto es lo que debemos cambiar en nuestro código inicial, para lograr el estrangulamiento deseado:

var results = await WhenAll(tasks, concurrencyLevel: 2);

Existe una diferencia con respecto al manejo de las excepciones. El nativo Task.WhenAllespera a que se completen todas las tareas y agrega todas las excepciones. La implementación anterior finaliza inmediatamente después de completar la primera tarea con errores.

Theodor Zoulias
fuente
La implementación de AC # 8 que devuelve un IAsyncEnumerable<T>se puede encontrar aquí .
Theodor Zoulias
-1

Aunque es posible que se pongan en cola 1000 tareas muy rápidamente, la biblioteca de Tareas paralelas solo puede manejar tareas simultáneas iguales a la cantidad de núcleos de CPU en la máquina. Eso significa que si tiene una máquina de cuatro núcleos, solo se ejecutarán 4 tareas en un momento dado (a menos que reduzca el MaxDegreeOfParallelism).

Scottm
fuente
8
Sí, pero eso no se relaciona con las operaciones de E / S asíncronas. El código anterior activará más de 1000 descargas simultáneas incluso si se ejecuta en un solo hilo.
Grief Coder
No vi la awaitpalabra clave allí. Eliminar eso debería resolver el problema, ¿correcto?
scottm
2
La biblioteca ciertamente puede manejar más tareas en ejecución (con el Runningestado) al mismo tiempo que la cantidad de núcleos. Este será especialmente el caso con tareas vinculadas de E / S.
svick
@svick: sí. ¿Sabe cómo controlar de manera eficiente el máximo de tareas TPL concurrentes (no subprocesos)?
Grief Coder
-1

Deben utilizarse cálculos en paralelo para acelerar las operaciones vinculadas a la CPU. Aquí estamos hablando de operaciones vinculadas de E / S. Su implementación debe ser puramente asincrónica , a menos que esté abrumando el ocupado núcleo único en su CPU de múltiples núcleos.

EDITAR Me gusta la sugerencia hecha por usr de usar un "semáforo asíncrono" aquí.

GregC
fuente
¡Buen punto! Aunque cada tarea aquí contendrá código asincrónico y de sincronización (la página se descarga de forma asincrónica y luego se procesa de manera sincronizada). Estoy tratando de distribuir la parte de sincronización del código entre las CPU y, al mismo tiempo, limitar la cantidad de operaciones de E / S asíncronas simultáneas.
Grief Coder
¿Por qué? Porque lanzar más de 1000 solicitudes http simultáneamente podría no ser una tarea adecuada para la capacidad de red del usuario.
gastador
Las extensiones paralelas también se pueden utilizar como una forma de multiplexar operaciones de E / S sin tener que implementar manualmente una solución asincrónica pura. Lo que estoy de acuerdo podría considerarse descuidado, pero siempre que mantenga un límite estricto en el número de operaciones simultáneas, probablemente no forzará demasiado el hilo.
Sean U
3
No creo que esta respuesta proporcione una respuesta. Ser puramente asincrónico no es suficiente aquí: realmente queremos acelerar los IO físicos de una manera sin bloqueo.
usr
1
Hmm ... no estoy seguro de estar de acuerdo ... cuando se trabaja en un proyecto grande, si demasiados desarrolladores adoptan este punto de vista, se morirá de hambre aunque la contribución de cada desarrollador de forma aislada no sea suficiente para llevar las cosas al límite. Dado que solo hay un ThreadPool, incluso si lo está tratando con semi-respeto ... si todos los demás están haciendo lo mismo, pueden surgir problemas. Como tal, siempre aconsejo no ejecutar cosas largas en ThreadPool.
gasto
-1

Use MaxDegreeOfParallelism, que es una opción que puede especificar en Parallel.ForEach():

var options = new ParallelOptions { MaxDegreeOfParallelism = 20 };

Parallel.ForEach(urls, options,
    url =>
        {
            var client = new HttpClient();
            var html = client.GetStringAsync(url);
            // do stuff with html
        });
Sean U
fuente
4
No creo que esto funcione. GetStringAsync(url)está destinado a ser llamado con await. Si inspecciona el tipo de var html, es un Task<string>, no el resultado string.
Neal Ehardt
2
@NealEhardt tiene razón. Parallel.ForEach(...)está diseñado para ejecutar bloques de código síncrono en paralelo (por ejemplo, en diferentes subprocesos).
Theo Yaung
-1

Esencialmente, querrá crear una Acción o Tarea para cada URL que desee ingresar, ponerlas en una Lista y luego procesar esa lista, limitando el número que se puede procesar en paralelo.

Mi entrada de blog muestra cómo hacer esto tanto con Tareas como con Acciones, y proporciona un proyecto de muestra que puede descargar y ejecutar para ver ambos en acción.

Con acciones

Si usa Acciones, puede usar la función incorporada .Net Parallel.Invoke. Aquí lo limitamos a ejecutar como máximo 20 subprocesos en paralelo.

var listOfActions = new List<Action>();
foreach (var url in urls)
{
    var localUrl = url;
    // Note that we create the Task here, but do not start it.
    listOfTasks.Add(new Task(() => CallUrl(localUrl)));
}

var options = new ParallelOptions {MaxDegreeOfParallelism = 20};
Parallel.Invoke(options, listOfActions.ToArray());

Con tareas

Con Tasks no hay una función incorporada. Sin embargo, puede utilizar el que proporciono en mi blog.

    /// <summary>
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel.
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
    /// </summary>
    /// <param name="tasksToRun">The tasks to run.</param>
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken())
    {
        await StartAndWaitAllThrottledAsync(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken);
    }

    /// <summary>
    /// Starts the given tasks and waits for them to complete. This will run the specified number of tasks in parallel.
    /// <para>NOTE: If a timeout is reached before the Task completes, another Task may be started, potentially running more than the specified maximum allowed.</para>
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
    /// </summary>
    /// <param name="tasksToRun">The tasks to run.</param>
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
    /// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken())
    {
        // Convert to a list of tasks so that we don't enumerate over it multiple times needlessly.
        var tasks = tasksToRun.ToList();

        using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel))
        {
            var postTaskTasks = new List<Task>();

            // Have each task notify the throttler when it completes so that it decrements the number of tasks currently running.
            tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release())));

            // Start running each task.
            foreach (var task in tasks)
            {
                // Increment the number of tasks currently running and wait if too many are running.
                await throttler.WaitAsync(timeoutInMilliseconds, cancellationToken);

                cancellationToken.ThrowIfCancellationRequested();
                task.Start();
            }

            // Wait for all of the provided tasks to complete.
            // We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler's using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object.
            await Task.WhenAll(postTaskTasks.ToArray());
        }
    }

Y luego, creando su lista de tareas y llamando a la función para que se ejecuten, con un máximo de 20 simultáneas a la vez, podría hacer esto:

var listOfTasks = new List<Task>();
foreach (var url in urls)
{
    var localUrl = url;
    // Note that we create the Task here, but do not start it.
    listOfTasks.Add(new Task(async () => await CallUrl(localUrl)));
}
await Tasks.StartAndWaitAllThrottledAsync(listOfTasks, 20);
perro mortal
fuente
Creo que solo está especificando initialCount para SemaphoreSlim y necesita especificar el segundo parámetro, es decir, maxCount en el constructor de SemaphoreSlim.
Jay Shah
Quiero que cada respuesta de cada tarea se procese en una lista. ¿Cómo puedo obtener un resultado de devolución o una respuesta?
venkat
-1

esta no es una buena práctica ya que cambia una variable global. tampoco es una solución general para async. pero es fácil para todas las instancias de HttpClient, si eso es todo lo que busca. simplemente puedes probar:

System.Net.ServicePointManager.DefaultConnectionLimit = 20;
simbionte
fuente