We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
tuple index out of range of threading.py
After upgrading from ddtrace==0.46.0 to version ddtrace==0.55.4 my service crash with IndexError.
```
Traceback (most recent call last):
  File "/my_service/services/base_service.py", line 105, in run
    futures.append(executor.submit(fn=self._single_entry_point_run, entry_point=entry_point))
  File "/my_service/venv/lib/python3.7/site-packages/ddtrace/contrib/futures/threading.py", line 26, in _wrap_submit
    fn = args[0]
IndexError: tuple index out of range
```

I'm facing this issue even when setting futures=False.
`patch_all(celery=True, django=True, psycopg2=True, redis=True, futures=True)`


--- 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/futures/threading.py`
Content:
```
1 import ddtrace
2 
3 
4 def _wrap_submit(func, instance, args, kwargs):
5     """
6     Wrap `Executor` method used to submit a work executed in another
7     thread. This wrapper ensures that a new `Context` is created and
8     properly propagated using an intermediate function.
9     """
10     # If there isn't a currently active context, then do not create one
11     # DEV: Calling `.active()` when there isn't an active context will create a new context
12     # DEV: We need to do this in case they are either:
13     #        - Starting nested futures
14     #        - Starting futures from outside of an existing context
15     #
16     #      In either of these cases we essentially will propagate the wrong context between futures
17     #
18     #      The resolution is to not create/propagate a new context if one does not exist, but let the
19     #      future's thread create the context instead.
20     current_ctx = None
21     if ddtrace.tracer.context_provider._has_active_context():
22         current_ctx = ddtrace.tracer.context_provider.active()
23 
24     # extract the target function that must be executed in
25     # a new thread and the `target` arguments
26     fn = args[0]
27     fn_args = args[1:]
28     return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)
29 
30 
31 def _wrap_execution(ctx, fn, args, kwargs):
32     """
33     Intermediate target function that is executed in a new thread;
34     it receives the original function with arguments and keyword
35     arguments, including our tracing `Context`. The current context
36     provider sets the Active context in a thread local storage
37     variable because it's outside the asynchronous loop.
38     """
39     if ctx is not None:
40         ddtrace.tracer.context_provider.activate(ctx)
41     return fn(*args, **kwargs)
42
```
--- 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/futures/threading.py b/ddtrace/contrib/futures/threading.py
--- a/ddtrace/contrib/futures/threading.py
+++ b/ddtrace/contrib/futures/threading.py
@@ -21,10 +21,12 @@
     if ddtrace.tracer.context_provider._has_active_context():
         current_ctx = ddtrace.tracer.context_provider.active()
 
-    # extract the target function that must be executed in
-    # a new thread and the `target` arguments
-    fn = args[0]
-    fn_args = args[1:]
+    # The target function can be provided as a kwarg argument "fn" or the first positional argument
+    if "fn" in kwargs:
+        fn = kwargs.pop("fn")
+        fn_args = args
+    else:
+        fn, fn_args = args[0], args[1:]
     return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)

{"golden_diff": "diff --git a/ddtrace/contrib/futures/threading.py b/ddtrace/contrib/futures/threading.py\n--- a/ddtrace/contrib/futures/threading.py\n+++ b/ddtrace/contrib/futures/threading.py\n@@ -21,10 +21,12 @@\n     if ddtrace.tracer.context_provider._has_active_context():\n         current_ctx = ddtrace.tracer.context_provider.active()\n \n-    # extract the target function that must be executed in\n-    # a new thread and the `target` arguments\n-    fn = args[0]\n-    fn_args = args[1:]\n+    # The target function can be provided as a kwarg argument \"fn\" or the first positional argument\n+    if \"fn\" in kwargs:\n+        fn = kwargs.pop(\"fn\")\n+        fn_args = args\n+    else:\n+        fn, fn_args = args[0], args[1:]\n     return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)\n", "issue": "tuple index out of range of threading.py\nAfter upgrading from ddtrace==0.46.0 to version ddtrace==0.55.4 my service crash with IndexError.\r\n```\r\nTraceback (most recent call last):\r\n  File \"/my_service/services/base_service.py\", line 105, in run\r\n    futures.append(executor.submit(fn=self._single_entry_point_run, entry_point=entry_point))\r\n  File \"/my_service/venv/lib/python3.7/site-packages/ddtrace/contrib/futures/threading.py\", line 26, in _wrap_submit\r\n    fn = args[0]\r\nIndexError: tuple index out of range\r\n```\r\n\r\nI'm facing this issue even when setting futures=False.\r\n`patch_all(celery=True, django=True, psycopg2=True, redis=True, futures=True)`\r\n\n", "before_files": [{"content": "import ddtrace\n\n\ndef _wrap_submit(func, instance, args, kwargs):\n    \"\"\"\n    Wrap `Executor` method used to submit a work executed in another\n    thread. This wrapper ensures that a new `Context` is created and\n    properly propagated using an intermediate function.\n    \"\"\"\n    # If there isn't a currently active context, then do not create one\n    # DEV: Calling `.active()` when there isn't an active context will create a new context\n    # DEV: We need to do this in case they are either:\n    #        - Starting nested futures\n    #        - Starting futures from outside of an existing context\n    #\n    #      In either of these cases we essentially will propagate the wrong context between futures\n    #\n    #      The resolution is to not create/propagate a new context if one does not exist, but let the\n    #      future's thread create the context instead.\n    current_ctx = None\n    if ddtrace.tracer.context_provider._has_active_context():\n        current_ctx = ddtrace.tracer.context_provider.active()\n\n    # extract the target function that must be executed in\n    # a new thread and the `target` arguments\n    fn = args[0]\n    fn_args = args[1:]\n    return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)\n\n\ndef _wrap_execution(ctx, fn, args, kwargs):\n    \"\"\"\n    Intermediate target function that is executed in a new thread;\n    it receives the original function with arguments and keyword\n    arguments, including our tracing `Context`. The current context\n    provider sets the Active context in a thread local storage\n    variable because it's outside the asynchronous loop.\n    \"\"\"\n    if ctx is not None:\n        ddtrace.tracer.context_provider.activate(ctx)\n    return fn(*args, **kwargs)\n", "path": "ddtrace/contrib/futures/threading.py"}], "after_files": [{"content": "import ddtrace\n\n\ndef _wrap_submit(func, instance, args, kwargs):\n    \"\"\"\n    Wrap `Executor` method used to submit a work executed in another\n    thread. This wrapper ensures that a new `Context` is created and\n    properly propagated using an intermediate function.\n    \"\"\"\n    # If there isn't a currently active context, then do not create one\n    # DEV: Calling `.active()` when there isn't an active context will create a new context\n    # DEV: We need to do this in case they are either:\n    #        - Starting nested futures\n    #        - Starting futures from outside of an existing context\n    #\n    #      In either of these cases we essentially will propagate the wrong context between futures\n    #\n    #      The resolution is to not create/propagate a new context if one does not exist, but let the\n    #      future's thread create the context instead.\n    current_ctx = None\n    if ddtrace.tracer.context_provider._has_active_context():\n        current_ctx = ddtrace.tracer.context_provider.active()\n\n    # The target function can be provided as a kwarg argument \"fn\" or the first positional argument\n    if \"fn\" in kwargs:\n        fn = kwargs.pop(\"fn\")\n        fn_args = args\n    else:\n        fn, fn_args = args[0], args[1:]\n    return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)\n\n\ndef _wrap_execution(ctx, fn, args, kwargs):\n    \"\"\"\n    Intermediate target function that is executed in a new thread;\n    it receives the original function with arguments and keyword\n    arguments, including our tracing `Context`. The current context\n    provider sets the Active context in a thread local storage\n    variable because it's outside the asynchronous loop.\n    \"\"\"\n    if ctx is not None:\n        ddtrace.tracer.context_provider.activate(ctx)\n    return fn(*args, **kwargs)\n", "path": "ddtrace/contrib/futures/threading.py"}]}