// Drain will remove interest but continue callbacks until all messages
// have been processed.
197188
func (s *Subscription) Drain() error {
	if s == nil {
		return ErrBadSubscription
	}
	s.mu.Lock()
	conn := s.conn
	s.mu.Unlock()
	if conn == nil {
		return ErrBadSubscription
	}
	return conn.unsubscribe(s, 0, true)
}
1342587
func (n *NeutrinoNotifier) notificationDispatcher() {
	defer n.wg.Done()
out:
	for {
		select {
		case cancelMsg := <-n.notificationCancels:
			switch msg := cancelMsg.(type) {
			case *epochCancel:
				chainntnfs.Log.Infof("Cancelling epoch "+
					"notification, epoch_id=%v", msg.epochID)

				// First, we'll lookup the original
				// registration in order to stop the active
				// queue goroutine.
				reg := n.blockEpochClients[msg.epochID]
				reg.epochQueue.Stop()

				// Next, close the cancel channel for this
				// specific client, and wait for the client to
				// exit.
				close(n.blockEpochClients[msg.epochID].cancelChan)
				n.blockEpochClients[msg.epochID].wg.Wait()

				// Once the client has exited, we can then
				// safely close the channel used to send epoch
				// notifications, in order to notify any
				// listeners that the intent has been
				// cancelled.
				close(n.blockEpochClients[msg.epochID].epochChan)
				delete(n.blockEpochClients, msg.epochID)
			}

		case registerMsg := <-n.notificationRegistry:
			switch msg := registerMsg.(type) {
			case *chainntnfs.HistoricalConfDispatch:
				// We'll start a historical rescan chain of the
				// chain asynchronously to prevent blocking
				// potentially long rescans.
				n.wg.Add(1)
				go func() {
					defer n.wg.Done()

					confDetails, err := n.historicalConfDetails(
						msg.ConfRequest,
						msg.StartHeight, msg.EndHeight,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
						return
					}

					// If the historical dispatch finished
					// without error, we will invoke
					// UpdateConfDetails even if none were
					// found. This allows the notifier to
					// begin safely updating the height hint
					// cache at tip, since any pending
					// rescans have now completed.
					err = n.txNotifier.UpdateConfDetails(
						msg.ConfRequest, confDetails,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
					}
				}()

			case *blockEpochRegistration:
				chainntnfs.Log.Infof("New block epoch subscription")

				n.blockEpochClients[msg.epochID] = msg

				// If the client did not provide their best
				// known block, then we'll immediately dispatch
				// a notification for the current tip.
				if msg.bestBlock == nil {
					n.notifyBlockEpochClient(
						msg, n.bestBlock.Height,
						n.bestBlock.Hash,
					)

					msg.errorChan <- nil
					continue
				}

				// Otherwise, we'll attempt to deliver the
				// backlog of notifications from their best
				// known block.
				n.bestBlockMtx.Lock()
				bestHeight := n.bestBlock.Height
				n.bestBlockMtx.Unlock()

				missedBlocks, err := chainntnfs.GetClientMissedBlocks(
					n.chainConn, msg.bestBlock, bestHeight,
					false,
				)
				if err != nil {
					msg.errorChan <- err
					continue
				}

				for _, block := range missedBlocks {
					n.notifyBlockEpochClient(
						msg, block.Height, block.Hash,
					)
				}

				msg.errorChan <- nil

			case *rescanFilterUpdate:
				err := n.chainView.Update(msg.updateOptions...)
				if err != nil {
					chainntnfs.Log.Errorf("Unable to "+
						"update rescan filter: %v", err)
				}
				msg.errChan <- err
			}

		case item := <-n.chainUpdates.ChanOut():
			update := item.(*filteredBlock)
			if update.connect {
				n.bestBlockMtx.Lock()
				// Since neutrino has no way of knowing what
				// height to rewind to in the case of a reorged
				// best known height, there is no point in
				// checking that the previous hash matches the
				// the hash from our best known height the way
				// the other notifiers do when they receive
				// a new connected block. Therefore, we just
				// compare the heights.
				if update.height != uint32(n.bestBlock.Height+1) {
					// Handle the case where the notifier
					// missed some blocks from its chain
					// backend
					chainntnfs.Log.Infof("Missed blocks, " +
						"attempting to catch up")

					_, missedBlocks, err :=
						chainntnfs.HandleMissedBlocks(
							n.chainConn,
							n.txNotifier,
							n.bestBlock,
							int32(update.height),
							false,
						)
					if err != nil {
						chainntnfs.Log.Error(err)
						n.bestBlockMtx.Unlock()
						continue
					}

					for _, block := range missedBlocks {
						filteredBlock, err :=
							n.getFilteredBlock(block)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
						err = n.handleBlockConnected(filteredBlock)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
					}

				}

				err := n.handleBlockConnected(update)
				if err != nil {
					chainntnfs.Log.Error(err)
				}

				n.bestBlockMtx.Unlock()
				continue
			}

			n.bestBlockMtx.Lock()
			if update.height != uint32(n.bestBlock.Height) {
				chainntnfs.Log.Infof("Missed disconnected " +
					"blocks, attempting to catch up")
			}
			newBestBlock, err := chainntnfs.RewindChain(
				n.chainConn, n.txNotifier, n.bestBlock,
				int32(update.height-1),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to rewind chain "+
					"from height %d to height %d: %v",
					n.bestBlock.Height, update.height-1, err)
			}

			// Set the bestHeight here in case a chain rewind
			// partially completed.
			n.bestBlock = newBestBlock
			n.bestBlockMtx.Unlock()

		case txUpdate := <-n.txUpdates.ChanOut():
			// A new relevant transaction notification has been
			// received from the backend. We'll attempt to process
			// it to determine if it fulfills any outstanding
			// confirmation and/or spend requests and dispatch
			// notifications for them.
			update := txUpdate.(*relevantTx)
			err := n.txNotifier.ProcessRelevantSpendTx(
				update.tx, uint32(update.details.Height),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to process "+
					"transaction %v: %v", update.tx.Hash(),
					err)
			}

		case err := <-n.rescanErr:
			chainntnfs.Log.Errorf("Error during rescan: %v", err)

		case <-n.quit:
			return

		}
	}
}
622863
func (p *Processor) Callback(w http.ResponseWriter, r *http.Request) {
	p.RLock()
	defer p.RUnlock()

	// Validate the JWT token through the handler.
	token, originURL, status, err := p.userTokenHandler.Callback(r)
	if err != nil {
		http.Error(w, fmt.Sprintf("Invalid code %s:", err), http.StatusInternalServerError)
		return
	}

	cookie := &http.Cookie{
		Name:     "X-APORETO-AUTH",
		Value:    token,
		HttpOnly: true,
		Secure:   true,
		Path:     "/",
	}

	http.SetCookie(w, cookie)

	// We transmit the information in the return payload for applications
	// that choose to use it directly without a cookie.
	data, err := json.MarshalIndent(cookie, " ", " ")
	if err != nil {
		http.Error(w, "Bad data", http.StatusInternalServerError)
		return
	}

	// We redirect here to the original URL that the application attempted
	// to access.
	w.Header().Add("Location", originURL)
	http.Error(w, string(data), status)
}
245279
func (c *Conn) drainResults() error {
	for {
		data, err := c.readEphemeralPacket()
		if err != nil {
			return NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
		}
		if isEOFPacket(data) {
			c.recycleReadPacket()
			return nil
		} else if isErrorPacket(data) {
			defer c.recycleReadPacket()
			return ParseErrorPacket(data)
		}
		c.recycleReadPacket()
	}
}
1122081
func (gc *gossipChannel) LeaveChannel() {
	gc.Lock()
	defer gc.Unlock()

	atomic.StoreInt32(&gc.leftChannel, 1)

	var chaincodes []*proto.Chaincode
	var height uint64
	if prevMsg := gc.stateInfoMsg; prevMsg != nil {
		chaincodes = prevMsg.GetStateInfo().Properties.Chaincodes
		height = prevMsg.GetStateInfo().Properties.LedgerHeight
	}
	gc.updateProperties(height, chaincodes, true)
}
118888
func (nc *Conn) drainConnection() {
	// Snapshot subs list.
	nc.mu.Lock()
	subs := make([]*Subscription, 0, len(nc.subs))
	for _, s := range nc.subs {
		subs = append(subs, s)
	}
	errCB := nc.Opts.AsyncErrorCB
	drainWait := nc.Opts.DrainTimeout
	nc.mu.Unlock()

	// for pushing errors with context.
	pushErr := func(err error) {
		nc.mu.Lock()
		nc.err = err
		if errCB != nil {
			nc.ach.push(func() { errCB(nc, nil, err) })
		}
		nc.mu.Unlock()
	}

	// Do subs first
	for _, s := range subs {
		if err := s.Drain(); err != nil {
			// We will notify about these but continue.
			pushErr(err)
		}
	}

	// Wait for the subscriptions to drop to zero.
	timeout := time.Now().Add(drainWait)
	for time.Now().Before(timeout) {
		if nc.NumSubscriptions() == 0 {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}

	// Check if we timed out.
	if nc.NumSubscriptions() != 0 {
		pushErr(ErrDrainTimeout)
	}

	// Flip State
	nc.mu.Lock()
	nc.status = DRAINING_PUBS
	nc.mu.Unlock()

	// Do publish drain via Flush() call.
	err := nc.Flush()
	if err != nil {
		pushErr(err)
		nc.Close()
		return
	}

	// Move to closed state.
	nc.Close()
}
275390
func (r *runtimeSample) drain(stats Stats) {
	stats.Gauge("runtime.NumGoroutine", float32(r.NumGoroutine), 1.0, nil)
	stats.Gauge("runtime.MemStats.Alloc", float32(r.MemStats.Alloc), 1.0, nil)
	stats.Gauge("runtime.MemStats.Frees", float32(r.MemStats.Frees), 1.0, nil)
	stats.Gauge("runtime.MemStats.HeapAlloc", float32(r.MemStats.HeapAlloc), 1.0, nil)
	stats.Gauge("runtime.MemStats.HeapIdle", float32(r.MemStats.HeapIdle), 1.0, nil)
	stats.Gauge("runtime.MemStats.HeapObjects", float32(r.MemStats.HeapObjects), 1.0, nil)
	stats.Gauge("runtime.MemStats.HeapReleased", float32(r.MemStats.HeapReleased), 1.0, nil)
	stats.Gauge("runtime.MemStats.HeapSys", float32(r.MemStats.HeapSys), 1.0, nil)
	stats.Gauge("runtime.MemStats.LastGC", float32(r.MemStats.LastGC), 1.0, nil)
	stats.Gauge("runtime.MemStats.Lookups", float32(r.MemStats.Lookups), 1.0, nil)
	stats.Gauge("runtime.MemStats.Mallocs", float32(r.MemStats.Mallocs), 1.0, nil)
	stats.Gauge("runtime.MemStats.MCacheInuse", float32(r.MemStats.MCacheInuse), 1.0, nil)
	stats.Gauge("runtime.MemStats.MCacheSys", float32(r.MemStats.MCacheSys), 1.0, nil)
	stats.Gauge("runtime.MemStats.MSpanInuse", float32(r.MemStats.MSpanInuse), 1.0, nil)
	stats.Gauge("runtime.MemStats.MSpanSys", float32(r.MemStats.MSpanSys), 1.0, nil)
	stats.Gauge("runtime.MemStats.NextGC", float32(r.MemStats.NextGC), 1.0, nil)
	stats.Gauge("runtime.MemStats.NumGC", float32(r.MemStats.NumGC), 1.0, nil)
	stats.Gauge("runtime.MemStats.StackInuse", float32(r.MemStats.StackInuse), 1.0, nil)
}
141101
func (q *Queue) CallbackWatch(matcher events.Matcher) (eventq chan events.Event, cancel func()) {
	chanSink, ch := q.sinkGen.NewChannelSink()
	lq := queue.NewLimitQueue(chanSink, q.limit)
	sink := events.Sink(lq)

	if matcher != nil {
		sink = events.NewFilter(sink, matcher)
	}

	q.broadcast.Add(sink)

	cancelFunc := func() {
		q.broadcast.Remove(sink)
		ch.Close()
		sink.Close()
	}

	externalCancelFunc := func() {
		q.mu.Lock()
		cancelFunc := q.cancelFuncs[sink]
		delete(q.cancelFuncs, sink)
		q.mu.Unlock()

		if cancelFunc != nil {
			cancelFunc()
		}
	}

	q.mu.Lock()
	q.cancelFuncs[sink] = cancelFunc
	q.mu.Unlock()

	// If the output channel shouldn't be closed and the queue is limitless,
	// there's no need for an additional goroutine.
	if !q.closeOutChan && q.limit == 0 {
		return ch.C, externalCancelFunc
	}

	outChan := make(chan events.Event)
	go func() {
		for {
			select {
			case <-ch.Done():
				// Close the output channel if the ChannelSink is Done for any
				// reason. This can happen if the cancelFunc is called
				// externally or if it has been closed by a wrapper sink, such
				// as the TimeoutSink.
				if q.closeOutChan {
					close(outChan)
				}
				externalCancelFunc()
				return
			case <-lq.Full():
				// Close the output channel and tear down the Queue if the
				// LimitQueue becomes full.
				if q.closeOutChan {
					close(outChan)
				}
				externalCancelFunc()
				return
			case event := <-ch.C:
				outChan <- event
			}
		}
	}()

	return outChan, externalCancelFunc
}
1056015
func (c *ContainerCache) Drain(newSize int) {
	c.Lock()
	defer c.Unlock()
	// Make a new one, leave the old one for GC
	c.cache = make(map[string]*docker.Container, newSize)
}
419292
func (c *chatLocal) ClearCache() {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.selfConvInfos = nil
}
1289178
func (tp *BytePool) Close() {
	tp.Drain()
	if tp.drainTicker != nil {
		tp.drainTicker.Stop()
		tp.drainTicker = nil
	}
}
631940
func (log *logCache) Flush(writer io.Writer) {
	log.flushing.Lock()
	defer log.flushing.Unlock()

	for _, message := range log.messages {
		writer.Write(message)
	}
	log.messages = [][]byte{}
}
321008
func (n *NodeDrainer) Update(node *structs.Node) {
	n.l.Lock()
	defer n.l.Unlock()

	if node == nil {
		return
	}

	draining, ok := n.nodes[node.ID]
	if !ok {
		draining = NewDrainingNode(node, n.state)
		n.nodes[node.ID] = draining
	} else {
		// Update it
		draining.Update(node)
	}

	// TODO test the notifier is updated
	if inf, deadline := node.DrainStrategy.DeadlineTime(); !inf {
		n.deadlineNotifier.Watch(node.ID, deadline)
	} else {
		// There is an infinite deadline so it shouldn't be tracked for
		// deadlining
		n.deadlineNotifier.Remove(node.ID)
	}

	// TODO Test this
	// Register interest in the draining jobs.
	jobs, err := draining.DrainingJobs()
	if err != nil {
		n.logger.Error("error retrieving draining jobs on node", "node_id", node.ID, "error", err)
		return
	}
	n.logger.Trace("node has draining jobs on it", "node_id", node.ID, "num_jobs", len(jobs))
	n.jobWatcher.RegisterJobs(jobs)

	// TODO Test at this layer as well that a node drain on a node without
	// allocs immediately gets unmarked as draining
	// Check if the node is done such that if an operator drains a node with
	// nothing on it we unset drain
	done, err := draining.IsDone()
	if err != nil {
		n.logger.Error("failed to check if node is done draining", "node_id", node.ID, "error", err)
		return
	}

	if done {
		// Node is done draining. Stop remaining system allocs before
		// marking node as complete.
		remaining, err := draining.RemainingAllocs()
		if err != nil {
			n.logger.Error("error getting remaining allocs on drained node", "node_id", node.ID, "error", err)
		} else if len(remaining) > 0 {
			future := structs.NewBatchFuture()
			n.drainAllocs(future, remaining)
			if err := future.Wait(); err != nil {
				n.logger.Error("failed to drain remaining allocs from done node", "num_allocs", len(remaining), "node_id", node.ID, "error", err)
			}
		}

		// Create the node event
		event := structs.NewNodeEvent().
			SetSubsystem(structs.NodeEventSubsystemDrain).
			SetMessage(NodeDrainEventComplete)

		index, err := n.raft.NodesDrainComplete([]string{node.ID}, event)
		if err != nil {
			n.logger.Error("failed to unset drain for node", "node_id", node.ID, "error", err)
		} else {
			n.logger.Info("node completed draining at index", "node_id", node.ID, "index", index)
		}
	}
}
829191
func (n *NodeDrainer) drainAllocs(future *structs.BatchFuture, allocs []*structs.Allocation) {
	// Compute the effected jobs and make the transition map
	jobs := make(map[string]*structs.Allocation, 4)
	transitions := make(map[string]*structs.DesiredTransition, len(allocs))
	for _, alloc := range allocs {
		transitions[alloc.ID] = &structs.DesiredTransition{
			Migrate: helper.BoolToPtr(true),
		}
		jobs[alloc.JobID] = alloc
	}

	evals := make([]*structs.Evaluation, 0, len(jobs))
	for job, alloc := range jobs {
		evals = append(evals, &structs.Evaluation{
			ID:          uuid.Generate(),
			Namespace:   alloc.Namespace,
			Priority:    alloc.Job.Priority,
			Type:        alloc.Job.Type,
			TriggeredBy: structs.EvalTriggerNodeDrain,
			JobID:       job,
			Status:      structs.EvalStatusPending,
		})
	}

	// Commit this update via Raft
	var finalIndex uint64
	for _, u := range partitionAllocDrain(defaultMaxIdsPerTxn, transitions, evals) {
		index, err := n.raft.AllocUpdateDesiredTransition(u.Transitions, u.Evals)
		if err != nil {
			future.Respond(0, err)
			return
		}
		finalIndex = index
	}

	future.Respond(finalIndex, nil)
}
1165834
func (ch *Chunk) Drain() {
	bytesAhead := ch.Size - ch.Pos
	for bytesAhead > 0 {
		readSize := int64(bytesAhead)

		if _, err := io.CopyN(ioutil.Discard, ch.R, readSize); err != nil {
			return
		}
		bytesAhead -= int(readSize)
	}
}
56820
func (f *StringForwarder) invokeCallback(callback func(string), msg string) {
	f.mu.Unlock()
	defer f.mu.Lock()
	callback(msg)
}
1001116
func (p *AuroraAdminClient) DrainHosts(ctx context.Context, hosts *Hosts) (r *Response, err error) {
  var _args336 AuroraAdminDrainHostsArgs
  _args336.Hosts = hosts
  var _result337 AuroraAdminDrainHostsResult
  if err = p.Client_().Call(ctx, "drainHosts", &_args336, &_result337); err != nil {
    return
  }
  return _result337.GetSuccess(), nil
}