store_io_coss.c

一个功能非常全面的代理服务器源代码程序,

    dlinkAdd(newmb, &newmb->node, &cs->membufs);

    coss_stats.stripes++;
    return newmb;
}

/*
 * This creates a memory buffer but assumes its going to be at the end
 * of the "LRU" and thusly will delete expire objects which appear under
 * it.
 */
static CossMemBuf *
storeCossCreateMemBuf(SwapDir * SD, int stripe, sfileno curfn, int *collision)
{
    CossMemBuf *newmb, *t;
    StoreEntry *e;
    dlink_node *m, *n;
    int numreleased = 0;
    CossInfo *cs = (CossInfo *) SD->fsdata;
    off_t start = (off_t) stripe * COSS_MEMBUF_SZ;
    off_t o;
    static time_t last_warn = 0;
    assert(start >= 0);

    if (cs->numfullstripes >= cs->maxfullstripes) {
	if (last_warn + 15 < squid_curtime) {
	    debug(79, 1) ("storeCossCreateMemBuf: Maximum number of full buffers reached on %s. You may need to increase the maxfullbuffers option for this cache_dir\n", stripePath(SD));
	    last_warn = squid_curtime;
	}
	return NULL;
    }
    /* No, we shouldn't ever try to create a membuf if we haven't freed the one on
     * this stripe. Grr */
    assert(cs->stripes[stripe].membuf == NULL);
    cs->curstripe = stripe;

    newmb = cbdataAlloc(CossMemBuf);
    cs->stripes[stripe].membuf = newmb;
    newmb->diskstart = start;
    newmb->stripe = stripe;
    debug(79, 2) ("storeCossCreateMemBuf: %s: creating new membuf at stripe %d,  %" PRId64 " (%p)\n", stripePath(SD), stripe, (int64_t) newmb->diskstart, newmb);
    newmb->diskend = newmb->diskstart + COSS_MEMBUF_SZ;
    newmb->flags.full = 0;
    newmb->flags.writing = 0;
    newmb->lockcount = 0;
    newmb->numobjs = 0;
    newmb->SD = SD;
    /* XXX This should be reversed, with the new buffer last in the chain */
    dlinkAdd(newmb, &newmb->node, &cs->membufs);
    assert(newmb->diskstart >= 0);
    assert(newmb->diskend >= 0);

    /* Print out the list of membufs */
    debug(79, 3) ("storeCossCreateMemBuf: %s: membuflist:\n", stripePath(SD));
    for (m = cs->membufs.head; m; m = m->next) {
	t = m->data;
	membuf_describe(t, 3, __LINE__);
    }

    /*
     * Kill objects from the tail to make space for a new chunk
     */
    m = cs->stripes[stripe].objlist.head;
    while (m != NULL) {
	n = m->next;
	e = m->data;
	o = storeCossFilenoToDiskOffset(e->swap_filen, cs);
	if (curfn > -1 && curfn == e->swap_filen)
	    *collision = 1;	/* Mark an object alloc collision */
	assert((o >= newmb->diskstart) && (o < newmb->diskend));
	debug(79, 3) ("COSS: %s: stripe %d, releasing filen %d (offset %" PRINTF_OFF_T ")\n", stripePath(SD), stripe, e->swap_filen, (squid_off_t) o);
	storeRelease(e);
	numreleased++;
	m = n;
    }
    if (numreleased > 0)
	debug(79, 3) ("storeCossCreateMemBuf: this allocation released %d storeEntries\n", numreleased);
    coss_stats.stripes++;
    return newmb;
}

/*
 * Creates the initial membuf after rebuild
 */
void
storeCossStartMembuf(SwapDir * sd)
{
    CossInfo *cs = (CossInfo *) sd->fsdata;
    CossMemBuf *newmb;
    CBDATA_INIT_TYPE_FREECB(storeIOState, storeCossIOFreeEntry);
    CBDATA_INIT_TYPE_FREECB(CossMemBuf, NULL);
    CBDATA_INIT_TYPE_FREECB(storeIOState, storeCossIOFreeEntry);
    CBDATA_INIT_TYPE_FREECB(CossPendingReloc, NULL);
    /*
     * XXX for now we start at the beginning of the disk;
     * The rebuild logic doesn't 'know' to pad out the current
     * offset to make it a multiple of COSS_MEMBUF_SZ.
     */
    newmb = storeCossCreateMemBuf(sd, 0, -1, NULL);
    assert(!cs->current_membuf);
    cs->current_membuf = newmb;

    newmb = storeCossCreateMemOnlyBuf(sd);
    assert(!cs->current_memonly_membuf);
    cs->current_memonly_membuf = newmb;

    cs->current_memonly_offset = cs->current_memonly_membuf->diskstart;
}

/*
 * Clean up any references from the SIO before it get's released.
 */
static void
storeCossIOFreeEntry(void *sio)
{
    memPoolFree(coss_state_pool, ((storeIOState *) sio)->fsstate);
}

static off_t
storeCossFilenoToDiskOffset(sfileno f, CossInfo * cs)
{
    off_t doff;

    doff = (off_t) f;
    doff <<= cs->blksz_bits;
    assert(doff >= 0);
    return doff;
}

static sfileno
storeCossDiskOffsetToFileno(off_t o, CossInfo * cs)
{
    assert(0 == (o & cs->blksz_mask));
    return o >> cs->blksz_bits;
}

static void
membuf_describe(CossMemBuf * t, int level, int line)
{
    assert(t->lockcount >= 0);
    debug(79, level) ("membuf id:%d (%p), LC:%02d, ST:%010lu, FL:%c%c%c\n",
	t->stripe,
	t,
	t->lockcount,
	(unsigned long) t->diskstart,
	t->flags.full ? 'F' : '.',
	t->flags.writing ? 'W' : '.',
	t->flags.written ? 'T' : '.');
}

int
storeCossFilenoToStripe(CossInfo * cs, sfileno filen)
{
    off_t o;
    /* Calculate sfileno to disk offset */
    o = ((off_t) filen) << cs->blksz_bits;
    /* Now, divide by COSS_MEMBUF_SZ to get which stripe it is in */
    return (int) (o / (off_t) COSS_MEMBUF_SZ);
}

/*
 * New stuff
 */
void
storeCossNewPendingRelocate(CossInfo * cs, storeIOState * sio, sfileno original_filen, sfileno new_filen)
{
    CossPendingReloc *pr;
    CossMemBuf *membuf;
    char *p;
    off_t disk_offset;
    int stripe;

    pr = cbdataAlloc(CossPendingReloc);
    cbdataLock(pr);
    pr->cs = cs;
    pr->original_filen = original_filen;
    pr->new_filen = new_filen;
    pr->len = sio->e->swap_file_sz;
    debug(79, 3) ("COSS Pending Relocate: %d -> %d: beginning\n", pr->original_filen, pr->new_filen);
    cs->pending_reloc_count++;
    dlinkAddTail(pr, &pr->node, &cs->pending_relocs);

    /* Update the stripe count */
    stripe = storeCossFilenoToStripe(cs, original_filen);
    assert(stripe >= 0);
    assert(stripe < cs->numstripes);
    assert(cs->stripes[stripe].pending_relocs >= 0);
    cs->stripes[stripe].pending_relocs++;

    /* And now; we begin the IO */
    p = storeCossMemPointerFromDiskOffset(cs, storeCossFilenoToDiskOffset(new_filen, cs), &membuf);
    pr->p = p;

    /* Lock the destination membuf */
    storeCossMemBufLockPending(pr, membuf);

    disk_offset = storeCossFilenoToDiskOffset(original_filen, cs);
    debug(79, 3) ("COSS Pending Relocate: size %" PRINTF_OFF_T ", disk_offset %" PRIu64 "\n", (squid_off_t) sio->e->swap_file_sz, (int64_t) disk_offset);
#if USE_AUFSOPS
    /* NOTE: the damned buffer isn't passed into aioRead! */
    debug(79, 3) ("COSS: aioRead: FD %d, from %d -> %d, offset %" PRIu64 ", len: %ld\n", cs->fd, pr->original_filen, pr->new_filen, (int64_t) disk_offset, (long int) pr->len);
    aioRead(cs->fd, (off_t) disk_offset, pr->len, storeCossCompletePendingReloc, pr);
#else
    a_file_read(&cs->aq, cs->fd,
	p,
	pr->len,
	disk_offset,
	storeCossCompletePendingReloc,
	pr);
#endif
}

CossPendingReloc *
storeCossGetPendingReloc(CossInfo * cs, sfileno new_filen)
{
    dlink_node *n;
    CossPendingReloc *pr;

    n = cs->pending_relocs.head;
    while (n != NULL) {
	pr = n->data;
	if (pr->new_filen == new_filen) {
	    return pr;
	}
	n = n->next;
    }
    return NULL;
}
#if USE_AUFSOPS
void
storeCossCompletePendingReloc(int fd, void *my_data, const char *buf, int aio_return, int aio_errno)
#else
void
storeCossCompletePendingReloc(int fd, const char *buf, int r_len, int r_errflag, void *my_data)
#endif
{
    CossPendingReloc *pr = my_data;
    CossReadOp *op;
    CossInfo *cs = pr->cs;
    int stripe;
    int errflag, len;
#if USE_AUFSOPS
    char *p;
#endif

#if USE_AUFSOPS
    len = aio_return;
    if (aio_errno)
	errflag = aio_errno == ENOSPC ? DISK_NO_SPACE_LEFT : DISK_ERROR;
    else
	errflag = DISK_OK;
#else
    errflag = r_errflag;
    len = r_len;
#endif

    debug(79, 3) ("storeCossCompletePendingReloc: %p\n", pr);
    assert(cbdataValid(pr));
    if (errflag != 0) {
	coss_stats.read.fail++;
	if (errflag > 0) {
	    errno = errflag;
	    debug(79, 1) ("storeCossCompletePendingReloc: error: %s\n", xstrerror());
	} else {
	    debug(79, 1) ("storeCossCompletePendingReloc: got failure (%d)\n", errflag);
	}
    } else {
	debug(79, 3) ("COSS Pending Relocate: %d -> %d: completed\n", pr->original_filen, pr->new_filen);
	coss_stats.read.success++;
    }
    /* aufs aioRead() doesn't take a buffer, it reads into its own. Grr */
#if USE_AUFSOPS
    p = storeCossMemPointerFromDiskOffset(cs, storeCossFilenoToDiskOffset(pr->new_filen, cs), NULL);
    assert(p != NULL);
    assert(p == pr->p);
    xmemcpy(p, buf, len);
#endif

    /* Nope, we're not a pending relocate anymore! */
    dlinkDelete(&pr->node, &cs->pending_relocs);

    /* Update the stripe count */
    stripe = storeCossFilenoToStripe(cs, pr->original_filen);
    assert(stripe >= 0);
    assert(stripe < cs->numstripes);
    assert(cs->stripes[stripe].pending_relocs >= 1);
    cs->stripes[stripe].pending_relocs--;

    /* Relocate has completed; we can now complete pending read ops on this particular entry */
    while (pr->ops.head != NULL) {
	op = pr->ops.head->data;
	debug(79, 3) ("storeCossCompletePendingReloc: %p: dequeueing op %p\n", pr, op);
	op->pr = NULL;
	dlinkDelete(&op->pending_op_node, &pr->ops);
	storeCossCompleteReadOp(cs, op, errflag);
	/* XXX again, this shouldn't be here (find the dlinkAddTail() in storeCossKickReadOp); these should
	 * be abstracted out. */
    }
    /* Unlock (and possibly write/free) the destination membuf */
    storeCossMemBufUnlockPending(pr, cs);
    /* Good, now we can delete it */
    cbdataUnlock(pr);
    cbdataFree(pr);
    assert(cs->pending_reloc_count != 0);
    cs->pending_reloc_count--;
}

/*
 * Begin a read operation
 *
 * the current 'state' of the read operation has already been set in storeIOState.
 *
 * We assume that the read operation will be from a currently in-memory MemBuf.
 */
CossReadOp *
storeCossCreateReadOp(CossInfo * cs, storeIOState * sio)
{
    CossReadOp *op;
    CossState *cstate = sio->fsstate;

    /* Create entry */
    op = memPoolAlloc(coss_op_pool);

    debug(79, 3) ("COSS: Creating Read operation: %p: filen %d, offset %" PRId64 ", size %" PRId64 "\n", op, sio->swap_filen, (int64_t) cstate->requestoffset, (int64_t) cstate->requestlen);

    /* Fill in details */
    op->type = COSS_OP_READ;
    op->sio = sio;
    cbdataLock(op->sio);
    op->requestlen = cstate->requestlen;
    op->requestoffset = cstate->requestoffset;
    op->reqdiskoffset = cstate->reqdiskoffset;
    op->requestbuf = cstate->requestbuf;

    /* Add to list */
    dlinkAddTail(op, &op->node, &cs->pending_ops);
    return op;
}

void
storeCossCompleteReadOp(CossInfo * cs, CossReadOp * op, int error)
{
    storeIOState *sio = op->sio;
    STRCB *callback = NULL;
    void *callback_data = NULL;
    CossState *cstate = sio->fsstate;
    ssize_t rlen = -1;
    char *p;
    SwapDir *SD = INDEXSD(sio->swap_dirn);

    debug(79, 3) ("storeCossCompleteReadOp: op %p, op dependencies satisfied, completing\n", op);

    assert(storeCossGetPendingReloc(cs, sio->swap_filen) == NULL);
    /* and make sure we aren't on a pending op list! */
    assert(op->pr == NULL);
    /* Is the callback still valid? If so; copy the data and callback */
    if (cbdataValid(sio) && cbdataValid(sio->read.callback_data)) {
	callback = sio->read.callback;
	callback_data = sio->read.callback_data;
	assert(callback);
	assert(callback_data);
	sio->read.callback = NULL;
	sio->read.callback_data = NULL;
	if (error == 0) {
	    /* P is the beginning of the object data we're interested in */
	    p = storeCossMemPointerFromDiskOffset(cs, storeCossFilenoToDiskOffset(sio->swap_filen, SD->fsdata), NULL);
	    assert(p != NULL);
	    /* cstate->requestlen contains the current copy length */
	    assert(cstate->requestlen == op->requestlen);
	    assert(cstate->requestbuf == op->requestbuf);
	    assert(cstate->requestoffset == op->requestoffset);
	    xmemcpy(cstate->requestbuf, &p[cstate->requestoffset], cstate->requestlen);
	    rlen = cstate->requestlen;
	}
	callback(callback_data, cstate->requestbuf, rlen);
    }
    cbdataUnlock(sio);		/* sio could have been freed here */
    op->sio = NULL;
    /* Remove from the operation list */
    dlinkDelete(&op->node, &cs->pending_ops);

    /* Completed! */
    memPoolFree(coss_op_pool, op);
}

/* See if the read op can be satisfied now */
void
storeCossKickReadOp(CossInfo * cs, CossReadOp * op)
{
    CossPendingReloc *pr;

    debug(79, 3) ("storeCossKickReadOp: op %p\n", op);

    if ((pr = storeCossGetPendingReloc(cs, op->sio->swap_filen)) == NULL) {
	debug(79, 3) ("COSS: filen: %d, tis already in memory; serving.\n", op->sio->swap_filen);
	storeCossCompleteReadOp(cs, op, 0);
    } else {
	debug(79, 3) ("COSS: filen: %d, not in memory, she'll have to wait.\n", op->sio->swap_filen);
	/* XXX Eww, hack! It has to be done; but doing it here is yuck */
	if (op->pr == NULL) {
	    debug(79, 3) ("storeCossKickReadOp: %p: op not bound to a pending read %p; binding\n", op, pr);
	    dlinkAddTail(op, &op->pending_op_node, &pr->ops);
	    op->pr = pr;
	}
    }
}

static void
membufsPrint(StoreEntry * e, CossMemBuf * t, const char *prefix)
{
    storeAppendPrintf(e, "%s: %d, lockcount: %d, numobjects %d, flags: %s,%s,%s,%s\n",
	prefix, t->stripe, t->lockcount, t->numobjs,
	t->flags.full ? "FULL" : "NOTFULL",
	t->flags.writing ? "WRITING" : "NOTWRITING",
	t->flags.written ? "WRITTEN" : "NOTWRITTEN",
	t->flags.memonly ? "MEMONLY" : "DISK");
}

void
membufsDump(CossInfo * cs, StoreEntry * e)
{
    dlink_node *m;
    int i;
    m = cs->membufs.head;
    while (m != NULL) {
	CossMemBuf *t = m->data;
	membufsPrint(e, t, "Stripe");
	m = m->next;
    }
    m = cs->dead_membufs.head;
    while (m != NULL) {
	CossMemBuf *t = m->data;
	membufsPrint(e, t, "Dead Stripe");
	m = m->next;
    }
    storeAppendPrintf(e, "Pending Relocations:\n");
    for (i = 0; i < cs->numstripes; i++) {
	if (cs->stripes[i].pending_relocs > 0) {
	    storeAppendPrintf(e, "  Stripe: %d   Number: %d\n", i, cs->stripes[i].pending_relocs);
	}
    }
}