// CHECK-DAG: %[[CONST_1:.*]] = stablehlo.constant dense<{{.*}}> : tensor<8xf32>
// CHECK: %[[BROADCAST_0:.*]] = stablehlo.broadcast_in_dim %[[CONST_0]], dims = [3] : (tensor<8xf32>) -> tensor<1x1x2x8xf32>
// CHECK: %[[MUL:.*]] = stablehlo.multiply %[[ARG]], %[[BROADCAST_0]] : tensor<1x1x2x8xf32>
// CHECK: %[[BROADCAST_1:.*]] = stablehlo.broadcast_in_dim %[[CONST_1]], dims = [3] : (tensor<8xf32>) -> tensor<1x1x2x8xf32>

    }

    def "can remove a listener which tries to notify another broadcaster"() {
        given:
        def listener1 = {
            sleep 1000
            // Try to get broadcaster, should not block
            def broadcaster = manager.getBroadcaster(TestBazListener)
            // Notify broadcaster, should not block
            broadcaster.baz()
        } as TestFooListener
        def listener2 = {

        synchronized (lock) {
            EventBroadcast<T> broadcaster = Cast.uncheckedCast(broadcasters.get(listenerClass));
            if (broadcaster == null) {
                broadcaster = new EventBroadcast<T>(listenerClass);
                broadcasters.put(listenerClass, broadcaster);
                for (ListenerDetails listener : allListeners.values()) {
                    broadcaster.maybeAdd(listener);
                }

            if (broadcaster != null) {
                broadcaster.add(listener);
            }
        }

        void addListener(Dispatch<MethodInvocation> listener) {
            subscribedListeners.add(listener);
            if (broadcaster != null) {
                broadcaster.add(listener);
            }
        }

        void removeListener(Object listener) {
            subscribedListeners.remove(listener);

  if (!applyPatterns(func, target, stage1FrozenPatterns)) {
    return signalPassFailure();
  }
  // Explict BroadcastTo addition for left-over broadcast-able ops.
  // The following pattern matchings should be done after the other legalization
  // rules in order not to add unnecessary BroadcastTo ops.
  RewritePatternSet stage2Patterns(&getContext());

  addPatterns(context, stage2Patterns, this->preserve_assert_op_);

	// Start events processing pipeline.
	tc.broadcaster.StartStructuredLogging(3)
	if tc.client != nil {
		logger.Info("Sending events to api server")
		tc.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: tc.client.CoreV1().Events("")})
	} else {
		logger.Error(nil, "kubeClient is nil", "controller", tc.name)
		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
	}
	defer tc.broadcaster.Shutdown()

  func.return %3 : tensor<1x10xf32>
}

// -----

// Converts tfl.broadcast_to to tfl.reshape if input and output have the same
// number of elements.
// CHECK-LABEL: broadcast_to_to_reshape
func.func @broadcast_to_to_reshape(%arg0: tensor<4x4x4xf32>, %arg1 : tensor<4xi32>) -> tensor<1x4x4x4xf32> {
  %0 = "tfl.broadcast_to"(%arg0, %arg1) : (tensor<4x4x4xf32>, tensor<4xi32>) -> tensor<1x4x4x4xf32>
  // CHECK: "tfl.reshape"

		portRange:     portRange,
		alloc:         alloc,
		leaks:         map[int]int{},
		broadcaster:   eventBroadcaster,
		recorder:      recorder,
	}
}

// RunUntil starts the controller until the provided ch is closed.
func (c *Repair) RunUntil(onFirstSuccess func(), stopCh chan struct{}) {
	c.broadcaster.StartRecordingToSink(stopCh)
	defer c.broadcaster.Shutdown()

	var once sync.Once
	wait.Until(func() {

		allocatorByFamily: allocatorByFamily,

		leaksByFamily: leaksByFamily,
		broadcaster:   eventBroadcaster,
		recorder:      recorder,
	}
}

// RunUntil starts the controller until the provided ch is closed.
func (c *Repair) RunUntil(onFirstSuccess func(), stopCh chan struct{}) {
	c.broadcaster.StartRecordingToSink(stopCh)
	defer c.broadcaster.Shutdown()

	var once sync.Once
	wait.Until(func() {

	defer utilruntime.HandleCrash()

	// Start event processing pipeline.
	r.broadcaster.StartStructuredLogging(3)
	logger := klog.FromContext(ctx)
	logger.Info("Sending events to api server")
	r.broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: r.client.CoreV1().Events("")})
	defer r.broadcaster.Shutdown()

	defer r.queue.ShutDown()

	logger.Info("Starting range CIDR allocator")