view release on metacpan or  search on metacpan

External.xs  view on Meta::CPAN

            if (digit < 0x80) break; 
        }

        {
            /* Recover the stringified scalar from disk. */
            SV   *item_sv = newSV(item_length + 1);
            char  flags;

            SvCUR_set(item_sv, item_length);
            SvPOK_on(item_sv);
            check = PerlIO_read(fh, SvPVX(item_sv), item_length);
            if (check < (int)item_length) {
                croak("PerlIO error: read %d bytes, expected %"UVuf" bytes", 
                    check, (UV)item_length);
            }
            self->pos += check;
            *(SvEND(item_sv)) = '\0'; /* Null-terminate. */

            /* Restore UTF8 and taint flags. */
            check = PerlIO_read(fh, &flags, 1);
            self->pos++;
            if (check < 1) croak("PerlIO failed: %s", strerror(errno));
            if (flags & SORTEX_UTF8_FLAG)    SvUTF8_on(item_sv);
            if (flags & SORTEX_TAINTED_FLAG) SvTAINTED_on(item_sv);

            /* Add to the buffarray. */
            av_push(buffarray_av, item_sv);

            /* Track how much we've read so far. */
            num_items++;
        }
    }

    return num_items;
}

MODULE = Sort::External     PACKAGE = Sort::External

PROTOTYPES: DISABLE

SV*
_new(class_name, working_dir, sortsub, cache_size, mem_threshold, tempfile_fh) 
    char *class_name;
    SV *working_dir;
    SV *sortsub;
    IV cache_size;
    IV mem_threshold;
    SV *tempfile_fh;
CODE:
{
    SortExternal *self = SortEx_new(working_dir, sortsub, cache_size, 
        mem_threshold, tempfile_fh);
    RETVAL = newSV(0);
    sv_setref_pv(RETVAL, class_name, (void*)self);
}
OUTPUT: RETVAL

void
DESTROY(self)
    SortExternal *self;
PPCODE:
    SortEx_destroy(self);

void
feed(self, ...)
    SortExternal *self;
PPCODE:
{
    AV *const item_cache = self->item_cache;
    I32 start    = av_len(item_cache) + 1;
    I32 new_size = start + items - 1;
    IV  space    = (OVERHEAD + sizeof(SV*)) * (items - 1);
    I32 i;
    int need_to_flush = 0;

    /* Push arguments onto cache array. */
    av_extend(self->item_cache, new_size);
    for (i = 1; i < items; i++) {
        SV *const element = ST(i);
        SV *const copy    = newSVsv(element);
        av_push(item_cache, copy);

        /* Calculate a rough estimate of the memory occupied by the arguments:
         * sum the allocated string lengths, plus 15 bytes per scalar as
         * overhead.  Also, sv_len() has a side-effect of stringifying the SV.
         */
        space += sv_len(copy);
    }
    self->mem_bytes += space;

    if (self->cache_size > 0) {
        /* Use the number of cache elements as a flush-trigger. */
        if (av_len(self->item_cache) + 1 >= self->cache_size) 
            need_to_flush = 1;
    }
    else if (self->mem_bytes > self->mem_threshold) {
        /* Use memory consumption as a flush-trigger. */
        need_to_flush = 1;
    }
    if (need_to_flush) {
        ENTER;
        SAVETMPS;
        PUSHMARK(SP);
        XPUSHs( ST(0) );
        call_method("_write_item_cache_to_tempfile", G_VOID|G_DISCARD);
        PUTBACK;
        FREETMPS;
        LEAVE;
    }
}

void
fetch(self)
    SortExternal *self;
PPCODE:
{
    if (self->fetch_tick > av_len(self->item_cache)) {
        ENTER;
        SAVETMPS;
        PUSHMARK(SP);
        XPUSHs( ST(0) );
        call_method("_gatekeeper", G_VOID|G_DISCARD);
        PUTBACK;
        FREETMPS;
        LEAVE;
    }

    if (self->fetch_tick > av_len(self->item_cache)) {
        /* If empty, return false in both scalar and list context. */
        XSRETURN(0);
    }
    else {
        SV *elem = av_delete(self->item_cache, self->fetch_tick, 0);
        self->fetch_tick++;
        ST(0) = elem;
        XSRETURN(1);
    }
}

SV*
_get_something(self)
    SortExternal *self;
ALIAS:
    _get_sortsub       = 1
    _get_item_cache    = 2
    _get_runs          = 3
    _get_tempfile_fh   = 4
    _get_working_dir   = 5
CODE:
{
    switch(ix) {
        case 1: RETVAL = self->sortsub 
                       ? newSVsv(self->sortsub) 
                       : newSV(0);
                break;

        case 2: RETVAL = newRV_inc((SV*)self->item_cache);
                break;

        case 3: RETVAL = newRV_inc((SV*)self->runs);
                break;

        case 4: RETVAL = newSVsv(self->tempfile_fh);
                break;

        case 5: RETVAL = newSVsv(self->working_dir);
                break;

        default:
            croak("unrecognized alias number %d", ix);
    }
}
OUTPUT: RETVAL
    
void
_set_mem_bytes(self, mem_bytes)
    SortExternal *self;
    IV mem_bytes;
PPCODE:
    self->mem_bytes = mem_bytes;

void
_set_fetch_tick(self, fetch_tick)
    SortExternal *self;
    IV fetch_tick;
PPCODE:
    self->fetch_tick = fetch_tick;

void
_set_temp_fh(self, tempfile_fh)
    SortExternal *self;
    SV *tempfile_fh;
PPCODE:
    SvREFCNT_dec(self->tempfile_fh);
    self->tempfile_fh   = newSVsv(tempfile_fh);
    self->fh            = IoIFP( sv_2io(tempfile_fh) );

void
_print_to_sortfile(self, input_av, fh)
    SortExternal *self;
    AV *input_av;
    PerlIO *fh;
PPCODE:
    SortEx_print_to_sortfile(self, input_av, fh);

void
_utf8_on(sv)
    SV *sv;
PPCODE:
    /* Testing only. */
    SvUTF8_on(sv);

MODULE = Sort::External   PACKAGE = Sort::External::SortExRun

SortExRun*
_new(class_sv, tempfile_fh, start, end)
    SV *class_sv;
    SV *tempfile_fh;
    NV  start;
    NV  end;
CODE:
    RETVAL = SortExRun_new(tempfile_fh, start, end);
    (void)class_sv; /* Silence "unused var" compiler warning. */
OUTPUT: RETVAL

void
DESTROY(self)
    SortExRun *self;
PPCODE:
    SortExRun_destroy(self);

SV*
_get_buffarray(self)
    SortExRun *self;
CODE:
    RETVAL = newRV_inc((SV*)self->buffarray);
OUTPUT: RETVAL

void
_set_buffarray(self, buffarray);
    SortExRun *self;
    AV *buffarray;
PPCODE:
    SvREFCNT_dec((SV*)self->buffarray);
    self->buffarray = (AV*)SvREFCNT_inc((SV*)buffarray);

IV
_refill_buffer(self)
    SortExRun *self;
CODE:
    RETVAL = SortExRun_refill_buffer(self);
OUTPUT: RETVAL