Class: MultiCompress::Inflater

Inherits:

Object

• Object
• MultiCompress::Inflater

Includes:

InflaterDefaults

Defined in:

ext/multi_compress/multi_compress.c

Instance Method Summary

• #close ⇒ Object
• #closed? ⇒ Boolean
• #finish ⇒ Object
• #initialize(*args) ⇒ Object constructor
• #reset ⇒ Object
• #write(chunk) ⇒ Object

Constructor Details

#initialize(*args) ⇒ Object

# File 'ext/multi_compress/multi_compress.c', line 2815

static VALUE inflater_initialize(int argc, VALUE *argv, VALUE self) {
    VALUE opts;
    scan_zero_required_keywords(argc, argv, &opts);

    mc_opts_t parsed_opts;
    mc_parse_opts(opts, &parsed_opts);
    reject_algorithm_keyword(&parsed_opts);

    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);

    VALUE algo_sym = parsed_opts.algo;
    VALUE dict_val = parsed_opts.dictionary;
    limits_config_t limits;
    parse_limits_from_parsed_opts(&parsed_opts, &limits);

    inf->algo = NIL_P(algo_sym) ? ALGO_ZSTD : sym_to_algo(algo_sym);
    inf->closed = 0;
    inf->finished = 0;
    inf->max_output_size = limits.max_output_size;
    inf->total_output = 0;
    inf->total_input = 0;
    inf->max_ratio_enabled = limits.max_ratio_enabled;
    inf->max_ratio = limits.max_ratio;

    dictionary_t *dict = NULL;
    if (!NIL_P(dict_val)) {
        if (inf->algo == ALGO_LZ4) {
            rb_raise(eUnsupportedError, "LZ4 does not support dictionaries");
        }
        dict = opt_dictionary(dict_val);
        dictionary_ivar_set(self, dict_val);
    }

    switch (inf->algo) {
    case ALGO_ZSTD:
        inf->ctx.zstd = ZSTD_createDStream();
        if (!inf->ctx.zstd)
            rb_raise(eMemError, "zstd: failed to create dstream");
        if (dict) {
            ZSTD_DCtx_reset(inf->ctx.zstd, ZSTD_reset_session_only);
            size_t r = ZSTD_DCtx_loadDictionary(inf->ctx.zstd, dict->data, dict->size);
            if (ZSTD_isError(r))
                rb_raise(eError, "zstd dict load: %s", ZSTD_getErrorName(r));
        } else {
            ZSTD_initDStream(inf->ctx.zstd);
        }
        break;
    case ALGO_BROTLI:
        inf->ctx.brotli = BrotliDecoderCreateInstance(NULL, NULL, NULL);
        if (!inf->ctx.brotli)
            rb_raise(eMemError, "brotli: failed to create decoder");
        if (dict) {
            BrotliDecoderAttachDictionary(inf->ctx.brotli, BROTLI_SHARED_DICTIONARY_RAW, dict->size,
                                          dict->data);
        }
        break;
    case ALGO_LZ4:
        inf->lz4_buf.cap = 16 * 1024;
        inf->lz4_buf.buf = ALLOC_N(char, inf->lz4_buf.cap);
        inf->lz4_buf.len = 0;
        inf->lz4_buf.offset = 0;
        break;
    }

    return self;
}

Instance Method Details

#close ⇒ Object

# File 'ext/multi_compress/multi_compress.c', line 3214

static VALUE inflater_close(VALUE self) {
    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);
    if (inf->closed)
        return Qnil;

    switch (inf->algo) {
    case ALGO_ZSTD:
        if (inf->ctx.zstd) {
            ZSTD_freeDStream(inf->ctx.zstd);
            inf->ctx.zstd = NULL;
        }
        break;
    case ALGO_BROTLI:
        if (inf->ctx.brotli) {
            BrotliDecoderDestroyInstance(inf->ctx.brotli);
            inf->ctx.brotli = NULL;
        }
        break;
    case ALGO_LZ4:
        break;
    }
    inf->closed = 1;
    return Qnil;
}

#closed? ⇒ Boolean

Returns:

• (Boolean)

# File 'ext/multi_compress/multi_compress.c', line 3240

static VALUE inflater_closed_p(VALUE self) {
    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);
    return inf->closed ? Qtrue : Qfalse;
}

#finish ⇒ Object

# File 'ext/multi_compress/multi_compress.c', line 3160

static VALUE inflater_finish(VALUE self) {
    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);
    if (inf->closed)
        rb_raise(eStreamError, "stream is closed");
    inf->finished = 1;
    return rb_binary_str_new("", 0);
}

#reset ⇒ Object

# File 'ext/multi_compress/multi_compress.c', line 3169

static VALUE inflater_reset(VALUE self) {
    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);

    VALUE dict_val = dictionary_ivar_get(self);
    dictionary_t *dict = NULL;
    if (!NIL_P(dict_val)) {
        TypedData_Get_Struct(dict_val, dictionary_t, &dictionary_type, dict);
    }

    switch (inf->algo) {
    case ALGO_ZSTD:
        if (inf->ctx.zstd) {
            ZSTD_DCtx_reset(inf->ctx.zstd, ZSTD_reset_session_only);
            if (dict) {
                size_t r = ZSTD_DCtx_loadDictionary(inf->ctx.zstd, dict->data, dict->size);
                if (ZSTD_isError(r))
                    rb_raise(eError, "zstd dict reload on reset: %s", ZSTD_getErrorName(r));
            }
        }
        break;
    case ALGO_BROTLI:
        if (inf->ctx.brotli) {
            BrotliDecoderDestroyInstance(inf->ctx.brotli);
            inf->ctx.brotli = BrotliDecoderCreateInstance(NULL, NULL, NULL);
            if (!inf->ctx.brotli)
                rb_raise(eMemError, "brotli: failed to recreate decoder");
            if (dict) {
                BrotliDecoderAttachDictionary(inf->ctx.brotli, BROTLI_SHARED_DICTIONARY_RAW,
                                              dict->size, dict->data);
            }
        }
        break;
    case ALGO_LZ4:
        inf->lz4_buf.len = 0;
        inf->lz4_buf.offset = 0;
        break;
    }
    inf->closed = 0;
    inf->finished = 0;
    inf->total_output = 0;
    inf->total_input = 0;
    return self;
}

#write(chunk) ⇒ Object

# File 'ext/multi_compress/multi_compress.c', line 2883

static VALUE inflater_write(VALUE self, VALUE chunk) {
    inflater_t *inf;
    TypedData_Get_Struct(self, inflater_t, &inflater_type, inf);
    if (inf->closed)
        rb_raise(eStreamError, "stream is closed");
    StringValue(chunk);

    const char *src = RSTRING_PTR(chunk);
    size_t slen = RSTRING_LEN(chunk);
    const algo_policy_t *policy = algo_policy(inf->algo);
    if (slen == 0)
        return rb_binary_str_new("", 0);

    size_t input_accounted_before = inf->total_input;

    switch (inf->algo) {
    case ALGO_ZSTD: {
        ZSTD_inBuffer input = {src, slen, 0};
        size_t out_cap = ZSTD_DStreamOutSize();
        size_t result_cap = out_cap > slen * 2 ? out_cap : slen * 2;
        size_t remaining_total_budget =
            inf->max_output_size > inf->total_output ? inf->max_output_size - inf->total_output : 0;
        if (remaining_total_budget == 0)
            rb_raise(eDataError, "decompressed output exceeds limit (%zu bytes)",
                     inf->max_output_size);
        if (result_cap > remaining_total_budget)
            result_cap = remaining_total_budget;
        VALUE result = rb_binary_str_buf_reserve(result_cap);
        size_t result_len = 0;
        VALUE scheduler = current_fiber_scheduler();

        while (input.pos < input.size) {
            size_t remaining_budget = inf->max_output_size - inf->total_output - result_len;
            if (remaining_budget == 0)
                rb_raise(eDataError, "decompressed output exceeds limit (%zu bytes)",
                         inf->max_output_size);

            if (result_len + out_cap > result_cap) {
                size_t next_cap = result_cap * 2;
                if (next_cap > inf->max_output_size - inf->total_output)
                    next_cap = inf->max_output_size - inf->total_output;
                result_cap = next_cap;
                rb_str_resize(result, result_cap);
            }

            size_t current_out_cap = out_cap;
            if (current_out_cap > remaining_budget)
                current_out_cap = remaining_budget;

            ZSTD_outBuffer output = {RSTRING_PTR(result) + result_len, current_out_cap, 0};
            size_t ret;

            if (select_fiber_or_direct_mode(scheduler, input.size - input.pos,
                                            policy->fiber_stream_threshold) == WORK_EXEC_FIBER) {
                zstd_decompress_stream_chunk_fiber_t args = {
                    .dstream = inf->ctx.zstd,
                    .output = output,
                    .input = input,
                    .result = 0,
                };
                RUN_VIA_FIBER_WORKER(zstd_decompress_stream_chunk_fiber_nogvl, args);
                output.pos = args.output.pos;
                input.pos = args.input.pos;
                ret = args.result;
            } else {
                ret = ZSTD_decompressStream(inf->ctx.zstd, &output, &input);
            }

            if (ZSTD_isError(ret))
                rb_raise(eDataError, "zstd decompress stream: %s", ZSTD_getErrorName(ret));
            result_len = checked_add_size(result_len, output.pos,
                                          "decompressed output exceeds representable size");
            size_t total_output = checked_add_size(
                inf->total_output, result_len, "decompressed output exceeds representable size");
            size_t total_input = checked_add_size(input_accounted_before, input.pos,
                                                  "compressed input exceeds representable size");
            enforce_output_and_ratio_limits(total_output, total_input, inf->max_output_size,
                                            inf->max_ratio_enabled, inf->max_ratio);
            if (ret == 0)
                break;
        }
        inf->total_input = checked_add_size(input_accounted_before, input.pos,
                                            "compressed input exceeds representable size");
        inf->total_output = checked_add_size(inf->total_output, result_len,
                                             "decompressed output exceeds representable size");
        rb_str_set_len(result, result_len);
        RB_GC_GUARD(chunk);
        return result;
    }
    case ALGO_BROTLI: {
        size_t available_in = slen;
        const uint8_t *next_in = (const uint8_t *)src;
        size_t remaining_total_budget =
            inf->max_output_size > inf->total_output ? inf->max_output_size - inf->total_output : 0;
        if (remaining_total_budget == 0)
            rb_raise(eDataError, "decompressed output exceeds limit (%zu bytes)",
                     inf->max_output_size);
        size_t result_cap = slen * 2;
        if (result_cap < 1024)
            result_cap = 1024;
        if (result_cap > remaining_total_budget)
            result_cap = remaining_total_budget;
        VALUE result = rb_binary_str_buf_reserve(result_cap);
        size_t result_len = 0;
        VALUE scheduler = current_fiber_scheduler();

        while (available_in > 0 || BrotliDecoderHasMoreOutput(inf->ctx.brotli)) {
            size_t available_out = 0;
            uint8_t *next_out = NULL;
            BrotliDecoderResult res;

            if (select_fiber_or_direct_mode(scheduler, available_in,
                                            policy->fiber_stream_threshold) == WORK_EXEC_FIBER) {
                brotli_decompress_stream_fiber_t sargs = {
                    .dec = inf->ctx.brotli,
                    .available_in = available_in,
                    .next_in = next_in,
                    .available_out = available_out,
                    .next_out = next_out,
                    .result = BROTLI_DECODER_RESULT_ERROR,
                };
                RUN_VIA_FIBER_WORKER(brotli_decompress_stream_fiber_nogvl, sargs);
                available_in = sargs.available_in;
                next_in = sargs.next_in;
                available_out = sargs.available_out;
                next_out = sargs.next_out;
                res = sargs.result;
            } else {
                res = BrotliDecoderDecompressStream(inf->ctx.brotli, &available_in, &next_in,
                                                    &available_out, &next_out, NULL);
            }
            if (res == BROTLI_DECODER_RESULT_ERROR)
                rb_raise(eDataError, "brotli decompress stream: %s",
                         BrotliDecoderErrorString(BrotliDecoderGetErrorCode(inf->ctx.brotli)));
            const uint8_t *out_data;
            size_t out_size = 0;
            out_data = BrotliDecoderTakeOutput(inf->ctx.brotli, &out_size);
            if (out_size > 0) {
                size_t total_output = checked_add_size(
                    inf->total_output,
                    checked_add_size(result_len, out_size,
                                     "decompressed output exceeds representable size"),
                    "decompressed output exceeds representable size");
                size_t total_input =
                    checked_add_size(input_accounted_before, slen - available_in,
                                     "compressed input exceeds representable size");
                enforce_output_and_ratio_limits(total_output, total_input, inf->max_output_size,
                                                inf->max_ratio_enabled, inf->max_ratio);

                if (result_len + out_size > result_cap) {
                    result_cap = result_len + out_size;
                    rb_str_resize(result, result_cap);
                }

                memcpy(RSTRING_PTR(result) + result_len, out_data, out_size);
                result_len += out_size;
            }
            if (res == BROTLI_DECODER_RESULT_SUCCESS)
                break;
            if (res == BROTLI_DECODER_RESULT_NEEDS_MORE_INPUT && available_in == 0)
                break;
        }
        inf->total_input = checked_add_size(input_accounted_before, slen - available_in,
                                            "compressed input exceeds representable size");
        inf->total_output = checked_add_size(inf->total_output, result_len,
                                             "decompressed output exceeds representable size");
        rb_str_set_len(result, result_len);
        RB_GC_GUARD(chunk);
        return result;
    }
    case ALGO_LZ4: {
        size_t data_len = inf->lz4_buf.len - inf->lz4_buf.offset;
        size_t needed =
            checked_add_size(data_len, slen, "lz4 stream input buffer exceeds representable size");

        if (needed > inf->lz4_buf.cap) {
            size_t new_cap = needed > SIZE_MAX / 2 ? needed : needed * 2;
            if (inf->lz4_buf.offset > 0) {
                char *new_buf = ALLOC_N(char, new_cap);
                if (data_len > 0)
                    memcpy(new_buf, inf->lz4_buf.buf + inf->lz4_buf.offset, data_len);
                xfree(inf->lz4_buf.buf);
                inf->lz4_buf.buf = new_buf;
                inf->lz4_buf.offset = 0;
                inf->lz4_buf.len = data_len;
                inf->lz4_buf.cap = new_cap;
            } else {
                REALLOC_N(inf->lz4_buf.buf, char, new_cap);
                inf->lz4_buf.cap = new_cap;
            }
        } else if (inf->lz4_buf.offset > inf->lz4_buf.cap / 2) {
            if (data_len > 0)
                memmove(inf->lz4_buf.buf, inf->lz4_buf.buf + inf->lz4_buf.offset, data_len);
            inf->lz4_buf.offset = 0;
            inf->lz4_buf.len = data_len;
        }

        memcpy(inf->lz4_buf.buf + inf->lz4_buf.len, src, slen);
        inf->lz4_buf.len += slen;

        size_t remaining_total_budget =
            inf->max_output_size > inf->total_output ? inf->max_output_size - inf->total_output : 0;
        if (remaining_total_budget == 0)
            rb_raise(eDataError, "decompressed output exceeds limit (%zu bytes)",
                     inf->max_output_size);
        size_t result_cap = slen * 4;
        if (result_cap < 256)
            result_cap = 256;
        if (result_cap > remaining_total_budget)
            result_cap = remaining_total_budget;
        VALUE result = rb_binary_str_buf_new(result_cap);
        size_t result_len = 0;
        int use_fiber = has_fiber_scheduler();
        size_t fiber_counter = 0;

        size_t pos = inf->lz4_buf.offset;
        while (pos + 4 <= inf->lz4_buf.len) {
            const uint8_t *p = (const uint8_t *)(inf->lz4_buf.buf + pos);
            uint32_t orig_size = read_le_u32(p);
            if (orig_size == 0) {
                inf->finished = 1;
                pos += 4;
                break;
            }
            if (pos + 8 > inf->lz4_buf.len)
                break;
            uint32_t comp_size = read_le_u32(p + 4);
            if (pos + 8 + comp_size > inf->lz4_buf.len)
                break;
            if (orig_size > 64 * 1024 * 1024)
                rb_raise(eDataError, "lz4 stream: block too large (%u)", orig_size);

            size_t total_output =
                checked_add_size(inf->total_output,
                                 checked_add_size(result_len, orig_size,
                                                  "decompressed output exceeds representable size"),
                                 "decompressed output exceeds representable size");
            size_t total_input = checked_add_size(
                input_accounted_before, (pos + 8 + (size_t)comp_size) - inf->lz4_buf.offset,
                "compressed input exceeds representable size");
            enforce_output_and_ratio_limits(total_output, total_input, inf->max_output_size,
                                            inf->max_ratio_enabled, inf->max_ratio);

            if (result_len + orig_size > result_cap) {
                result_cap = result_len + orig_size;
                rb_str_resize(result, result_cap);
            }

            int dsize =
                LZ4_decompress_safe(inf->lz4_buf.buf + pos + 8, RSTRING_PTR(result) + result_len,
                                    (int)comp_size, (int)orig_size);
            if (dsize < 0)
                rb_raise(eDataError, "lz4 stream decompress block failed");

            result_len += dsize;
            pos += 8 + comp_size;
            if (use_fiber) {
                int did_yield = 0;
                fiber_counter = fiber_maybe_yield(fiber_counter, (size_t)dsize,
                                                  policy->fiber_yield_chunk, &did_yield);
                (void)did_yield;
            }
        }

        inf->total_input = checked_add_size(input_accounted_before, pos - inf->lz4_buf.offset,
                                            "compressed input exceeds representable size");
        inf->lz4_buf.offset = pos;
        inf->total_output = checked_add_size(inf->total_output, result_len,
                                             "decompressed output exceeds representable size");
        rb_str_set_len(result, result_len);
        RB_GC_GUARD(chunk);
        return result;
    }
    }
    return rb_binary_str_new("", 0);
}