pq_event.c

Go to the documentation of this file.

/*
   This file is part of GNUnet
   Copyright (C) 2021, 2023 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 "pq.h"
#include <pthread.h>
 
 
struct GNUNET_DB_EventHandler
{
  struct GNUNET_ShortHashCode sh;
 
  GNUNET_DB_EventCallback cb;
 
  void *cb_cls;
 
  struct GNUNET_PQ_Context *db;
 
  struct GNUNET_SCHEDULER_Task *timeout_task;
};
 
 
static void
es_to_sh (const struct GNUNET_DB_EventHeaderP *es,
          struct GNUNET_ShortHashCode *sh)
{
  struct GNUNET_HashCode h_channel;
 
  GNUNET_CRYPTO_hash (es,
                      ntohs (es->size),
                      &h_channel);
  GNUNET_static_assert (sizeof (*sh) <= sizeof (h_channel));
  memcpy (sh,
          &h_channel,
          sizeof (*sh));
}
 
 
static char *
sh_to_channel (struct GNUNET_ShortHashCode *sh,
               char identifier[64])
{
  char *end;
 
  end = GNUNET_STRINGS_data_to_string (sh,
                                       sizeof (*sh),
                                       identifier,
                                       63);
  GNUNET_assert (NULL != end);
  *end = '\0';
  return end;
}
 
 
static enum GNUNET_GenericReturnValue
channel_to_sh (const char *identifier,
               struct GNUNET_ShortHashCode *sh)
{
  return GNUNET_STRINGS_string_to_data (identifier,
                                        strlen (identifier),
                                        sh,
                                        sizeof (*sh));
}
 
 
static char *
es_to_channel (const struct GNUNET_DB_EventHeaderP *es,
               char identifier[64])
{
  struct GNUNET_ShortHashCode sh;
 
  es_to_sh (es,
            &sh);
  return sh_to_channel (&sh,
                        identifier);
}
 
 
struct NotifyContext
{
  void *extra;
 
  size_t extra_size;
};
 
 
static enum GNUNET_GenericReturnValue
do_notify (void *cls,
           const struct GNUNET_ShortHashCode *sh,
           void *value)
{
  struct NotifyContext *ctx = cls;
  struct GNUNET_DB_EventHandler *eh = value;
 
  eh->cb (eh->cb_cls,
          ctx->extra,
          ctx->extra_size);
  return GNUNET_OK;
}
 
 
void
GNUNET_PQ_event_do_poll (struct GNUNET_PQ_Context *db)
{
  PGnotify *n;
  unsigned int cnt = 0;
 
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "PG poll job active\n");
  if (1 !=
      PQconsumeInput (db->conn))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to read from Postgres: %s\n",
                PQerrorMessage (db->conn));
    if (CONNECTION_BAD != PQstatus (db->conn))
      return;
    GNUNET_PQ_reconnect (db);
    return;
  }
  while (NULL != (n = PQnotifies (db->conn)))
  {
    struct GNUNET_ShortHashCode sh;
    struct NotifyContext ctx = {
      .extra = NULL
    };
 
    cnt++;
    if ('X' != toupper ((int) n->relname[0]))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Ignoring notification for unsupported channel identifier `%s'\n",
                  n->relname);
      PQfreemem (n);
      continue;
    }
    if (GNUNET_OK !=
        channel_to_sh (&n->relname[1],
                       &sh))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Ignoring notification for unsupported channel identifier `%s'\n",
                  n->relname);
      PQfreemem (n);
      continue;
    }
    if ( (NULL != n->extra) &&
         (GNUNET_OK !=
          GNUNET_STRINGS_string_to_data_alloc (n->extra,
                                               strlen (n->extra),
                                               &ctx.extra,
                                               &ctx.extra_size)))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Ignoring notification for unsupported extra data `%s' on channel `%s'\n",
                  n->extra,
                  n->relname);
      PQfreemem (n);
      continue;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Received notification %s with extra data `%.*s'\n",
                n->relname,
                (int) ctx.extra_size,
                (const char *) ctx.extra);
    GNUNET_CONTAINER_multishortmap_get_multiple (db->channel_map,
                                                 &sh,
                                                 &do_notify,
                                                 &ctx);
    GNUNET_free (ctx.extra);
    PQfreemem (n);
  }
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "PG poll job finishes after %u events\n",
              cnt);
}
 
 
static void
do_scheduler_notify (void *cls)
{
  struct GNUNET_PQ_Context *db = cls;
 
  db->event_task = NULL;
  if (NULL == db->rfd)
    GNUNET_PQ_reconnect (db);
  GNUNET_PQ_event_do_poll (db);
  if (NULL != db->event_task)
    return;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Resubscribing\n");
  if (NULL == db->rfd)
  {
    db->resubscribe_backoff
      = GNUNET_TIME_relative_max (db->resubscribe_backoff,
                                  GNUNET_TIME_UNIT_SECONDS);
    db->resubscribe_backoff
      = GNUNET_TIME_STD_BACKOFF (db->resubscribe_backoff);
    db->event_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
                                                   &do_scheduler_notify,
                                                   db);
    return;
  }
  db->resubscribe_backoff = GNUNET_TIME_UNIT_SECONDS;
  db->event_task
    = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                     db->rfd,
                                     &do_scheduler_notify,
                                     db);
}
 
 
static void
scheduler_fd_cb (void *cls,
                 int fd)
{
  struct GNUNET_PQ_Context *db = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "New poll FD is %d\n",
              fd);
  if (NULL != db->event_task)
  {
    GNUNET_SCHEDULER_cancel (db->event_task);
    db->event_task = NULL;
  }
  GNUNET_free (db->rfd);
  if (-1 == fd)
    return;
  if (0 == GNUNET_CONTAINER_multishortmap_size (db->channel_map))
    return;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Activating poll job on %d\n",
              fd);
  db->rfd = GNUNET_NETWORK_socket_box_native (fd);
  db->event_task
    = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_ZERO,
                                     db->rfd,
                                     &do_scheduler_notify,
                                     db);
}
 
 
static void
manage_subscribe (struct GNUNET_PQ_Context *db,
                  const char *cmd,
                  struct GNUNET_DB_EventHandler *eh)
{
  char sql[16 + 64];
  char *end;
  PGresult *result;
 
  if (NULL == db->conn)
    return;
  end = stpcpy (sql,
                cmd);
  end = sh_to_channel (&eh->sh,
                       end);
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Executing PQ command `%s'\n",
              sql);
  result = PQexec (db->conn,
                   sql);
  if (PGRES_COMMAND_OK != PQresultStatus (result))
  {
    GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
                     "pq",
                     "Failed to execute `%s': %s/%s/%s/%s/%s",
                     sql,
                     PQresultErrorField (result,
                                         PG_DIAG_MESSAGE_PRIMARY),
                     PQresultErrorField (result,
                                         PG_DIAG_MESSAGE_DETAIL),
                     PQresultErrorMessage (result),
                     PQresStatus (PQresultStatus (result)),
                     PQerrorMessage (db->conn));
  }
  PQclear (result);
}
 
 
static enum GNUNET_GenericReturnValue
register_notify (void *cls,
                 const struct GNUNET_ShortHashCode *sh,
                 void *value)
{
  struct GNUNET_PQ_Context *db = cls;
  struct GNUNET_DB_EventHandler *eh = value;
 
  manage_subscribe (db,
                    "LISTEN X",
                    eh);
  return GNUNET_OK;
}
 
 
void
GNUNET_PQ_event_reconnect_ (struct GNUNET_PQ_Context *db,
                            int fd)
{
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Change in PQ event FD to %d\n",
              fd);
  scheduler_fd_cb (db,
                   fd);
  GNUNET_CONTAINER_multishortmap_iterate (db->channel_map,
                                          &register_notify,
                                          db);
}
 
 
static void
event_timeout (void *cls)
{
  struct GNUNET_DB_EventHandler *eh = cls;
 
  eh->timeout_task = NULL;
  eh->cb (eh->cb_cls,
          NULL,
          0);
}
 
 
struct GNUNET_DB_EventHandler *
GNUNET_PQ_event_listen (struct GNUNET_PQ_Context *db,
                        const struct GNUNET_DB_EventHeaderP *es,
                        struct GNUNET_TIME_Relative timeout,
                        GNUNET_DB_EventCallback cb,
                        void *cb_cls)
{
  struct GNUNET_DB_EventHandler *eh;
  bool sub;
 
  eh = GNUNET_new (struct GNUNET_DB_EventHandler);
  eh->db = db;
  es_to_sh (es,
            &eh->sh);
  eh->cb = cb;
  eh->cb_cls = cb_cls;
  sub = (NULL ==
         GNUNET_CONTAINER_multishortmap_get (db->channel_map,
                                             &eh->sh));
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multishortmap_put (db->channel_map,
                                                     &eh->sh,
                                                     eh,
                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
  if (NULL == db->event_task)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Starting event scheduler\n");
    scheduler_fd_cb (db,
                     PQsocket (db->conn));
  }
  if (sub)
    manage_subscribe (db,
                      "LISTEN X",
                      eh);
  eh->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout,
                                                   &event_timeout,
                                                   eh);
  return eh;
}
 
 
void
GNUNET_PQ_event_listen_cancel (struct GNUNET_DB_EventHandler *eh)
{
  struct GNUNET_PQ_Context *db = eh->db;
 
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multishortmap_remove (db->channel_map,
                                                        &eh->sh,
                                                        eh));
  if (NULL ==
      GNUNET_CONTAINER_multishortmap_get (db->channel_map,
                                          &eh->sh))
    manage_subscribe (db,
                      "UNLISTEN X",
                      eh);
  if (0 == GNUNET_CONTAINER_multishortmap_size (db->channel_map))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Stopping PQ event scheduler job\n");
    GNUNET_free (db->rfd);
    if (NULL != db->event_task)
    {
      GNUNET_SCHEDULER_cancel (db->event_task);
      db->event_task = NULL;
    }
  }
  if (NULL != eh->timeout_task)
  {
    GNUNET_SCHEDULER_cancel (eh->timeout_task);
    eh->timeout_task = NULL;
  }
  GNUNET_free (eh);
}
 
 
char *
GNUNET_PQ_get_event_notify_channel (const struct GNUNET_DB_EventHeaderP *es)
{
  char sql[16 + 64 + 8];
  char *end;
 
  end = stpcpy (sql,
                "X");
  end = es_to_channel (es,
                       end);
  GNUNET_assert (NULL != end);
  return GNUNET_strdup (sql);
}
 
 
void
GNUNET_PQ_event_notify (struct GNUNET_PQ_Context *db,
                        const struct GNUNET_DB_EventHeaderP *es,
                        const void *extra,
                        size_t extra_size)
{
  char sql[16 + 64 + extra_size * 8 / 5 + 8];
  char *end;
  PGresult *result;
 
  end = stpcpy (sql,
                "NOTIFY X");
  end = es_to_channel (es,
                       end);
  end = stpcpy (end,
                ", '");
  end = GNUNET_STRINGS_data_to_string (extra,
                                       extra_size,
                                       end,
                                       sizeof (sql) - (end - sql) - 1);
  GNUNET_assert (NULL != end);
  *end = '\0';
  end = stpcpy (end,
                "'");
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Executing command `%s'\n",
              sql);
  result = PQexec (db->conn,
                   sql);
  if (PGRES_COMMAND_OK != PQresultStatus (result))
  {
    GNUNET_log_from (GNUNET_ERROR_TYPE_ERROR,
                     "pq",
                     "Failed to execute `%s': %s/%s/%s/%s/%s",
                     sql,
                     PQresultErrorField (result,
                                         PG_DIAG_MESSAGE_PRIMARY),
                     PQresultErrorField (result,
                                         PG_DIAG_MESSAGE_DETAIL),
                     PQresultErrorMessage (result),
                     PQresStatus (PQresultStatus (result)),
                     PQerrorMessage (db->conn));
  }
  PQclear (result);
  GNUNET_PQ_event_do_poll (db);
}
 
 
/* end of pq_event.c */