from typing import AsyncGenerator, ContextManager, TypeVar

import anyio
from anyio import CapacityLimiter
from starlette.concurrency import iterate_in_threadpool as iterate_in_threadpool  # noqa
from starlette.concurrency import run_in_threadpool as run_in_threadpool  # noqa
from starlette.concurrency import (  # noqa
    run_until_first_complete as run_until_first_complete,
)

_T = TypeVar("_T")


@asynccontextmanager

Concurrency is not parallelism....

Concurrency
===========

This document describes the concurrency considerations for http/2 connections and the connection pool within OkHttp.

## HTTP/2 Connections

The HttpURLConnection API is a blocking API. You make a blocking write to send a request, and a blocking read to receive the response.

#### Blocking APIs

    
    In this release, concurrency configuration has been tweaked to be consistent across deployment types.
    The new logic will use the `ProxyConfig.Concurrency` setting (which can be configured mesh wide or per-pod), if set,
    and otherwise set concurrency based on the CPU limit allocated to the container
    For example, a limit of 2500m would set concurrency to 3.
    

	}
}

func TestXGlobal(t *testing.T) {
	testFiles(t, global, "testdata/hello.txt", "hello, world\n")
	testString(t, concurrency, "concurrency", "Concurrency is not parallelism.\n")
	testString(t, string(glass), "glass", "I can eat glass and it doesn't hurt me.\n")
	testString(t, concurrency2, "concurrency2", "Concurrency is not parallelism.\n")
	testString(t, string(glass2), "glass2", "I can eat glass and it doesn't hurt me.\n")

		expected *meshconfig.ProxyConfig
	}{
		{
			name: "concurrency",
			first: &meshconfig.ProxyConfig{
				Concurrency: v(1),
			},
			second: &meshconfig.ProxyConfig{
				Concurrency: v(2),
			},
			expected: &meshconfig.ProxyConfig{
				Concurrency: v(1),
			},
		},
		{
			name: "concurrency value 0",
			first: &meshconfig.ProxyConfig{
				Concurrency: v(0),
			},
			second: &meshconfig.ProxyConfig{

	go func() {
		defer xioutil.SafeClose(ch)

		concurrency := opts.concurrencyStart

		if opts.autotune {
			// if we have less drives than concurrency then choose
			// only the concurrency to be number of drives to start
			// with - since default '32' might be big and may not
			// complete in total time of 10s.
			if globalEndpoints.NEndpoints() < concurrency {
				concurrency = globalEndpoints.NEndpoints()
			}

        int concurrency = 20
        AtomicInteger total = addAndGetLazyConcurrently(concurrency, lazy)

        then: 'all threads observe the same value'
        total.get() % concurrency == 0
    }

    private AtomicInteger addAndGetLazyConcurrently(int concurrency, Lazy<Integer> lazy) {
        def total = new AtomicInteger()

  // methods is identical, save for method being invoked.

  /** Invokes {@code latch.}{@link CountDownLatch#await() await()} uninterruptibly. */
  @J2ktIncompatible
  @GwtIncompatible // concurrency
  public static void awaitUninterruptibly(CountDownLatch latch) {
    boolean interrupted = false;
    try {
      while (true) {
        try {
          latch.await();
          return;

    @Timeout(value = 5, unit = TimeUnit.SECONDS)
    def "supports #concurrency concurrent computations"() {
        def latch = new CountDownLatch(concurrency)
        def executor = Executors.newFixedThreadPool(concurrency)
        def cache = new ManualEvictionInMemoryCache<String, String>()

        when:
        concurrency.times { idx ->
            executor.submit {
                String key = "key${idx}"