Mirrors/criu
1
0
Fork 0
mirror of https://github.com/checkpoint-restore/criu.git synced 2026-07-17 16:47:50 +00:00
Code Issues Projects Releases Wiki Activity

mem/pagemap: Decode LZ4 pages before PIE

Browse source 
The PIE restorer cannot link against LZ4. Decode mappings containing
LZ4 payloads through CRIU's premap path, while retaining delayed PIE
restore for raw and zero ranges when a worker pool cannot help.

Classify image ranges as LZ4, raw, or zero. Decode LZ4 blocks while CRIU
still runs in its normal address space, read aligned raw ranges directly
into their final VMAs, and clear zero ranges in place. Omit compression
metadata for entries containing only raw blocks so they retain the
ordinary restore path.

Share one encoded-read context across each incremental parent chain and
reuse its worker pool for bounded batches, COW reads, streamed images,
and page-server chunks. Use the same jobs to fill zero runs of at least
1 MiB when two blocks and two CPUs are available. Zero-only ranges that
cannot benefit retain the PIE path, while zero blocks in an existing
encoded batch remain inline without effective parallel capacity. Hold
input buffers only while encoded work is active, and join every worker
before child restore or the PIE handoff.

Account calling restore threads and workers against a shared CPU budget.
A requested width of zero selects automatic concurrency, one keeps LZ4
decoding serial and disables zero-fill workers, and larger values cap
aggregate worker concurrency. Bound the active width by CPU affinity and
useful work. Release encoded-work leases before ordinary reads so idle
contexts cannot block ready workers.

Local payload-heavy restores otherwise alternate between reading an
encoded batch and decoding it. This leaves either storage or worker CPUs
idle. When a second encoded-input lease is immediately available,
dispatch the current batch to the pool and use the already-accounted
calling thread to read the next adjacent payload. Keep synchronous reads
for serial restore, streams, direct I/O, small batches, and contended
leases.

Select direct AIO from the ranges actually delayed for PIE. Batch
inherited COW comparisons and streamed payload reads, and align
page-server chunks to compression regions. Validate metadata and offset
arithmetic before allocation or I/O, roll back partially enqueued work,
and propagate worker, I/O, and deferred deduplication failures before
remapping.

On a 256 MiB pseudorandom workload with the worker limit set to eight
and three measured runs, median restore time was 114.5 ms uncompressed,
113.4 ms with LZ4 per page, and 116.5 ms with 64 KiB regions. The
previous compressed restore path took about 227 ms in either mode.

With automatic concurrency and 256 KiB regions, parallel zero filling
reduced median restore time over five runs from 103.1 ms to 31.7 ms for
zero-filled memory and from 89.3 ms to 60.0 ms for mixed memory.

On 512 MiB TEXT and ELF workloads with automatic concurrency and 256 KiB
regions, five-run median restore time fell from 117.0 ms to 110.7 ms and
from 115.4 ms to 109.5 ms, respectively.

Assisted-by: Codex:GPT-5
Assisted-by: Claude:claude-fable-5
Signed-off-by: Radostin Stoyanov <rstoyanov@fedoraproject.org>
This commit is contained in:
Radostin Stoyanov 2026-07-11 12:12:40 +01:00
parent 506ab0d9da
commit e2c84bfba7
14 changed files with 3306 additions and 576 deletions
Download patch file Download diff file Expand all files Collapse all files

822
criu/compression.c
View file

@ -1,16 +1,335 @@
#include <errno.h>
#include <limits.h>
#include <pthread.h>
#include <sched.h>
#include <signal.h>
#include <stdint.h>
#include <string.h>
#include <unistd.h>
#include <lz4.h>
#include "page.h"
#include "log.h"
#include "common/bug.h"
#include "compression.h"
#include "common/xmalloc.h"
#undef LOG_PREFIX
#define LOG_PREFIX "compression: "
#define PARALLEL_DECOMPRESS_MIN_BYTES_PER_THREAD (512UL << 10)
#define PARALLEL_DECOMPRESS_STACK_SIZE (256UL << 10)
#define PARALLEL_DECOMPRESS_MAX_JOB_CHUNK 8
#define PARALLEL_DECOMPRESS_MAX_BATCHES 2
/*
* asyncd can restore several shmem or memfd objects concurrently. Each of
* those requests may enter decompress_jobs_parallel_pool(), so limiting one
* invocation to the available CPUs is not sufficient: independent calls can
* otherwise multiply the worker count. Account for the calling thread as
* well as workers. Restore processes use one budget in shared restore memory;
* callers outside restore fall back to a process-wide budget.
*/
static pthread_once_t decompress_budget_once = PTHREAD_ONCE_INIT;
static pthread_mutex_t decompress_budget_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t decompress_budget_cond = PTHREAD_COND_INITIALIZER;
static unsigned int decompress_budget_available = 1;
static unsigned int decompress_budget_capacity_cached = 1;
static pthread_mutex_t decompress_batch_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t decompress_batch_cond = PTHREAD_COND_INITIALIZER;
static unsigned int decompress_batches_available = PARALLEL_DECOMPRESS_MAX_BATCHES;
static struct decompression_shared_budget *decompress_shared_budget;
static unsigned int decompression_available_cpus(void)
{
cpu_set_t *affinity;
size_t affinity_size;
size_t nr_cpus = CPU_SETSIZE;
long available_cpus;
int affinity_errno;
available_cpus = sysconf(_SC_NPROCESSORS_CONF);
if (available_cpus > (long)nr_cpus)
nr_cpus = (size_t)available_cpus;
for (;;) {
affinity = CPU_ALLOC(nr_cpus);
if (!affinity) {
pr_warn("Unable to allocate a CPU affinity mask, using serial decompression\n");
return 1;
}
affinity_size = CPU_ALLOC_SIZE(nr_cpus);
CPU_ZERO_S(affinity_size, affinity);
if (sched_getaffinity(0, affinity_size, affinity) == 0) {
available_cpus = CPU_COUNT_S(affinity_size, affinity);
CPU_FREE(affinity);
goto found;
}
affinity_errno = errno;
CPU_FREE(affinity);
if (affinity_errno != EINVAL) {
pr_warn("Unable to read the CPU affinity mask: %s; using serial decompression\n",
strerror(affinity_errno));
return 1;
}
if (nr_cpus > UINT_MAX / 2) {
pr_warn("CPU affinity mask is too large, using serial decompression\n");
return 1;
}
nr_cpus *= 2;
}
found:
if (available_cpus < 1)
return 1;
return (unsigned int)available_cpus;
}
unsigned int decompression_thread_limit(unsigned int requested, unsigned int available_cpus)
{
if (!available_cpus)
available_cpus = 1;
if (!requested)
return available_cpus;
return min(requested, available_cpus);
}
static unsigned int decompression_cpu_limit(unsigned int requested)
{
return decompression_thread_limit(requested, decompression_available_cpus());
}
/*
* Automatic (0) and serial (1) settings must not serialize independent asyncd
* jobs: both leave the restore-wide budget at the available-CPU capacity, and a
* serial setting still limits each worker-pool call to its caller. Explicit
* parallel settings cap all callers to the requested number of CPUs.
*/
static unsigned int decompression_budget_capacity(unsigned int configured_threads)
{
if (configured_threads < 2)
return decompression_cpu_limit(0);
return decompression_cpu_limit(configured_threads);
}
static void decompress_budget_init(void)
{
decompress_budget_capacity_cached = decompression_cpu_limit(0);
decompress_budget_available = decompress_budget_capacity_cached;
}
void decompression_shared_budget_init(struct decompression_shared_budget *budget, unsigned int requested_threads)
{
unsigned int threads = decompression_budget_capacity(requested_threads);
/* Reduce an explicit request that exceeds the CPUs available to CRIU. */
if (requested_threads > threads)
pr_warn("Reducing compressed-page worker concurrency from %u to %u (available CPUs)\n",
requested_threads, threads);
futex_set(&budget->threads, threads);
budget->thread_capacity = threads;
/*
* Batch leases bound active encoded working sets independently of the CPU
* budget. Keep at most two 32 MiB input buffers so one local reader can
* overlap its next read with decoding without retaining unbounded memory.
*/
futex_set(&budget->batches, PARALLEL_DECOMPRESS_MAX_BATCHES);
}
void decompression_use_shared_budget(struct decompression_shared_budget *budget)
{
/* Borrowed from shared restore memory and valid for the restore lifetime. */
decompress_shared_budget = budget;
}
bool compressed_restore_has_parallel_capacity(unsigned int requested_threads)
{
unsigned int capacity;
int err;
if (requested_threads == 1)
return false;
if (decompress_shared_budget)
capacity = decompress_shared_budget->thread_capacity;
else {
err = pthread_once(&decompress_budget_once, decompress_budget_init);
if (err)
return false;
capacity = decompress_budget_capacity_cached;
}
return decompression_thread_limit(requested_threads, capacity) > 1;
}
static unsigned int shared_budget_acquire(futex_t *available, unsigned int requested)
{
for (;;) {
unsigned int current = futex_get(available);
unsigned int granted;
unsigned int previous;
if (!current) {
futex_wait_while_eq(available, 0);
continue;
}
granted = min(requested, current);
previous = atomic_cmpxchg(&available->raw, current, current - granted);
if (previous == current)
return granted;
}
}
static bool shared_budget_try_acquire(futex_t *available)
{
unsigned int current = futex_get(available);
while (current) {
unsigned int previous;
previous = atomic_cmpxchg(&available->raw, current, current - 1);
if (previous == current)
return true;
current = previous;
}
return false;
}
static void shared_budget_release(futex_t *available, unsigned int amount)
{
atomic_add(amount, &available->raw);
futex_wake(available);
}
static unsigned int decompress_budget_acquire(unsigned int requested)
{
unsigned int granted;
int err;
if (decompress_shared_budget)
return shared_budget_acquire(&decompress_shared_budget->threads, requested);
err = pthread_once(&decompress_budget_once, decompress_budget_init);
if (err) {
pr_warn("Unable to initialize decompression CPU budget: %s\n", strerror(err));
return 0;
}
err = pthread_mutex_lock(&decompress_budget_lock);
if (err) {
pr_warn("Unable to lock decompression CPU budget: %s\n", strerror(err));
return 0;
}
while (!decompress_budget_available) {
err = pthread_cond_wait(&decompress_budget_cond, &decompress_budget_lock);
if (err) {
pr_warn("Unable to wait for decompression CPU budget: %s\n", strerror(err));
pthread_mutex_unlock(&decompress_budget_lock);
return 0;
}
}
granted = min(requested, decompress_budget_available);
decompress_budget_available -= granted;
pthread_mutex_unlock(&decompress_budget_lock);
return granted;
}
static void decompress_budget_release(unsigned int nr_threads)
{
int err;
if (!nr_threads)
return;
if (decompress_shared_budget) {
shared_budget_release(&decompress_shared_budget->threads, nr_threads);
return;
}
err = pthread_mutex_lock(&decompress_budget_lock);
if (err) {
pr_err("Unable to lock decompression CPU budget for release: %s\n", strerror(err));
BUG();
}
decompress_budget_available += nr_threads;
pthread_cond_broadcast(&decompress_budget_cond);
pthread_mutex_unlock(&decompress_budget_lock);
}
static void decompress_budget_trim_reservation(unsigned int *threads_held, unsigned int threads_to_keep)
{
BUG_ON(threads_to_keep > *threads_held);
decompress_budget_release(*threads_held - threads_to_keep);
*threads_held = threads_to_keep;
}
void decompression_batch_acquire(void)
{
int err;
if (decompress_shared_budget) {
shared_budget_acquire(&decompress_shared_budget->batches, 1);
return;
}
err = pthread_mutex_lock(&decompress_batch_lock);
if (err) {
pr_err("Unable to lock decompression batch budget: %s\n", strerror(err));
BUG();
}
while (!decompress_batches_available) {
err = pthread_cond_wait(&decompress_batch_cond, &decompress_batch_lock);
if (err) {
pr_err("Unable to wait for decompression batch budget: %s\n", strerror(err));
BUG();
}
}
decompress_batches_available--;
pthread_mutex_unlock(&decompress_batch_lock);
}
bool decompression_batch_try_acquire(void)
{
bool acquired = false;
int err;
if (decompress_shared_budget)
return shared_budget_try_acquire(&decompress_shared_budget->batches);
err = pthread_mutex_lock(&decompress_batch_lock);
if (err) {
pr_warn("Unable to lock compressed-page batch budget: %s\n", strerror(err));
return false;
}
if (decompress_batches_available) {
decompress_batches_available--;
acquired = true;
}
pthread_mutex_unlock(&decompress_batch_lock);
return acquired;
}
void decompression_batch_release(void)
{
int err;
if (decompress_shared_budget) {
shared_budget_release(&decompress_shared_budget->batches, 1);
return;
}
err = pthread_mutex_lock(&decompress_batch_lock);
if (err) {
pr_err("Unable to lock decompression batch budget for release: %s\n", strerror(err));
BUG();
}
decompress_batches_available++;
pthread_cond_signal(&decompress_batch_cond);
pthread_mutex_unlock(&decompress_batch_lock);
}
int compress_data(const char *input_data, size_t input_size,
char *compressed_data, size_t output_size,
int acceleration)
@ -165,3 +484,506 @@ int decompress_region(const char *src, int compressed_size,
return 0;
}
/* Parallel decompression queue and reusable worker pool. */
struct decompress_queue {
struct decompress_job *jobs;
size_t nr_jobs;
size_t job_chunk;
/*
* Concurrent scheduling and result accesses use compiler atomics. Keep all
* three fields under that API because next_job needs size_t, while CRIU's
* atomic_t is int-sized.
*/
size_t next_job;
int failed;
int failed_block;
};
struct decompress_worker {
pthread_t tid;
struct decompression_pool *pool;
unsigned int index;
};
/*
* The lock protects the batch scheduler fields below. decompression_pool_start_batch()
* publishes the caller-owned jobs as immutable batch metadata. Workers update
* only the queue's atomic scheduling and result fields.
* decompression_pool_finish_batch() waits for every selected worker before
* releasing those jobs.
*/
struct decompression_pool {
pthread_mutex_t lock;
pthread_cond_t work_ready;
pthread_cond_t work_done;
struct decompress_worker *workers;
struct decompress_queue queue;
unsigned int nr_workers;
unsigned int batch_workers;
unsigned int pending_batch_workers;
unsigned long batch_generation;
bool batch_active;
bool stop;
};
static int decompress_job_run(const struct decompress_job *job)
{
size_t block_bytes;
if (!job->pages || job->pages > MAX_REGION_PAGES)
return -1;
block_bytes = (size_t)job->pages * PAGE_SIZE;
if (!job->compressed_size) {
memset(job->dst, 0, block_bytes);
return 0;
}
/* Raw blocks are handled inline by the pagemap reader. */
if (job->compressed_size >= block_bytes)
return -1;
return decompress_data_nolog(job->src, job->compressed_size, block_bytes, job->dst);
}
static void decompress_queue_init(struct decompress_queue *queue, struct decompress_job *jobs, size_t nr_jobs,
unsigned int nr_threads)
{
queue->jobs = jobs;
queue->nr_jobs = nr_jobs;
queue->job_chunk = nr_jobs / nr_threads / 4;
if (!queue->job_chunk)
queue->job_chunk = 1;
if (queue->job_chunk > PARALLEL_DECOMPRESS_MAX_JOB_CHUNK)
queue->job_chunk = PARALLEL_DECOMPRESS_MAX_JOB_CHUNK;
queue->next_job = 0;
queue->failed = 0;
queue->failed_block = INT_MAX;
}
static bool decompress_queue_failed(const struct decompress_queue *queue)
{
return __atomic_load_n(&queue->failed, __ATOMIC_ACQUIRE);
}
static bool decompress_queue_claim_chunk(struct decompress_queue *queue, size_t *first, size_t *end)
{
if (decompress_queue_failed(queue))
return false;
/* This relaxed counter only assigns disjoint chunks; it publishes no data. */
*first = __atomic_fetch_add(&queue->next_job, queue->job_chunk, __ATOMIC_RELAXED);
if (*first >= queue->nr_jobs)
return false;
*end = *first + queue->job_chunk;
if (*end > queue->nr_jobs)
*end = queue->nr_jobs;
return true;
}
static void decompress_queue_record_failure(struct decompress_queue *queue, int block_index)
{
int earliest_block = __atomic_load_n(&queue->failed_block, __ATOMIC_RELAXED);
/* Keep the lowest failing block index when several workers fail. */
while (block_index < earliest_block) {
if (__atomic_compare_exchange_n(&queue->failed_block, &earliest_block, block_index, false,
__ATOMIC_RELAXED, __ATOMIC_RELAXED))
break;
}
/* Publish cancellation only after failed_block has been updated. */
__atomic_store_n(&queue->failed, 1, __ATOMIC_RELEASE);
}
static void decompress_queue_run(struct decompress_queue *queue)
{
size_t first, end;
while (decompress_queue_claim_chunk(queue, &first, &end)) {
size_t i;
for (i = first; i < end; i++) {
struct decompress_job *job = &queue->jobs[i];
if (!decompress_job_run(job))
continue;
decompress_queue_record_failure(queue, job->block_index);
break;
}
}
}
static int decompress_jobs_serial(struct decompress_job *jobs, size_t nr_jobs)
{
struct decompress_queue queue;
decompress_queue_init(&queue, jobs, nr_jobs, 1);
decompress_queue_run(&queue);
if (!decompress_queue_failed(&queue))
return 0;
pr_err("Decompression failed at block %d\n", queue.failed_block);
return -1;
}
static unsigned int decompress_batch_threads(size_t nr_jobs, size_t total_uncompressed, unsigned int requested_threads)
{
size_t work_limit;
unsigned int nr_threads;
if (total_uncompressed < PARALLEL_RESTORE_MIN_BATCH_BYTES)
return 1;
nr_threads = decompression_cpu_limit(requested_threads);
work_limit = nr_jobs;
if (work_limit < nr_threads)
nr_threads = (unsigned int)work_limit;
work_limit = total_uncompressed / PARALLEL_DECOMPRESS_MIN_BYTES_PER_THREAD;
if (work_limit < nr_threads)
nr_threads = (unsigned int)work_limit;
if (nr_threads < 2)
return 1;
return nr_threads;
}
static void decompression_worker_signal_set(sigset_t *set)
{
/*
* CRIU's process-global signal handling belongs to the restore thread.
* Worker threads keep asynchronous signals blocked for their lifetime.
* Synchronous faults must remain deliverable.
*/
sigfillset(set);
sigdelset(set, SIGABRT);
sigdelset(set, SIGBUS);
sigdelset(set, SIGFPE);
sigdelset(set, SIGILL);
sigdelset(set, SIGSEGV);
sigdelset(set, SIGSYS);
sigdelset(set, SIGTRAP);
}
static void *decompression_pool_worker(void *arg)
{
struct decompress_worker *worker = arg;
struct decompression_pool *pool = worker->pool;
unsigned long seen_generation = 0;
int err;
err = pthread_mutex_lock(&pool->lock);
if (err) {
pr_err("Unable to lock decompression worker pool: %s\n", strerror(err));
BUG();
}
for (;;) {
struct decompress_queue *queue;
/* Wait for a new batch that selected this worker. */
while (!pool->stop) {
if (seen_generation != pool->batch_generation && worker->index < pool->batch_workers)
break;
err = pthread_cond_wait(&pool->work_ready, &pool->lock);
if (err) {
pr_err("Unable to wait for decompression work: %s\n", strerror(err));
BUG();
}
}
if (pool->stop)
break;
seen_generation = pool->batch_generation;
queue = &pool->queue;
pthread_mutex_unlock(&pool->lock);
decompress_queue_run(queue);
err = pthread_mutex_lock(&pool->lock);
if (err) {
pr_err("Unable to lock completed decompression work: %s\n", strerror(err));
BUG();
}
BUG_ON(!pool->pending_batch_workers);
pool->pending_batch_workers--;
if (!pool->pending_batch_workers)
pthread_cond_signal(&pool->work_done);
}
pthread_mutex_unlock(&pool->lock);
return NULL;
}
static void decompression_pool_spawn_workers(struct decompression_pool *pool, unsigned int nr_workers,
pthread_attr_t *attrp)
{
unsigned int attempt;
/* Successful workers are packed; a failed attempt does not consume a slot. */
for (attempt = 0; attempt < nr_workers; attempt++) {
struct decompress_worker *worker = &pool->workers[pool->nr_workers];
int err;
worker->pool = pool;
worker->index = pool->nr_workers;
err = pthread_create(&worker->tid, attrp, decompression_pool_worker, worker);
if (err) {
pr_warn("Unable to start decompression worker %u: %s\n", attempt, strerror(err));
break;
}
pool->nr_workers++;
}
}
static struct decompression_pool *decompression_pool_create(unsigned int nr_threads, bool *mask_restore_failed)
{
struct decompression_pool *pool;
pthread_attr_t attr;
sigset_t set, old;
unsigned int nr_workers;
size_t stack_size = PARALLEL_DECOMPRESS_STACK_SIZE;
int err;
*mask_restore_failed = false;
if (nr_threads <= 1)
return NULL;
nr_workers = nr_threads - 1;
pool = xzalloc(sizeof(*pool));
if (!pool)
return NULL;
pool->workers = xzalloc((size_t)nr_workers * sizeof(*pool->workers));
if (!pool->workers)
goto free_pool;
/* Initialize every synchronization object before starting workers. */
err = pthread_mutex_init(&pool->lock, NULL);
if (err) {
pr_warn("Unable to initialize decompression pool lock: %s\n", strerror(err));
goto free_workers;
}
err = pthread_cond_init(&pool->work_ready, NULL);
if (err) {
pr_warn("Unable to initialize decompression work condition: %s\n", strerror(err));
goto destroy_lock;
}
err = pthread_cond_init(&pool->work_done, NULL);
if (err) {
pr_warn("Unable to initialize decompression completion condition: %s\n", strerror(err));
goto destroy_ready;
}
/* A bounded worker stack keeps large restore pools inexpensive. */
err = pthread_attr_init(&attr);
if (err) {
pr_warn("Unable to initialize decompression worker attributes: %s\n", strerror(err));
goto destroy_done;
}
if (stack_size < PTHREAD_STACK_MIN)
stack_size = PTHREAD_STACK_MIN;
err = pthread_attr_setstacksize(&attr, stack_size);
if (err) {
pr_warn("Unable to set decompression worker stack size: %s\n", strerror(err));
goto destroy_attr;
}
/* Workers inherit blocked asynchronous signals from their creator. */
decompression_worker_signal_set(&set);
err = pthread_sigmask(SIG_BLOCK, &set, &old);
if (err) {
pr_warn("Unable to block signals for decompression workers: %s\n", strerror(err));
goto destroy_attr;
}
decompression_pool_spawn_workers(pool, nr_workers, &attr);
/* Restore the calling thread's mask after all workers have started. */
err = pthread_sigmask(SIG_SETMASK, &old, NULL);
if (err) {
pr_err("Unable to restore signal mask after starting decompression workers: %s\n", strerror(err));
*mask_restore_failed = true;
}
destroy_attr:
pthread_attr_destroy(&attr);
if (pool->nr_workers)
return pool;
destroy_done:
pthread_cond_destroy(&pool->work_done);
destroy_ready:
pthread_cond_destroy(&pool->work_ready);
destroy_lock:
pthread_mutex_destroy(&pool->lock);
free_workers:
xfree(pool->workers);
free_pool:
xfree(pool);
return NULL;
}
void decompression_pool_destroy(struct decompression_pool *pool)
{
unsigned int w;
int err;
if (!pool)
return;
err = pthread_mutex_lock(&pool->lock);
if (err) {
pr_err("Unable to lock decompression pool for shutdown: %s\n", strerror(err));
BUG();
}
BUG_ON(pool->batch_active || pool->pending_batch_workers);
pool->stop = true;
pthread_cond_broadcast(&pool->work_ready);
pthread_mutex_unlock(&pool->lock);
for (w = 0; w < pool->nr_workers; w++) {
err = pthread_join(pool->workers[w].tid, NULL);
if (err) {
pr_err("Unable to join decompression worker %u: %s\n", w, strerror(err));
BUG();
}
}
pthread_cond_destroy(&pool->work_done);
pthread_cond_destroy(&pool->work_ready);
pthread_mutex_destroy(&pool->lock);
xfree(pool->workers);
xfree(pool);
}
/* Publish one immutable job array to the selected subset of pool workers. */
static void decompression_pool_start_batch(struct decompression_pool *pool, struct decompress_job *jobs,
size_t nr_jobs, unsigned int active_threads)
{
int err;
err = pthread_mutex_lock(&pool->lock);
if (err) {
pr_err("Unable to lock decompression pool for dispatch: %s\n", strerror(err));
BUG();
}
BUG_ON(pool->batch_active || pool->pending_batch_workers);
decompress_queue_init(&pool->queue, jobs, nr_jobs, active_threads);
pool->batch_active = true;
pool->batch_workers = active_threads - 1;
pool->pending_batch_workers = pool->batch_workers;
pool->batch_generation++;
pthread_cond_broadcast(&pool->work_ready);
pthread_mutex_unlock(&pool->lock);
}
/* Wait until workers have dropped every reference to the caller-owned jobs. */
static bool decompression_pool_finish_batch(struct decompression_pool *pool, int *failed_block)
{
bool failed;
int err;
err = pthread_mutex_lock(&pool->lock);
if (err) {
pr_err("Unable to lock decompression pool for completion: %s\n", strerror(err));
BUG();
}
while (pool->pending_batch_workers) {
err = pthread_cond_wait(&pool->work_done, &pool->lock);
if (err) {
pr_err("Unable to wait for decompression completion: %s\n", strerror(err));
BUG();
}
}
failed = decompress_queue_failed(&pool->queue);
*failed_block = pool->queue.failed_block;
pool->queue.jobs = NULL;
pool->queue.nr_jobs = 0;
pool->batch_active = false;
pool->batch_workers = 0;
pthread_mutex_unlock(&pool->lock);
return failed;
}
int decompress_jobs_parallel_pool_with_caller_work(
struct decompression_pool **poolp, struct decompress_job *jobs,
size_t nr_jobs, size_t total_uncompressed,
unsigned int requested_threads,
decompression_caller_work_fn caller_work, void *caller_work_arg)
{
struct decompression_pool *pool;
unsigned int active_threads, useful_threads, threads_held;
bool mask_restore_failed = false;
int failed_block = INT_MAX, ret = 0;
if (!nr_jobs)
return 0;
if (!jobs || !poolp)
return -1;
/* Reserve restore-wide CPU slots after choosing this batch's useful width. */
useful_threads = decompress_batch_threads(nr_jobs, total_uncompressed, requested_threads);
threads_held = decompress_budget_acquire(useful_threads);
if (!threads_held)
return decompress_jobs_serial(jobs, nr_jobs);
if (threads_held == 1) {
ret = decompress_jobs_serial(jobs, nr_jobs);
goto out_release_budget;
}
pool = *poolp;
if (pool && threads_held > pool->nr_workers + 1) {
decompression_pool_destroy(pool);
pool = NULL;
*poolp = NULL;
}
if (!pool) {
pool = decompression_pool_create(threads_held, &mask_restore_failed);
*poolp = pool;
}
if (mask_restore_failed) {
decompression_pool_destroy(pool);
*poolp = NULL;
ret = -1;
goto out_release_budget;
}
if (!pool) {
pr_warn("Unable to start decompression workers, using serial decompression\n");
decompress_budget_trim_reservation(&threads_held, 1);
ret = decompress_jobs_serial(jobs, nr_jobs);
goto out_release_budget;
}
/* A partially created pool may need fewer slots than were reserved. */
active_threads = min(threads_held, pool->nr_workers + 1);
decompress_budget_trim_reservation(&threads_held, active_threads);
decompression_pool_start_batch(pool, jobs, nr_jobs, active_threads);
if (caller_work)
caller_work(caller_work_arg);
/* After optional caller work, consume any jobs the pool has not claimed. */
decompress_queue_run(&pool->queue);
if (decompression_pool_finish_batch(pool, &failed_block)) {
pr_err("Decompression failed at block %d\n", failed_block);
ret = -1;
}
out_release_budget:
decompress_budget_release(threads_held);
return ret;
}
int decompress_jobs_parallel_pool(struct decompression_pool **poolp,
struct decompress_job *jobs, size_t nr_jobs,
size_t total_uncompressed,
unsigned int requested_threads)
{
return decompress_jobs_parallel_pool_with_caller_work(
poolp, jobs, nr_jobs, total_uncompressed, requested_threads,
NULL, NULL);
}

1
criu/config.c
View file

@ -433,6 +433,7 @@ void init_opts(void)
opts.image_io_mode = IMAGE_IO_DEFAULT;
opts.network_lock_method = NETWORK_LOCK_DEFAULT;
opts.ghost_fiemap = FIEMAP_DEFAULT;
opts.decompress_threads = 1;
}
bool deprecated_ok(char *what)

16
criu/cr-restore.c
View file

@ -50,6 +50,7 @@
#include "uffd.h"
#include "namespaces.h"
#include "asyncd.h"
#include "compression.h"
#include "mem.h"
#include "mount.h"
#include "fsnotify.h"
@ -2395,6 +2396,9 @@ int prepare_task_entries(void)
mutex_init(&task_entries->userns_sync_lock);
mutex_init(&task_entries->cgroupd_sync_lock);
mutex_init(&task_entries->last_pid_mutex);
decompression_shared_budget_init(&task_entries->decompression_budget,
opts.decompress_threads);
decompression_use_shared_budget(&task_entries->decompression_budget);
return 0;
}
@ -3402,6 +3406,18 @@ static int sigreturn_restore(pid_t pid, struct task_restore_args *task_args, uns
RST_MEM_FIXUP_PPTR(task_args->helpers);
RST_MEM_FIXUP_PPTR(task_args->zombies);
RST_MEM_FIXUP_PPTR(task_args->vma_ios);
{
struct restore_vma_io *rio = task_args->vma_ios;
unsigned int n;
for (n = 0; n < task_args->vma_ios_n; n++) {
if (rio->compressed_size)
RST_MEM_FIXUP_PPTR(rio->compressed_size);
if (rio->block_pages)
RST_MEM_FIXUP_PPTR(rio->block_pages);
rio = (struct restore_vma_io *)((char *)rio + RIO_SIZE(rio->nr_iovs));
}
}
RST_MEM_FIXUP_PPTR(task_args->inotify_fds);
task_args->compatible_mode = core_is_compat(core);

126
criu/include/compression.h
View file

@ -5,6 +5,7 @@
#include <stdint.h>
#include <sys/types.h>
#include "common/lock.h"
#include "page.h"
/*
@ -27,6 +28,9 @@ enum compress_mode {
#define MAX_REGION_PAGES (MAX_REGION_SIZE / PAGE_SIZE)
#define DEFAULT_REGION_PAGES (DEFAULT_REGION_SIZE / PAGE_SIZE)
/* Minimum useful batch before starting compressed-page restore workers. */
#define PARALLEL_RESTORE_MIN_BATCH_BYTES (1UL << 20)
/* LZ4 worst-case compressed size for one page: src + src/255 + 16 */
#define PAGE_COMPRESSED_SIZE_BOUND (PAGE_SIZE + (PAGE_SIZE / 255) + 16)
@ -87,6 +91,34 @@ static inline bool page_is_all_zero(const char *page)
return true;
}
/*
* One block that can be restored independently. Jobs passed together must
* write to disjoint destinations. A zero compressed size clears the block;
* other jobs hold LZ4 payloads. Raw-fallback blocks remain caller-owned.
*/
struct decompress_job {
const char *src;
char *dst;
uint32_t compressed_size;
uint16_t pages;
int block_index;
};
struct decompression_pool;
typedef void (*decompression_caller_work_fn)(void *arg);
/*
* Restore-wide limits live in the shared restore mapping. All task restore
* processes and asyncd workers inherit the same object, bounding active
* compressed-page workers and large encoded working sets across siblings.
*/
struct decompression_shared_budget {
futex_t threads;
futex_t batches;
/* Immutable capacity used for restore-path selection. */
unsigned int thread_capacity;
};
#ifdef CONFIG_LZ4
int compress_data(const char *input_data, size_t input_size,
@ -116,6 +148,50 @@ int compress_region(const char *src, unsigned int n_pages, char *dst,
int decompress_region(const char *src, int compressed_size,
unsigned int n_pages, char *dst);
/*
* Reuse worker threads through @pool across related batches. A later wider
* batch may replace the pool. @requested_threads includes the caller; zero
* selects automatic concurrency and one keeps the call serial. The active
* width is bounded by available CPUs, useful batch work, and the shared CPU
* budget. Small batches run serially. Calls sharing a pool must be serialized.
* @pool must initially be NULL and must be destroyed before the caller forks
* or remaps its address space.
*/
int decompress_jobs_parallel_pool(struct decompression_pool **pool,
struct decompress_job *jobs,
size_t nr_jobs,
size_t total_uncompressed,
unsigned int requested_threads);
/*
* After dispatching a parallel batch, run @caller_work once in the calling
* thread before it helps the pool. Serial fallbacks skip the callback. The
* callback must record its own result and must not mutate @jobs, re-enter
* @pool, or acquire another worker-budget reservation.
*/
int decompress_jobs_parallel_pool_with_caller_work(
struct decompression_pool **pool, struct decompress_job *jobs,
size_t nr_jobs, size_t total_uncompressed,
unsigned int requested_threads,
decompression_caller_work_fn caller_work, void *caller_work_arg);
void decompression_pool_destroy(struct decompression_pool *pool);
/*
* Initialize the restore-wide CPU and encoded-working-set budgets. Automatic
* (0) and serial (1) per-call settings still permit independent restore
* requests to run on separate CPUs; values above one cap their aggregate
* worker width.
*/
void decompression_shared_budget_init(struct decompression_shared_budget *budget,
unsigned int requested_threads);
void decompression_use_shared_budget(struct decompression_shared_budget *budget);
void decompression_batch_acquire(void);
bool decompression_batch_try_acquire(void);
void decompression_batch_release(void);
bool compressed_restore_has_parallel_capacity(unsigned int requested_threads);
/* Apply the requested auto/explicit setting to the detected CPU capacity. */
unsigned int decompression_thread_limit(unsigned int requested,
unsigned int available_cpus);
#else /* !CONFIG_LZ4 */
static inline int compress_data(const char *in, size_t in_sz, char *out,
@ -142,6 +218,56 @@ static inline int decompress_region(const char *src, int comp_sz,
return -1;
}
static inline int decompress_jobs_parallel_pool(
struct decompression_pool **pool, struct decompress_job *jobs,
size_t nr_jobs, size_t total_uncompressed,
unsigned int requested_threads)
{
return -1;
}
static inline int decompress_jobs_parallel_pool_with_caller_work(
struct decompression_pool **pool, struct decompress_job *jobs,
size_t nr_jobs, size_t total_uncompressed,
unsigned int requested_threads,
decompression_caller_work_fn caller_work, void *caller_work_arg)
{
return -1;
}
static inline void decompression_pool_destroy(struct decompression_pool *pool)
{
}
static inline void decompression_shared_budget_init(
struct decompression_shared_budget *budget, unsigned int requested_threads)
{
}
static inline void decompression_use_shared_budget(
struct decompression_shared_budget *budget)
{
}
static inline void decompression_batch_acquire(void)
{
}
static inline bool decompression_batch_try_acquire(void)
{
return false;
}
static inline void decompression_batch_release(void)
{
}
static inline bool compressed_restore_has_parallel_capacity(
unsigned int requested_threads)
{
return false;
}
#endif /* CONFIG_LZ4 */
#endif /* __CR_COMPRESSION_H__ */

8
criu/include/page-pipe.h
View file

@ -96,6 +96,7 @@ struct page_pipe_buf {
unsigned long pipe_off; /* where this buf is started in a pipe */
unsigned long pages_in; /* how many pages are there */
#define PPB_LAZY (1 << 0)
#define PPB_FORCE_RAW (1 << 1)
unsigned int flags;
struct iovec *iov; /* vaddr:len map */
struct list_head l; /* links into page_pipe->bufs */
@ -105,10 +106,11 @@ struct page_pipe_buf {
* Page pipe buffers with different flags cannot share the same pipe.
* We track the last ppb that was used for each type separately in the
* prev[] array in the struct page_pipe (below).
* Currently we have 2 types: the buffers that are always stored in
* the images and the buffers that are lazily migrated
* PPB_LAZY and PPB_FORCE_RAW form the type index. Keeping force-raw
* buffers separate prevents a pipe segment that may be LZ4-compressed
* from being transferred under the force-raw policy (or vice versa).
*/
#define PP_PIPE_TYPES 2
#define PP_PIPE_TYPES 4
#define PP_HOLE_PARENT (1 << 0)

2
criu/include/page-xfer.h
View file

@ -31,6 +31,8 @@ struct page_xfer {
*/
unsigned long offset;
bool transfer_lazy;
/* Store the current page-pipe segment as compressed-format raw/zero blocks. */
bool force_raw;
/*
* Local non-streaming dump only: the pages image fd was switched to

40
criu/include/pagemap.h
View file

@ -45,6 +45,8 @@
* All this is implemented in read_pagemap_page.
*/
struct encoded_read_ctx;
struct page_read {
/* reads page from current pagemap */
int (*read_pages)(struct page_read *, unsigned long vaddr, unsigned long nr, void *, unsigned flags);
@ -128,6 +130,15 @@ struct page_read {
unsigned long cached_region_vaddr;
size_t cached_region_size;
/*
* Bounded encoded buffers and workers. All readers in an incremental
* parent chain use the context owned by encoded_read_owner, so a sync
* initiated by a parent cannot acquire a second batch lease and deadlock
* against its child. Only the owner releases the context at close.
*/
struct encoded_read_ctx *encoded_read_ctx;
struct page_read *encoded_read_owner;
/* Record consequent neighbour iov-ecs to punch together */
struct iovec bunch;
@ -169,6 +180,35 @@ struct task_restore_args;
int pagemap_enqueue_iovec(struct page_read *pr, void *buf, unsigned long len, struct list_head *to);
int pagemap_render_iovec(struct list_head *from, struct task_restore_args *ta);
/*
* Return true when every file-backed queued read can be submitted with
* O_DIRECT. Encoded storage is rejected; raw storage and all destination
* extents must be page-aligned. Zero entries have no file extent, and a
* zero-only list returns false because it needs no AIO setup.
*/
bool pagemap_iovec_is_direct_compatible(const struct list_head *from);
/*
* Return whether any page-image block overlapping [start, end) contains an
* actual LZ4 payload. Raw-fallback and zero blocks return false. Parent-image
* entries are followed without changing either reader's cursor.
*
* A local lookup is O(log(nr_pmes) + overlapping entries/blocks). Each
* inherited span repeats that lookup in the parent, so a fragmented parent
* chain additionally pays one logarithmic lookup per inherited span. A
* negative return indicates invalid input or an inconsistent parent chain.
*/
int page_read_range_has_lz4(struct page_read *pr, unsigned long start,
unsigned long end);
/* Also return true when raw/zero runs exceed the bounded direct-PIE limit. */
int page_read_range_needs_premap(struct page_read *pr, unsigned long start,
unsigned long end);
/* Return whether the top image delegates any part of [start, end) to a parent. */
int page_read_range_has_parent(struct page_read *pr, unsigned long start,
unsigned long end);
/*
* Try to enable O_DIRECT on a pages-image fd and verify with one
* aligned probe read.

16
criu/include/restorer.h
View file

@ -135,9 +135,25 @@ struct thread_restore_args {
typedef long (*thread_restore_fcall_t)(struct thread_restore_args *args);
/*
* Internal representation of a pages-image range queued for PIE restore.
*
* PACKED_RAW and ZERO originate from entries which still carry compression
* metadata. They are separate from UNCOMPRESSED so that the restorer can
* bypass LZ4 without treating their packed image offsets as ordinary pages
* image offsets (in particular, --auto-dedup must not punch PACKED_RAW).
*/
enum restore_vma_io_storage {
VMA_IO_UNCOMPRESSED,
VMA_IO_ENCODED,
VMA_IO_PACKED_RAW,
VMA_IO_ZERO,
};
struct restore_vma_io {
int nr_iovs;
loff_t off;
enum restore_vma_io_storage storage;
uint32_t *compressed_size;
uint64_t total_compressed_size;
int n_compressed_size;

2
criu/include/rst_info.h
View file

@ -6,6 +6,7 @@
#include "common/list.h"
#include "vma.h"
#include "kerndat.h"
#include "compression.h"
#include "images/mm.pb-c.h"
#include "images/core.pb-c.h"
@ -17,6 +18,7 @@ struct task_entries {
mutex_t userns_sync_lock;
mutex_t cgroupd_sync_lock;
mutex_t last_pid_mutex;
struct decompression_shared_budget decompression_budget;
int asyncd_sk_id;
};

211
criu/mem.c
View file

@ -229,9 +229,14 @@ static int generate_iovs(struct pstree_item *item, struct vma_area *vma, struct
unsigned long pages[3] = {};
unsigned long vaddr;
bool dump_all_pages;
bool self_contained;
bool force_raw;
int ret = 0;
self_contained = (vma->e->flags & MAP_HUGETLB) ||
(vma->e->status & VMA_EXT_PLUGIN);
dump_all_pages = should_dump_entire_vma(vma->e);
force_raw = opts.compress_mode && self_contained;
/*
* In region-compression mode, force the first page of this VMA to
@ -262,6 +267,8 @@ static int generate_iovs(struct pstree_item *item, struct vma_area *vma, struct
if (vma_entry_can_be_lazy(vma->e) && !is_stack(item, vaddr))
ppb_flags |= PPB_LAZY;
if (force_raw)
ppb_flags |= PPB_FORCE_RAW;
/*
* If we're doing incremental dump (parent images
@ -270,7 +277,15 @@ static int generate_iovs(struct pstree_item *item, struct vma_area *vma, struct
* page. The latter would be checked in page-xfer.
*/
if (has_parent && page_in_parent(page_info.softdirty)) {
/*
* Hugetlb and external-plugin VMAs cannot use the generic premap
* path, and delayed PIE I/O cannot redirect a PE_PARENT range to a
* different pages image. Keep these rare mappings self-contained at
* every image level. With compression enabled, force_raw additionally
* ensures PIE never has to decode LZ4.
*/
if (has_parent && !self_contained &&
page_in_parent(page_info.softdirty)) {
ret = page_pipe_add_hole(pp, vaddr, PP_HOLE_PARENT);
st = 0;
} else {
@ -1093,6 +1108,11 @@ static int premap_priv_vmas(struct pstree_item *t, struct vm_area_list *vmas, vo
filemap_ctx_init(true);
list_for_each_entry(vma, &vmas->h, list) {
bool exceptional;
int has_lz4 = 0;
int has_parent = 0;
int needs_premap = 0;
if (vma_area_is(vma, VMA_AREA_GUARD))
continue;
@ -1109,15 +1129,77 @@ static int premap_priv_vmas(struct pstree_item *t, struct vm_area_list *vmas, vo
if (!vma_area_is_private(vma, kdat.task_size))
continue;
exceptional = (vma->e->flags & MAP_HUGETLB) ||
(vma->e->status & VMA_EXT_PLUGIN);
if (vma->e->flags & MAP_HUGETLB)
/*
* PIE cannot link against liblz4. Premap ordinary private VMAs which
* contain an actual LZ4 block so their content is decoded by the
* normal page reader before switching to PIE. Also premap large zero
* runs through the compressed-page worker pool, and entries whose
* raw/zero run count exceeds the direct-PIE bound. Short runs then
* coalesce in-process, while long runs retain direct I/O. Uniform and
* lightly fragmented ranges keep the faster delayed path.
*
* Hugetlb and external-plugin VMAs cannot use the generic premap
* path. The dump side therefore guarantees that their compressed
* pagemap entries contain raw/zero blocks only. Reject an image
* which violates that invariant before the destructive restore.
*/
if (opts.compress_mode) {
if (exceptional) {
has_lz4 = page_read_range_has_lz4(pr, vma->e->start, vma->e->end);
if (has_lz4 < 0) {
ret = -1;
break;
}
} else {
needs_premap = page_read_range_needs_premap(pr, vma->e->start, vma->e->end);
if (needs_premap < 0) {
ret = -1;
break;
}
}
}
if (exceptional) {
has_parent = page_read_range_has_parent(pr, vma->e->start, vma->e->end);
if (has_parent < 0) {
ret = -1;
break;
}
if (has_parent) {
pr_err("Non-premapped VMA %#" PRIx64 "-%#" PRIx64
" contains pages inherited from a parent image\n",
vma->e->start, vma->e->end);
ret = -1;
break;
}
}
if (vma->e->flags & MAP_HUGETLB) {
if (has_lz4) {
pr_err("Hugetlb VMA %#" PRIx64 "-%#" PRIx64
" contains an LZ4 block\n",
vma->e->start, vma->e->end);
ret = -1;
break;
}
continue;
}
/* VMA offset may change due to plugin so we cannot premap */
if (vma->e->status & VMA_EXT_PLUGIN)
if (vma->e->status & VMA_EXT_PLUGIN) {
if (has_lz4) {
pr_err("External-plugin VMA %#" PRIx64 "-%#" PRIx64
" contains an LZ4 block\n",
vma->e->start, vma->e->end);
ret = -1;
break;
}
continue;
}
if (vma->pvma == NULL && pr->pieok && !vma_force_premap(vma, &vmas->h)) {
if (vma->pvma == NULL && pr->pieok && !needs_premap && !vma_force_premap(vma, &vmas->h)) {
/*
* VMA in question is not shared with anyone. We'll
* restore it with its contents in restorer.
@ -1146,6 +1228,8 @@ static int premap_priv_vmas(struct pstree_item *t, struct vm_area_list *vmas, vo
return ret;
}
#define COW_READ_BATCH_PAGES (1UL << 8)
static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
{
struct vma_area *vma;
@ -1162,18 +1246,11 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
unsigned int nr_lazy = 0;
unsigned long va;
void *buf = NULL;
int memerr;
bool page_read_closed = false;
vma = list_first_entry(vmas, struct vma_area, list);
rsti(t)->pages_img_id = pr->pages_img_id;
/* O_DIRECT may require the buffer to be aligned. */
memerr = posix_memalign(&buf, PAGE_SIZE, PAGE_SIZE);
if (memerr) {
pr_err("Can't allocate COW buffer: %s\n", strerror(memerr));
return -1;
}
/*
* Read page contents.
*/
@ -1198,7 +1275,8 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
continue;
}
for (i = 0; i < nr_pages; i++) {
i = 0;
while (i < nr_pages) {
void *p;
/*
@ -1241,7 +1319,7 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
va += len;
len >>= PAGE_SHIFT;
nr_restored += len;
i += len - 1;
i += len;
nr_enqueued++;
continue;
@ -1254,24 +1332,64 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
off = (va - vma->e->start) / PAGE_SIZE;
p = decode_pointer((off)*PAGE_SIZE + vma->premmaped_addr);
set_bit(off, vma->page_bitmap);
if (vma_inherited(vma)) {
clear_bit(off, vma->pvma->page_bitmap);
unsigned long nr, j;
int memerr;
ret = pr->read_pages(pr, va, 1, buf, 0);
if (ret < 0)
goto err_read;
/*
* A page-at-a-time read turns compressed COW restore
* into one pread() and one LZ4 call per page. Read a
* bounded aligned batch, then retain the existing
* page-by-page sharing decision.
*/
nr = min_t(unsigned long, nr_pages - i,
(vma->e->end - va) / PAGE_SIZE);
nr = min(nr, COW_READ_BATCH_PAGES);
if (pr->pe->has_region_pages && pr->pe->region_pages &&
nr < nr_pages - i) {
unsigned long aligned =
nr - nr % pr->pe->region_pages;
va += PAGE_SIZE;
nr_compared++;
if (memcmp(p, buf, PAGE_SIZE) == 0) {
nr_shared++; /* the page is cowed */
continue;
if (aligned)
nr = aligned;
}
if (!buf) {
memerr = posix_memalign(&buf, PAGE_SIZE,
COW_READ_BATCH_PAGES * PAGE_SIZE);
if (memerr) {
pr_err("Can't allocate COW buffer: %s\n",
strerror(memerr));
ret = -1;
goto err_read;
}
}
nr_restored++;
memcpy(p, buf, PAGE_SIZE);
ret = pr->read_pages(pr, va, nr, buf, PR_ASYNC);
if (ret < 0)
goto err_read;
if (pr->sync(pr)) {
ret = -1;
goto err_read;
}
for (j = 0; j < nr; j++) {
void *src = (char *)buf + j * PAGE_SIZE;
void *dst = (char *)p + j * PAGE_SIZE;
set_bit(off + j, vma->page_bitmap);
clear_bit(off + j, vma->pvma->page_bitmap);
nr_compared++;
if (memcmp(dst, src, PAGE_SIZE) == 0) {
nr_shared++; /* the page is cowed */
continue;
}
nr_restored++;
memcpy(dst, src, PAGE_SIZE);
}
va += nr * PAGE_SIZE;
i += nr;
} else {
int nr;
@ -1286,13 +1404,14 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
nr = min_t(int, nr_pages - i, (vma->e->end - va) / PAGE_SIZE);
set_bit(off, vma->page_bitmap);
ret = pr->read_pages(pr, va, nr, p, PR_ASYNC);
if (ret < 0)
goto err_read;
va += nr * PAGE_SIZE;
nr_restored += nr;
i += nr - 1;
i += nr;
bitmap_set(vma->page_bitmap, off + 1, nr - 1);
}
@ -1300,10 +1419,17 @@ static int restore_priv_vma_content(struct pstree_item *t, struct page_read *pr)
}
err_read:
if (pr->sync(pr))
goto out;
{
int sync_ret = pr->sync(pr);
pr->close(pr);
page_read_closed = true;
if (sync_ret) {
ret = -1;
goto out;
}
}
pr->close(pr);
if (ret < 0) {
exit_code = ret;
goto out;
@ -1354,6 +1480,11 @@ err_read:
err_addr:
pr_err("Page entry address %lx outside of VMA %lx-%lx\n", va, (long)vma->e->start, (long)vma->e->end);
out:
if (!page_read_closed) {
if (pr->sync(pr))
exit_code = -1;
pr->close(pr);
}
xfree(buf);
return exit_code;
}
@ -1529,6 +1660,7 @@ int open_vmas(struct pstree_item *t)
static int prepare_vma_ios(struct pstree_item *t, struct task_restore_args *ta)
{
struct cr_img *pages;
int ret;
/*
* We optimize the case when rsti(t)->vma_io is empty.
@ -1554,15 +1686,28 @@ static int prepare_vma_ios(struct pstree_item *t, struct task_restore_args *ta)
return -1;
ta->vma_ios_fd = img_raw_fd(pages);
if (ta->vma_ios_fd >= 0 && opts.image_io_mode == IMAGE_IO_DIRECT) {
/*
* Select direct AIO from the actual delayed ranges, not the inventory-wide
* compression mode. LZ4 ranges were premapped above, while aligned raw
* fallbacks and zero ranges remain safe for the PIE fast path.
*/
if (ta->vma_ios_fd >= 0 && opts.image_io_mode == IMAGE_IO_DIRECT &&
pagemap_iovec_is_direct_compatible(&rsti(t)->vma_io)) {
int direct = probe_pages_o_direct(ta->vma_ios_fd);
if (direct < 0) {
close_image(pages);
ta->vma_ios_fd = -1;
return -1;
}
ta->vma_ios_use_direct = (direct == 1);
}
return pagemap_render_iovec(&rsti(t)->vma_io, ta);
ret = pagemap_render_iovec(&rsti(t)->vma_io, ta);
if (ret) {
close_image(pages);
ta->vma_ios_fd = -1;
}
return ret;
}
int prepare_vmas(struct pstree_item *t, struct task_restore_args *ta)

25
criu/page-pipe.c
View file

@ -68,9 +68,20 @@ static inline int ppb_resize_pipe(struct page_pipe_buf *ppb)
return 0;
}
static inline unsigned int ppb_type(unsigned int ppb_flags)
{
unsigned int type = 0;
if (ppb_flags & PPB_LAZY && opts.lazy_pages)
type |= PPB_LAZY;
if (ppb_flags & PPB_FORCE_RAW)
type |= PPB_FORCE_RAW;
return type;
}
static struct page_pipe_buf *pp_prev_ppb(struct page_pipe *pp, unsigned int ppb_flags)
{
int type = 0;
/* don't allow to reuse a pipe in the PP_CHUNK_MODE mode */
if (pp->flags & PP_CHUNK_MODE)
@ -79,20 +90,12 @@ static struct page_pipe_buf *pp_prev_ppb(struct page_pipe *pp, unsigned int ppb_
if (list_empty(&pp->bufs))
return NULL;
if (ppb_flags & PPB_LAZY && opts.lazy_pages)
type = 1;
return pp->prev[type];
return pp->prev[ppb_type(ppb_flags)];
}
static void pp_update_prev_ppb(struct page_pipe *pp, struct page_pipe_buf *ppb, unsigned int ppb_flags)
{
int type = 0;
if (ppb_flags & PPB_LAZY && opts.lazy_pages)
type = 1;
pp->prev[type] = ppb;
pp->prev[ppb_type(ppb_flags)] = ppb;
}
static struct page_pipe_buf *ppb_alloc(struct page_pipe *pp, unsigned int ppb_flags)

273
criu/page-xfer.c
View file

@ -8,6 +8,7 @@
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <stdlib.h>
#undef LOG_PREFIX
#define LOG_PREFIX "page-xfer: "
@ -264,6 +265,8 @@ static int write_pages_to_server_compressed(struct page_xfer *xfer, int p, unsig
if (page_is_all_zero(buf)) {
compressed_size = 0;
} else if (xfer->force_raw) {
compressed_size = PAGE_SIZE;
} else {
int r = compress_data(buf, PAGE_SIZE, compressed_buf, PAGE_COMPRESSED_SIZE_BOUND, acceleration);
if (r < 0)
@ -540,6 +543,40 @@ static int read_pipe_full(int fd, void *buf, size_t count)
return 0;
}
static bool region_is_all_zero(const char *buf, unsigned int nr_pages)
{
unsigned int i;
for (i = 0; i < nr_pages; i++)
if (!page_is_all_zero(buf + (size_t)i * PAGE_SIZE))
return false;
return true;
}
static bool pending_entry_is_all_raw(const struct page_xfer *xfer)
{
unsigned int region_pages = xfer->pending_pe.region_pages;
size_t i;
for (i = 0; i < xfer->pending_pe.total_blocks; i++) {
size_t block_bytes = PAGE_SIZE;
if (region_pages) {
unsigned long pages_done = i * (unsigned long)region_pages;
unsigned long pages_left = xfer->pending_pe.nr_pages - pages_done;
unsigned int block_pages = pages_left < region_pages ? (unsigned int)pages_left : region_pages;
block_bytes = (size_t)block_pages * PAGE_SIZE;
}
if (xfer->pending_pe.compressed_size[i] != block_bytes)
return false;
}
return true;
}
static int write_pages_loc_compressed(struct page_xfer *xfer, int p, unsigned long len)
{
unsigned long off = 0;
@ -578,6 +615,13 @@ static int write_pages_loc_compressed(struct page_xfer *xfer, int p, unsigned lo
xfer->pending_pe.compressed_size[idx] = 0;
continue;
}
if (xfer->force_raw) {
xfer->pending_pe.compressed_size[idx] = PAGE_SIZE;
xfer->pending_pe.total_compressed_size += PAGE_SIZE;
if (write_compressed_payload(xfer, buf, PAGE_SIZE, true))
return -1;
continue;
}
cs = compress_data(buf, PAGE_SIZE, compressed_buf, PAGE_COMPRESSED_SIZE_BOUND, acceleration);
if (cs < 0)
@ -608,8 +652,8 @@ static int write_pages_loc_compressed(struct page_xfer *xfer, int p, unsigned lo
int rc = -1;
src_buf = xmalloc(region_bytes_max);
dst_buf = xmalloc(cap);
if (!src_buf || !dst_buf)
dst_buf = xfer->force_raw ? NULL : xmalloc(cap);
if (!src_buf || (!xfer->force_raw && !dst_buf))
goto region_out;
pages_done = (unsigned long)xfer->pending_pe.n_compressed * region_pages;
@ -618,6 +662,7 @@ static int write_pages_loc_compressed(struct page_xfer *xfer, int p, unsigned lo
unsigned long pages_left = xfer->pending_pe.nr_pages - pages_done;
unsigned int this_region = pages_left < region_pages ? pages_left : region_pages;
size_t region_bytes = (size_t)this_region * PAGE_SIZE;
const char *payload = dst_buf;
int cs;
size_t idx;
@ -633,13 +678,22 @@ static int write_pages_loc_compressed(struct page_xfer *xfer, int p, unsigned lo
idx = xfer->pending_pe.n_compressed++;
cs = compress_region(src_buf, this_region, dst_buf, cap, acceleration);
if (cs < 0)
goto region_out;
if (xfer->force_raw) {
if (region_is_all_zero(src_buf, this_region)) {
cs = 0;
} else {
cs = (int)region_bytes;
payload = src_buf;
}
} else {
cs = compress_region(src_buf, this_region, dst_buf, cap, acceleration);
if (cs < 0)
goto region_out;
}
xfer->pending_pe.compressed_size[idx] = cs;
xfer->pending_pe.total_compressed_size += cs;
if (cs > 0 && write_compressed_payload(xfer, dst_buf, cs, (size_t)cs == region_bytes))
if (cs > 0 && write_compressed_payload(xfer, payload, cs, (size_t)cs == region_bytes))
goto region_out;
}
@ -654,20 +708,28 @@ region_out:
/* When all blocks are compressed, flush the pagemap entry */
if (xfer->pending_pe.n_compressed == xfer->pending_pe.total_blocks) {
PagemapEntry pe = PAGEMAP_ENTRY__INIT;
bool all_raw = pending_entry_is_all_raw(xfer);
pe.vaddr = xfer->pending_pe.vaddr;
pe.nr_pages = xfer->pending_pe.nr_pages;
pe.has_flags = true;
pe.flags = xfer->pending_pe.flags;
pe.has_nr_pages = true;
pe.compressed_size = xfer->pending_pe.compressed_size;
pe.n_compressed_size = xfer->pending_pe.total_blocks;
pe.has_total_compressed_size = true;
pe.total_compressed_size = xfer->pending_pe.total_compressed_size;
/*
* An entry for which every block fell back to raw has exactly the
* same contiguous payload as an ordinary page image. Omitting the
* compression metadata lets restore take the uncompressed fast path.
*/
if (!all_raw) {
pe.compressed_size = xfer->pending_pe.compressed_size;
pe.n_compressed_size = xfer->pending_pe.total_blocks;
pe.has_total_compressed_size = true;
pe.total_compressed_size = xfer->pending_pe.total_compressed_size;
if (region_pages > 0) {
pe.has_region_pages = true;
pe.region_pages = region_pages;
if (region_pages > 0) {
pe.has_region_pages = true;
pe.region_pages = region_pages;
}
}
if (pb_write_one(xfer->pmi, &pe, PB_PAGEMAP) < 0)
@ -876,6 +938,7 @@ out:
xfer->pending_pe.compressed_size = NULL;
xfer->pending_pe.payload_started = false;
xfer->pages_image_offset = 0;
xfer->force_raw = false;
if (compress) {
xfer->write_pagemap = write_pagemap_loc_compressed;
xfer->write_pages = write_pages_loc_compressed;
@ -899,6 +962,7 @@ int open_page_xfer(struct page_xfer *xfer, int fd_type, unsigned long img_id)
xfer->offset = 0;
xfer->transfer_lazy = true;
xfer->force_raw = false;
if (opts.use_page_server)
return open_page_server_xfer(xfer, fd_type, img_id);
@ -1337,6 +1401,7 @@ int page_xfer_predump_pages(int pid, struct page_xfer *xfer, struct page_pipe *p
pr_debug("\t p %p - %p\n", iov.iov_base, iov.iov_base + iov.iov_len);
flags = ppb_xfer_flags(xfer, ppb);
xfer->force_raw = ppb->flags & PPB_FORCE_RAW;
if (xfer->write_pagemap(xfer, &iov, flags))
goto err;
@ -1385,6 +1450,7 @@ int page_xfer_dump_pages(struct page_xfer *xfer, struct page_pipe *pp)
pr_debug("\tp %p - %p\n", iov.iov_base, iov.iov_base + iov.iov_len);
flags = ppb_xfer_flags(xfer, ppb);
xfer->force_raw = ppb->flags & PPB_FORCE_RAW;
if (xfer->write_pagemap(xfer, &iov, flags))
return -1;
@ -1601,6 +1667,7 @@ static int page_server_add_compressed(int sk, struct page_server_iov *pi, u32 fl
unsigned long i;
uint32_t *compressed_size;
uint64_t total_compressed_size = 0;
bool all_raw = true;
bool payload_started = false;
if (validate_page_server_iov(pi, flags))
@ -1649,6 +1716,8 @@ static int page_server_add_compressed(int sk, struct page_server_iov *pi, u32 fl
compressed_size[i] = cs;
total_compressed_size += cs;
if (cs != PAGE_SIZE)
all_raw = false;
if (cs > 0) {
char buf[PAGE_COMPRESSED_SIZE_BOUND];
@ -1681,10 +1750,12 @@ static int page_server_add_compressed(int sk, struct page_server_iov *pi, u32 fl
pe.has_flags = true;
pe.flags = flags;
pe.has_nr_pages = true;
pe.compressed_size = compressed_size;
pe.n_compressed_size = pi->nr_pages;
pe.has_total_compressed_size = true;
pe.total_compressed_size = total_compressed_size;
if (!all_raw) {
pe.compressed_size = compressed_size;
pe.n_compressed_size = pi->nr_pages;
pe.has_total_compressed_size = true;
pe.total_compressed_size = total_compressed_size;
}
if (pb_write_one(lxfer->pmi, &pe, PB_PAGEMAP) < 0) {
xfree(compressed_size);
@ -1969,11 +2040,153 @@ static int page_server_serve(int sk)
return ret;
}
static int fill_page_pipe(struct page_read *pr, struct page_pipe *pp)
/* Match the restore batch cap; individual reads are aligned below. */
#define PAGE_SERVER_DECODE_BUFFER_SIZE (32UL << 20)
static bool page_read_requires_buffered_copy(const struct page_read *pr)
{
int i;
/* splice(2) is not portable for O_DIRECT input descriptors. */
if (pr->use_direct)
return true;
for (i = 0; i < pr->nr_pmes; i++)
if (pagemap_present(pr->pmes[i]) &&
(pr->pmes[i]->n_compressed_size ||
pagemap_payload_aligned(pr->pmes[i])))
return true;
return false;
}
static int splice_page_pipe(struct page_read *pr, struct page_pipe *pp)
{
struct page_pipe_buf *ppb;
int fd = img_raw_fd(pr->pi);
int i;
if (fd < 0) {
pr_err("Failed getting pages image fd\n");
return -1;
}
list_for_each_entry(ppb, &pp->bufs, l) {
for (i = 0; i < ppb->nr_segs; i++) {
size_t left = ppb->iov[i].iov_len;
while (left) {
ssize_t ret = splice(fd, NULL, ppb->p[1], NULL, left, SPLICE_F_MOVE);
if (ret < 0) {
if (errno == EINTR)
continue;
pr_perror("Splice from pages image failed");
return -1;
}
if (!ret) {
pr_err("Pages image ended while filling page pipe\n");
return -1;
}
left -= ret;
}
}
}
return 0;
}
static int decode_page_pipe(struct page_read *pr, struct page_pipe *pp)
{
const size_t buffer_size = max((size_t)PAGE_SIZE, (size_t)PAGE_SERVER_DECODE_BUFFER_SIZE);
const unsigned long buffer_pages = buffer_size / PAGE_SIZE;
struct page_pipe_buf *ppb;
void *buf = NULL;
int i, ret;
ret = posix_memalign(&buf, PAGE_SIZE, buffer_size);
if (ret) {
pr_err("Failed to allocate page-pipe decode buffer: %s\n", strerror(ret));
return -1;
}
pr->reset(pr);
list_for_each_entry(ppb, &pp->bufs, l) {
for (i = 0; i < ppb->nr_segs; i++) {
struct iovec iov = ppb->iov[i];
unsigned long vaddr = encode_pointer(iov.iov_base);
unsigned long nr_pages = iov.iov_len / PAGE_SIZE;
unsigned long pages_done = 0;
if (iov.iov_len % PAGE_SIZE) {
pr_err("Unaligned page-pipe iovec length %zu\n", iov.iov_len);
goto err;
}
while (pages_done < nr_pages) {
unsigned int region_pages;
unsigned long entry_pages;
unsigned long read_pages;
unsigned long page_vaddr = vaddr + pages_done * PAGE_SIZE;
ret = pr->seek_pagemap(pr, page_vaddr);
if (ret <= 0 || !pr->pe || pagemap_in_parent(pr->pe)) {
pr_err("Missing local page at %#lx while filling page pipe\n", page_vaddr);
goto err;
}
entry_pages = pr->pe->nr_pages - ((page_vaddr - pr->pe->vaddr) / PAGE_SIZE);
read_pages = entry_pages;
read_pages = min(read_pages, nr_pages - pages_done);
region_pages = pr->pe->has_region_pages ? pr->pe->region_pages : 0;
if (read_pages > buffer_pages ||
(region_pages && read_pages == buffer_pages &&
entry_pages > read_pages)) {
unsigned long batch_pages = buffer_pages;
/*
* Region sizes are arbitrary page multiples, so a fixed
* 32 MiB buffer is not divisible by all of them. End a
* bounded chunk on a region boundary; otherwise the next
* chunk falls back to synchronous partial decompression.
*/
if (region_pages)
batch_pages -= batch_pages % region_pages;
if (!batch_pages) {
pr_err("Compression region %u exceeds page-server decode buffer\n",
region_pages);
goto err;
}
read_pages = batch_pages;
}
ret = pr->read_pages(pr, page_vaddr, read_pages, buf, PR_ASYNC);
if (ret < 0)
goto err;
if (pr->sync(pr))
goto err;
if (write_fd_full(ppb->p[1], buf, read_pages * PAGE_SIZE)) {
pr_err("Failed writing decoded pages to page pipe\n");
goto err;
}
pages_done += read_pages;
}
}
}
xfree(buf);
return 0;
err:
xfree(buf);
return -1;
}
static int fill_page_pipe(struct page_read *pr, struct page_pipe *pp)
{
unsigned long i;
int ret;
pr->reset(pr);
while (pr->advance(pr)) {
@ -1991,17 +2204,17 @@ static int fill_page_pipe(struct page_read *pr, struct page_pipe *pp)
}
}
list_for_each_entry(ppb, &pp->bufs, l) {
for (i = 0; i < ppb->nr_segs; i++) {
struct iovec iov = ppb->iov[i];
if (splice(img_raw_fd(pr->pi), NULL, ppb->p[1], NULL, iov.iov_len, SPLICE_F_MOVE) !=
iov.iov_len) {
pr_perror("Splice failed");
return -1;
}
}
}
/*
* Ordinary page images retain the original zero-copy send path. A page
* image with compression metadata needs the page reader to synthesize
* zero blocks, copy raw fallbacks, and decode LZ4 blocks.
*/
if (page_read_requires_buffered_copy(pr))
ret = decode_page_pipe(pr, pp);
else
ret = splice_page_pipe(pr, pp);
if (ret)
return ret;
debug_show_page_pipe(pp);
@ -2018,6 +2231,8 @@ static int page_pipe_from_pagemap(struct page_pipe **pp, int pid)
pr_err("Failed to open page read for %d\n", pid);
return -1;
}
/* Building the serving pipe must not modify the source image on failure. */
page_read_disable_dedup(&pr);
while (pr.advance(&pr))
if (pagemap_present(pr.pe))

1987
criu/pagemap.c
View file

File diff suppressed because it is too large Load diff

353
criu/pie/restorer.c
View file

@ -1,5 +1,6 @@
#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
#include <linux/securebits.h>
#include <linux/capability.h>
@ -1680,8 +1681,10 @@ static int process_aio_event(struct task_restore_args *args, aio_context_t aio_c
}
cb = (struct iocb *)(unsigned long)ev->obj;
idx = cb - iocbs;
r = rio_ptrs[idx];
if (args->auto_dedup) {
if (args->auto_dedup && r->storage == VMA_IO_UNCOMPRESSED) {
long fr = sys_fallocate(fd, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
cb->aio_offset, res);
if (fr < 0)
@ -1691,8 +1694,6 @@ static int process_aio_event(struct task_restore_args *args, aio_context_t aio_c
if (res == ev->data)
return 0;
idx = cb - iocbs;
r = rio_ptrs[idx];
advance_vma_io_retry(r, res, &iov_next, &nr_next);
if (!iov_next || nr_next == 0) {
pr_err("AIO retry advance produced no work after %zd bytes\n", res);
@ -1760,6 +1761,148 @@ static int reap_aio_events(struct task_restore_args *args, aio_context_t aio_ctx
return 0;
}
static int restore_vma_preadv_one(struct task_restore_args *args,
struct restore_vma_io *rio,
bool allow_dedup);
static int validate_direct_vma_io(struct restore_vma_io *rio)
{
uint64_t payload_bytes = 0;
uint64_t output_bytes = 0;
uint64_t iov_bytes = 0;
int i;
if (rio->storage != VMA_IO_PACKED_RAW && rio->storage != VMA_IO_ZERO)
return -1;
if (rio->n_compressed_size <= 0 || !rio->compressed_size) {
pr_err("Direct compressed VMA IO has no block metadata\n");
return -1;
}
if (rio->region_pages && !rio->block_pages) {
pr_err("Direct compressed region VMA IO has no page counts\n");
return -1;
}
for (i = 0; i < rio->n_compressed_size; i++) {
unsigned int block_pages = rio->region_pages ?
rio->block_pages[i] : 1;
uint64_t block_bytes;
uint32_t compressed_size = rio->compressed_size[i];
if (!block_pages ||
(rio->region_pages && block_pages > rio->region_pages)) {
pr_err("Invalid direct VMA IO block page count %u\n",
block_pages);
return -1;
}
block_bytes = (uint64_t)block_pages * PAGE_SIZE;
if (rio->storage == VMA_IO_PACKED_RAW &&
compressed_size != block_bytes) {
pr_err("Packed-raw VMA IO block %d has size %u, expected %llu\n",
i, compressed_size,
(unsigned long long)block_bytes);
return -1;
}
if (rio->storage == VMA_IO_ZERO && compressed_size) {
pr_err("Zero VMA IO block %d has payload size %u\n", i,
compressed_size);
return -1;
}
if (payload_bytes > UINT64_MAX - compressed_size ||
output_bytes > UINT64_MAX - block_bytes) {
pr_err("Direct VMA IO size overflows\n");
return -1;
}
payload_bytes += compressed_size;
output_bytes += block_bytes;
}
for (i = 0; i < rio->nr_iovs; i++) {
if (iov_bytes > UINT64_MAX - rio->iovs[i].iov_len) {
pr_err("Direct VMA IO destination size overflows\n");
return -1;
}
iov_bytes += rio->iovs[i].iov_len;
}
if (payload_bytes != rio->total_compressed_size ||
output_bytes != iov_bytes || rio->n_pages <= 0 ||
output_bytes != (uint64_t)rio->n_pages * PAGE_SIZE) {
pr_err("Inconsistent direct VMA IO sizes: payload=%llu metadata=%llu output=%llu iov=%llu\n",
(unsigned long long)payload_bytes,
(unsigned long long)rio->total_compressed_size,
(unsigned long long)output_bytes,
(unsigned long long)iov_bytes);
return -1;
}
return 0;
}
static int restore_vma_preadv_mixed(struct task_restore_args *args)
{
struct restore_vma_io *rio = args->vma_ios;
unsigned int i;
int ret = -1;
/*
* Actual LZ4 blocks are restored into premapped VMAs before PIE. A
* delayed VMA can therefore contain only ordinary uncompressed pages,
* packed raw-fallback blocks, or zero blocks, all of which PIE can
* restore with native syscalls.
*/
for (i = 0; i < args->vma_ios_n; i++) {
if (rio->storage == VMA_IO_UNCOMPRESSED) {
if (args->vma_ios_fd == -1) {
pr_err("No pages image fd for uncompressed VMA IO entry\n");
goto out;
}
if (restore_vma_preadv_one(args, rio, true))
goto out;
goto next;
}
if (rio->storage == VMA_IO_PACKED_RAW) {
if (args->vma_ios_fd == -1) {
pr_err("No pages image fd for packed-raw VMA IO entry\n");
goto out;
}
if (validate_direct_vma_io(rio) ||
restore_vma_preadv_one(args, rio, false))
goto out;
goto next;
}
if (rio->storage == VMA_IO_ZERO) {
int j;
if (validate_direct_vma_io(rio))
goto out;
for (j = 0; j < rio->nr_iovs; j++)
memset(rio->iovs[j].iov_base, 0,
rio->iovs[j].iov_len);
goto next;
}
if (rio->storage != VMA_IO_ENCODED) {
pr_err("Unknown VMA IO storage kind %d\n", rio->storage);
goto out;
}
pr_err("Delayed VMA IO unexpectedly contains an LZ4 block\n");
goto out;
next:
rio = (struct restore_vma_io *)((char *)rio + RIO_SIZE(rio->nr_iovs));
}
ret = 0;
out:
if (args->vma_ios_fd != -1) {
sys_close(args->vma_ios_fd);
args->vma_ios_fd = -1;
}
return ret;
}
/*
* Call preadv() but limit size of the read. Zero `max_to_read` skips the limit.
*/
@ -1794,76 +1937,61 @@ static ssize_t preadv_limited(int fd, struct iovec *iovs, int nr, off_t offs, si
return ret;
}
/*
* Restore private VMA page contents via synchronous preadv().
*
* Fallback engine selected when the pages image fd does not support
* O_DIRECT (probe_pages_o_direct() returned 0). Reads each page-io
* iovec batch with sys_preadv() and honors --auto-dedup by punching
* holes in the consumed range. Closes args->vma_ios_fd on exit.
*/
static int restore_vma_preadv(struct task_restore_args *args)
static int restore_vma_preadv_one(struct task_restore_args *args,
struct restore_vma_io *rio,
bool allow_dedup)
{
struct restore_vma_io *rio;
unsigned int i;
struct iovec *iovs = rio->iovs;
int nr = rio->nr_iovs;
ssize_t r;
rio = args->vma_ios;
for (i = 0; i < args->vma_ios_n; i++) {
struct iovec *iovs = rio->iovs;
int nr = rio->nr_iovs;
ssize_t r;
while (nr) {
pr_debug("Preadv %lx:%d... (%d iovs)\n", (unsigned long)iovs->iov_base, (int)iovs->iov_len, nr);
/*
* If we're requested to punch holes in the file after reading we do
* it to save memory. Limit the reads then to an arbitrary block size.
*/
r = preadv_limited(args->vma_ios_fd, iovs, nr, rio->off,
args->auto_dedup ? AUTO_DEDUP_OVERHEAD_BYTES : 0);
if (r < 0) {
pr_err("Can't read pages data (%d)\n", (int)r);
return -1;
}
if (r == 0) {
pr_err("Unexpected EOF reading pages data at offset %ld (%d iovs remaining)\n",
(long)rio->off, nr);
return -1;
}
pr_debug("`- returned %ld\n", (long)r);
/* If the file is open for writing, then it means we should punch holes
* in it. */
if (args->auto_dedup) {
int fr = sys_fallocate(args->vma_ios_fd, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
rio->off, r);
if (fr < 0) {
pr_debug("Failed to punch holes with fallocate: %d\n", fr);
}
}
rio->off += r;
/* Advance the iovecs */
do {
if (iovs->iov_len <= r) {
pr_debug(" `- skip pagemap\n");
r -= iovs->iov_len;
iovs++;
nr--;
continue;
}
iovs->iov_base += r;
iovs->iov_len -= r;
break;
} while (nr > 0);
while (nr) {
pr_debug("Preadv %lx:%d... (%d iovs)\n", (unsigned long)iovs->iov_base, (int)iovs->iov_len, nr);
/*
* If we're requested to punch holes in the file after reading we do
* it to save memory. Limit the reads then to an arbitrary block size.
*/
r = preadv_limited(args->vma_ios_fd, iovs, nr, rio->off,
args->auto_dedup && allow_dedup ?
AUTO_DEDUP_OVERHEAD_BYTES : 0);
if (r < 0) {
pr_err("Can't read pages data (%d)\n", (int)r);
return -1;
}
rio = (struct restore_vma_io *)((char *)rio + RIO_SIZE(rio->nr_iovs));
}
if (r == 0) {
pr_err("Unexpected EOF reading pages data at offset %ld (%d iovs remaining)\n",
(long)rio->off, nr);
return -1;
}
if (args->vma_ios_fd != -1)
sys_close(args->vma_ios_fd);
pr_debug("`- returned %ld\n", (long)r);
/*
* If the file is open for writing, then it means we should
* punch holes in it.
*/
if (args->auto_dedup && allow_dedup) {
int fr = sys_fallocate(args->vma_ios_fd, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
rio->off, r);
if (fr < 0)
pr_debug("Failed to punch holes with fallocate: %d\n", fr);
}
rio->off += r;
/* Advance the iovecs */
do {
if (iovs->iov_len <= r) {
pr_debug(" `- skip pagemap\n");
r -= iovs->iov_len;
iovs++;
nr--;
continue;
}
iovs->iov_base += r;
iovs->iov_len -= r;
break;
} while (nr > 0);
}
return 0;
}
@ -1871,10 +1999,10 @@ static int restore_vma_preadv(struct task_restore_args *args)
/*
* Restore private VMA page contents via Linux native AIO.
*
* Submits io_submit() requests against the O_DIRECT-prepared pages image
* fd in a bounded AIO_BATCH window. Handles MAX_RW_COUNT (0x7FFFF000)
* short reads by masking to page alignment and resubmitting. Closes
* args->vma_ios_fd on exit.
* Submits aligned uncompressed and packed-raw requests against the
* O_DIRECT-prepared pages image in a bounded AIO_BATCH window. Zero records
* are cleared without I/O. Handles MAX_RW_COUNT (0x7FFFF000) short reads by
* masking to page alignment and resubmitting. Closes args->vma_ios_fd on exit.
*/
static int restore_vma_aio(struct task_restore_args *args)
{
@ -1888,40 +2016,75 @@ static int restore_vma_aio(struct task_restore_args *args)
struct restore_vma_io *rio;
struct io_event *events;
unsigned long alloc_sz;
unsigned int submitted = 0, completed = 0;
const unsigned long event_sz =
AIO_BATCH * sizeof(struct io_event);
const unsigned long request_sz = sizeof(struct iocb) +
sizeof(struct iocb *) + sizeof(struct restore_vma_io *);
unsigned int submitted = 0, completed = 0, read_count = 0;
unsigned int i;
int ret = -1;
if (__builtin_mul_overflow((unsigned long)n, request_sz, &alloc_sz) ||
__builtin_add_overflow(alloc_sz, event_sz, &alloc_sz)) {
pr_err("AIO restore metadata size overflows for %u requests\n", n);
sys_close(fd);
args->vma_ios_fd = -1;
return -1;
}
aio_ret = sys_io_setup(AIO_BATCH, &aio_ctx);
if (aio_ret < 0) {
pr_err("io_setup(%d) failed: %ld\n", AIO_BATCH, aio_ret);
sys_close(fd);
args->vma_ios_fd = -1;
return -1;
}
alloc_sz = n * sizeof(struct iocb) + n * sizeof(struct iocb *) +
n * sizeof(struct restore_vma_io *) +
AIO_BATCH * sizeof(struct io_event);
iocbs = (void *)sys_mmap(NULL, alloc_sz, PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
if (IS_ERR(iocbs)) {
pr_err("Can't mmap AIO buffers: %ld\n", PTR_ERR(iocbs));
sys_io_destroy(aio_ctx);
sys_close(fd);
args->vma_ios_fd = -1;
return -1;
}
iocbps = (struct iocb **)((char *)iocbs + n * sizeof(struct iocb));
rio_ptrs = (struct restore_vma_io **)((char *)iocbps + n * sizeof(struct iocb *));
events = (struct io_event *)((char *)rio_ptrs + n * sizeof(struct restore_vma_io *));
/* Build all iocbs from vma_ios */
/* Build compact iocbs for entries which actually read from the image. */
rio = args->vma_ios;
for (i = 0; i < n; i++) {
struct iocb *cb = &iocbs[i];
struct iocb *cb;
size_t expected = 0;
int j;
for (j = 0; j < rio->nr_iovs; j++)
expected += rio->iovs[j].iov_len;
if (rio->storage == VMA_IO_ZERO) {
if (validate_direct_vma_io(rio))
goto out;
for (j = 0; j < rio->nr_iovs; j++)
memset(rio->iovs[j].iov_base, 0,
rio->iovs[j].iov_len);
goto next;
}
if (rio->storage == VMA_IO_PACKED_RAW) {
if (validate_direct_vma_io(rio))
goto out;
} else if (rio->storage != VMA_IO_UNCOMPRESSED) {
pr_err("AIO restore received storage kind %d\n", rio->storage);
goto out;
}
for (j = 0; j < rio->nr_iovs; j++) {
if (expected > SIZE_MAX - rio->iovs[j].iov_len) {
pr_err("AIO restore request size overflows\n");
goto out;
}
expected += rio->iovs[j].iov_len;
}
cb = &iocbs[read_count];
memset(cb, 0, sizeof(*cb));
cb->aio_fildes = fd;
cb->aio_lio_opcode = IOCB_CMD_PREADV;
@ -1930,18 +2093,21 @@ static int restore_vma_aio(struct task_restore_args *args)
cb->aio_offset = rio->off;
/* io_getevents() returns this as event.data for short-read checks. */
cb->aio_data = expected;
iocbps[i] = cb;
rio_ptrs[i] = rio;
iocbps[read_count] = cb;
rio_ptrs[read_count] = rio;
read_count++;
next:
rio = (struct restore_vma_io *)((char *)rio + RIO_SIZE(rio->nr_iovs));
}
/* Submit and reap in batches */
while (submitted < n || completed < n) {
if (submit_aio_batch(aio_ctx, iocbps, n, &submitted, completed) < 0)
while (submitted < read_count || completed < read_count) {
if (submit_aio_batch(aio_ctx, iocbps, read_count,
&submitted, completed) < 0)
goto out;
if (completed >= n)
if (completed >= read_count)
continue;
/*
@ -2257,13 +2423,20 @@ __visible long __export_restore_task(struct task_restore_args *args)
* Engine is selected from args->vma_ios_use_direct, populated by
* probe_pages_o_direct() at restore-args build time:
* true -> restore_vma_aio() (io_submit, O_DIRECT, batched)
* false -> restore_vma_preadv() (sys_preadv, buffered, sequential)
* Both helpers consume args->vma_ios and close args->vma_ios_fd on exit.
* false -> restore_vma_preadv_mixed() (buffered, sequential I/O)
* The selected reader consumes args->vma_ios and closes
* args->vma_ios_fd on exit. Compressed images can use AIO when every
* delayed file-backed range is raw and page-aligned; encoded ranges are
* premapped and decoded before PIE.
*/
if (args->vma_ios_n > 0 && args->vma_ios_fd != -1) {
int rc = args->vma_ios_use_direct
? restore_vma_aio(args)
: restore_vma_preadv(args);
int rc;
if (args->vma_ios_use_direct)
pr_debug("Restoring delayed VMA I/O with native AIO\n");
rc = args->vma_ios_use_direct ?
restore_vma_aio(args) :
restore_vma_preadv_mixed(args);
if (rc < 0)
goto core_restore_end;
}