// pingAndPurgeIfNeeded pings other by publishing to others ping key. If it
// does not receive a pong reply within some amount of time, it will
// assume the pool is stale and purge it.
974471
func (p *Pool) pingAndPurgeIfNeeded(other *Pool) error {
	ping := redisPool.Get()
	pong := redis.PubSubConn{redisPool.Get()}
	// Listen for pongs by subscribing to the other pool's pong key
	pong.Subscribe(other.pongKey())
	// Ping the other pool by publishing to its ping key
	ping.Do("PUBLISH", other.pingKey(), 1)
	// Use a select statement to either receive the pong or timeout
	pongChan := make(chan interface{})
	errChan := make(chan error)
	go func() {
		defer func() {
			pong.Close()
			ping.Close()
		}()
		select {
		case <-p.exit:
			return
		default:
		}
		for {
			reply := pong.Receive()
			switch reply.(type) {
			case redis.Message:
				// The pong was received
				pongChan <- reply
				return
			case error:
				// There was some unexpected error
				err := reply.(error)
				errChan <- err
				return
			}
		}
	}()
	timeout := time.After(p.config.StaleTimeout)
	select {
	case <-pongChan:
		// The other pool responded with a pong
		return nil
	case err := <-errChan:
		// Received an error from the pubsub conn
		return err
	case <-timeout:
		// The pool is considered stale and should be purged
		t := newTransaction()
		other.RLock()
		otherId := other.id
		other.RUnlock()
		t.purgeStalePool(otherId)
		if err := t.exec(); err != nil {
			return err
		}
	}
	return nil
}
998979
func (nc *Conn) clearPendingFlushCalls() {
	// Clear any queued pongs, e.g. pending flush calls.
	for _, ch := range nc.pongs {
		if ch != nil {
			close(ch)
		}
	}
	nc.pongs = nil
}
1292217
func IsIngressNetworkNeeded(s *api.Service) bool {
	if s == nil {
		return false
	}

	if s.Spec.Endpoint == nil {
		return false
	}

	for _, p := range s.Spec.Endpoint.Ports {
		// The service to which this task belongs is trying to
		// expose ports with PublishMode as Ingress to the
		// external world. Automatically attach the task to
		// the ingress network.
		if p.PublishMode == api.PublishModeIngress {
			return true
		}
	}

	return false
}
200333
func (w *Producer) DeferredPublish(topic string, delay time.Duration, body []byte) error {
	return w.sendCommand(DeferredPublish(topic, delay, body))
}
988025
func (c *client) sendPong() {
	c.traceOutOp("PONG", nil)
	c.sendProto([]byte("PONG\r\n"), true)
}
1271379
func (s *Storage) EphemeralPurge(ctx context.Context, convID chat1.ConversationID, uid gregor1.UID, purgeInfo *chat1.EphemeralPurgeInfo) (newPurgeInfo *chat1.EphemeralPurgeInfo, explodedMsgs []chat1.MessageUnboxed, err Error) {
	defer s.Trace(ctx, func() error { return err }, "EphemeralPurge")()
	lock := locks.StorageLockTab.AcquireOnName(ctx, s.G(), convID.String())
	defer lock.Release(ctx)

	if purgeInfo == nil {
		return nil, nil, nil
	}

	// Fetch secret key
	key, ierr := GetSecretBoxKey(ctx, s.G().ExternalG(), DefaultSecretUI)
	if ierr != nil {
		return nil, nil, MiscError{Msg: "unable to get secret key: " + ierr.Error()}
	}

	ctx, err = s.engine.Init(ctx, key, convID, uid)
	if err != nil {
		return nil, nil, err
	}

	maxMsgID, err := s.idtracker.getMaxMessageID(ctx, convID, uid)
	if err != nil {
		return nil, nil, err
	}

	// We don't care about holes.
	maxHoles := int(maxMsgID-purgeInfo.MinUnexplodedID) + 1
	var target int
	if purgeInfo.MinUnexplodedID == 0 {
		target = 0 // we need to traverse the whole conversation
	} else {
		target = maxHoles
	}
	rc := NewHoleyResultCollector(maxHoles, NewSimpleResultCollector(target))
	err = s.engine.ReadMessages(ctx, rc, convID, uid, maxMsgID, 0)
	switch err.(type) {
	case nil:
		// ok
		if len(rc.Result()) == 0 {
			err := s.ephemeralTracker.inactivatePurgeInfo(ctx, convID, uid)
			return nil, nil, err
		}
	case MissError:
		s.Debug(ctx, "record-only ephemeralTracker: no local messages")
		// We don't have these messages in cache, so don't retry this
		// conversation until further notice.
		err := s.ephemeralTracker.inactivatePurgeInfo(ctx, convID, uid)
		return nil, nil, err
	default:
		return nil, nil, err
	}
	newPurgeInfo, explodedMsgs, err = s.ephemeralPurgeHelper(ctx, convID, uid, rc.Result())
	if err != nil {
		return nil, nil, err
	}
	err = s.ephemeralTracker.setPurgeInfo(ctx, convID, uid, newPurgeInfo)
	return newPurgeInfo, explodedMsgs, err
}
39817
func CleanupLeftovers(ipvs utilipvs.Interface, ipt utiliptables.Interface, ipset utilipset.Interface, cleanupIPVS bool) (encounteredError bool) {
	if cleanupIPVS {
		// Return immediately when ipvs interface is nil - Probably initialization failed in somewhere.
		if ipvs == nil {
			return true
		}
		encounteredError = false
		err := ipvs.Flush()
		if err != nil {
			klog.Errorf("Error flushing IPVS rules: %v", err)
			encounteredError = true
		}
	}
	// Delete dummy interface created by ipvs Proxier.
	nl := NewNetLinkHandle(false)
	err := nl.DeleteDummyDevice(DefaultDummyDevice)
	if err != nil {
		klog.Errorf("Error deleting dummy device %s created by IPVS proxier: %v", DefaultDummyDevice, err)
		encounteredError = true
	}
	// Clear iptables created by ipvs Proxier.
	encounteredError = cleanupIptablesLeftovers(ipt) || encounteredError
	// Destroy ip sets created by ipvs Proxier.  We should call it after cleaning up
	// iptables since we can NOT delete ip set which is still referenced by iptables.
	for _, set := range ipsetInfo {
		err = ipset.DestroySet(set.name)
		if err != nil {
			if !utilipset.IsNotFoundError(err) {
				klog.Errorf("Error removing ipset %s, error: %v", set.name, err)
				encounteredError = true
			}
		}
	}
	return encounteredError
}
790867
func indirectPing(n *Node, target string, amount int, timeout time.Duration) (reached bool, errs []error) {
	resCh := sendPingRequests(n, target, amount, timeout)

	// wait for responses from the ping-reqs
	for result := range resCh {
		switch res := result.(type) {
		case *pingResponse:
			if res.Ok {
				return true, errs
			}
			// If the ping to the target was not-ok we want to wait for more results.

		case error:
			errs = append(errs, res)
		}
	}

	return false, errs
}
1065476
func (c *Client) processPingreq() error {
	// send a pingresp packet
	err := c.send(packet.NewPingresp(), true)
	if err != nil {
		return c.die(TransportError, err)
	}

	return nil
}
953410
func (p *Service) SanitizedAddress() (sanitizedAddress string, err error) {
	var (
		canonicalizedAddress string
	)

	if p.Outputaddress != "" {
		return p.Outputaddress, nil
	}

	if strings.HasPrefix(p.Address, "postgres://") || strings.HasPrefix(p.Address, "postgresql://") {
		if canonicalizedAddress, err = parseURL(p.Address); err != nil {
			return sanitizedAddress, err
		}
	} else {
		canonicalizedAddress = p.Address
	}

	sanitizedAddress = kvMatcher.ReplaceAllString(canonicalizedAddress, "")

	return sanitizedAddress, err
}
233911
func receivePings(pingCh chan ping, pings []ping,
	kindMatch string, kindSeen bool) ([]ping, bool) {
	for {
		select {
		case pingVal := <-pingCh:
			pings = append(pings, pingVal)
			if pingVal.kind == kindMatch {
				kindSeen = true
			}

		default:
			return pings, kindSeen
		}
	}
}
395598
func (_class NetworkClass) GetPurpose(sessionID SessionRef, self NetworkRef) (_retval []NetworkPurpose, _err error) {
	_method := "network.get_purpose"
	_sessionIDArg, _err := convertSessionRefToXen(fmt.Sprintf("%s(%s)", _method, "session_id"), sessionID)
	if _err != nil {
		return
	}
	_selfArg, _err := convertNetworkRefToXen(fmt.Sprintf("%s(%s)", _method, "self"), self)
	if _err != nil {
		return
	}
	_result, _err := _class.client.APICall(_method, _sessionIDArg, _selfArg)
	if _err != nil {
		return
	}
	_retval, _err = convertEnumNetworkPurposeSetToGo(_method + " -> ", _result.Value)
	return
}
1062058
func (mt *MultiplexTransport) Cleanup(p Peer) {
	mt.conns.RemoveAddr(p.RemoteAddr())
	_ = p.CloseConn()
}
1092170
func (mm *messageManager) GeneratePurgeQuery(timeCutoff int64) (string, map[string]*querypb.BindVariable) {
	return mm.purgeQuery.Query, map[string]*querypb.BindVariable{
		"time_scheduled": sqltypes.Int64BindVariable(timeCutoff),
	}
}
1163900
func replyToPings(pings []ping) {
	for _, ping := range pings {
		if ping.pongCh != nil {
			close(ping.pongCh)
			ping.pongCh = nil
		}
	}
}
765714
func (p *Pool) respondToPings() error {
	pong := redisPool.Get()
	ping := redis.PubSubConn{redisPool.Get()}
	defer func() {
		pong.Close()
		ping.Close()
	}()
	// Subscribe to the ping key for this pool to receive pings.
	if err := ping.Subscribe(p.pingKey()); err != nil {
		return err
	}
	for {
		// Whenever we recieve a ping, reply immediately with a pong by
		// publishing to the pong key for this pool.
		switch reply := ping.Receive().(type) {
		case redis.Message:
			if _, err := pong.Do("PUBLISH", p.pongKey(), 0); err != nil {
				return err
			}
		case error:
			err := reply.(error)
			return err
		}
		time.Sleep(1 * time.Millisecond)
	}
}
781676
func (nc *Conn) sendPing(ch chan struct{}) {
	nc.pongs = append(nc.pongs, ch)
	nc.bw.WriteString(pingProto)
	// Flush in place.
	nc.bw.Flush()
}