/*

 * Copyright (c) 2020 Red Hat, Inc.

 *

 * This program is free software; you can redistribute it and/or

 * modify it under the terms of the GNU General Public License

 * as published by the Free Software Foundation; either version 2

 * of the License, or (at your option) any later version.

 * 

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 * 

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

 * 02110-1301, USA. 

 *

 * $Id: //eng/uds-releases/jasper/src/uds/util/eventCount.c#2 $

 */

/**

 * This EventCount implementation uses a posix semaphore for portability,

 * although a futex would be slightly superior to use and easy to substitute.

 * It is designed to make signalling as cheap as possible, since that is the

 * code path likely triggered on most updates to a lock-free data structure.

 * Waiters are likely going to sleep, so optimizing for that case isn't

 * necessary.

 *

 * The critical field is the state, which is really two fields that can be

 * atomically updated in unison: an event counter and a waiter count. Every

 * call to eventCountPrepare() issues a wait token by atomically incrementing

 * the waiter count. The key invariant is a strict accounting of the number of

 * tokens issued. Every token returned by eventCountPrepare() is a contract

 * that the caller will call acquireSemaphore() and a signaller will call

 * releaseSemaphore(), each exactly once. Atomic updates to the state field

 * ensure that each token is counted once and that tokens are not lost.

 * Cancelling a token attempts to take a fast-path by simply decrementing the

 * waiters field, but if the token has already been claimed by a signaller,

 * the canceller must still wait on the semaphore to consume the transferred

 * token.

 *

 * The state field is 64 bits, partitioned into a 16-bit waiter field and a

 * 48-bit counter. We are unlikely to have 2^16 threads, much less 2^16

 * threads waiting on any single event transition. 2^48 microseconds is

 * several years, so a token holder would have to wait that long for the

 * counter to wrap around, and then call eventCountWait() at the exact right

 * time to see the re-used counter, in order to lose a wakeup due to counter

 * wrap-around. Using a 32-bit state field would greatly increase that chance,

 * but if forced to do so, the implementation could likely tolerate it since

 * callers are supposed to hold tokens for miniscule periods of time.

 * Fortunately, x64 has 64-bit compare-and-swap, and the performance of

 * interlocked 64-bit operations appears to be about the same as for 32-bit

 * ones, so being paranoid and using 64 bits costs us nothing.

 *

 * Here are some sequences of calls and state transitions:

 *

 *    action                    postcondition

 *                        counter   waiters   semaphore

 *    initialized           0          0          0

 *    prepare               0          1          0

 *    wait (blocks)         0          1          0

 *    signal                1          0          1

 *    wait (unblocks)       1          0          0

 *

 *    signal (fast-path)    1          0          0

 *    signal (fast-path)    1          0          0

 *

 *    prepare A             1          1          0

 *    prepare B             1          2          0

 *    signal                2          0          2

 *    wait B (fast-path)    2          0          1

 *    wait A (fast-path)    2          0          0

 *

 *    prepare               2          1          0

 *    cancel (fast-path)    2          0          0

 *

 *    prepare               2          1          0

 *    signal                3          0          1

 *    cancel (must wait)    3          0          0

 *

 * The EventCount structure is aligned, sized, and allocated to cache line

 * boundaries to avoid any false sharing between the EventCount and other

 * shared state. The state field and semaphore should fit on a single cache

 * line. The instrumentation counters increase the size of the structure so it

 * rounds up to use two (64-byte x86) cache lines.

 *

 * XXX Need interface to access or display instrumentation counters.

 **/

#include "eventCount.h"

#include "atomicDefs.h"

#include "common.h"

#include "compiler.h"

#include "cpu.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "threads.h"

enum {

  ONE_WAITER   = 1,               // value used to increment the waiters field

  ONE_EVENT    = (1 << 16),       // value used to increment the event counter

  WAITERS_MASK = (ONE_EVENT - 1), // bit mask to access the waiters field

  EVENTS_MASK  = ~WAITERS_MASK,   // bit mask to access the event counter

};

struct eventCount {

  // Atomically mutable state:

  // low  16 bits: the number of wait tokens not posted to the semaphore

  // high 48 bits: current event counter

  atomic64_t state;

  // Semaphore used to block threads when waiting is required.

  Semaphore semaphore;

  // Instrumentation counters.

  // Declare alignment so we don't share a cache line.

} __attribute__((aligned(CACHE_LINE_BYTES)));

/**

 * Test the event field in two tokens for equality.

 *

 * @return  true iff the tokens contain the same event field value

 **/

static INLINE bool sameEvent(EventToken token1, EventToken token2)

{

  return ((token1 & EVENTS_MASK) == (token2 & EVENTS_MASK));

}

/**********************************************************************/

void eventCountBroadcast(EventCount *ec)

{

  // Even if there are no waiters (yet), we will need a memory barrier.

  smp_mb();

  uint64_t waiters;

  uint64_t state = atomic64_read(&ec->state);

  uint64_t oldState = state;

  do {

    // Check if there are any tokens that have not yet been been transferred

    // to the semaphore. This is the fast no-waiters path.

    waiters = (state & WAITERS_MASK);

    if (waiters == 0) {

      // Fast path first time through--no need to signal or post if there are

      // no observers.

      return;

    }

    /*

     * Attempt to atomically claim all the wait tokens and bump the event count

     * using an atomic compare-and-swap.  This operation contains a memory

     * barrier.

     */

    EventToken newState = ((state & ~WAITERS_MASK) + ONE_EVENT);

    oldState = state;

    state = atomic64_cmpxchg(&ec->state, oldState, newState);

    // The cmpxchg fails when we lose a race with a new waiter or another

    // signaller, so try again.

  } while (unlikely(state != oldState));

  /*

   * Wake the waiters by posting to the semaphore. This effectively transfers

   * the wait tokens to the semaphore. There's sadly no bulk post for posix

   * semaphores, so we've got to loop to do them all.

   */

  while (waiters-- > 0) {

    releaseSemaphore(&ec->semaphore);

  }

}

/**

 * Attempt to cancel a prepared wait token by decrementing the

 * number of waiters in the current state. This can only be done

 * safely if the event count hasn't been bumped.

 *

 * @param ec     the event count on which the wait token was issued

 * @param token  the wait to cancel

 *

 * @return true if the wait was cancelled, false if the caller must

 *         still wait on the semaphore

 **/

static INLINE bool fastCancel(EventCount *ec, EventToken token)

{

  EventToken currentToken = atomic64_read(&ec->state);

  while (sameEvent(currentToken, token)) {

    // Try to decrement the waiter count via compare-and-swap as if we had

    // never prepared to wait.

    EventToken et = atomic64_cmpxchg(&ec->state, currentToken,

                                     currentToken - 1);

    if (et == currentToken) {

      return true;

    }

    currentToken = et;

  }

  return false;

}

/**

 * Consume a token from the semaphore, waiting (with an optional timeout) if

 * one is not currently available. Also attempts to count the number of times

 * we'll actually have to wait because there are no tokens (permits) available

 * in the semaphore, and the number of times the wait times out.

 *

 * @param ec       the event count instance

 * @param timeout  an optional timeout value to pass to attemptSemaphore()

 *

 * @return true if a token was consumed, otherwise false only if a timeout

 *         was specified and we timed out

 **/

static bool consumeWaitToken(EventCount *ec, const RelTime *timeout)

{

  // Try to grab a token without waiting.

  if (attemptSemaphore(&ec->semaphore, 0)) {

    return true;

  }

  if (timeout == NULL) {

    acquireSemaphore(&ec->semaphore);

  } else if (!attemptSemaphore(&ec->semaphore, *timeout)) {

    return false;

  }

  return true;

}

/**********************************************************************/

int makeEventCount(EventCount **ecPtr)

{

  // The event count will be allocated on a cache line boundary so there will

  // not be false sharing of the line with any other data structure.

  EventCount *ec = NULL;

  int result = ALLOCATE(1, EventCount, "event count", &ec);

  if (result != UDS_SUCCESS) {

    return result;

  }

  atomic64_set(&ec->state, 0);

  result = initializeSemaphore(&ec->semaphore, 0);

  if (result != UDS_SUCCESS) {

    FREE(ec);

    return result;

  }

  *ecPtr = ec;

  return UDS_SUCCESS;

}

/**********************************************************************/

void freeEventCount(EventCount *ec)

{

  if (ec == NULL) {

    return;

  }

  destroySemaphore(&ec->semaphore);

  FREE(ec);

}

/**********************************************************************/

EventToken eventCountPrepare(EventCount *ec)

{

  return atomic64_add_return(ONE_WAITER, &ec->state);

}

/**********************************************************************/

void eventCountCancel(EventCount *ec, EventToken token)

{

  // Decrement the waiter count if the event hasn't been signalled.

  if (fastCancel(ec, token)) {

    return;

  }

  // A signaller has already transferred (or promised to transfer) our token

  // to the semaphore, so we must consume it from the semaphore by waiting.

  eventCountWait(ec, token, NULL);

}

/**********************************************************************/

bool eventCountWait(EventCount *ec, EventToken token, const RelTime *timeout)

{

  for (;;) {

    // Wait for a signaller to transfer our wait token to the semaphore.

    if (!consumeWaitToken(ec, timeout)) {

      // The wait timed out, so we must cancel the token instead. Try to

      // decrement the waiter count if the event hasn't been signalled.

      if (fastCancel(ec, token)) {

        return false;

      }

      /*

       * We timed out, but a signaller came in before we could cancel the

       * wait. We have no choice but to wait for the semaphore to be posted.

       * Since signaller has promised to do it, the wait will be short. The

       * timeout and the signal happened at about the same time, so either

       * outcome could be returned. It's simpler to ignore the timeout.

       */

      timeout = NULL;

      continue;

    }

    // A wait token has now been consumed from the semaphore.

    // Stop waiting if the count has changed since the token was acquired.

    if (!sameEvent(token, atomic64_read(&ec->state))) {

      return true;

    }

    // We consumed someone else's wait token. Put it back in the semaphore,

    // which will wake another waiter, hopefully one who can stop waiting.

    releaseSemaphore(&ec->semaphore);

    // Attempt to give an earlier waiter a shot at the semaphore.

    yieldScheduler();

  }

}