src/datastore/plugin_datastore_mysql.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file datastore/plugin_datastore_mysql.c
  23  * @brief mysql-based datastore backend
  24  * @author Igor Wronsky
  25  * @author Christian Grothoff
  26  *
  27  * NOTE: This db module does NOT work with mysql prior to 4.1 since
  28  * it uses prepared statements.  MySQL 5.0.46 promises to fix a bug
  29  * in MyISAM that is causing us grief.  At the time of this writing,
  30  * that version is yet to be released.  In anticipation, the code
  31  * will use MyISAM with 5.0.46 (and higher).  If you run such a
  32  * version, please run "make check" to verify that the MySQL bug
  33  * was actually fixed in your version (and if not, change the
  34  * code below to use MyISAM for gn071).
  35  *
  36  * HIGHLIGHTS
  37  *
  38  * Pros
  39  * + On up-to-date hardware where mysql can be used comfortably, this
  40  *   module will have better performance than the other db choices
  41  *   (according to our tests).
  42  * + Its often possible to recover the mysql database from internal
  43  *   inconsistencies. The other db choices do not support repair!
  44  * Cons
  45  * - Memory usage (Comment: "I have 1G and it never caused me trouble")
  46  * - Manual setup
  47  *
  48  * MANUAL SETUP INSTRUCTIONS
  49  *
  50  * 1) in /etc/gnunet.conf, set
  51  * @verbatim
  52        [datastore]
  53        DATABASE = "mysql"
  54    @endverbatim
  55  * 2) Then access mysql as root,
  56  * @verbatim
  57      $ mysql -u root -p
  58    @endverbatim
  59  *    and do the following. [You should replace $USER with the username
  60  *    that will be running the gnunetd process].
  61  * @verbatim
  62       CREATE DATABASE gnunet;
  63       GRANT select,insert,update,delete,create,alter,drop,create temporary tables
  64          ON gnunet.* TO $USER@localhost;
  65       SET PASSWORD FOR $USER@localhost=PASSWORD('$the_password_you_like');
  66       FLUSH PRIVILEGES;
  67    @endverbatim
  68  * 3) In the $HOME directory of $USER, create a ".my.cnf" file
  69  *    with the following lines
  70  * @verbatim
  71       [client]
  72       user=$USER
  73       password=$the_password_you_like
  74    @endverbatim
  75  *
  76  * Thats it. Note that .my.cnf file is a security risk unless its on
  77  * a safe partition etc. The $HOME/.my.cnf can of course be a symbolic
  78  * link. Even greater security risk can be achieved by setting no
  79  * password for $USER.  Luckily $USER has only priviledges to mess
  80  * up GNUnet's tables, nothing else (unless you give him more,
  81  * of course).<p>
  82  *
  83  * 4) Still, perhaps you should briefly try if the DB connection
  84  *    works. First, login as $USER. Then use,
  85  *
  86  * @verbatim
  87      $ mysql -u $USER -p $the_password_you_like
  88      mysql> use gnunet;
  89    @endverbatim
  90  *
  91  *    If you get the message &quot;Database changed&quot; it probably works.
  92  *
  93  *    [If you get &quot;ERROR 2002: Can't connect to local MySQL server
  94  *     through socket '/tmp/mysql.sock' (2)&quot; it may be resolvable by
  95  *     &quot;ln -s /var/run/mysqld/mysqld.sock /tmp/mysql.sock&quot;
  96  *     so there may be some additional trouble depending on your mysql setup.]
  97  *
  98  * REPAIRING TABLES
  99  *
 100  * - Its probably healthy to check your tables for inconsistencies
 101  *   every now and then.
 102  * - If you get odd SEGVs on gnunetd startup, it might be that the mysql
 103  *   databases have been corrupted.
 104  * - The tables can be verified/fixed in two ways;
 105  *   1) by running mysqlcheck -A, or
 106  *   2) by executing (inside of mysql using the GNUnet database):
 107  * @verbatim
 108      mysql> REPAIR TABLE gn090;
 109    @endverbatim
 110  *
 111  * PROBLEMS?
 112  *
 113  * If you have problems related to the mysql module, your best
 114  * friend is probably the mysql manual. The first thing to check
 115  * is that mysql is basically operational, that you can connect
 116  * to it, create tables, issue queries etc.
 117  */
 118
 119 #include "platform.h"
 120 #include "gnunet_datastore_plugin.h"
 121 #include "gnunet_util_lib.h"
 122 #include <mysql/mysql.h>
 123
 124 #define DEBUG_MYSQL GNUNET_NO
 125
 126 #define MAX_DATUM_SIZE 65536
 127
 128 /**
 129  * Maximum number of supported parameters for a prepared
 130  * statement.  Increase if needed.
 131  */
 132 #define MAX_PARAM 16
 133
 134 /**
 135  * Die with an error message that indicates
 136  * a failure of the command 'cmd' with the message given
 137  * by strerror(errno).
 138  */
 139 #define DIE_MYSQL(cmd, dbh) do { GNUNET_log(GNUNET_ERROR_TYPE_ERROR, _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, mysql_error((dbh)->dbf)); abort(); } while(0);
 140
 141 /**
 142  * Log an error message at log-level 'level' that indicates
 143  * a failure of the command 'cmd' on file 'filename'
 144  * with the message given by strerror(errno).
 145  */
 146 #define LOG_MYSQL(level, cmd, dbh) do { GNUNET_log(level, _("`%s' failed at %s:%d with error: %s\n"), cmd, __FILE__, __LINE__, mysql_error((dbh)->dbf)); } while(0);
 147
 148
 149 struct GNUNET_MysqlStatementHandle
 150 {
 151   struct GNUNET_MysqlStatementHandle *next;
 152
 153   struct GNUNET_MysqlStatementHandle *prev;
 154
 155   char *query;
 156
 157   MYSQL_STMT *statement;
 158
 159   int valid;
 160
 161 };
 162
 163
 164 /**
 165  * Context for all functions in this plugin.
 166  */
 167 struct Plugin
 168 {
 169   /**
 170    * Our execution environment.
 171    */
 172   struct GNUNET_DATASTORE_PluginEnvironment *env;
 173
 174   /**
 175    * Handle to talk to MySQL.
 176    */
 177   MYSQL *dbf;
 178
 179   /**
 180    * We keep all prepared statements in a DLL.  This is the head.
 181    */
 182   struct GNUNET_MysqlStatementHandle *shead;
 183
 184   /**
 185    * We keep all prepared statements in a DLL.  This is the tail.
 186    */
 187   struct GNUNET_MysqlStatementHandle *stail;
 188
 189   /**
 190    * Filename of "my.cnf" (msyql configuration).
 191    */
 192   char *cnffile;
 193
 194   /**
 195    * Prepared statements.
 196    */
 197 #define INSERT_ENTRY "INSERT INTO gn090 (repl,type,prio,anonLevel,expire,rvalue,hash,vhash,value) VALUES (?,?,?,?,?,?,?,?,?)"
 198   struct GNUNET_MysqlStatementHandle *insert_entry;
 199
 200 #define DELETE_ENTRY_BY_UID "DELETE FROM gn090 WHERE uid=?"
 201   struct GNUNET_MysqlStatementHandle *delete_entry_by_uid;
 202
 203 #define COUNT_ENTRY_BY_HASH "SELECT count(*) FROM gn090 FORCE INDEX (idx_hash) WHERE hash=?"
 204   struct GNUNET_MysqlStatementHandle *count_entry_by_hash;
 205
 206 #define SELECT_ENTRY_BY_HASH "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash) WHERE hash=? ORDER BY uid LIMIT 1 OFFSET ?"
 207   struct GNUNET_MysqlStatementHandle *select_entry_by_hash;
 208
 209 #define COUNT_ENTRY_BY_HASH_AND_VHASH "SELECT count(*) FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=?"
 210   struct GNUNET_MysqlStatementHandle *count_entry_by_hash_and_vhash;
 211
 212 #define SELECT_ENTRY_BY_HASH_AND_VHASH "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=? ORDER BY uid LIMIT 1 OFFSET ?"
 213   struct GNUNET_MysqlStatementHandle *select_entry_by_hash_and_vhash;
 214
 215 #define COUNT_ENTRY_BY_HASH_AND_TYPE "SELECT count(*) FROM gn090 FORCE INDEX (idx_hash_type_uid) WHERE hash=? AND type=?"
 216   struct GNUNET_MysqlStatementHandle *count_entry_by_hash_and_type;
 217
 218 #define SELECT_ENTRY_BY_HASH_AND_TYPE "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_type_uid) WHERE hash=? AND type=? ORDER BY uid LIMIT 1 OFFSET ?"
 219   struct GNUNET_MysqlStatementHandle *select_entry_by_hash_and_type;
 220
 221 #define COUNT_ENTRY_BY_HASH_VHASH_AND_TYPE "SELECT count(*) FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=? AND type=?"
 222   struct GNUNET_MysqlStatementHandle *count_entry_by_hash_vhash_and_type;
 223
 224 #define SELECT_ENTRY_BY_HASH_VHASH_AND_TYPE "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=? AND type=? ORDER BY uid ASC LIMIT 1 OFFSET ?"
 225   struct GNUNET_MysqlStatementHandle *select_entry_by_hash_vhash_and_type;
 226
 227 #define UPDATE_ENTRY "UPDATE gn090 SET prio=prio+?,expire=IF(expire>=?,expire,?) WHERE uid=?"
 228   struct GNUNET_MysqlStatementHandle *update_entry;
 229
 230 #define DEC_REPL "UPDATE gn090 SET repl=GREATEST (0, repl - 1) WHERE uid=?"
 231   struct GNUNET_MysqlStatementHandle *dec_repl;
 232
 233 #define SELECT_SIZE "SELECT SUM(BIT_LENGTH(value) DIV 8) FROM gn090"
 234   struct GNUNET_MysqlStatementHandle *get_size;
 235
 236 #define SELECT_IT_NON_ANONYMOUS "SELECT type,prio,anonLevel,expire,hash,value,uid "\
 237    "FROM gn090 FORCE INDEX (idx_anonLevel_type_rvalue) "\
 238    "WHERE anonLevel=0 AND type=? AND "\
 239    "(rvalue >= ? OR"\
 240    "  NOT EXISTS (SELECT 1 FROM gn090 FORCE INDEX (idx_anonLevel_type_rvalue) WHERE anonLevel=0 AND type=? AND rvalue>=?)) "\
 241    "ORDER BY rvalue ASC LIMIT 1"
 242   struct GNUNET_MysqlStatementHandle *zero_iter;
 243
 244 #define SELECT_IT_EXPIRATION "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_expire) WHERE expire < ? ORDER BY expire ASC LIMIT 1"
 245   struct GNUNET_MysqlStatementHandle *select_expiration;
 246
 247 #define SELECT_IT_PRIORITY "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_prio) ORDER BY prio ASC LIMIT 1"
 248   struct GNUNET_MysqlStatementHandle *select_priority;
 249
 250 #define SELECT_IT_REPLICATION "SELECT type,prio,anonLevel,expire,hash,value,uid "\
 251   "FROM gn090 FORCE INDEX (idx_repl_rvalue) "\
 252   "WHERE repl=? AND "\
 253   " (rvalue>=? OR"\
 254   "  NOT EXISTS (SELECT 1 FROM gn090 FORCE INDEX (idx_repl_rvalue) WHERE repl=? AND rvalue>=?)) "\
 255   "ORDER BY rvalue ASC "\
 256   "LIMIT 1"
 257   struct GNUNET_MysqlStatementHandle *select_replication;
 258
 259 #define SELECT_MAX_REPL "SELECT MAX(repl) FROM gn090"
 260   struct GNUNET_MysqlStatementHandle *max_repl;
 261
 262 };
 263
 264
 265 /**
 266  * Obtain the location of ".my.cnf".
 267  *
 268  * @param cfg our configuration
 269  * @return NULL on error
 270  */
 271 static char *
 272 get_my_cnf_path (const struct GNUNET_CONFIGURATION_Handle *cfg)
 273 {
 274   char *cnffile;
 275   char *home_dir;
 276   struct stat st;
 277
 278 #ifndef WINDOWS
 279   struct passwd *pw;
 280 #endif
 281   int configured;
 282
 283 #ifndef WINDOWS
 284   pw = getpwuid (getuid ());
 285   if (!pw)
 286   {
 287     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "getpwuid");
 288     return NULL;
 289   }
 290   if (GNUNET_YES ==
 291       GNUNET_CONFIGURATION_have_value (cfg, "datastore-mysql", "CONFIG"))
 292   {
 293     GNUNET_assert (GNUNET_OK ==
 294                    GNUNET_CONFIGURATION_get_value_filename (cfg,
 295                                                             "datastore-mysql",
 296                                                             "CONFIG",
 297                                                             &cnffile));
 298     configured = GNUNET_YES;
 299   }
 300   else
 301   {
 302     home_dir = GNUNET_strdup (pw->pw_dir);
 303     GNUNET_asprintf (&cnffile, "%s/.my.cnf", home_dir);
 304     GNUNET_free (home_dir);
 305     configured = GNUNET_NO;
 306   }
 307 #else
 308   home_dir = (char *) GNUNET_malloc (_MAX_PATH + 1);
 309   plibc_conv_to_win_path ("~/", home_dir);
 310   GNUNET_asprintf (&cnffile, "%s/.my.cnf", home_dir);
 311   GNUNET_free (home_dir);
 312   configured = GNUNET_NO;
 313 #endif
 314
 315   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 316               _("Trying to use file `%s' for MySQL configuration.\n"), cnffile);
 317   if ((0 != STAT (cnffile, &st)) || (0 != ACCESS (cnffile, R_OK)) ||
 318       (!S_ISREG (st.st_mode)))
 319   {
 320     if (configured == GNUNET_YES)
 321       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 322                   _("Could not access file `%s': %s\n"), cnffile,
 323                   STRERROR (errno));
 324     GNUNET_free (cnffile);
 325     return NULL;
 326   }
 327   return cnffile;
 328 }
 329
 330
 331 /**
 332  * Close database connection and all prepared statements (we got a DB
 333  * disconnect error).
 334  *
 335  * @param plugin plugin context
 336  */
 337 static int
 338 iclose (struct Plugin *plugin)
 339 {
 340   struct GNUNET_MysqlStatementHandle *s;
 341
 342   for (s = plugin->shead; s != NULL; s = s->next)
 343   {
 344     if (s->valid)
 345     {
 346       mysql_stmt_close (s->statement);
 347       s->valid = GNUNET_NO;
 348     }
 349   }
 350   if (plugin->dbf != NULL)
 351   {
 352     mysql_close (plugin->dbf);
 353     plugin->dbf = NULL;
 354   }
 355   return GNUNET_OK;
 356 }
 357
 358
 359 /**
 360  * Open the connection with the database (and initialize
 361  * our default options).
 362  *
 363  * @param plugin plugin context
 364  * @return GNUNET_OK on success
 365  */
 366 static int
 367 iopen (struct Plugin *plugin)
 368 {
 369   char *mysql_dbname;
 370   char *mysql_server;
 371   char *mysql_user;
 372   char *mysql_password;
 373   unsigned long long mysql_port;
 374   my_bool reconnect;
 375   unsigned int timeout;
 376
 377   plugin->dbf = mysql_init (NULL);
 378   if (plugin->dbf == NULL)
 379     return GNUNET_SYSERR;
 380   if (plugin->cnffile != NULL)
 381     mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_FILE, plugin->cnffile);
 382   mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_GROUP, "client");
 383   reconnect = 0;
 384   mysql_options (plugin->dbf, MYSQL_OPT_RECONNECT, &reconnect);
 385   timeout = 120;                /* in seconds */
 386   mysql_options (plugin->dbf, MYSQL_OPT_CONNECT_TIMEOUT,
 387                  (const void *) &timeout);
 388   mysql_options (plugin->dbf, MYSQL_SET_CHARSET_NAME, "UTF8");
 389   timeout = 60;                 /* in seconds */
 390   mysql_options (plugin->dbf, MYSQL_OPT_READ_TIMEOUT, (const void *) &timeout);
 391   mysql_options (plugin->dbf, MYSQL_OPT_WRITE_TIMEOUT, (const void *) &timeout);
 392   mysql_dbname = NULL;
 393   if (GNUNET_YES ==
 394       GNUNET_CONFIGURATION_have_value (plugin->env->cfg, "datastore-mysql",
 395                                        "DATABASE"))
 396     GNUNET_assert (GNUNET_OK ==
 397                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 398                                                           "datastore-mysql",
 399                                                           "DATABASE",
 400                                                           &mysql_dbname));
 401   else
 402     mysql_dbname = GNUNET_strdup ("gnunet");
 403   mysql_user = NULL;
 404   if (GNUNET_YES ==
 405       GNUNET_CONFIGURATION_have_value (plugin->env->cfg, "datastore-mysql",
 406                                        "USER"))
 407   {
 408     GNUNET_assert (GNUNET_OK ==
 409                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 410                                                           "datastore-mysql",
 411                                                           "USER", &mysql_user));
 412   }
 413   mysql_password = NULL;
 414   if (GNUNET_YES ==
 415       GNUNET_CONFIGURATION_have_value (plugin->env->cfg, "datastore-mysql",
 416                                        "PASSWORD"))
 417   {
 418     GNUNET_assert (GNUNET_OK ==
 419                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 420                                                           "datastore-mysql",
 421                                                           "PASSWORD",
 422                                                           &mysql_password));
 423   }
 424   mysql_server = NULL;
 425   if (GNUNET_YES ==
 426       GNUNET_CONFIGURATION_have_value (plugin->env->cfg, "datastore-mysql",
 427                                        "HOST"))
 428   {
 429     GNUNET_assert (GNUNET_OK ==
 430                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 431                                                           "datastore-mysql",
 432                                                           "HOST",
 433                                                           &mysql_server));
 434   }
 435   mysql_port = 0;
 436   if (GNUNET_YES ==
 437       GNUNET_CONFIGURATION_have_value (plugin->env->cfg, "datastore-mysql",
 438                                        "PORT"))
 439   {
 440     GNUNET_assert (GNUNET_OK ==
 441                    GNUNET_CONFIGURATION_get_value_number (plugin->env->cfg,
 442                                                           "datastore-mysql",
 443                                                           "PORT", &mysql_port));
 444   }
 445
 446   GNUNET_assert (mysql_dbname != NULL);
 447   mysql_real_connect (plugin->dbf, mysql_server, mysql_user, mysql_password,
 448                       mysql_dbname, (unsigned int) mysql_port, NULL,
 449                       CLIENT_IGNORE_SIGPIPE);
 450   GNUNET_free_non_null (mysql_server);
 451   GNUNET_free_non_null (mysql_user);
 452   GNUNET_free_non_null (mysql_password);
 453   GNUNET_free (mysql_dbname);
 454   if (mysql_error (plugin->dbf)[0])
 455   {
 456     LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR, "mysql_real_connect", plugin);
 457     return GNUNET_SYSERR;
 458   }
 459   return GNUNET_OK;
 460 }
 461
 462
 463 /**
 464  * Run the given MySQL statement.
 465  *
 466  * @param plugin plugin context
 467  * @param statement SQL statement to run
 468  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 469  */
 470 static int
 471 run_statement (struct Plugin *plugin, const char *statement)
 472 {
 473   if ((NULL == plugin->dbf) && (GNUNET_OK != iopen (plugin)))
 474     return GNUNET_SYSERR;
 475   mysql_query (plugin->dbf, statement);
 476   if (mysql_error (plugin->dbf)[0])
 477   {
 478     LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR, "mysql_query", plugin);
 479     iclose (plugin);
 480     return GNUNET_SYSERR;
 481   }
 482   return GNUNET_OK;
 483 }
 484
 485
 486 /**
 487  * Create a prepared statement.
 488  *
 489  * @param plugin plugin context
 490  * @param statement SQL statement text to prepare
 491  * @return NULL on error
 492  */
 493 static struct GNUNET_MysqlStatementHandle *
 494 prepared_statement_create (struct Plugin *plugin, const char *statement)
 495 {
 496   struct GNUNET_MysqlStatementHandle *ret;
 497
 498   ret = GNUNET_malloc (sizeof (struct GNUNET_MysqlStatementHandle));
 499   ret->query = GNUNET_strdup (statement);
 500   GNUNET_CONTAINER_DLL_insert (plugin->shead, plugin->stail, ret);
 501   return ret;
 502 }
 503
 504
 505 /**
 506  * Prepare a statement for running.
 507  *
 508  * @param plugin plugin context
 509  * @param ret handle to prepared statement
 510  * @return GNUNET_OK on success
 511  */
 512 static int
 513 prepare_statement (struct Plugin *plugin,
 514                    struct GNUNET_MysqlStatementHandle *ret)
 515 {
 516   if (GNUNET_YES == ret->valid)
 517     return GNUNET_OK;
 518   if ((NULL == plugin->dbf) && (GNUNET_OK != iopen (plugin)))
 519     return GNUNET_SYSERR;
 520   ret->statement = mysql_stmt_init (plugin->dbf);
 521   if (ret->statement == NULL)
 522   {
 523     iclose (plugin);
 524     return GNUNET_SYSERR;
 525   }
 526   if (mysql_stmt_prepare (ret->statement, ret->query, strlen (ret->query)))
 527   {
 528     GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR, "mysql",
 529                      _("Failed to prepare statement `%s'\n"), ret->query);
 530     LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR, "mysql_stmt_prepare", plugin);
 531     mysql_stmt_close (ret->statement);
 532     ret->statement = NULL;
 533     iclose (plugin);
 534     return GNUNET_SYSERR;
 535   }
 536   ret->valid = GNUNET_YES;
 537   return GNUNET_OK;
 538
 539 }
 540
 541
 542 /**
 543  * Bind the parameters for the given MySQL statement
 544  * and run it.
 545  *
 546  * @param plugin plugin context
 547  * @param s statement to bind and run
 548  * @param ap arguments for the binding
 549  * @return GNUNET_SYSERR on error, GNUNET_OK on success
 550  */
 551 static int
 552 init_params (struct Plugin *plugin, struct GNUNET_MysqlStatementHandle *s,
 553              va_list ap)
 554 {
 555   MYSQL_BIND qbind[MAX_PARAM];
 556   unsigned int pc;
 557   unsigned int off;
 558   enum enum_field_types ft;
 559
 560   pc = mysql_stmt_param_count (s->statement);
 561   if (pc > MAX_PARAM)
 562   {
 563     /* increase internal constant! */
 564     GNUNET_break (0);
 565     return GNUNET_SYSERR;
 566   }
 567   memset (qbind, 0, sizeof (qbind));
 568   off = 0;
 569   ft = 0;
 570   while ((pc > 0) && (-1 != (int) (ft = va_arg (ap, enum enum_field_types))))
 571   {
 572     qbind[off].buffer_type = ft;
 573     switch (ft)
 574     {
 575     case MYSQL_TYPE_FLOAT:
 576       qbind[off].buffer = va_arg (ap, float *);
 577
 578       break;
 579     case MYSQL_TYPE_LONGLONG:
 580       qbind[off].buffer = va_arg (ap, unsigned long long *);
 581       qbind[off].is_unsigned = va_arg (ap, int);
 582
 583       break;
 584     case MYSQL_TYPE_LONG:
 585       qbind[off].buffer = va_arg (ap, unsigned int *);
 586       qbind[off].is_unsigned = va_arg (ap, int);
 587
 588       break;
 589     case MYSQL_TYPE_VAR_STRING:
 590     case MYSQL_TYPE_STRING:
 591     case MYSQL_TYPE_BLOB:
 592       qbind[off].buffer = va_arg (ap, void *);
 593       qbind[off].buffer_length = va_arg (ap, unsigned long);
 594       qbind[off].length = va_arg (ap, unsigned long *);
 595
 596       break;
 597     default:
 598       /* unsupported type */
 599       GNUNET_break (0);
 600       return GNUNET_SYSERR;
 601     }
 602     pc--;
 603     off++;
 604   }
 605   if (!((pc == 0) && (-1 != (int) ft) && (va_arg (ap, int) == -1)))
 606   {
 607     GNUNET_assert (0);
 608     return GNUNET_SYSERR;
 609   }
 610   if (mysql_stmt_bind_param (s->statement, qbind))
 611   {
 612     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 613                 _("`%s' failed at %s:%d with error: %s\n"),
 614                 "mysql_stmt_bind_param", __FILE__, __LINE__,
 615                 mysql_stmt_error (s->statement));
 616     iclose (plugin);
 617     return GNUNET_SYSERR;
 618   }
 619   if (mysql_stmt_execute (s->statement))
 620   {
 621     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 622                 _("`%s' for `%s' failed at %s:%d with error: %s\n"),
 623                 "mysql_stmt_execute", s->query, __FILE__, __LINE__,
 624                 mysql_stmt_error (s->statement));
 625     iclose (plugin);
 626     return GNUNET_SYSERR;
 627   }
 628   return GNUNET_OK;
 629 }
 630
 631
 632 /**
 633  * Run a prepared SELECT statement.
 634  *
 635  * @param plugin plugin context
 636  * @param s statement to run
 637  * @param result_size number of elements in results array
 638  * @param results pointer to already initialized MYSQL_BIND
 639  *        array (of sufficient size) for passing results
 640  * @param ap pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 641  *        values (size + buffer-reference for pointers); terminated
 642  *        with "-1"
 643  * @return GNUNET_SYSERR on error, otherwise GNUNET_OK or GNUNET_NO (no result)
 644  */
 645 static int
 646 prepared_statement_run_select_va (struct Plugin *plugin,
 647                                   struct GNUNET_MysqlStatementHandle *s,
 648                                   unsigned int result_size,
 649                                   MYSQL_BIND * results, va_list ap)
 650 {
 651   int ret;
 652   unsigned int rsize;
 653
 654   if (GNUNET_OK != prepare_statement (plugin, s))
 655   {
 656     GNUNET_break (0);
 657     return GNUNET_SYSERR;
 658   }
 659   if (GNUNET_OK != init_params (plugin, s, ap))
 660   {
 661     GNUNET_break (0);
 662     return GNUNET_SYSERR;
 663   }
 664   rsize = mysql_stmt_field_count (s->statement);
 665   if (rsize > result_size)
 666   {
 667     GNUNET_break (0);
 668     return GNUNET_SYSERR;
 669   }
 670   if (mysql_stmt_bind_result (s->statement, results))
 671   {
 672     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 673                 _("`%s' failed at %s:%d with error: %s\n"),
 674                 "mysql_stmt_bind_result", __FILE__, __LINE__,
 675                 mysql_stmt_error (s->statement));
 676     iclose (plugin);
 677     return GNUNET_SYSERR;
 678   }
 679   ret = mysql_stmt_fetch (s->statement);
 680   if (ret == MYSQL_NO_DATA)
 681     return GNUNET_NO;
 682   if (ret != 0)
 683   {
 684     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 685                 _("`%s' failed at %s:%d with error: %s\n"), "mysql_stmt_fetch",
 686                 __FILE__, __LINE__, mysql_stmt_error (s->statement));
 687     iclose (plugin);
 688     return GNUNET_SYSERR;
 689   }
 690   mysql_stmt_reset (s->statement);
 691   return GNUNET_OK;
 692 }
 693
 694
 695 /**
 696  * Run a prepared SELECT statement.
 697  *
 698  * @param plugin plugin context
 699  * @param s statement to run
 700  * @param result_size number of elements in results array
 701  * @param results pointer to already initialized MYSQL_BIND
 702  *        array (of sufficient size) for passing results
 703  * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 704  *        values (size + buffer-reference for pointers); terminated
 705  *        with "-1"
 706  * @return GNUNET_SYSERR on error, otherwise
 707  *         the number of successfully affected (or queried) rows
 708  */
 709 static int
 710 prepared_statement_run_select (struct Plugin *plugin,
 711                                struct GNUNET_MysqlStatementHandle *s,
 712                                unsigned int result_size, MYSQL_BIND * results,
 713                                ...)
 714 {
 715   va_list ap;
 716   int ret;
 717
 718   va_start (ap, results);
 719   ret = prepared_statement_run_select_va (plugin, s, result_size, results, ap);
 720   va_end (ap);
 721   return ret;
 722 }
 723
 724
 725 /**
 726  * Run a prepared statement that does NOT produce results.
 727  *
 728  * @param plugin plugin context
 729  * @param s statement to run
 730  * @param insert_id NULL or address where to store the row ID of whatever
 731  *        was inserted (only for INSERT statements!)
 732  * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 733  *        values (size + buffer-reference for pointers); terminated
 734  *        with "-1"
 735  * @return GNUNET_SYSERR on error, otherwise
 736  *         the number of successfully affected rows
 737  */
 738 static int
 739 prepared_statement_run (struct Plugin *plugin,
 740                         struct GNUNET_MysqlStatementHandle *s,
 741                         unsigned long long *insert_id, ...)
 742 {
 743   va_list ap;
 744   int affected;
 745
 746   if (GNUNET_OK != prepare_statement (plugin, s))
 747     return GNUNET_SYSERR;
 748   va_start (ap, insert_id);
 749   if (GNUNET_OK != init_params (plugin, s, ap))
 750   {
 751     va_end (ap);
 752     return GNUNET_SYSERR;
 753   }
 754   va_end (ap);
 755   affected = mysql_stmt_affected_rows (s->statement);
 756   if (NULL != insert_id)
 757     *insert_id = (unsigned long long) mysql_stmt_insert_id (s->statement);
 758   mysql_stmt_reset (s->statement);
 759   return affected;
 760 }
 761
 762
 763 /**
 764  * Delete an entry from the gn090 table.
 765  *
 766  * @param plugin plugin context
 767  * @param uid unique ID of the entry to delete
 768  * @return GNUNET_OK on success, GNUNET_NO if no such value exists, GNUNET_SYSERR on error
 769  */
 770 static int
 771 do_delete_entry (struct Plugin *plugin, unsigned long long uid)
 772 {
 773   int ret;
 774
 775 #if DEBUG_MYSQL
 776   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Deleting value %llu from gn090 table\n",
 777               uid);
 778 #endif
 779   ret =
 780       prepared_statement_run (plugin, plugin->delete_entry_by_uid, NULL,
 781                               MYSQL_TYPE_LONGLONG, &uid, GNUNET_YES, -1);
 782   if (ret >= 0)
 783     return GNUNET_OK;
 784   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 785               "Deleting value %llu from gn090 table failed\n", uid);
 786   return ret;
 787 }
 788
 789
 790 /**
 791  * Get an estimate of how much space the database is
 792  * currently using.
 793  *
 794  * @param cls our "struct Plugin *"
 795  * @return number of bytes used on disk
 796  */
 797 static unsigned long long
 798 mysql_plugin_estimate_size (void *cls)
 799 {
 800   struct Plugin *plugin = cls;
 801   MYSQL_BIND cbind[1];
 802   long long total;
 803
 804   memset (cbind, 0, sizeof (cbind));
 805   total = 0;
 806   cbind[0].buffer_type = MYSQL_TYPE_LONGLONG;
 807   cbind[0].buffer = &total;
 808   cbind[0].is_unsigned = GNUNET_NO;
 809   if (GNUNET_OK !=
 810       prepared_statement_run_select (plugin, plugin->get_size, 1, cbind, -1))
 811     return 0;
 812   return total;
 813 }
 814
 815
 816 /**
 817  * Store an item in the datastore.
 818  *
 819  * @param cls closure
 820  * @param key key for the item
 821  * @param size number of bytes in data
 822  * @param data content stored
 823  * @param type type of the content
 824  * @param priority priority of the content
 825  * @param anonymity anonymity-level for the content
 826  * @param replication replication-level for the content
 827  * @param expiration expiration time for the content
 828  * @param msg set to error message
 829  * @return GNUNET_OK on success
 830  */
 831 static int
 832 mysql_plugin_put (void *cls, const GNUNET_HashCode * key, uint32_t size,
 833                   const void *data, enum GNUNET_BLOCK_Type type,
 834                   uint32_t priority, uint32_t anonymity, uint32_t replication,
 835                   struct GNUNET_TIME_Absolute expiration, char **msg)
 836 {
 837   struct Plugin *plugin = cls;
 838   unsigned int irepl = replication;
 839   unsigned int ipriority = priority;
 840   unsigned int ianonymity = anonymity;
 841   unsigned long long lexpiration = expiration.abs_value;
 842   unsigned long long lrvalue =
 843       (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
 844                                                      UINT64_MAX);
 845   unsigned long hashSize;
 846   unsigned long hashSize2;
 847   unsigned long lsize;
 848   GNUNET_HashCode vhash;
 849
 850   if (size > MAX_DATUM_SIZE)
 851   {
 852     GNUNET_break (0);
 853     return GNUNET_SYSERR;
 854   }
 855   hashSize = sizeof (GNUNET_HashCode);
 856   hashSize2 = sizeof (GNUNET_HashCode);
 857   lsize = size;
 858   GNUNET_CRYPTO_hash (data, size, &vhash);
 859   if (GNUNET_OK !=
 860       prepared_statement_run (plugin, plugin->insert_entry, NULL,
 861                               MYSQL_TYPE_LONG, &irepl, GNUNET_YES,
 862                               MYSQL_TYPE_LONG, &type, GNUNET_YES,
 863                               MYSQL_TYPE_LONG, &ipriority, GNUNET_YES,
 864                               MYSQL_TYPE_LONG, &ianonymity, GNUNET_YES,
 865                               MYSQL_TYPE_LONGLONG, &lexpiration, GNUNET_YES,
 866                               MYSQL_TYPE_LONGLONG, &lrvalue, GNUNET_YES,
 867                               MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
 868                               MYSQL_TYPE_BLOB, &vhash, hashSize2, &hashSize2,
 869                               MYSQL_TYPE_BLOB, data, lsize, &lsize, -1))
 870     return GNUNET_SYSERR;
 871 #if DEBUG_MYSQL
 872   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 873               "Inserted value `%s' with size %u into gn090 table\n",
 874               GNUNET_h2s (key), (unsigned int) size);
 875 #endif
 876   if (size > 0)
 877     plugin->env->duc (plugin->env->cls, size);
 878   return GNUNET_OK;
 879 }
 880
 881
 882 /**
 883  * Update the priority for a particular key in the datastore.  If
 884  * the expiration time in value is different than the time found in
 885  * the datastore, the higher value should be kept.  For the
 886  * anonymity level, the lower value is to be used.  The specified
 887  * priority should be added to the existing priority, ignoring the
 888  * priority in value.
 889  *
 890  * Note that it is possible for multiple values to match this put.
 891  * In that case, all of the respective values are updated.
 892  *
 893  * @param cls our "struct Plugin*"
 894  * @param uid unique identifier of the datum
 895  * @param delta by how much should the priority
 896  *     change?  If priority + delta < 0 the
 897  *     priority should be set to 0 (never go
 898  *     negative).
 899  * @param expire new expiration time should be the
 900  *     MAX of any existing expiration time and
 901  *     this value
 902  * @param msg set to error message
 903  * @return GNUNET_OK on success
 904  */
 905 static int
 906 mysql_plugin_update (void *cls, uint64_t uid, int delta,
 907                      struct GNUNET_TIME_Absolute expire, char **msg)
 908 {
 909   struct Plugin *plugin = cls;
 910   unsigned long long vkey = uid;
 911   unsigned long long lexpire = expire.abs_value;
 912   int ret;
 913
 914 #if DEBUG_MYSQL
 915   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 916               "Updating value %llu adding %d to priority and maxing exp at %llu\n",
 917               vkey, delta, lexpire);
 918 #endif
 919   ret =
 920       prepared_statement_run (plugin, plugin->update_entry, NULL,
 921                               MYSQL_TYPE_LONG, &delta, GNUNET_NO,
 922                               MYSQL_TYPE_LONGLONG, &lexpire, GNUNET_YES,
 923                               MYSQL_TYPE_LONGLONG, &lexpire, GNUNET_YES,
 924                               MYSQL_TYPE_LONGLONG, &vkey, GNUNET_YES, -1);
 925   if (ret != GNUNET_OK)
 926   {
 927     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Failed to update value %llu\n",
 928                 vkey);
 929   }
 930   return ret;
 931 }
 932
 933
 934 /**
 935  * Run the given select statement and call 'proc' on the resulting
 936  * values (which must be in particular positions).
 937  *
 938  * @param plugin the plugin handle
 939  * @param stmt select statement to run
 940  * @param proc function to call on result
 941  * @param proc_cls closure for proc
 942  * @param ... arguments to initialize stmt
 943  */
 944 static void
 945 execute_select (struct Plugin *plugin, struct GNUNET_MysqlStatementHandle *stmt,
 946                 PluginDatumProcessor proc, void *proc_cls, ...)
 947 {
 948   va_list ap;
 949   int ret;
 950   unsigned int type;
 951   unsigned int priority;
 952   unsigned int anonymity;
 953   unsigned long long exp;
 954   unsigned long hashSize;
 955   unsigned long size;
 956   unsigned long long uid;
 957   char value[GNUNET_DATASTORE_MAX_VALUE_SIZE];
 958   GNUNET_HashCode key;
 959   struct GNUNET_TIME_Absolute expiration;
 960   MYSQL_BIND rbind[7];
 961
 962   hashSize = sizeof (GNUNET_HashCode);
 963   memset (rbind, 0, sizeof (rbind));
 964   rbind[0].buffer_type = MYSQL_TYPE_LONG;
 965   rbind[0].buffer = &type;
 966   rbind[0].is_unsigned = 1;
 967   rbind[1].buffer_type = MYSQL_TYPE_LONG;
 968   rbind[1].buffer = &priority;
 969   rbind[1].is_unsigned = 1;
 970   rbind[2].buffer_type = MYSQL_TYPE_LONG;
 971   rbind[2].buffer = &anonymity;
 972   rbind[2].is_unsigned = 1;
 973   rbind[3].buffer_type = MYSQL_TYPE_LONGLONG;
 974   rbind[3].buffer = &exp;
 975   rbind[3].is_unsigned = 1;
 976   rbind[4].buffer_type = MYSQL_TYPE_BLOB;
 977   rbind[4].buffer = &key;
 978   rbind[4].buffer_length = hashSize;
 979   rbind[4].length = &hashSize;
 980   rbind[5].buffer_type = MYSQL_TYPE_BLOB;
 981   rbind[5].buffer = value;
 982   rbind[5].buffer_length = size = sizeof (value);
 983   rbind[5].length = &size;
 984   rbind[6].buffer_type = MYSQL_TYPE_LONGLONG;
 985   rbind[6].buffer = &uid;
 986   rbind[6].is_unsigned = 1;
 987
 988   va_start (ap, proc_cls);
 989   ret = prepared_statement_run_select_va (plugin, stmt, 7, rbind, ap);
 990   va_end (ap);
 991   if (ret <= 0)
 992   {
 993     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
 994     return;
 995   }
 996   GNUNET_assert (size <= sizeof (value));
 997   if ((rbind[4].buffer_length != sizeof (GNUNET_HashCode)) ||
 998       (hashSize != sizeof (GNUNET_HashCode)))
 999   {
1000     GNUNET_break (0);
1001     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1002     return;
1003   }
1004 #if DEBUG_MYSQL
1005   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1006               "Found %u-byte value under key `%s' with prio %u, anon %u, expire %llu selecting from gn090 table\n",
1007               (unsigned int) size, GNUNET_h2s (&key), priority, anonymity, exp);
1008 #endif
1009   GNUNET_assert (size < MAX_DATUM_SIZE);
1010   expiration.abs_value = exp;
1011   ret =
1012       proc (proc_cls, &key, size, value, type, priority, anonymity, expiration,
1013             uid);
1014   if (ret == GNUNET_NO)
1015   {
1016     do_delete_entry (plugin, uid);
1017     if (size != 0)
1018       plugin->env->duc (plugin->env->cls, -size);
1019   }
1020 }
1021
1022
1023
1024 /**
1025  * Get one of the results for a particular key in the datastore.
1026  *
1027  * @param cls closure
1028  * @param offset offset of the result (modulo num-results);
1029  *               specific ordering does not matter for the offset
1030  * @param key key to match, never NULL
1031  * @param vhash hash of the value, maybe NULL (to
1032  *        match all values that have the right key).
1033  *        Note that for DBlocks there is no difference
1034  *        betwen key and vhash, but for other blocks
1035  *        there may be!
1036  * @param type entries of which type are relevant?
1037  *     Use 0 for any type.
1038  * @param proc function to call on the matching value,
1039  *        with NULL for if no value matches
1040  * @param proc_cls closure for proc
1041  */
1042 static void
1043 mysql_plugin_get_key (void *cls, uint64_t offset, const GNUNET_HashCode * key,
1044                       const GNUNET_HashCode * vhash,
1045                       enum GNUNET_BLOCK_Type type, PluginDatumProcessor proc,
1046                       void *proc_cls)
1047 {
1048   struct Plugin *plugin = cls;
1049   int ret;
1050   MYSQL_BIND cbind[1];
1051   long long total;
1052   unsigned long hashSize;
1053   unsigned long hashSize2;
1054   unsigned long long off;
1055
1056   GNUNET_assert (key != NULL);
1057   GNUNET_assert (NULL != proc);
1058   hashSize = sizeof (GNUNET_HashCode);
1059   hashSize2 = sizeof (GNUNET_HashCode);
1060   memset (cbind, 0, sizeof (cbind));
1061   total = -1;
1062   cbind[0].buffer_type = MYSQL_TYPE_LONGLONG;
1063   cbind[0].buffer = &total;
1064   cbind[0].is_unsigned = GNUNET_NO;
1065   if (type != 0)
1066   {
1067     if (vhash != NULL)
1068     {
1069       ret =
1070           prepared_statement_run_select (plugin,
1071                                          plugin->
1072                                          count_entry_by_hash_vhash_and_type, 1,
1073                                          cbind, MYSQL_TYPE_BLOB, key, hashSize,
1074                                          &hashSize, MYSQL_TYPE_BLOB, vhash,
1075                                          hashSize2, &hashSize2, MYSQL_TYPE_LONG,
1076                                          &type, GNUNET_YES, -1);
1077     }
1078     else
1079     {
1080       ret =
1081           prepared_statement_run_select (plugin,
1082                                          plugin->count_entry_by_hash_and_type,
1083                                          1, cbind, MYSQL_TYPE_BLOB, key,
1084                                          hashSize, &hashSize, MYSQL_TYPE_LONG,
1085                                          &type, GNUNET_YES, -1);
1086     }
1087   }
1088   else
1089   {
1090     if (vhash != NULL)
1091     {
1092       ret =
1093           prepared_statement_run_select (plugin,
1094                                          plugin->count_entry_by_hash_and_vhash,
1095                                          1, cbind, MYSQL_TYPE_BLOB, key,
1096                                          hashSize, &hashSize, MYSQL_TYPE_BLOB,
1097                                          vhash, hashSize2, &hashSize2, -1);
1098
1099     }
1100     else
1101     {
1102       ret =
1103           prepared_statement_run_select (plugin, plugin->count_entry_by_hash, 1,
1104                                          cbind, MYSQL_TYPE_BLOB, key, hashSize,
1105                                          &hashSize, -1);
1106     }
1107   }
1108   if ((ret != GNUNET_OK) || (0 >= total))
1109   {
1110     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1111     return;
1112   }
1113   offset = offset % total;
1114   off = (unsigned long long) offset;
1115 #if DEBUG_MYSQL
1116   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1117               "Obtaining %llu/%lld result for GET `%s'\n", off, total,
1118               GNUNET_h2s (key));
1119 #endif
1120
1121   if (type != GNUNET_BLOCK_TYPE_ANY)
1122   {
1123     if (NULL != vhash)
1124     {
1125       execute_select (plugin, plugin->select_entry_by_hash_vhash_and_type, proc,
1126                       proc_cls, MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1127                       MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize,
1128                       MYSQL_TYPE_LONG, &type, GNUNET_YES, MYSQL_TYPE_LONGLONG,
1129                       &off, GNUNET_YES, -1);
1130     }
1131     else
1132     {
1133       execute_select (plugin, plugin->select_entry_by_hash_and_type, proc,
1134                       proc_cls, MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1135                       MYSQL_TYPE_LONG, &type, GNUNET_YES, MYSQL_TYPE_LONGLONG,
1136                       &off, GNUNET_YES, -1);
1137     }
1138   }
1139   else
1140   {
1141     if (NULL != vhash)
1142     {
1143       execute_select (plugin, plugin->select_entry_by_hash_and_vhash, proc,
1144                       proc_cls, MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1145                       MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize,
1146                       MYSQL_TYPE_LONGLONG, &off, GNUNET_YES, -1);
1147     }
1148     else
1149     {
1150       execute_select (plugin, plugin->select_entry_by_hash, proc, proc_cls,
1151                       MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1152                       MYSQL_TYPE_LONGLONG, &off, GNUNET_YES, -1);
1153     }
1154   }
1155 }
1156
1157
1158 /**
1159  * Get a zero-anonymity datum from the datastore.
1160  *
1161  * @param cls our "struct Plugin*"
1162  * @param offset offset of the result
1163  * @param type entries of which type should be considered?
1164  *        Use 0 for any type.
1165  * @param proc function to call on a matching value or NULL
1166  * @param proc_cls closure for iter
1167  */
1168 static void
1169 mysql_plugin_get_zero_anonymity (void *cls, uint64_t offset,
1170                                  enum GNUNET_BLOCK_Type type,
1171                                  PluginDatumProcessor proc, void *proc_cls)
1172 {
1173   struct Plugin *plugin = cls;
1174   unsigned long long rvalue =
1175       (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1176                                                      UINT64_MAX);
1177
1178   execute_select (plugin, plugin->zero_iter, proc, proc_cls, MYSQL_TYPE_LONG,
1179                   &type, GNUNET_YES, MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES,
1180                   MYSQL_TYPE_LONG, &type, GNUNET_YES, MYSQL_TYPE_LONGLONG,
1181                   &rvalue, GNUNET_YES, -1);
1182 }
1183
1184
1185 /**
1186  * Context for 'repl_proc' function.
1187  */
1188 struct ReplCtx
1189 {
1190
1191   /**
1192    * Plugin handle.
1193    */
1194   struct Plugin *plugin;
1195
1196   /**
1197    * Function to call for the result (or the NULL).
1198    */
1199   PluginDatumProcessor proc;
1200
1201   /**
1202    * Closure for proc.
1203    */
1204   void *proc_cls;
1205 };
1206
1207
1208 /**
1209  * Wrapper for the processor for 'mysql_plugin_get_replication'.
1210  * Decrements the replication counter and calls the original
1211  * iterator.
1212  *
1213  * @param cls closure
1214  * @param key key for the content
1215  * @param size number of bytes in data
1216  * @param data content stored
1217  * @param type type of the content
1218  * @param priority priority of the content
1219  * @param anonymity anonymity-level for the content
1220  * @param expiration expiration time for the content
1221  * @param uid unique identifier for the datum;
1222  *        maybe 0 if no unique identifier is available
1223  *
1224  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
1225  *         (continue on call to "next", of course),
1226  *         GNUNET_NO to delete the item and continue (if supported)
1227  */
1228 static int
1229 repl_proc (void *cls, const GNUNET_HashCode * key, uint32_t size,
1230            const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority,
1231            uint32_t anonymity, struct GNUNET_TIME_Absolute expiration,
1232            uint64_t uid)
1233 {
1234   struct ReplCtx *rc = cls;
1235   struct Plugin *plugin = rc->plugin;
1236   unsigned long long oid;
1237   int ret;
1238   int iret;
1239
1240   ret =
1241       rc->proc (rc->proc_cls, key, size, data, type, priority, anonymity,
1242                 expiration, uid);
1243   if (NULL != key)
1244   {
1245     oid = (unsigned long long) uid;
1246     iret =
1247         prepared_statement_run (plugin, plugin->dec_repl, NULL,
1248                                 MYSQL_TYPE_LONGLONG, &oid, GNUNET_YES, -1);
1249     if (iret == GNUNET_SYSERR)
1250     {
1251       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1252                   "Failed to reduce replication counter\n");
1253       return GNUNET_SYSERR;
1254     }
1255   }
1256   return ret;
1257 }
1258
1259
1260 /**
1261  * Get a random item for replication.  Returns a single, not expired,
1262  * random item from those with the highest replication counters.  The
1263  * item's replication counter is decremented by one IF it was positive
1264  * before.  Call 'proc' with all values ZERO or NULL if the datastore
1265  * is empty.
1266  *
1267  * @param cls closure
1268  * @param proc function to call the value (once only).
1269  * @param proc_cls closure for proc
1270  */
1271 static void
1272 mysql_plugin_get_replication (void *cls, PluginDatumProcessor proc,
1273                               void *proc_cls)
1274 {
1275   struct Plugin *plugin = cls;
1276   struct ReplCtx rc;
1277   unsigned long long rvalue;
1278   unsigned long repl;
1279   MYSQL_BIND results;
1280
1281   rc.plugin = plugin;
1282   rc.proc = proc;
1283   rc.proc_cls = proc_cls;
1284   memset (&results, 0, sizeof (results));
1285   results.buffer_type = MYSQL_TYPE_LONG;
1286   results.buffer = &repl;
1287   results.is_unsigned = GNUNET_YES;
1288
1289   if (1 !=
1290       prepared_statement_run_select (plugin, plugin->max_repl, 1, &results, -1))
1291   {
1292     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
1293     return;
1294   }
1295
1296   rvalue =
1297       (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1298                                                      UINT64_MAX);
1299   execute_select (plugin, plugin->select_replication, &repl_proc, &rc,
1300                   MYSQL_TYPE_LONG, &repl, GNUNET_YES, MYSQL_TYPE_LONGLONG,
1301                   &rvalue, GNUNET_YES, MYSQL_TYPE_LONG, &repl, GNUNET_YES,
1302                   MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES, -1);
1303
1304 }
1305
1306
1307 /**
1308  * Context for 'expi_proc' function.
1309  */
1310 struct ExpiCtx
1311 {
1312
1313   /**
1314    * Plugin handle.
1315    */
1316   struct Plugin *plugin;
1317
1318   /**
1319    * Function to call for the result (or the NULL).
1320    */
1321   PluginDatumProcessor proc;
1322
1323   /**
1324    * Closure for proc.
1325    */
1326   void *proc_cls;
1327 };
1328
1329
1330
1331 /**
1332  * Wrapper for the processor for 'mysql_plugin_get_expiration'.
1333  * If no expired value was found, we do a second query for
1334  * low-priority content.
1335  *
1336  * @param cls closure
1337  * @param key key for the content
1338  * @param size number of bytes in data
1339  * @param data content stored
1340  * @param type type of the content
1341  * @param priority priority of the content
1342  * @param anonymity anonymity-level for the content
1343  * @param expiration expiration time for the content
1344  * @param uid unique identifier for the datum;
1345  *        maybe 0 if no unique identifier is available
1346  *
1347  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
1348  *         (continue on call to "next", of course),
1349  *         GNUNET_NO to delete the item and continue (if supported)
1350  */
1351 static int
1352 expi_proc (void *cls, const GNUNET_HashCode * key, uint32_t size,
1353            const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority,
1354            uint32_t anonymity, struct GNUNET_TIME_Absolute expiration,
1355            uint64_t uid)
1356 {
1357   struct ExpiCtx *rc = cls;
1358   struct Plugin *plugin = rc->plugin;
1359
1360   if (NULL == key)
1361   {
1362     execute_select (plugin, plugin->select_priority, rc->proc, rc->proc_cls,
1363                     -1);
1364     return GNUNET_SYSERR;
1365   }
1366   return rc->proc (rc->proc_cls, key, size, data, type, priority, anonymity,
1367                    expiration, uid);
1368 }
1369
1370
1371 /**
1372  * Get a random item for expiration.
1373  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
1374  *
1375  * @param cls closure
1376  * @param proc function to call the value (once only).
1377  * @param proc_cls closure for proc
1378  */
1379 static void
1380 mysql_plugin_get_expiration (void *cls, PluginDatumProcessor proc,
1381                              void *proc_cls)
1382 {
1383   struct Plugin *plugin = cls;
1384   long long nt;
1385   struct ExpiCtx rc;
1386
1387   rc.plugin = plugin;
1388   rc.proc = proc;
1389   rc.proc_cls = proc_cls;
1390   nt = (long long) GNUNET_TIME_absolute_get ().abs_value;
1391   execute_select (plugin, plugin->select_expiration, expi_proc, &rc,
1392                   MYSQL_TYPE_LONGLONG, &nt, GNUNET_YES, -1);
1393
1394 }
1395
1396
1397 /**
1398  * Drop database.
1399  *
1400  * @param cls the "struct Plugin*"
1401  */
1402 static void
1403 mysql_plugin_drop (void *cls)
1404 {
1405   struct Plugin *plugin = cls;
1406
1407   if (GNUNET_OK != run_statement (plugin, "DROP TABLE gn090"))
1408     return;                     /* error */
1409   plugin->env->duc (plugin->env->cls, 0);
1410 }
1411
1412
1413 /**
1414  * Entry point for the plugin.
1415  *
1416  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1417  * @return our "struct Plugin*"
1418  */
1419 void *
1420 libgnunet_plugin_datastore_mysql_init (void *cls)
1421 {
1422   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1423   struct GNUNET_DATASTORE_PluginFunctions *api;
1424   struct Plugin *plugin;
1425
1426   plugin = GNUNET_malloc (sizeof (struct Plugin));
1427   plugin->env = env;
1428   plugin->cnffile = get_my_cnf_path (env->cfg);
1429   if (GNUNET_OK != iopen (plugin))
1430   {
1431     iclose (plugin);
1432     GNUNET_free_non_null (plugin->cnffile);
1433     GNUNET_free (plugin);
1434     return NULL;
1435   }
1436 #define MRUNS(a) (GNUNET_OK != run_statement (plugin, a) )
1437 #define PINIT(a,b) (NULL == (a = prepared_statement_create(plugin, b)))
1438   if (MRUNS
1439       ("CREATE TABLE IF NOT EXISTS gn090 ("
1440        " repl INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1441        " type INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1442        " prio INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1443        " anonLevel INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1444        " expire BIGINT UNSIGNED NOT NULL DEFAULT 0,"
1445        " rvalue BIGINT UNSIGNED NOT NULL,"
1446        " hash BINARY(64) NOT NULL DEFAULT '',"
1447        " vhash BINARY(64) NOT NULL DEFAULT '',"
1448        " value BLOB NOT NULL DEFAULT ''," " uid BIGINT NOT NULL AUTO_INCREMENT,"
1449        " PRIMARY KEY (uid)," " INDEX idx_hash (hash(64)),"
1450        " INDEX idx_hash_uid (hash(64),uid),"
1451        " INDEX idx_hash_vhash (hash(64),vhash(64)),"
1452        " INDEX idx_hash_type_uid (hash(64),type,rvalue),"
1453        " INDEX idx_prio (prio)," " INDEX idx_repl_rvalue (repl,rvalue),"
1454        " INDEX idx_expire (expire),"
1455        " INDEX idx_anonLevel_type_rvalue (anonLevel,type,rvalue)"
1456        ") ENGINE=InnoDB") || MRUNS ("SET AUTOCOMMIT = 1") ||
1457       PINIT (plugin->insert_entry, INSERT_ENTRY) ||
1458       PINIT (plugin->delete_entry_by_uid, DELETE_ENTRY_BY_UID) ||
1459       PINIT (plugin->select_entry_by_hash, SELECT_ENTRY_BY_HASH) ||
1460       PINIT (plugin->select_entry_by_hash_and_vhash,
1461              SELECT_ENTRY_BY_HASH_AND_VHASH) ||
1462       PINIT (plugin->select_entry_by_hash_and_type,
1463              SELECT_ENTRY_BY_HASH_AND_TYPE) ||
1464       PINIT (plugin->select_entry_by_hash_vhash_and_type,
1465              SELECT_ENTRY_BY_HASH_VHASH_AND_TYPE) ||
1466       PINIT (plugin->count_entry_by_hash, COUNT_ENTRY_BY_HASH) ||
1467       PINIT (plugin->get_size, SELECT_SIZE) ||
1468       PINIT (plugin->count_entry_by_hash_and_vhash,
1469              COUNT_ENTRY_BY_HASH_AND_VHASH) ||
1470       PINIT (plugin->count_entry_by_hash_and_type, COUNT_ENTRY_BY_HASH_AND_TYPE)
1471       || PINIT (plugin->count_entry_by_hash_vhash_and_type,
1472                 COUNT_ENTRY_BY_HASH_VHASH_AND_TYPE) ||
1473       PINIT (plugin->update_entry, UPDATE_ENTRY) ||
1474       PINIT (plugin->dec_repl, DEC_REPL) ||
1475       PINIT (plugin->zero_iter, SELECT_IT_NON_ANONYMOUS) ||
1476       PINIT (plugin->select_expiration, SELECT_IT_EXPIRATION) ||
1477       PINIT (plugin->select_priority, SELECT_IT_PRIORITY) ||
1478       PINIT (plugin->max_repl, SELECT_MAX_REPL) ||
1479       PINIT (plugin->select_replication, SELECT_IT_REPLICATION))
1480   {
1481     iclose (plugin);
1482     GNUNET_free_non_null (plugin->cnffile);
1483     GNUNET_free (plugin);
1484     return NULL;
1485   }
1486 #undef PINIT
1487 #undef MRUNS
1488
1489   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1490   api->cls = plugin;
1491   api->estimate_size = &mysql_plugin_estimate_size;
1492   api->put = &mysql_plugin_put;
1493   api->update = &mysql_plugin_update;
1494   api->get_key = &mysql_plugin_get_key;
1495   api->get_replication = &mysql_plugin_get_replication;
1496   api->get_expiration = &mysql_plugin_get_expiration;
1497   api->get_zero_anonymity = &mysql_plugin_get_zero_anonymity;
1498   api->drop = &mysql_plugin_drop;
1499   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO, "mysql",
1500                    _("Mysql database running\n"));
1501   return api;
1502 }
1503
1504
1505 /**
1506  * Exit point from the plugin.
1507  * @param cls our "struct Plugin*"
1508  * @return always NULL
1509  */
1510 void *
1511 libgnunet_plugin_datastore_mysql_done (void *cls)
1512 {
1513   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1514   struct Plugin *plugin = api->cls;
1515   struct GNUNET_MysqlStatementHandle *s;
1516
1517   iclose (plugin);
1518   while (NULL != (s = plugin->shead))
1519   {
1520     GNUNET_CONTAINER_DLL_remove (plugin->shead, plugin->stail, s);
1521     GNUNET_free (s->query);
1522     GNUNET_free (s);
1523   }
1524   GNUNET_free_non_null (plugin->cnffile);
1525   GNUNET_free (plugin);
1526   GNUNET_free (api);
1527   mysql_library_end ();
1528   return NULL;
1529 }
1530
1531 /* end of plugin_datastore_mysql.c */