# Concurrency and Async Tinker client methods return an [`APIFuture`](https://tinker-docs.thinkingmachines.ai/tinker/api-reference/apifuture/index.md) which supports both asyncio and synchronous contexts: - `await client.method(...)` in asyncio contexts - `client.method(...).result()` in sync contexts Methods that create other clients are also async, so make sure to use `await` or `.result()` before using sampling and training clients: ```python sampling_client = await service_client.create_sampling_client(...) training_client = await service_client.create_lora_training_client(...) ``` `APIFuture`s also easily interoperate with standard Python asyncio and concurrency utilities: - Use `api_future.coroutine()` to create a coroutine that can be used with Python built-ins like `asyncio.create_task` and `asyncio.gather`. - Use `api_future.future()` to create a standard `concurrent.futures.Future`. ### Pipeline training operations See [Clock Cycles & Pipelining](https://tinker-docs.thinkingmachines.ai/tinker/under-the-hood/index.md) for more details on how Tinker clock cycles work behind the scenes. The general pattern is to submit both training operations in order before waiting: ```python # no await, just submit. Tinker guarantees execution order. forward_backward_future = training_client.forward_backward(batch, "cross_entropy") optim_future = training_client.optim_step(adam_params) # await the futures after submitting so that they are on the same clock cycle forward_backward_result = await forward_backward_future optim_result = await optim_future ``` ## Migration guide for 0.37.0 and below Prior to `0.38.0`, Tinker SDK methods did not consistently return `APIFuture`s, so there were explicitly `_async` variants of many methods. To migrate, follow the below instructions (or send them to a coding agent). ### Client creation #### Async ```python # old sampling_client = await service_client.create_sampling_client_async(...) training_client = await service_client.create_lora_training_client_async(...) # new sampling_client = await service_client.create_sampling_client(...) training_client = await service_client.create_lora_training_client(...) ``` #### Sync ```python # old sampling_client = service_client.create_sampling_client(...) training_client = service_client.create_lora_training_client(...) # new sampling_client = service_client.create_sampling_client(...).result() training_client = service_client.create_lora_training_client(...).result() ``` ### Rest Client ```python # old rest_client = service_client.create_rest_client(...) # new # for managing runs rest_client = tinker.RestClient() # for admin operations like billing admin_client = tinker.AdminClient() ``` ### Training Operations #### Async ```python # old fwd_bwd_fut = await training_client.forward_backward_async(batch, "cross_entropy") optim_fut = await training_client.optim_step_async(adam_params) await fwd_bwd_fut await optim_fut # new fwd_bwd_fut = training_client.forward_backward(batch, "cross_entropy") optim_fut = training_client.optim_step(adam_params) await fwd_bwd_fut await optim_fut ``` #### Sync ```python # old fwd_bwd_fut = training_client.forward_backward(batch, "cross_entropy") optim_fut = training_client.optim_step(adam_params) fwd_bwd_fut.result() optim_fut.result() # new: no change fwd_bwd_fut = training_client.forward_backward(batch, "cross_entropy") optim_fut = training_client.optim_step(adam_params) fwd_bwd_fut.result() optim_fut.result() ``` ### Sampling #### Async ```python # old response = await sampling_client.sample_async(prompt, num_samples=1, sampling_params=params) logprobs = await sampling_client.compute_logprobs_async(prompt) # new response = await sampling_client.sample(prompt, num_samples=1, sampling_params=params) logprobs = await sampling_client.compute_logprobs(prompt) ``` #### Sync ```python # old response = sampling_client.sample(prompt, num_samples=1, sampling_params=params).result() logprobs = sampling_client.compute_logprobs(prompt).result() # new: no change response = sampling_client.sample(prompt, num_samples=1, sampling_params=params).result() logprobs = sampling_client.compute_logprobs(prompt).result() ``` `sample()` and `compute_logprobs()` used to return a `concurrent.futures.Future`. Calling its methods, such as `done()` or `add_done_callback()`, on the returned future still works but warns; call `.future()` first instead. #### Concurrent requests with `asyncio.gather` ```python # old responses = await asyncio.gather( *[sampling_client.sample_async(prompt, num_samples=1, sampling_params=params) for prompt in prompts] ) # new responses = await asyncio.gather( *[sampling_client.sample(prompt, num_samples=1, sampling_params=params).coroutine() for prompt in prompts] ) ``` Each `sample()` call submits its request immediately; `.coroutine()` only changes how you wait for it. #### Sampling clients from a training client `save_weights_and_get_sampling_client()` and `TrainingClient.create_sampling_client()` now return a future of the client, like the `ServiceClient` factories. ```python # old (async) sampling_client = await training_client.save_weights_and_get_sampling_client_async(name="step-100") # new (async) sampling_client = await training_client.save_weights_and_get_sampling_client(name="step-100") # old (sync) sampling_client = training_client.save_weights_and_get_sampling_client(name="step-100") # new (sync) sampling_client = training_client.save_weights_and_get_sampling_client(name="step-100").result() ``` ### Full Replacement table The replacements use `await`; in synchronous code, call `.result()` on the returned future instead. Deprecated methods still run but raise a `FutureWarning`; methods marked removed no longer exist. | Previous API | Replacement | | ----------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `await training_client.get_info_async()` | `training_client.get_info()` (no network request) | | `await sampling_client.get_base_model_async()` | `sampling_client.get_base_model()` (no network request) | | `await service_client.create_lora_training_client_async(...)` | `await service_client.create_lora_training_client(...)` | | `await service_client.create_training_client_from_state_async(path)` | `await service_client.create_training_client_from_state(path)` | | `await service_client.create_training_client_from_state_with_optimizer_async(path)` | `await service_client.create_training_client_from_state_with_optimizer(path)` | | `await service_client.create_sampling_client_async(...)` | `await service_client.create_sampling_client(...)` | | `await training_client.create_sampling_client_async(path)` | `await training_client.create_sampling_client(path)` | | `await training_client.save_weights_and_get_sampling_client_async(...)` | `await training_client.save_weights_and_get_sampling_client(...)` | | `training_client = service_client.create_lora_training_client(...)` | `training_client = service_client.create_lora_training_client(...).result()` | | `training_client = service_client.create_training_client_from_state(path)` | `training_client = service_client.create_training_client_from_state(path).result()` | | `training_client = service_client.create_training_client_from_state_with_optimizer(path)` | `training_client = service_client.create_training_client_from_state_with_optimizer(path).result()` | | `sampling_client = service_client.create_sampling_client(...)` | `sampling_client = service_client.create_sampling_client(...).result()` | | `sampling_client = training_client.create_sampling_client(path)` | `sampling_client = training_client.create_sampling_client(path).result()` | | `sampling_client = training_client.save_weights_and_get_sampling_client(...)` | `sampling_client = training_client.save_weights_and_get_sampling_client(...).result()` | | `await sampling_client.sample_async(...)` | `await sampling_client.sample(...)` | | `await sampling_client.compute_logprobs_async(...)` | `await sampling_client.compute_logprobs(...)` | | `fut = await training_client.forward_async(...)` | `fut = training_client.forward(...)` | | `fut = await training_client.forward_backward_async(...)` | `fut = training_client.forward_backward(...)` | | `fut = await training_client.forward_backward_custom_async(...)` | `fut = training_client.forward_backward_custom(...)` | | `fut = await training_client.optim_step_async(...)` | `fut = training_client.optim_step(...)` | | `fut = await training_client.save_state_async(...)` | `fut = training_client.save_state(...)` | | `fut = await training_client.save_weights_for_sampler_async(...)` | `fut = training_client.save_weights_for_sampler(...)` | | `fut = await training_client.save_weights_external_async(...)` | `fut = training_client.save_weights_external(...)` | | `training_client.load_state(path)` / `load_state_async(path)` | `training_client = await service_client.create_training_client_from_state(path)` | | `training_client.load_state_with_optimizer(path)` / `load_state_with_optimizer_async(path)` | `training_client = await service_client.create_training_client_from_state_with_optimizer(path)` | | `service_client.get_server_capabilities()` / `await service_client.get_server_capabilities_async()` | `await rest_client.get_server_capabilities()` | | `service_client.create_rest_client()` | `rest_client = tinker.RestClient()` | | `await rest_client.get_training_run_async(training_run_id)` | `await rest_client.get_training_run(training_run_id)` | | `await rest_client.get_training_run_by_tinker_path_async(tinker_path)` | `await rest_client.get_training_run_by_tinker_path(tinker_path)` | | `await rest_client.list_training_runs_async(...)` | `await rest_client.list_training_runs(...)` | | `await rest_client.list_checkpoints_async(training_run_id)` | `await rest_client.list_checkpoints(training_run_id)` | | `await rest_client.list_user_checkpoints_async(...)` | `await rest_client.list_user_checkpoints(...)` | | `await rest_client.get_external_weights_urls_async(tinker_path)` | `await rest_client.get_external_weights_urls(tinker_path)` | | `await rest_client.delete_checkpoint_async(training_run_id, checkpoint_id)` | `await rest_client.delete_checkpoint(training_run_id, checkpoint_id)` | | `await rest_client.delete_checkpoint_from_tinker_path_async(tinker_path)` | `await rest_client.delete_checkpoint_from_tinker_path(tinker_path)` | | `await rest_client.copy_weights_async(path, ...)` | `await rest_client.copy_weights(path, ...)` | | `await rest_client.publish_checkpoint_from_tinker_path_async(tinker_path)` | `await rest_client.publish_checkpoint_from_tinker_path(tinker_path)` | | `await rest_client.unpublish_checkpoint_from_tinker_path_async(tinker_path)` | `await rest_client.unpublish_checkpoint_from_tinker_path(tinker_path)` | | `await rest_client.set_checkpoint_ttl_from_tinker_path_async(tinker_path, ttl_seconds)` | `await rest_client.set_checkpoint_ttl_from_tinker_path(tinker_path, ttl_seconds)` | | `await rest_client.get_session_async(session_id)` | `await rest_client.get_session(session_id)` | | `await rest_client.list_sessions_async(...)` | `await rest_client.list_sessions(...)` | | `await rest_client.assign_session_project_async(session_id, project_id)` | `await rest_client.assign_session_project(session_id, project_id)` | | `await rest_client.export_session_trace_async(session_id)` | `await rest_client.export_session_trace(session_id)` | | `await rest_client.get_sampler_async(sampler_id)` | `await rest_client.get_sampler(sampler_id)` | | `await rest_client.search_sessions_async(request)` (removed) | `await rest_client.search_sessions(request)` | | `rest_client.get_billing_usage(...)` / `get_billing_usage_async(...)` | `await admin_client.get_billing_usage(...)` | | `rest_client.get_current_checkpoint_storage_usage()` / `get_current_checkpoint_storage_usage_async()` | `await admin_client.get_current_checkpoint_storage_usage()` | | `rest_client.get_audit_log(...)` / `get_audit_log_async(...)` | `await admin_client.get_audit_log(...)` | | `await admin_client.get_billing_usage_async(...)` | `await admin_client.get_billing_usage(...)` | | `await admin_client.get_current_checkpoint_storage_usage_async()` | `await admin_client.get_current_checkpoint_storage_usage()` | | `await admin_client.get_audit_log_async(...)` | `await admin_client.get_audit_log(...)` | | `done()` / `add_done_callback()` on a `RestClient` or `AdminClient` result, or passing it to `concurrent.futures.wait()` / `as_completed()` (removed) | Call `.future()` on the result first | | `done()` / `add_done_callback()` on a `sample()` or `compute_logprobs()` result | Call `.future()` on the result first | | `rest_client.get_checkpoint_archive_url(...)` / `get_checkpoint_archive_url_from_tinker_path(...)` (removed in 0.36.0) | Export with `save_weights_external`, then use `get_external_weights_urls`; see [Download weights](https://tinker-docs.thinkingmachines.ai/tinker/howto/checkpoints/#download-weights). | For the training rows, keep awaiting `fut` as before, and submit the next operation before awaiting, as in [Pipeline training operations](#pipeline-training-operations).