// It returns true if the result is sent else false if returns when doneCh is closed.
	send := func(result acceptResult) bool {
		select {
		case listener.acceptCh <- result:
			// Successfully written to acceptCh
			return true
		case <-listener.ctx.Done():
			return false
		}
	}

	// Closure to handle TCPListener until done channel is closed.
	handleListener := func(idx int, listener net.Listener) {
		for {

				xfer := make([]expireObjInfo, len(toDel))
				copy(xfer, toDel)

				var done bool
				select {
				case <-ctx.Done():
					done = true
				case expireCh <- xfer:
					toDel = toDel[:0] // resetting toDel
				}
				if done {
					break
				}
			}
			var match BatchJobExpireFilter
			var found bool
			for _, rule := range r.Rules {

			entry := lastN[(cnt+i)%last]
			if (entry == log.Info{}) {
				continue
			}
			select {
			case subCh <- entry:
			case <-doneCh:
				return nil
			}
		}
	}
	return sys.pubsub.Subscribe(madmin.LogMaskAll, subCh, doneCh, filter)
}

// Init if HTTPConsoleLoggerSys is valid, always returns nil right now
func (sys *HTTPConsoleLoggerSys) Init(_ context.Context) error {
	return nil

	}
	go func() {
		<-p.ctx.Done()
		for i := 0; i < LargeWorkerCount; i++ {
			xioutil.SafeClose(p.lrgworkers[i])
		}
	}()
}

// AddLargeWorker adds a replication worker to the static pool for large uploads.
func (p *ReplicationPool) AddLargeWorker(input <-chan ReplicationWorkerOperation) {
	for {
		select {
		case <-p.ctx.Done():
			return
		case oi, ok := <-input:

// Listen - listen on peers.
func (client *peerRESTClient) Listen(ctx context.Context, listenCh chan<- []byte, v url.Values) {
	go func() {
		for {
			client.doListen(ctx, listenCh, v)
			select {
			case <-ctx.Done():
				return
			default:
				// There was error in the REST request, retry after sometime as probably the peer is down.
				time.Sleep(5 * time.Second)
			}
		}
	}()
}

func (sys *BucketQuotaSys) Init(objAPI ObjectLayer) {
	bucketStorageCache.InitOnce(10*time.Second,
		cachevalue.Opts{ReturnLastGood: true, NoWait: true},
		func() (DataUsageInfo, error) {
			ctx, done := context.WithTimeout(context.Background(), 2*time.Second)
			defer done()

			return loadDataUsageFromBackend(ctx, objAPI)
		},
	)
}

// GetBucketUsageInfo return bucket usage info for a given bucket

				if h.hasEnded() && h.endTime.Add(keepHealSeqStateDuration).Before(now) {
					delete(ahs.healSeqMap, path)
				}
			}
			ahs.Unlock()

			periodicTimer.Reset(time.Minute * 5)
		case <-ctx.Done():
			// server could be restarting - need
			// to exit immediately
			return
		}
	}
}

// getHealSequenceByToken - Retrieve a heal sequence by token. The second

	}

	done := keepHTTPResponseAlive(w)

	result, err := selfSpeedTest(r.Context(), speedTestOpts{
		objectSize:   size,
		concurrency:  concurrent,
		duration:     duration,
		storageClass: storageClass,
		bucketName:   bucketName,
		enableSha256: enableSha256,
	})
	if err != nil {
		result.Error = err.Error()
	}

	done(nil)

					return
				}
			case <-ctx.Done():
				c.connChange.L.Unlock()
				return
			}
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return context.Cause(ctx)
		case newState := <-changed:
			if newState == StateConnected {
				return nil
			}
		}
	}
}

/*
var ErrDone = errors.New("done for now")

	if r == nil {
		return
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-r.wTimer.C:
			r.wlock.Lock()
			r.workers.update()
			r.wlock.Unlock()

		}
	}
}

func (r *ReplicationStats) collectQueueMetrics(ctx context.Context) {
	if r == nil {
		return
	}

	for {
		select {
		case <-ctx.Done():
			return
		case <-r.qTimer.C:
			r.qCache.update()
		}