package semaphore

import (
	"sync"
	"time"

	"golang.org/x/net/context"
)

type empty struct{}

// The Semaphore interface specifies the operations permitted by a semaphore.
// semImpl is the canonical implementation.
type Semaphore interface {
	Resize(int)
	Acquire() bool
	AcquireContext(context.Context) (bool, time.Duration)
	AcquireWithin(time.Duration) (bool, time.Duration)
	Release()
	Size() int
	QueueSize() int
	Available() int
	MinAvailable() int
}

type semImpl struct {
	size         int
	queueSize    int
	requestQueue chan chan empty

	available    int
	minAvailable int
	mu           sync.RWMutex
}

// New creates a new semaphore with a maximum capacity of size and
// a queuing capacity of size queueSize.
func New(size, queueSize int) Semaphore {
	return &semImpl{
		size:         size,
		queueSize:    queueSize,
		requestQueue: make(chan chan empty, queueSize),

		available:    size,
		minAvailable: size,
	}
}

// MinAvailable returns the minimum number of available permits which existed at some time
// since the last call to MinAvailable (or since the semaphore's creation if it was not called).
// NOTE: this API is ONLY to be used by applications which are single consumers of this API.
// This API could become deprecated in the future if we get requests to support multiple consumers.
func (s *semImpl) MinAvailable() int {
	s.mu.Lock()
	m := s.minAvailable
	s.minAvailable = s.available
	s.mu.Unlock()
	return m
}

// Resize changes the maximum capacity to size.  TODO: support changing the queue size
func (s *semImpl) Resize(size int) {
	s.mu.Lock()
	defer s.mu.Unlock()

	oldSize := s.size
	s.size = size

	// For a period of time, available may be greater than size but
	// it will gracefully return to the state where available <= size as
	// Release() is called by clients.
	sizeDiff := size - oldSize
	s.available += sizeDiff
}

// Acquire returns true if there is an available permit to instantly acquire
// and returns false otherwise.
func (s *semImpl) Acquire() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.available > 0 {
		s.available -= 1
		if s.available < s.minAvailable {
			s.minAvailable = s.available
		}
		return true
	}
	return false
}

// AcquireContext will either successfully acquire a permit (blocking until resources are available)
// or fail to acquire a permit if the given context is cancelled - whichever comes first.
// AcquireContext returns as its first value true if the permit is acquired, and false otherwise.
// AcquireContext always returns as its second value the time spent in this function.
func (s *semImpl) AcquireContext(ctx context.Context) (bool, time.Duration) {
	return s.doAcquireCtx(ctx)
}

// AcquireWithin will either successfully acquire a permit (blocking until resources are available)
// or fail to acquire a permit if the timeout theshold is exceeded - whichever comes first.
// AcquireWithin returns as its first value true if a permit is acquired, and false otherwise.
// AcquireWithin always returns as its second value the time spent in this function.
// NOTE that using AcquireContext is preferred to AcquireWithin.
func (s *semImpl) AcquireWithin(timeout time.Duration) (bool, time.Duration) {
	if timeout <= 0 {
		return s.Acquire(), 0
	}

	ctx, _ := context.WithTimeout(context.Background(), timeout)
	return s.doAcquireCtx(ctx)
}

func (s *semImpl) doAcquireCtx(ctx context.Context) (bool, time.Duration) {
	startTime := time.Now()
	s.mu.Lock()
	// If there are readily available permits, acquire one immediately.
	if s.available > 0 {
		s.available -= 1
		if s.available < s.minAvailable {
			s.minAvailable = s.available
		}
		s.mu.Unlock()
		return true, 0
	}
	// If we have too many requests queued up, fail instantly.
	if len(s.requestQueue) == s.queueSize {
		s.mu.Unlock()
		return false, 0
	}
	// Make a channel for us to receive a permit on, and push this channel onto the queue.
	req := make(chan empty, 1)
	s.requestQueue <- req
	s.mu.Unlock()
	select {
	case <-req:
		return true, time.Since(startTime)
	case <-ctx.Done():
		// At some point the pool will give us our permit, so release it immediately.
		go func() {
			<-req
			s.Release()
		}()
		return false, time.Since(startTime)
	}
}

// Release releases a permit.
func (s *semImpl) Release() {
	s.mu.Lock()
	defer s.mu.Unlock()
	// If we've recently reduced our pool size, don't hand this resource to another caller.
	if s.available < 0 {
		s.available += 1
		return
	}
	// If there's already a consumer queueing on a permit, give it to them.  Otherwise,
	// increase the number of available permits.
	select {
	case req := <-s.requestQueue:
		req <- empty{}
	default:
		s.available += 1
	}
}

// Size returns the maximum capacity of the semaphore.
func (s *semImpl) Size() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.size
}

// Size returns the maximum queueing capacity of the semaphore.
func (s *semImpl) QueueSize() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.queueSize
}

// Available returns how many permits are currently available to be acquired.
func (s *semImpl) Available() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.available
}
