diff --git a/async_generator/__init__.py b/async_generator/__init__.py
index e81a8fe..9c27da2 100644
--- a/async_generator/__init__.py
+++ b/async_generator/__init__.py
@@ -7,11 +7,18 @@
     isasyncgenfunction,
     get_asyncgen_hooks,
     set_asyncgen_hooks,
+    supports_native_asyncgens,
 )
 from ._util import aclosing, asynccontextmanager
+from functools import partial as _partial
+
+async_generator_native = _partial(async_generator, allow_native=True)
+async_generator_legacy = _partial(async_generator, allow_native=False)
 
 __all__ = [
     "async_generator",
+    "async_generator_native",
+    "async_generator_legacy",
     "yield_",
     "yield_from_",
     "aclosing",
@@ -20,4 +27,5 @@
     "asynccontextmanager",
     "get_asyncgen_hooks",
     "set_asyncgen_hooks",
+    "supports_native_asyncgens",
 ]
diff --git a/async_generator/_impl.py b/async_generator/_impl.py
index f35b07f..dcb9bf3 100644
--- a/async_generator/_impl.py
+++ b/async_generator/_impl.py
@@ -1,6 +1,6 @@
 import sys
-from functools import wraps
-from types import coroutine, CodeType
+from functools import wraps, partial
+from types import coroutine, CodeType, FunctionType
 import inspect
 from inspect import (
     getcoroutinestate, CORO_CREATED, CORO_CLOSED, CORO_SUSPENDED
@@ -50,8 +50,11 @@ def inner():
 
 if sys.implementation.name == "cpython" and sys.version_info >= (3, 6):
     # On 3.6, with native async generators, we want to use the same
-    # wrapper type that native generators use. This lets @async_generators
-    # yield_from_ native async generators and vice versa.
+    # wrapper type that native generators use. This lets @async_generator
+    # create a native async generator under most circumstances, which
+    # improves performance while still permitting 3.5-compatible syntax.
+    # It also lets non-native @async_generators (e.g. those that return
+    # non-None values) yield_from_ native ones and vice versa.
 
     import ctypes
     from types import AsyncGeneratorType, GeneratorType
@@ -315,10 +318,11 @@ def set_asyncgen_hooks(firstiter=UNSPECIFIED, finalizer=UNSPECIFIED):
 
 
 class AsyncGenerator:
-    def __init__(self, coroutine):
+    def __init__(self, coroutine, *, warn_on_native_differences=False):
         self._coroutine = coroutine
+        self._warn_on_native_differences = warn_on_native_differences
         self._it = coroutine.__await__()
-        self.ag_running = False
+        self._running = False
         self._finalizer = None
         self._closed = False
         self._hooks_inited = False
@@ -351,6 +355,28 @@ def ag_code(self):
     def ag_frame(self):
         return self._coroutine.cr_frame
 
+    @property
+    def ag_await(self):
+        return self._coroutine.cr_await
+
+    @property
+    def ag_running(self):
+        if self._running != self._coroutine.cr_running and self._warn_on_native_differences:
+            import warnings
+            warnings.warn(
+                "Native async generators incorrectly set ag_running = False "
+                "when the generator is awaiting a trap to the event loop and "
+                "not suspended via a yield to its caller. Your code examines "
+                "ag_running under such conditions, and will change behavior "
+                "when async_generator starts using native generators by default "
+                "(where available) in the next release. "
+                "Use @async_generator_legacy to keep the current behavior, or "
+                "@async_generator_native if you're OK with the change.",
+                category=FutureWarning,
+                stacklevel=2
+            )
+        return self._running
+
     ################################################################
     # Core functionality
     ################################################################
@@ -387,10 +413,10 @@ async def step():
             if self.ag_running:
                 raise ValueError("async generator already executing")
             try:
-                self.ag_running = True
+                self._running = True
                 return await ANextIter(self._it, start_fn, *args)
             finally:
-                self.ag_running = False
+                self._running = False
 
         return step()
 
@@ -462,10 +488,128 @@ def __del__(self):
     collections.abc.AsyncGenerator.register(AsyncGenerator)
 
 
-def async_generator(coroutine_maker):
-    @wraps(coroutine_maker)
+def _find_return_of_not_none(co):
+    """Inspect the code object *co* for the presence of return statements that
+    might return a value other than None. If any such statement is found,
+    return the source line number (in file ``co.co_filename``) on which it occurs.
+    If all return statements definitely return the value None, return None.
+    """
+
+    # 'return X' for simple/constant X seems to always compile to
+    # LOAD_CONST + RETURN_VALUE immediately following one another,
+    # and LOAD_CONST(None) + RETURN_VALUE definitely does mean return None,
+    # so we'll search for RETURN_VALUE not preceded by LOAD_CONST or preceded
+    # by a LOAD_CONST that does not load None.
+    import dis
+    current_line = co.co_firstlineno
+    prev_inst = None
+    for inst in dis.Bytecode(co):
+        if inst.starts_line is not None:
+            current_line = inst.starts_line
+        if inst.opname == "RETURN_VALUE" and (prev_inst is None or
+                                              prev_inst.opname != "LOAD_CONST"
+                                              or prev_inst.argval is not None):
+            return current_line
+        prev_inst = inst
+    return None
+
+
+def _as_native_asyncgen_function(afn):
+    """Given a non-generator async function *afn*, which contains some ``await yield_(...)``
+    and/or ``await yield_from_(...)`` calls, create the analogous async generator function.
+    *afn* must not return values other than None; doing so is likely to crash the interpreter.
+    Use :func:`_find_return_of_not_none` to check this.
+    """
+
+    # An async function that contains 'await yield_()' statements is a perfectly
+    # cromulent async generator, except that it doesn't get marked with CO_ASYNC_GENERATOR
+    # because it doesn't contain any 'yield' statements. Create a new code object that
+    # does have CO_ASYNC_GENERATOR set. This is preferable to using a wrapper because
+    # it makes ag_code and ag_frame work the same way they would for a native async
+    # generator with 'yield' statements, and the same as for an async_generator
+    # asyncgen with allow_native=False.
+
+    from inspect import CO_COROUTINE, CO_ASYNC_GENERATOR
+    co = afn.__code__
+    new_code = CodeType(
+        co.co_argcount, co.co_kwonlyargcount, co.co_nlocals, co.co_stacksize,
+        (co.co_flags & ~CO_COROUTINE) | CO_ASYNC_GENERATOR, co.co_code,
+        co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name,
+        co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars
+    )
+    asyncgen_fn = FunctionType(
+        new_code, afn.__globals__, afn.__name__, afn.__defaults__,
+        afn.__closure__
+    )
+    asyncgen_fn.__kwdefaults__ = afn.__kwdefaults__
+    return wraps(afn)(asyncgen_fn)
+
+
+def async_generator(afn=None, *, allow_native=None, uses_return=None):
+    if afn is None:
+        return partial(
+            async_generator,
+            uses_return=uses_return,
+            allow_native=allow_native
+        )
+
+    uses_wrapper = False
+    if not inspect.iscoroutinefunction(afn):
+        underlying = afn
+        while hasattr(underlying, "__wrapped__"):
+            underlying = getattr(underlying, "__wrapped__")
+            if inspect.iscoroutinefunction(underlying):
+                break
+        else:
+            raise TypeError(
+                "expected an async function, not {!r}".format(
+                    type(afn).__name__
+                )
+            )
+        # A sync wrapper around an async function is fine, in the sense
+        # that we can call it to get a coroutine just like we could for
+        # an async function; but it's a bit suboptimal, in the sense that
+        # we can't turn it into an async generator. One way to get here
+        # is to put trio's @enable_ki_protection decorator below
+        # @async_generator rather than above it.
+        uses_wrapper = True
+
+    if sys.implementation.name == "cpython" and not uses_wrapper:
+        # 'return' statements with non-None arguments are syntactically forbidden when
+        # compiling a true async generator, but the flags mutation in
+        # _convert_to_native_asyncgen_function sidesteps that check, which could raise
+        # an assertion in genobject.c when a non-None return is executed.
+        # To avoid crashing the interpreter due to a user error, we need to examine the
+        # code of the function we're converting.
+
+        co = afn.__code__
+        nontrivial_return_line = _find_return_of_not_none(co)
+        seems_to_use_return = nontrivial_return_line is not None
+        if uses_return is None:
+            uses_return = seems_to_use_return
+        elif uses_return != seems_to_use_return:
+            prefix = "{} declared using @async_generator(uses_return={}) but ".format(
+                afn.__qualname__, uses_return
+            )
+            if seems_to_use_return:
+                raise RuntimeError(
+                    prefix + "might return a value other than None at {}:{}".
+                    format(co.co_filename, nontrivial_return_line)
+                )
+            else:
+                raise RuntimeError(
+                    prefix + "never returns a value other than None"
+                )
+
+        if allow_native and not uses_return and supports_native_asyncgens:
+            return _as_native_asyncgen_function(afn)
+
+    @wraps(afn)
     def async_generator_maker(*args, **kwargs):
-        return AsyncGenerator(coroutine_maker(*args, **kwargs))
+        return AsyncGenerator(
+            afn(*args, **kwargs),
+            warn_on_native_differences=(allow_native is not False)
+        )
 
     async_generator_maker._async_gen_function = id(async_generator_maker)
     return async_generator_maker
diff --git a/async_generator/_tests/conftest.py b/async_generator/_tests/conftest.py
index 6c42d45..09d5072 100644
--- a/async_generator/_tests/conftest.py
+++ b/async_generator/_tests/conftest.py
@@ -34,3 +34,41 @@ def wrapper(**kwargs):
                 pass
 
         pyfuncitem.obj = wrapper
+
+
+# On CPython 3.6+, tests that accept an 'async_generator' fixture
+# will be called twice, once to test behavior with legacy pure-Python
+# async generators (i.e. _impl.AsyncGenerator) and once to test behavior
+# with wrapped native async generators. Tests should decorate their
+# asyncgens with their local @async_generator, and may inspect
+# async_generator.is_native to know which sort is being used.
+# On CPython 3.5 or PyPy, where native async generators do not
+# exist, async_generator will always call _impl.async_generator()
+# and tests will be invoked only once.
+
+from .. import supports_native_asyncgens
+maybe_native = pytest.param(
+    "native",
+    marks=pytest.mark.skipif(
+        not supports_native_asyncgens,
+        reason="native async generators are not supported on this Python version"
+    )
+)
+
+
+@pytest.fixture(params=["legacy", maybe_native])
+def async_generator(request):
+    from .. import async_generator as real_async_generator
+
+    def wrapper(afn=None, *, uses_return=False):
+        if afn is None:
+            return partial(wrapper, uses_return=uses_return)
+        if request.param == "native":
+            return real_async_generator(afn, allow_native=True)
+        # make sure the uses_return= argument matches what async_generator detects
+        real_async_generator(afn, allow_native=False, uses_return=uses_return)
+        # make sure autodetection without uses_return= works too
+        return real_async_generator(afn, allow_native=False)
+
+    wrapper.is_native = request.param == "native"
+    return wrapper
diff --git a/async_generator/_tests/test_async_generator.py b/async_generator/_tests/test_async_generator.py
index ff3e2a4..9258ecf 100644
--- a/async_generator/_tests/test_async_generator.py
+++ b/async_generator/_tests/test_async_generator.py
@@ -3,17 +3,17 @@
 import types
 import sys
 import collections.abc
-from functools import wraps
+from functools import wraps, partial
 import gc
 import inspect
 
 from .conftest import mock_sleep
 from .. import (
-    async_generator,
     yield_,
     yield_from_,
     isasyncgen,
     isasyncgenfunction,
+    supports_native_asyncgens,
     get_asyncgen_hooks,
     set_asyncgen_hooks,
 )
@@ -33,32 +33,39 @@ async def collect(ait):
 #
 ################################################################
 
-
-@async_generator
-async def async_range(count):
-    for i in range(count):
-        print("Calling yield_({})".format(i))
-        await yield_(i)
+# Async generators shared by multiple tests must be wrapped in
+# pytest fixtures in order to follow the async_generator
+# choice in effect for their test. That is, don't call async_range()
+# directly; instead, accept an async_range fixture and call that.
 
 
-@async_generator
-async def double(ait):
-    async for value in ait:
-        await yield_(value * 2)
-        await mock_sleep()
+@pytest.fixture
+def async_range(async_generator):
+    @async_generator
+    async def async_range(count):
+        for i in range(count):
+            print("Calling yield_({})".format(i))
+            await yield_(i)
 
+    return async_range
 
-class HasAsyncGenMethod:
-    def __init__(self, factor):
-        self._factor = factor
 
+async def test_async_generator(async_generator, async_range):
     @async_generator
-    async def async_multiplied(self, ait):
+    async def double(ait):
         async for value in ait:
-            await yield_(value * self._factor)
+            await yield_(value * 2)
+            await mock_sleep()
 
+    class HasAsyncGenMethod:
+        def __init__(self, factor):
+            self._factor = factor
+
+        @async_generator
+        async def async_multiplied(self, ait):
+            async for value in ait:
+                await yield_(value * self._factor)
 
-async def test_async_generator():
     assert await collect(async_range(10)) == list(range(10))
     assert (await collect(double(async_range(5))) == [0, 2, 4, 6, 8])
 
@@ -69,12 +76,11 @@ async def test_async_generator():
     )
 
 
-@async_generator
-async def agen_yield_no_arg():
-    await yield_()
-
+async def test_yield_no_arg(async_generator):
+    @async_generator
+    async def agen_yield_no_arg():
+        await yield_()
 
-async def test_yield_no_arg():
     assert await collect(agen_yield_no_arg()) == [None]
 
 
@@ -85,14 +91,13 @@ async def test_yield_no_arg():
 ################################################################
 
 
-@async_generator
-async def async_gen_with_non_None_return():
-    await yield_(1)
-    await yield_(2)
-    return "hi"
-
+async def test_bad_return_value(async_generator):
+    @async_generator(uses_return=True)
+    async def async_gen_with_non_None_return():
+        await yield_(1)
+        await yield_(2)
+        return "hi"
 
-async def test_bad_return_value():
     gen = async_gen_with_non_None_return()
     async for item in gen:  # pragma: no branch
         assert item == 1
@@ -106,6 +111,38 @@ async def test_bad_return_value():
         assert e.args[0] == "hi"
 
 
+@pytest.mark.skipif(
+    sys.implementation.name != "cpython",
+    reason="relies on bytecode inspection"
+)
+async def test_bad_return_detection():
+    from .. import async_generator
+
+    with pytest.raises(RuntimeError):
+
+        @async_generator(uses_return=False)
+        async def agen():  # pragma: no cover
+            return 42
+
+    with pytest.raises(RuntimeError):
+
+        @async_generator(uses_return=False)
+        async def agen():  # pragma: no cover
+            if await yield_(1):
+                return await yield_(2)
+            await yield_(3)
+
+    with pytest.raises(RuntimeError):
+
+        @async_generator(uses_return=True)
+        async def agen():  # pragma: no cover
+            if await yield_(1):
+                await yield_(2)
+                return None
+            await yield_(3)
+            return
+
+
 ################################################################
 #
 # Exhausitve tests of the different ways to re-enter a coroutine.
@@ -146,19 +183,6 @@ def next_me():
     assert (yield "next me") is None
 
 
-@async_generator
-async def yield_after_different_entries():
-    await yield_(1)
-    try:
-        await hit_me()
-    except MyTestError:
-        await yield_(2)
-    await number_me()
-    await yield_(3)
-    await next_me()
-    await yield_(4)
-
-
 def hostile_coroutine_runner(coro):
     coro_iter = coro.__await__()
     value = None
@@ -175,13 +199,25 @@ def hostile_coroutine_runner(coro):
             return exc.value
 
 
-def test_yield_different_entries():
+def test_yield_different_entries(async_generator):
+    @async_generator
+    async def yield_after_different_entries():
+        await yield_(1)
+        try:
+            await hit_me()
+        except MyTestError:
+            await yield_(2)
+        await number_me()
+        await yield_(3)
+        await next_me()
+        await yield_(4)
+
     coro = collect(yield_after_different_entries())
     yielded = hostile_coroutine_runner(coro)
     assert yielded == [1, 2, 3, 4]
 
 
-async def test_reentrance_forbidden():
+async def test_reentrance_forbidden(async_generator):
     @async_generator
     async def recurse():
         async for obj in agen:  # pragma: no branch
@@ -194,7 +230,9 @@ async def recurse():
 
 
 async def test_reentrance_forbidden_simultaneous_asends():
-    @async_generator
+    from .. import async_generator_legacy
+
+    @async_generator_legacy
     async def f():
         await mock_sleep()
 
@@ -210,12 +248,26 @@ async def f():
 
 
 # https://bugs.python.org/issue32526
-async def test_reentrance_forbidden_while_suspended_in_coroutine_runner():
+@pytest.mark.parametrize("explicitly_non_native", [False, True])
+async def test_reentrance_forbidden_while_suspended_in_coroutine_runner(
+        explicitly_non_native
+):
+    from .. import async_generator
+    if explicitly_non_native:
+        async_generator = partial(async_generator, allow_native=False)
+
     @async_generator
     async def f():
         await mock_sleep()
         await yield_("final yield")
 
+    def warns_if_not_explicit():
+        if explicitly_non_native:
+            from contextlib import contextmanager
+            return contextmanager(lambda: (yield))()
+        else:
+            return pytest.warns(FutureWarning)
+
     ag = f()
     asend_coro = ag.asend(None)
     fut = asend_coro.send(None)
@@ -223,8 +275,9 @@ async def f():
     # Now the async generator's frame is not executing, but a call to asend()
     # *is* executing. Make sure that in this case, ag_running is True, and we
     # can't start up another call to asend().
-    assert ag.ag_running
-    with pytest.raises(ValueError):
+    with warns_if_not_explicit():
+        assert ag.ag_running
+    with warns_if_not_explicit(), pytest.raises(ValueError):
         await ag.asend(None)
     # Clean up
     with pytest.raises(StopIteration):
@@ -240,13 +293,12 @@ async def f():
 ################################################################
 
 
-@async_generator
-async def asend_me():
-    assert (await yield_(1)) == 2
-    assert (await yield_(3)) == 4
-
+async def test_asend(async_generator):
+    @async_generator
+    async def asend_me():
+        assert (await yield_(1)) == 2
+        assert (await yield_(3)) == 4
 
-async def test_asend():
     aiter = asend_me()
     assert (await aiter.__anext__()) == 1
     assert (await aiter.asend(2)) == 3
@@ -261,16 +313,15 @@ async def test_asend():
 ################################################################
 
 
-@async_generator
-async def athrow_me():
-    with pytest.raises(KeyError):
-        await yield_(1)
-    with pytest.raises(ValueError):
-        await yield_(2)
-    await yield_(3)
-
+async def test_athrow(async_generator):
+    @async_generator
+    async def athrow_me():
+        with pytest.raises(KeyError):
+            await yield_(1)
+        with pytest.raises(ValueError):
+            await yield_(2)
+        await yield_(3)
 
-async def test_athrow():
     aiter = athrow_me()
     assert (await aiter.__anext__()) == 1
     assert (await aiter.athrow(KeyError("oops"))) == 2
@@ -286,18 +337,22 @@ async def test_athrow():
 ################################################################
 
 
-@async_generator
-async def close_me_aiter(track):
-    try:
-        await yield_(1)
-    except GeneratorExit:
-        track[0] = "closed"
-        raise
-    else:  # pragma: no cover
-        track[0] = "wtf"
+@pytest.fixture
+def close_me_aiter(async_generator):
+    @async_generator
+    async def close_me_aiter(track):
+        try:
+            await yield_(1)
+        except GeneratorExit:
+            track[0] = "closed"
+            raise
+        else:  # pragma: no cover
+            track[0] = "wtf"
 
+    return close_me_aiter
 
-async def test_aclose():
+
+async def test_aclose(close_me_aiter):
     track = [None]
     aiter = close_me_aiter(track)
     async for obj in aiter:  # pragma: no branch
@@ -308,37 +363,35 @@ async def test_aclose():
     assert track[0] == "closed"
 
 
-async def test_aclose_on_unstarted_generator():
+async def test_aclose_on_unstarted_generator(close_me_aiter):
     aiter = close_me_aiter([None])
     await aiter.aclose()
     async for obj in aiter:
         assert False  # pragma: no cover
 
 
-async def test_aclose_on_finished_generator():
+async def test_aclose_on_finished_generator(async_range):
     aiter = async_range(3)
     async for obj in aiter:
         pass  # pragma: no cover
     await aiter.aclose()
 
 
-@async_generator
-async def sync_yield_during_aclose():
-    try:
-        await yield_(1)
-    finally:
-        await mock_sleep()
-
-
-@async_generator
-async def async_yield_during_aclose():
-    try:
-        await yield_(1)
-    finally:
-        await yield_(2)
+async def test_aclose_yielding(async_generator):
+    @async_generator
+    async def sync_yield_during_aclose():
+        try:
+            await yield_(1)
+        finally:
+            await mock_sleep()
 
+    @async_generator
+    async def async_yield_during_aclose():
+        try:
+            await yield_(1)
+        finally:
+            await yield_(2)
 
-async def test_aclose_yielding():
     aiter = sync_yield_during_aclose()
     assert (await aiter.__anext__()) == 1
     # Doesn't raise:
@@ -357,11 +410,15 @@ async def test_aclose_yielding():
 ################################################################
 
 
-@async_generator
-async def async_range_twice(count):
-    await yield_from_(async_range(count))
-    await yield_(None)
-    await yield_from_(async_range(count))
+@pytest.fixture
+def async_range_twice(async_range, async_generator):
+    @async_generator
+    async def async_range_twice(count):
+        await yield_from_(async_range(count))
+        await yield_(None)
+        await yield_from_(async_range(count))
+
+    return async_range_twice
 
 
 if sys.version_info >= (3, 6):
@@ -381,7 +438,9 @@ async def native_async_range_twice(count, async_range):
     )
 
 
-async def test_async_yield_from_():
+async def test_async_yield_from_(
+        async_range_twice, async_range, async_generator
+):
     assert await collect(async_range_twice(3)) == [
         0,
         1,
@@ -412,18 +471,16 @@ async def yield_from_native():
         ]
 
 
-@async_generator
-async def doubles_sends(value):
-    while True:
-        value = await yield_(2 * value)
-
-
-@async_generator
-async def wraps_doubles_sends(value):
-    await yield_from_(doubles_sends(value))
+async def test_async_yield_from_asend(async_generator):
+    @async_generator
+    async def doubles_sends(value):
+        while True:
+            value = await yield_(2 * value)
 
+    @async_generator
+    async def wraps_doubles_sends(value):
+        await yield_from_(doubles_sends(value))
 
-async def test_async_yield_from_asend():
     gen = wraps_doubles_sends(10)
     await gen.__anext__() == 20
     assert (await gen.asend(2)) == 4
@@ -432,31 +489,29 @@ async def test_async_yield_from_asend():
     await gen.aclose()
 
 
-async def test_async_yield_from_athrow():
+async def test_async_yield_from_athrow(async_range_twice):
     gen = async_range_twice(2)
     assert (await gen.__anext__()) == 0
     with pytest.raises(ValueError):
         await gen.athrow(ValueError)
 
 
-@async_generator
-async def returns_1():
-    await yield_(0)
-    return 1
-
-
-@async_generator
-async def yields_from_returns_1():
-    await yield_(await yield_from_(returns_1()))
+async def test_async_yield_from_return_value(async_generator):
+    @async_generator(uses_return=True)
+    async def returns_1():
+        await yield_(0)
+        return 1
 
+    @async_generator
+    async def yields_from_returns_1():
+        await yield_(await yield_from_(returns_1()))
 
-async def test_async_yield_from_return_value():
     assert await collect(yields_from_returns_1()) == [0, 1]
 
 
 # Special cases to get coverage
-async def test_yield_from_empty():
-    @async_generator
+async def test_yield_from_empty(async_generator):
+    @async_generator(uses_return=True)
     async def empty():
         return "done"
 
@@ -467,7 +522,7 @@ async def yield_from_empty():
     assert await collect(yield_from_empty()) == []
 
 
-async def test_yield_from_non_generator():
+async def test_yield_from_non_generator(async_generator):
     class Countdown:
         def __init__(self, count):
             self.count = count
@@ -491,7 +546,7 @@ async def __anext__(self):
         async def aclose(self):
             self.closed = True
 
-    @async_generator
+    @async_generator(uses_return=True)
     async def yield_from_countdown(count, happenings):
         try:
             c = Countdown(count)
@@ -536,7 +591,7 @@ async def yield_from_countdown(count, happenings):
     assert h == ["countdown closed", "raise"]
 
 
-async def test_yield_from_non_generator_with_no_aclose():
+async def test_yield_from_non_generator_with_no_aclose(async_generator):
     class Countdown:
         def __init__(self, count):
             self.count = count
@@ -557,7 +612,7 @@ async def __anext__(self):
                 raise StopAsyncIteration("boom")
             return self.count
 
-    @async_generator
+    @async_generator(uses_return=True)
     async def yield_from_countdown(count):
         return await yield_from_(Countdown(count))
 
@@ -570,7 +625,7 @@ async def yield_from_countdown(count):
     await agen.aclose()
 
 
-async def test_yield_from_with_old_style_aiter():
+async def test_yield_from_with_old_style_aiter(async_generator):
     # old-style 'async def __aiter__' should still work even on newer pythons
     class Countdown:
         def __init__(self, count):
@@ -587,15 +642,15 @@ async def __anext__(self):
                 raise StopAsyncIteration("boom")
             return self.count
 
-    @async_generator
+    @async_generator(uses_return=True)
     async def yield_from_countdown(count):
         return await yield_from_(Countdown(count))
 
     assert await collect(yield_from_countdown(3)) == [2, 1, 0]
 
 
-async def test_yield_from_athrow_raises_StopAsyncIteration():
-    @async_generator
+async def test_yield_from_athrow_raises_StopAsyncIteration(async_generator):
+    @async_generator(uses_return=True)
     async def catch():
         try:
             while True:
@@ -603,7 +658,7 @@ async def catch():
         except Exception as exc:
             return ("bye", exc)
 
-    @async_generator
+    @async_generator(uses_return=True)
     async def yield_from_catch():
         return await yield_from_(catch())
 
@@ -624,7 +679,7 @@ async def yield_from_catch():
 ################################################################
 
 
-async def test___del__(capfd):
+async def test___del__(async_generator, capfd):
     completions = 0
 
     @async_generator
@@ -680,11 +735,18 @@ async def awaits_when_unwinding():
         elif stop_after_turn == 2:
             # Stopped in the middle of a try/finally that awaits in the finally,
             # so __del__ can't cleanup.
-            with pytest.raises(RuntimeError) as info:
+            if async_generator.is_native:
+                # Native asyncgens suppress the exception __del__-style even if
+                # __del__ is not being called during GC.
                 gen.__del__()
-            assert "awaited during finalization; install a finalization hook" in str(
-                info.value
-            )
+                assert "RuntimeError: async generator ignored GeneratorExit" in capfd.readouterr(
+                ).err
+            else:
+                with pytest.raises(RuntimeError) as info:
+                    gen.__del__()
+                assert "awaited during finalization; install a finalization hook" in str(
+                    info.value
+                )
         else:
             # Can clean up without awaiting, so __del__ is fine
             gen.__del__()
@@ -700,8 +762,15 @@ async def yields_when_unwinding():
 
     gen = yields_when_unwinding()
     assert await gen.__anext__() == 1
-    with pytest.raises(RuntimeError) as info:
+    if async_generator.is_native:
+        # Native asyncgens suppress the exception __del__-style even if
+        # __del__ is not being called during GC.
         gen.__del__()
+        assert "RuntimeError: async generator ignored GeneratorExit" in capfd.readouterr(
+        ).err
+    else:
+        with pytest.raises(RuntimeError):
+            gen.__del__()
 
 
 ################################################################
@@ -709,7 +778,7 @@ async def yields_when_unwinding():
 ################################################################
 
 
-def test_isasyncgen():
+def test_isasyncgen(async_range):
     assert not isasyncgen(async_range)
     assert isasyncgen(async_range(10))
 
@@ -718,7 +787,7 @@ def test_isasyncgen():
         assert isasyncgen(native_async_range(10))
 
 
-def test_isasyncgenfunction():
+def test_isasyncgenfunction(async_range):
     assert isasyncgenfunction(async_range)
     assert not isasyncgenfunction(list)
     assert not isasyncgenfunction(async_range(10))
@@ -745,7 +814,7 @@ def test_isasyncgenfunction():
 # correct, inspect.isasyncgenfunction-compliant behavior, we have async_cm
 # introspecting as an async context manager, and async_cm.__wrapped__
 # introspecting as an async generator.
-def test_isasyncgenfunction_is_not_inherited_by_wrappers():
+def test_isasyncgenfunction_is_not_inherited_by_wrappers(async_range):
     @wraps(async_range)
     def async_range_wrapper(*args, **kwargs):  # pragma: no cover
         return async_range(*args, **kwargs)
@@ -754,12 +823,12 @@ def async_range_wrapper(*args, **kwargs):  # pragma: no cover
     assert isasyncgenfunction(async_range_wrapper.__wrapped__)
 
 
-def test_collections_abc_AsyncGenerator():
+def test_collections_abc_AsyncGenerator(async_range):
     if hasattr(collections.abc, "AsyncGenerator"):
         assert isinstance(async_range(10), collections.abc.AsyncGenerator)
 
 
-async def test_ag_attributes():
+async def test_ag_attributes(async_generator):
     @async_generator
     async def f():
         x = 1
@@ -769,6 +838,7 @@ async def f():
     assert agen.ag_code.co_name == "f"
     async for _ in agen:  # pragma: no branch
         assert agen.ag_frame.f_locals["x"] == 1
+        assert agen.ag_await.cr_code is yield_.__code__
         break
 
 
@@ -794,8 +864,7 @@ def test_refcnt():
 
 
 @pytest.mark.skipif(
-    not hasattr(None, "__sizeof__") or not hasattr(inspect, "isasyncgen"),
-    reason="CPython with native asyncgens only"
+    not supports_native_asyncgens, reason="relevant to native asyncgens only"
 )
 def test_gen_agen_size():
     def gen():  # pragma: no cover
@@ -821,7 +890,8 @@ def gen():  # pragma: no cover
 
 
 def test_unwrap_not_wrapped():
-    with pytest.raises((TypeError, AttributeError)):
+    with pytest.raises(TypeError
+                       if supports_native_asyncgens else AttributeError):
         _impl._unwrap(42)
 
 
@@ -836,12 +906,11 @@ def test_unwrap_not_wrapped():
 # generator should produce a RuntimeError with the __cause__ set to the
 # original exception. Note that contextlib.asynccontextmanager depends on this
 # behavior.
-@async_generator
-async def lets_exception_out():
-    await yield_()
-
+async def test_throw_StopIteration_or_StopAsyncIteration(async_generator):
+    @async_generator
+    async def lets_exception_out():
+        await yield_()
 
-async def test_throw_StopIteration_or_StopAsyncIteration():
     for cls in [StopIteration, StopAsyncIteration]:
         agen = lets_exception_out()
         await agen.asend(None)
@@ -854,7 +923,7 @@ async def test_throw_StopIteration_or_StopAsyncIteration():
 
 # No "coroutine was never awaited" warnings for async generators that are not
 # iterated
-async def test_no_spurious_unawaited_coroutine_warning(recwarn):
+async def test_no_spurious_unawaited_coroutine_warning(recwarn, async_range):
     agen = async_range(10)
     del agen
 
@@ -871,6 +940,51 @@ async def test_no_spurious_unawaited_coroutine_warning(recwarn):
         assert not issubclass(msg.category, RuntimeWarning)
 
 
+# Reasonable error on @async_generator of something other than an async function
+async def test_asyncgen_type_error(async_generator):
+    with pytest.raises(TypeError) as info:
+
+        @async_generator
+        def whoops1():  # pragma: no cover
+            pass
+
+    assert "expected an async function, not 'function'" in str(info.value)
+
+    with pytest.raises(TypeError) as info:
+
+        def whoops2():  # pragma: no cover
+            yield
+
+        async_generator(whoops2())
+    assert "expected an async function, not 'generator'" in str(info.value)
+
+    # Make sure a sync function wrapping an async one is OK, and produces
+    # a non-native generator.
+    async def ok(val):
+        await yield_(2 * val)
+
+    @wraps(ok)
+    def wrap1(val):
+        return ok(3 * val)
+
+    @wraps(wrap1)
+    def wrap2(val):
+        return wrap1(5 * val)
+
+    gen0 = async_generator(ok)
+    gen1 = async_generator(wrap1)
+    gen2 = async_generator(wrap2)
+
+    assert await collect(gen0(10)) == [20]
+    assert await collect(gen1(10)) == [60]
+    assert await collect(gen2(10)) == [300]
+
+    if async_generator.is_native:
+        assert inspect.isasyncgenfunction(gen0)
+        assert not inspect.isasyncgenfunction(gen1)
+        assert not inspect.isasyncgenfunction(gen2)
+
+
 ################################################################
 #
 # GC hooks
@@ -926,7 +1040,7 @@ def in_thread(results):
     assert get_asyncgen_hooks() == (one, two)
 
 
-async def test_gc_hooks_behavior(local_asyncgen_hooks):
+async def test_gc_hooks_behavior(async_generator, local_asyncgen_hooks):
     events = []
     to_finalize = []
 
diff --git a/async_generator/_tests/test_util.py b/async_generator/_tests/test_util.py
index 16961d1..812a45b 100644
--- a/async_generator/_tests/test_util.py
+++ b/async_generator/_tests/test_util.py
@@ -1,18 +1,17 @@
 import pytest
 
-from .. import aclosing, async_generator, yield_, asynccontextmanager
+from .. import aclosing, yield_, asynccontextmanager, supports_native_asyncgens
 
 
-@async_generator
-async def async_range(count, closed_slot):
-    try:
-        for i in range(count):  # pragma: no branch
-            await yield_(i)
-    except GeneratorExit:
-        closed_slot[0] = True
-
+async def test_aclosing(async_generator):
+    @async_generator
+    async def async_range(count, closed_slot):
+        try:
+            for i in range(count):  # pragma: no branch
+                await yield_(i)
+        except GeneratorExit:
+            closed_slot[0] = True
 
-async def test_aclosing():
     closed_slot = [False]
     async with aclosing(async_range(10, closed_slot)) as gen:
         it = iter(range(10))
@@ -35,7 +34,9 @@ async def test_aclosing():
     assert closed_slot[0]
 
 
-async def test_contextmanager_do_not_unchain_non_stopiteration_exceptions():
+async def test_contextmanager_do_not_unchain_non_stopiteration_exceptions(
+        async_generator
+):
     @asynccontextmanager
     @async_generator
     async def manager_issue29692():
@@ -78,8 +79,7 @@ async def noop_async_context_manager():
 
 
 # Native async generators are only available from Python 3.6 and onwards
-nativeasyncgenerators = True
-try:
+if supports_native_asyncgens:
     exec(
         """
 @asynccontextmanager
@@ -90,12 +90,10 @@ async def manager_issue29692_2():
         raise RuntimeError('issue29692:Chained') from exc
 """
     )
-except SyntaxError:
-    nativeasyncgenerators = False
 
 
 @pytest.mark.skipif(
-    not nativeasyncgenerators,
+    not supports_native_asyncgens,
     reason="Python < 3.6 doesn't have native async generators"
 )
 async def test_native_contextmanager_do_not_unchain_non_stopiteration_exceptions(
@@ -115,7 +113,7 @@ async def test_native_contextmanager_do_not_unchain_non_stopiteration_exceptions
         assert excinfo.value.__cause__ is None
 
 
-async def test_asynccontextmanager_exception_passthrough():
+async def test_asynccontextmanager_exception_passthrough(async_generator):
     # This was the cause of annoying coverage flapping, see gh-140
     @asynccontextmanager
     @async_generator
@@ -132,7 +130,7 @@ async def noop_async_context_manager():
         pass
 
 
-async def test_asynccontextmanager_catches_exception():
+async def test_asynccontextmanager_catches_exception(async_generator):
     @asynccontextmanager
     @async_generator
     async def catch_it():
@@ -143,7 +141,7 @@ async def catch_it():
         raise ValueError
 
 
-async def test_asynccontextmanager_different_exception():
+async def test_asynccontextmanager_different_exception(async_generator):
     @asynccontextmanager
     @async_generator
     async def switch_it():
@@ -157,7 +155,7 @@ async def switch_it():
             raise KeyError
 
 
-async def test_asynccontextmanager_nice_message_on_sync_enter():
+async def test_asynccontextmanager_nice_message_on_sync_enter(async_generator):
     @asynccontextmanager
     @async_generator
     async def xxx():  # pragma: no cover
@@ -175,7 +173,7 @@ async def xxx():  # pragma: no cover
         pass
 
 
-async def test_asynccontextmanager_no_yield():
+async def test_asynccontextmanager_no_yield(async_generator):
     @asynccontextmanager
     @async_generator
     async def yeehaw():
@@ -188,7 +186,7 @@ async def yeehaw():
     assert "didn't yield" in str(excinfo.value)
 
 
-async def test_asynccontextmanager_too_many_yields():
+async def test_asynccontextmanager_too_many_yields(async_generator):
     closed_count = 0
 
     @asynccontextmanager
diff --git a/docs/source/reference.rst b/docs/source/reference.rst
index cf3f406..df4123e 100644
--- a/docs/source/reference.rst
+++ b/docs/source/reference.rst
@@ -117,6 +117,98 @@ support for ``yield from`` and for returning non-None values from
 an async generator (under the theory that these may well be added
 to native async generators one day).
 
+There are actually two implementations of the ``@async_generator``
+decorator internally:
+
+* ``@async_generator_legacy`` (or ``@async_generator(allow_native=False)``)
+  always uses the pure-Python version that requires no interpreter
+  support beyond basic async/await. It is rather slow for yield-bound
+  workloads (see :ref:`perf`), but supports PyPy, CPython 3.5, and
+  async generators that return non-None values.
+
+* ``@async_generator_native`` (or ``@async_generator(allow_native=True)``)
+  uses some ``co_flags`` trickery to create a *native* async generator
+  where possible, with 3.5-compatible syntax but closer to 3.6-native
+  performance (and some minor semantic differences, described below).
+  Native async generators can only be created on CPython 3.6+ and only
+  for functions that never return values other than None; if these
+  conditions are not met, ``@async_generator_native`` silently behaves
+  like ``@async_generator_legacy``.
+
+Currently, ``@async_generator`` behaves like ``@async_generator_legacy``,
+and also emits warnings if the resulting generator is used in a way
+that would produce different results under ``@async_generator_native``.
+The next release of this library will change the default to
+``@async_generator_native``.
+
+There are a couple of known differences between the semantics of the
+two implementations:
+
+* Native async generators suffer from a `bug <https://bugs.python.org/issue32526>`__
+  where they will not be considered "running" if the generator has
+  awaited some async function that is currently suspended in the
+  event loop. Thus, you can start a second ``asend()`` call (or similar)
+  before the first one completes, resulting in behavior that is extremely
+  confusing at best. The legacy implementation does not have this quirk,
+  and will throw an exception on any attempt to reenter the generator.
+
+* Native async generators provide a rather unhelpful error message if
+  they are garbage-collected with no :ref:`finalizer <finalization>` installed
+  and some async operations to perform during stack unwinding: they say
+  "async generator ignored GeneratorExit", the same as if the generator
+  tried to ``yield`` in a ``finally:`` block. The legacy implementation
+  provides a clearer error that describes the possible remedies.
+
+Both types of ``@async_generator`` behave the same with respect to
+:ref:`finalization semantics <finalization>`, :ref:`yield_from interoperability
+with native generators <yieldfrom>`, and so on; you don't need to use
+``@async_generator_native`` for those if you're on 3.6.
+
+The value ``async_generator.supports_native_asyncgens`` will be True
+if we're running on a platform where the above-described tricks are
+possible.
+
+.. note::
+   Producing a native async generator requires this library to determine
+   whether an async function being decorated with ``@async_generator`` might
+   return something other than None. (Returns with an argument are
+   syntactically forbidden in native async generators, and if through
+   trickery we were to manage to execute one, the interpreter would probably
+   crash.) Normally the presence of such returns is autodetected through
+   some fairly straightforward bytecode inspection, but it's possible to
+   verify that the check is behaving as you expect: if you decorate a
+   function with ``@async_generator(uses_return=True)`` and it doesn't
+   appear to have non-None returns, or with
+   ``@async_generator(uses_return=False)`` and it does, the decorator
+   will throw a RuntimeError.
+
+
+.. _perf:
+
+Performance
+~~~~~~~~~~~
+
+`PEP 525 <https://www.python.org/dev/peps/pep-0525/#improvements-over-asynchronous-iterators>`__
+includes a microbenchmark demonstrating that a yield-bound async
+generator (that yields out ten million integers without awaiting
+anything) runs 2.3x faster than the equivalent async iterator.
+The pure-Python version of ``@async_generator`` cannot claim such
+improvements; it's about 10x *slower* than the async iterator on
+that workload. On the other hand, the version that produces a native
+async generator with ``await yield_(...)`` calls is only about 1.7x
+slower than the async iterator (4.3x slower than a backwards-incompatible
+fully native async generator); the overhead appears to be entirely
+due to the two extra async function calls that must be executed per
+``yield_`` compared to a native ``yield``.
+
+When considering these numbers, remember that most async generators
+are not yield-bound; if they didn't have to wait for something to
+happen in the event loop, they probably wouldn't be async! (Pure
+Python ``@async_generator`` functions do add a bit of overhead per
+event loop trap as well, but native ones do not.)
+
+
+.. _finalization:
 
 Garbage collection hooks
 ~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/newsfragments/17.feature.rst b/newsfragments/17.feature.rst
new file mode 100644
index 0000000..ed6f9c3
--- /dev/null
+++ b/newsfragments/17.feature.rst
@@ -0,0 +1 @@
+Add the ability for ``@async_generator`` to produce a native async generator on CPython 3.6+. Since there are minor semantic differences between native async generators and the existing pure-Python ``@async_generator`` generators, production of native generators is disabled by default in this release and you must use ``@async_generator_native`` to request it. Native mode will become the default (on supported platforms) in the next release of this library. ``@async_generator_legacy`` will always means the old pure-Python style of async generator.