// extendPending extends all the pending messages
486290
func (veq *visibilityExtenderQueue) extendPending() {
	veq.Lock()
	defer veq.Unlock()

	fails, err := veq.extendBatch(veq.entries)
	if err != nil {
		veq.svc.Logger("Error extending batch: %s", err)
	}

	var retries uint
	for len(fails) > 0 && retries <= 5 {
		retries++
		time.Sleep(time.Duration((2<<retries)*50) * time.Millisecond)
		fails, err = veq.extendBatch(fails)
		if err != nil {
			veq.svc.Logger("Error extending batch: %s", err)
		}
	}
}
524641
func (s *Service) reportPending(conn *websocket.Conn) error {
	// Retrieve the pending count from the local blockchain
	var pending int
	if s.eth != nil {
		pending, _ = s.eth.TxPool().Stats()
	} else {
		pending = s.les.TxPool().Stats()
	}
	// Assemble the transaction stats and send it to the server
	log.Trace("Sending pending transactions to ethstats", "count", pending)

	stats := map[string]interface{}{
		"id": s.node,
		"stats": &pendStats{
			Pending: pending,
		},
	}
	report := map[string][]interface{}{
		"emit": {"pending", stats},
	}
	return websocket.JSON.Send(conn, report)
}
626789
func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) {
	notifier, supported := rpc.NotifierFromContext(ctx)
	if !supported {
		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
	}

	rpcSub := notifier.CreateSubscription()

	go func() {
		txHashes := make(chan []common.Hash, 128)
		pendingTxSub := api.events.SubscribePendingTxs(txHashes)

		for {
			select {
			case hashes := <-txHashes:
				// To keep the original behaviour, send a single tx hash in one notification.
				// TODO(rjl493456442) Send a batch of tx hashes in one notification
				for _, h := range hashes {
					notifier.Notify(rpcSub.ID, h)
				}
			case <-rpcSub.Err():
				pendingTxSub.Unsubscribe()
				return
			case <-notifier.Closed():
				pendingTxSub.Unsubscribe()
				return
			}
		}
	}()

	return rpcSub, nil
}
738403
func (s *ReplicationInstance) SetPendingModifiedValues(v *ReplicationPendingModifiedValues) *ReplicationInstance {
	s.PendingModifiedValues = v
	return s
}
883040
func (ss *FileSubStore) AddSeqPending(subid, seqno uint64) error {
	ss.Lock()
	ss.updateSub.ID, ss.updateSub.Seqno = subid, seqno
	if err := ss.writeRecord(nil, subRecMsg, &ss.updateSub); err != nil {
		ss.Unlock()
		return err
	}
	si := ss.subs[subid]
	if si != nil {
		s := si.(*subscription)
		if seqno > s.sub.LastSent {
			s.sub.LastSent = seqno
		}
		s.seqnos[seqno] = struct{}{}
	}
	ss.Unlock()
	return nil
}
862414
func (ec *EthereumClient) GetPendingCodeAt(ctx *Context, account *Address) (code []byte, _ error) {
	return ec.client.PendingCodeAt(ctx.context, account.address)
}
27613
func (p *FlushByRamOrCountsPolicy) markLargestWriterPending(control *DocumentsWriterFlushControl,
	perThreadState *ThreadState, currentBytesPerThread int64) {
	control.setFlushPending(p.findLargestNonPendingWriter(control, perThreadState))
}
1346347
func (s *Cluster) SetPendingModifiedValues(v *PendingModifiedValues) *Cluster {
	s.PendingModifiedValues = v
	return s
}
590603
func (s *ModifyCurrentDBClusterCapacityOutput) SetPendingCapacity(v int64) *ModifyCurrentDBClusterCapacityOutput {
	s.PendingCapacity = &v
	return s
}
1262345
func (v *vaultClient) extendExpiration(ttlSeconds int) {
	v.currentExpirationLock.Lock()
	v.currentExpiration = time.Now().Add(time.Duration(ttlSeconds) * time.Second)
	v.currentExpirationLock.Unlock()
}
1062430
func (c *Overlay) SetSuspended(suspended bool) (*gcdmessage.ChromeResponse, error) {
	var v OverlaySetSuspendedParams
	v.Suspended = suspended
	return c.SetSuspendedWithParams(&v)
}
327548
func (p ResourcePersistence) getOnePending(resID, pendingID string) (resourceDoc, error) {
	logger.Tracef("querying db for resource %q (pending %q)", resID, pendingID)
	id := pendingResourceID(resID, pendingID)
	var doc resourceDoc
	if err := p.base.One(resourcesC, id, &doc); err != nil {
		return doc, errors.Trace(err)
	}
	return doc, nil
}
1135182
func addPendingTiming(c context.Context, build *ui.MiloBuildLegacy, sr *swarming.SwarmingRpcsTaskResult) {
	created, err := time.Parse(SwarmingTimeLayout, sr.CreatedTs)
	if err != nil {
		return
	}
	build.Summary.PendingTime = ui.NewInterval(c, created, build.Summary.ExecutionTime.Started)
}
580258
func (store *store) ExtendLease(key lease.Key, request lease.Request, _ <-chan struct{}) error {
	return store.request(key.Lease, request, store.extendLeaseOps, "extending")
}
814159
func (cl *deployClient) AddPendingResources(applicationID string, chID charmstore.CharmID, csMac *macaroon.Macaroon, resources []charmresource.Resource) ([]string, error) {
	return cl.Client.AddPendingResources(client.AddPendingResourcesArgs{
		ApplicationID:      applicationID,
		CharmID:            chID,
		CharmStoreMacaroon: csMac,
		Resources:          resources,
	})
}
295512
func (a *AccountState) PendingPayments(ctx context.Context, limit int) ([]stellar1.PaymentSummary, error) {
	a.RLock()
	defer a.RUnlock()
	a.enqueueRefreshReq()
	if limit > 0 && limit < len(a.pending) {
		return a.pending[:limit], nil
	}
	return a.pending, nil
}
987416
func (s *CreateWorkspacesOutput) SetPendingRequests(v []*Workspace) *CreateWorkspacesOutput {
	s.PendingRequests = v
	return s
}