new batch insertion code
This commit is contained in:
parent
dbfd4e252a
commit
eba2a5d90c
@ -114,6 +114,8 @@ run (void *cls)
|
|||||||
&value));
|
&value));
|
||||||
now = GNUNET_TIME_absolute_get ();
|
now = GNUNET_TIME_absolute_get ();
|
||||||
ts = GNUNET_TIME_timestamp_get ();
|
ts = GNUNET_TIME_timestamp_get ();
|
||||||
|
for (unsigned int r=0;r<10;r++)
|
||||||
|
{
|
||||||
plugin->start (plugin->cls,
|
plugin->start (plugin->cls,
|
||||||
"test_by_exchange_j");
|
"test_by_exchange_j");
|
||||||
for (unsigned int k = 0; k<batch_size; k++)
|
for (unsigned int k = 0; k<batch_size; k++)
|
||||||
@ -129,6 +131,7 @@ run (void *cls)
|
|||||||
4));
|
4));
|
||||||
}
|
}
|
||||||
plugin->commit (plugin->cls);
|
plugin->commit (plugin->cls);
|
||||||
|
}
|
||||||
duration = GNUNET_TIME_absolute_get_duration (now);
|
duration = GNUNET_TIME_absolute_get_duration (now);
|
||||||
fprintf (stdout,
|
fprintf (stdout,
|
||||||
"for a batchsize equal to %d it took %s\n",
|
"for a batchsize equal to %d it took %s\n",
|
||||||
@ -136,6 +139,7 @@ run (void *cls)
|
|||||||
GNUNET_STRINGS_relative_time_to_string (duration,
|
GNUNET_STRINGS_relative_time_to_string (duration,
|
||||||
GNUNET_NO) );
|
GNUNET_NO) );
|
||||||
}
|
}
|
||||||
|
result = 0;
|
||||||
drop:
|
drop:
|
||||||
GNUNET_break (GNUNET_OK ==
|
GNUNET_break (GNUNET_OK ==
|
||||||
plugin->drop_tables (plugin->cls));
|
plugin->drop_tables (plugin->cls));
|
||||||
|
@ -62,13 +62,12 @@ notify_on_reserve (struct PostgresClosure *pg,
|
|||||||
|
|
||||||
enum GNUNET_DB_QueryStatus
|
enum GNUNET_DB_QueryStatus
|
||||||
TEH_PG_batch_reserves_in_insert (void *cls,
|
TEH_PG_batch_reserves_in_insert (void *cls,
|
||||||
const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
|
const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
|
||||||
unsigned int reserves_length,
|
unsigned int reserves_length,
|
||||||
enum GNUNET_DB_QueryStatus *results)
|
enum GNUNET_DB_QueryStatus *results)
|
||||||
{
|
{
|
||||||
struct PostgresClosure *pg = cls;
|
struct PostgresClosure *pg = cls;
|
||||||
enum GNUNET_DB_QueryStatus qs1;
|
enum GNUNET_DB_QueryStatus qs1;
|
||||||
struct TALER_EXCHANGEDB_Reserve reserve;
|
|
||||||
struct GNUNET_TIME_Timestamp expiry;
|
struct GNUNET_TIME_Timestamp expiry;
|
||||||
struct GNUNET_TIME_Timestamp gc;
|
struct GNUNET_TIME_Timestamp gc;
|
||||||
struct TALER_PaytoHashP h_payto;
|
struct TALER_PaytoHashP h_payto;
|
||||||
@ -78,7 +77,14 @@ TEH_PG_batch_reserves_in_insert (void *cls,
|
|||||||
struct GNUNET_TIME_Timestamp reserve_expiration
|
struct GNUNET_TIME_Timestamp reserve_expiration
|
||||||
= GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
|
= GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
|
||||||
|
|
||||||
reserve.pub = reserves->reserve_pub;
|
PREPARE (pg,
|
||||||
|
"reserve_create",
|
||||||
|
"SELECT "
|
||||||
|
"out_reserve_found AS conflicted"
|
||||||
|
",transaction_duplicate"
|
||||||
|
",ruuid AS reserve_uuid"
|
||||||
|
" FROM batch_reserves_in"
|
||||||
|
" ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11);");
|
||||||
expiry = GNUNET_TIME_absolute_to_timestamp (
|
expiry = GNUNET_TIME_absolute_to_timestamp (
|
||||||
GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
|
GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
|
||||||
pg->idle_reserve_expiration_time));
|
pg->idle_reserve_expiration_time));
|
||||||
@ -95,21 +101,22 @@ TEH_PG_batch_reserves_in_insert (void *cls,
|
|||||||
time; we do this before adding the actual transaction to "reserves_in",
|
time; we do this before adding the actual transaction to "reserves_in",
|
||||||
as for a new reserve it can't be a duplicate 'add' operation, and as
|
as for a new reserve it can't be a duplicate 'add' operation, and as
|
||||||
the 'add' operation needs the reserve entry as a foreign key. */
|
the 'add' operation needs the reserve entry as a foreign key. */
|
||||||
|
for (unsigned int i=0;i<reserves_length;i++)
|
||||||
{
|
{
|
||||||
|
const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
|
||||||
struct GNUNET_PQ_QueryParam params[] = {
|
struct GNUNET_PQ_QueryParam params[] = {
|
||||||
GNUNET_PQ_query_param_auto_from_type (&reserves->reserve_pub), /*$1*/
|
GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub), /*$1*/
|
||||||
GNUNET_PQ_query_param_timestamp (&expiry), /*$4*/
|
GNUNET_PQ_query_param_timestamp (&expiry), /*$4*/
|
||||||
GNUNET_PQ_query_param_timestamp (&gc), /*$5*/
|
GNUNET_PQ_query_param_timestamp (&gc), /*$5*/
|
||||||
GNUNET_PQ_query_param_uint64 (&reserves->wire_reference), /*6*/
|
GNUNET_PQ_query_param_uint64 (&reserve->wire_reference), /*6*/
|
||||||
TALER_PQ_query_param_amount (&reserves->balance), /*7+8*/
|
TALER_PQ_query_param_amount (&reserve->balance), /*7+8*/
|
||||||
GNUNET_PQ_query_param_string (reserves->exchange_account_name), /*9*/
|
GNUNET_PQ_query_param_string (reserve->exchange_account_name), /*9*/
|
||||||
GNUNET_PQ_query_param_timestamp (&reserves->execution_time), /*10*/
|
GNUNET_PQ_query_param_timestamp (&reserve->execution_time), /*10*/
|
||||||
GNUNET_PQ_query_param_auto_from_type (&h_payto), /*11*/
|
GNUNET_PQ_query_param_auto_from_type (&h_payto), /*11*/
|
||||||
GNUNET_PQ_query_param_string (reserves->sender_account_details),/*12*/
|
GNUNET_PQ_query_param_string (reserve->sender_account_details),/*12*/
|
||||||
GNUNET_PQ_query_param_timestamp (&reserve_expiration),/*13*/
|
GNUNET_PQ_query_param_timestamp (&reserve_expiration),/*13*/
|
||||||
GNUNET_PQ_query_param_end
|
GNUNET_PQ_query_param_end
|
||||||
};
|
};
|
||||||
|
|
||||||
/* We should get all our results into results[]*/
|
/* We should get all our results into results[]*/
|
||||||
struct GNUNET_PQ_ResultSpec rs[] = {
|
struct GNUNET_PQ_ResultSpec rs[] = {
|
||||||
GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
|
GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
|
||||||
@ -121,32 +128,23 @@ TEH_PG_batch_reserves_in_insert (void *cls,
|
|||||||
GNUNET_PQ_result_spec_end
|
GNUNET_PQ_result_spec_end
|
||||||
};
|
};
|
||||||
|
|
||||||
|
TALER_payto_hash (reserve->sender_account_details,
|
||||||
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
|
&h_payto);
|
||||||
"Reserve does not exist; creating a new one\n");
|
|
||||||
/* Note: query uses 'on conflict do nothing' */
|
/* Note: query uses 'on conflict do nothing' */
|
||||||
PREPARE (pg,
|
|
||||||
"reserve_create",
|
|
||||||
"SELECT "
|
|
||||||
"out_reserve_found AS conflicted"
|
|
||||||
",transaction_duplicate"
|
|
||||||
",ruuid AS reserve_uuid"
|
|
||||||
" FROM batch_reserves_in"
|
|
||||||
" ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11);");
|
|
||||||
|
|
||||||
qs1 = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
|
qs1 = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
|
||||||
"reserve_create",
|
"reserve_create",
|
||||||
params,
|
params,
|
||||||
rs);
|
rs);
|
||||||
|
|
||||||
|
|
||||||
if (qs1 < 0)
|
if (qs1 < 0)
|
||||||
return qs1;
|
return qs1;
|
||||||
|
notify_on_reserve (pg,
|
||||||
|
&reserve->reserve_pub);
|
||||||
|
GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
|
||||||
|
results[i] = (transaction_duplicate)
|
||||||
|
? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
|
||||||
|
: GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
|
||||||
|
if ( (! conflicted) && transaction_duplicate)
|
||||||
|
TEH_PG_rollback(pg);
|
||||||
}
|
}
|
||||||
if ((int)conflicted == 0 && (int)transaction_duplicate == 1)
|
return reserves_length;
|
||||||
TEH_PG_rollback(pg);
|
|
||||||
notify_on_reserve (pg,
|
|
||||||
&reserves->reserve_pub);
|
|
||||||
|
|
||||||
return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
|
|
||||||
}
|
}
|
||||||
|
@ -7,7 +7,7 @@ BASE_URL = http://localhost/
|
|||||||
[exchangedb-postgres]
|
[exchangedb-postgres]
|
||||||
|
|
||||||
#The connection string the plugin has to use for connecting to the database
|
#The connection string the plugin has to use for connecting to the database
|
||||||
CONFIG = postgres:///talercheck
|
CONFIG = postgres://dab:test@localhost/talercheck
|
||||||
|
|
||||||
# Where are the SQL files to setup our tables?
|
# Where are the SQL files to setup our tables?
|
||||||
SQL_DIR = $DATADIR/sql/exchange/
|
SQL_DIR = $DATADIR/sql/exchange/
|
||||||
|
@ -114,8 +114,10 @@ run (void *cls)
|
|||||||
&value));
|
&value));
|
||||||
now = GNUNET_TIME_absolute_get ();
|
now = GNUNET_TIME_absolute_get ();
|
||||||
ts = GNUNET_TIME_timestamp_get ();
|
ts = GNUNET_TIME_timestamp_get ();
|
||||||
plugin->start (plugin->cls,
|
for (unsigned int r=0;r<10;r++)
|
||||||
"test_by_j");
|
{
|
||||||
|
plugin->start_read_committed (plugin->cls,
|
||||||
|
"test_by_j");
|
||||||
|
|
||||||
for (unsigned int k = 0; k<batch_size; k++)
|
for (unsigned int k = 0; k<batch_size; k++)
|
||||||
{
|
{
|
||||||
@ -127,13 +129,14 @@ run (void *cls)
|
|||||||
reserves[k].wire_reference = k;
|
reserves[k].wire_reference = k;
|
||||||
|
|
||||||
}
|
}
|
||||||
FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
|
FAILIF (batch_size !=
|
||||||
plugin->batch_reserves_in_insert (plugin->cls,
|
plugin->batch_reserves_in_insert (plugin->cls,
|
||||||
reserves,
|
reserves,
|
||||||
batch_size,
|
batch_size,
|
||||||
results));
|
results));
|
||||||
|
|
||||||
plugin->commit (plugin->cls);
|
plugin->commit (plugin->cls);
|
||||||
|
}
|
||||||
duration = GNUNET_TIME_absolute_get_duration (now);
|
duration = GNUNET_TIME_absolute_get_duration (now);
|
||||||
fprintf (stdout,
|
fprintf (stdout,
|
||||||
"for a batchsize equal to %d it took %s\n",
|
"for a batchsize equal to %d it took %s\n",
|
||||||
@ -141,6 +144,7 @@ run (void *cls)
|
|||||||
GNUNET_STRINGS_relative_time_to_string (duration,
|
GNUNET_STRINGS_relative_time_to_string (duration,
|
||||||
GNUNET_NO) );
|
GNUNET_NO) );
|
||||||
}
|
}
|
||||||
|
result = 0;
|
||||||
drop:
|
drop:
|
||||||
GNUNET_break (GNUNET_OK ==
|
GNUNET_break (GNUNET_OK ==
|
||||||
plugin->drop_tables (plugin->cls));
|
plugin->drop_tables (plugin->cls));
|
||||||
|
Loading…
Reference in New Issue
Block a user