A thread-local that contains the following attributes.

        current_task_id: For the main thread, this field is the ID of this
            worker's current running task; for other threads, this field is a
            fake random ID.
        task_index: The number of tasks that have been submitted from the
            current task.
        put_index: The number of objects that have been put from the current
            task.

def task_context(self):
        """A thread-local that contains the following attributes.

        current_task_id: For the main thread, this field is the ID of this
            worker's current running task; for other threads, this field is a
            fake random ID.
        task_index: The number of tasks that have been submitted from the
            current task.
        put_index: The number of objects that have been put from the current
            task.
        """
        if not hasattr(self._task_context, "initialized"):
            # Initialize task_context for the current thread.
            if ray.utils.is_main_thread():
                # If this is running on the main thread, initialize it to
                # NIL. The actual value will set when the worker receives
                # a task from raylet backend.
                self._task_context.current_task_id = TaskID.nil()
            else:
                # If this is running on a separate thread, then the mapping
                # to the current task ID may not be correct. Generate a
                # random task ID so that the backend can differentiate
                # between different threads.
                self._task_context.current_task_id = TaskID(_random_string())
                if getattr(self, "_multithreading_warned", False) is not True:
                    logger.warning(
                        "Calling ray.get or ray.wait in a separate thread "
                        "may lead to deadlock if the main thread blocks on "
                        "this thread and there are not enough resources to "
                        "execute more tasks")
                    self._multithreading_warned = True

            self._task_context.task_index = 0
            self._task_context.put_index = 1
            self._task_context.initialized = True
        return self._task_context