#define DEBUG_POSTGRES GNUNET_NO
-#define SELECT_IT_LOW_PRIORITY "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+#define SELECT_IT_LOW_PRIORITY "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (prio = $1 AND oid > $2) " \
"ORDER BY prio ASC,oid ASC LIMIT 1) "\
"UNION "\
- "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+ "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (prio > $1 AND oid != $2)"\
"ORDER BY prio ASC,oid ASC LIMIT 1)"\
"ORDER BY prio ASC,oid ASC LIMIT 1"
-#define SELECT_IT_NON_ANONYMOUS "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+#define SELECT_IT_NON_ANONYMOUS "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (prio = $1 AND oid < $2)"\
" AND anonLevel=0 ORDER BY prio DESC,oid DESC LIMIT 1) "\
"UNION "\
- "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+ "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (prio < $1 AND oid != $2)"\
" AND anonLevel=0 ORDER BY prio DESC,oid DESC LIMIT 1) "\
"ORDER BY prio DESC,oid DESC LIMIT 1"
-#define SELECT_IT_EXPIRATION_TIME "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+#define SELECT_IT_EXPIRATION_TIME "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (expire = $1 AND oid > $2) "\
"ORDER BY expire ASC,oid ASC LIMIT 1) "\
"UNION "\
- "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+ "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (expire > $1 AND oid != $2) " \
"ORDER BY expire ASC,oid ASC LIMIT 1)"\
"ORDER BY expire ASC,oid ASC LIMIT 1"
-#define SELECT_IT_MIGRATION_ORDER "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+#define SELECT_IT_MIGRATION_ORDER "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (expire = $1 AND oid < $2)"\
" AND expire > $3 AND type!=3"\
" ORDER BY expire DESC,oid DESC LIMIT 1) "\
"UNION "\
- "(SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "\
+ "(SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "\
"WHERE (expire < $1 AND oid != $2)" \
" AND expire > $3 AND type!=3"\
" ORDER BY expire DESC,oid DESC LIMIT 1)"\
#define BUSY_TIMEOUT GNUNET_TIME_UNIT_SECONDS
+/**
+ * Closure for 'postgres_next_request_cont'.
+ */
struct NextRequestClosure
{
+ /**
+ * Global plugin data.
+ */
struct Plugin *plugin;
+
+ /**
+ * Function to call for each matching entry.
+ */
PluginIterator iter;
+
+ /**
+ * Closure for 'iter'.
+ */
void *iter_cls;
+
+ /**
+ * Parameters for the prepared statement.
+ */
const char *paramValues[5];
+
+ /**
+ * Name of the prepared statement to run.
+ */
const char *pname;
+
+ /**
+ * Size of values pointed to by paramValues.
+ */
int paramLengths[5];
- int nparams; // nparams
- struct GNUNET_TIME_Absolute now;
+
+ /**
+ * Number of paramters in paramValues/paramLengths.
+ */
+ int nparams;
+
+ /**
+ * Current time (possible parameter), big-endian.
+ */
+ uint64_t bnow;
+
+ /**
+ * Key (possible parameter)
+ */
GNUNET_HashCode key;
+
+ /**
+ * Hash of value (possible parameter)
+ */
GNUNET_HashCode vhash;
+
+ /**
+ * Number of entries found so far
+ */
long long count;
- long long off;
- long long limit_off;
+
+ /**
+ * Offset this iteration starts at.
+ */
+ uint64_t off;
+
+ /**
+ * Current offset to use in query, big-endian.
+ */
+ uint64_t blimit_off;
+
+ /**
+ * Overall number of matching entries.
+ */
unsigned long long total;
- unsigned long long last_expire;
- unsigned int last_rowid; // last_rowid
- unsigned int last_prio;
- enum GNUNET_BLOCK_Type type;
+
+ /**
+ * Expiration value of previous result (possible parameter), big-endian.
+ */
+ uint64_t blast_expire;
+
+ /**
+ * Row ID of last result (possible paramter), big-endian.
+ */
+ uint32_t blast_rowid;
+
+ /**
+ * Priority of last result (possible parameter), big-endian.
+ */
+ uint32_t blast_prio;
+
+ /**
+ * Type of block (possible paramter), big-endian.
+ */
+ uint32_t btype;
+
+ /**
+ * Flag set to GNUNET_YES to stop iteration.
+ */
int end_it;
};
*/
GNUNET_SCHEDULER_TaskIdentifier next_task;
- unsigned long long payload;
-
- unsigned int lastSync;
-
};
* the desired status code. If not, log an error, clear the
* result and return GNUNET_SYSERR.
*
+ * @param plugin global context
+ * @param ret result to check
+ * @param expected_status expected return value
+ * @param command name of SQL command that was run
+ * @param args arguments to SQL command
+ * @param line line number for error reporting
* @return GNUNET_OK if the result is acceptable
*/
static int
/**
* Run simple SQL statement (without results).
+ *
+ * @param plugin global context
+ * @param sql statement to run
+ * @param line code line for error reporting
*/
static int
pq_exec (struct Plugin *plugin,
/**
* Prepare SQL statement.
+ *
+ * @param plugin global context
+ * @param name name for the prepared SQL statement
+ * @param sql SQL code to prepare
+ * @param nparams number of parameters in sql
+ * @param line code line for error reporting
+ * @return GNUNET_OK on success
*/
static int
pq_prepare (struct Plugin *plugin,
- const char *name, const char *sql, int nparms, int line)
+ const char *name, const char *sql, int nparams, int line)
{
PGresult *ret;
- ret = PQprepare (plugin->dbh, name, sql, nparms, NULL);
+ ret = PQprepare (plugin->dbh, name, sql, nparams, NULL);
if (GNUNET_OK !=
check_result (plugin,
ret, PGRES_COMMAND_OK, "PQprepare", sql, line))
/**
* @brief Get a database handle
+ *
+ * @param plugin global context
* @return GNUNET_OK on success, GNUNET_SYSERR on error
*/
static int
if (PQstatus (plugin->dbh) != CONNECTION_OK)
{
GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
- "postgres",
+ "datastore-postgres",
_("Unable to initialize Postgres: %s"),
PQerrorMessage (plugin->dbh));
PQfinish (plugin->dbh);
return GNUNET_SYSERR;
}
ret = PQexec (plugin->dbh,
- "CREATE TABLE gn080 ("
- " size INTEGER NOT NULL DEFAULT 0,"
+ "CREATE TABLE gn090 ("
" type INTEGER NOT NULL DEFAULT 0,"
" prio INTEGER NOT NULL DEFAULT 0,"
" anonLevel INTEGER NOT NULL DEFAULT 0,"
PG_DIAG_SQLSTATE)))))
{
check_result (plugin,
- ret, PGRES_COMMAND_OK, "CREATE TABLE", "gn080", __LINE__);
+ ret, PGRES_COMMAND_OK, "CREATE TABLE", "gn090", __LINE__);
PQfinish (plugin->dbh);
plugin->dbh = NULL;
return GNUNET_SYSERR;
if (PQresultStatus (ret) == PGRES_COMMAND_OK)
{
if ((GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_hash ON gn080 (hash)", __LINE__)) ||
+ pq_exec (plugin, "CREATE INDEX idx_hash ON gn090 (hash)", __LINE__)) ||
(GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_hash_vhash ON gn080 (hash,vhash)",
+ pq_exec (plugin, "CREATE INDEX idx_hash_vhash ON gn090 (hash,vhash)",
__LINE__))
|| (GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_prio ON gn080 (prio)", __LINE__))
+ pq_exec (plugin, "CREATE INDEX idx_prio ON gn090 (prio)", __LINE__))
|| (GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_expire ON gn080 (expire)", __LINE__))
+ pq_exec (plugin, "CREATE INDEX idx_expire ON gn090 (expire)", __LINE__))
|| (GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_comb3 ON gn080 (prio,anonLevel)",
+ pq_exec (plugin, "CREATE INDEX idx_comb3 ON gn090 (prio,anonLevel)",
__LINE__))
|| (GNUNET_OK !=
pq_exec
- (plugin, "CREATE INDEX idx_comb4 ON gn080 (prio,hash,anonLevel)",
+ (plugin, "CREATE INDEX idx_comb4 ON gn090 (prio,hash,anonLevel)",
__LINE__))
|| (GNUNET_OK !=
- pq_exec (plugin, "CREATE INDEX idx_comb7 ON gn080 (expire,hash)",
+ pq_exec (plugin, "CREATE INDEX idx_comb7 ON gn090 (expire,hash)",
__LINE__)))
{
PQclear (ret);
PQclear (ret);
#if 1
ret = PQexec (plugin->dbh,
- "ALTER TABLE gn080 ALTER value SET STORAGE EXTERNAL");
+ "ALTER TABLE gn090 ALTER value SET STORAGE EXTERNAL");
if (GNUNET_OK !=
check_result (plugin,
ret, PGRES_COMMAND_OK,
- "ALTER TABLE", "gn080", __LINE__))
+ "ALTER TABLE", "gn090", __LINE__))
{
PQfinish (plugin->dbh);
plugin->dbh = NULL;
}
PQclear (ret);
ret = PQexec (plugin->dbh,
- "ALTER TABLE gn080 ALTER hash SET STORAGE PLAIN");
+ "ALTER TABLE gn090 ALTER hash SET STORAGE PLAIN");
if (GNUNET_OK !=
check_result (plugin,
ret, PGRES_COMMAND_OK,
- "ALTER TABLE", "gn080", __LINE__))
+ "ALTER TABLE", "gn090", __LINE__))
{
PQfinish (plugin->dbh);
plugin->dbh = NULL;
}
PQclear (ret);
ret = PQexec (plugin->dbh,
- "ALTER TABLE gn080 ALTER vhash SET STORAGE PLAIN");
+ "ALTER TABLE gn090 ALTER vhash SET STORAGE PLAIN");
if (GNUNET_OK !=
check_result (plugin,
- ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn080", __LINE__))
+ ret, PGRES_COMMAND_OK, "ALTER TABLE", "gn090", __LINE__))
{
PQfinish (plugin->dbh);
plugin->dbh = NULL;
if ((GNUNET_OK !=
pq_prepare (plugin,
"getvt",
- "SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "
+ "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
"WHERE hash=$1 AND vhash=$2 AND type=$3 "
- "AND oid >= $4 ORDER BY oid ASC LIMIT 1 OFFSET $5",
+ "AND oid > $4 ORDER BY oid ASC LIMIT 1 OFFSET $5",
5,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"gett",
- "SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "
+ "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
"WHERE hash=$1 AND type=$2"
- "AND oid >= $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
+ "AND oid > $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
4,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"getv",
- "SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "
+ "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
"WHERE hash=$1 AND vhash=$2"
- "AND oid >= $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
+ "AND oid > $3 ORDER BY oid ASC LIMIT 1 OFFSET $4",
4,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"get",
- "SELECT size, type, prio, anonLevel, expire, hash, value, oid FROM gn080 "
+ "SELECT type, prio, anonLevel, expire, hash, value, oid FROM gn090 "
"WHERE hash=$1"
- "AND oid >= $2 ORDER BY oid ASC LIMIT 1 OFFSET $3",
+ "AND oid > $2 ORDER BY oid ASC LIMIT 1 OFFSET $3",
3,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"put",
- "INSERT INTO gn080 (size, type, prio, anonLevel, expire, hash, vhash, value) "
- "VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
+ "INSERT INTO gn090 (type, prio, anonLevel, expire, hash, vhash, value) "
+ "VALUES ($1, $2, $3, $4, $5, $6, $7)",
8,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"update",
- "UPDATE gn080 SET prio = prio + $1, expire = CASE WHEN expire < $2 THEN $2 ELSE expire END "
+ "UPDATE gn090 SET prio = prio + $1, expire = CASE WHEN expire < $2 THEN $2 ELSE expire END "
"WHERE oid = $3",
3,
__LINE__)) ||
(GNUNET_OK !=
pq_prepare (plugin,
"delrow",
- "DELETE FROM gn080 " "WHERE oid=$1", 1, __LINE__)))
+ "DELETE FROM gn090 " "WHERE oid=$1", 1, __LINE__)))
{
PQfinish (plugin->dbh);
plugin->dbh = NULL;
* Delete the row identified by the given rowid (qid
* in postgres).
*
+ * @param plugin global context
+ * @param rowid which row to delete
* @return GNUNET_OK on success
*/
static int
postgres_plugin_get_size (void *cls)
{
struct Plugin *plugin = cls;
- double ret;
+ unsigned long long total;
+ PGresult *ret;
- ret = plugin->payload;
- return (unsigned long long) (ret * 1.00);
- /* benchmarking shows XX% overhead */
+ ret = PQexecParams (plugin->dbh,
+ "SELECT SUM(LENGTH(value))+256*COUNT(*) FROM gn090",
+ 0, NULL, NULL, NULL, NULL, 1);
+ if (GNUNET_OK != check_result (plugin,
+ ret,
+ PGRES_TUPLES_OK,
+ "PQexecParams",
+ "get_size",
+ __LINE__))
+ {
+ return 0;
+ }
+ if ((PQntuples (ret) != 1) ||
+ (PQnfields (ret) != 1) ||
+ (PQgetlength (ret, 0, 0) != sizeof (unsigned long long)))
+ {
+ GNUNET_break (0);
+ PQclear (ret);
+ return 0;
+ }
+ total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
+ PQclear (ret);
+ return total;
}
struct Plugin *plugin = cls;
GNUNET_HashCode vhash;
PGresult *ret;
+ uint32_t btype = htonl (type);
+ uint32_t bprio = htonl (priority);
+ uint32_t banon = htonl (anonymity);
+ uint64_t bexpi = GNUNET_TIME_absolute_hton (expiration).abs_value__;
const char *paramValues[] = {
- (const char *) &size,
- (const char *) &type,
- (const char *) &priority,
- (const char *) &anonymity,
- (const char *) &expiration.value,
+ (const char *) &btype,
+ (const char *) &bprio,
+ (const char *) &banon,
+ (const char *) &bexpi,
(const char *) key,
(const char *) &vhash,
(const char *) data
};
int paramLengths[] = {
- sizeof (size),
- sizeof (type),
- sizeof (priority),
- sizeof (anonymity),
- sizeof (expiration.value),
+ sizeof (btype),
+ sizeof (bprio),
+ sizeof (banon),
+ sizeof (bexpi),
sizeof (GNUNET_HashCode),
sizeof (GNUNET_HashCode),
size
};
- const int paramFormats[] = { 1, 1, 1, 1, 1, 1, 1, 1 };
+ const int paramFormats[] = { 1, 1, 1, 1, 1, 1, 1 };
GNUNET_CRYPTO_hash (data, size, &vhash);
ret = PQexecPrepared (plugin->dbh,
- "put", 8, paramValues, paramLengths, paramFormats, 1);
+ "put", 7, paramValues, paramLengths, paramFormats, 1);
if (GNUNET_OK != check_result (plugin, ret,
PGRES_COMMAND_OK,
"PQexecPrepared", "put", __LINE__))
return GNUNET_SYSERR;
PQclear (ret);
- plugin->payload += size;
+ plugin->env->duc (plugin->env->cls, size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Stored %u bytes in database\n",
+ (unsigned int) size);
+#endif
return GNUNET_OK;
}
* asking the database plugin to call the iterator
* with the next item.
*
- * @param cls the 'struct NextRequestClosure'
+ * @param next_cls the 'struct NextRequestClosure'
* @param tc scheduler context
*/
static void
int iret;
PGresult *res;
enum GNUNET_BLOCK_Type type;
- unsigned int anonymity;
- unsigned int priority;
- unsigned int size;
+ uint32_t anonymity;
+ uint32_t priority;
+ uint32_t size;
unsigned int rowid;
struct GNUNET_TIME_Absolute expiration_time;
GNUNET_HashCode key;
plugin->next_task = GNUNET_SCHEDULER_NO_TASK;
plugin->next_task_nc = NULL;
- if (GNUNET_YES == nrc->end_it)
+ if ( (GNUNET_YES == nrc->end_it) ||
+ (nrc->count == nrc->total) )
{
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Ending iteration (%s)\n",
+ (GNUNET_YES == nrc->end_it) ? "client requested it" : "completed result set");
+#endif
nrc->iter (nrc->iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
GNUNET_free (nrc);
return;
}
-
if (nrc->count == 0)
- nrc->limit_off = nrc->off;
+ nrc->blimit_off = GNUNET_htonll (nrc->off);
else
- nrc->limit_off = 0;
+ nrc->blimit_off = GNUNET_htonll (0);
+ if (nrc->count + nrc->off == nrc->total)
+ nrc->blast_rowid = htonl (0); /* back to start */
res = PQexecPrepared (plugin->dbh,
nrc->pname,
nrc->pname,
__LINE__))
{
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Ending iteration (postgres error)\n");
+#endif
nrc->iter (nrc->iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
if (0 == PQntuples (res))
{
/* no result */
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Ending iteration (no more results)\n");
+#endif
nrc->iter (nrc->iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
return;
}
if ((1 != PQntuples (res)) ||
- (8 != PQnfields (res)) ||
- (sizeof (unsigned int) != PQfsize (res, 0)) ||
- (sizeof (unsigned int) != PQfsize (res, 7)))
+ (7 != PQnfields (res)) ||
+ (sizeof (uint32_t) != PQfsize (res, 0)) ||
+ (sizeof (uint32_t) != PQfsize (res, 6)))
{
GNUNET_break (0);
nrc->iter (nrc->iter_cls,
GNUNET_free (nrc);
return;
}
- rowid = *(unsigned int *) PQgetvalue (res, 0, 7);
- size = *(unsigned int *) PQgetvalue (res, 0, 0);
- if ((sizeof (unsigned int) != PQfsize (res, 1)) ||
- (sizeof (unsigned int) != PQfsize (res, 2)) ||
- (sizeof (unsigned int) != PQfsize (res, 3)) ||
- (sizeof (unsigned long long) != PQfsize (res, 4)) ||
- (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 5)) ||
- (size != PQgetlength (res, 0, 6)))
+ rowid = ntohl (*(uint32_t *) PQgetvalue (res, 0, 6));
+ if ((sizeof (uint32_t) != PQfsize (res, 0)) ||
+ (sizeof (uint32_t) != PQfsize (res, 1)) ||
+ (sizeof (uint32_t) != PQfsize (res, 2)) ||
+ (sizeof (uint64_t) != PQfsize (res, 3)) ||
+ (sizeof (GNUNET_HashCode) != PQgetlength (res, 0, 4)) )
{
GNUNET_break (0);
PQclear (res);
return;
}
- type = *(unsigned int *) PQgetvalue (res, 0, 1);
- priority = *(unsigned int *) PQgetvalue (res, 0, 2);
- anonymity = *(unsigned int *) PQgetvalue (res, 0, 3);
- expiration_time.value = *(unsigned long long *) PQgetvalue (res, 0, 4);
- size = PQgetlength (res, 0, 6);
- memcpy (&key, PQgetvalue (res, 0, 5), sizeof (GNUNET_HashCode));
+ type = ntohl (*(uint32_t *) PQgetvalue (res, 0, 0));
+ priority = ntohl (*(uint32_t *) PQgetvalue (res, 0, 1));
+ anonymity = ntohl ( *(uint32_t *) PQgetvalue (res, 0, 2));
+ expiration_time.abs_value = GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, 0, 3));
+ memcpy (&key, PQgetvalue (res, 0, 4), sizeof (GNUNET_HashCode));
+ size = PQgetlength (res, 0, 5);
- nrc->last_prio = priority;
- nrc->last_expire = expiration_time.value;
- nrc->last_rowid = rowid + 1;
+ nrc->blast_prio = htonl (priority);
+ nrc->blast_expire = GNUNET_htonll (expiration_time.abs_value);
+ nrc->blast_rowid = htonl (rowid);
nrc->count++;
+
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Found result of size %u bytes and type %u in database\n",
+ (unsigned int) size,
+ (unsigned int) type);
+#endif
iret = nrc->iter (nrc->iter_cls,
nrc,
&key,
size,
- PQgetvalue (res, 0, 6),
+ PQgetvalue (res, 0, 5),
(enum GNUNET_BLOCK_Type) type,
priority,
anonymity,
rowid);
PQclear (res);
if (iret == GNUNET_SYSERR)
- return;
+ {
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Ending iteration (client error)\n");
+#endif
+ return;
+ }
if (iret == GNUNET_NO)
{
- plugin->payload -= size;
- delete_by_rowid (plugin, rowid);
+ if (GNUNET_OK == delete_by_rowid (plugin, rowid))
+ {
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Deleting %u bytes from database\n",
+ (unsigned int) size);
+#endif
+ plugin->env->duc (plugin->env->cls,
+ - (size + GNUNET_DATASTORE_ENTRY_OVERHEAD));
+#if DEBUG_POSTGRES
+ GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
+ "datastore-postgres",
+ "Deleted %u bytes from database\n",
+ (unsigned int) size);
+#endif
+ }
}
}
-
/**
* Function invoked on behalf of a "PluginIterator"
* asking the database plugin to call the iterator
char **msg)
{
struct Plugin *plugin = cls;
- unsigned int oid = (unsigned int) uid; /* only 32 bit for postgres */
PGresult *ret;
-
+ int32_t bdelta = (int32_t) htonl ((uint32_t) delta);
+ uint32_t boid = htonl ( (uint32_t) uid);
+ uint64_t bexpire = GNUNET_TIME_absolute_hton (expire).abs_value__;
const char *paramValues[] = {
- (const char *) &delta,
- (const char *) &expire.value,
- (const char *) &oid,
+ (const char *) &bdelta,
+ (const char *) &bexpire,
+ (const char *) &boid,
};
int paramLengths[] = {
- sizeof (delta),
- sizeof (expire.value),
- sizeof (oid),
+ sizeof (bdelta),
+ sizeof (bexpire),
+ sizeof (boid),
};
const int paramFormats[] = { 1, 1, 1 };
* Call a method for each key in the database and
* call the callback method on it.
*
+ * @param plugin global context
* @param type entries of which type should be considered?
+ * @param is_asc ascending or descending iteration?
+ * @param iter_select which SELECT method should be used?
* @param iter maybe NULL (to just count); iter
* should return GNUNET_SYSERR to abort the
* iteration, GNUNET_NO to delete the entry and
* continue and GNUNET_OK to continue iterating
+ * @param iter_cls closure for 'iter'
*/
static void
postgres_iterate (struct Plugin *plugin,
struct NextRequestClosure *nrc;
nrc = GNUNET_malloc (sizeof (struct NextRequestClosure));
+ nrc->count = UINT32_MAX;
nrc->plugin = plugin;
nrc->iter = iter;
nrc->iter_cls = iter_cls;
if (is_asc)
{
- nrc->last_prio = 0;
- nrc->last_rowid = 0;
- nrc->last_expire = 0;
+ nrc->blast_prio = htonl (0);
+ nrc->blast_rowid = htonl (0);
+ nrc->blast_expire = htonl (0);
}
else
{
- nrc->last_prio = 0x7FFFFFFFL;
- nrc->last_rowid = 0xFFFFFFFF;
- nrc->last_expire = 0x7FFFFFFFFFFFFFFFLL;
+ nrc->blast_prio = htonl (0x7FFFFFFFL);
+ nrc->blast_rowid = htonl (0xFFFFFFFF);
+ nrc->blast_expire = GNUNET_htonll (0x7FFFFFFFFFFFFFFFLL);
}
switch (iter_select)
{
case 0:
nrc->pname = "select_low_priority";
nrc->nparams = 2;
- nrc->paramValues[0] = (const char *) &nrc->last_prio;
- nrc->paramValues[1] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[0] = sizeof (nrc->last_prio);
- nrc->paramLengths[1] = sizeof (nrc->last_rowid);
+ nrc->paramValues[0] = (const char *) &nrc->blast_prio;
+ nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[0] = sizeof (nrc->blast_prio);
+ nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
break;
case 1:
nrc->pname = "select_non_anonymous";
nrc->nparams = 2;
- nrc->paramValues[0] = (const char *) &nrc->last_prio;
- nrc->paramValues[1] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[0] = sizeof (nrc->last_prio);
- nrc->paramLengths[1] = sizeof (nrc->last_rowid);
+ nrc->paramValues[0] = (const char *) &nrc->blast_prio;
+ nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[0] = sizeof (nrc->blast_prio);
+ nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
break;
case 2:
nrc->pname = "select_expiration_time";
nrc->nparams = 2;
- nrc->paramValues[0] = (const char *) &nrc->last_expire;
- nrc->paramValues[1] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[0] = sizeof (nrc->last_expire);
- nrc->paramLengths[1] = sizeof (nrc->last_rowid);
+ nrc->paramValues[0] = (const char *) &nrc->blast_expire;
+ nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[0] = sizeof (nrc->blast_expire);
+ nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
break;
case 3:
nrc->pname = "select_migration_order";
nrc->nparams = 3;
- nrc->paramValues[0] = (const char *) &nrc->last_expire;
- nrc->paramValues[1] = (const char *) &nrc->last_rowid;
- nrc->paramValues[2] = (const char *) &nrc->now;
- nrc->paramLengths[0] = sizeof (nrc->last_expire);
- nrc->paramLengths[1] = sizeof (nrc->last_rowid);
- nrc->paramLengths[2] = sizeof (nrc->now);
+ nrc->paramValues[0] = (const char *) &nrc->blast_expire;
+ nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
+ nrc->paramValues[2] = (const char *) &nrc->bnow;
+ nrc->paramLengths[0] = sizeof (nrc->blast_expire);
+ nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
+ nrc->paramLengths[2] = sizeof (nrc->bnow);
break;
default:
GNUNET_break (0);
iter (iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
+ GNUNET_free (nrc);
return;
}
- nrc->now = GNUNET_TIME_absolute_get ();
+ nrc->bnow = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ()).abs_value__;
postgres_plugin_next_request (nrc,
GNUNET_NO);
}
nrc->vhash = *vhash;
nrc->paramValues[0] = (const char*) &nrc->key;
nrc->paramLengths[0] = sizeof (GNUNET_HashCode);
- nrc->type = type;
+ nrc->btype = htonl (type);
if (type != 0)
{
if (vhash != NULL)
{
nrc->paramValues[1] = (const char *) &nrc->vhash;
nrc->paramLengths[1] = sizeof (nrc->vhash);
- nrc->paramValues[2] = (const char *) &nrc->type;
- nrc->paramLengths[2] = sizeof (nrc->type);
- nrc->paramValues[3] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[3] = sizeof (nrc->last_rowid);
- nrc->paramValues[4] = (const char *) &nrc->limit_off;
- nrc->paramLengths[4] = sizeof (nrc->limit_off);
+ nrc->paramValues[2] = (const char *) &nrc->btype;
+ nrc->paramLengths[2] = sizeof (nrc->btype);
+ nrc->paramValues[3] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[3] = sizeof (nrc->blast_rowid);
+ nrc->paramValues[4] = (const char *) &nrc->blimit_off;
+ nrc->paramLengths[4] = sizeof (nrc->blimit_off);
nrc->nparams = 5;
nrc->pname = "getvt";
ret = PQexecParams (plugin->dbh,
- "SELECT count(*) FROM gn080 WHERE hash=$1 AND vhash=$2 AND type=$3",
+ "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2 AND type=$3",
3,
NULL,
nrc->paramValues,
}
else
{
- nrc->paramValues[1] = (const char *) &nrc->type;
- nrc->paramLengths[1] = sizeof (nrc->type);
- nrc->paramValues[2] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[2] = sizeof (nrc->last_rowid);
- nrc->paramValues[3] = (const char *) &nrc->limit_off;
- nrc->paramLengths[3] = sizeof (nrc->limit_off);
+ nrc->paramValues[1] = (const char *) &nrc->btype;
+ nrc->paramLengths[1] = sizeof (nrc->btype);
+ nrc->paramValues[2] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[2] = sizeof (nrc->blast_rowid);
+ nrc->paramValues[3] = (const char *) &nrc->blimit_off;
+ nrc->paramLengths[3] = sizeof (nrc->blimit_off);
nrc->nparams = 4;
nrc->pname = "gett";
ret = PQexecParams (plugin->dbh,
- "SELECT count(*) FROM gn080 WHERE hash=$1 AND type=$2",
+ "SELECT count(*) FROM gn090 WHERE hash=$1 AND type=$2",
2,
NULL,
nrc->paramValues,
{
nrc->paramValues[1] = (const char *) &nrc->vhash;
nrc->paramLengths[1] = sizeof (nrc->vhash);
- nrc->paramValues[2] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[2] = sizeof (nrc->last_rowid);
- nrc->paramValues[3] = (const char *) &nrc->limit_off;
- nrc->paramLengths[3] = sizeof (nrc->limit_off);
+ nrc->paramValues[2] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[2] = sizeof (nrc->blast_rowid);
+ nrc->paramValues[3] = (const char *) &nrc->blimit_off;
+ nrc->paramLengths[3] = sizeof (nrc->blimit_off);
nrc->nparams = 4;
nrc->pname = "getv";
ret = PQexecParams (plugin->dbh,
- "SELECT count(*) FROM gn080 WHERE hash=$1 AND vhash=$2",
+ "SELECT count(*) FROM gn090 WHERE hash=$1 AND vhash=$2",
2,
NULL,
nrc->paramValues,
}
else
{
- nrc->paramValues[1] = (const char *) &nrc->last_rowid;
- nrc->paramLengths[1] = sizeof (nrc->last_rowid);
- nrc->paramValues[2] = (const char *) &nrc->limit_off;
- nrc->paramLengths[2] = sizeof (nrc->limit_off);
+ nrc->paramValues[1] = (const char *) &nrc->blast_rowid;
+ nrc->paramLengths[1] = sizeof (nrc->blast_rowid);
+ nrc->paramValues[2] = (const char *) &nrc->blimit_off;
+ nrc->paramLengths[2] = sizeof (nrc->blimit_off);
nrc->nparams = 3;
nrc->pname = "get";
ret = PQexecParams (plugin->dbh,
- "SELECT count(*) FROM gn080 WHERE hash=$1",
+ "SELECT count(*) FROM gn090 WHERE hash=$1",
1,
NULL,
nrc->paramValues,
iter (iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
+ GNUNET_free (nrc);
return;
}
if ((PQntuples (ret) != 1) ||
iter (iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
+ GNUNET_free (nrc);
return;
}
nrc->total = GNUNET_ntohll (*(const unsigned long long *) PQgetvalue (ret, 0, 0));
iter (iter_cls,
NULL, NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
+ GNUNET_free (nrc);
return;
}
nrc->off = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
}
-
/**
* Select a subset of the items in the datastore and call
* the given iterator for each of them.
}
-
/**
* Select a subset of the items in the datastore and call
* the given iterator for each of them.
{
struct Plugin *plugin = cls;
- pq_exec (plugin, "DROP TABLE gn080", __LINE__);
+ pq_exec (plugin, "DROP TABLE gn090", __LINE__);
}
api->iter_all_now = &postgres_plugin_iter_all_now;
api->drop = &postgres_plugin_drop;
GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
- "postgres", _("Postgres database running\n"));
+ "datastore-postgres",
+ _("Postgres database running\n"));
return api;
}