*
* Copyright 2025 Adam Masciola
*
* This program 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 3 of the License, or
* (at your option) any later version.
*
* This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
*
* SPDX-License-Identifier: GPL-3.0-or-later
*/
#define G_LOG_DOMAIN "BAZAAR::ENTRY-CACHE"
#define BAZAAR_MODULE "entry-cache"
#define MAX_CONCURRENT_WRITES 16
#define WATCH_CLEANUP_INTERVAL_MSEC 5000
#include <malloc.h>
#include "bz-entry-cache-manager.h"
#include "bz-flatpak-entry.h"
#include "bz-serializable.h"
#include "env.h"
#include "io.h"
#include "util.h"
G_DEFINE_QUARK (bz-entry-cache-error-quark, bz_entry_cache_error);
struct _BzEntryCacheManager
{
GObject parent_instance;
GMutex mutex;
guint living_entries;
DexScheduler *scheduler;
guint64 memory_usage;
DexPromise *init;
GHashTable *alive_hash;
GHashTable *writing_hash;
GHashTable *reading_hash;
BzGuard *ongoing_gates[MAX_CONCURRENT_WRITES];
GMutex ongoing_mutexes[MAX_CONCURRENT_WRITES];
guint ongoing_queued[MAX_CONCURRENT_WRITES];
GMutex ongoing_queueing_mutex;
BzGuard *alive_gate;
GMutex alive_mutex;
BzGuard *reading_gate;
GMutex reading_mutex;
BzGuard *writing_gate;
GMutex writing_mutex;
DexFuture *watch_task;
};
G_DEFINE_FINAL_TYPE (BzEntryCacheManager, bz_entry_cache_manager, G_TYPE_OBJECT);
enum
{
PROP_0,
PROP_LIVING_ENTRIES,
LAST_PROP
};
static GParamSpec *props[LAST_PROP] = { 0 };
static DexFuture *
watch_init_fiber (GWeakRef *wr);
static DexFuture *
watch_cb (DexFuture *future,
GWeakRef *wr);
static DexFuture *
watch_work_fiber (GWeakRef *wr);
static DexFuture *
notify_props_fiber (GWeakRef *wr);
BZ_DEFINE_DATA (
living_entry,
LivingEntry,
{
GWeakRef wr;
BzGuard *gate;
GMutex mutex;
GTimer *cached;
},
BZ_RELEASE_DATA (gate, bz_guard_destroy);
g_mutex_clear (&self->mutex);
g_weak_ref_clear (&self->wr);
BZ_RELEASE_DATA (cached, g_timer_destroy));
BZ_DEFINE_DATA (
write_task,
WriteTask,
{
GWeakRef *self;
char *unique_id_checksum;
BzEntry *entry;
},
BZ_RELEASE_DATA (self, bz_weak_release);
BZ_RELEASE_DATA (unique_id_checksum, g_free);
BZ_RELEASE_DATA (entry, g_object_unref);)
static DexFuture *
write_task_fiber (WriteTaskData *data);
BZ_DEFINE_DATA (
read_task,
ReadTask,
{
GWeakRef *self;
char *unique_id_checksum;
},
BZ_RELEASE_DATA (self, bz_weak_release);
BZ_RELEASE_DATA (unique_id_checksum, g_free))
static DexFuture *
read_task_fiber (ReadTaskData *data);
static DexFuture *
enumerate_disk_fiber (GWeakRef *wr);
static void
bz_entry_cache_manager_dispose (GObject *object)
{
BzEntryCacheManager *self = BZ_ENTRY_CACHE_MANAGER (object);
g_mutex_clear (&self->mutex);
dex_clear (&self->scheduler);
dex_clear (&self->watch_task);
g_clear_pointer (&self->init, dex_unref);
g_clear_pointer (&self->alive_hash, g_hash_table_unref);
g_clear_pointer (&self->writing_hash, g_hash_table_unref);
g_clear_pointer (&self->reading_hash, g_hash_table_unref);
for (guint i = 0; i < G_N_ELEMENTS (self->ongoing_gates); i++)
g_clear_pointer (&self->ongoing_gates[i], bz_guard_destroy);
for (guint i = 0; i < G_N_ELEMENTS (self->ongoing_mutexes); i++)
g_mutex_clear (&self->ongoing_mutexes[i]);
g_mutex_clear (&self->ongoing_queueing_mutex);
g_clear_pointer (&self->alive_gate, bz_guard_destroy);
g_clear_pointer (&self->reading_gate, bz_guard_destroy);
g_clear_pointer (&self->writing_gate, bz_guard_destroy);
g_mutex_clear (&self->alive_mutex);
g_mutex_clear (&self->reading_mutex);
g_mutex_clear (&self->writing_mutex);
G_OBJECT_CLASS (bz_entry_cache_manager_parent_class)->dispose (object);
}
static void
bz_entry_cache_manager_get_property (GObject *object,
guint prop_id,
GValue *value,
GParamSpec *pspec)
{
BzEntryCacheManager *self = BZ_ENTRY_CACHE_MANAGER (object);
switch (prop_id)
{
case PROP_LIVING_ENTRIES:
g_value_set_uint (value, bz_entry_cache_manager_get_living_entries (self));
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
}
}
static void
bz_entry_cache_manager_set_property (GObject *object,
guint prop_id,
const GValue *value,
GParamSpec *pspec)
{
BzEntryCacheManager *self = BZ_ENTRY_CACHE_MANAGER (object);
(void) self;
switch (prop_id)
{
case PROP_LIVING_ENTRIES:
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
}
}
static void
bz_entry_cache_manager_class_init (BzEntryCacheManagerClass *klass)
{
GObjectClass *object_class = G_OBJECT_CLASS (klass);
object_class->set_property = bz_entry_cache_manager_set_property;
object_class->get_property = bz_entry_cache_manager_get_property;
object_class->dispose = bz_entry_cache_manager_dispose;
props[PROP_LIVING_ENTRIES] =
g_param_spec_uint (
"living-entries",
NULL, NULL,
0, G_MAXUINT, 0,
G_PARAM_READABLE | G_PARAM_STATIC_STRINGS | G_PARAM_EXPLICIT_NOTIFY);
g_object_class_install_properties (object_class, LAST_PROP, props);
}
static void
bz_entry_cache_manager_init (BzEntryCacheManager *self)
{
g_mutex_init (&self->mutex);
self->scheduler = dex_thread_pool_scheduler_new ();
self->init = dex_promise_new ();
self->alive_hash = g_hash_table_new_full (
g_str_hash, g_str_equal, g_free, living_entry_data_unref);
self->writing_hash = g_hash_table_new_full (
g_str_hash, g_str_equal, g_free, dex_unref);
self->reading_hash = g_hash_table_new_full (
g_str_hash, g_str_equal, g_free, dex_unref);
for (guint i = 0; i < G_N_ELEMENTS (self->ongoing_mutexes); i++)
g_mutex_init (&self->ongoing_mutexes[i]);
g_mutex_init (&self->ongoing_queueing_mutex);
g_mutex_init (&self->alive_mutex);
g_mutex_init (&self->reading_mutex);
g_mutex_init (&self->writing_mutex);
self->watch_task = dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) watch_init_fiber,
bz_track_weak (self), bz_weak_release);
}
BzEntryCacheManager *
bz_entry_cache_manager_new (void)
{
return g_object_new (BZ_TYPE_ENTRY_CACHE_MANAGER, NULL);
}
guint
bz_entry_cache_manager_get_living_entries (BzEntryCacheManager *self)
{
g_autoptr (GMutexLocker) locker = NULL;
g_return_val_if_fail (BZ_IS_ENTRY_CACHE_MANAGER (self), 0);
locker = g_mutex_locker_new (&self->mutex);
return self->living_entries;
}
DexFuture *
bz_entry_cache_manager_add (BzEntryCacheManager *self,
BzEntry *entry)
{
g_autoptr (WriteTaskData) data = NULL;
g_autoptr (DexFuture) future = NULL;
dex_return_error_if_fail (BZ_IS_ENTRY_CACHE_MANAGER (self));
dex_return_error_if_fail (BZ_IS_ENTRY (entry));
dex_return_error_if_fail (!bz_entry_is_holding (entry));
data = write_task_data_new ();
data->self = bz_track_weak (self);
data->unique_id_checksum = g_strdup (bz_entry_get_unique_id_checksum (entry));
data->entry = g_object_ref (entry);
future = dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) write_task_fiber,
write_task_data_ref (data),
write_task_data_unref);
return g_steal_pointer (&future);
}
DexFuture *
bz_entry_cache_manager_get (BzEntryCacheManager *self,
const char *unique_id)
{
g_autoptr (ReadTaskData) data = NULL;
g_autoptr (DexFuture) future = NULL;
dex_return_error_if_fail (BZ_IS_ENTRY_CACHE_MANAGER (self));
dex_return_error_if_fail (unique_id != NULL);
data = read_task_data_new ();
data->self = bz_track_weak (self);
data->unique_id_checksum = g_compute_checksum_for_string (G_CHECKSUM_MD5, unique_id, -1);
future = dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) read_task_fiber,
read_task_data_ref (data),
read_task_data_unref);
return g_steal_pointer (&future);
}
DexFuture *
bz_entry_cache_manager_get_by_checksum (BzEntryCacheManager *self,
const char *unique_id_checksum)
{
g_autoptr (ReadTaskData) data = NULL;
g_autoptr (DexFuture) future = NULL;
dex_return_error_if_fail (BZ_IS_ENTRY_CACHE_MANAGER (self));
dex_return_error_if_fail (unique_id_checksum != NULL);
data = read_task_data_new ();
data->self = bz_track_weak (self);
data->unique_id_checksum = g_strdup (unique_id_checksum);
future = dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) read_task_fiber,
read_task_data_ref (data),
read_task_data_unref);
return g_steal_pointer (&future);
}
DexFuture *
bz_entry_cache_manager_enumerate_disk (BzEntryCacheManager *self)
{
g_autoptr (DexFuture) future = NULL;
dex_return_error_if_fail (BZ_IS_ENTRY_CACHE_MANAGER (self));
future = dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) enumerate_disk_fiber,
bz_track_weak (self), bz_weak_release);
return g_steal_pointer (&future);
}
static DexFuture *
write_task_fiber (WriteTaskData *data)
{
g_autoptr (BzEntryCacheManager) self = NULL;
char *unique_id_checksum = data->unique_id_checksum;
BzEntry *entry = data->entry;
g_autoptr (GError) local_error = NULL;
g_autoptr (BzGuard) slot_guard = NULL;
g_autoptr (BzGuard) other_guard = NULL;
g_autoptr (GMutexLocker) locker = NULL;
guint slot_queued = G_MAXUINT;
guint slot_index = 0;
DexFuture *writing_future = NULL;
g_autoptr (LivingEntryData) living = NULL;
g_autoptr (DexPromise) promise = NULL;
g_autoptr (GVariantBuilder) builder = NULL;
g_autoptr (GVariant) variant = NULL;
g_autoptr (GBytes) bytes = NULL;
gsize bytes_size = 0;
gconstpointer bytes_data = 0;
g_autofree char *main_cache = NULL;
g_autoptr (GFile) parent_file = NULL;
g_autofree char *save_file_path = NULL;
g_autoptr (GFile) save_file = NULL;
gsize existing_contents_size = 0;
g_autofree char *existing_contents = NULL;
g_autoptr (GFileOutputStream) output = NULL;
gssize bytes_written = 0;
gboolean result = FALSE;
g_autoptr (GError) ret_error = NULL;
bz_weak_get_or_return_reject (self, data->self);
if (!BZ_IS_FLATPAK_ENTRY (entry))
return dex_future_new_reject (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Entry with unique ID checksum '%s' cannot be "
"cached because it is not a flatpak entry",
unique_id_checksum);
* when refresh triggers a flood of requests
*
* Here we make sure to pick the slot with the
* least tasks waiting in line
*/
locker = g_mutex_locker_new (&self->ongoing_queueing_mutex);
for (guint i = 0; i < G_N_ELEMENTS (self->ongoing_gates); i++)
{
if (self->ongoing_queued[i] < slot_queued)
{
slot_queued = self->ongoing_queued[i];
slot_index = i;
}
}
self->ongoing_queued[slot_index]++;
g_clear_pointer (&locker, g_mutex_locker_free);
BZ_BEGIN_GUARD_WITH_CONTEXT (&slot_guard,
&self->ongoing_mutexes[slot_index],
&self->ongoing_gates[slot_index]);
locker = g_mutex_locker_new (&self->ongoing_queueing_mutex);
self->ongoing_queued[slot_index]--;
g_clear_pointer (&locker, g_mutex_locker_free);
dex_await (dex_ref (self->init), NULL);
BZ_BEGIN_GUARD_WITH_CONTEXT (&other_guard,
&self->writing_mutex,
&self->writing_gate);
{
writing_future = g_hash_table_lookup (self->writing_hash, unique_id_checksum);
if (writing_future != NULL)
return dex_future_new_reject (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Entry with unique ID '%s' is already being cached right now",
unique_id_checksum);
promise = dex_promise_new ();
g_hash_table_replace (self->writing_hash,
g_strdup (unique_id_checksum),
dex_ref (promise));
}
bz_clear_guard (&other_guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&other_guard,
&self->alive_mutex,
&self->alive_gate);
{
living = g_hash_table_lookup (self->alive_hash, unique_id_checksum);
if (living != NULL)
living_entry_data_ref (living);
else
{
living = living_entry_data_new ();
g_weak_ref_init (&living->wr, NULL);
g_mutex_init (&living->mutex);
living->cached = g_timer_new ();
g_hash_table_replace (self->alive_hash,
g_strdup (unique_id_checksum),
living_entry_data_ref (living));
}
}
bz_clear_guard (&other_guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&other_guard,
&living->mutex,
&living->gate);
{
builder = g_variant_builder_new (G_VARIANT_TYPE_VARDICT);
bz_serializable_serialize (BZ_SERIALIZABLE (entry), builder);
variant = g_variant_builder_end (builder);
bytes = g_variant_get_data_as_bytes (variant);
bytes_data = g_bytes_get_data (bytes, &bytes_size);
main_cache = bz_dup_module_dir ();
parent_file = g_file_new_for_path (main_cache);
result = g_file_make_directory_with_parents (parent_file, NULL, &local_error);
if (!result)
{
if (g_error_matches (local_error, G_IO_ERROR, G_IO_ERROR_EXISTS))
g_clear_pointer (&local_error, g_error_free);
else
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Failed to make parent directory '%s' when caching '%s': %s",
main_cache, unique_id_checksum, local_error->message);
goto done;
}
}
save_file_path = g_build_filename (main_cache, unique_id_checksum, NULL);
save_file = g_file_new_for_path (save_file_path);
result = g_file_get_contents (
save_file_path, &existing_contents,
&existing_contents_size, NULL);
if (!result ||
existing_contents_size != bytes_size ||
memcmp (existing_contents, bytes_data, bytes_size) != 0)
{
output = g_file_replace (
save_file,
NULL,
FALSE,
G_FILE_CREATE_REPLACE_DESTINATION,
NULL,
&local_error);
if (output == NULL)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Failed to open write stream when caching '%s': %s",
unique_id_checksum, local_error->message);
goto done;
}
bytes_written = g_output_stream_write_bytes (G_OUTPUT_STREAM (output), bytes, NULL, &local_error);
if (bytes_written < 0)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Failed to write data to stream when caching '%s': %s",
unique_id_checksum, local_error->message);
goto done;
}
result = g_output_stream_close (G_OUTPUT_STREAM (output), NULL, &local_error);
if (!result)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_CACHE_FAILED,
"Failed to close stream when caching '%s': %s",
unique_id_checksum, local_error->message);
goto done;
}
}
g_timer_start (living->cached);
}
done:
bz_clear_guard (&slot_guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&other_guard,
&self->writing_mutex,
&self->writing_gate);
{
if (ret_error != NULL)
dex_promise_reject (promise, g_error_copy (ret_error));
else
dex_promise_resolve_boolean (promise, TRUE);
g_hash_table_remove (self->writing_hash, unique_id_checksum);
}
bz_clear_guard (&other_guard);
if (ret_error != NULL)
return dex_future_new_for_error (g_steal_pointer (&ret_error));
else
return dex_future_new_true ();
}
static DexFuture *
read_task_fiber (ReadTaskData *data)
{
g_autoptr (BzEntryCacheManager) self = NULL;
char *unique_id_checksum = data->unique_id_checksum;
g_autoptr (GError) local_error = NULL;
g_autoptr (BzGuard) guard = NULL;
g_autoptr (GMutexLocker) locker = NULL;
g_autoptr (DexFuture) writing_future = NULL;
g_autoptr (LivingEntryData) living = NULL;
DexFuture *reading_future = NULL;
g_autoptr (DexPromise) promise = NULL;
g_autofree char *main_cache = NULL;
g_autofree char *path = NULL;
g_autoptr (GFile) file = NULL;
g_autoptr (GBytes) bytes = NULL;
g_autoptr (GVariant) variant = NULL;
g_autoptr (BzFlatpakEntry) entry = NULL;
gboolean result = FALSE;
g_autoptr (GError) ret_error = NULL;
bz_weak_get_or_return_reject (self, data->self);
dex_await (dex_ref (self->init), NULL);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard,
&self->writing_mutex,
&self->writing_gate);
{
writing_future = g_hash_table_lookup (self->writing_hash, unique_id_checksum);
if (writing_future != NULL)
{
dex_ref (writing_future);
bz_clear_guard (&guard);
dex_await (g_steal_pointer (&writing_future), NULL);
}
}
bz_clear_guard (&guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard,
&self->reading_mutex,
&self->reading_gate);
{
reading_future = g_hash_table_lookup (self->reading_hash, unique_id_checksum);
if (reading_future != NULL)
return dex_ref (reading_future);
promise = dex_promise_new ();
g_hash_table_replace (self->reading_hash,
g_strdup (unique_id_checksum),
dex_ref (promise));
}
bz_clear_guard (&guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard,
&self->alive_mutex,
&self->alive_gate);
{
living = g_hash_table_lookup (self->alive_hash, unique_id_checksum);
if (living != NULL)
{
g_autoptr (BzEntry) living_entry = NULL;
living_entry_data_ref (living);
bz_clear_guard (&guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard, &living->mutex, &living->gate);
living_entry = g_weak_ref_get (&living->wr);
if (living_entry != NULL)
{
bz_clear_guard (&guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard,
&self->reading_mutex,
&self->reading_gate);
{
g_hash_table_remove (self->reading_hash, unique_id_checksum);
}
bz_clear_guard (&guard);
dex_promise_resolve_object (promise, g_object_ref (living_entry));
return dex_future_new_for_object (living_entry);
}
}
else
{
living = living_entry_data_new ();
g_weak_ref_init (&living->wr, NULL);
g_mutex_init (&living->mutex);
living->cached = g_timer_new ();
g_hash_table_replace (self->alive_hash,
g_strdup (unique_id_checksum),
living_entry_data_ref (living));
bz_clear_guard (&guard);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard, &living->mutex, &living->gate);
}
}
main_cache = bz_dup_module_dir ();
path = g_build_filename (main_cache, unique_id_checksum, NULL);
file = g_file_new_for_path (path);
bytes = g_file_load_bytes (file, NULL, NULL, &local_error);
if (bytes == NULL)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_DECACHE_FAILED,
"Failed to de-cache variant from file: %s",
local_error->message);
goto done;
}
variant = g_variant_new_from_bytes (G_VARIANT_TYPE_VARDICT, bytes, FALSE);
if (variant == NULL)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_DECACHE_FAILED,
"Failed to interpret variant from %s: %s",
path, local_error->message);
goto done;
}
entry = g_object_new (BZ_TYPE_FLATPAK_ENTRY, NULL);
result = bz_serializable_deserialize (BZ_SERIALIZABLE (entry), variant, &local_error);
if (!result)
{
ret_error = g_error_new (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_DECACHE_FAILED,
"Failed to deserialize entry from %s: %s",
path, local_error->message);
goto done;
}
g_weak_ref_init (&living->wr, entry);
done:
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard,
&self->reading_mutex,
&self->reading_gate);
{
if (ret_error != NULL)
dex_promise_reject (promise, g_error_copy (ret_error));
else
dex_promise_resolve_object (promise, g_object_ref (entry));
g_hash_table_remove (self->reading_hash, unique_id_checksum);
}
bz_clear_guard (&guard);
if (ret_error != NULL)
return dex_future_new_for_error (g_steal_pointer (&ret_error));
else
return dex_future_new_for_object (entry);
}
static DexFuture *
enumerate_disk_fiber (GWeakRef *wr)
{
g_autoptr (BzEntryCacheManager) self = NULL;
g_autoptr (GError) local_error = NULL;
g_autoptr (BzGuard) guard = NULL;
g_autoptr (GHashTable) set = NULL;
g_autofree char *main_cache = NULL;
g_autoptr (GFile) main_cache_file = NULL;
g_autoptr (GFileEnumerator) enumerator = NULL;
bz_weak_get_or_return_reject (self, wr);
set = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, NULL);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard, &self->alive_mutex, &self->alive_gate);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard, &self->reading_mutex, &self->reading_gate);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard, &self->writing_mutex, &self->writing_gate);
main_cache = bz_dup_module_dir ();
if (!g_file_test (main_cache, G_FILE_TEST_EXISTS))
goto done;
main_cache_file = g_file_new_for_path (main_cache);
enumerator = g_file_enumerate_children (
main_cache_file,
G_FILE_ATTRIBUTE_STANDARD_IS_SYMLINK
"," G_FILE_ATTRIBUTE_STANDARD_NAME
"," G_FILE_ATTRIBUTE_STANDARD_TYPE,
G_FILE_QUERY_INFO_NOFOLLOW_SYMLINKS,
NULL,
&local_error);
if (enumerator == NULL)
return dex_future_new_reject (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_ENUMERATE_FAILED,
"Could not initialize directory enumerator at %s: %s",
main_cache, local_error->message);
for (;;)
{
g_autoptr (GFileInfo) info = NULL;
g_autoptr (GFile) child = NULL;
g_autofree char *basename = NULL;
info = g_file_enumerator_next_file (enumerator, NULL, &local_error);
if (info == NULL)
{
if (local_error != NULL)
return dex_future_new_reject (
BZ_ENTRY_CACHE_ERROR,
BZ_ENTRY_CACHE_ERROR_ENUMERATE_FAILED,
"Could not enumerate children of cache directory at %s: %s",
main_cache, local_error->message);
else
break;
}
child = g_file_enumerator_get_child (enumerator, info);
if (g_file_info_get_is_symlink (info) ||
g_file_info_get_file_type (info) != G_FILE_TYPE_REGULAR)
continue;
basename = g_file_get_basename (child);
if (basename != NULL)
g_hash_table_replace (set, g_steal_pointer (&basename), NULL);
}
done:
return dex_future_new_take_boxed (G_TYPE_HASH_TABLE, g_steal_pointer (&set));
}
static DexFuture *
watch_init_fiber (GWeakRef *wr)
{
g_autoptr (BzEntryCacheManager) self = NULL;
bz_weak_get_or_return_reject (self, wr);
dex_promise_resolve_boolean (self->init, TRUE);
return dex_future_finally_loop (
dex_timeout_new_msec (WATCH_CLEANUP_INTERVAL_MSEC),
(DexFutureCallback) watch_cb,
bz_track_weak (self), bz_weak_release);
}
static DexFuture *
watch_cb (DexFuture *future,
GWeakRef *wr)
{
g_autoptr (BzEntryCacheManager) self = NULL;
bz_weak_get_or_return_reject (self, wr);
return dex_scheduler_spawn (
self->scheduler,
bz_get_dex_stack_size (),
(DexFiberFunc) watch_work_fiber,
bz_track_weak (self), bz_weak_release);
}
static DexFuture *
watch_work_fiber (GWeakRef *wr)
{
g_autoptr (BzEntryCacheManager) self = NULL;
g_autoptr (BzGuard) guard0 = NULL;
GHashTableIter iter = { 0 };
g_autoptr (GTimer) timer = NULL;
guint total = 0;
guint active = 0;
guint alive = 0;
guint pruned = 0;
bz_weak_get_or_return_reject (self, wr);
timer = g_timer_new ();
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard0, &self->alive_mutex, &self->alive_gate);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard0, &self->reading_mutex, &self->reading_gate);
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard0, &self->writing_mutex, &self->writing_gate);
g_hash_table_iter_init (&iter, self->alive_hash);
for (;;)
{
char *unique_id_checksum = NULL;
LivingEntryData *living = NULL;
g_autoptr (BzGuard) guard1 = NULL;
g_autoptr (BzEntry) entry = NULL;
if (!g_hash_table_iter_next (&iter, (gpointer *) &unique_id_checksum, (gpointer *) &living))
break;
total++;
if (g_hash_table_contains (self->reading_hash, unique_id_checksum) ||
g_hash_table_contains (self->writing_hash, unique_id_checksum))
{
active++;
continue;
}
BZ_BEGIN_GUARD_WITH_CONTEXT (&guard1, &living->mutex, &living->gate);
entry = g_weak_ref_get (&living->wr);
if (entry != NULL)
alive++;
else
{
bz_clear_guard (&guard1);
g_hash_table_iter_remove (&iter);
pruned++;
}
}
bz_clear_guard (&guard0);
#ifdef __GLIBC__
malloc_trim (0);
#endif
g_debug ("Sweep report: finished in %.4f seconds, including time to acquire guards\n"
" Out of a total of %d entries considered:\n"
" %d were skipped due to active tasks being associated with them\n"
" %d application entries were otherwise kept alive\n"
" %d entries were forgotten by the application and were pruned\n"
" Another sweep will take place in %d msec",
g_timer_elapsed (timer, NULL),
total, active, alive, pruned, WATCH_CLEANUP_INTERVAL_MSEC);
g_mutex_lock (&self->mutex);
self->living_entries = active + alive;
g_mutex_unlock (&self->mutex);
dex_future_disown (dex_scheduler_spawn (
dex_scheduler_get_default (),
bz_get_dex_stack_size (),
(DexFiberFunc) notify_props_fiber,
bz_track_weak (self),
bz_weak_release));
return dex_timeout_new_msec (WATCH_CLEANUP_INTERVAL_MSEC);
}
static DexFuture *
notify_props_fiber (GWeakRef *wr)
{
g_autoptr (BzEntryCacheManager) self = NULL;
bz_weak_get_or_return_reject (self, wr);
g_object_notify_by_pspec (G_OBJECT (self), props[PROP_LIVING_ENTRIES]);
return dex_future_new_true ();
}
