/* ----------------------------------------------------------------------------
 *
 * Raima Database Manager
 *
 * Copyright (c) 2012 Raima Inc.,  All rights reserved.
 *
 * Use of this software, whether in source code format, or in executable,
 * binary object code form, is governed by the Raima LICENSE which
 * is fully described in the LICENSE.TXT file, included within this
 * distribution of files.
 *
 * --------------------------------------------------------------------------*/
#include <process.h>
#include <AclAPI.h>
#include <stdio.h>

/* rdm.h is required in all files using the RDM Database */
#include "rdm.h"

/* perf09_db.h was generated from the DDL file perf09_db.sdl. */
#include "perf09_db.h"

/* perf09_db_dbd.h was generated from the DDL file perf09_db.ddl. using ddlp with the -c option */
#include "perf09_db_dbd.h"

static int8_t quit = 0;
static int32_t rowsUpdated = 0;
static int32_t queriesExecuted = 0;

/*! Number of updates in a transaction block */
#define TRANS_BLOCK 1000

/*! Number of records to initialize the database with */
#define NUM_RECORDS 50000

/*! Context structure for the Perf09 example */
typedef struct _PERF09_CTX
{
    TFS_HANDLE      hTFS;            /*!< RDM TFS handle */
} PERF09_CTX;

/*! \file perf09.c
 * \brief Source code for the RDM Perf09 example
 */
 
/*!
 * \mainpage Perf09
 * \author Jeffrey R. Parsons, Raima Sr. Engineer
 * \date 2012
 *
 * \section Description
 * EXAMPLE - Perf09
 *
 * This simple performance example creates 50,000 rows in a simple table
 * in an on-disk database.  It then spawns three threads.  The first one
 * updates rows in the table inside transactions with 1000 updates per
 * transaction.  The second thread attempts to read all the records to 
 * sum their contents.  The third thread just periodically reports the
 * results from the other two threads.  The reading thread uses standard
 * locks.  This can be compared against the performance in Perf10 where
 * the reading thread uses Read-Only Transactions.
 *
 * \section Perf09 Database
 * The DDL (Database Definition Language) specification for the database used
 * in this example is located in the file perf09_db.ddl.
 *
 * \include perf09_db.ddl
 *
 * The schema was compiled using the RDM ddlp utility with the -c option to
 * generate an embedded database dictionary (used by the RDM d_open_ptr API).
 *
 * \code ddlp -c -d perf09_db_db.ddl \endcode
 *
 * The database dictionary information is embedded inside the application so 
 * an external DBD file stored on the file system is not required.
 *
 * \section RDM Functions Used
 * The following RDM functions are illustrated:
 *   * d_tfsinitEx/d_tfsterm
 *   * d_opentask/d_closetask
 *   * d_open_ptr/d_close
 *   * d_initialize
 *   * d_fillnew
 *   * d_keyfind
 *   * d_keynext
 *   * d_recfrst/d_recnext
 *   * d_recread
 *   * d_recwrite
 *
 * They are fully documented in http://docs.raima.com/rdm/.
 */

/*! \brief Display a message and the elapsed time to perform database operations
 *
 */
void display_elapsed_time(
    const char *msg,         /**< [in] Message to be display along with the elapsed time */
    uint64_t    start_time,  /**< [in] Start of elapsed time to display */
    uint64_t    end_time)    /**< [in] End of elapsed */
{
    uint64_t elapsed_time = end_time - start_time;
    printf("%s" UINT64_SPEC " millisecs\n", msg, elapsed_time/1000000);
}

/*! \brief Initialize the RDM runtime library for use in the perf09 example
 *
 * This function initializes the RDM Transactional File Server (TFS) to use 
 * the DIRECT-LINK TFS implementation.  It also creates the database and initializes
 * it with 50,000 records.
 *
 * @return Returns a 32-bit RDM return code (S_OKAY if successful)
 */
int32_t initialize(
	PERF09_CTX  *ctx)    /**< [in/out] Pointer to the application context */
{
    DB_TASK       *task;
    int32_t        rdm_status = S_OKAY;
    uint64_t       start_time;
    uint64_t       end_time;
    int32_t        ii;
    struct simple  simple_rec;

    /* Initialize the TFS to use the DIRECT-LINK configuration (TFS) */    
    start_time = psp_timeNanoSecs();
    rdm_status = d_tfsinitEx(NULL, NULL, &ctx->hTFS, TFS_TYPE_TFS);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error initializing the TFS (%s - %s)\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        return rdm_status;
    }

    /* Create a RDM DB_TASK */
    rdm_status = d_opentask(&task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error opening the RDM task (%s - %s)\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        return rdm_status;
    }

    /* Open the perf09_db database in exclusive mode */
    rdm_status = d_open_ptr("perf09_db", "x", &perf09_db_dbd, sizeof(perf09_db_dbd), task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error opening the perf09_db database (%s - %s)\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        return rdm_status;
    }

    /* Initialize the database (creates new database files truncating any existing data) */    
    rdm_status = d_initialize(task, CURR_DB);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error initializing the perf09_db database (%s - %s)\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        return rdm_status;
    }

    /* Display information */
    printf("\n");
    printf("Initializing with DIRECT-LINK configuration\n");
    
    printf("\tOne, simple record type\n");
    printf("\tOn disk\n");
    printf("\tAdding initialial %d records\n", NUM_RECORDS);

    /* Add the initialize records to the database */
    d_trbegin("add records", task);
    simple_rec.int_col = 0;
    for (ii=1; ii<NUM_RECORDS+1 && rdm_status == S_OKAY; ii++)
    {
        if(ii % 10000 == 0)
        {
            d_trend(task);
            d_trbegin("add records", task);
        }
        simple_rec.id = ii;
        rdm_status = d_fillnew(SIMPLE, &simple_rec, task, CURR_DB);
    }

    /* End final transactions and close db/task */
    d_trend(task);
    d_close(task);
    d_closetask(task);

    end_time = psp_timeNanoSecs();
    display_elapsed_time("Initializing:\t\t\t", start_time, end_time);
    return rdm_status;
}

/*! \brief Cleanup the RDM runtime library
 *
 * This functions terminates the STANDALONE TFS used
 * in the example. 
 *
 */
void cleanup(
    PERF09_CTX *ctx)   /**< [in] Pointer to the application context */
{
    uint64_t start_time;
    uint64_t end_time;

    start_time = psp_timeNanoSecs();

    /* Terminate the TFS */
    d_tfsterm(ctx->hTFS);

    end_time = psp_timeNanoSecs();

    printf("\n");
    display_elapsed_time("Cleanup:\t\t\t\t\t", start_time, end_time);
}

/*! \brief Thread to update rows in chunks of 1,000
 *
 * The function initializes the RDM environment and runs the update operations
 * in transaction blocks of 1,000 updates.
 *
 * @return Returns a 32-bit RDM return code (S_OKAY if successful)
 */
static uint32_t __stdcall UpdateThreadFcn(
    void *arg) /**< [in] Unused thread argument */
{
    DB_TASK       *task;
    uint32_t       ii;
    uint32_t       jj;
    int32_t        starting_id;
    int32_t        rdm_status;
    struct simple  simple_rec;

    /* Create a task for this thread, each thread needs to use a private task
     * unless there is some way to ensure that only one thread can use a task
     * at a time.  Multiple threads using the same RDM task simultaneously
     * can/will result in application crashes and database corruption
     */
    rdm_status = d_opentask(&task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error open DB_TASK in UpdateThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        _endthreadex( rdm_status );
        return rdm_status;
    }

    /* Open the database in shared mode */
    rdm_status = d_open("perf09_db", "s", task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error opening database in UpdateThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        d_closetask(task);
        
        _endthreadex( rdm_status );
        return rdm_status;
    }

    /* Loop until we get an error or are told to quit */
    for (ii = 0; !quit && rdm_status == S_OKAY; ii++)
    {
        /* Start a transaction */
        rdm_status = d_trbegin("AddRecord", task);
        if (rdm_status == S_OKAY)
        {
            starting_id = (ii % (NUM_RECORDS/TRANS_BLOCK)) * TRANS_BLOCK + 1;

            /* Lock the SIMPLE record */
            rdm_status = d_reclock(SIMPLE, "w", task, CURR_DB);
            if (rdm_status == S_OKAY)
            {
                rdm_status = d_keyfind(ID, &starting_id, task, CURR_DB);
                if(rdm_status != S_OKAY)
                {
                    fprintf(stderr, "Error looking up record in UpdateThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
                    break;
                }

                for (jj = 0; jj < TRANS_BLOCK && rdm_status == S_OKAY; jj++)
                {
                    /* Read the current simple record */
                    rdm_status = d_recread(&simple_rec, task, CURR_DB);

                    /* Increment and update the int_col field */
                    simple_rec.int_col++;
                    rdm_status = d_recwrite(&simple_rec, task, CURR_DB);

                    /* Go to the next SIMPLE record */
                    rdm_status = d_keynext(ID, task, CURR_DB);
                    if(rdm_status == S_NOTFOUND)
                    {
                        rdm_status = S_OKAY;
                        break;
                    }
                }
            }
            else
            {
                    fprintf(stderr, "Error write-locking SIMPLE record in UpdateThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
                    break;
            }

            /* End the transaction (will free out locks) */
            if(rdm_status == S_OKAY)
            {
                d_trend(task);
            }
            else
            {
                d_trabort(task);
            }
            rowsUpdated += TRANS_BLOCK;
        }
    }

    /* Close the database and task */
    d_close(task);
    d_closetask(task);
    
    _endthreadex( rdm_status );
    return rdm_status;
}

/*! \brief Thread to perform sum queries on the SIMPLE record
 *
 * The function initializes the RDM environment and runs a series of read
 * operations on the SIMPLE record
 *
 * @return Returns a 32-bit RDM return code (S_OKAY if successful)
 */
static uint32_t __stdcall ReadThreadFcn(
    void *arg)           /**< [in] Unused thread argument */
{
    DB_TASK       *task;
    int32_t        rdm_status;
    struct simple  simple_rec;
    uint32_t       int_col_sum;

    /* Create a task for this thread, each thread needs to use a private task
     * unless there is some way to ensure that only one thread can use a task
     * at a time.  Multiple threads using the same RDM task simultaneously
     * can/will result in application crashes and database corruption
     */
    rdm_status = d_opentask(&task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error open DB_TASK in ReadThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        _endthreadex( rdm_status );
        return rdm_status;
    }

    /* Open the database in shared mode */
    rdm_status = d_open("perf09_db", "s", task);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Error opening database in ReadThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
        d_closetask(task);
        
        _endthreadex( rdm_status );
        return rdm_status;
    }

    /* Loop until we get an error or are told to quit */
    while (!quit && rdm_status == S_OKAY)
    {
        int_col_sum = 0;
        rdm_status = d_reclock(SIMPLE, "r", task, CURR_DB);
        if (rdm_status == S_OKAY)
        {
            rdm_status = d_recfrst(SIMPLE, task, CURR_DB);
            while (rdm_status == S_OKAY)
            {
                /* Read the current simple record and increment our sum*/
                rdm_status = d_recread(&simple_rec, task, CURR_DB);
                int_col_sum += simple_rec.int_col;

                /* go to the next record */
                rdm_status = d_recnext(task, CURR_DB);
            }

            /* We expect to break out of the loop with the value of S_NOTFOUND */
            if(rdm_status == S_NOTFOUND)
            {
                rdm_status = S_OKAY;
            }

            d_recfree(SIMPLE, task, CURR_DB);
            queriesExecuted++;
        }
        else
        {
            fprintf(stderr, "Error read-locking SIMPLE record in ReadThreadFcn - %s: %s\n", d_errorname(rdm_status), d_errorstr(rdm_status));
            break;
        }
    }

    /* Close the database and task */
    d_close(task);
    d_closetask(task);
    
    _endthreadex( rdm_status );
    return rdm_status;
}

/*! \brief Thread to report progress
 *
 * The function reports on the progress of the update and read threads
 *
 */
static int32_t __stdcall DisplayThreadFcn(
    void *arg)
{
    uint32_t       secs;

    for (secs = 1; !quit; secs++)
    {
        Sleep(1000);

        printf("Updated %d rows (%lf/sec) and executed %d queries (%lf/sec) in %d secs\n", rowsUpdated, ((double) rowsUpdated / secs), queriesExecuted, ((double) queriesExecuted / secs), secs);
    }

    
    _endthreadex( 0 );
    return 0;
}

/*! \brief Main function for perf09 example
 *
 * The function initializes the RDM environment and runs the insert operations.
 * over a number of transaction block sizes
 *
 * @return Returns 0 on success, and 1 if there were any errors
 */
int32_t main()
{
    PERF09_CTX  ctx;
    int32_t     rdm_status;
    uint32_t    update_tid;
    uint32_t    read_tid;
    uint32_t    display_tid;
    HANDLE      hUpdateThread;
    HANDLE      hReadThread;
    HANDLE      hDisplayThread;

    /* Display the example title */
    printf("%s", RDM_EXAMPLE_DESC("Perf09"));

    memset(&ctx, 0, sizeof(PERF09_CTX));

    /* Initialize the TFS */
    rdm_status = initialize(&ctx);
    if(rdm_status != S_OKAY)
    {
        fprintf(stderr, "Application is aborting due to initialization error\n");
        cleanup(&ctx);
        return 1;
    }

    /* Start threads */
    printf ("\nStarting threads...\n");
    hUpdateThread = (HANDLE) _beginthreadex(NULL, 0, UpdateThreadFcn, NULL, 0, &update_tid);
    hReadThread  = (HANDLE) _beginthreadex(NULL, 0, ReadThreadFcn, NULL, 0, &read_tid);
    hDisplayThread   = (HANDLE) _beginthreadex(NULL, 0, DisplayThreadFcn, NULL, 0, &display_tid);
    
    printf ("Press <Enter> when you are ready to terminate them . . .\n");
    getchar ();

    /* Tell the threads to quit */    
    quit = 1;

    /* Wait for the Display thread */
    WaitForSingleObject(hDisplayThread, INFINITE);
    printf("Display thread finished\n");
    CloseHandle(hDisplayThread);

    /* Wait for the Query thread */
    WaitForSingleObject(hReadThread, INFINITE);
    printf("Read thread finished\n");
    CloseHandle(hReadThread);    

    /* Wait for the Update thread */
    WaitForSingleObject(hUpdateThread, INFINITE);
    printf("Update thread finished\n");
    CloseHandle(hUpdateThread);    

    cleanup(&ctx);

    printf("\nHit <Enter> to continue . . .");
    getchar();

	return 0;
}
