mysql.c

GNUnet是一个安全的点对点网络框架

                     GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("`%s' failed at %s:%d with error: %s\n"),
                     "mysql_stmt_bind_result",
                     __FILE__, __LINE__,
                     mysql_stmt_error (dbh->select_value));
      delete_entry_by_vkey (vkey);
      content_size -= ntohl (datum->size);
      GNUNET_mutex_unlock (lock);
      GNUNET_free (datum);
      return NULL;
    }
  mysql_stmt_reset (dbh->select_value);
  GNUNET_mutex_unlock (lock);
  return datum;
}

/**
 * Store an item in the datastore.
 *
 * @return GNUNET_OK on success, GNUNET_SYSERR on error
 */
static int
put (const GNUNET_HashCode * key, const GNUNET_DatastoreValue * value)
{
  unsigned long contentSize;
  unsigned long hashSize;
  unsigned long hashSize2;
  unsigned int size;
  unsigned int type;
  unsigned int prio;
  unsigned int level;
  unsigned long long expiration;
  unsigned long long vkey;
  GNUNET_HashCode vhash;
  MYSQL_BIND qbind[8];
#if DEBUG_MYSQL
  GNUNET_EncName enc;
#endif

  if (((ntohl (value->size) < sizeof (GNUNET_DatastoreValue))) ||
      ((ntohl (value->size) - sizeof (GNUNET_DatastoreValue)) >
       MAX_DATUM_SIZE))
    {
      GNUNET_GE_BREAK (ectx, 0);
      return GNUNET_SYSERR;
    }
  hashSize = sizeof (GNUNET_HashCode);
  hashSize2 = sizeof (GNUNET_HashCode);
  size = ntohl (value->size);
  type = ntohl (value->type);
  prio = ntohl (value->priority);
  level = ntohl (value->anonymity_level);
  expiration = GNUNET_ntohll (value->expiration_time);
  contentSize = ntohl (value->size) - sizeof (GNUNET_DatastoreValue);
  GNUNET_hash (&value[1], contentSize, &vhash);
  GNUNET_mutex_lock (lock);
  mysql_thread_init ();
  if (GNUNET_OK != CHECK_DBH)
    {
      mysql_thread_end ();
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }
  if (GNUNET_OK != insert_value (&value[1], contentSize, &vkey))
    {
      mysql_thread_end ();
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }
#if DEBUG_MYSQL
  IF_GELOG (ectx, GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
            GNUNET_hash_to_enc (key, &enc));
  GNUNET_GE_LOG (ectx, GNUNET_GE_DEBUG | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 "Storing in database block with type %u and key %s.\n", type,
                 &enc);
#endif
  GNUNET_GE_ASSERT (ectx, mysql_stmt_param_count (dbh->insert_entry) == 8);
  memset (qbind, 0, sizeof (qbind));
  qbind[0].buffer_type = MYSQL_TYPE_LONG;       /* size */
  qbind[0].buffer = &size;
  qbind[0].is_unsigned = 1;
  qbind[1].buffer_type = MYSQL_TYPE_LONG;       /* type */
  qbind[1].is_unsigned = 1;
  qbind[1].buffer = &type;
  qbind[2].buffer_type = MYSQL_TYPE_LONG;       /* priority */
  qbind[2].is_unsigned = 1;
  qbind[2].buffer = &prio;
  qbind[3].buffer_type = MYSQL_TYPE_LONG;       /* anon level */
  qbind[3].is_unsigned = 1;
  qbind[3].buffer = &level;
  qbind[4].buffer_type = MYSQL_TYPE_LONGLONG;   /* expiration */
  qbind[4].is_unsigned = 1;
  qbind[4].buffer = &expiration;
  qbind[5].buffer_type = MYSQL_TYPE_BLOB;       /* GNUNET_hash */
  qbind[5].buffer = (void *) key;
  qbind[5].length = &hashSize;
  qbind[5].buffer_length = hashSize;
  qbind[6].buffer_type = MYSQL_TYPE_BLOB;       /* vhash */
  qbind[6].buffer = (void *) &vhash;
  qbind[6].length = &hashSize2;
  qbind[6].buffer_length = hashSize2;
  qbind[7].buffer_type = MYSQL_TYPE_LONGLONG;   /* vkey */
  qbind[7].is_unsigned = 1;
  qbind[7].buffer = &vkey;

  if (mysql_stmt_bind_param (dbh->insert_entry, qbind))
    {
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("`%s' failed at %s:%d with error: %s\n"),
                     "mysql_stmt_bind_param",
                     __FILE__, __LINE__,
                     mysql_stmt_error (dbh->insert_entry));
      delete_value (vkey);
      iclose ();
      mysql_thread_end ();
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }

  if (mysql_stmt_execute (dbh->insert_entry))
    {
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                     _("`%s' failed at %s:%d with error: %s\n"),
                     "mysql_stmt_execute",
                     __FILE__, __LINE__,
                     mysql_stmt_error (dbh->insert_entry));
      delete_value (vkey);
      iclose ();
      mysql_thread_end ();
      GNUNET_mutex_unlock (lock);
      return GNUNET_SYSERR;
    }
  mysql_thread_end ();
  content_size += ntohl (value->size);
  GNUNET_mutex_unlock (lock);
  return GNUNET_OK;
}




/**
 * Iterate over the items in the datastore
 * using the given query to select and order
 * the items.
 *
 * @param type entries of which type should be considered?
 *        Use 0 for any type.
 * @param iter never NULL
 * @param is_asc are we using ascending order?
 * @param is_prio is the extra ordering by priority (otherwise by expiration)
 * @param is_migr is this IT_MIGRATON_ORDER (with expire)
 * @return the number of results, GNUNET_SYSERR if the
 *   iter is non-NULL and aborted the iteration
 */
static int
iterateHelper (unsigned int type,
               int is_asc,
               int is_prio,
               int is_migr,
               unsigned int iter_select, GNUNET_DatastoreValueIterator iter,
               void *closure)
{
  GNUNET_DatastoreValue *datum;
  int count;
  int ret;
  unsigned int last_prio;
  unsigned long long last_expire;
  unsigned long long last_vkey;
  unsigned int size;
  unsigned int rtype;
  unsigned int prio;
  unsigned int level;
  unsigned long long expiration;
  unsigned long long vkey;
  unsigned long hashSize;
  GNUNET_HashCode key;
  GNUNET_CronTime now;
  MYSQL_BIND qbind[6];
  MYSQL_BIND rbind[7];
  MYSQL_STMT *stmt;

  GNUNET_GE_ASSERT (NULL, ((is_migr == 1) || (is_migr == 0)));
  if (is_asc)
    {
      last_prio = 0;
      last_vkey = 0;
      last_expire = 0;
    }
  else
    {
      last_prio = 0x7FFFFFFFL;
      last_vkey = 0x7FFFFFFFFFFFFFFFLL; /* MySQL only supports 63 bits */
      last_expire = 0x7FFFFFFFFFFFFFFFLL;       /* MySQL only supports 63 bits */
    }
  memset (qbind, 0, sizeof (qbind));
  if (is_prio)
    {
      qbind[0].buffer_type = MYSQL_TYPE_LONG;
      qbind[0].buffer = &last_prio;
      qbind[0].is_unsigned = 1;
      qbind[2 + is_migr].buffer_type = MYSQL_TYPE_LONG;
      qbind[2 + is_migr].buffer = &last_prio;
      qbind[2 + is_migr].is_unsigned = 1;
    }
  else
    {
      qbind[0].buffer_type = MYSQL_TYPE_LONGLONG;
      qbind[0].buffer = &last_expire;
      qbind[0].is_unsigned = 1;
      qbind[2 + is_migr].buffer_type = MYSQL_TYPE_LONGLONG;
      qbind[2 + is_migr].buffer = &last_expire;
      qbind[2 + is_migr].is_unsigned = 1;
    }
  qbind[1].buffer_type = MYSQL_TYPE_LONGLONG;
  qbind[1].buffer = &last_vkey;
  qbind[1].is_unsigned = 1;
  qbind[3 + is_migr].buffer_type = MYSQL_TYPE_LONGLONG;
  qbind[3 + is_migr].buffer = &last_vkey;
  qbind[3 + is_migr].is_unsigned = 1;
  if (is_migr)
    {
      qbind[2].buffer_type = MYSQL_TYPE_LONGLONG;
      qbind[2].buffer = &now;
      qbind[2].is_unsigned = 1;
      qbind[5].buffer_type = MYSQL_TYPE_LONGLONG;
      qbind[5].buffer = &now;
      qbind[5].is_unsigned = 1;
    }

  hashSize = sizeof (GNUNET_HashCode);
  memset (rbind, 0, sizeof (rbind));
  rbind[0].buffer_type = MYSQL_TYPE_LONG;
  rbind[0].buffer = &size;
  rbind[0].is_unsigned = 1;
  rbind[1].buffer_type = MYSQL_TYPE_LONG;
  rbind[1].buffer = &rtype;
  rbind[1].is_unsigned = 1;
  rbind[2].buffer_type = MYSQL_TYPE_LONG;
  rbind[2].buffer = &prio;
  rbind[2].is_unsigned = 1;
  rbind[3].buffer_type = MYSQL_TYPE_LONG;
  rbind[3].buffer = &level;
  rbind[3].is_unsigned = 1;
  rbind[4].buffer_type = MYSQL_TYPE_LONGLONG;
  rbind[4].buffer = &expiration;
  rbind[4].is_unsigned = 1;
  rbind[5].buffer_type = MYSQL_TYPE_BLOB;
  rbind[5].buffer = &key;
  rbind[5].buffer_length = hashSize;
  rbind[5].length = &hashSize;
  rbind[6].buffer_type = MYSQL_TYPE_LONGLONG;
  rbind[6].buffer = &vkey;
  rbind[6].is_unsigned = GNUNET_YES;

  mysql_thread_init ();
  count = 0;
  while (1)
    {
      GNUNET_mutex_lock (lock);
      if (GNUNET_OK != CHECK_DBH)
        {
          GNUNET_mutex_unlock (lock);
          mysql_thread_end ();
          return GNUNET_SYSERR;
        }
      stmt = dbh->iter[iter_select];
      GNUNET_GE_ASSERT (ectx, mysql_stmt_param_count (stmt) <= 6);
      GNUNET_GE_ASSERT (ectx, mysql_stmt_field_count (stmt) == 7);
      now = GNUNET_get_time ();
      if (mysql_stmt_bind_param (stmt, qbind))
        {
          GNUNET_GE_LOG (ectx,
                         GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                         _("`%s' failed at %s:%d with error: %s\n"),
                         "mysql_stmt_bind_param",
                         __FILE__, __LINE__, mysql_stmt_error (stmt));
          iclose ();
          GNUNET_mutex_unlock (lock);
          mysql_thread_end ();
          return GNUNET_SYSERR;
        }
      if (mysql_stmt_execute (stmt))
        {
          GNUNET_GE_LOG (ectx,
                         GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                         _("`%s' failed at %s:%d with error: %s\n"),
                         "mysql_stmt_execute",
                         __FILE__, __LINE__, mysql_stmt_error (stmt));
          iclose ();
          GNUNET_mutex_unlock (lock);
          mysql_thread_end ();
          return GNUNET_SYSERR;
        }
      if (mysql_stmt_bind_result (stmt, rbind))
        {
          GNUNET_GE_LOG (ectx,
                         GNUNET_GE_ERROR | GNUNET_GE_BULK | GNUNET_GE_USER,
                         _("`%s' failed at %s:%d with error: %s\n"),
                         "mysql_stmt_bind_result",
                         __FILE__, __LINE__, mysql_stmt_error (stmt));
          iclose ();
          mysql_thread_end ();
          GNUNET_mutex_unlock (lock);
          return GNUNET_SYSERR;
        }
      datum = NULL;
      if (0 != mysql_stmt_fetch (stmt))
        {
          mysql_stmt_reset (stmt);
          GNUNET_mutex_unlock (lock);
          break;
        }
      mysql_stmt_reset (stmt);
      GNUNET_mutex_unlock (lock);
      last_vkey = vkey;
      last_prio = prio;
      last_expire = expiration;
      count++;
      if (iter != NULL)
        {
          datum = assembleDatum (rbind);
          if (datum == NULL)
            continue;
          ret = iter (&key, datum, closure, vkey);
          if (ret == GNUNET_SYSERR)
            {
              GNUNET_free (datum);
              break;
            }
          if (ret == GNUNET_NO)
            {
              GNUNET_mutex_lock (lock);
              delete_value (vkey);
              delete_entry_by_vkey (vkey);
              content_size -= ntohl (datum->size);
              GNUNET_mutex_unlock (lock);
            }
          GNUNET_free (datum);
        }
    }
  mysql_thread_end ();
  return count;
}

/**
 * Iterate over the items in the datastore in ascending
 * order of priority.
 *
 * @param type entries of which type should be considered?
 *        Use 0 for any type.
 * @param iter never NULL
 * @return the number of results, GNUNET_SYSERR if the
 *   iter is non-NULL and aborted the iteration
 */
static int
iterateLowPriority (unsigned int type, GNUNET_DatastoreValueIterator iter,
                    void *closure)
{
  return iterateHelper (type, GNUNET_YES, GNUNET_YES, GNUNET_NO, 0, iter,
                        closure);
}

/**
 * Iterate over the items in the datastore that
 * have anonymity level 0.
 *
 * @param type entries of which type should be considered?
 *        Use 0 for any type.
 * @param iter never NULL
 * @return the number of results, GNUNET_SYSERR if the
 *   iter is non-NULL and aborted the iteration