<commit_before>/* -*- mode: C; c-file-style: "gnu"; indent-tabs-mode: nil; -*-
 *
 * Copyright (C) 2011,2012,2013 <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 *
 * Author: <NAME> <<EMAIL>>
 */

/**
 * See:
 * https://mail.gnome.org/archives/ostree-list/2012-August/msg00021.html
 *
 * DESIGN:
 *
 * Pull refs
 *   For each ref:
 *     Queue scan of commit
 *
 * Mainloop:
 *  Process requests, await idle scan
 *  
 * Async queue:
 *  Scan commit
 *   If already cached, recursively scan content
 *   If not, queue fetch
 * 
 *  For each commit:
 *    Verify checksum
 *    Import
 *    Traverse and queue dirtree/dirmeta
 * 
 * Pull dirtrees:
 *  For each dirtree:
 *    Verify checksum
 *    Import
 *    Traverse and queue content/dirtree/dirmeta
 *
 * Pull content meta:
 *  For each content:
 *    Pull meta
 *    If contentcontent needed:
 *      Queue contentcontent
 *    else:
 *      Import
 *
 * Pull contentcontent:
 *  For each contentcontent
 *    Verify checksum
 *    Import
 *    
 *  
 */

#include "config.h"

#include "ostree.h"
#include "ostree-fetcher.h"

typedef struct {
  enum {
    PULL_MSG_SCAN_IDLE,
    PULL_MSG_MAIN_IDLE,
    PULL_MSG_FETCH,
    PULL_MSG_SCAN,
    PULL_MSG_QUIT
  } t;
  union {
    guint     idle_serial;
    GVariant *item;
  } d;
} PullWorkerMessage;

typedef struct {
  OstreeRepo   *repo;
  OstreeRepoPullFlags flags;
  char         *remote_name;
  OstreeRepoMode remote_mode;
  OstreeFetcher *fetcher;
  SoupURI      *base_uri;

  GMainContext    *main_context;
  GMainLoop    *loop;
  GCancellable *cancellable;

  gboolean      transaction_resuming;
  volatile gint n_scanned_metadata;
  guint         outstanding_uri_requests;

  GThread          *metadata_thread;
  GMainContext     *metadata_thread_context;
  GMainLoop        *metadata_thread_loop;
  OtWaitableQueue  *metadata_objects_to_scan;
  OtWaitableQueue  *metadata_objects_to_fetch;
  GHashTable       *scanned_metadata; /* Maps object name to itself */
  GHashTable       *requested_metadata; /* Maps object name to itself */
  GHashTable       *requested_content; /* Maps object name to itself */
  guint             metadata_scan_idle : 1; /* TRUE if we passed through an idle message */
  guint             idle_serial; /* Incremented when we get a SCAN_IDLE message */
  guint             n_outstanding_metadata_fetches;
  guint             n_outstanding_metadata_stage_requests;
  guint             n_outstanding_content_fetches;
  guint             n_outstanding_content_stage_requests;
  gint              n_requested_metadata;
  gint              n_requested_content;
  guint             n_fetched_metadata;
  guint             n_fetched_content;

  gboolean      have_previous_bytes;
  guint64       previous_bytes_sec;
  guint64       previous_total_downloaded;

  GError      **async_error;
  gboolean      caught_error;
} OtPullData;

typedef struct {
  OtPullData  *pull_data;
  GVariant    *object;
  GFile       *temp_path;
} FetchObjectData;

static SoupURI *
suburi_new (SoupURI   *base,
            const char *first,
            ...) G_GNUC_NULL_TERMINATED;

static gboolean scan_one_metadata_object (OtPullData         *pull_data,
                                          const guchar       *csum,
                                          OstreeObjectType    objtype,
                                          guint               recursion_depth,
                                          GCancellable       *cancellable,
                                          GError            **error);
static gboolean scan_one_metadata_object_v_name (OtPullData         *pull_data,
                                                 GVariant           *object,
                                                 GCancellable       *cancellable,
                                                 GError            **error);


static SoupURI *
suburi_new (SoupURI   *base,
            const char *first,
            ...)
{
  va_list args;
  GPtrArray *arg_array;
  const char *arg;
  char *subpath;
  SoupURI *ret;

  arg_array = g_ptr_array_new ();
  g_ptr_array_add (arg_array, (char*)soup_uri_get_path (base));
  g_ptr_array_add (arg_array, (char*)first);

  va_start (args, first);
  
  while ((arg = va_arg (args, const char *)) != NULL)
    g_ptr_array_add (arg_array, (char*)arg);
  g_ptr_array_add (arg_array, NULL);

  subpath = g_build_filenamev ((char**)arg_array->pdata);
  g_ptr_array_unref (arg_array);
  
  ret = soup_uri_copy (base);
  soup_uri_set_path (ret, subpath);
  g_free (subpath);
  
  va_end (args);
  
  return ret;
}

static gboolean
uri_fetch_update_status (gpointer user_data)
{
  OtPullData *pull_data = user_data;
  gs_free char *fetcher_status = NULL;
  GString *status;
  guint outstanding_stages;
  guint outstanding_fetches;
 
  status = g_string_new ("");

  if (!pull_data->metadata_scan_idle)
    g_string_append_printf (status, "scan: %u; ",
                            g_atomic_int_get (&pull_data->n_scanned_metadata));

  outstanding_stages = pull_data->n_outstanding_content_stage_requests + pull_data->n_outstanding_metadata_stage_requests;
  if (outstanding_stages > 0)
    g_string_append_printf (status, "writing: %u; ", outstanding_stages);

  outstanding_fetches = pull_data->n_outstanding_content_fetches + pull_data->n_outstanding_metadata_fetches;
  if (outstanding_fetches)
    {
      g_string_append_printf (status, "fetch: %u/%u meta %u/%u; ",
                              pull_data->n_fetched_metadata,
                              pull_data->n_requested_metadata,
                              pull_data->n_fetched_content,
                              pull_data->n_requested_content);

      fetcher_status = ostree_fetcher_query_state_text (pull_data->fetcher);
      g_string_append (status, fetcher_status);
    }

  gs_console_begin_status_line (gs_console_get (), status->str, NULL, NULL);

  g_string_free (status, TRUE);

  return TRUE;
}

static PullWorkerMessage *
pull_worker_message_new (int msgtype, gpointer data)
{
  PullWorkerMessage *msg = g_new (PullWorkerMessage, 1);
  msg->t = msgtype;
  switch (msgtype)
    {
    case PULL_MSG_SCAN_IDLE:
    case PULL_MSG_MAIN_IDLE:
      msg->d.idle_serial = GPOINTER_TO_UINT (data);
      break;
    case PULL_MSG_SCAN:
    case PULL_MSG_FETCH:<commit_msg>pull: Tweak status line again, do what git does<commit_after>/* -*- mode: C; c-file-style: "gnu"; indent-tabs-mode: nil; -*-
 *
 * Copyright (C) 2011,2012,2013 <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 *
 * Author: <NAME> <<EMAIL>>
 */

/**
 * See:
 * https://mail.gnome.org/archives/ostree-list/2012-August/msg00021.html
 *
 * DESIGN:
 *
 * Pull refs
 *   For each ref:
 *     Queue scan of commit
 *
 * Mainloop:
 *  Process requests, await idle scan
 *  
 * Async queue:
 *  Scan commit
 *   If already cached, recursively scan content
 *   If not, queue fetch
 * 
 *  For each commit:
 *    Verify checksum
 *    Import
 *    Traverse and queue dirtree/dirmeta
 * 
 * Pull dirtrees:
 *  For each dirtree:
 *    Verify checksum
 *    Import
 *    Traverse and queue content/dirtree/dirmeta
 *
 * Pull content meta:
 *  For each content:
 *    Pull meta
 *    If contentcontent needed:
 *      Queue contentcontent
 *    else:
 *      Import
 *
 * Pull contentcontent:
 *  For each contentcontent
 *    Verify checksum
 *    Import
 *    
 *  
 */

#include "config.h"

#include "ostree.h"
#include "ostree-fetcher.h"

typedef struct {
  enum {
    PULL_MSG_SCAN_IDLE,
    PULL_MSG_MAIN_IDLE,
    PULL_MSG_FETCH,
    PULL_MSG_SCAN,
    PULL_MSG_QUIT
  } t;
  union {
    guint     idle_serial;
    GVariant *item;
  } d;
} PullWorkerMessage;

typedef struct {
  OstreeRepo   *repo;
  OstreeRepoPullFlags flags;
  char         *remote_name;
  OstreeRepoMode remote_mode;
  OstreeFetcher *fetcher;
  SoupURI      *base_uri;

  GMainContext    *main_context;
  GMainLoop    *loop;
  GCancellable *cancellable;

  gboolean      transaction_resuming;
  volatile gint n_scanned_metadata;
  guint         outstanding_uri_requests;

  GThread          *metadata_thread;
  GMainContext     *metadata_thread_context;
  GMainLoop        *metadata_thread_loop;
  OtWaitableQueue  *metadata_objects_to_scan;
  OtWaitableQueue  *metadata_objects_to_fetch;
  GHashTable       *scanned_metadata; /* Maps object name to itself */
  GHashTable       *requested_metadata; /* Maps object name to itself */
  GHashTable       *requested_content; /* Maps object name to itself */
  guint             metadata_scan_idle : 1; /* TRUE if we passed through an idle message */
  guint             idle_serial; /* Incremented when we get a SCAN_IDLE message */
  guint             n_outstanding_metadata_fetches;
  guint             n_outstanding_metadata_stage_requests;
  guint             n_outstanding_content_fetches;
  guint             n_outstanding_content_stage_requests;
  gint              n_requested_metadata;
  gint              n_requested_content;
  guint             n_fetched_metadata;
  guint             n_fetched_content;

  gboolean      have_previous_bytes;
  guint64       previous_bytes_sec;
  guint64       previous_total_downloaded;

  GError      **async_error;
  gboolean      caught_error;
} OtPullData;

typedef struct {
  OtPullData  *pull_data;
  GVariant    *object;
  GFile       *temp_path;
} FetchObjectData;

static SoupURI *
suburi_new (SoupURI   *base,
            const char *first,
            ...) G_GNUC_NULL_TERMINATED;

static gboolean scan_one_metadata_object (OtPullData         *pull_data,
                                          const guchar       *csum,
                                          OstreeObjectType    objtype,
                                          guint               recursion_depth,
                                          GCancellable       *cancellable,
                                          GError            **error);
static gboolean scan_one_metadata_object_v_name (OtPullData         *pull_data,
                                                 GVariant           *object,
                                                 GCancellable       *cancellable,
                                                 GError            **error);


static SoupURI *
suburi_new (SoupURI   *base,
            const char *first,
            ...)
{
  va_list args;
  GPtrArray *arg_array;
  const char *arg;
  char *subpath;
  SoupURI *ret;

  arg_array = g_ptr_array_new ();
  g_ptr_array_add (arg_array, (char*)soup_uri_get_path (base));
  g_ptr_array_add (arg_array, (char*)first);

  va_start (args, first);
  
  while ((arg = va_arg (args, const char *)) != NULL)
    g_ptr_array_add (arg_array, (char*)arg);
  g_ptr_array_add (arg_array, NULL);

  subpath = g_build_filenamev ((char**)arg_array->pdata);
  g_ptr_array_unref (arg_array);
  
  ret = soup_uri_copy (base);
  soup_uri_set_path (ret, subpath);
  g_free (subpath);
  
  va_end (args);
  
  return ret;
}

static gboolean
uri_fetch_update_status (gpointer user_data)
{
  OtPullData *pull_data = user_data;
  gs_free char *fetcher_status = NULL;
  GString *status;
  guint outstanding_stages;
  guint outstanding_fetches;
 
  status = g_string_new ("");

  outstanding_fetches = pull_data->n_outstanding_content_fetches + pull_data->n_outstanding_metadata_fetches;
  outstanding_stages = pull_data->n_outstanding_content_stage_requests + pull_data->n_outstanding_metadata_stage_requests;

  if (outstanding_fetches)
    {
      guint64 bytes_transferred = ostree_fetcher_bytes_transferred (pull_data->fetcher);
      guint fetched = pull_data->n_fetched_metadata + pull_data->n_fetched_content;
      guint requested = pull_data->n_requested_metadata + pull_data->n_requested_content;
      gs_free char *formatted_bytes_transferred = NULL;

      formatted_bytes_transferred = g_format_size_full (bytes_transferred, 0);

      g_string_append_printf (status, "Receiving objects: %u%% (%u/%u) %s",
                              (guint)((((double)fetched) / requested) * 100),
                              fetched, requested, formatted_bytes_transferred);
    }
  else if (outstanding_stages > 0)
    g_string_append_printf (status, "Writing objects: %u", outstanding_stages);
  else if (!pull_data->metadata_scan_idle)
    g_string_append_printf (status, "Scanning metadata: %u",
                            g_atomic_int_get (&pull_data->n_scanned_metadata));
  else
    g_string_append_printf (status, "Idle");

  gs_console_begin_status_line (gs_console_get (), status->str, NULL, NULL);

  g_string_free (status, TRUE);

  return TRUE;
}

static PullWorkerMessage *
pull_worker_message_new (int msgtype, gpointer data)
{
  PullWorkerMessage *msg = g_new (PullWorkerMessage, 1);
  msg->t = msgtype;
  switch (msgtype)
    {
    case PULL_MSG_SCAN_IDLE:
    case PULL_MSG_MAIN_IDLE:
      msg->d.idle_serial = GPOINTER_TO_UINT (data);
      break;
    case PULL_MSG_SCAN:
    case PULL_MSG_FETCH:
{'fraction_non_alphanumeric': 0.07979910714285714, 'fraction_numerical': 0.006417410714285714, 'mean_word_length': 2.5148320666830104, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 47, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3376293', 'n_tokens_mistral': 4600, 'n_tokens_neox': 4445, 'n_words': 1110}