func blockingSyscall(d time.Duration, done chan<- error) {
	r, w, err := os.Pipe()
	if err != nil {
		done <- err
		return
	}
	start := time.Now()
	msg := []byte("hello")
	time.AfterFunc(d, func() { w.Write(msg) })
	_, err = syscall.Read(int(r.Fd()), make([]byte, len(msg)))
	if err == nil && time.Since(start) < d {
		err = fmt.Errorf("syscall returned too early: want=%s got=%s", d, time.Since(start))
	}

	go func() {
		q.Run(stop)
		signal <- struct{}{}
	}()

	q.Push(func() error {
		t.Log("mock exec")
		return nil
	})

	// mock queue block wait cond signal
	time.AfterFunc(10*time.Millisecond, func() {
		close(stop)
	})

	select {
	case <-time.After(200 * time.Millisecond):
		t.Error("close stop, method exit timeout.")
	case <-signal:
		t.Log("queue return.")
	}
}

func (c *Cluster) Run(mesh mesh.Watcher, handlers []handler, action ACTION) {
	if features.RemoteClusterTimeout > 0 {
		time.AfterFunc(features.RemoteClusterTimeout, func() {
			if !c.initialSync.Load() {
				log.Errorf("remote cluster %s failed to sync after %v", c.ID, features.RemoteClusterTimeout)
				timeouts.With(clusterLabel.Value(string(c.ID))).Increment()

		err := f(ctx, fireAt, args)
		if err != nil {
			logger.Error(err, "TaintEvictionController: timed worker failed")
		}
	}
	if delay <= 0 {
		go fWithErrorLogging()
		return nil
	}
	timer := clock.AfterFunc(delay, fWithErrorLogging)
	return &TimedWorker{
		WorkItem:  args,
		CreatedAt: createdAt,
		FireAt:    fireAt,
		Timer:     timer,
	}
}

// Cancel cancels the execution of function by the `TimedWorker`

	if shuttingdown {
		return false
	}

	// Run the task.
	if err := task.task(); err != nil {
		delay := q.delay
		log.Infof("Work item handle failed (%v), retry after delay %v", err, delay)
		time.AfterFunc(delay, func() {
			q.Push(task.task)
		})
	}
	q.metrics.workDuration.Record(time.Since(task.startTime).Seconds())

	return true
}

func (q *queueImpl) HasSynced() bool {

	a.cancelOnce.Do(func() {
		a.cancelContext()
		_ = a.client.CloseSend()
		if a.conn != nil {
			_ = a.conn.Close()
		}
	})
}

func (a *AdsTest) adsReceiveChannel() {
	context.AfterFunc(a.context, a.Cleanup)
	for {
		resp, err := a.client.Recv()
		if err != nil {
			if isUnexpectedError(err) {
				log.Warnf("ads received error: %v", err)
			}
			select {
			case a.error <- err:

	a.cancelOnce.Do(func() {
		a.cancelContext()
		_ = a.client.CloseSend()
		if a.conn != nil {
			_ = a.conn.Close()
		}
	})
}

func (a *DeltaAdsTest) adsReceiveChannel() {
	context.AfterFunc(a.context, a.Cleanup)
	for {
		resp, err := a.client.Recv()
		if err != nil {
			if isUnexpectedError(err) {
				log.Warnf("ads received error: %v", err)
			}
			select {
			case a.error <- err:

			fd.pfd.SetWriteDeadline(aLongTimeAgo)
		} else {
			if deadline, ok := ctx.Deadline(); ok && !deadline.IsZero() {
				fd.pfd.SetWriteDeadline(deadline)
			}

			done := make(chan struct{})
			stop := context.AfterFunc(ctx, func() {
				// Force the runtime's poller to immediately give
				// up waiting for writability.
				fd.pfd.SetWriteDeadline(aLongTimeAgo)
				close(done)
			})
			defer func() {
				if !stop() {

		}
		return
	}

	// Time in the future, setup a timer to cancel in the future.
	if dur := time.Until(t); dur > 0 {
		if closed {
			d.cancel = make(chan struct{})
		}
		d.timer = time.AfterFunc(dur, func() {
			close(d.cancel)
		})
		return
	}

	// Time in the past, so close immediately.
	if !closed {
		close(d.cancel)
	}
}

func SetVolumeOwnership(mounter Mounter, dir string, fsGroup *int64, fsGroupChangePolicy *v1.PodFSGroupChangePolicy, completeFunc func(types.CompleteFuncParam)) error {
	if fsGroup == nil {
		return nil
	}

	timer := time.AfterFunc(30*time.Second, func() {
		klog.Warningf("Setting volume ownership for %s and fsGroup set. If the volume has a lot of files then setting volume ownership could be slow, see https://github.com/kubernetes/kubernetes/issues/69699", dir)