| # Ptwrite utilities. |
| # Copyright (C) 2023-2025 Free Software Foundation, Inc. |
| |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 3 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| |
| """Utilities for working with ptwrite filters.""" |
| |
| import gdb |
| |
| # _ptwrite_filter contains the per thread copies of the filter function. |
| # The keys are tuples of inferior id and thread id. |
| # The filter functions are created for each thread by calling the |
| # _ptwrite_filter_factory. |
| _ptwrite_filter = {} |
| _ptwrite_filter_factory = None |
| |
| |
| def _ptwrite_exit_handler(event): |
| """Exit handler to prune _ptwrite_filter on thread exit.""" |
| _ptwrite_filter.pop(event.inferior_thread.ptid, None) |
| |
| |
| gdb.events.thread_exited.connect(_ptwrite_exit_handler) |
| |
| |
| def _clear_traces(): |
| """Helper function to clear the trace of all threads.""" |
| current_thread = gdb.selected_thread() |
| |
| for inferior in gdb.inferiors(): |
| for thread in inferior.threads(): |
| thread.switch() |
| recording = gdb.current_recording() |
| if recording is not None: |
| recording.clear() |
| |
| current_thread.switch() |
| |
| |
| def register_filter_factory(filter_factory_): |
| """Register the ptwrite filter factory.""" |
| if filter_factory_ is not None and not callable(filter_factory_): |
| raise TypeError("The filter factory must be callable or 'None'.") |
| |
| # Clear the traces of all threads of all inferiors to force |
| # re-decoding with the new filter. |
| _clear_traces() |
| |
| _ptwrite_filter.clear() |
| global _ptwrite_filter_factory |
| _ptwrite_filter_factory = filter_factory_ |
| |
| |
| def get_filter(): |
| """Returns the filter of the current thread.""" |
| thread = gdb.selected_thread() |
| key = thread.ptid |
| |
| # Create a new filter for new threads. |
| if key not in _ptwrite_filter: |
| if _ptwrite_filter_factory is not None: |
| _ptwrite_filter[key] = _ptwrite_filter_factory(thread) |
| else: |
| return None |
| |
| return _ptwrite_filter[key] |