    def _multi_worker_islice(
        self,
        iterable: Iterable[_T],
    ) -> Iterator[_T]:
        """
        This is just like `shard_iterable` but is for internal use only.

        It has some additional logic to handle `max_instances` based on the distributed
        or multi-process context, and whether or not sharding is handled manually
        in the `_read()` method.
        """
        # This has some complicated logic because any given reader may or may not
        # implement manual multi-process and manual distributed sharding itself.
        # We have to handle all possibilities.

        sharded_slice: Iterator[_T] = iter(iterable)

        # We'll adjust max_instances as we go, depending on what sort of sharding is done.
        # At the end, we want to ensure the total number of instances collected across
        # all workers processes is equal to self.max_instances.
        max_instances = self.max_instances

        if self._distributed_info is not None:
            if max_instances is not None:
                # Need to scale down max_instances because otherwise each node would read self.max_instances,
                # but we really want self.max_instances total across all nodes.
                if self._distributed_info.global_rank < (
                    max_instances % self._distributed_info.world_size
                ):
                    max_instances = max_instances // self._distributed_info.world_size + 1
                else:
                    max_instances = max_instances // self._distributed_info.world_size

            if not self.manual_distributed_sharding:
                sharded_slice = itertools.islice(
                    sharded_slice,
                    self._distributed_info.global_rank,
                    None,
                    self._distributed_info.world_size,
                )

        if self._worker_info is not None:
            if max_instances is not None:
                # Like in the distributed case above, we need to adjust max_instances.
                if self._worker_info.id < (max_instances % self._worker_info.num_workers):
                    max_instances = max_instances // self._worker_info.num_workers + 1
                else:
                    max_instances = max_instances // self._worker_info.num_workers

            if not self.manual_multiprocess_sharding:
                warnings.warn(
                    "Using multi-process data loading without setting "
                    "DatasetReader.manual_multiprocess_sharding to True.\n"
                    "Did you forget to set this?\n"
                    "If you're not handling the multi-process sharding logic within your "
                    "_read() method, there is probably no benefit to using more than one "
                    "worker.",
                    UserWarning,
                )
                sharded_slice = itertools.islice(
                    sharded_slice, self._worker_info.id, None, self._worker_info.num_workers
                )

        if max_instances is not None:
            sharded_slice = itertools.islice(sharded_slice, max_instances)

        return sharded_slice
This is just like `shard_iterable` but is for internal use only.

It has some additional logic to handle `max_instances` based on the distributed
or multi-process context, and whether or not sharding is handled manually
in the `_read()` method.
allennlp/data/dataset_readers/dataset_reader.py