/*

 * 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/funnelQueue.c#2 $

 */

#include "funnelQueue.h"

#include "memoryAlloc.h"

#include "permassert.h"

#include "uds.h"

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

int makeFunnelQueue(FunnelQueue **queuePtr)

{

  // Allocate the queue on a cache line boundary so the producer and consumer

  // fields in the structure will land on separate cache lines.

  FunnelQueue *queue;

  int result = ALLOCATE(1, FunnelQueue, "funnel queue", &queue);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Initialize the stub entry and put it in the queue, establishing the

  // invariant that queue->newest and queue->oldest are never null.

  queue->stub.next = NULL;

  queue->newest = &queue->stub;

  queue->oldest = &queue->stub;

  *queuePtr = queue;

  return UDS_SUCCESS;

}

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

void freeFunnelQueue(FunnelQueue *queue)

{

  FREE(queue);

}

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

static FunnelQueueEntry *getOldest(FunnelQueue *queue)

{

 /*

  * Barrier requirements: We need a read barrier between reading a "next"

  * field pointer value and reading anything it points to. There's an

  * accompanying barrier in funnelQueuePut between its caller setting up the

  * entry and making it visible.

  */

  FunnelQueueEntry *oldest = queue->oldest;

  FunnelQueueEntry *next   = oldest->next;

  if (oldest == &queue->stub) {

    // When the oldest entry is the stub and it has no successor, the queue is

    // logically empty.

    if (next == NULL) {

      return NULL;

    }

    // The stub entry has a successor, so the stub can be dequeued and ignored

    // without breaking the queue invariants.

    oldest = next;

    queue->oldest = oldest;

    smp_read_barrier_depends();

    next = oldest->next;

  }

  // We have a non-stub candidate to dequeue. If it lacks a successor, we'll

  // need to put the stub entry back on the queue first.

  if (next == NULL) {

    FunnelQueueEntry *newest = queue->newest;

    if (oldest != newest) {

      // Another thread has already swung queue->newest atomically, but not

      // yet assigned previous->next. The queue is really still empty.

      return NULL;

    }

    // Put the stub entry back on the queue, ensuring a successor will

    // eventually be seen.

    funnelQueuePut(queue, &queue->stub);

    // Check again for a successor.

    next = oldest->next;

    if (next == NULL) {

      // We lost a race with a producer who swapped queue->newest before we

      // did, but who hasn't yet updated previous->next. Try again later.

      return NULL;

    }

  }

  return oldest;

}

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

FunnelQueueEntry *funnelQueuePoll(FunnelQueue *queue)

{

  FunnelQueueEntry *oldest = getOldest(queue);

  if (oldest == NULL) {

    return oldest;

  }

  /*

   * Dequeue the oldest entry and return it. Only one consumer thread may call

   * this function, so no locking, atomic operations, or fences are needed;

   * queue->oldest is owned by the consumer and oldest->next is never used by

   * a producer thread after it is swung from NULL to non-NULL.

   */

  queue->oldest = oldest->next;

  /*

   * Make sure the caller sees the proper stored data for this entry.

   *

   * Since we've already fetched the entry pointer we stored in

   * "queue->oldest", this also ensures that on entry to the next call we'll

   * properly see the dependent data.

   */

  smp_rmb();

  /*

   * If "oldest" is a very light-weight work item, we'll be looking

   * for the next one very soon, so prefetch it now.

   */

  prefetchAddress(queue->oldest, true);

  oldest->next = NULL;

  return oldest;

}

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

bool isFunnelQueueEmpty(FunnelQueue *queue)

{

  return getOldest(queue) == NULL;

}

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

bool isFunnelQueueIdle(FunnelQueue *queue)

{

  /*

   * Oldest is not the stub, so there's another entry, though if next is

   * NULL we can't retrieve it yet.

   */

  if (queue->oldest != &queue->stub) {

    return false;

  }

  /*

   * Oldest is the stub, but newest has been updated by _put(); either

   * there's another, retrievable entry in the list, or the list is

   * officially empty but in the intermediate state of having an entry

   * added.

   *

   * Whether anything is retrievable depends on whether stub.next has

   * been updated and become visible to us, but for idleness we don't

   * care. And due to memory ordering in _put(), the update to newest

   * would be visible to us at the same time or sooner.

   */

  if (queue->newest != &queue->stub) {

    return false;

  }

  // Otherwise, we're idle.

  return true;

}