We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
aioredis raises CancelledError in _finish_span 
### Which version of dd-trace-py are you using?

~~0.53.0~~ 0.58.0

### Which version of pip are you using?

21.3.1

### Which version of the libraries are you using?

django==3.2.11
django-redis==5.0.0
channels==3.0.4
daphne==3.0.2

### How can we reproduce your problem?

I am using code similar to the following:

asgi.py

```
import django
from channels.routing import get_default_application
from ddtrace.contrib.asgi import TraceMiddleware

django.setup()
application = TraceMiddleware(get_default_application())
```

routing.py

```
from django.urls import re_path
import my_app.consumers

websocket_urlpatterns = [
    re_path(r"^ws/test/$", consumers.TestConsumer.as_asgi()),
]
```

my_app/consumers.py

```
from channels.generic.websocket import WebsocketConsumer

class TestConsumer(WebsocketConsumer):
    groups = ["broadcast"]

    def connect(self):
        self.accept()

    def receive(self, text_data=None, bytes_data=None):
        raise Exception("An test exception")
```

I am running the application with: `ddtrace-run daphne asgi:application --bind 0.0.0.0 --port 8001`

### What is the result that you get?

I don't get any traces at all, and my logs show this:

```
handle: <Handle traced_13_execute_command.<locals>._finish_span(<Future cancelled>) at /usr/local/lib/python3.10/site-packages/ddtrace/contrib/aioredis/patch.py:140>

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/lib/python3.10/site-packages/ddtrace/contrib/aioredis/patch.py", line 146, in _finish_span
    future.result()
asyncio.exceptions.CancelledError
```


### What is the result that you expected?

No errors


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ddtrace/contrib/aioredis/patch.py`
Content:
```
1 import asyncio
2 import sys
3 
4 import aioredis
5 
6 from ddtrace import config
7 from ddtrace.internal.utils.wrappers import unwrap as _u
8 from ddtrace.pin import Pin
9 from ddtrace.vendor.wrapt import wrap_function_wrapper as _w
10 
11 from .. import trace_utils
12 from ...constants import ANALYTICS_SAMPLE_RATE_KEY
13 from ...constants import SPAN_MEASURED_KEY
14 from ...ext import SpanTypes
15 from ...ext import net
16 from ...ext import redis as redisx
17 from ...internal.utils.formats import stringify_cache_args
18 from ..redis.util import _trace_redis_cmd
19 from ..redis.util import _trace_redis_execute_pipeline
20 
21 
22 try:
23     from aioredis.commands.transaction import _RedisBuffer
24 except ImportError:
25     _RedisBuffer = None
26 
27 config._add("aioredis", dict(_default_service="redis"))
28 
29 aioredis_version_str = getattr(aioredis, "__version__", "0.0.0")
30 aioredis_version = tuple([int(i) for i in aioredis_version_str.split(".")])
31 
32 
33 def patch():
34     if getattr(aioredis, "_datadog_patch", False):
35         return
36     setattr(aioredis, "_datadog_patch", True)
37     pin = Pin()
38     if aioredis_version >= (2, 0):
39         _w("aioredis.client", "Redis.execute_command", traced_execute_command)
40         _w("aioredis.client", "Redis.pipeline", traced_pipeline)
41         _w("aioredis.client", "Pipeline.execute", traced_execute_pipeline)
42         pin.onto(aioredis.client.Redis)
43     else:
44         _w("aioredis", "Redis.execute", traced_13_execute_command)
45         _w("aioredis", "Redis.pipeline", traced_13_pipeline)
46         _w("aioredis.commands.transaction", "Pipeline.execute", traced_13_execute_pipeline)
47         pin.onto(aioredis.Redis)
48 
49 
50 def unpatch():
51     if not getattr(aioredis, "_datadog_patch", False):
52         return
53 
54     setattr(aioredis, "_datadog_patch", False)
55     if aioredis_version >= (2, 0):
56         _u(aioredis.client.Redis, "execute_command")
57         _u(aioredis.client.Redis, "pipeline")
58         _u(aioredis.client.Pipeline, "execute")
59     else:
60         _u(aioredis.Redis, "execute")
61         _u(aioredis.Redis, "pipeline")
62         _u(aioredis.commands.transaction.Pipeline, "execute")
63 
64 
65 async def traced_execute_command(func, instance, args, kwargs):
66     pin = Pin.get_from(instance)
67     if not pin or not pin.enabled():
68         return await func(*args, **kwargs)
69 
70     with _trace_redis_cmd(pin, config.aioredis, instance, args):
71         return await func(*args, **kwargs)
72 
73 
74 def traced_pipeline(func, instance, args, kwargs):
75     pipeline = func(*args, **kwargs)
76     pin = Pin.get_from(instance)
77     if pin:
78         pin.onto(pipeline)
79     return pipeline
80 
81 
82 async def traced_execute_pipeline(func, instance, args, kwargs):
83     pin = Pin.get_from(instance)
84     if not pin or not pin.enabled():
85         return await func(*args, **kwargs)
86 
87     cmds = [stringify_cache_args(c) for c, _ in instance.command_stack]
88     resource = "\n".join(cmds)
89     with _trace_redis_execute_pipeline(pin, config.aioredis, resource, instance):
90         return await func(*args, **kwargs)
91 
92 
93 def traced_13_pipeline(func, instance, args, kwargs):
94     pipeline = func(*args, **kwargs)
95     pin = Pin.get_from(instance)
96     if pin:
97         pin.onto(pipeline)
98     return pipeline
99 
100 
101 def traced_13_execute_command(func, instance, args, kwargs):
102     # If we have a _RedisBuffer then we are in a pipeline
103     if isinstance(instance.connection, _RedisBuffer):
104         return func(*args, **kwargs)
105 
106     pin = Pin.get_from(instance)
107     if not pin or not pin.enabled():
108         return func(*args, **kwargs)
109 
110     # Don't activate the span since this operation is performed as a future which concludes sometime later on in
111     # execution so subsequent operations in the stack are not necessarily semantically related
112     # (we don't want this span to be the parent of all other spans created before the future is resolved)
113     parent = pin.tracer.current_span()
114     span = pin.tracer.start_span(
115         redisx.CMD,
116         service=trace_utils.ext_service(pin, config.aioredis),
117         span_type=SpanTypes.REDIS,
118         activate=False,
119         child_of=parent,
120     )
121 
122     span.set_tag(SPAN_MEASURED_KEY)
123     query = stringify_cache_args(args)
124     span.resource = query
125     span.set_tag(redisx.RAWCMD, query)
126     if pin.tags:
127         span.set_tags(pin.tags)
128 
129     span.set_tags(
130         {
131             net.TARGET_HOST: instance.address[0],
132             net.TARGET_PORT: instance.address[1],
133             redisx.DB: instance.db or 0,
134         }
135     )
136     span.set_metric(redisx.ARGS_LEN, len(args))
137     # set analytics sample rate if enabled
138     span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())
139 
140     def _finish_span(future):
141         try:
142             # Accessing the result will raise an exception if:
143             #   - The future was cancelled
144             #   - There was an error executing the future (`future.exception()`)
145             #   - The future is in an invalid state
146             future.result()
147         except Exception:
148             span.set_exc_info(*sys.exc_info())
149         finally:
150             span.finish()
151 
152     task = func(*args, **kwargs)
153     # Execute command returns a coroutine when no free connections are available
154     # https://github.com/aio-libs/aioredis-py/blob/v1.3.1/aioredis/pool.py#L191
155     task = asyncio.ensure_future(task)
156     task.add_done_callback(_finish_span)
157     return task
158 
159 
160 async def traced_13_execute_pipeline(func, instance, args, kwargs):
161     pin = Pin.get_from(instance)
162     if not pin or not pin.enabled():
163         return await func(*args, **kwargs)
164 
165     cmds = []
166     for _, cmd, cmd_args, _ in instance._pipeline:
167         parts = [cmd]
168         parts.extend(cmd_args)
169         cmds.append(stringify_cache_args(parts))
170     resource = "\n".join(cmds)
171     with pin.tracer.trace(
172         redisx.CMD,
173         resource=resource,
174         service=trace_utils.ext_service(pin, config.aioredis),
175         span_type=SpanTypes.REDIS,
176     ) as span:
177 
178         span.set_tags(
179             {
180                 net.TARGET_HOST: instance._pool_or_conn.address[0],
181                 net.TARGET_PORT: instance._pool_or_conn.address[1],
182                 redisx.DB: instance._pool_or_conn.db or 0,
183             }
184         )
185 
186         span.set_tag(SPAN_MEASURED_KEY)
187         span.set_tag(redisx.RAWCMD, resource)
188         span.set_metric(redisx.PIPELINE_LEN, len(instance._pipeline))
189         # set analytics sample rate if enabled
190         span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())
191 
192         return await func(*args, **kwargs)
193
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ddtrace/contrib/aioredis/patch.py b/ddtrace/contrib/aioredis/patch.py
--- a/ddtrace/contrib/aioredis/patch.py
+++ b/ddtrace/contrib/aioredis/patch.py
@@ -140,11 +140,12 @@
     def _finish_span(future):
         try:
             # Accessing the result will raise an exception if:
-            #   - The future was cancelled
+            #   - The future was cancelled (CancelledError)
             #   - There was an error executing the future (`future.exception()`)
             #   - The future is in an invalid state
             future.result()
-        except Exception:
+        # CancelledError exceptions extend from BaseException as of Python 3.8, instead of usual Exception
+        except BaseException:
             span.set_exc_info(*sys.exc_info())
         finally:
             span.finish()

{"golden_diff": "diff --git a/ddtrace/contrib/aioredis/patch.py b/ddtrace/contrib/aioredis/patch.py\n--- a/ddtrace/contrib/aioredis/patch.py\n+++ b/ddtrace/contrib/aioredis/patch.py\n@@ -140,11 +140,12 @@\n     def _finish_span(future):\n         try:\n             # Accessing the result will raise an exception if:\n-            #   - The future was cancelled\n+            #   - The future was cancelled (CancelledError)\n             #   - There was an error executing the future (`future.exception()`)\n             #   - The future is in an invalid state\n             future.result()\n-        except Exception:\n+        # CancelledError exceptions extend from BaseException as of Python 3.8, instead of usual Exception\n+        except BaseException:\n             span.set_exc_info(*sys.exc_info())\n         finally:\n             span.finish()\n", "issue": "aioredis raises CancelledError in _finish_span \n### Which version of dd-trace-py are you using?\r\n\r\n~~0.53.0~~ 0.58.0\r\n\r\n### Which version of pip are you using?\r\n\r\n21.3.1\r\n\r\n### Which version of the libraries are you using?\r\n\r\ndjango==3.2.11\r\ndjango-redis==5.0.0\r\nchannels==3.0.4\r\ndaphne==3.0.2\r\n\r\n### How can we reproduce your problem?\r\n\r\nI am using code similar to the following:\r\n\r\nasgi.py\r\n\r\n```\r\nimport django\r\nfrom channels.routing import get_default_application\r\nfrom ddtrace.contrib.asgi import TraceMiddleware\r\n\r\ndjango.setup()\r\napplication = TraceMiddleware(get_default_application())\r\n```\r\n\r\nrouting.py\r\n\r\n```\r\nfrom django.urls import re_path\r\nimport my_app.consumers\r\n\r\nwebsocket_urlpatterns = [\r\n    re_path(r\"^ws/test/$\", consumers.TestConsumer.as_asgi()),\r\n]\r\n```\r\n\r\nmy_app/consumers.py\r\n\r\n```\r\nfrom channels.generic.websocket import WebsocketConsumer\r\n\r\nclass TestConsumer(WebsocketConsumer):\r\n    groups = [\"broadcast\"]\r\n\r\n    def connect(self):\r\n        self.accept()\r\n\r\n    def receive(self, text_data=None, bytes_data=None):\r\n        raise Exception(\"An test exception\")\r\n```\r\n\r\nI am running the application with: `ddtrace-run daphne asgi:application --bind 0.0.0.0 --port 8001`\r\n\r\n### What is the result that you get?\r\n\r\nI don't get any traces at all, and my logs show this:\r\n\r\n```\r\nhandle: <Handle traced_13_execute_command.<locals>._finish_span(<Future cancelled>) at /usr/local/lib/python3.10/site-packages/ddtrace/contrib/aioredis/patch.py:140>\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.10/asyncio/events.py\", line 80, in _run\r\n    self._context.run(self._callback, *self._args)\r\n  File \"/usr/local/lib/python3.10/site-packages/ddtrace/contrib/aioredis/patch.py\", line 146, in _finish_span\r\n    future.result()\r\nasyncio.exceptions.CancelledError\r\n```\r\n\r\n\r\n### What is the result that you expected?\r\n\r\nNo errors\r\n\n", "before_files": [{"content": "import asyncio\nimport sys\n\nimport aioredis\n\nfrom ddtrace import config\nfrom ddtrace.internal.utils.wrappers import unwrap as _u\nfrom ddtrace.pin import Pin\nfrom ddtrace.vendor.wrapt import wrap_function_wrapper as _w\n\nfrom .. import trace_utils\nfrom ...constants import ANALYTICS_SAMPLE_RATE_KEY\nfrom ...constants import SPAN_MEASURED_KEY\nfrom ...ext import SpanTypes\nfrom ...ext import net\nfrom ...ext import redis as redisx\nfrom ...internal.utils.formats import stringify_cache_args\nfrom ..redis.util import _trace_redis_cmd\nfrom ..redis.util import _trace_redis_execute_pipeline\n\n\ntry:\n    from aioredis.commands.transaction import _RedisBuffer\nexcept ImportError:\n    _RedisBuffer = None\n\nconfig._add(\"aioredis\", dict(_default_service=\"redis\"))\n\naioredis_version_str = getattr(aioredis, \"__version__\", \"0.0.0\")\naioredis_version = tuple([int(i) for i in aioredis_version_str.split(\".\")])\n\n\ndef patch():\n    if getattr(aioredis, \"_datadog_patch\", False):\n        return\n    setattr(aioredis, \"_datadog_patch\", True)\n    pin = Pin()\n    if aioredis_version >= (2, 0):\n        _w(\"aioredis.client\", \"Redis.execute_command\", traced_execute_command)\n        _w(\"aioredis.client\", \"Redis.pipeline\", traced_pipeline)\n        _w(\"aioredis.client\", \"Pipeline.execute\", traced_execute_pipeline)\n        pin.onto(aioredis.client.Redis)\n    else:\n        _w(\"aioredis\", \"Redis.execute\", traced_13_execute_command)\n        _w(\"aioredis\", \"Redis.pipeline\", traced_13_pipeline)\n        _w(\"aioredis.commands.transaction\", \"Pipeline.execute\", traced_13_execute_pipeline)\n        pin.onto(aioredis.Redis)\n\n\ndef unpatch():\n    if not getattr(aioredis, \"_datadog_patch\", False):\n        return\n\n    setattr(aioredis, \"_datadog_patch\", False)\n    if aioredis_version >= (2, 0):\n        _u(aioredis.client.Redis, \"execute_command\")\n        _u(aioredis.client.Redis, \"pipeline\")\n        _u(aioredis.client.Pipeline, \"execute\")\n    else:\n        _u(aioredis.Redis, \"execute\")\n        _u(aioredis.Redis, \"pipeline\")\n        _u(aioredis.commands.transaction.Pipeline, \"execute\")\n\n\nasync def traced_execute_command(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    with _trace_redis_cmd(pin, config.aioredis, instance, args):\n        return await func(*args, **kwargs)\n\n\ndef traced_pipeline(func, instance, args, kwargs):\n    pipeline = func(*args, **kwargs)\n    pin = Pin.get_from(instance)\n    if pin:\n        pin.onto(pipeline)\n    return pipeline\n\n\nasync def traced_execute_pipeline(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    cmds = [stringify_cache_args(c) for c, _ in instance.command_stack]\n    resource = \"\\n\".join(cmds)\n    with _trace_redis_execute_pipeline(pin, config.aioredis, resource, instance):\n        return await func(*args, **kwargs)\n\n\ndef traced_13_pipeline(func, instance, args, kwargs):\n    pipeline = func(*args, **kwargs)\n    pin = Pin.get_from(instance)\n    if pin:\n        pin.onto(pipeline)\n    return pipeline\n\n\ndef traced_13_execute_command(func, instance, args, kwargs):\n    # If we have a _RedisBuffer then we are in a pipeline\n    if isinstance(instance.connection, _RedisBuffer):\n        return func(*args, **kwargs)\n\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return func(*args, **kwargs)\n\n    # Don't activate the span since this operation is performed as a future which concludes sometime later on in\n    # execution so subsequent operations in the stack are not necessarily semantically related\n    # (we don't want this span to be the parent of all other spans created before the future is resolved)\n    parent = pin.tracer.current_span()\n    span = pin.tracer.start_span(\n        redisx.CMD,\n        service=trace_utils.ext_service(pin, config.aioredis),\n        span_type=SpanTypes.REDIS,\n        activate=False,\n        child_of=parent,\n    )\n\n    span.set_tag(SPAN_MEASURED_KEY)\n    query = stringify_cache_args(args)\n    span.resource = query\n    span.set_tag(redisx.RAWCMD, query)\n    if pin.tags:\n        span.set_tags(pin.tags)\n\n    span.set_tags(\n        {\n            net.TARGET_HOST: instance.address[0],\n            net.TARGET_PORT: instance.address[1],\n            redisx.DB: instance.db or 0,\n        }\n    )\n    span.set_metric(redisx.ARGS_LEN, len(args))\n    # set analytics sample rate if enabled\n    span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())\n\n    def _finish_span(future):\n        try:\n            # Accessing the result will raise an exception if:\n            #   - The future was cancelled\n            #   - There was an error executing the future (`future.exception()`)\n            #   - The future is in an invalid state\n            future.result()\n        except Exception:\n            span.set_exc_info(*sys.exc_info())\n        finally:\n            span.finish()\n\n    task = func(*args, **kwargs)\n    # Execute command returns a coroutine when no free connections are available\n    # https://github.com/aio-libs/aioredis-py/blob/v1.3.1/aioredis/pool.py#L191\n    task = asyncio.ensure_future(task)\n    task.add_done_callback(_finish_span)\n    return task\n\n\nasync def traced_13_execute_pipeline(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    cmds = []\n    for _, cmd, cmd_args, _ in instance._pipeline:\n        parts = [cmd]\n        parts.extend(cmd_args)\n        cmds.append(stringify_cache_args(parts))\n    resource = \"\\n\".join(cmds)\n    with pin.tracer.trace(\n        redisx.CMD,\n        resource=resource,\n        service=trace_utils.ext_service(pin, config.aioredis),\n        span_type=SpanTypes.REDIS,\n    ) as span:\n\n        span.set_tags(\n            {\n                net.TARGET_HOST: instance._pool_or_conn.address[0],\n                net.TARGET_PORT: instance._pool_or_conn.address[1],\n                redisx.DB: instance._pool_or_conn.db or 0,\n            }\n        )\n\n        span.set_tag(SPAN_MEASURED_KEY)\n        span.set_tag(redisx.RAWCMD, resource)\n        span.set_metric(redisx.PIPELINE_LEN, len(instance._pipeline))\n        # set analytics sample rate if enabled\n        span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())\n\n        return await func(*args, **kwargs)\n", "path": "ddtrace/contrib/aioredis/patch.py"}], "after_files": [{"content": "import asyncio\nimport sys\n\nimport aioredis\n\nfrom ddtrace import config\nfrom ddtrace.internal.utils.wrappers import unwrap as _u\nfrom ddtrace.pin import Pin\nfrom ddtrace.vendor.wrapt import wrap_function_wrapper as _w\n\nfrom .. import trace_utils\nfrom ...constants import ANALYTICS_SAMPLE_RATE_KEY\nfrom ...constants import SPAN_MEASURED_KEY\nfrom ...ext import SpanTypes\nfrom ...ext import net\nfrom ...ext import redis as redisx\nfrom ...internal.utils.formats import stringify_cache_args\nfrom ..redis.util import _trace_redis_cmd\nfrom ..redis.util import _trace_redis_execute_pipeline\n\n\ntry:\n    from aioredis.commands.transaction import _RedisBuffer\nexcept ImportError:\n    _RedisBuffer = None\n\nconfig._add(\"aioredis\", dict(_default_service=\"redis\"))\n\naioredis_version_str = getattr(aioredis, \"__version__\", \"0.0.0\")\naioredis_version = tuple([int(i) for i in aioredis_version_str.split(\".\")])\n\n\ndef patch():\n    if getattr(aioredis, \"_datadog_patch\", False):\n        return\n    setattr(aioredis, \"_datadog_patch\", True)\n    pin = Pin()\n    if aioredis_version >= (2, 0):\n        _w(\"aioredis.client\", \"Redis.execute_command\", traced_execute_command)\n        _w(\"aioredis.client\", \"Redis.pipeline\", traced_pipeline)\n        _w(\"aioredis.client\", \"Pipeline.execute\", traced_execute_pipeline)\n        pin.onto(aioredis.client.Redis)\n    else:\n        _w(\"aioredis\", \"Redis.execute\", traced_13_execute_command)\n        _w(\"aioredis\", \"Redis.pipeline\", traced_13_pipeline)\n        _w(\"aioredis.commands.transaction\", \"Pipeline.execute\", traced_13_execute_pipeline)\n        pin.onto(aioredis.Redis)\n\n\ndef unpatch():\n    if not getattr(aioredis, \"_datadog_patch\", False):\n        return\n\n    setattr(aioredis, \"_datadog_patch\", False)\n    if aioredis_version >= (2, 0):\n        _u(aioredis.client.Redis, \"execute_command\")\n        _u(aioredis.client.Redis, \"pipeline\")\n        _u(aioredis.client.Pipeline, \"execute\")\n    else:\n        _u(aioredis.Redis, \"execute\")\n        _u(aioredis.Redis, \"pipeline\")\n        _u(aioredis.commands.transaction.Pipeline, \"execute\")\n\n\nasync def traced_execute_command(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    with _trace_redis_cmd(pin, config.aioredis, instance, args):\n        return await func(*args, **kwargs)\n\n\ndef traced_pipeline(func, instance, args, kwargs):\n    pipeline = func(*args, **kwargs)\n    pin = Pin.get_from(instance)\n    if pin:\n        pin.onto(pipeline)\n    return pipeline\n\n\nasync def traced_execute_pipeline(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    cmds = [stringify_cache_args(c) for c, _ in instance.command_stack]\n    resource = \"\\n\".join(cmds)\n    with _trace_redis_execute_pipeline(pin, config.aioredis, resource, instance):\n        return await func(*args, **kwargs)\n\n\ndef traced_13_pipeline(func, instance, args, kwargs):\n    pipeline = func(*args, **kwargs)\n    pin = Pin.get_from(instance)\n    if pin:\n        pin.onto(pipeline)\n    return pipeline\n\n\ndef traced_13_execute_command(func, instance, args, kwargs):\n    # If we have a _RedisBuffer then we are in a pipeline\n    if isinstance(instance.connection, _RedisBuffer):\n        return func(*args, **kwargs)\n\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return func(*args, **kwargs)\n\n    # Don't activate the span since this operation is performed as a future which concludes sometime later on in\n    # execution so subsequent operations in the stack are not necessarily semantically related\n    # (we don't want this span to be the parent of all other spans created before the future is resolved)\n    parent = pin.tracer.current_span()\n    span = pin.tracer.start_span(\n        redisx.CMD,\n        service=trace_utils.ext_service(pin, config.aioredis),\n        span_type=SpanTypes.REDIS,\n        activate=False,\n        child_of=parent,\n    )\n\n    span.set_tag(SPAN_MEASURED_KEY)\n    query = stringify_cache_args(args)\n    span.resource = query\n    span.set_tag(redisx.RAWCMD, query)\n    if pin.tags:\n        span.set_tags(pin.tags)\n\n    span.set_tags(\n        {\n            net.TARGET_HOST: instance.address[0],\n            net.TARGET_PORT: instance.address[1],\n            redisx.DB: instance.db or 0,\n        }\n    )\n    span.set_metric(redisx.ARGS_LEN, len(args))\n    # set analytics sample rate if enabled\n    span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())\n\n    def _finish_span(future):\n        try:\n            # Accessing the result will raise an exception if:\n            #   - The future was cancelled (CancelledError)\n            #   - There was an error executing the future (`future.exception()`)\n            #   - The future is in an invalid state\n            future.result()\n        # CancelledError exceptions extend from BaseException as of Python 3.8, instead of usual Exception\n        except BaseException:\n            span.set_exc_info(*sys.exc_info())\n        finally:\n            span.finish()\n\n    task = func(*args, **kwargs)\n    # Execute command returns a coroutine when no free connections are available\n    # https://github.com/aio-libs/aioredis-py/blob/v1.3.1/aioredis/pool.py#L191\n    task = asyncio.ensure_future(task)\n    task.add_done_callback(_finish_span)\n    return task\n\n\nasync def traced_13_execute_pipeline(func, instance, args, kwargs):\n    pin = Pin.get_from(instance)\n    if not pin or not pin.enabled():\n        return await func(*args, **kwargs)\n\n    cmds = []\n    for _, cmd, cmd_args, _ in instance._pipeline:\n        parts = [cmd]\n        parts.extend(cmd_args)\n        cmds.append(stringify_cache_args(parts))\n    resource = \"\\n\".join(cmds)\n    with pin.tracer.trace(\n        redisx.CMD,\n        resource=resource,\n        service=trace_utils.ext_service(pin, config.aioredis),\n        span_type=SpanTypes.REDIS,\n    ) as span:\n\n        span.set_tags(\n            {\n                net.TARGET_HOST: instance._pool_or_conn.address[0],\n                net.TARGET_PORT: instance._pool_or_conn.address[1],\n                redisx.DB: instance._pool_or_conn.db or 0,\n            }\n        )\n\n        span.set_tag(SPAN_MEASURED_KEY)\n        span.set_tag(redisx.RAWCMD, resource)\n        span.set_metric(redisx.PIPELINE_LEN, len(instance._pipeline))\n        # set analytics sample rate if enabled\n        span.set_tag(ANALYTICS_SAMPLE_RATE_KEY, config.aioredis.get_analytics_sample_rate())\n\n        return await func(*args, **kwargs)\n", "path": "ddtrace/contrib/aioredis/patch.py"}]}