Berkeley DB Family

Hi,
This problem has bugged me for a while and I just couldn't figure out what is wrong.
I have a program which receives real-time data. The program will process the data and store the results in the Berkeley DB. The data needs to be written into BDB is averaging 20K record per second, each record is 20 bytes, ie. 400KB/sec. Everyday the program will generate a database file around 10GB. The database is designed for simple data access so that ACID is not required. Now the main problem is that BDB can hardly catch up this 20K/s speed. After the cache is used up and it starts to swap pages to harddisk, the speed is slowing down and only averaging around 10K records per second.

Currently I am using an environment with 512MB cache size and the page size is 8KB. The main reason that I use environment is to be able to use memp_trickle function, since the sync will take a rather long time.

The data is in the format of (timestamp, value). And there are thousands of different categories of data with different index. So:
The key of the database is:
struct dp_db_key{
int dp_index;
double x_value; // timestamp in msec
};
The data of the dababase is a double y_value.

I have tried to use different comparison functions to sort the data in the BDB, like sort by dp_index first, x_value second, or sort by x_value first, dp_index second. Although the second way is supposed to perform better since it follows the locality rule (since data is come in in the sequence of time), it actually turned out not better than the first method.

Is there anyway to speed up the BDB write so that it is capable to handle the data stream? Any suggestions are welcomed.


int open_env(const char env_name, DB_ENV *envpp){
DB_ENV dbenv; / Env structure handle */
u_int32_t env_flags; /* env open flags */
int ret; /* function return value */

ret = db_env_create(&dbenv, 0);
if (ret != 0) {
fprintf(stderr, "Error creating env handle: %s\n", db_strerror(ret));
return ret;
}

dbenv->set_errfile(dbenv, fd);
dbenv->set_errpfx(dbenv, "TALIESIN");

if((ret = dbenv->set_cachesize(dbenv, 0, gl_conf.db_cache_size, 0)) != 0){
dbenv->err(dbenv, ret, "set_cachesize");
goto err;
}

env_flags = DB_CREATE | DB_INIT_MPOOL | DB_THREAD; // If the environment does not exist, create it.

ret = dbenv->open(dbenv, env_name, env_flags, 0);
if(ret != 0){
fprintf(stderr, "Environment open failed: %s", db_strerror(ret));
return ret;
}

*envpp = dbenv;

return ret;

err:
(void)dbenv->close(dbenv, 0);
return (NULL);
}


The database file is opened as DB_BTREE:

int open_db(const char filename, DB *dbpp, const char dbname, DB_ENV db_env_p, DBTYPE db_type, u_int32_t db_flags, int (* comp_fct)(DB *, const DBT *, const DBT *)){
DB* dbp;
int ret;
if ((ret = db_create(&dbp, db_env_p, 0)) != 0) {
fprintf(stderr, "%s: db_create: %s\n", filename, db_strerror(ret));
return (EXIT_FAILURE);
}

if(db_flags){
if((ret = dbp->set_flags(dbp, db_flags)) != 0){
dbp->err(dbp, ret, "set_flags");
goto err1;
}
}

if(comp_fct){
dbp->set_bt_compare(dbp, comp_fct);
}

if ((ret = dbp->open(dbp, NULL, filename, dbname, db_type, DB_CREATE, 0664)) != 0) {
dbp->err(dbp, ret, "%s: open %s", filename, dbname);
goto err1;
}
*dbpp = dbp;
return (EXIT_SUCCESS);

err1:
(void)dbp->close(dbp, 0);
return (EXIT_FAILURE);
}

and the comparison function is:
int compare_dp_db_key(DB dbp, const DBT k1, const DBT *k2){
struct dp_db_key key1 = (struct dp_db_key )k1->data;
struct dp_db_key key2 = (struct dp_db_key )k2->data;

if(key1->dp_index > key2->dp_index)
return 1;
else if(key1->dp_index < key2->dp_index)
return -1;
else{
if(key1->x_value > key2->x_value)
return 1;
else if(key1->x_value == key2->x_value)
return 0;
else
return -1;
}
}

The write function:
int write_dp_point_data_to_db(double y_value, double x_value, struct data_point_list *curr){
int ret;
DB *dbp = curr->dp_list_dbp;
DBT key, data;
struct dp_db_key key_data;

//fprintf(fd, "write_dp_point_data_to_db::%s %d %d\n", curr->dp_list_name, curr->dp_list_index, t);
//fflush(fd);

if(dbp == NULL)
return (EXIT_FAILURE);

memset(&key, 0, sizeof(DBT));
memset(&data, 0, sizeof(DBT));
key_data.dp_index = curr->dp_list_index;
key_data.x_value = x_value;

key.data = &key_data;
key.size = sizeof(struct dp_db_key);
data.data = &y_value;
data.size = sizeof(y_value);
ret = dbp->put(dbp, NULL, &key, &data, DB_NOOVERWRITE);
if(ret != 0){
COleDateTime TempTime = (time_t)(hi_curr_time);
fprintf(fd, "%s::%s: x=%f, y=%f\n", TempTime.Format("%H:%M:%S"), curr->dp_list_name, x_value, y_value);
fflush(fd);
dbp->err(dbp, ret, "DB->put");
goto err1;
}

return (EXIT_SUCCESS);

err1:
return (EXIT_FAILURE);
}