async.c

eXtremeDB数据库在Linux平台下的使用源码。内含多个示例程序

/************************************************************
 *                                                          *
 * 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;

	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);

	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);
}