Repository: redis/redis-py
File: redis/maint_notifications.py (Python)

Code before:
            )
            # mark for reconnect all in use connections to the node - this will force them to
            # disconnect after they complete their current commands
            # Some of them might be used by sub sub and we don't know which ones - so we disconnect
            # all in flight connections after they are done with current command execution
            for conn in (
                current_node.redis_connection.connection_pool._get_in_use_connections()
            ):
                conn.mark_for_reconnect()

            if (
                current_node
                not in self.cluster_client.nodes_manager.nodes_cache.values()
            ):
                # disconnect all free connections to the node - this node will be dropped
                # from the cluster, so we don't need to revert the timeouts
                for conn in current_node.redis_connection.connection_pool._get_free_connections():
                    conn.disconnect()
            else:
                if self.config.is_relaxed_timeouts_enabled():
                    # reset the timeouts for the node to which the connection is connected
                    # Perform check if other maintenance ops are in progress for the same node
                    # and if so, don't reset the timeouts and wait for the last maintenance
                    # to complete
                    for conn in (
                        *current_node.redis_connection.connection_pool._get_in_use_connections(),
                        *current_node.redis_connection.connection_pool._get_free_connections(),
                    ):
                        if (
                            len(conn.get_processed_start_notifications())
                            > len(conn.get_skipped_end_notifications()) + 1
                        ):
                            # we have received more start notifications than end notifications
                            # for this connection - we should not reset the timeouts
                            # and add the notification id to the set of skipped end notifications
                            conn.add_skipped_end_notification(notification.id)
                        else:
                            conn.reset_tmp_settings(reset_relaxed_timeout=True)
                            conn.update_current_socket_timeout(relaxed_timeout=-1)
                            conn.maintenance_state = MaintenanceState.NONE
                            conn.reset_received_notifications()

            # mark the notification as processed
            self._processed_notifications.add(notification)
            self._in_progress.remove(notification)


Review comment (suggestion):
I think we need to somehow control access to connections, in multi-threaded environment in use and free connections are frequently accessed and could be modified while we're loop over them

Code after:
            )
            with current_node.redis_connection.connection_pool._lock:
                # mark for reconnect all in use connections to the node - this will force them to
                # disconnect after they complete their current commands
                # Some of them might be used by sub sub and we don't know which ones - so we disconnect
                # all in flight connections after they are done with current command execution
                for conn in current_node.redis_connection.connection_pool._get_in_use_connections():
                    conn.mark_for_reconnect()

                if (
                    current_node
                    not in self.cluster_client.nodes_manager.nodes_cache.values()
                ):
                    # disconnect all free connections to the node - this node will be dropped
                    # from the cluster, so we don't need to revert the timeouts
                    for conn in current_node.redis_connection.connection_pool._get_free_connections():
                        conn.disconnect()
                else:
                    if self.config.is_relaxed_timeouts_enabled():
                        # reset the timeouts for the node to which the connection is connected
                        # Perform check if other maintenance ops are in progress for the same node
                        # and if so, don't reset the timeouts and wait for the last maintenance
                        # to complete
                        for conn in (
                            *current_node.redis_connection.connection_pool._get_in_use_connections(),
                            *current_node.redis_connection.connection_pool._get_free_connections(),
                        ):
                            if (
                                len(conn.get_processed_start_notifications())
                                > len(conn.get_skipped_end_notifications()) + 1
                            ):
                                # we have received more start notifications than end notifications
                                # for this connection - we should not reset the timeouts
                                # and add the notification id to the set of skipped end notifications
                                conn.add_skipped_end_notification(notification.id)
                            else:
                                conn.reset_tmp_settings(reset_relaxed_timeout=True)
                                conn.update_current_socket_timeout(relaxed_timeout=-1)
                                conn.maintenance_state = MaintenanceState.NONE
                                conn.reset_received_notifications()

            # mark the notification as processed
            self._processed_notifications.add(notification)
            self._in_progress.remove(notification)