store_io_coss.c

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

	     * by adding the object into the link list on the current stripe
	     */
	    storeCossAdd(SD, e, cs->curstripe);
	} else {
	    /* Relocate the object in COSS, but not in other layers */
	    storeCossNewPendingRelocate(cs, sio, sio->swap_filen, nf);
	    sio->swap_filen = nf;
	    cstate->flags.reloc = 1;

	    /*
	     * lock the new buffer so it doesn't get swapped out on us
	     * this will get unlocked in storeCossClose
	     */
	    storeCossMemBufLock(SD, sio);
	}
    }
    coss_stats.open.success++;
    return sio;
}

/*
 * Aha! The unlocked membuf.
 *
 * If its storeCossCreate, then it was locked. Fine.
 * If it was storeCossOpen() and we found the object in-stripe then cool,
 *   its locked.
 * If it was storeCossOpen() and we didn't find the object in-stripe then
 *   we reallocated the object into the current stripe and locked THAT.
 */
void
storeCossClose(SwapDir * SD, storeIOState * sio)
{
    debug(79, 3) ("storeCossClose: %p: offset %d\n", sio, sio->swap_filen);
    coss_stats.close.ops++;
    coss_stats.close.success++;
    storeCossMemBufUnlock(SD, sio);
    storeCossIOCallback(sio, 0);
}

void
storeCossRead(SwapDir * SD, storeIOState * sio, char *buf, size_t size, squid_off_t offset, STRCB * callback, void *callback_data)
{
    CossState *cstate = (CossState *) sio->fsstate;
    CossInfo *cs = (CossInfo *) SD->fsdata;
    CossReadOp *op;

    coss_stats.read.ops++;
    assert(sio->read.callback == NULL);
    assert(sio->read.callback_data == NULL);
    sio->read.callback = callback;
    sio->read.callback_data = callback_data;
    debug(79, 3) ("storeCossRead: %s: file number %d offset %ld\n", stripePath(SD), sio->swap_filen, (long int) offset);
    sio->offset = offset;
    cstate->flags.reading = 1;
    if ((offset + size) > sio->st_size)
	size = sio->st_size - offset;
    cstate->requestlen = size;
    cstate->requestbuf = buf;
    cstate->requestoffset = offset;
    /* All of these reads should be treated as pending ones */
    /* Ie, we create a read op; then we 'kick' the read op to see if it can be completed now */
    op = storeCossCreateReadOp(cs, sio);
    storeCossKickReadOp(cs, op);
}

void
storeCossWrite(SwapDir * SD, storeIOState * sio, char *buf, size_t size, squid_off_t offset, FREE * free_func)
{
    char *dest;
    CossMemBuf *membuf;
    off_t diskoffset;

    /*
     * If we get handed an object with a size of -1,
     * the squid code is broken
     */
    assert(sio->e->mem_obj->object_sz != -1);
    coss_stats.write.ops++;

    if (sio->offset != offset) {
	debug(79, 1) ("storeCossWrite: Possible data corruption on fileno %d, offsets do not match (Current:%" PRINTF_OFF_T " Want:%" PRINTF_OFF_T ")\n", sio->swap_filen, sio->offset, offset);
    }
    debug(79, 3) ("storeCossWrite: %s: offset %ld, len %lu\n", stripePath(SD),
	(long int) sio->offset, (unsigned long int) size);
    diskoffset = storeCossFilenoToDiskOffset(sio->swap_filen, SD->fsdata) + sio->offset;
    dest = storeCossMemPointerFromDiskOffset(SD->fsdata, diskoffset, &membuf);
    assert(dest != NULL);
    xmemcpy(dest, buf, size);
    sio->offset += size;
    if (free_func)
	(free_func) (buf);
    coss_stats.write.success++;
}


/*  === STATIC =========================================================== */

static void
storeCossIOCallback(storeIOState * sio, int errflag)
{
    CossState *cstate = (CossState *) sio->fsstate;
    debug(79, 3) ("storeCossIOCallback: errflag=%d\n", errflag);
    assert(NULL == cstate->locked_membuf);
    if (cbdataValid(sio->callback_data))
	sio->callback(sio->callback_data, errflag, sio);
    cbdataUnlock(sio->callback_data);
    sio->callback_data = NULL;
    cbdataFree(sio);
}

static char *
storeCossMemPointerFromDiskOffset(CossInfo * cs, off_t offset, CossMemBuf ** mb)
{
    CossMemBuf *t;
    dlink_node *m;

    for (m = cs->membufs.head; m; m = m->next) {
	t = m->data;
	if ((offset >= t->diskstart) && (offset < t->diskend)) {
	    if (mb)
		*mb = t;
	    return &t->buffer[offset - t->diskstart];
	}
    }

    if (mb)
	*mb = NULL;
    return NULL;
}

static CossMemBuf *
storeCossFilenoToMembuf(SwapDir * SD, sfileno s)
{
    CossMemBuf *t = NULL;
    dlink_node *m;
    CossInfo *cs = (CossInfo *) SD->fsdata;
    off_t o = storeCossFilenoToDiskOffset(s, cs);
    for (m = cs->membufs.head; m; m = m->next) {
	t = m->data;
	if ((o >= t->diskstart) && (o < t->diskend))
	    break;
    }
    assert(t);
    return t;
}

static void
storeCossMemBufLockPending(CossPendingReloc * pr, CossMemBuf * t)
{
    assert(t->flags.dead == 0);
    assert(pr->locked_membuf == NULL);
    debug(79, 3) ("storeCossMemBufLockPending: locking %p, lockcount %d\n",
	t, t->lockcount);
    pr->locked_membuf = t;
    t->lockcount++;
}

static void
storeCossMemBufUnlockPending(CossPendingReloc * pr, CossInfo * cs)
{
    CossMemBuf *t = pr->locked_membuf;
    if (NULL == t)
	return;
    assert(t->flags.dead == 0);
    debug(79, 3) ("storeCossMemBufLockPending: unlocking %p, lockcount %d\n",
	t, t->lockcount);
    t->lockcount--;
    pr->locked_membuf = NULL;

    if (!t->flags.written) {
	storeCossMaybeWriteMemBuf(t->SD, t);
    } else {
	/* cs->current_membuf may be invalid at this point */
	storeCossMaybeFreeBuf(cs, t);
    }
}

static void
storeCossMemBufLock(SwapDir * SD, storeIOState * sio)
{
    CossMemBuf *t = storeCossFilenoToMembuf(SD, sio->swap_filen);
    CossState *cstate = (CossState *) sio->fsstate;
    assert(cstate->locked_membuf == NULL);
    assert(t->flags.dead == 0);
    debug(79, 3) ("storeCossMemBufLock: locking %p, lockcount %d\n",
	t, t->lockcount);
    cstate->locked_membuf = t;
    t->lockcount++;
}

static void
storeCossMemBufUnlock(SwapDir * SD, storeIOState * sio)
{
    CossState *cstate = (CossState *) sio->fsstate;
    CossInfo *cs = SD->fsdata;
    CossMemBuf *t = cstate->locked_membuf;
    if (NULL == t)
	return;
    assert(t->flags.dead == 0);
    debug(79, 3) ("storeCossMemBufUnlock: unlocking %p, lockcount %d\n",
	t, t->lockcount);
    t->lockcount--;
    cstate->locked_membuf = NULL;
    if (!t->flags.written) {
	storeCossMaybeWriteMemBuf(SD, t);
    } else {
	/* cs->current_membuf may be invalid at this point */
	storeCossMaybeFreeBuf(cs, t);
    }
}

static void
storeCossMaybeWriteMemBuf(SwapDir * SD, CossMemBuf * t)
{
    //CossInfo *cs = SD->fsdata;
    membuf_describe(t, 3, __LINE__);
    assert(t->flags.dead == 0);
    if (!t->flags.full)
	debug(79, 3) ("membuf %p not full\n", t);
    else if (t->flags.writing)
	debug(79, 3) ("membuf %p writing\n", t);
    else if (t->lockcount)
	debug(79, 3) ("membuf %p lockcount=%d\n", t, t->lockcount);
    else if (t->flags.written)
	debug(79, 3) ("membuf %p written\n", t);
    else
	storeCossWriteMemBuf(SD, t);
    /* t may be invalid at this point */
}

void
storeCossSync(SwapDir * SD)
{
    CossInfo *cs = (CossInfo *) SD->fsdata;
    dlink_node *m;
    off_t end;

    /* First, flush pending IO ops */
#if USE_AUFSOPS
    aioSync(SD);
#else
    a_file_syncqueue(&cs->aq);
#endif

    /* Then, flush any in-memory partial membufs */
    if (!cs->membufs.head)
	return;
    for (m = cs->membufs.head; m; m = m->next) {
	CossMemBuf *t = m->data;
	if (t->flags.writing) {
	    debug(79, 1) ("WARNING: sleeping for 5 seconds in storeCossSync()\n");
	    sleep(5);		/* XXX EEEWWW! */
	}
	lseek(cs->fd, t->diskstart, SEEK_SET);
	end = (t == cs->current_membuf) ? cs->current_offset : t->diskend;
	FD_WRITE_METHOD(cs->fd, t->buffer, end - t->diskstart);
    }
}

static void
storeCossWriteMemBuf(SwapDir * SD, CossMemBuf * t)
{
    CossInfo *cs = (CossInfo *) SD->fsdata;
    int cur_load_interval = (squid_curtime / cs->load_interval) % 2;
    int prev_load_interval = ((squid_curtime + cs->load_interval) / cs->load_interval) % 2;
    assert(t->flags.dead == 0);
    debug(79, 3) ("storeCossWriteMemBuf: %p: offset %ld, len %ld\n", t,
	(long int) t->diskstart, (long int) (t->diskend - t->diskstart));
    t->flags.writing = 1;
    /* Check to see whether anything has a pending relocate (ie, a disk read)
     * scheduled from the disk data we're about to overwrite.
     * According to the specification this should never, ever happen - all the
     * objects underneath this stripe were deallocated before we started
     * using them - but there is a possibility that an object was opened
     * before the objects underneath the membufs stripe were purged and there
     * is still a pending relocate for it. Its a slim chance but it might happen.
     */
    if (!(t->flags.memonly)) {
	coss_stats.stripe_write.ops++;
	assert(t->stripe < cs->numstripes);
	if (cs->stripes[t->stripe].pending_relocs > 0) {
	    debug(79, 1) ("WARNING: %s: One or more pending relocate (reads) from stripe %d are queued - and I'm now writing over that part of the disk. This may result in object data corruption!\n", stripePath(SD), t->stripe);
	}
	/* Update load stats */
	cs->loadcalc[cur_load_interval] += 1;
	cs->loadcalc[prev_load_interval] = 0;

	/*
	 * normally nothing should have this node locked here - but between the time
	 * we call a_file_write and the IO completes someone might have snuck in and
	 * attached itself somehow. This is why there's a distinction between "written"
	 * and "writing". Read the rest of the code for more details.
	 */
#if USE_AUFSOPS
	/* XXX The last stripe, for now, ain't the coss stripe size for some reason */
	/* XXX This may cause problems later on; worry about figuring it out later on */
	//assert(t->diskend - t->diskstart == COSS_MEMBUF_SZ);
	debug(79, 3) ("aioWrite: FD %d: disk start: %" PRIu64 ", size %" PRIu64 "\n", cs->fd, (uint64_t) t->diskstart, (uint64_t) t->diskend - t->diskstart);
	aioWrite(cs->fd, t->diskstart, &(t->buffer[0]), COSS_MEMBUF_SZ, storeCossWriteMemBufDone, t, NULL);
#else
	a_file_write(&cs->aq, cs->fd, t->diskstart, &t->buffer,
	    COSS_MEMBUF_SZ, storeCossWriteMemBufDone, t, NULL);
#endif
    } else {
	/* No need to write, just mark as written and free */
	t->flags.written = 1;
	t->flags.writing = 0;
	storeCossMaybeFreeBuf(cs, t);
    }
}

/*
 * Check if a memory buffer can be freed.
 * Memory buffers can be freed if their refcount is 0 and they've been written.
 */
static void
storeCossMaybeFreeBuf(CossInfo * cs, CossMemBuf * mb)
{
    assert(mb->lockcount >= 0);
    /* It'd be nice if we could walk all the pending sio's somehow to see if some has this membuf locked .. */
    if (mb->flags.dead == 1) {
	debug(79, 1) ("storeCossMaybeFreeBuf: %p: dead; it'll be freed soon enough\n", mb);
	return;
    }
    /* Place on dead list rather than free
     * the asyncio code fails over to a 'sync' path; which may mean a membuf is
     * deallocated somewhere deep in the stack level. This way we just mark them
     * as dead and deallocate membufs early in the stack frame (ie, before we
     * call the asyncio disk completion handler.)
     */
    if (mb->lockcount == 0 && mb->flags.written == 1) {
	/* We need to wait until here to mark the membuf as
	 * free so we can re-alocate it
	 */
	if (mb->flags.memonly) {
	    assert(cs->memstripes[mb->stripe].membuf == mb);
	    cs->memstripes[mb->stripe].membuf = NULL;
	} else {
	    cs->numfullstripes--;
	}
	debug(79, 3) ("storeCossMaybeFreeBuf: %p: lockcount = 0, written = 1: marking dead\n", mb);
	mb->flags.dead = 1;
	dlinkDelete(&mb->node, &cs->membufs);
	dlinkAddTail(mb, &mb->node, &cs->dead_membufs);
	coss_stats.dead_stripes++;
	coss_stats.stripes--;
    }
}

void
storeCossFreeDeadMemBufs(CossInfo * cs)
{
    CossMemBuf *mb;
    while (cs->dead_membufs.head != NULL) {
	mb = cs->dead_membufs.head->data;
	assert(mb->flags.dead == 1);
	debug(79, 3) ("storeCossFreeDeadMemBufs: %p: freeing\n", mb);
	dlinkDelete(&mb->node, &cs->dead_membufs);
	cbdataFree(mb);
	coss_stats.dead_stripes--;
    }
}

/*
 * Writing a membuf has completed. Set the written flag to 1; membufs might have been
 * locked for read between the initial membuf write and the completion of the disk
 * write.
 */
#if USE_AUFSOPS
static void
storeCossWriteMemBufDone(int fd, void *my_data, const char *buf, int aio_return, int aio_errno)
#else
static void
storeCossWriteMemBufDone(int fd, int r_errflag, size_t r_len, void *my_data)
#endif
{
    CossMemBuf *t = my_data;
    CossInfo *cs = (CossInfo *) t->SD->fsdata;
    int errflag;
    int len;
#if USE_AUFSOPS
    len = aio_return;
    if (aio_errno)
	errflag = aio_errno == ENOSPC ? DISK_NO_SPACE_LEFT : DISK_ERROR;
    else
	errflag = DISK_OK;
#else
    len = r_len;
    errflag = r_errflag;
#endif

    debug(79, 3) ("storeCossWriteMemBufDone: stripe %d, buf %p, len %ld\n", t->stripe, t, (long int) len);
    if (errflag) {
	coss_stats.stripe_write.fail++;
	debug(79, 1) ("storeCossWriteMemBufDone: got failure (%d)\n", errflag);
	debug(79, 1) ("FD %d, size=%d\n", fd, (int) (t->diskend - t->diskstart));
    } else {
	coss_stats.stripe_write.success++;
    }
    assert(cs->stripes[t->stripe].membuf == t);
    debug(79, 2) ("storeCossWriteMemBufDone: %s: stripe %d: numobjs written: %d, lockcount %d\n", stripePath(t->SD), t->stripe, t->numobjs, t->lockcount);
    cs->stripes[t->stripe].numdiskobjs = t->numobjs;
    cs->stripes[t->stripe].membuf = NULL;
    t->flags.written = 1;
    t->flags.writing = 0;
    storeCossMaybeFreeBuf(cs, t);
}

static CossMemBuf *
storeCossCreateMemOnlyBuf(SwapDir * SD)
{
    CossMemBuf *newmb;
    CossInfo *cs = (CossInfo *) SD->fsdata;
    off_t start;
    int stripe;
    static time_t last_warn = 0;

    /* TODO: Maybe make this a simple search for a free membuf */
    for (stripe = 0; stripe < cs->nummemstripes; stripe++) {
	if (cs->memstripes[stripe].membuf == NULL)
	    break;
    }
    if (stripe >= cs->nummemstripes) {
	if (last_warn + 15 < squid_curtime) {
	    debug(79, 1) ("storeCossCreateMemOnlyBuf: no free membufs.  You may need to increase the value of membufs on the %s cache_dir\n", stripePath(SD));
	    last_warn = squid_curtime;
	}
	return NULL;
    }
    cs->curmemstripe = stripe;

    start = (off_t) stripe *COSS_MEMBUF_SZ;
    newmb = cbdataAlloc(CossMemBuf);

    cs->memstripes[stripe].membuf = newmb;
    newmb->diskstart = ((off_t) SD->max_size << 10) + start;
    newmb->stripe = stripe;
    newmb->diskend = newmb->diskstart + COSS_MEMBUF_SZ;
    newmb->flags.full = 0;
    newmb->flags.writing = 0;
    newmb->flags.memonly = 1;
    newmb->lockcount = 0;
    newmb->numobjs = 0;
    newmb->SD = SD;