fslib.c

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

/*
     This file is part of GNUnet
     (C) 2004, 2005, 2006, 2007, 2008 Christian Grothoff (and other contributing authors)

     GNUnet is free software; you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published
     by the Free Software Foundation; either version 2, or (at your
     option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     General Public License for more details.

     You should have received a copy of the GNU General Public License
     along with GNUnet; see the file COPYING.  If not, write to the
     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
     Boston, MA 02111-1307, USA.
*/

/**
 * @file applications/fs/lib/fslib.c
 * @brief convenience methods to access the FS application from clients
 * @author Christian Grothoff
 */

#include "platform.h"
#include "gnunet_fs_lib.h"
#include "gnunet_protocols.h"
#include "fs.h"
#include "ecrs_core.h"

#define DEBUG_FSLIB GNUNET_NO

/**
 * How often should we automatically retry a request
 * that failed? (Note that searches are retried
 * indefinitely in any case; this only applies
 * to upload/delete operations).
 */
#define AUTO_RETRY 5

/**
 * In memory, the search handle is followed
 * by a copy of the corresponding request of
 * type "CS_fs_request_search_MESSAGE *".
 */
struct GNUNET_FS_SearchHandle
{
  /**
   * This is a linked list.
   */
  struct GNUNET_FS_SearchHandle *next;

  /**
   * Function to call with results.
   */
  GNUNET_DatastoreValueIterator callback;

  /**
   * Extra argument to pass to callback.
   */
  void *closure;
};

/**
 * Context for a set of search operations.
 */
struct GNUNET_FS_SearchContext
{
  /**
   * Configuration data.
   */
  struct GNUNET_GC_Configuration *cfg;

  /**
   * Error logging.
   */
  struct GNUNET_GE_Context *ectx;

  /**
   * Connection to gnunetd.
   */
  struct GNUNET_ClientServerConnection *sock;

  /**
   * Thread listening for replies.
   */
  struct GNUNET_ThreadHandle *thread;

  /**
   * Lock for access to this struct.
   */
  struct GNUNET_Mutex *lock;

  /**
   * List of active requests.
   */
  struct GNUNET_FS_SearchHandle *handles;

  /**
   * Flag to signal that we should abort.
   */
  int abort;

  /**
   * Counter for how many times this context has
   * been suspended.  Results will not be passed
   * on until the counter is zero.
   */
  unsigned int block_results;

#if DEBUG_FSLIB
  unsigned int total_received;

  unsigned int total_requested;
#endif
};

/**
 * Retransmit all of the requests to gnunetd
 * (used after a disconnect).
 */
static int
reissue_requests (struct GNUNET_FS_SearchContext *ctx)
{
  const CS_fs_request_search_MESSAGE *req;
  struct GNUNET_FS_SearchHandle *pos;

  pos = ctx->handles;
  while (pos != NULL)
    {
      req = (const CS_fs_request_search_MESSAGE *) &pos[1];
      if (GNUNET_OK !=
          GNUNET_client_connection_write (ctx->sock, &req->header))
        return GNUNET_SYSERR;
      pos = pos->next;
    }
  if (GNUNET_SYSERR == GNUNET_client_connection_ensure_connected (ctx->sock))
    return GNUNET_SYSERR;
  return GNUNET_OK;
}

/**
 * Thread that processes replies from gnunetd and
 * calls the appropriate callback.
 */
static void *
reply_process_thread (void *cls)
{
  struct GNUNET_FS_SearchContext *ctx = cls;
  GNUNET_MessageHeader *hdr;
  int matched;
  const CS_fs_reply_content_MESSAGE *rep;
  GNUNET_HashCode query;
  unsigned int size;
  GNUNET_CronTime delay;
  const CS_fs_request_search_MESSAGE *req;
  GNUNET_DatastoreValue *value;
  struct GNUNET_FS_SearchHandle *pos;
  struct GNUNET_FS_SearchHandle *spos;
  struct GNUNET_FS_SearchHandle *prev;
  int unique;

  delay = 100 * GNUNET_CRON_MILLISECONDS;
  while (ctx->abort == GNUNET_NO)
    {
      hdr = NULL;
      if (GNUNET_OK == GNUNET_client_connection_read (ctx->sock, &hdr))
        {
#if DEBUG_FSLIB
          fprintf (stderr, "FSLIB: received message from gnunetd\n");
#endif
          delay = 100 * GNUNET_CRON_MILLISECONDS;
          /* verify hdr, if reply, process, otherwise
             signal protocol problem; if ok, find
             matching callback, call on value */
          if ((ntohs (hdr->size) < sizeof (CS_fs_reply_content_MESSAGE)) ||
              (ntohs (hdr->type) != GNUNET_CS_PROTO_GAP_RESULT))
            {
              GNUNET_GE_BREAK (ctx->ectx, 0);
              GNUNET_free (hdr);
              continue;
            }
          rep = (const CS_fs_reply_content_MESSAGE *) hdr;
          size = ntohs (hdr->size) - sizeof (CS_fs_reply_content_MESSAGE);
          if (GNUNET_OK != GNUNET_EC_file_block_check_and_get_query (size, (GNUNET_EC_DBlock *) & rep[1], GNUNET_NO,    /* gnunetd will have checked already */
                                                                     &query))
            {
              GNUNET_GE_BREAK (ctx->ectx, 0);
              GNUNET_free (hdr);
              continue;
            }
          unique =
            GNUNET_EC_file_block_get_type (size,
                                           (GNUNET_EC_DBlock *) & rep[1]) ==
            GNUNET_ECRS_BLOCKTYPE_DATA;
          value = GNUNET_malloc (sizeof (GNUNET_DatastoreValue) + size);
          value->size = htonl (size + sizeof (GNUNET_DatastoreValue));
          value->type =
            htonl (GNUNET_EC_file_block_get_type
                   (size, (GNUNET_EC_DBlock *) & rep[1]));
          value->priority = htonl (0);
          value->anonymity_level = rep->anonymity_level;
          value->expiration_time = rep->expiration_time;
          memcpy (&value[1], &rep[1], size);
          matched = 0;
          GNUNET_mutex_lock (ctx->lock);
          while (ctx->block_results > 0)
            {
              GNUNET_mutex_unlock (ctx->lock);
              GNUNET_thread_sleep (100 * GNUNET_CRON_MILLISECONDS);
              GNUNET_mutex_lock (ctx->lock);
            }
          prev = NULL;
          pos = ctx->handles;
          while (pos != NULL)
            {
              req = (const CS_fs_request_search_MESSAGE *) &pos[1];
              if (0 ==
                  memcmp (&query, &req->query[0], sizeof (GNUNET_HashCode)))
                {
                  matched++;
                  spos = pos;
                  if (unique)
                    {
                      if (prev == NULL)
                        ctx->handles = pos->next;
                      else
                        prev->next = pos->next;
                      if (prev == NULL)
                        pos = ctx->handles;
                      else
                        pos = prev->next;
                    }
                  else
                    {
                      prev = pos;
                      pos = pos->next;
                    }
#if DEBUG_FSLIB
                  fprintf (stderr,
                           "FSLIB passes response %u to client (%d)\n",
                           ctx->total_received++, unique);
#endif
                  if ((spos->callback != NULL) &&
                      (GNUNET_SYSERR == spos->callback (&query,
                                                        value,
                                                        spos->closure, 0)))
                    spos->callback = NULL;
                  if (unique)
                    GNUNET_free (spos);
                }
              else
                {
                  prev = pos;
                  pos = pos->next;
                }
            }
          GNUNET_free (value);
#if DEBUG_FSLIB
          if (matched == 0)
            fprintf (stderr,
                     "FSLIB: received content but have no pending request\n");
#endif
          GNUNET_mutex_unlock (ctx->lock);
        }
      else
        {
          while (GNUNET_NO == ctx->abort)
            {
              GNUNET_thread_sleep (delay);
              delay *= 2;
              if (delay > 60 * GNUNET_CRON_SECONDS)
                delay = 60 * GNUNET_CRON_SECONDS;
              if ((GNUNET_OK ==
                   GNUNET_client_connection_ensure_connected (ctx->sock))
                  && (GNUNET_OK == reissue_requests (ctx)))
                break;          /* we're back, continue outer loop! */
            }
        }
      GNUNET_free_non_null (hdr);
    }
  return NULL;
}

struct GNUNET_FS_SearchContext *
GNUNET_FS_create_search_context (struct GNUNET_GE_Context *ectx,
                                 struct GNUNET_GC_Configuration *cfg)
{
  struct GNUNET_FS_SearchContext *ret;

  ret = GNUNET_malloc (sizeof (struct GNUNET_FS_SearchContext));
  memset (ret, 0, sizeof (struct GNUNET_FS_SearchContext));
  ret->ectx = ectx;
  ret->cfg = cfg;
  ret->lock = GNUNET_mutex_create (GNUNET_YES);
  ret->sock = GNUNET_client_connection_create (ectx, cfg);
  if (ret->sock == NULL)
    {
      GNUNET_mutex_destroy (ret->lock);
      GNUNET_free (ret);
      return NULL;
    }
  ret->handles = NULL;
  ret->abort = GNUNET_NO;
  ret->thread = GNUNET_thread_create (&reply_process_thread, ret, 128 * 1024);
  if (ret->thread == NULL)
    GNUNET_GE_DIE_STRERROR (ectx,
                            GNUNET_GE_FATAL | GNUNET_GE_ADMIN |
                            GNUNET_GE_BULK, "PTHREAD_CREATE");
  return ret;
}


/**
 * Resume the search context (start sending results again).
 */
void
GNUNET_FS_resume_search_context (struct GNUNET_FS_SearchContext *ctx)
{
  ctx->block_results--;
  GNUNET_thread_stop_sleep (ctx->thread);
}

void
GNUNET_FS_suspend_search_context (struct GNUNET_FS_SearchContext *ctx)
{
  GNUNET_mutex_lock (ctx->lock);
  ctx->block_results++;
  GNUNET_mutex_unlock (ctx->lock);
}


void
GNUNET_FS_destroy_search_context (struct GNUNET_FS_SearchContext *ctx)
{
  void *unused;
  struct GNUNET_FS_SearchHandle *pos;

  ctx->abort = GNUNET_YES;
  GNUNET_client_connection_close_forever (ctx->sock);
  GNUNET_thread_stop_sleep (ctx->thread);
  GNUNET_thread_join (ctx->thread, &unused);
  GNUNET_client_connection_destroy (ctx->sock);
  while (ctx->handles != NULL)
    {
      pos = ctx->handles;
      ctx->handles = pos->next;
      GNUNET_free (pos);
    }
  GNUNET_mutex_destroy (ctx->lock);
  GNUNET_free (ctx);
}