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 #ifndef WINDOWS
 278   struct passwd *pw;
 279 #endif
 280   int configured;
 281
 282 #ifndef WINDOWS
 283   pw = getpwuid (getuid ());
 284   if (!pw)
 285     {
 286       GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
 287                            "getpwuid");
 288       return NULL;
 289     }
 290   if (GNUNET_YES ==
 291       GNUNET_CONFIGURATION_have_value (cfg,
 292                                        "datastore-mysql", "CONFIG"))
 293     {
 294       GNUNET_assert (GNUNET_OK ==
 295                      GNUNET_CONFIGURATION_get_value_filename (cfg,
 296                                                               "datastore-mysql", "CONFIG", &cnffile));
 297       configured = GNUNET_YES;
 298     }
 299   else
 300     {
 301       home_dir = GNUNET_strdup (pw->pw_dir);
 302 #else
 303       home_dir = (char *) GNUNET_malloc (_MAX_PATH + 1);
 304       plibc_conv_to_win_path ("~/", home_dir);
 305 #endif
 306       GNUNET_asprintf (&cnffile, "%s/.my.cnf", home_dir);
 307       GNUNET_free (home_dir);
 308       configured = GNUNET_NO;
 309     }
 310   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 311               _("Trying to use file `%s' for MySQL configuration.\n"),
 312               cnffile);
 313   if ((0 != STAT (cnffile, &st)) ||
 314       (0 != ACCESS (cnffile, R_OK)) || (!S_ISREG (st.st_mode)))
 315     {
 316       if (configured == GNUNET_YES)
 317         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 318                     _("Could not access file `%s': %s\n"), cnffile,
 319                     STRERROR (errno));
 320       GNUNET_free (cnffile);
 321       return NULL;
 322     }
 323   return cnffile;
 324 }
 325
 326
 327 /**
 328  * Close database connection and all prepared statements (we got a DB
 329  * disconnect error).
 330  *
 331  * @param plugin plugin context
 332  */
 333 static int
 334 iclose (struct Plugin *plugin)
 335 {
 336   struct GNUNET_MysqlStatementHandle *s;
 337
 338   for (s = plugin->shead; s != NULL; s = s->next)
 339     {
 340       if (s->valid)
 341         {
 342           mysql_stmt_close (s->statement);
 343           s->valid = GNUNET_NO;
 344         }
 345     }
 346   if (plugin->dbf != NULL)
 347     {
 348       mysql_close (plugin->dbf);
 349       plugin->dbf = NULL;
 350     }
 351   return GNUNET_OK;
 352 }
 353
 354
 355 /**
 356  * Open the connection with the database (and initialize
 357  * our default options).
 358  *
 359  * @param plugin plugin context
 360  * @return GNUNET_OK on success
 361  */
 362 static int
 363 iopen (struct Plugin *plugin)
 364 {
 365   char *mysql_dbname;
 366   char *mysql_server;
 367   char *mysql_user;
 368   char *mysql_password;
 369   unsigned long long mysql_port;
 370   my_bool reconnect;
 371   unsigned int timeout;
 372
 373   plugin->dbf = mysql_init (NULL);
 374   if (plugin->dbf == NULL)
 375     return GNUNET_SYSERR;
 376   if (plugin->cnffile != NULL)
 377     mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_FILE, plugin->cnffile);
 378   mysql_options (plugin->dbf, MYSQL_READ_DEFAULT_GROUP, "client");
 379   reconnect = 0;
 380   mysql_options (plugin->dbf, MYSQL_OPT_RECONNECT, &reconnect);
 381   timeout = 120; /* in seconds */
 382   mysql_options (plugin->dbf,
 383                  MYSQL_OPT_CONNECT_TIMEOUT, (const void *) &timeout);
 384   mysql_options(plugin->dbf, MYSQL_SET_CHARSET_NAME, "UTF8");
 385   timeout = 60; /* in seconds */
 386   mysql_options (plugin->dbf, MYSQL_OPT_READ_TIMEOUT, (const void *) &timeout);
 387   mysql_options (plugin->dbf, MYSQL_OPT_WRITE_TIMEOUT, (const void *) &timeout);
 388   mysql_dbname = NULL;
 389   if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
 390                                                      "datastore-mysql", "DATABASE"))
 391     GNUNET_assert (GNUNET_OK ==
 392                    GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 393                                                           "datastore-mysql", "DATABASE",
 394                                                           &mysql_dbname));
 395   else
 396     mysql_dbname = GNUNET_strdup ("gnunet");
 397   mysql_user = NULL;
 398   if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
 399                                                      "datastore-mysql", "USER"))
 400     {
 401       GNUNET_assert (GNUNET_OK ==
 402                     GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 403                                                            "datastore-mysql", "USER",
 404                                                            &mysql_user));
 405     }
 406   mysql_password = NULL;
 407   if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
 408                                                      "datastore-mysql", "PASSWORD"))
 409     {
 410       GNUNET_assert (GNUNET_OK ==
 411                     GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 412                                                            "datastore-mysql", "PASSWORD",
 413                                                            &mysql_password));
 414     }
 415   mysql_server = NULL;
 416   if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
 417                                                      "datastore-mysql", "HOST"))
 418     {
 419       GNUNET_assert (GNUNET_OK ==
 420                     GNUNET_CONFIGURATION_get_value_string (plugin->env->cfg,
 421                                                            "datastore-mysql", "HOST",
 422                                                            &mysql_server));
 423     }
 424   mysql_port = 0;
 425   if (GNUNET_YES == GNUNET_CONFIGURATION_have_value (plugin->env->cfg,
 426                                                      "datastore-mysql", "PORT"))
 427     {
 428       GNUNET_assert (GNUNET_OK ==
 429                     GNUNET_CONFIGURATION_get_value_number (plugin->env->cfg, "datastore-mysql",
 430                                                            "PORT", &mysql_port));
 431     }
 432
 433   GNUNET_assert (mysql_dbname != NULL);
 434   mysql_real_connect (plugin->dbf,
 435                       mysql_server,
 436                       mysql_user, mysql_password,
 437                       mysql_dbname,
 438                       (unsigned int) mysql_port, NULL,
 439                       CLIENT_IGNORE_SIGPIPE);
 440   GNUNET_free_non_null (mysql_server);
 441   GNUNET_free_non_null (mysql_user);
 442   GNUNET_free_non_null (mysql_password);
 443   GNUNET_free (mysql_dbname);
 444   if (mysql_error (plugin->dbf)[0])
 445     {
 446       LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR,
 447                  "mysql_real_connect", plugin);
 448       return GNUNET_SYSERR;
 449     }
 450   return GNUNET_OK;
 451 }
 452
 453
 454 /**
 455  * Run the given MySQL statement.
 456  *
 457  * @param plugin plugin context
 458  * @param statement SQL statement to run
 459  * @return GNUNET_OK on success, GNUNET_SYSERR on error
 460  */
 461 static int
 462 run_statement (struct Plugin *plugin,
 463                const char *statement)
 464 {
 465   if ((NULL == plugin->dbf) && (GNUNET_OK != iopen (plugin)))
 466     return GNUNET_SYSERR;
 467   mysql_query (plugin->dbf, statement);
 468   if (mysql_error (plugin->dbf)[0])
 469     {
 470       LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR,
 471                  "mysql_query", plugin);
 472       iclose (plugin);
 473       return GNUNET_SYSERR;
 474     }
 475   return GNUNET_OK;
 476 }
 477
 478
 479 /**
 480  * Create a prepared statement.
 481  *
 482  * @param plugin plugin context
 483  * @param statement SQL statement text to prepare
 484  * @return NULL on error
 485  */
 486 static struct GNUNET_MysqlStatementHandle *
 487 prepared_statement_create (struct Plugin *plugin,
 488                            const char *statement)
 489 {
 490   struct GNUNET_MysqlStatementHandle *ret;
 491
 492   ret = GNUNET_malloc (sizeof (struct GNUNET_MysqlStatementHandle));
 493   ret->query = GNUNET_strdup (statement);
 494   GNUNET_CONTAINER_DLL_insert (plugin->shead,
 495                                plugin->stail,
 496                                ret);
 497   return ret;
 498 }
 499
 500
 501 /**
 502  * Prepare a statement for running.
 503  *
 504  * @param plugin plugin context
 505  * @param ret handle to prepared statement
 506  * @return GNUNET_OK on success
 507  */
 508 static int
 509 prepare_statement (struct Plugin *plugin,
 510                    struct GNUNET_MysqlStatementHandle *ret)
 511 {
 512   if (GNUNET_YES == ret->valid)
 513     return GNUNET_OK;
 514   if ((NULL == plugin->dbf) &&
 515       (GNUNET_OK != iopen (plugin)))
 516     return GNUNET_SYSERR;
 517   ret->statement = mysql_stmt_init (plugin->dbf);
 518   if (ret->statement == NULL)
 519     {
 520       iclose (plugin);
 521       return GNUNET_SYSERR;
 522     }
 523   if (mysql_stmt_prepare (ret->statement,
 524                           ret->query,
 525                           strlen (ret->query)))
 526     {
 527       GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
 528                        "mysql",
 529                        _("Failed to prepare statement `%s'\n"),
 530                        ret->query);
 531       LOG_MYSQL (GNUNET_ERROR_TYPE_ERROR,
 532                  "mysql_stmt_prepare",
 533                  plugin);
 534       mysql_stmt_close (ret->statement);
 535       ret->statement = NULL;
 536       iclose (plugin);
 537       return GNUNET_SYSERR;
 538     }
 539   ret->valid = GNUNET_YES;
 540   return GNUNET_OK;
 541
 542 }
 543
 544
 545 /**
 546  * Bind the parameters for the given MySQL statement
 547  * and run it.
 548  *
 549  * @param plugin plugin context
 550  * @param s statement to bind and run
 551  * @param ap arguments for the binding
 552  * @return GNUNET_SYSERR on error, GNUNET_OK on success
 553  */
 554 static int
 555 init_params (struct Plugin *plugin,
 556              struct GNUNET_MysqlStatementHandle *s,
 557              va_list ap)
 558 {
 559   MYSQL_BIND qbind[MAX_PARAM];
 560   unsigned int pc;
 561   unsigned int off;
 562   enum enum_field_types ft;
 563
 564   pc = mysql_stmt_param_count (s->statement);
 565   if (pc > MAX_PARAM)
 566     {
 567       /* increase internal constant! */
 568       GNUNET_break (0);
 569       return GNUNET_SYSERR;
 570     }
 571   memset (qbind, 0, sizeof (qbind));
 572   off = 0;
 573   ft = 0;
 574   while ((pc > 0) && (-1 != (int) (ft = va_arg (ap, enum enum_field_types))))
 575     {
 576       qbind[off].buffer_type = ft;
 577       switch (ft)
 578         {
 579         case MYSQL_TYPE_FLOAT:
 580           qbind[off].buffer = va_arg (ap, float *);
 581           break;
 582         case MYSQL_TYPE_LONGLONG:
 583           qbind[off].buffer = va_arg (ap, unsigned long long *);
 584           qbind[off].is_unsigned = va_arg (ap, int);
 585           break;
 586         case MYSQL_TYPE_LONG:
 587           qbind[off].buffer = va_arg (ap, unsigned int *);
 588           qbind[off].is_unsigned = va_arg (ap, int);
 589           break;
 590         case MYSQL_TYPE_VAR_STRING:
 591         case MYSQL_TYPE_STRING:
 592         case MYSQL_TYPE_BLOB:
 593           qbind[off].buffer = va_arg (ap, void *);
 594           qbind[off].buffer_length = va_arg (ap, unsigned long);
 595           qbind[off].length = va_arg (ap, unsigned long *);
 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",
 615                   __FILE__, __LINE__, 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",
 624                   s->query,
 625                   __FILE__, __LINE__, mysql_stmt_error (s->statement));
 626       iclose (plugin);
 627       return GNUNET_SYSERR;
 628     }
 629   return GNUNET_OK;
 630 }
 631
 632
 633 /**
 634  * Run a prepared SELECT statement.
 635  *
 636  * @param plugin plugin context
 637  * @param s statement to run
 638  * @param result_size number of elements in results array
 639  * @param results pointer to already initialized MYSQL_BIND
 640  *        array (of sufficient size) for passing results
 641  * @param ap pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 642  *        values (size + buffer-reference for pointers); terminated
 643  *        with "-1"
 644  * @return GNUNET_SYSERR on error, otherwise GNUNET_OK or GNUNET_NO (no result)
 645  */
 646 static int
 647 prepared_statement_run_select_va (struct Plugin *plugin,
 648                                   struct GNUNET_MysqlStatementHandle *s,
 649                                   unsigned int result_size,
 650                                   MYSQL_BIND *results,
 651                                   va_list ap)
 652 {
 653   int ret;
 654   unsigned int rsize;
 655
 656   if (GNUNET_OK != prepare_statement (plugin, s))
 657     {
 658       GNUNET_break (0);
 659       return GNUNET_SYSERR;
 660     }
 661   if (GNUNET_OK != init_params (plugin, s, ap))
 662     {
 663       GNUNET_break (0);
 664       return GNUNET_SYSERR;
 665     }
 666   rsize = mysql_stmt_field_count (s->statement);
 667   if (rsize > result_size)
 668     {
 669       GNUNET_break (0);
 670       return GNUNET_SYSERR;
 671     }
 672   if (mysql_stmt_bind_result (s->statement, results))
 673     {
 674       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 675                   _("`%s' failed at %s:%d with error: %s\n"),
 676                   "mysql_stmt_bind_result",
 677                   __FILE__, __LINE__, mysql_stmt_error (s->statement));
 678       iclose (plugin);
 679       return GNUNET_SYSERR;
 680     }
 681   ret = mysql_stmt_fetch (s->statement);
 682   if (ret == MYSQL_NO_DATA)
 683     return GNUNET_NO;
 684   if (ret != 0)
 685     {
 686       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 687                   _("`%s' failed at %s:%d with error: %s\n"),
 688                   "mysql_stmt_fetch",
 689                   __FILE__, __LINE__, mysql_stmt_error (s->statement));
 690       iclose (plugin);
 691       return GNUNET_SYSERR;
 692     }
 693   mysql_stmt_reset (s->statement);
 694   return GNUNET_OK;
 695 }
 696
 697
 698 /**
 699  * Run a prepared SELECT statement.
 700  *
 701  * @param plugin plugin context
 702  * @param s statement to run
 703  * @param result_size number of elements in results array
 704  * @param results pointer to already initialized MYSQL_BIND
 705  *        array (of sufficient size) for passing results
 706  * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 707  *        values (size + buffer-reference for pointers); terminated
 708  *        with "-1"
 709  * @return GNUNET_SYSERR on error, otherwise
 710  *         the number of successfully affected (or queried) rows
 711  */
 712 static int
 713 prepared_statement_run_select (struct Plugin *plugin,
 714                                struct GNUNET_MysqlStatementHandle *s,
 715                                unsigned int result_size,
 716                                MYSQL_BIND *results,
 717                                ...)
 718 {
 719   va_list ap;
 720   int ret;
 721
 722   va_start (ap, results);
 723   ret = prepared_statement_run_select_va (plugin, s,
 724                                           result_size, results,
 725                                           ap);
 726   va_end (ap);
 727   return ret;
 728 }
 729
 730
 731 /**
 732  * Run a prepared statement that does NOT produce results.
 733  *
 734  * @param plugin plugin context
 735  * @param s statement to run
 736  * @param insert_id NULL or address where to store the row ID of whatever
 737  *        was inserted (only for INSERT statements!)
 738  * @param ... pairs and triplets of "MYSQL_TYPE_XXX" keys and their respective
 739  *        values (size + buffer-reference for pointers); terminated
 740  *        with "-1"
 741  * @return GNUNET_SYSERR on error, otherwise
 742  *         the number of successfully affected rows
 743  */
 744 static int
 745 prepared_statement_run (struct Plugin *plugin,
 746                         struct GNUNET_MysqlStatementHandle *s,
 747                         unsigned long long *insert_id, ...)
 748 {
 749   va_list ap;
 750   int affected;
 751
 752   if (GNUNET_OK != prepare_statement (plugin, s))
 753     return GNUNET_SYSERR;
 754   va_start (ap, insert_id);
 755   if (GNUNET_OK != init_params (plugin, s, ap))
 756     {
 757       va_end (ap);
 758       return GNUNET_SYSERR;
 759     }
 760   va_end (ap);
 761   affected = mysql_stmt_affected_rows (s->statement);
 762   if (NULL != insert_id)
 763     *insert_id = (unsigned long long) mysql_stmt_insert_id (s->statement);
 764   mysql_stmt_reset (s->statement);
 765   return affected;
 766 }
 767
 768
 769 /**
 770  * Delete an entry from the gn090 table.
 771  *
 772  * @param plugin plugin context
 773  * @param uid unique ID of the entry to delete
 774  * @return GNUNET_OK on success, GNUNET_NO if no such value exists, GNUNET_SYSERR on error
 775  */
 776 static int
 777 do_delete_entry (struct Plugin *plugin,
 778                  unsigned long long uid)
 779 {
 780   int ret;
 781
 782 #if DEBUG_MYSQL
 783   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 784               "Deleting value %llu from gn090 table\n",
 785               uid);
 786 #endif
 787   ret = prepared_statement_run (plugin,
 788                                 plugin->delete_entry_by_uid,
 789                                 NULL,
 790                                 MYSQL_TYPE_LONGLONG, &uid, GNUNET_YES,
 791                                 -1);
 792   if (ret >= 0)
 793     return GNUNET_OK;
 794   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 795               "Deleting value %llu from gn090 table failed\n",
 796               uid);
 797   return ret;
 798 }
 799
 800
 801 /**
 802  * Get an estimate of how much space the database is
 803  * currently using.
 804  *
 805  * @param cls our "struct Plugin *"
 806  * @return number of bytes used on disk
 807  */
 808 static unsigned long long
 809 mysql_plugin_estimate_size (void *cls)
 810 {
 811   struct Plugin *plugin = cls;
 812   MYSQL_BIND cbind[1];
 813   long long total;
 814
 815   memset (cbind, 0, sizeof (cbind));
 816   total = 0;
 817   cbind[0].buffer_type = MYSQL_TYPE_LONGLONG;
 818   cbind[0].buffer = &total;
 819   cbind[0].is_unsigned = GNUNET_NO;
 820   if (GNUNET_OK !=
 821       prepared_statement_run_select (plugin,
 822                                      plugin->get_size,
 823                                      1, cbind,
 824                                      -1))
 825     return 0;
 826   return total;
 827 }
 828
 829
 830 /**
 831  * Store an item in the datastore.
 832  *
 833  * @param cls closure
 834  * @param key key for the item
 835  * @param size number of bytes in data
 836  * @param data content stored
 837  * @param type type of the content
 838  * @param priority priority of the content
 839  * @param anonymity anonymity-level for the content
 840  * @param replication replication-level for the content
 841  * @param expiration expiration time for the content
 842  * @param msg set to error message
 843  * @return GNUNET_OK on success
 844  */
 845 static int
 846 mysql_plugin_put (void *cls,
 847                   const GNUNET_HashCode * key,
 848                   uint32_t size,
 849                   const void *data,
 850                   enum GNUNET_BLOCK_Type type,
 851                   uint32_t priority,
 852                   uint32_t anonymity,
 853                   uint32_t replication,
 854                   struct GNUNET_TIME_Absolute expiration,
 855                   char **msg)
 856 {
 857   struct Plugin *plugin = cls;
 858   unsigned int irepl = replication;
 859   unsigned int ipriority = priority;
 860   unsigned int ianonymity = anonymity;
 861   unsigned long long lexpiration = expiration.abs_value;
 862   unsigned long long lrvalue = (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
 863                                                                               UINT64_MAX);
 864   unsigned long hashSize;
 865   unsigned long hashSize2;
 866   unsigned long lsize;
 867   GNUNET_HashCode vhash;
 868
 869   if (size > MAX_DATUM_SIZE)
 870     {
 871       GNUNET_break (0);
 872       return GNUNET_SYSERR;
 873     }
 874   hashSize = sizeof (GNUNET_HashCode);
 875   hashSize2 = sizeof (GNUNET_HashCode);
 876   lsize = size;
 877   GNUNET_CRYPTO_hash (data, size, &vhash);
 878   if (GNUNET_OK !=
 879       prepared_statement_run (plugin,
 880                               plugin->insert_entry,
 881                               NULL,
 882                               MYSQL_TYPE_LONG, &irepl, GNUNET_YES,
 883                               MYSQL_TYPE_LONG, &type, GNUNET_YES,
 884                               MYSQL_TYPE_LONG, &ipriority, GNUNET_YES,
 885                               MYSQL_TYPE_LONG, &ianonymity, GNUNET_YES,
 886                               MYSQL_TYPE_LONGLONG, &lexpiration, GNUNET_YES,
 887                               MYSQL_TYPE_LONGLONG, &lrvalue, GNUNET_YES,
 888                               MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
 889                               MYSQL_TYPE_BLOB, &vhash, hashSize2, &hashSize2,
 890                               MYSQL_TYPE_BLOB, data, lsize, &lsize,
 891                               -1))
 892     return GNUNET_SYSERR;
 893 #if DEBUG_MYSQL
 894   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 895               "Inserted value `%s' with size %u into gn090 table\n",
 896               GNUNET_h2s (key),
 897               (unsigned int) size);
 898 #endif
 899   if (size > 0)
 900     plugin->env->duc (plugin->env->cls,
 901                       size);
 902   return GNUNET_OK;
 903 }
 904
 905
 906 /**
 907  * Update the priority for a particular key in the datastore.  If
 908  * the expiration time in value is different than the time found in
 909  * the datastore, the higher value should be kept.  For the
 910  * anonymity level, the lower value is to be used.  The specified
 911  * priority should be added to the existing priority, ignoring the
 912  * priority in value.
 913  *
 914  * Note that it is possible for multiple values to match this put.
 915  * In that case, all of the respective values are updated.
 916  *
 917  * @param cls our "struct Plugin*"
 918  * @param uid unique identifier of the datum
 919  * @param delta by how much should the priority
 920  *     change?  If priority + delta < 0 the
 921  *     priority should be set to 0 (never go
 922  *     negative).
 923  * @param expire new expiration time should be the
 924  *     MAX of any existing expiration time and
 925  *     this value
 926  * @param msg set to error message
 927  * @return GNUNET_OK on success
 928  */
 929 static int
 930 mysql_plugin_update (void *cls,
 931                      uint64_t uid,
 932                      int delta,
 933                      struct GNUNET_TIME_Absolute expire,
 934                      char **msg)
 935 {
 936   struct Plugin *plugin = cls;
 937   unsigned long long vkey = uid;
 938   unsigned long long lexpire = expire.abs_value;
 939   int ret;
 940
 941 #if DEBUG_MYSQL
 942   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 943               "Updating value %llu adding %d to priority and maxing exp at %llu\n",
 944               vkey,
 945               delta,
 946               lexpire);
 947 #endif
 948   ret = prepared_statement_run (plugin,
 949                                 plugin->update_entry,
 950                                 NULL,
 951                                 MYSQL_TYPE_LONG, &delta, GNUNET_NO,
 952                                 MYSQL_TYPE_LONGLONG, &lexpire, GNUNET_YES,
 953                                 MYSQL_TYPE_LONGLONG, &lexpire, GNUNET_YES,
 954                                 MYSQL_TYPE_LONGLONG, &vkey, GNUNET_YES,
 955                                 -1);
 956   if (ret != GNUNET_OK)
 957     {
 958       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 959                   "Failed to update value %llu\n",
 960                   vkey);
 961     }
 962   return ret;
 963 }
 964
 965
 966 /**
 967  * Run the given select statement and call 'proc' on the resulting
 968  * values (which must be in particular positions).
 969  *
 970  * @param plugin the plugin handle
 971  * @param stmt select statement to run
 972  * @param proc function to call on result
 973  * @param proc_cls closure for proc
 974  * @param ... arguments to initialize stmt
 975  */
 976 static void
 977 execute_select (struct Plugin *plugin,
 978                 struct GNUNET_MysqlStatementHandle *stmt,
 979                 PluginDatumProcessor proc, void *proc_cls,
 980                 ...)
 981 {
 982   va_list ap;
 983   int ret;
 984   unsigned int type;
 985   unsigned int priority;
 986   unsigned int anonymity;
 987   unsigned long long exp;
 988   unsigned long hashSize;
 989   unsigned long size;
 990   unsigned long long uid;
 991   char value[GNUNET_DATASTORE_MAX_VALUE_SIZE];
 992   GNUNET_HashCode key;
 993   struct GNUNET_TIME_Absolute expiration;
 994   MYSQL_BIND rbind[7];
 995
 996   hashSize = sizeof (GNUNET_HashCode);
 997   memset (rbind, 0, sizeof (rbind));
 998   rbind[0].buffer_type = MYSQL_TYPE_LONG;
 999   rbind[0].buffer = &type;
1000   rbind[0].is_unsigned = 1;
1001   rbind[1].buffer_type = MYSQL_TYPE_LONG;
1002   rbind[1].buffer = &priority;
1003   rbind[1].is_unsigned = 1;
1004   rbind[2].buffer_type = MYSQL_TYPE_LONG;
1005   rbind[2].buffer = &anonymity;
1006   rbind[2].is_unsigned = 1;
1007   rbind[3].buffer_type = MYSQL_TYPE_LONGLONG;
1008   rbind[3].buffer = &exp;
1009   rbind[3].is_unsigned = 1;
1010   rbind[4].buffer_type = MYSQL_TYPE_BLOB;
1011   rbind[4].buffer = &key;
1012   rbind[4].buffer_length = hashSize;
1013   rbind[4].length = &hashSize;
1014   rbind[5].buffer_type = MYSQL_TYPE_BLOB;
1015   rbind[5].buffer = value;
1016   rbind[5].buffer_length = size = sizeof (value);
1017   rbind[5].length = &size;
1018   rbind[6].buffer_type = MYSQL_TYPE_LONGLONG;
1019   rbind[6].buffer = &uid;
1020   rbind[6].is_unsigned = 1;
1021
1022   va_start (ap, proc_cls);
1023   ret = prepared_statement_run_select_va (plugin,
1024                                           stmt,
1025                                           7, rbind,
1026                                           ap);
1027   va_end (ap);
1028   if (ret <= 0)
1029     {
1030       proc (proc_cls,
1031             NULL, 0, NULL, 0, 0, 0,
1032             GNUNET_TIME_UNIT_ZERO_ABS, 0);
1033       return;
1034     }
1035   GNUNET_assert (size <= sizeof(value));
1036   if ( (rbind[4].buffer_length != sizeof (GNUNET_HashCode)) ||
1037        (hashSize != sizeof (GNUNET_HashCode)) )
1038     {
1039       GNUNET_break (0);
1040       proc (proc_cls,
1041             NULL, 0, NULL, 0, 0, 0,
1042             GNUNET_TIME_UNIT_ZERO_ABS, 0);
1043       return;
1044     }
1045 #if DEBUG_MYSQL
1046   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1047               "Found %u-byte value under key `%s' with prio %u, anon %u, expire %llu selecting from gn090 table\n",
1048               (unsigned int) size,
1049               GNUNET_h2s (&key),
1050               priority,
1051               anonymity,
1052               exp);
1053 #endif
1054   GNUNET_assert (size < MAX_DATUM_SIZE);
1055   expiration.abs_value = exp;
1056   ret = proc (proc_cls,
1057               &key,
1058               size, value,
1059               type, priority, anonymity, expiration,
1060               uid);
1061   if (ret == GNUNET_NO)
1062     {
1063       do_delete_entry (plugin, uid);
1064       if (size != 0)
1065         plugin->env->duc (plugin->env->cls,
1066                           - size);
1067     }
1068 }
1069
1070
1071
1072 /**
1073  * Get one of the results for a particular key in the datastore.
1074  *
1075  * @param cls closure
1076  * @param offset offset of the result (modulo num-results);
1077  *               specific ordering does not matter for the offset
1078  * @param key key to match, never NULL
1079  * @param vhash hash of the value, maybe NULL (to
1080  *        match all values that have the right key).
1081  *        Note that for DBlocks there is no difference
1082  *        betwen key and vhash, but for other blocks
1083  *        there may be!
1084  * @param type entries of which type are relevant?
1085  *     Use 0 for any type.
1086  * @param proc function to call on the matching value,
1087  *        with NULL for if no value matches
1088  * @param proc_cls closure for proc
1089  */
1090 static void
1091 mysql_plugin_get_key (void *cls,
1092                       uint64_t offset,
1093                       const GNUNET_HashCode *key,
1094                       const GNUNET_HashCode *vhash,
1095                       enum GNUNET_BLOCK_Type type,
1096                       PluginDatumProcessor proc, void *proc_cls)
1097 {
1098   struct Plugin *plugin = cls;
1099   int ret;
1100   MYSQL_BIND cbind[1];
1101   long long total;
1102   unsigned long hashSize;
1103   unsigned long hashSize2;
1104   unsigned long long off;
1105
1106   GNUNET_assert (key != NULL);
1107   GNUNET_assert (NULL != proc);
1108   hashSize = sizeof (GNUNET_HashCode);
1109   hashSize2 = sizeof (GNUNET_HashCode);
1110   memset (cbind, 0, sizeof (cbind));
1111   total = -1;
1112   cbind[0].buffer_type = MYSQL_TYPE_LONGLONG;
1113   cbind[0].buffer = &total;
1114   cbind[0].is_unsigned = GNUNET_NO;
1115   if (type != 0)
1116     {
1117       if (vhash != NULL)
1118         {
1119           ret =
1120             prepared_statement_run_select (plugin,
1121                                            plugin->count_entry_by_hash_vhash_and_type,
1122                                            1, cbind,
1123                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1124                                            MYSQL_TYPE_BLOB, vhash, hashSize2, &hashSize2,
1125                                            MYSQL_TYPE_LONG, &type, GNUNET_YES,
1126                                            -1);
1127         }
1128       else
1129         {
1130           ret =
1131             prepared_statement_run_select (plugin,
1132                                            plugin->count_entry_by_hash_and_type,
1133                                            1, cbind,
1134                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1135                                            MYSQL_TYPE_LONG, &type, GNUNET_YES,
1136                                            -1);
1137         }
1138     }
1139   else
1140     {
1141       if (vhash != NULL)
1142         {
1143           ret =
1144             prepared_statement_run_select (plugin,
1145                                            plugin->count_entry_by_hash_and_vhash,
1146                                            1, cbind,
1147                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1148                                            MYSQL_TYPE_BLOB, vhash, hashSize2, &hashSize2,
1149                                            -1);
1150
1151         }
1152       else
1153         {
1154           ret =
1155             prepared_statement_run_select (plugin,
1156                                            plugin->count_entry_by_hash,
1157                                            1, cbind,
1158                                            MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1159                                            -1);
1160         }
1161     }
1162   if ((ret != GNUNET_OK) || (0 >= total))
1163     {
1164       proc (proc_cls,
1165             NULL, 0, NULL, 0, 0, 0,
1166             GNUNET_TIME_UNIT_ZERO_ABS, 0);
1167       return;
1168     }
1169   offset = offset % total;
1170   off = (unsigned long long) offset;
1171 #if DEBUG_MYSQL
1172   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1173               "Obtaining %llu/%lld result for GET `%s'\n",
1174               off,
1175               total,
1176               GNUNET_h2s (key));
1177 #endif
1178
1179   if (type != GNUNET_BLOCK_TYPE_ANY)
1180     {
1181       if (NULL != vhash)
1182         {
1183           execute_select (plugin,
1184                           plugin->select_entry_by_hash_vhash_and_type,
1185                           proc, proc_cls,
1186                           MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1187                           MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize,
1188                           MYSQL_TYPE_LONG, &type, GNUNET_YES,
1189                           MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
1190                           -1);
1191         }
1192       else
1193         {
1194           execute_select (plugin,
1195                           plugin->select_entry_by_hash_and_type,
1196                           proc, proc_cls,
1197                           MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1198                           MYSQL_TYPE_LONG, &type, GNUNET_YES,
1199                           MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
1200                           -1);
1201         }
1202     }
1203   else
1204     {
1205       if (NULL != vhash)
1206         {
1207           execute_select (plugin,
1208                           plugin->select_entry_by_hash_and_vhash,
1209                           proc, proc_cls,
1210                           MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1211                           MYSQL_TYPE_BLOB, vhash, hashSize, &hashSize,
1212                           MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
1213                           -1);
1214         }
1215       else
1216         {
1217           execute_select (plugin,
1218                           plugin->select_entry_by_hash,
1219                           proc, proc_cls,
1220                           MYSQL_TYPE_BLOB, key, hashSize, &hashSize,
1221                           MYSQL_TYPE_LONGLONG, &off, GNUNET_YES,
1222                           -1);
1223         }
1224     }
1225 }
1226
1227
1228 /**
1229  * Get a zero-anonymity datum from the datastore.
1230  *
1231  * @param cls our "struct Plugin*"
1232  * @param offset offset of the result
1233  * @param type entries of which type should be considered?
1234  *        Use 0 for any type.
1235  * @param proc function to call on a matching value or NULL
1236  * @param proc_cls closure for iter
1237  */
1238 static void
1239 mysql_plugin_get_zero_anonymity (void *cls,
1240                                  uint64_t offset,
1241                                  enum GNUNET_BLOCK_Type type,
1242                                  PluginDatumProcessor proc, void *proc_cls)
1243 {
1244   struct Plugin *plugin = cls;
1245   unsigned long long rvalue = (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1246                                                                              UINT64_MAX);
1247   execute_select (plugin,
1248                   plugin->zero_iter,
1249                   proc, proc_cls,
1250                   MYSQL_TYPE_LONG, &type, GNUNET_YES,
1251                   MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES,
1252                   MYSQL_TYPE_LONG, &type, GNUNET_YES,
1253                   MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES,
1254                   -1);
1255 }
1256
1257
1258 /**
1259  * Context for 'repl_proc' function.
1260  */
1261 struct ReplCtx
1262 {
1263
1264   /**
1265    * Plugin handle.
1266    */
1267   struct Plugin *plugin;
1268
1269   /**
1270    * Function to call for the result (or the NULL).
1271    */
1272   PluginDatumProcessor proc;
1273
1274   /**
1275    * Closure for proc.
1276    */
1277   void *proc_cls;
1278 };
1279
1280
1281 /**
1282  * Wrapper for the processor for 'mysql_plugin_get_replication'.
1283  * Decrements the replication counter and calls the original
1284  * iterator.
1285  *
1286  * @param cls closure
1287  * @param key key for the content
1288  * @param size number of bytes in data
1289  * @param data content stored
1290  * @param type type of the content
1291  * @param priority priority of the content
1292  * @param anonymity anonymity-level for the content
1293  * @param expiration expiration time for the content
1294  * @param uid unique identifier for the datum;
1295  *        maybe 0 if no unique identifier is available
1296  *
1297  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
1298  *         (continue on call to "next", of course),
1299  *         GNUNET_NO to delete the item and continue (if supported)
1300  */
1301 static int
1302 repl_proc (void *cls,
1303            const GNUNET_HashCode *key,
1304            uint32_t size,
1305            const void *data,
1306            enum GNUNET_BLOCK_Type type,
1307            uint32_t priority,
1308            uint32_t anonymity,
1309            struct GNUNET_TIME_Absolute expiration,
1310            uint64_t uid)
1311 {
1312   struct ReplCtx *rc = cls;
1313   struct Plugin *plugin = rc->plugin;
1314   unsigned long long oid;
1315   int ret;
1316   int iret;
1317
1318   ret = rc->proc (rc->proc_cls,
1319                   key,
1320                   size, data,
1321                   type, priority, anonymity, expiration,
1322                   uid);
1323   if (NULL != key)
1324     {
1325       oid = (unsigned long long) uid;
1326       iret = prepared_statement_run (plugin,
1327                                      plugin->dec_repl,
1328                                      NULL,
1329                                      MYSQL_TYPE_LONGLONG, &oid, GNUNET_YES,
1330                                      -1);
1331       if (iret == GNUNET_SYSERR)
1332         {
1333           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1334                       "Failed to reduce replication counter\n");
1335           return GNUNET_SYSERR;
1336         }
1337     }
1338   return ret;
1339 }
1340
1341
1342 /**
1343  * Get a random item for replication.  Returns a single, not expired,
1344  * random item from those with the highest replication counters.  The
1345  * item's replication counter is decremented by one IF it was positive
1346  * before.  Call 'proc' with all values ZERO or NULL if the datastore
1347  * is empty.
1348  *
1349  * @param cls closure
1350  * @param proc function to call the value (once only).
1351  * @param proc_cls closure for proc
1352  */
1353 static void
1354 mysql_plugin_get_replication (void *cls,
1355                               PluginDatumProcessor proc, void *proc_cls)
1356 {
1357   struct Plugin *plugin = cls;
1358   struct ReplCtx rc;
1359   unsigned long long rvalue;
1360   unsigned long repl;
1361   MYSQL_BIND results;
1362
1363   rc.plugin = plugin;
1364   rc.proc = proc;
1365   rc.proc_cls = proc_cls;
1366   memset (&results, 0, sizeof (results));
1367   results.buffer_type = MYSQL_TYPE_LONG;
1368   results.buffer = &repl;
1369   results.is_unsigned = GNUNET_YES;
1370
1371   if (1 !=
1372       prepared_statement_run_select (plugin,
1373                                      plugin->max_repl,
1374                                      1,
1375                                      &results,
1376                                      -1))
1377     {
1378       proc (proc_cls,
1379             NULL, 0, NULL, 0, 0, 0,
1380             GNUNET_TIME_UNIT_ZERO_ABS, 0);
1381       return;
1382     }
1383
1384   rvalue = (unsigned long long) GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1385                                                           UINT64_MAX);
1386   execute_select (plugin,
1387                   plugin->select_replication,
1388                   &repl_proc, &rc,
1389                   MYSQL_TYPE_LONG, &repl, GNUNET_YES,
1390                   MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES,
1391                   MYSQL_TYPE_LONG, &repl, GNUNET_YES,
1392                   MYSQL_TYPE_LONGLONG, &rvalue, GNUNET_YES,
1393                   -1);
1394
1395 }
1396
1397
1398 /**
1399  * Context for 'expi_proc' function.
1400  */
1401 struct ExpiCtx
1402 {
1403
1404   /**
1405    * Plugin handle.
1406    */
1407   struct Plugin *plugin;
1408
1409   /**
1410    * Function to call for the result (or the NULL).
1411    */
1412   PluginDatumProcessor proc;
1413
1414   /**
1415    * Closure for proc.
1416    */
1417   void *proc_cls;
1418 };
1419
1420
1421
1422 /**
1423  * Wrapper for the processor for 'mysql_plugin_get_expiration'.
1424  * If no expired value was found, we do a second query for
1425  * low-priority content.
1426  *
1427  * @param cls closure
1428  * @param key key for the content
1429  * @param size number of bytes in data
1430  * @param data content stored
1431  * @param type type of the content
1432  * @param priority priority of the content
1433  * @param anonymity anonymity-level for the content
1434  * @param expiration expiration time for the content
1435  * @param uid unique identifier for the datum;
1436  *        maybe 0 if no unique identifier is available
1437  *
1438  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
1439  *         (continue on call to "next", of course),
1440  *         GNUNET_NO to delete the item and continue (if supported)
1441  */
1442 static int
1443 expi_proc (void *cls,
1444            const GNUNET_HashCode *key,
1445            uint32_t size,
1446            const void *data,
1447            enum GNUNET_BLOCK_Type type,
1448            uint32_t priority,
1449            uint32_t anonymity,
1450            struct GNUNET_TIME_Absolute expiration,
1451            uint64_t uid)
1452 {
1453   struct ExpiCtx *rc = cls;
1454   struct Plugin *plugin = rc->plugin;
1455
1456   if (NULL == key)
1457     {
1458       execute_select (plugin,
1459                       plugin->select_priority,
1460                       rc->proc, rc->proc_cls,
1461                       -1);
1462       return GNUNET_SYSERR;
1463     }
1464   return rc->proc (rc->proc_cls,
1465                    key,
1466                    size, data,
1467                    type, priority, anonymity, expiration,
1468                    uid);
1469 }
1470
1471
1472 /**
1473  * Get a random item for expiration.
1474  * Call 'proc' with all values ZERO or NULL if the datastore is empty.
1475  *
1476  * @param cls closure
1477  * @param proc function to call the value (once only).
1478  * @param proc_cls closure for proc
1479  */
1480 static void
1481 mysql_plugin_get_expiration (void *cls,
1482                              PluginDatumProcessor proc, void *proc_cls)
1483 {
1484   struct Plugin *plugin = cls;
1485   long long nt;
1486   struct ExpiCtx rc;
1487
1488   rc.plugin = plugin;
1489   rc.proc = proc;
1490   rc.proc_cls = proc_cls;
1491   nt = (long long) GNUNET_TIME_absolute_get().abs_value;
1492   execute_select (plugin,
1493                   plugin->select_expiration,
1494                   expi_proc, &rc,
1495                   MYSQL_TYPE_LONGLONG, &nt, GNUNET_YES,
1496                   -1);
1497
1498 }
1499
1500
1501 /**
1502  * Drop database.
1503  *
1504  * @param cls the "struct Plugin*"
1505  */
1506 static void
1507 mysql_plugin_drop (void *cls)
1508 {
1509   struct Plugin *plugin = cls;
1510
1511   if (GNUNET_OK != run_statement (plugin,
1512                                   "DROP TABLE gn090"))
1513     return;           /* error */
1514   plugin->env->duc (plugin->env->cls, 0);
1515 }
1516
1517
1518 /**
1519  * Entry point for the plugin.
1520  *
1521  * @param cls the "struct GNUNET_DATASTORE_PluginEnvironment*"
1522  * @return our "struct Plugin*"
1523  */
1524 void *
1525 libgnunet_plugin_datastore_mysql_init (void *cls)
1526 {
1527   struct GNUNET_DATASTORE_PluginEnvironment *env = cls;
1528   struct GNUNET_DATASTORE_PluginFunctions *api;
1529   struct Plugin *plugin;
1530
1531   plugin = GNUNET_malloc (sizeof (struct Plugin));
1532   plugin->env = env;
1533   plugin->cnffile = get_my_cnf_path (env->cfg);
1534   if (GNUNET_OK != iopen (plugin))
1535     {
1536       iclose (plugin);
1537       GNUNET_free_non_null (plugin->cnffile);
1538       GNUNET_free (plugin);
1539       return NULL;
1540     }
1541 #define MRUNS(a) (GNUNET_OK != run_statement (plugin, a) )
1542 #define PINIT(a,b) (NULL == (a = prepared_statement_create(plugin, b)))
1543   if (MRUNS ("CREATE TABLE IF NOT EXISTS gn090 ("
1544              " repl INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1545              " type INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1546              " prio INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1547              " anonLevel INT(11) UNSIGNED NOT NULL DEFAULT 0,"
1548              " expire BIGINT UNSIGNED NOT NULL DEFAULT 0,"
1549              " rvalue BIGINT UNSIGNED NOT NULL,"
1550              " hash BINARY(64) NOT NULL DEFAULT '',"
1551              " vhash BINARY(64) NOT NULL DEFAULT '',"
1552              " value BLOB NOT NULL DEFAULT '',"
1553              " uid BIGINT NOT NULL AUTO_INCREMENT,"
1554              " PRIMARY KEY (uid),"
1555              " INDEX idx_hash (hash(64)),"
1556              " INDEX idx_hash_uid (hash(64),uid),"
1557              " INDEX idx_hash_vhash (hash(64),vhash(64)),"
1558              " INDEX idx_hash_type_uid (hash(64),type,rvalue),"
1559              " INDEX idx_prio (prio),"
1560              " INDEX idx_repl_rvalue (repl,rvalue),"
1561              " INDEX idx_expire (expire),"
1562              " INDEX idx_anonLevel_type_rvalue (anonLevel,type,rvalue)"
1563              ") ENGINE=InnoDB") ||
1564       MRUNS ("SET AUTOCOMMIT = 1") ||
1565       PINIT (plugin->insert_entry, INSERT_ENTRY) ||
1566       PINIT (plugin->delete_entry_by_uid, DELETE_ENTRY_BY_UID) ||
1567       PINIT (plugin->select_entry_by_hash, SELECT_ENTRY_BY_HASH) ||
1568       PINIT (plugin->select_entry_by_hash_and_vhash, SELECT_ENTRY_BY_HASH_AND_VHASH)
1569       || PINIT (plugin->select_entry_by_hash_and_type, SELECT_ENTRY_BY_HASH_AND_TYPE)
1570       || PINIT (plugin->select_entry_by_hash_vhash_and_type,
1571                 SELECT_ENTRY_BY_HASH_VHASH_AND_TYPE)
1572       || PINIT (plugin->count_entry_by_hash, COUNT_ENTRY_BY_HASH)
1573       || PINIT (plugin->get_size, SELECT_SIZE)
1574       || PINIT (plugin->count_entry_by_hash_and_vhash, COUNT_ENTRY_BY_HASH_AND_VHASH)
1575       || PINIT (plugin->count_entry_by_hash_and_type, COUNT_ENTRY_BY_HASH_AND_TYPE)
1576       || PINIT (plugin->count_entry_by_hash_vhash_and_type,
1577                 COUNT_ENTRY_BY_HASH_VHASH_AND_TYPE)
1578       || PINIT (plugin->update_entry, UPDATE_ENTRY)
1579       || PINIT (plugin->dec_repl, DEC_REPL)
1580       || PINIT (plugin->zero_iter, SELECT_IT_NON_ANONYMOUS)
1581       || PINIT (plugin->select_expiration, SELECT_IT_EXPIRATION)
1582       || PINIT (plugin->select_priority, SELECT_IT_PRIORITY)
1583       || PINIT (plugin->max_repl, SELECT_MAX_REPL)
1584       || PINIT (plugin->select_replication, SELECT_IT_REPLICATION) )
1585     {
1586       iclose (plugin);
1587       GNUNET_free_non_null (plugin->cnffile);
1588       GNUNET_free (plugin);
1589       return NULL;
1590     }
1591 #undef PINIT
1592 #undef MRUNS
1593
1594   api = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_PluginFunctions));
1595   api->cls = plugin;
1596   api->estimate_size = &mysql_plugin_estimate_size;
1597   api->put = &mysql_plugin_put;
1598   api->update = &mysql_plugin_update;
1599   api->get_key = &mysql_plugin_get_key;
1600   api->get_replication = &mysql_plugin_get_replication;
1601   api->get_expiration = &mysql_plugin_get_expiration;
1602   api->get_zero_anonymity = &mysql_plugin_get_zero_anonymity;
1603   api->drop = &mysql_plugin_drop;
1604   GNUNET_log_from (GNUNET_ERROR_TYPE_INFO,
1605                    "mysql", _("Mysql database running\n"));
1606   return api;
1607 }
1608
1609
1610 /**
1611  * Exit point from the plugin.
1612  * @param cls our "struct Plugin*"
1613  * @return always NULL
1614  */
1615 void *
1616 libgnunet_plugin_datastore_mysql_done (void *cls)
1617 {
1618   struct GNUNET_DATASTORE_PluginFunctions *api = cls;
1619   struct Plugin *plugin = api->cls;
1620   struct GNUNET_MysqlStatementHandle *s;
1621
1622   iclose (plugin);
1623   while (NULL != (s = plugin->shead))
1624     {
1625       GNUNET_CONTAINER_DLL_remove (plugin->shead,
1626                                    plugin->stail,
1627                                    s);
1628       GNUNET_free (s->query);
1629       GNUNET_free (s);
1630     }
1631   GNUNET_free_non_null (plugin->cnffile);
1632   GNUNET_free (plugin);
1633   GNUNET_free (api);
1634   mysql_library_end ();
1635   return NULL;
1636 }
1637
1638 /* end of plugin_datastore_mysql.c */