class_docstr: Performs sampling based on input seeds grouped into batches by
  a batch id.  Writes the output minibatches to parquet, with
  partition sizes specified by the user.  Allows controlling the
  number of input seeds per sampling function call.  Supports
  basic logging.
  
  Batches in each partition that are empty are discarded, and the remaining non-empty
  batches are renumbered to be contiguous starting from the first
  batch id in the partition.
  This means that the output batch ids may not match the input batch ids.
  See GitHub issue #3794 for more details.
class_docstr_tok: ["Performs","sampling","based","on","input","seeds","grouped","into","batches","by","a","batch","id",".","Writes","the","output","minibatches","to","parquet",",","with","partition","sizes","specified","by","the","user",".","Allows","controlling","the","number","of","input","seeds","per","sampling","function","call",".","Supports","basic","logging",".","Batches","in","each","partition","that","are","empty","are","discarded",",","and","the","remaining","non-empty","batches","are","renumbered","to","be","contiguous","starting","from","the","first","batch","id","in","the","partition",".","This","means","that","the","output","batch","ids","may","not","match","the","input","batch","ids",".","See","GitHub","issue","#","3794","for","more","details","."]
docstr: Computes all uncomputed batches
docstr_tok: ["Computes","all","uncomputed","batches"]
code: def flush(self) -> None:
      """
      Computes all uncomputed batches
      """
      if self.size == 0:
          return
  
      start_time_calc_batches = time.perf_counter()
      if isinstance(self.__batches, dask_cudf.DataFrame):
          self.__batches = self.__batches.persist()
  
      min_batch_id = self.__batches[self.batch_col_name].min()
      if isinstance(self.__batches, dask_cudf.DataFrame):
          min_batch_id = min_batch_id.persist()
      else:
          min_batch_id = int(min_batch_id)
  
      partition_size = self.batches_per_partition * self.batch_size
      partitions_per_call = (
          self.seeds_per_call + partition_size - 1
      ) // partition_size
      npartitions = partitions_per_call
  
      max_batch_id = (
          min_batch_id + npartitions * self.batches_per_partition - 1
      )
      if isinstance(self.__batches, dask_cudf.DataFrame):
          max_batch_id = max_batch_id.persist()
  
      batch_id_filter = (
          self.__batches[self.batch_col_name] <= max_batch_id
      )
      if isinstance(batch_id_filter, dask_cudf.Series):
          batch_id_filter = batch_id_filter.persist()
  
      end_time_calc_batches = time.perf_counter()
      self.__logger.info(
          f"Calculated batches to sample; min = {min_batch_id}"
          f" and max = {max_batch_id};"
          f" took {end_time_calc_batches - start_time_calc_batches:.4f} s"
      )
  
      if isinstance(
          self.__graph._plc_graph, pylibcugraph.graphs.SGGraph
      ):
          sample_fn = cugraph.uniform_neighbor_sample
      else:
          sample_fn = cugraph.dask.uniform_neighbor_sample
          self.__sample_call_args.update(
              {
                  "_multiple_clients": True,
                  "keep_batches_together": True,
                  "min_batch_id": min_batch_id,
                  "max_batch_id": max_batch_id,
              }
          )
  
      start_time_sample_call = time.perf_counter()
  
      # Call uniform neighbor sample
      output = sample_fn(
          self.__graph,
          **self.__sample_call_args,
          start_list=self.__batches[
              [self.start_col_name, self.batch_col_name]
          ][batch_id_filter],
          with_batch_ids=True,
          with_edge_properties=True,
          return_offsets=True,
          renumber=self.__renumber,
          # use_legacy_names=False,
      )
  
      if self.__renumber:
          samples, offsets, renumber_map = output
      else:
          samples, offsets = output
          renumber_map = None
  
      end_time_sample_call = time.perf_counter()
      sample_runtime = end_time_sample_call - start_time_sample_call
  
      self.__logger.info(
          f"Called uniform neighbor sample, took {sample_runtime:.4f} s"
      )
  
      # Filter batches to remove those already processed
      self.__batches = self.__batches[~batch_id_filter]
      del batch_id_filter
      if isinstance(self.__batches, dask_cudf.DataFrame):
          self.__batches = self.__batches.persist()
  
      start_time_write = time.perf_counter()
  
      # Write batches to parquet
      self.__write(samples, offsets, renumber_map)
      if isinstance(self.__batches, dask_cudf.DataFrame):
          futures = [f.release() for f in futures_of(samples)] + [
              f.release() for f in futures_of(offsets)
          ]
          if renumber_map is not None:
              futures += [f.release() for f in futures_of(renumber_map)]
          wait(futures)
  
      del samples
      del offsets
      if renumber_map is not None:
          del renumber_map
  
      end_time_write = time.perf_counter()
      write_runtime = end_time_write - start_time_write
      self.__logger.info(
          f"Wrote samples to parquet, took {write_runtime} seconds"
      )
  
      current_size = self.size
      if current_size > 0:
          self.__logger.info(
              f"There are still {current_size} samples remaining, "
              "calling flush() again..."
          )
          self.flush()
  
imports: import os
  from typing import Union
  import cudf
  import dask_cudf
  from dask.distributed import wait
  from dask.distributed import futures_of
  import cugraph
  import pylibcugraph
  from cugraph.gnn.data_loading.bulk_sampler_io import write_samples
  import warnings
  import logging
  import time
label_desc: Use image node_id 8 for calling the BulkSampler obj's underlying member method code with example usage: obj.flush() without return types