<commit_before>package scheduler

import (
	"sync/atomic"
	"time"

	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
	log "github.com/golang/glog"
	"github.com/mesos/mesos-go/mesos"
)

type OfferFilter func(*mesos.Offer) bool

type OfferRegistry interface {
	// Initialize the instance, spawning necessary housekeeping go routines.
	Init()
	Add([]*mesos.Offer)
	AwaitNew(id string, f OfferFilter) <-chan empty

	// invoked when offers are rescinded or expired
	Delete(string)

	Get(offerId string) (PerishableOffer, bool)

	Walk(Walker) error

	// invalidate one or all (when offerId="") offers; offers are not declined,
	// but are simply flagged as expired in the offer history
	Invalidate(offerId string)
}

// callback that is invoked during a walk through a series of active offers,
// returning with stop=true when the walk should stop.
type Walker func(offer PerishableOffer) (stop bool, err error)

type OfferRegistryConfig struct {
	declineOffer  func(offerId string) error
	ttl           time.Duration // determines a perishable offer's expiration deadline: now+ttl
	lingerTtl     time.Duration // if zero, offers will not linger in the FIFO past their expiration deadline
	walkDelay     time.Duration // specifies the sleep time between expired offer harvests
	listenerDelay time.Duration // specifies the sleep time between offer listener notifications
}

type offerStorage struct {
	OfferRegistryConfig
	offers    *cache.FIFO // collection of timedOffer's
	listeners *cache.FIFO // collection of <-chan empty
}

type timedOffer struct {
	*mesos.Offer
	expiration time.Time
	acquired   int32 // 1 = acquired, 0 = free
}

type expiredOffer struct{}

type PerishableOffer interface {
	// returns true if this offer has expired
	hasExpired() bool
	// if not yet expired, return mesos offer details; otherwise nil
	details() *mesos.Offer
	// mark this offer as acquired, returning true if it was previously unacquired. thread-safe.
	acquire() bool
	// mark this offer as un-acquired. thread-safe.
	release()
}

func (e *expiredOffer) hasExpired() bool {
	return true
}

func (e *expiredOffer) details() *mesos.Offer {
	return nil
}

func (e *expiredOffer) acquire() bool {
	return false
}

func (e *expiredOffer) release() {}

func (to *timedOffer) hasExpired() bool {
	return time.Now().After(to.expiration)
}

func (to *timedOffer) details() *mesos.Offer {
	return to.Offer
}

func (to *timedOffer) acquire() bool {
	return atomic.CompareAndSwapInt32(&to.acquired, 0, 1)
}

func (to *timedOffer) release() {
	atomic.CompareAndSwapInt32(&to.acquired, 1, 0)
}

func CreateOfferRegistry(c OfferRegistryConfig) OfferRegistry {
	return &offerStorage{c, cache.NewFIFO(), cache.NewFIFO()}
}

func (s *offerStorage) Add(offers []*mesos.Offer) {
	now := time.Now()
	for _, offer := range offers {
		offerId := offer.Id.GetValue()
		log.V(3).Infof("Receiving offer %v", offerId)
		s.offers.Add(offerId, &timedOffer{offer, now.Add(s.ttl), 0})
	}
}

// delete an offer from storage, meaning that we expire it
func (s *offerStorage) Delete(offerId string) {
	if offer, ok := s.Get(offerId); ok {
		log.V(3).Infof("Deleting offer %v", offerId)
		// attempt to block others from consuming the offer. if it's already been
		// claimed and is not yet lingering then don't decline it - just mark it as
		// expired in the history: allow a prior claimant to attempt to launch with it
		myoffer := offer.acquire()
		if offer.details() != nil {
			if myoffer {
				log.V(3).Infof("Declining offer %v", offerId)
				if err := s.declineOffer(offerId); err != nil {
					log.Warningf("Failed to decline offer %v: %v", offerId, err)
				}
			} else {
				// some pod has acquired this and may attempt to launch a task with it
				// failed schedule/launch attempts are requried to release() any claims on the offer
				go func() {
					// TODO(jdef): not sure what a good value is here. the goal is to provide a
					// launchTasks (driver) operation enough time to complete so that we don't end
					// up delining an offer that we're actually attempting to use.
					time.Sleep(2 * s.ttl)

					// at this point the offer is in one of five states:
					// a) permanently deleted: expired due to timeout
					// b) permanently deleted: expired due to having been rescinded
					// c) lingering: expired due to timeout
					// d) lingering: expired due to having been rescinded
					// e) claimed: task launched and it using resources from this offer
					// we want to **avoid** delining an offer that's claimed: attempt to acquire
					if offer.acquire() {
						// previously claimed offer was released, perhaps due to a launch
						// failure, so we should attempt to decline
						if err := s.declineOffer(offerId); err != nil {
							log.Warningf("Failed to decline (previously claimed) offer %v: %v", offerId, err)
						}
					}
				}()
			}
		}
		s.expireOffer(offer)
	} // else, ignore offers not in the history
}

// expire all known, live offers
func (s *offerStorage) Invalidate(offerId string) {
	if offerId != "" {
		s.invalidateOne(offerId)
		return
	}
	obj := s.offers.List()
	for _, o := range obj {
		offer, ok := o.(PerishableOffer)
		if !ok {
			log.Errorf("Expected perishable offer, not %v", o)
			continue
		}
		offer.acquire() // attempt to block others from using it
		s.expireOffer(offer)
		// don't reject it from the controller, we already know that it's an invalid offer
	}
}

func (s *offerStorage) invalidateOne(offerId string) {
	if offer, ok := s.Get(offerId); ok {
		offer.acquire() // attempt to block others from using it
		s.expireOffer(offer)
		// don't reject it from the controller, we already know that it's an invalid offer
	}
}

// Walk the collection of offers, Delete()ing those that have expired,
// visiting those that haven't with the Walked. The walk stops either as
// indicated by the Walker or when the end of the offer list is reached.
func (s *offerStorage) Walk(w Walker) error {
	for offerId := range s.offers.Contains() {
		offer, ok := s.Get(offerId)
		if !ok {
			// offer disappeared...
			continue
		}
		// delete any non-lingering offers that have expired
		if offer.hasExpired() && offer.details() != nil {
			s.Delete(offerId)
			continue
		}
		if stop, err := w(offer); err != nil {
			return err
		} else if stop {
			return nil
		}
	}
	return nil
}

func (s *offerStorage) expireOffer(offer PerishableOffer) {
	// the offer may or may not be expired due to TTL so check for details
	// since that's a more reliable determinant of lingering status
	if details := offer.details(); details != nil {
		// recently expired, should linger
		offerId := details.Id.GetValue()
		log.V(3).Infof("Expiring offer %v", offerId)
		if s.lingerTtl > 0 {
			log.V(3).Infof("offer will linger: %v", offerId)
			s.offers.Update(offerId, &expiredOffer{})
			go func() {
				time.Sleep(s.lingerTtl)<commit_msg>never pass expired offers to walkers<commit_after>package scheduler

import (
	"sync/atomic"
	"time"

	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
	log "github.com/golang/glog"
	"github.com/mesos/mesos-go/mesos"
)

type OfferFilter func(*mesos.Offer) bool

type OfferRegistry interface {
	// Initialize the instance, spawning necessary housekeeping go routines.
	Init()
	Add([]*mesos.Offer)
	AwaitNew(id string, f OfferFilter) <-chan empty

	// invoked when offers are rescinded or expired
	Delete(string)

	Get(offerId string) (PerishableOffer, bool)

	Walk(Walker) error

	// invalidate one or all (when offerId="") offers; offers are not declined,
	// but are simply flagged as expired in the offer history
	Invalidate(offerId string)
}

// callback that is invoked during a walk through a series of active offers,
// returning with stop=true when the walk should stop.
type Walker func(offer PerishableOffer) (stop bool, err error)

type OfferRegistryConfig struct {
	declineOffer  func(offerId string) error
	ttl           time.Duration // determines a perishable offer's expiration deadline: now+ttl
	lingerTtl     time.Duration // if zero, offers will not linger in the FIFO past their expiration deadline
	walkDelay     time.Duration // specifies the sleep time between expired offer harvests
	listenerDelay time.Duration // specifies the sleep time between offer listener notifications
}

type offerStorage struct {
	OfferRegistryConfig
	offers    *cache.FIFO // collection of timedOffer's
	listeners *cache.FIFO // collection of <-chan empty
}

type timedOffer struct {
	*mesos.Offer
	expiration time.Time
	acquired   int32 // 1 = acquired, 0 = free
}

type expiredOffer struct{}

type PerishableOffer interface {
	// returns true if this offer has expired
	hasExpired() bool
	// if not yet expired, return mesos offer details; otherwise nil
	details() *mesos.Offer
	// mark this offer as acquired, returning true if it was previously unacquired. thread-safe.
	acquire() bool
	// mark this offer as un-acquired. thread-safe.
	release()
}

func (e *expiredOffer) hasExpired() bool {
	return true
}

func (e *expiredOffer) details() *mesos.Offer {
	return nil
}

func (e *expiredOffer) acquire() bool {
	return false
}

func (e *expiredOffer) release() {}

func (to *timedOffer) hasExpired() bool {
	return time.Now().After(to.expiration)
}

func (to *timedOffer) details() *mesos.Offer {
	return to.Offer
}

func (to *timedOffer) acquire() bool {
	return atomic.CompareAndSwapInt32(&to.acquired, 0, 1)
}

func (to *timedOffer) release() {
	atomic.CompareAndSwapInt32(&to.acquired, 1, 0)
}

func CreateOfferRegistry(c OfferRegistryConfig) OfferRegistry {
	return &offerStorage{c, cache.NewFIFO(), cache.NewFIFO()}
}

func (s *offerStorage) Add(offers []*mesos.Offer) {
	now := time.Now()
	for _, offer := range offers {
		offerId := offer.Id.GetValue()
		log.V(3).Infof("Receiving offer %v", offerId)
		s.offers.Add(offerId, &timedOffer{offer, now.Add(s.ttl), 0})
	}
}

// delete an offer from storage, meaning that we expire it
func (s *offerStorage) Delete(offerId string) {
	if offer, ok := s.Get(offerId); ok {
		log.V(3).Infof("Deleting offer %v", offerId)
		// attempt to block others from consuming the offer. if it's already been
		// claimed and is not yet lingering then don't decline it - just mark it as
		// expired in the history: allow a prior claimant to attempt to launch with it
		myoffer := offer.acquire()
		if offer.details() != nil {
			if myoffer {
				log.V(3).Infof("Declining offer %v", offerId)
				if err := s.declineOffer(offerId); err != nil {
					log.Warningf("Failed to decline offer %v: %v", offerId, err)
				}
			} else {
				// some pod has acquired this and may attempt to launch a task with it
				// failed schedule/launch attempts are requried to release() any claims on the offer
				go func() {
					// TODO(jdef): not sure what a good value is here. the goal is to provide a
					// launchTasks (driver) operation enough time to complete so that we don't end
					// up delining an offer that we're actually attempting to use.
					time.Sleep(2 * s.ttl)

					// at this point the offer is in one of five states:
					// a) permanently deleted: expired due to timeout
					// b) permanently deleted: expired due to having been rescinded
					// c) lingering: expired due to timeout
					// d) lingering: expired due to having been rescinded
					// e) claimed: task launched and it using resources from this offer
					// we want to **avoid** delining an offer that's claimed: attempt to acquire
					if offer.acquire() {
						// previously claimed offer was released, perhaps due to a launch
						// failure, so we should attempt to decline
						if err := s.declineOffer(offerId); err != nil {
							log.Warningf("Failed to decline (previously claimed) offer %v: %v", offerId, err)
						}
					}
				}()
			}
		}
		s.expireOffer(offer)
	} // else, ignore offers not in the history
}

// expire all known, live offers
func (s *offerStorage) Invalidate(offerId string) {
	if offerId != "" {
		s.invalidateOne(offerId)
		return
	}
	obj := s.offers.List()
	for _, o := range obj {
		offer, ok := o.(PerishableOffer)
		if !ok {
			log.Errorf("Expected perishable offer, not %v", o)
			continue
		}
		offer.acquire() // attempt to block others from using it
		s.expireOffer(offer)
		// don't reject it from the controller, we already know that it's an invalid offer
	}
}

func (s *offerStorage) invalidateOne(offerId string) {
	if offer, ok := s.Get(offerId); ok {
		offer.acquire() // attempt to block others from using it
		s.expireOffer(offer)
		// don't reject it from the controller, we already know that it's an invalid offer
	}
}

// Walk the collection of offers, Delete()ing those that have expired,
// visiting those that haven't with the Walked. The walk stops either as
// indicated by the Walker or when the end of the offer list is reached.
func (s *offerStorage) Walk(w Walker) error {
	for offerId := range s.offers.Contains() {
		offer, ok := s.Get(offerId)
		if !ok {
			// offer disappeared...
			continue
		}
		if offer.hasExpired() {
			// delete any non-lingering offers that have expired
			if offer.details() != nil {
				s.Delete(offerId)
			}
			// never pass expired offers to walkers
			continue
		}
		if stop, err := w(offer); err != nil {
			return err
		} else if stop {
			return nil
		}
	}
	return nil
}

func (s *offerStorage) expireOffer(offer PerishableOffer) {
	// the offer may or may not be expired due to TTL so check for details
	// since that's a more reliable determinant of lingering status
	if details := offer.details(); details != nil {
		// recently expired, should linger
		offerId := details.Id.GetValue()
		log.V(3).Infof("Expiring offer %v", offerId)
		if s.lingerTtl > 0 {
			log.V(3).Infof("offer will linger: %v", offerId)
			s.offers.Update(offerId, &expiredOffer{})
			go func() {
				time.Sleep(s.lingerTtl)
{'fraction_non_alphanumeric': 0.10620233127068643, 'fraction_numerical': 0.002878111958555188, 'mean_word_length': 3.486442866365397, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6301597', 'n_tokens_mistral': 4933, 'n_tokens_neox': 4321, 'n_words': 1609}