Skip to content

Custom Task

Quantum methods for running custom tasks.

CustomTask

For running custom tasks.

Source code in src/scruby/mixins/custom_task.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class CustomTask:
    """For running custom tasks."""

    @final
    async def run_custom_task(
        self,
        custom_task: Any,
        filter_fn: Callable = lambda _: True,
    ) -> Any:
        """For run a custom task.

        Attention:
            - The search is based on the effect of a quantum loop.
            - The search effectiveness depends on the number of processor threads.

        Args:
            custom_task (Any): Custom task class.
            filter_fn (Callable): A function that execute the conditions of filtering.
                                  By default, it searches all documents.

        Returns:
            The result of a custom task.
        """
        hash_reduce_left: int = self._hash_reduce_left
        assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `run_custom_task` method."

        search_task_fn = self._task_find
        branch_numbers = range(self._max_number_branch)
        db_root = self._db_root
        class_model = self._class_model
        mode = (self._mode,)
        stop_signal = Event()
        stop_outer_loop: bool = False

        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    search_task_fn,
                    filter_fn,
                    branch_number,
                    hash_reduce_left,
                    db_root,
                    class_model,
                    mode,
                    stop_signal,
                )
                for branch_number in branch_numbers
            ]

            for future in as_completed(futures):
                docs = await future.result()

                if docs is not None:
                    for doc in docs:
                        custom_task.accept(doc)
                        if custom_task.stop_signal:
                            # Cancel all pending tasks in the queue instantly
                            executor.shutdown(wait=False, cancel_futures=True)
                            # Trigger the event to tell running tasks to exit
                            stop_signal.set()
                            # Stop loops
                            stop_outer_loop = True
                            break

                    if stop_outer_loop:
                        break

        return custom_task.result()

run_custom_task(custom_task, filter_fn=lambda _: True) async

For run a custom task.

Attention
  • The search is based on the effect of a quantum loop.
  • The search effectiveness depends on the number of processor threads.

Parameters:

Name Type Description Default
custom_task Any

Custom task class.

 required
filter_fn Callable

A function that execute the conditions of filtering. By default, it searches all documents.

 lambda _: True

Returns:

Type Description
Any

The result of a custom task.
Source code in src/scruby/mixins/custom_task.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@final
async def run_custom_task(
    self,
    custom_task: Any,
    filter_fn: Callable = lambda _: True,
) -> Any:
    """For run a custom task.

    Attention:
        - The search is based on the effect of a quantum loop.
        - The search effectiveness depends on the number of processor threads.

    Args:
        custom_task (Any): Custom task class.
        filter_fn (Callable): A function that execute the conditions of filtering.
                              By default, it searches all documents.

    Returns:
        The result of a custom task.
    """
    hash_reduce_left: int = self._hash_reduce_left
    assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `run_custom_task` method."

    search_task_fn = self._task_find
    branch_numbers = range(self._max_number_branch)
    db_root = self._db_root
    class_model = self._class_model
    mode = (self._mode,)
    stop_signal = Event()
    stop_outer_loop: bool = False

    # Run quantum loop
    with ThreadPoolExecutor(self._max_workers) as executor:
        futures: list[Future] = [
            executor.submit(
                search_task_fn,
                filter_fn,
                branch_number,
                hash_reduce_left,
                db_root,
                class_model,
                mode,
                stop_signal,
            )
            for branch_number in branch_numbers
        ]

        for future in as_completed(futures):
            docs = await future.result()

            if docs is not None:
                for doc in docs:
                    custom_task.accept(doc)
                    if custom_task.stop_signal:
                        # Cancel all pending tasks in the queue instantly
                        executor.shutdown(wait=False, cancel_futures=True)
                        # Trigger the event to tell running tasks to exit
                        stop_signal.set()
                        # Stop loops
                        stop_outer_loop = True
                        break

                if stop_outer_loop:
                    break

    return custom_task.result()