async.c

PB 熟悉的哥们希望大家可以互相学习一下

/************************************************************
 *                                                          *
 * Copyright (c) 2001-2007 McObject LLC. All Right Reserved.*
 *                                                          *
 ************************************************************/
/*
 * This sample demonstrates asynchronous database events
 * handling. The database schema decalres three events:
 * <new> object, object <update> and object <delete>. Each
 * of the events is waited for in a separate thread (task).
 * The thread is blocked until the event happened or until
 * it has been explicitely released by calling one of the
 * release runtime functions.
 */

#include "async.h"

/* These wait time allow the thread with the appropriate handler
 * to catch up and execute the handler. In order to demonstrate
 * how the events are lost change the "waits". Some of the events
 * could be left unhandled, since the thread had no chance to run!
 */
#define WAIT_FOR_NEW      100
#define WAIT_FOR_UPDATE   100
#define WAIT_FOR_DELETE   100

const char dbName[] = "eventsDB";
const int MAP_ADDRESS = 0x20000000;

static ThrParam all_tp[NTHREADS];

void _SH_(void)
{

    char text[] = 
    {
        "\nThis sample demonstrates eXtremeDB asynchronous events.\n"
            "The schema declares <new>, <update> and <delete> events\n"
            "The appropriate event handlers are called by the runtime\n"
            "when a new object is created, deleted or updated.\n"
            "Each event-handler is executed in a separate thread, which\n"
            "is blocked waiting for the event or the explicit release via\n"
            "the mco_async_release() runtime function.\n"
    };
    char text1[] = 
    {
        "Copyright (c) 2001-2007 McObject LLC. All Right Reserved.\n\n"
    };

    printf("%s\neXtremeDB runtime version %d.%d, build %d\n%s\n\nPress Enter to start", text, MCO_COMP_VER_MAJOR,
           MCO_COMP_VER_MINOR, MCO_COMP_BUILD_NUM, text1);

    getchar();
}


/* Creates a new object. That triggers the <new_object> event
 */
int new_object(mco_db_h db, int8* id)
{
    mco_trans_h t;
    MyClass obj;
    MCO_RET rc;

    mco_trans_start(db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t);
    rc = MyClass_new(t, &obj);
    MyClass_autoid_get(&obj, id);

    if (rc)
    {
        mco_trans_rollback(t);
    }

    rc = mco_trans_commit(t);
    return rc;

}


/* Updates an existing object. That triggetrs <update> events
 */
int update_object(mco_db_h db, int8 autoid)
{

    mco_trans_h t;
    MyClass obj;
    MCO_RET rc;
    uint4 u4;
    uint1 bf = 0;
    uint1 barr = 0;

    rc = mco_trans_start(db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t);
    if (rc)
    {
        return rc;
    }

    rc = MyClass_autoid_find(t, autoid, &obj);
    if (rc)
    {
        mco_trans_rollback(t);
        return rc;
    }

    MyClass_u4_get(&obj, &u4);
    u4++;
    MyClass_u4_put(&obj, u4);

    /* bfield update*/
    MyClass_bfield_get(&obj, &bf);
    bf=~bf;
    MyClass_bfield_put(&obj, bf);

    /* bitArr update*/
    MyClass_bitArr_get(&obj,&barr);
    barr=~barr;
    MyClass_bitArr_put_range(&obj,0,5, &barr);


    rc = mco_trans_commit(t);
    return rc;
}


int del_object(mco_db_h db, int8 id)
{

    MCO_RET rc;
    mco_trans_h t;
    MyClass obj;

    mco_trans_start(db, MCO_READ_WRITE, MCO_TRANS_FOREGROUND, &t);

    rc = MyClass_autoid_find(t, id, &obj);

    if (rc)
    {
        mco_trans_rollback(t);
        return rc;
    }

    rc = MyClass_delete(&obj);
    rc = mco_trans_commit(t);
    return rc;

}


/* fatal error handler */
static void errhandler(int n)
{
    printf("\neXtremeDB runtime fatal error: %d\n", n);
    getchar();
    exit( - 1);

}

int main(void)
{

    int i;
    int8 u8 = 
    {
        0
    };
    MCO_RET rc;
    mco_db_h db = 0;
    mco_runtime_info_t info;
    void* start_mem;

    _SH_();

    /* check wheather evants are supported by the runtime */
    mco_get_runtime_info(&info);
    if (!info.mco_multithreaded)
    {
        printf("This sample requires multi-threaded runtime\n");
        exit(1);
    }

    if (info.mco_shm_supported)
    {
        start_mem = (void*)((unsigned)MAP_ADDRESS);
    }
    else
    {
        start_mem = malloc(DBSIZE);
        if (!start_mem)
        {
            printf("Couldn't allocated memory\n");
            exit(1);
        }
    };

    /* set fatal error handler */
    mco_error_set_handler(&errhandler);

    mco_runtime_start();
    rc = mco_db_open(dbName, evdb_get_dictionary(), start_mem, DBSIZE, (uint2)PAGESIZE);

    if (rc)
    {
        printf("\nerror creating database, %d", rc);
        if (!info.mco_shm_supported)
        {
            free(start_mem);
        }
        exit(1);
    }
    /* connect to the database, obtain a database handle
     */
    mco_db_connect(dbName, &db);

    /* create threads and start waiting for the database events
     */
    for (i = 0; i < NTHREADS; i++)
    {
        all_tp[i].db = db;
    }

    threads(all_tp);

    for (i = 0; i < NOBJECTS; i++)
    {
        /* create a number of new objects
         */
        rc = new_object(db, &u8);
        if (rc)
        {
            printf("Error creating object\n");
            break;
        }

        /* allow the handler to run and handnle the event. Events are lost
         * (are unhandled) if they happend or released when are not "waited for"
         * or by a thread, or the waiting thread has no chance to run. Putting the
         * current thread to  sleep whould probably allow the handler to execute.
         */
        Sleep(WAIT_FOR_NEW);

        rc = update_object(db, u8);
        if (rc)
        {
            printf("Error updating object %d\n", rc);
            break;
        }

        Sleep(WAIT_FOR_UPDATE);
        rc = del_object(db, u8);
        if (rc)
        {
            printf("Error deleteing object\n");
            break;
        }
        printf("\n");
        Sleep(WAIT_FOR_DELETE);
    }

    /* release all the database events. This will cause the appropriate
    wait function to return MCO_S_EVENT_RELEASED */
    mco_async_event_release_all(db);

    Sleep(100);

    /* wait till all the threads are finished
     */

    for (;;)
    {
        int n, done = 1;

        for (n = 0; n < (NTHREADS); n++)
        {
            if (!all_tp[n].finished)
            {
                done = 0;
                break;
            }
        }
        if (done)
        {
            break;
        }
        Sleep(100);
    }

    printf("\nNumber of <new> events = %d\n", (int)ev_new_count);
    printf("Number of <update> events = %d\n", (int)ev_update_count);
    printf("Number of <deltete> events = %d\n", (int)ev_del_count);

    /* disconnect from the database, db is no longer valid */
    mco_db_disconnect(db);
    /* destroys the db instance */
    mco_db_close(dbName);

    mco_runtime_stop();

    if (!info.mco_shm_supported)
    {
        free(start_mem);
    }

    printf("\n\nPress any key to exit");
    getchar();

    PROG_EXIT(0);
}