<commit_before>/* batch_fsync.c --- efficiently fsync multiple targets
 *
 * ====================================================================
 *    Licensed to the Apache Software Foundation (ASF) under one
 *    or more contributor license agreements.  See the NOTICE file
 *    distributed with this work for additional information
 *    regarding copyright ownership.  The ASF licenses this file
 *    to you under the Apache License, Version 2.0 (the
 *    "License"); you may not use this file except in compliance
 *    with the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing,
 *    software distributed under the License is distributed on an
 *    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *    KIND, either express or implied.  See the License for the
 *    specific language governing permissions and limitations
 *    under the License.
 * ====================================================================
 */

#include <apr_thread_pool.h>
#include <apr_thread_cond.h>

#include "batch_fsync.h"
#include "svn_pools.h"
#include "svn_hash.h"
#include "svn_dirent_uri.h"
#include "svn_private_config.h"

#include "private/svn_dep_compat.h"
#include "private/svn_mutex.h"
#include "private/svn_subr_private.h"

/* Handy macro to check APR function results and turning them into
 * svn_error_t upon failure. */
#define WRAP_APR_ERR(x,msg)                     \
  {                                             \
    apr_status_t status_ = (x);                 \
    if (status_)                                \
      return svn_error_wrap_apr(status_, msg);  \
  }


/* A simple SVN-wrapper around the apr_thread_cond_* API */
#if APR_HAS_THREADS
typedef apr_thread_cond_t svn_thread_cond__t;
#else
typedef int svn_thread_cond__t;
#endif

static svn_error_t *
svn_thread_cond__create(svn_thread_cond__t **cond,
                        apr_pool_t *result_pool)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_create(cond, result_pool),
               _("Can't create condition variable"));

#else

  *cond = apr_pcalloc(result_pool, sizeof(**cond));

#endif

  return SVN_NO_ERROR;
}

static svn_error_t *
svn_thread_cond__broadcast(svn_thread_cond__t *cond)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_broadcast(cond),
               _("Can't broadcast condition variable"));

#endif

  return SVN_NO_ERROR;
}

static svn_error_t *
svn_thread_cond__wait(svn_thread_cond__t *cond,
                      svn_mutex__t *mutex)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_wait(cond, svn_mutex__get(mutex)),
               _("Can't broadcast condition variable"));

#endif

  return SVN_NO_ERROR;
}

/* Utility construct:  Clients can efficiently wait for the encapsulated
 * counter to reach a certain value.  Currently, only increments have been
 * implemented.  This whole structure can be opaque to the API users.
 */
typedef struct waitable_counter_t
{
  /* Current value, initialized to 0. */
  int value;

  /* Synchronization objects. */
  svn_thread_cond__t *cond;
  svn_mutex__t *mutex;
} waitable_counter_t;

/* Set *COUNTER_P to a new waitable_counter_t instance allocated in
 * RESULT_POOL.  The initial counter value is 0. */
static svn_error_t *
waitable_counter__create(waitable_counter_t **counter_p,
                         apr_pool_t *result_pool)
{
  waitable_counter_t *counter = apr_pcalloc(result_pool, sizeof(*counter));
  counter->value = 0;

  SVN_ERR(svn_thread_cond__create(&counter->cond, result_pool));
  SVN_ERR(svn_mutex__init(&counter->mutex, TRUE, result_pool));

  *counter_p = counter;

  return SVN_NO_ERROR;
}

/* Increment the value in COUNTER by 1. */
static svn_error_t *
waitable_counter__increment(waitable_counter_t *counter)
{
  SVN_ERR(svn_mutex__lock(counter->mutex));
  counter->value++;

  SVN_ERR(svn_thread_cond__broadcast(counter->cond));
  SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));

  return SVN_NO_ERROR;
}

/* Efficiently wait for COUNTER to assume VALUE. */
static svn_error_t *
waitable_counter__wait_for(waitable_counter_t *counter,
                           int value)
{
  svn_boolean_t done = FALSE;

  /* This loop implicitly handles spurious wake-ups. */
  do
    {
      SVN_ERR(svn_mutex__lock(counter->mutex));

      if (counter->value == value)
        done = TRUE;
      else
        SVN_ERR(svn_thread_cond__wait(counter->cond, counter->mutex));

      SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));
    }
  while (!done);

  return SVN_NO_ERROR;
}

/* Set the value in COUNTER to 0. */
static svn_error_t *
waitable_counter__reset(waitable_counter_t *counter)
{
  SVN_ERR(svn_mutex__lock(counter->mutex));
  counter->value = 0;
  SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));

  SVN_ERR(svn_thread_cond__broadcast(counter->cond));

  return SVN_NO_ERROR;
}

/* Entry type for the svn_fs_x__batch_fsync_t collection.  There is one
 * instance per file handle.
 */
typedef struct to_sync_t
{
  /* Open handle of the file / directory to fsync. */
  apr_file_t *file;

  /* Pool to use with FILE.  It is private to FILE such that it can be
   * used safely together with FILE in a separate thread. */
  apr_pool_t *pool;

  /* Result of the file operations. */
  svn_error_t *result;

  /* Counter to increment when we completed the task. */
  waitable_counter_t *counter;
} to_sync_t;

/* The actual collection object. */
struct svn_fs_x__batch_fsync_t
{
  /* Maps open file handles: C-string path to to_sync_t *. */
  apr_hash_t *files;

  /* Counts the number of completed fsync tasks. */
  waitable_counter_t *counter;
};

/* Data structures for concurrent fsync execution are only available if
 * we have threading support.
 */
#if APR_HAS_THREADS

/* Number of microseconds that an unused thread remains in the pool before
 * being terminated.
 *
 * Higher values are useful if clients frequently send small requests and
 * you want to minimize the latency for those.
 */
#define THREADPOOL_THREAD_IDLE_LIMIT 1000000

/* Maximum number of threads in THREAD_POOL, i.e. number of paths we can
 * fsync concurrently throughout the process. */
#define MAX_THREADS 16

/* Thread pool to execute the fsync tasks. */
static apr_thread_pool_t *thread_pool = NULL;

#endif

/* We open non-directory files with these flags. */
#define FILE_FLAGS (APR_READ | APR_WRITE | APR_BUFFERED | APR_CREATE)

/* Destructor function that implicitly cleans up any running threads
   in the thread_pool given as DATA and releases their memory pools
   before they get destroyed themselves.

   Must be run as a pre-cleanup hook.
 */
static apr_status_t
thread_pool_pre_cleanup(void *data)
{
  apr_thread_pool_t *tp = data;
  return apr_thread_pool_destroy(tp);
}

svn_error_t *
svn_fs_x__batch_fsync_init(void)
{
#if APR_HAS_THREADS
  /* The thread-pool must be allocated from a thread-safe pool.
     GLOBAL_POOL may be single-threaded, though. */
  apr_pool_t *pool = svn_pool_create(NULL);
<commit_msg>Follow-up to r1694846: One code section wasn't marked as conditional at all before.<commit_after>/* batch_fsync.c --- efficiently fsync multiple targets
 *
 * ====================================================================
 *    Licensed to the Apache Software Foundation (ASF) under one
 *    or more contributor license agreements.  See the NOTICE file
 *    distributed with this work for additional information
 *    regarding copyright ownership.  The ASF licenses this file
 *    to you under the Apache License, Version 2.0 (the
 *    "License"); you may not use this file except in compliance
 *    with the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing,
 *    software distributed under the License is distributed on an
 *    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *    KIND, either express or implied.  See the License for the
 *    specific language governing permissions and limitations
 *    under the License.
 * ====================================================================
 */

#include <apr_thread_pool.h>
#include <apr_thread_cond.h>

#include "batch_fsync.h"
#include "svn_pools.h"
#include "svn_hash.h"
#include "svn_dirent_uri.h"
#include "svn_private_config.h"

#include "private/svn_dep_compat.h"
#include "private/svn_mutex.h"
#include "private/svn_subr_private.h"

/* Handy macro to check APR function results and turning them into
 * svn_error_t upon failure. */
#define WRAP_APR_ERR(x,msg)                     \
  {                                             \
    apr_status_t status_ = (x);                 \
    if (status_)                                \
      return svn_error_wrap_apr(status_, msg);  \
  }


/* A simple SVN-wrapper around the apr_thread_cond_* API */
#if APR_HAS_THREADS
typedef apr_thread_cond_t svn_thread_cond__t;
#else
typedef int svn_thread_cond__t;
#endif

static svn_error_t *
svn_thread_cond__create(svn_thread_cond__t **cond,
                        apr_pool_t *result_pool)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_create(cond, result_pool),
               _("Can't create condition variable"));

#else

  *cond = apr_pcalloc(result_pool, sizeof(**cond));

#endif

  return SVN_NO_ERROR;
}

static svn_error_t *
svn_thread_cond__broadcast(svn_thread_cond__t *cond)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_broadcast(cond),
               _("Can't broadcast condition variable"));

#endif

  return SVN_NO_ERROR;
}

static svn_error_t *
svn_thread_cond__wait(svn_thread_cond__t *cond,
                      svn_mutex__t *mutex)
{
#if APR_HAS_THREADS

  WRAP_APR_ERR(apr_thread_cond_wait(cond, svn_mutex__get(mutex)),
               _("Can't broadcast condition variable"));

#endif

  return SVN_NO_ERROR;
}

/* Utility construct:  Clients can efficiently wait for the encapsulated
 * counter to reach a certain value.  Currently, only increments have been
 * implemented.  This whole structure can be opaque to the API users.
 */
typedef struct waitable_counter_t
{
  /* Current value, initialized to 0. */
  int value;

  /* Synchronization objects. */
  svn_thread_cond__t *cond;
  svn_mutex__t *mutex;
} waitable_counter_t;

/* Set *COUNTER_P to a new waitable_counter_t instance allocated in
 * RESULT_POOL.  The initial counter value is 0. */
static svn_error_t *
waitable_counter__create(waitable_counter_t **counter_p,
                         apr_pool_t *result_pool)
{
  waitable_counter_t *counter = apr_pcalloc(result_pool, sizeof(*counter));
  counter->value = 0;

  SVN_ERR(svn_thread_cond__create(&counter->cond, result_pool));
  SVN_ERR(svn_mutex__init(&counter->mutex, TRUE, result_pool));

  *counter_p = counter;

  return SVN_NO_ERROR;
}

/* Increment the value in COUNTER by 1. */
static svn_error_t *
waitable_counter__increment(waitable_counter_t *counter)
{
  SVN_ERR(svn_mutex__lock(counter->mutex));
  counter->value++;

  SVN_ERR(svn_thread_cond__broadcast(counter->cond));
  SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));

  return SVN_NO_ERROR;
}

/* Efficiently wait for COUNTER to assume VALUE. */
static svn_error_t *
waitable_counter__wait_for(waitable_counter_t *counter,
                           int value)
{
  svn_boolean_t done = FALSE;

  /* This loop implicitly handles spurious wake-ups. */
  do
    {
      SVN_ERR(svn_mutex__lock(counter->mutex));

      if (counter->value == value)
        done = TRUE;
      else
        SVN_ERR(svn_thread_cond__wait(counter->cond, counter->mutex));

      SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));
    }
  while (!done);

  return SVN_NO_ERROR;
}

/* Set the value in COUNTER to 0. */
static svn_error_t *
waitable_counter__reset(waitable_counter_t *counter)
{
  SVN_ERR(svn_mutex__lock(counter->mutex));
  counter->value = 0;
  SVN_ERR(svn_mutex__unlock(counter->mutex, SVN_NO_ERROR));

  SVN_ERR(svn_thread_cond__broadcast(counter->cond));

  return SVN_NO_ERROR;
}

/* Entry type for the svn_fs_x__batch_fsync_t collection.  There is one
 * instance per file handle.
 */
typedef struct to_sync_t
{
  /* Open handle of the file / directory to fsync. */
  apr_file_t *file;

  /* Pool to use with FILE.  It is private to FILE such that it can be
   * used safely together with FILE in a separate thread. */
  apr_pool_t *pool;

  /* Result of the file operations. */
  svn_error_t *result;

  /* Counter to increment when we completed the task. */
  waitable_counter_t *counter;
} to_sync_t;

/* The actual collection object. */
struct svn_fs_x__batch_fsync_t
{
  /* Maps open file handles: C-string path to to_sync_t *. */
  apr_hash_t *files;

  /* Counts the number of completed fsync tasks. */
  waitable_counter_t *counter;
};

/* Data structures for concurrent fsync execution are only available if
 * we have threading support.
 */
#if APR_HAS_THREADS

/* Number of microseconds that an unused thread remains in the pool before
 * being terminated.
 *
 * Higher values are useful if clients frequently send small requests and
 * you want to minimize the latency for those.
 */
#define THREADPOOL_THREAD_IDLE_LIMIT 1000000

/* Maximum number of threads in THREAD_POOL, i.e. number of paths we can
 * fsync concurrently throughout the process. */
#define MAX_THREADS 16

/* Thread pool to execute the fsync tasks. */
static apr_thread_pool_t *thread_pool = NULL;

#endif

/* We open non-directory files with these flags. */
#define FILE_FLAGS (APR_READ | APR_WRITE | APR_BUFFERED | APR_CREATE)

#if APR_HAS_THREADS

/* Destructor function that implicitly cleans up any running threads
   in the thread_pool given as DATA and releases their memory pools
   before they get destroyed themselves.

   Must be run as a pre-cleanup hook.
 */
static apr_status_t
thread_pool_pre_cleanup(void *data)
{
  apr_thread_pool_t *tp = data;
  return apr_thread_pool_destroy(tp);
}

#endif

svn_error_t *
svn_fs_x__batch_fsync_init(void)
{
#if APR_HAS_THREADS
  /* The thread-pool must be allocated from a thread-safe pool.
     GLOBAL_POOL may be single-threaded, though. */
  apr_pool_t *pool = svn_pool_create(NULL);

{'fraction_non_alphanumeric': 0.1027623532719477, 'fraction_numerical': 0.003162999929711113, 'mean_word_length': 3.674113009198423, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6067268', 'n_tokens_mistral': 4725, 'n_tokens_neox': 4555, 'n_words': 1305}