code: def test_write(
      self, tmpdir, use_cuda_pinned_tensor, async_queue, cuda_device
  ):
      _skip_for_invalid_environment(
          use_cuda_device=cuda_device,
          use_cuda_pinned_tensor=use_cuda_pinned_tensor,
      )
  
      ref_files = []
      ref_buffers = []
      for i in range(async_queue):
          f, buf = _do_ref_write(tmpdir, i)
          ref_files.append(f)
          ref_buffers.append(buf)
  
      single_submit = True
      overlap_events = True
      h = (
          AsyncIOBuilder()
          .load()
          .aio_handle(
              BLOCK_SIZE,
              QUEUE_DEPTH,
              single_submit,
              overlap_events,
              IO_PARALLEL,
          )
      )
  
      aio_files = []
      aio_buffers = []
      for i in range(async_queue):
          if cuda_device:
              f, buf = _get_test_write_file_and_cuda_buffer(
                  tmpdir, ref_buffers[i], i
              )
          elif use_cuda_pinned_tensor:
              f, buf = _get_test_write_file_and_cpu_buffer(
                  tmpdir, ref_buffers[i], None, i
              )
          else:
              f, buf = _get_test_write_file_and_cpu_buffer(
                  tmpdir, ref_buffers[i], h, i
              )
          aio_files.append(f)
          aio_buffers.append(buf)
  
      use_cpu_locked_tensor = not (
          cuda_device or use_cuda_pinned_tensor
      )
  
      _validate_handle_state(h, single_submit, overlap_events)
  
      for i in range(async_queue):
          read_status = h.async_pwrite(aio_buffers[i], aio_files[i])
          assert read_status == 0
  
      wait_status = h.wait()
      assert wait_status == async_queue
  
      if use_cpu_locked_tensor:
          for t in aio_buffers:
              h.free_cpu_locked_tensor(t)
  
      for i in range(async_queue):
          assert os.path.isfile(aio_files[i])
  
          filecmp.clear_cache()
          assert filecmp.cmp(ref_files[i], aio_files[i], shallow=False)
  
imports: import pytest
  import os
  import filecmp
  import torch
  import deepspeed
  import deepspeed.comm
  from deepspeed.accelerator import get_accelerator
  from deepspeed.ops.op_builder import AsyncIOBuilder
  from unit.common import DistributedTest
label_desc: Use image node_id 2 for calling the TestAsyncQueue obj's underlying member method code with example usage: obj.test_write(tmpdir, use_cuda_pinned_tensor, async_queue, cuda_device) without return types