ceph/src/zstd/tests/regression/method.c

   1 /*
   2  * Copyright (c) 2016-present, Yann Collet, Facebook, Inc.
   3  * All rights reserved.
   4  *
   5  * This source code is licensed under both the BSD-style license (found in the
   6  * LICENSE file in the root directory of this source tree) and the GPLv2 (found
   7  * in the COPYING file in the root directory of this source tree).
   8  * You may select, at your option, one of the above-listed licenses.
   9  */
  10
  11 #include "method.h"
  12
  13 #include <stdio.h>
  14 #include <stdlib.h>
  15
  16 #define ZSTD_STATIC_LINKING_ONLY
  17 #include <zstd.h>
  18
  19 #define MIN(x, y) ((x) < (y) ? (x) : (y))
  20
  21 static char const* g_zstdcli = NULL;
  22
  23 void method_set_zstdcli(char const* zstdcli) {
  24     g_zstdcli = zstdcli;
  25 }
  26
  27 /**
  28  * Macro to get a pointer of type, given ptr, which is a member variable with
  29  * the given name, member.
  30  *
  31  *     method_state_t* base = ...;
  32  *     buffer_state_t* state = container_of(base, buffer_state_t, base);
  33  */
  34 #define container_of(ptr, type, member) \
  35     ((type*)(ptr == NULL ? NULL : (char*)(ptr)-offsetof(type, member)))
  36
  37 /** State to reuse the same buffers between compression calls. */
  38 typedef struct {
  39     method_state_t base;
  40     data_buffers_t inputs; /**< The input buffer for each file. */
  41     data_buffer_t dictionary; /**< The dictionary. */
  42     data_buffer_t compressed; /**< The compressed data buffer. */
  43     data_buffer_t decompressed; /**< The decompressed data buffer. */
  44 } buffer_state_t;
  45
  46 static size_t buffers_max_size(data_buffers_t buffers) {
  47     size_t max = 0;
  48     for (size_t i = 0; i < buffers.size; ++i) {
  49         if (buffers.buffers[i].size > max)
  50             max = buffers.buffers[i].size;
  51     }
  52     return max;
  53 }
  54
  55 static method_state_t* buffer_state_create(data_t const* data) {
  56     buffer_state_t* state = (buffer_state_t*)calloc(1, sizeof(buffer_state_t));
  57     if (state == NULL)
  58         return NULL;
  59     state->base.data = data;
  60     state->inputs = data_buffers_get(data);
  61     state->dictionary = data_buffer_get_dict(data);
  62     size_t const max_size = buffers_max_size(state->inputs);
  63     state->compressed = data_buffer_create(ZSTD_compressBound(max_size));
  64     state->decompressed = data_buffer_create(max_size);
  65     return &state->base;
  66 }
  67
  68 static void buffer_state_destroy(method_state_t* base) {
  69     if (base == NULL)
  70         return;
  71     buffer_state_t* state = container_of(base, buffer_state_t, base);
  72     free(state);
  73 }
  74
  75 static int buffer_state_bad(
  76     buffer_state_t const* state,
  77     config_t const* config) {
  78     if (state == NULL) {
  79         fprintf(stderr, "buffer_state_t is NULL\n");
  80         return 1;
  81     }
  82     if (state->inputs.size == 0 || state->compressed.data == NULL ||
  83         state->decompressed.data == NULL) {
  84         fprintf(stderr, "buffer state allocation failure\n");
  85         return 1;
  86     }
  87     if (config->use_dictionary && state->dictionary.data == NULL) {
  88         fprintf(stderr, "dictionary loading failed\n");
  89         return 1;
  90     }
  91     return 0;
  92 }
  93
  94 static result_t simple_compress(method_state_t* base, config_t const* config) {
  95     buffer_state_t* state = container_of(base, buffer_state_t, base);
  96
  97     if (buffer_state_bad(state, config))
  98         return result_error(result_error_system_error);
  99
 100     /* Keep the tests short by skipping directories, since behavior shouldn't
 101      * change.
 102      */
 103     if (base->data->type != data_type_file)
 104         return result_error(result_error_skip);
 105
 106     if (config->use_dictionary || config->no_pledged_src_size)
 107         return result_error(result_error_skip);
 108
 109     /* If the config doesn't specify a level, skip. */
 110     int const level = config_get_level(config);
 111     if (level == CONFIG_NO_LEVEL)
 112         return result_error(result_error_skip);
 113
 114     data_buffer_t const input = state->inputs.buffers[0];
 115
 116     /* Compress, decompress, and check the result. */
 117     state->compressed.size = ZSTD_compress(
 118         state->compressed.data,
 119         state->compressed.capacity,
 120         input.data,
 121         input.size,
 122         level);
 123     if (ZSTD_isError(state->compressed.size))
 124         return result_error(result_error_compression_error);
 125
 126     state->decompressed.size = ZSTD_decompress(
 127         state->decompressed.data,
 128         state->decompressed.capacity,
 129         state->compressed.data,
 130         state->compressed.size);
 131     if (ZSTD_isError(state->decompressed.size))
 132         return result_error(result_error_decompression_error);
 133     if (data_buffer_compare(input, state->decompressed))
 134         return result_error(result_error_round_trip_error);
 135
 136     result_data_t data;
 137     data.total_size = state->compressed.size;
 138     return result_data(data);
 139 }
 140
 141 static result_t compress_cctx_compress(
 142     method_state_t* base,
 143     config_t const* config) {
 144     buffer_state_t* state = container_of(base, buffer_state_t, base);
 145
 146     if (buffer_state_bad(state, config))
 147         return result_error(result_error_system_error);
 148
 149     if (config->no_pledged_src_size)
 150         return result_error(result_error_skip);
 151
 152     if (base->data->type != data_type_dir)
 153         return result_error(result_error_skip);
 154
 155     int const level = config_get_level(config);
 156
 157     ZSTD_CCtx* cctx = ZSTD_createCCtx();
 158     ZSTD_DCtx* dctx = ZSTD_createDCtx();
 159     if (cctx == NULL || dctx == NULL) {
 160         fprintf(stderr, "context creation failed\n");
 161         return result_error(result_error_system_error);
 162     }
 163
 164     result_t result;
 165     result_data_t data = {.total_size = 0};
 166     for (size_t i = 0; i < state->inputs.size; ++i) {
 167         data_buffer_t const input = state->inputs.buffers[i];
 168         ZSTD_parameters const params =
 169             config_get_zstd_params(config, input.size, state->dictionary.size);
 170
 171         if (level == CONFIG_NO_LEVEL)
 172             state->compressed.size = ZSTD_compress_advanced(
 173                 cctx,
 174                 state->compressed.data,
 175                 state->compressed.capacity,
 176                 input.data,
 177                 input.size,
 178                 config->use_dictionary ? state->dictionary.data : NULL,
 179                 config->use_dictionary ? state->dictionary.size : 0,
 180                 params);
 181         else if (config->use_dictionary)
 182             state->compressed.size = ZSTD_compress_usingDict(
 183                 cctx,
 184                 state->compressed.data,
 185                 state->compressed.capacity,
 186                 input.data,
 187                 input.size,
 188                 state->dictionary.data,
 189                 state->dictionary.size,
 190                 level);
 191         else
 192             state->compressed.size = ZSTD_compressCCtx(
 193                 cctx,
 194                 state->compressed.data,
 195                 state->compressed.capacity,
 196                 input.data,
 197                 input.size,
 198                 level);
 199
 200         if (ZSTD_isError(state->compressed.size)) {
 201             result = result_error(result_error_compression_error);
 202             goto out;
 203         }
 204
 205         if (config->use_dictionary)
 206             state->decompressed.size = ZSTD_decompress_usingDict(
 207                 dctx,
 208                 state->decompressed.data,
 209                 state->decompressed.capacity,
 210                 state->compressed.data,
 211                 state->compressed.size,
 212                 state->dictionary.data,
 213                 state->dictionary.size);
 214         else
 215             state->decompressed.size = ZSTD_decompressDCtx(
 216                 dctx,
 217                 state->decompressed.data,
 218                 state->decompressed.capacity,
 219                 state->compressed.data,
 220                 state->compressed.size);
 221         if (ZSTD_isError(state->decompressed.size)) {
 222             result = result_error(result_error_decompression_error);
 223             goto out;
 224         }
 225         if (data_buffer_compare(input, state->decompressed)) {
 226             result = result_error(result_error_round_trip_error);
 227             goto out;
 228         }
 229
 230         data.total_size += state->compressed.size;
 231     }
 232
 233     result = result_data(data);
 234 out:
 235     ZSTD_freeCCtx(cctx);
 236     ZSTD_freeDCtx(dctx);
 237     return result;
 238 }
 239
 240 /** Generic state creation function. */
 241 static method_state_t* method_state_create(data_t const* data) {
 242     method_state_t* state = (method_state_t*)malloc(sizeof(method_state_t));
 243     if (state == NULL)
 244         return NULL;
 245     state->data = data;
 246     return state;
 247 }
 248
 249 static void method_state_destroy(method_state_t* state) {
 250     free(state);
 251 }
 252
 253 static result_t cli_compress(method_state_t* state, config_t const* config) {
 254     if (config->cli_args == NULL)
 255         return result_error(result_error_skip);
 256
 257     /* We don't support no pledged source size with directories. Too slow. */
 258     if (state->data->type == data_type_dir && config->no_pledged_src_size)
 259         return result_error(result_error_skip);
 260
 261     if (g_zstdcli == NULL)
 262         return result_error(result_error_system_error);
 263
 264     /* '<zstd>' -cqr <args> [-D '<dict>'] '<file/dir>' */
 265     char cmd[1024];
 266     size_t const cmd_size = snprintf(
 267         cmd,
 268         sizeof(cmd),
 269         "'%s' -cqr %s %s%s%s %s '%s'",
 270         g_zstdcli,
 271         config->cli_args,
 272         config->use_dictionary ? "-D '" : "",
 273         config->use_dictionary ? state->data->dict.path : "",
 274         config->use_dictionary ? "'" : "",
 275         config->no_pledged_src_size ? "<" : "",
 276         state->data->data.path);
 277     if (cmd_size >= sizeof(cmd)) {
 278         fprintf(stderr, "command too large: %s\n", cmd);
 279         return result_error(result_error_system_error);
 280     }
 281     FILE* zstd = popen(cmd, "r");
 282     if (zstd == NULL) {
 283         fprintf(stderr, "failed to popen command: %s\n", cmd);
 284         return result_error(result_error_system_error);
 285     }
 286
 287     char out[4096];
 288     size_t total_size = 0;
 289     while (1) {
 290         size_t const size = fread(out, 1, sizeof(out), zstd);
 291         total_size += size;
 292         if (size != sizeof(out))
 293             break;
 294     }
 295     if (ferror(zstd) || pclose(zstd) != 0) {
 296         fprintf(stderr, "zstd failed with command: %s\n", cmd);
 297         return result_error(result_error_compression_error);
 298     }
 299
 300     result_data_t const data = {.total_size = total_size};
 301     return result_data(data);
 302 }
 303
 304 static int advanced_config(
 305     ZSTD_CCtx* cctx,
 306     buffer_state_t* state,
 307     config_t const* config) {
 308     ZSTD_CCtx_reset(cctx, ZSTD_reset_session_and_parameters);
 309     for (size_t p = 0; p < config->param_values.size; ++p) {
 310         param_value_t const pv = config->param_values.data[p];
 311         if (ZSTD_isError(ZSTD_CCtx_setParameter(cctx, pv.param, pv.value))) {
 312             return 1;
 313         }
 314     }
 315     if (config->use_dictionary) {
 316         if (ZSTD_isError(ZSTD_CCtx_loadDictionary(
 317                 cctx, state->dictionary.data, state->dictionary.size))) {
 318             return 1;
 319         }
 320     }
 321     return 0;
 322 }
 323
 324 static result_t advanced_one_pass_compress_output_adjustment(
 325     method_state_t* base,
 326     config_t const* config,
 327     size_t const subtract) {
 328     buffer_state_t* state = container_of(base, buffer_state_t, base);
 329
 330     if (buffer_state_bad(state, config))
 331         return result_error(result_error_system_error);
 332
 333     ZSTD_CCtx* cctx = ZSTD_createCCtx();
 334     result_t result;
 335
 336     if (!cctx || advanced_config(cctx, state, config)) {
 337         result = result_error(result_error_compression_error);
 338         goto out;
 339     }
 340
 341     result_data_t data = {.total_size = 0};
 342     for (size_t i = 0; i < state->inputs.size; ++i) {
 343         data_buffer_t const input = state->inputs.buffers[i];
 344
 345         if (!config->no_pledged_src_size) {
 346             if (ZSTD_isError(ZSTD_CCtx_setPledgedSrcSize(cctx, input.size))) {
 347                 result = result_error(result_error_compression_error);
 348                 goto out;
 349             }
 350         }
 351         size_t const size = ZSTD_compress2(
 352             cctx,
 353             state->compressed.data,
 354             ZSTD_compressBound(input.size) - subtract,
 355             input.data,
 356             input.size);
 357         if (ZSTD_isError(size)) {
 358             result = result_error(result_error_compression_error);
 359             goto out;
 360         }
 361         data.total_size += size;
 362     }
 363
 364     result = result_data(data);
 365 out:
 366     ZSTD_freeCCtx(cctx);
 367     return result;
 368 }
 369
 370 static result_t advanced_one_pass_compress(
 371     method_state_t* base,
 372     config_t const* config) {
 373   return advanced_one_pass_compress_output_adjustment(base, config, 0);
 374 }
 375
 376 static result_t advanced_one_pass_compress_small_output(
 377     method_state_t* base,
 378     config_t const* config) {
 379   return advanced_one_pass_compress_output_adjustment(base, config, 1);
 380 }
 381
 382 static result_t advanced_streaming_compress(
 383     method_state_t* base,
 384     config_t const* config) {
 385     buffer_state_t* state = container_of(base, buffer_state_t, base);
 386
 387     if (buffer_state_bad(state, config))
 388         return result_error(result_error_system_error);
 389
 390     ZSTD_CCtx* cctx = ZSTD_createCCtx();
 391     result_t result;
 392
 393     if (!cctx || advanced_config(cctx, state, config)) {
 394         result = result_error(result_error_compression_error);
 395         goto out;
 396     }
 397
 398     result_data_t data = {.total_size = 0};
 399     for (size_t i = 0; i < state->inputs.size; ++i) {
 400         data_buffer_t input = state->inputs.buffers[i];
 401
 402         if (!config->no_pledged_src_size) {
 403             if (ZSTD_isError(ZSTD_CCtx_setPledgedSrcSize(cctx, input.size))) {
 404                 result = result_error(result_error_compression_error);
 405                 goto out;
 406             }
 407         }
 408
 409         while (input.size > 0) {
 410             ZSTD_inBuffer in = {input.data, MIN(input.size, 4096)};
 411             input.data += in.size;
 412             input.size -= in.size;
 413             ZSTD_EndDirective const op =
 414                 input.size > 0 ? ZSTD_e_continue : ZSTD_e_end;
 415             size_t ret = 0;
 416             while (in.pos < in.size || (op == ZSTD_e_end && ret != 0)) {
 417                 ZSTD_outBuffer out = {state->compressed.data,
 418                                       MIN(state->compressed.capacity, 1024)};
 419                 ret = ZSTD_compressStream2(cctx, &out, &in, op);
 420                 if (ZSTD_isError(ret)) {
 421                     result = result_error(result_error_compression_error);
 422                     goto out;
 423                 }
 424                 data.total_size += out.pos;
 425             }
 426         }
 427     }
 428
 429     result = result_data(data);
 430 out:
 431     ZSTD_freeCCtx(cctx);
 432     return result;
 433 }
 434
 435 static int init_cstream(
 436     buffer_state_t* state,
 437     ZSTD_CStream* zcs,
 438     config_t const* config,
 439     int const advanced,
 440     ZSTD_CDict** cdict)
 441 {
 442     size_t zret;
 443     if (advanced) {
 444         ZSTD_parameters const params = config_get_zstd_params(config, 0, 0);
 445         ZSTD_CDict* dict = NULL;
 446         if (cdict) {
 447             *cdict = ZSTD_createCDict_advanced(
 448                 state->dictionary.data,
 449                 state->dictionary.size,
 450                 ZSTD_dlm_byRef,
 451                 ZSTD_dct_auto,
 452                 params.cParams,
 453                 ZSTD_defaultCMem);
 454             if (!*cdict) {
 455                 return 1;
 456             }
 457             zret = ZSTD_initCStream_usingCDict_advanced(
 458                 zcs, *cdict, params.fParams, ZSTD_CONTENTSIZE_UNKNOWN);
 459         } else {
 460             zret = ZSTD_initCStream_advanced(
 461                 zcs,
 462                 state->dictionary.data,
 463                 state->dictionary.size,
 464                 params,
 465                 ZSTD_CONTENTSIZE_UNKNOWN);
 466         }
 467     } else {
 468         int const level = config_get_level(config);
 469         if (cdict) {
 470             *cdict = ZSTD_createCDict(
 471                 state->dictionary.data,
 472                 state->dictionary.size,
 473                 level);
 474             if (!*cdict) {
 475                 return 1;
 476             }
 477             zret = ZSTD_initCStream_usingCDict(zcs, *cdict);
 478         } else if (config->use_dictionary) {
 479             zret = ZSTD_initCStream_usingDict(
 480                 zcs, state->dictionary.data, state->dictionary.size, level);
 481         } else {
 482             zret = ZSTD_initCStream(zcs, level);
 483         }
 484     }
 485     if (ZSTD_isError(zret)) {
 486         return 1;
 487     }
 488     return 0;
 489 }
 490
 491 static result_t old_streaming_compress_internal(
 492     method_state_t* base,
 493     config_t const* config,
 494     int const advanced,
 495     int const cdict) {
 496   buffer_state_t* state = container_of(base, buffer_state_t, base);
 497
 498   if (buffer_state_bad(state, config))
 499     return result_error(result_error_system_error);
 500
 501
 502   ZSTD_CStream* zcs = ZSTD_createCStream();
 503   ZSTD_CDict* cd = NULL;
 504   result_t result;
 505   if (zcs == NULL) {
 506     result = result_error(result_error_compression_error);
 507     goto out;
 508   }
 509   if (init_cstream(state, zcs, config, advanced, cdict ? &cd : NULL)) {
 510       result = result_error(result_error_compression_error);
 511       goto out;
 512   }
 513
 514   result_data_t data = {.total_size = 0};
 515   for (size_t i = 0; i < state->inputs.size; ++i) {
 516     data_buffer_t input = state->inputs.buffers[i];
 517     size_t zret = ZSTD_resetCStream(
 518         zcs,
 519         config->no_pledged_src_size ? ZSTD_CONTENTSIZE_UNKNOWN : input.size);
 520     if (ZSTD_isError(zret)) {
 521       result = result_error(result_error_compression_error);
 522       goto out;
 523     }
 524
 525     while (input.size > 0) {
 526       ZSTD_inBuffer in = {input.data, MIN(input.size, 4096)};
 527       input.data += in.size;
 528       input.size -= in.size;
 529       ZSTD_EndDirective const op =
 530           input.size > 0 ? ZSTD_e_continue : ZSTD_e_end;
 531       zret = 0;
 532       while (in.pos < in.size || (op == ZSTD_e_end && zret != 0)) {
 533         ZSTD_outBuffer out = {state->compressed.data,
 534                               MIN(state->compressed.capacity, 1024)};
 535         if (op == ZSTD_e_continue || in.pos < in.size)
 536           zret = ZSTD_compressStream(zcs, &out, &in);
 537         else
 538           zret = ZSTD_endStream(zcs, &out);
 539         if (ZSTD_isError(zret)) {
 540           result = result_error(result_error_compression_error);
 541           goto out;
 542         }
 543         data.total_size += out.pos;
 544       }
 545     }
 546   }
 547
 548   result = result_data(data);
 549 out:
 550     ZSTD_freeCStream(zcs);
 551     ZSTD_freeCDict(cd);
 552     return result;
 553 }
 554
 555 static result_t old_streaming_compress(
 556     method_state_t* base,
 557     config_t const* config)
 558 {
 559     return old_streaming_compress_internal(
 560         base, config, /* advanced */ 0, /* cdict */ 0);
 561 }
 562
 563 static result_t old_streaming_compress_advanced(
 564     method_state_t* base,
 565     config_t const* config)
 566 {
 567     return old_streaming_compress_internal(
 568         base, config, /* advanced */ 1, /* cdict */ 0);
 569 }
 570
 571 static result_t old_streaming_compress_cdict(
 572     method_state_t* base,
 573     config_t const* config)
 574 {
 575     return old_streaming_compress_internal(
 576         base, config, /* advanced */ 0, /* cdict */ 1);
 577 }
 578
 579 static result_t old_streaming_compress_cdict_advanced(
 580     method_state_t* base,
 581     config_t const* config)
 582 {
 583     return old_streaming_compress_internal(
 584         base, config, /* advanced */ 1, /* cdict */ 1);
 585 }
 586
 587 method_t const simple = {
 588     .name = "compress simple",
 589     .create = buffer_state_create,
 590     .compress = simple_compress,
 591     .destroy = buffer_state_destroy,
 592 };
 593
 594 method_t const compress_cctx = {
 595     .name = "compress cctx",
 596     .create = buffer_state_create,
 597     .compress = compress_cctx_compress,
 598     .destroy = buffer_state_destroy,
 599 };
 600
 601 method_t const advanced_one_pass = {
 602     .name = "advanced one pass",
 603     .create = buffer_state_create,
 604     .compress = advanced_one_pass_compress,
 605     .destroy = buffer_state_destroy,
 606 };
 607
 608 method_t const advanced_one_pass_small_out = {
 609     .name = "advanced one pass small out",
 610     .create = buffer_state_create,
 611     .compress = advanced_one_pass_compress,
 612     .destroy = buffer_state_destroy,
 613 };
 614
 615 method_t const advanced_streaming = {
 616     .name = "advanced streaming",
 617     .create = buffer_state_create,
 618     .compress = advanced_streaming_compress,
 619     .destroy = buffer_state_destroy,
 620 };
 621
 622 method_t const old_streaming = {
 623     .name = "old streaming",
 624     .create = buffer_state_create,
 625     .compress = old_streaming_compress,
 626     .destroy = buffer_state_destroy,
 627 };
 628
 629 method_t const old_streaming_advanced = {
 630     .name = "old streaming advanced",
 631     .create = buffer_state_create,
 632     .compress = old_streaming_compress,
 633     .destroy = buffer_state_destroy,
 634 };
 635
 636 method_t const old_streaming_cdict = {
 637     .name = "old streaming cdcit",
 638     .create = buffer_state_create,
 639     .compress = old_streaming_compress,
 640     .destroy = buffer_state_destroy,
 641 };
 642
 643 method_t const old_streaming_advanced_cdict = {
 644     .name = "old streaming advanced cdict",
 645     .create = buffer_state_create,
 646     .compress = old_streaming_compress,
 647     .destroy = buffer_state_destroy,
 648 };
 649
 650 method_t const cli = {
 651     .name = "zstdcli",
 652     .create = method_state_create,
 653     .compress = cli_compress,
 654     .destroy = method_state_destroy,
 655 };
 656
 657 static method_t const* g_methods[] = {
 658     &simple,
 659     &compress_cctx,
 660     &cli,
 661     &advanced_one_pass,
 662     &advanced_one_pass_small_out,
 663     &advanced_streaming,
 664     &old_streaming,
 665     &old_streaming_advanced,
 666     &old_streaming_cdict,
 667     &old_streaming_advanced_cdict,
 668     NULL,
 669 };
 670
 671 method_t const* const* methods = g_methods;