/*
** Copyright (C) 2004-2006 by Carnegie Mellon University.
**
** @OPENSOURCE_HEADER_START@
** 
** Use of the SILK system and related source code is subject to the terms 
** of the following licenses:
** 
** GNU Public License (GPL) Rights pursuant to Version 2, June 1991
** Government Purpose License Rights (GPLR) pursuant to DFARS 252.225-7013
** 
** NO WARRANTY
** 
** ANY INFORMATION, MATERIALS, SERVICES, INTELLECTUAL PROPERTY OR OTHER 
** PROPERTY OR RIGHTS GRANTED OR PROVIDED BY CARNEGIE MELLON UNIVERSITY 
** PURSUANT TO THIS LICENSE (HEREINAFTER THE "DELIVERABLES") ARE ON AN 
** "AS-IS" BASIS. CARNEGIE MELLON UNIVERSITY MAKES NO WARRANTIES OF ANY 
** KIND, EITHER EXPRESS OR IMPLIED AS TO ANY MATTER INCLUDING, BUT NOT 
** LIMITED TO, WARRANTY OF FITNESS FOR A PARTICULAR PURPOSE, 
** MERCHANTABILITY, INFORMATIONAL CONTENT, NONINFRINGEMENT, OR ERROR-FREE 
** OPERATION. CARNEGIE MELLON UNIVERSITY SHALL NOT BE LIABLE FOR INDIRECT, 
** SPECIAL OR CONSEQUENTIAL DAMAGES, SUCH AS LOSS OF PROFITS OR INABILITY 
** TO USE SAID INTELLECTUAL PROPERTY, UNDER THIS LICENSE, REGARDLESS OF 
** WHETHER SUCH PARTY WAS AWARE OF THE POSSIBILITY OF SUCH DAMAGES. 
** LICENSEE AGREES THAT IT WILL NOT MAKE ANY WARRANTY ON BEHALF OF 
** CARNEGIE MELLON UNIVERSITY, EXPRESS OR IMPLIED, TO ANY PERSON 
** CONCERNING THE APPLICATION OF OR THE RESULTS TO BE OBTAINED WITH THE 
** DELIVERABLES UNDER THIS LICENSE.
** 
** Licensee hereby agrees to defend, indemnify, and hold harmless Carnegie 
** Mellon University, its trustees, officers, employees, and agents from 
** all claims or demands made against them (and any related losses, 
** expenses, or attorney's fees) arising out of, or relating to Licensee's 
** and/or its sub licensees' negligent use or willful misuse of or 
** negligent conduct or willful misconduct regarding the Software, 
** facilities, or other rights or assistance granted by Carnegie Mellon 
** University under this License, including, but not limited to, any 
** claims of product liability, personal injury, death, damage to 
** property, or violation of any laws or regulations.
** 
** Carnegie Mellon University Software Engineering Institute authored 
** documents are sponsored by the U.S. Department of Defense under 
** Contract F19628-00-C-0003. Carnegie Mellon University retains 
** copyrights in all material produced under this contract. The U.S. 
** Government retains a non-exclusive, royalty-free license to publish or 
** reproduce these documents, or allow others to do so, for U.S. 
** Government purposes only pursuant to the copyright license under the 
** contract clause at 252.227.7013.
** 
** @OPENSOURCE_HEADER_END@
*/

/*
**  Merged deque implementation
**
*/


/* Includes */
#include "silk.h"

RCSIDENT("$SiLK: skdqmerged.c 3856 2006-05-26 15:38:34Z mwd $");

#include "skdqprivate.h"

/* Merged deque */
typedef struct merged_dq_t {
    skDeque_t q[2];		/* 0 == back, 1 == front */
} merged_dq_t;

/* Merged deque methods */
static skDQErr_t merged_status(skDeque_t self);
static skDQErr_t merged_pop(skDeque_t self, void **item, 
			    uint8_t block, uint8_t front, uint32_t seconds);
static skDQErr_t merged_peek(skDeque_t self, void **item, uint8_t front);
static skDQErr_t merged_push(skDeque_t self, void *item, uint8_t front);
static skDQErr_t merged_destroy(skDeque_t self);
static skDQErr_t merged_block(skDeque_t self, uint8_t flag);
static uint32_t merged_size(skDeque_t self);


/* Creates a new pseudo-deque which acts like a deque with all the
   elements of q1 in front of q2.  q1 and q2 continue behaving
   normally. */
skDeque_t skDequeCreateMerged(skDeque_t q1, skDeque_t q2)
{
    volatile skDeque_t deque;
    merged_dq_t *data;
    int oldtype;

    if (q1 == NULL || q2 == NULL ||
        q1->data == NULL || q2->data == NULL) {
        return NULL;
    }

    /* memory allocation */
    if ((deque = (skDeque_t)malloc(sizeof(sk_deque_t))) == NULL) {
        return NULL;
    }
    if ((data = (merged_dq_t *)malloc(sizeof(merged_dq_t))) == NULL) {
        free(deque);
        return NULL;
    }

    /* Initialize data */
    if ((data->q[1] = skDequeCopy(q1)) == NULL) {
        free(data);
        free(deque);
        return NULL;
    }
    if ((data->q[0] = skDequeCopy(q2)) == NULL) {
        skDequeDestroy(data->q[1]);
        free(data);
        free(deque);
        return NULL;
    }

    /* Initialize deque */
    deque->ref = 1;		/* Initialize to 1 */
    pthread_mutex_init(&deque->mutex_data, NULL);
    pthread_cond_init(&deque->cond_data, NULL);
    deque->mutex = &deque->mutex_data;
    deque->cond = &deque->cond_data;
    deque->methods.status = merged_status;
    deque->methods.pop = merged_pop;
    deque->methods.peek = merged_peek;
    deque->methods.push = merged_push;
    deque->methods.destroy = merged_destroy;
    deque->methods.block = merged_block;
    deque->methods.size = merged_size;
    deque->data = (void *)data;

    /** Reorganize subdeques' mutexes and condition variables **/

    /* Lock our own */
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &oldtype);
    pthread_cleanup_push((callbackfn_t)pthread_mutex_unlock, 
                         (void *)deque->mutex);
    pthread_mutex_lock(deque->mutex);

    /* Grab q[0] */
    pthread_cleanup_push((callbackfn_t)pthread_mutex_unlock, 
                         (void *)data->q[0]->mutex);
    pthread_mutex_lock(data->q[0]->mutex);
    /* Fix its mutex and condition variable to be ours */
    data->q[0]->mutex = deque->mutex;
    data->q[0]->cond = deque->cond;
    /* Wake up anything waiting on it so they pick up the new condition
       variable. */
    pthread_cond_broadcast(&data->q[0]->cond_data);
    pthread_cleanup_pop(1);	/* data->q[0]->mutex */

    /* Grab q[1] */
    pthread_cleanup_push((callbackfn_t)pthread_mutex_unlock, 
                         (void *)data->q[1]->mutex);
    pthread_mutex_lock(data->q[1]->mutex);
    /* Fix its mutex and condition variable to be ours */
    data->q[1]->mutex = deque->mutex;
    data->q[1]->cond = deque->cond;
    /* Wake up anything waiting on it so they pick up the new condition
       variable. */
    pthread_cond_broadcast(&data->q[1]->cond_data);
    pthread_cleanup_pop(1);	/* data->q[1]->mutex */

    pthread_cleanup_pop(1);	/* deque->mutex */
    pthread_setcanceltype(oldtype, NULL);
  
    return deque;
}

static skDQErr_t merged_status(skDeque_t self) 
{
    merged_dq_t *q = (merged_dq_t *)self->data;
    skDQErr_t retval;

    if (q == NULL) {
        return SKDQ_ERROR;
    }

    if ((retval = q->q[0]->methods.status(q->q[0])) == SKDQ_EMPTY) {
        retval = q->q[1]->methods.status(q->q[1]);
    }
  
    return retval;
}

static skDQErr_t merged_pop(skDeque_t self, void **item, 
			    uint8_t block, uint8_t f,
                            uint32_t seconds)
{
    merged_dq_t *q = (merged_dq_t *)self->data;
    skDQErr_t retval = SKDQ_SUCCESS;
    uint8_t b;
    int rv;
    struct timeval now;
    struct timespec to;

    if (q == NULL) {
        return SKDQ_ERROR;
    }

    if (block) {
        gettimeofday(&now, NULL);
        to.tv_sec = now.tv_sec + seconds;
        to.tv_nsec = now.tv_usec * 1000;
        while (self->data && merged_status(self) == SKDQ_EMPTY) {
            if (seconds) {
                rv = pthread_cond_timedwait(self->cond, self->mutex, &to);
                if (rv == ETIMEDOUT) {
                    return SKDQ_TIMEDOUT;
                }
            } else {
                pthread_cond_wait(self->cond, self->mutex);
            }
        }
    }

    if (self->data == NULL) {
        return SKDQ_DESTROYED;
    }

    if ((retval = merged_status(self)) != SKDQ_SUCCESS) {
        return retval;
    }

    b = 1 - f;

    retval = q->q[f]->methods.pop(q->q[f], item, 0, f, 0);
    if (retval == SKDQ_EMPTY) {
        retval = q->q[b]->methods.pop(q->q[b], item, 0, f, 0);
    }
  
    return retval;
}

static skDQErr_t merged_peek(skDeque_t self, void **item, uint8_t f) 
{
    merged_dq_t *q = (merged_dq_t *)self->data;
    skDQErr_t retval;
    uint8_t b;

    if (q == NULL) {
        return SKDQ_ERROR;
    }

    b = f - 1;

    if ((retval = q->q[f]->methods.peek(q->q[f], item, f)) == SKDQ_EMPTY) {
        retval = q->q[b]->methods.peek(q->q[b], item, f);
    }
  
    return retval;
}

static skDQErr_t merged_push(skDeque_t self, void *item, uint8_t f)
{
    merged_dq_t *q = (merged_dq_t *)self->data;

    if (q == NULL) {
        return SKDQ_ERROR;
    }

    return q->q[f]->methods.push(q->q[f], item, f);
}

static skDQErr_t merged_destroy(skDeque_t self)
{
    merged_dq_t *q = (merged_dq_t *)self->data;
    uint8_t i;

    if (q == NULL) {
        return SKDQ_ERROR;
    }

    for (i = 0; i <= 1; i++) {
        q->q[i]->mutex = &q->q[i]->mutex_data;
        q->q[i]->cond = &q->q[i]->cond_data;
        skDequeDestroy(q->q[i]);
    }

    free(q);

    return SKDQ_SUCCESS;
}

static skDQErr_t merged_block(skDeque_t self, uint8_t flag)
{
    merged_dq_t *q = (merged_dq_t *)self->data;
    skDQErr_t err = SKDQ_SUCCESS;
    uint8_t i;

    if (q == NULL) {
        return SKDQ_ERROR;
    }
  
    for (i = 0; i <= 1 && err == SKDQ_SUCCESS; i++) {
        err = q->q[i]->methods.block(q->q[i], flag);
    }
  
    return err;
}

static uint32_t merged_size(skDeque_t self)
{
    merged_dq_t *q = (merged_dq_t *)self->data;

    return (q->q[0]->methods.size(q->q[0]) + q->q[1]->methods.size(q->q[1]));
}