scalarproduct_api.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2013, 2014, 2016 GNUnet e.V.
 
     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     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
     Affero General Public License for more details.
 
     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_statistics_service.h"
#include "gnunet_scalarproduct_service.h"
#include "gnunet_protocols.h"
#include "scalarproduct.h"
 
#define LOG(kind, ...) GNUNET_log_from (kind, "scalarproduct-api", __VA_ARGS__)
 
 
typedef void
(*GNUNET_SCALARPRODUCT_ResponseMessageHandler) (
  struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
  const struct ClientResponseMessage *msg,
  enum GNUNET_SCALARPRODUCT_ResponseStatus status);
 
 
struct GNUNET_SCALARPRODUCT_ComputationHandle
{
  const struct GNUNET_CONFIGURATION_Handle *cfg;
 
  struct GNUNET_MQ_Handle *mq;
 
  GNUNET_SCALARPRODUCT_ContinuationWithStatus cont_status;
 
  GNUNET_SCALARPRODUCT_DatumProcessor cont_datum;
 
  void *cont_cls;
 
  GNUNET_SCALARPRODUCT_ResponseMessageHandler response_proc;
 
  struct GNUNET_HashCode key;
};
 
 
static int
check_response (void *cls,
                const struct ClientResponseMessage *message)
{
  if (ntohs (message->header.size) !=
      ntohl (message->product_length) + sizeof(struct ClientResponseMessage))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
process_status_message (struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
                        const struct ClientResponseMessage *msg,
                        enum GNUNET_SCALARPRODUCT_ResponseStatus status)
{
  if (NULL != h->cont_status)
    h->cont_status (h->cont_cls,
                    status);
  GNUNET_SCALARPRODUCT_cancel (h);
}
 
 
static void
handle_response (void *cls,
                 const struct ClientResponseMessage *message)
{
  struct GNUNET_SCALARPRODUCT_ComputationHandle *h = cls;
  enum GNUNET_SCALARPRODUCT_ResponseStatus status;
 
  status = (enum GNUNET_SCALARPRODUCT_ResponseStatus) ntohl (message->status);
  h->response_proc (h,
                    message,
                    status);
}
 
 
static int
check_unique (const struct GNUNET_SCALARPRODUCT_Element *elements,
              uint32_t element_count)
{
  struct GNUNET_CONTAINER_MultiHashMap *map;
  int ok;
 
  ok = GNUNET_OK;
  map = GNUNET_CONTAINER_multihashmap_create (2 * element_count,
                                              GNUNET_YES);
  for (uint32_t i = 0; i < element_count; i++)
    if (GNUNET_OK !=
        GNUNET_CONTAINER_multihashmap_put (map,
                                           &elements[i].key,
                                           map,
                                           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  _ ("Keys given to SCALARPRODUCT not unique!\n"));
      ok = GNUNET_SYSERR;
    }
  GNUNET_CONTAINER_multihashmap_destroy (map);
  return ok;
}
 
 
static void
mq_error_handler (void *cls,
                  enum GNUNET_MQ_Error error)
{
  struct GNUNET_SCALARPRODUCT_ComputationHandle *h = cls;
 
  LOG (GNUNET_ERROR_TYPE_INFO,
       "Disconnected from SCALARPRODUCT service.\n");
  h->response_proc (h,
                    NULL,
                    GNUNET_SCALARPRODUCT_STATUS_DISCONNECTED);
}
 
 
struct GNUNET_SCALARPRODUCT_ComputationHandle *
GNUNET_SCALARPRODUCT_accept_computation (
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  const struct GNUNET_HashCode *session_key,
  const struct GNUNET_SCALARPRODUCT_Element *elements,
  uint32_t element_count,
  GNUNET_SCALARPRODUCT_ContinuationWithStatus cont,
  void *cont_cls)
{
  struct GNUNET_SCALARPRODUCT_ComputationHandle *h
    = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_var_size (response,
                           GNUNET_MESSAGE_TYPE_SCALARPRODUCT_RESULT,
                           struct ClientResponseMessage,
                           h),
    GNUNET_MQ_handler_end ()
  };
  struct GNUNET_MQ_Envelope *env;
  struct BobComputationMessage *msg;
  struct ComputationBobCryptodataMultipartMessage *mmsg;
  uint32_t size;
  uint16_t possible;
  uint16_t todo;
  uint32_t element_count_transfered;
 
 
  if (GNUNET_SYSERR == check_unique (elements,
                                     element_count))
    return NULL;
  h->cont_status = cont;
  h->cont_cls = cont_cls;
  h->response_proc = &process_status_message;
  h->cfg = cfg;
  h->key = *session_key;
  h->mq = GNUNET_CLIENT_connect (cfg,
                                 "scalarproduct-bob",
                                 handlers,
                                 &mq_error_handler,
                                 h);
  if (NULL == h->mq)
  {
    /* scalarproduct configuration error */
    GNUNET_break (0);
    GNUNET_free (h);
    return NULL;
  }
  possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof(struct
                                                   BobComputationMessage))
             / sizeof(struct GNUNET_SCALARPRODUCT_Element);
  todo = GNUNET_MIN (possible,
                     element_count);
  size = todo * sizeof(struct GNUNET_SCALARPRODUCT_Element);
  env = GNUNET_MQ_msg_extra (msg,
                             size,
                             GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_BOB);
  msg->element_count_total = htonl (element_count);
  msg->element_count_contained = htonl (todo);
  msg->session_key = *session_key;
  GNUNET_memcpy (&msg[1],
                 elements,
                 size);
  element_count_transfered = todo;
  GNUNET_MQ_send (h->mq,
                  env);
  possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof(*mmsg))
             / sizeof(struct GNUNET_SCALARPRODUCT_Element);
  while (element_count_transfered < element_count)
  {
    todo = GNUNET_MIN (possible,
                       element_count - element_count_transfered);
    size = todo * sizeof(struct GNUNET_SCALARPRODUCT_Element);
    env = GNUNET_MQ_msg_extra (mmsg,
                               size,
                               GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_MULTIPART_BOB);
    mmsg->element_count_contained = htonl (todo);
    GNUNET_memcpy (&mmsg[1],
                   &elements[element_count_transfered],
                   size);
    element_count_transfered += todo;
    GNUNET_MQ_send (h->mq,
                    env);
  }
  return h;
}
 
 
static void
process_result_message (struct GNUNET_SCALARPRODUCT_ComputationHandle *h,
                        const struct ClientResponseMessage *msg,
                        enum GNUNET_SCALARPRODUCT_ResponseStatus status)
{
  uint32_t product_len;
  gcry_mpi_t result = NULL;
  gcry_error_t rc;
  gcry_mpi_t num;
  size_t rsize;
 
  if (GNUNET_SCALARPRODUCT_STATUS_SUCCESS == status)
  {
    result = gcry_mpi_new (0);
 
    product_len = ntohl (msg->product_length);
    if (0 < product_len)
    {
      rsize = 0;
      if (0 != (rc = gcry_mpi_scan (&num, GCRYMPI_FMT_STD,
                                    &msg[1],
                                    product_len,
                                    &rsize)))
      {
        LOG_GCRY (GNUNET_ERROR_TYPE_ERROR,
                  "gcry_mpi_scan",
                  rc);
        gcry_mpi_release (result);
        result = NULL;
        status = GNUNET_SCALARPRODUCT_STATUS_INVALID_RESPONSE;
      }
      else
      {
        if (0 < (int32_t) ntohl (msg->range))
          gcry_mpi_add (result, result, num);
        else
          gcry_mpi_sub (result, result, num);
        gcry_mpi_release (num);
      }
    }
  }
  if (NULL != h->cont_datum)
    h->cont_datum (h->cont_cls,
                   status,
                   result);
  if (NULL != result)
    gcry_mpi_release (result);
  GNUNET_SCALARPRODUCT_cancel (h);
}
 
 
struct GNUNET_SCALARPRODUCT_ComputationHandle *
GNUNET_SCALARPRODUCT_start_computation (
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  const struct GNUNET_HashCode *session_key,
  const struct GNUNET_PeerIdentity *peer,
  const struct GNUNET_SCALARPRODUCT_Element *elements,
  uint32_t element_count,
  GNUNET_SCALARPRODUCT_DatumProcessor cont,
  void *cont_cls)
{
  struct GNUNET_SCALARPRODUCT_ComputationHandle *h
    = GNUNET_new (struct GNUNET_SCALARPRODUCT_ComputationHandle);
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_var_size (response,
                           GNUNET_MESSAGE_TYPE_SCALARPRODUCT_RESULT,
                           struct ClientResponseMessage,
                           h),
    GNUNET_MQ_handler_end ()
  };
  struct GNUNET_MQ_Envelope *env;
  struct AliceComputationMessage *msg;
  struct ComputationBobCryptodataMultipartMessage *mmsg;
  uint32_t size;
  uint16_t possible;
  uint16_t todo;
  uint32_t element_count_transfered;
 
  if (GNUNET_SYSERR == check_unique (elements,
                                     element_count))
    return NULL;
  h->mq = GNUNET_CLIENT_connect (cfg,
                                 "scalarproduct-alice",
                                 handlers,
                                 &mq_error_handler,
                                 h);
  if (NULL == h->mq)
  {
    /* misconfigured scalarproduct service */
    GNUNET_break (0);
    GNUNET_free (h);
    return NULL;
  }
  h->cont_datum = cont;
  h->cont_cls = cont_cls;
  h->response_proc = &process_result_message;
  h->cfg = cfg;
  h->key = *session_key;
 
  possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof(struct
                                                   AliceComputationMessage))
             / sizeof(struct GNUNET_SCALARPRODUCT_Element);
  todo = GNUNET_MIN (possible,
                     element_count);
  size = todo * sizeof(struct GNUNET_SCALARPRODUCT_Element);
  env = GNUNET_MQ_msg_extra (msg,
                             size,
                             GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_ALICE);
  msg->element_count_total = htonl (element_count);
  msg->element_count_contained = htonl (todo);
  msg->reserved = htonl (0);
  msg->peer = *peer;
  msg->session_key = *session_key;
  GNUNET_memcpy (&msg[1],
                 elements,
                 size);
  GNUNET_MQ_send (h->mq,
                  env);
  element_count_transfered = todo;
  possible = (GNUNET_MAX_MESSAGE_SIZE - 1 - sizeof(*mmsg))
             / sizeof(struct GNUNET_SCALARPRODUCT_Element);
  while (element_count_transfered < element_count)
  {
    todo = GNUNET_MIN (possible,
                       element_count - element_count_transfered);
    size = todo * sizeof(struct GNUNET_SCALARPRODUCT_Element);
    env = GNUNET_MQ_msg_extra (mmsg,
                               size,
                               GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_MULTIPART_ALICE);
    mmsg->element_count_contained = htonl (todo);
    GNUNET_memcpy (&mmsg[1],
                   &elements[element_count_transfered],
                   size);
    element_count_transfered += todo;
    GNUNET_MQ_send (h->mq,
                    env);
  }
  return h;
}
 
 
void
GNUNET_SCALARPRODUCT_cancel (struct GNUNET_SCALARPRODUCT_ComputationHandle *h)
{
  if (NULL != h->mq)
  {
    GNUNET_MQ_destroy (h->mq);
    h->mq = NULL;
  }
  GNUNET_free (h);
}
 
 
/* end of scalarproduct_api.c */