16981563750

Committed 14 Aug 2025 10:49PM UTC coverage: 86.896% (+0.04%) from 86.852%

Build # 16981563750

Build Type

push

github

Committed by

web-flow

Commit Message

add support for Fn::Tranform in CFnV2 (#12966)

Co-authored-by: Simon Walker <simon.walker@localstack.cloud>

Coverage Stats

181 of 195 new or added lines in 6 files covered. (92.82%)

348 existing lines in 22 files now uncovered.

66915 of 77006 relevant lines covered (86.9%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.5

/localstack-core/localstack/services/sns/executor.py

import itertools
import logging
import os
import queue
import threading

LOG = logging.getLogger(__name__)


def _worker(work_queue: queue.Queue):
    try:
        while True:
            work_item = work_queue.get(block=True)
            if work_item is None:
                return
            work_item.run()
            # delete reference to the work item to avoid it being in memory until the next blocking `queue.get` call returns
            del work_item

    except Exception:
        LOG.error(
            "Exception in worker",
            exc_info=LOG.isEnabledFor(logging.DEBUG),
        )


class _WorkItem:
    def __init__(self, fn, args, kwargs):
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    def run(self):
        try:
            self.fn(*self.args, **self.kwargs)
        except Exception:
            LOG.error(
                "Unhandled Exception in while running %s",
                self.fn.__name__,
                exc_info=LOG.isEnabledFor(logging.DEBUG),
            )


class TopicPartitionedThreadPoolExecutor:
    """
    This topic partition the work between workers based on Topics.
    It guarantees that each Topic only has one worker assigned, and thus that the tasks will be executed sequentially.

    Loosely based on ThreadPoolExecutor for stdlib, but does not return Future as SNS does not need it (fire&forget)
    Could be extended if needed to fit other needs.

    Currently, we do not re-balance between workers if some of them have more load. This could be investigated.
    """

    # Used to assign unique thread names when thread_name_prefix is not supplied.
    _counter = itertools.count().__next__

    def __init__(self, max_workers: int = None, thread_name_prefix: str = ""):
        if max_workers is None:
            max_workers = min(32, (os.cpu_count() or 1) + 4)
        if max_workers <= 0:
            raise ValueError("max_workers must be greater than 0")

        self._max_workers = max_workers
        self._thread_name_prefix = (
            thread_name_prefix or f"TopicThreadPoolExecutor-{self._counter()}"
        )

        # for now, the pool isn't fair and is not redistributed depending on load
        self._pool = {}
        self._shutdown = False
        self._lock = threading.Lock()
        self._threads = set()
        self._work_queues = []
        self._cycle = itertools.cycle(range(max_workers))

    def _add_worker(self):
        work_queue = queue.SimpleQueue()
        self._work_queues.append(work_queue)
        thread_name = f"{self._thread_name_prefix}_{len(self._threads)}"
        t = threading.Thread(name=thread_name, target=_worker, args=(work_queue,))
        t.daemon = True
        t.start()
        self._threads.add(t)

    def _get_work_queue(self, topic: str) -> queue.SimpleQueue:
        if not (work_queue := self._pool.get(topic)):
            if len(self._threads) < self._max_workers:
                self._add_worker()

            # we cycle through the possible indexes for a work queue, in order to distribute the load across
            # once we get to the max amount of worker, the cycle will start back at 0
            index = next(self._cycle)
            work_queue = self._work_queues[index]

            # TODO: the pool is not cleaned up at the moment, think about the clean-up interface
            self._pool[topic] = work_queue
        return work_queue

    def submit(self, fn, topic, /, *args, **kwargs) -> None:
        with self._lock:
            work_queue = self._get_work_queue(topic)

            if self._shutdown:
                raise RuntimeError("cannot schedule new futures after shutdown")

            w = _WorkItem(fn, args, kwargs)
            work_queue.put(w)

    def shutdown(self, wait=True):
        with self._lock:
            self._shutdown = True

            # Send a wake-up to prevent threads calling
            # _work_queue.get(block=True) from permanently blocking.
            for work_queue in self._work_queues:
                work_queue.put(None)

        if wait:
            for t in self._threads:
                t.join()

1	import itertools	1✔
2	import logging	1✔
3	import os	1✔
4	import queue	1✔
5	import threading	1✔
6
7	LOG = logging.getLogger(__name__)	1✔
8
9
10	def _worker(work_queue: queue.Queue):	1✔
11	try:	1✔
12	while True:	1✔
13	work_item = work_queue.get(block=True)	1✔
14	if work_item is None:	1✔
15	return	1✔
16	work_item.run()	1✔
17	# delete reference to the work item to avoid it being in memory until the next blocking `queue.get` call returns
18	del work_item	1✔
19
20	except Exception:	×
21	LOG.error(	×
22	"Exception in worker",
23	exc_info=LOG.isEnabledFor(logging.DEBUG),
24	)
25
26
27	class _WorkItem:	1✔
28	def __init__(self, fn, args, kwargs):	1✔
29	self.fn = fn	1✔
30	self.args = args	1✔
31	self.kwargs = kwargs	1✔
32
33	def run(self):	1✔
34	try:	1✔
35	self.fn(self.args, *self.kwargs)	1✔
UNCOV 36	except Exception:	×
UNCOV 37	LOG.error(	×
38	"Unhandled Exception in while running %s",
39	self.fn.__name__,
40	exc_info=LOG.isEnabledFor(logging.DEBUG),
41	)
42
43
44	class TopicPartitionedThreadPoolExecutor:	1✔
45	"""
46	This topic partition the work between workers based on Topics.
47	It guarantees that each Topic only has one worker assigned, and thus that the tasks will be executed sequentially.
48
49	Loosely based on ThreadPoolExecutor for stdlib, but does not return Future as SNS does not need it (fire&forget)
50	Could be extended if needed to fit other needs.
51
52	Currently, we do not re-balance between workers if some of them have more load. This could be investigated.
53	"""
54
55	# Used to assign unique thread names when thread_name_prefix is not supplied.
56	_counter = itertools.count().__next__	1✔
57
58	def __init__(self, max_workers: int = None, thread_name_prefix: str = ""):	1✔
59	if max_workers is None:	1✔
UNCOV 60	max_workers = min(32, (os.cpu_count() or 1) + 4)	×
61	if max_workers <= 0:	1✔
UNCOV 62	raise ValueError("max_workers must be greater than 0")	×
63
64	self._max_workers = max_workers	1✔
65	self._thread_name_prefix = (	1✔
66	thread_name_prefix or f"TopicThreadPoolExecutor-{self._counter()}"
67	)
68
69	# for now, the pool isn't fair and is not redistributed depending on load
70	self._pool = {}	1✔
71	self._shutdown = False	1✔
72	self._lock = threading.Lock()	1✔
73	self._threads = set()	1✔
74	self._work_queues = []	1✔
75	self._cycle = itertools.cycle(range(max_workers))	1✔
76
77	def _add_worker(self):	1✔
78	work_queue = queue.SimpleQueue()	1✔
79	self._work_queues.append(work_queue)	1✔
80	thread_name = f"{self._thread_name_prefix}_{len(self._threads)}"	1✔
81	t = threading.Thread(name=thread_name, target=_worker, args=(work_queue,))	1✔
82	t.daemon = True	1✔
83	t.start()	1✔
84	self._threads.add(t)	1✔
85
86	def _get_work_queue(self, topic: str) -> queue.SimpleQueue:	1✔
87	if not (work_queue := self._pool.get(topic)):	1✔
88	if len(self._threads) < self._max_workers:	1✔
89	self._add_worker()	1✔
90
91	# we cycle through the possible indexes for a work queue, in order to distribute the load across
92	# once we get to the max amount of worker, the cycle will start back at 0
93	index = next(self._cycle)	1✔
94	work_queue = self._work_queues[index]	1✔
95
96	# TODO: the pool is not cleaned up at the moment, think about the clean-up interface
97	self._pool[topic] = work_queue	1✔
98	return work_queue	1✔
99
100	def submit(self, fn, topic, /, args, *kwargs) -> None:	1✔
101	with self._lock:	1✔
102	work_queue = self._get_work_queue(topic)	1✔
103
104	if self._shutdown:	1✔
UNCOV 105	raise RuntimeError("cannot schedule new futures after shutdown")	×
106
107	w = _WorkItem(fn, args, kwargs)	1✔
108	work_queue.put(w)	1✔
109
110	def shutdown(self, wait=True):	1✔
111	with self._lock:	1✔
112	self._shutdown = True	1✔
113
114	# Send a wake-up to prevent threads calling
115	# _work_queue.get(block=True) from permanently blocking.
116	for work_queue in self._work_queues:	1✔
117	work_queue.put(None)	1✔
118
119	if wait:	1✔
UNCOV 120	for t in self._threads:	×
UNCOV 121	t.join()	×

localstack / localstack / 16981563750

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous