store_dir_coss.c

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

    storeAppendPrintf(sentry, "dead_stripes:     %d\n", coss_stats.dead_stripes);
    storeAppendPrintf(sentry, "alloc.alloc:      %d\n", coss_stats.alloc.alloc);
    storeAppendPrintf(sentry, "alloc.realloc:    %d\n", coss_stats.alloc.realloc);
    storeAppendPrintf(sentry, "alloc.memalloc:   %d\n", coss_stats.alloc.memalloc);
    storeAppendPrintf(sentry, "alloc.collisions: %d\n", coss_stats.alloc.collisions);
    storeAppendPrintf(sentry, "disk_overflows:   %d\n", coss_stats.disk_overflows);
    storeAppendPrintf(sentry, "stripe_overflows: %d\n", coss_stats.stripe_overflows);
    storeAppendPrintf(sentry, "open_mem_hits:    %d\n", coss_stats.open_mem_hits);
    storeAppendPrintf(sentry, "open_mem_misses:  %d\n", coss_stats.open_mem_misses);
}

void
storeFsSetup_coss(storefs_entry_t * storefs)
{
    assert(!coss_initialised);

    storefs->parsefunc = storeCossDirParse;
    storefs->reconfigurefunc = storeCossDirReconfigure;
    storefs->donefunc = storeCossDirDone;
    coss_state_pool = memPoolCreate("COSS IO State data", sizeof(CossState));
    coss_index_pool = memPoolCreate("COSS index data", sizeof(CossIndexNode));
    coss_realloc_pool = memPoolCreate("COSS pending realloc", sizeof(CossPendingReloc));
    coss_op_pool = memPoolCreate("COSS pending operation", sizeof(CossReadOp));
    cachemgrRegister(SWAPDIR_COSS, "COSS Stats", storeCossStats, 0, 1);
    coss_initialised = 1;
}

/* New storedir rebuilding code! */

static void storeDirCoss_ReadStripe(RebuildState * rb);
static void storeDirCoss_ParseStripeBuffer(RebuildState * rb);
static void storeCoss_ConsiderStoreEntry(RebuildState * rb, const cache_key * key, StoreEntry * e);

#if USE_AUFSOPS
static void
storeDirCoss_ReadStripeComplete(int fd, void *my_data, const char *buf, int aio_return, int aio_errno)
#else
static void
storeDirCoss_ReadStripeComplete(int fd, const char *buf, int r_len, int r_errflag, void *my_data)
#endif
{
    RebuildState *rb = my_data;
    SwapDir *SD = rb->sd;
    CossInfo *cs = SD->fsdata;
#if USE_AUFSOPS
    int r_errflag;
    int r_len;
    r_len = aio_return;
    if (aio_errno)
	r_errflag = aio_errno == ENOSPC ? DISK_NO_SPACE_LEFT : DISK_ERROR;
    else
	r_errflag = DISK_OK;
    xmemcpy(cs->rebuild.buf, buf, r_len);
#endif

    debug(47, 2) ("COSS: %s: stripe %d, read %d bytes, status %d\n", stripePath(SD), cs->rebuild.curstripe, r_len, r_errflag);
    cs->rebuild.reading = 0;
    if (r_errflag != DISK_OK) {
	debug(47, 2) ("COSS: %s: stripe %d: error! Ignoring objects in this stripe.\n", stripePath(SD), cs->rebuild.curstripe);
	goto nextstripe;
    }
    cs->rebuild.buflen = r_len;
    /* parse the stripe contents */
    /* 
     * XXX note: the read should be put before the parsing so they can happen
     * simultaneously. This'll require some code-shifting so the read buffer
     * and parse buffer are different. This might speed up the read speed;
     * the disk throughput isn't being reached at the present.
     */
    storeDirCoss_ParseStripeBuffer(rb);

  nextstripe:
    cs->rebuild.curstripe++;
    if (cs->rebuild.curstripe >= cs->numstripes) {
	/* Completed the rebuild - move onto the next phase */
	debug(47, 2) ("COSS: %s: completed reading the stripes.\n", stripePath(SD));
	storeCossRebuildComplete(rb);
	return;
    } else {
	/* Next stripe */
	storeDirCoss_ReadStripe(rb);
    }
}

static void
storeDirCoss_ReadStripe(RebuildState * rb)
{
    SwapDir *SD = rb->sd;
    CossInfo *cs = SD->fsdata;

    assert(cs->rebuild.reading == 0);
    cs->rebuild.reading = 1;
    /* Use POSIX AIO for now */
    debug(47, 2) ("COSS: %s: reading stripe %d\n", stripePath(SD), cs->rebuild.curstripe);
    if (cs->rebuild.curstripe > rb->report_current) {
	debug(47, 1) ("COSS: %s: Rebuilding (%d %% completed - %d/%d stripes)\n", stripePath(SD),
	    cs->rebuild.curstripe * 100 / cs->numstripes, cs->rebuild.curstripe, cs->numstripes);
	rb->report_current += rb->report_interval;
    }
#if USE_AUFSOPS
    /* XXX this should be a prime candidate to use a modified aioRead which doesn't malloc a damned buffer */
    aioRead(cs->fd, (off_t) cs->rebuild.curstripe * COSS_MEMBUF_SZ, COSS_MEMBUF_SZ, storeDirCoss_ReadStripeComplete, rb);
#else
    a_file_read(&cs->aq, cs->fd, cs->rebuild.buf, COSS_MEMBUF_SZ, (off_t) cs->rebuild.curstripe * COSS_MEMBUF_SZ, storeDirCoss_ReadStripeComplete, rb);
#endif
}

static void
storeDirCoss_StartDiskRebuild(RebuildState * rb)
{
    SwapDir *SD = rb->sd;
    CossInfo *cs = SD->fsdata;
    assert(cs->rebuild.rebuilding == 0);
    assert(cs->numstripes > 0);
    assert(cs->rebuild.buf == NULL);
    assert(cs->fd >= 0);
    cs->rebuild.rebuilding = 1;
    cs->rebuild.curstripe = 0;
    cs->rebuild.buf = xmalloc(COSS_MEMBUF_SZ);
    rb->report_interval = cs->numstripes / COSS_REPORT_INTERVAL;
    rb->report_current = 0;
    debug(47, 2) ("COSS: %s: Beginning disk rebuild.\n", stripePath(SD));
    storeDirCoss_ReadStripe(rb);
}

/*
 * Take a stripe and attempt to place objects into it
 */
static void
storeDirCoss_ParseStripeBuffer(RebuildState * rb)
{
    SwapDir *SD = rb->sd;
    CossInfo *cs = SD->fsdata;
    tlv *t, *tlv_list;
    int j = 0;
    int bl = 0;
    int tmp;
    squid_off_t *l, len;
    int blocksize = cs->blksz_mask + 1;
    StoreEntry tmpe;
    cache_key key[MD5_DIGEST_CHARS];
    sfileno filen;

    assert(cs->rebuild.rebuilding == 1);
    assert(cs->numstripes > 0);
    assert(cs->rebuild.buf != NULL);

    if (cs->rebuild.buflen == 0) {
	debug(47, 3) ("COSS: %s: stripe %d: read 0 bytes, skipping stripe\n", stripePath(SD), cs->rebuild.curstripe);
	return;
    }
    while (j < cs->rebuild.buflen) {
	l = NULL;
	bl = 0;
	/* XXX there's no bounds checking on the buffer being passed into storeSwapMetaUnpack! */
	tlv_list = storeSwapMetaUnpack(cs->rebuild.buf + j, &bl);
	if (tlv_list == NULL) {
	    debug(47, 3) ("COSS: %s: stripe %d: offset %d gives NULL swapmeta data; end of stripe\n", stripePath(SD), cs->rebuild.curstripe, j);
	    return;
	}
	filen = (off_t) j / (off_t) blocksize + (off_t) ((off_t) cs->rebuild.curstripe * (off_t) COSS_MEMBUF_SZ / (off_t) blocksize);
	debug(47, 3) ("COSS: %s: stripe %d: filen %d: header size %d\n", stripePath(SD), cs->rebuild.curstripe, filen, bl);

	/* COSS objects will have an object size written into the metadata */
	memset(&tmpe, 0, sizeof(tmpe));
	memset(key, 0, sizeof(key));
	for (t = tlv_list; t; t = t->next) {
	    switch (t->type) {
	    case STORE_META_URL:
		debug(47, 3) ("    URL: %s\n", (char *) t->value);
		break;
	    case STORE_META_OBJSIZE:
		l = t->value;
		debug(47, 3) ("Size: %" PRINTF_OFF_T " (len %d)\n", *l, t->length);
		break;
	    case STORE_META_KEY:
		assert(t->length == MD5_DIGEST_CHARS);
		xmemcpy(key, t->value, MD5_DIGEST_CHARS);
		break;
#if SIZEOF_SQUID_FILE_SZ == SIZEOF_SIZE_T
	    case STORE_META_STD:
		assert(t->length == STORE_HDR_METASIZE);
		xmemcpy(&tmpe.timestamp, t->value, STORE_HDR_METASIZE);
		break;
#else
	    case STORE_META_STD_LFS:
		assert(t->length == STORE_HDR_METASIZE);
		xmemcpy(&tmpe.timestamp, t->value, STORE_HDR_METASIZE);
		break;
	    case STORE_META_STD:
		assert(t->length == STORE_HDR_METASIZE_OLD);
		{
		    struct {
			time_t timestamp;
			time_t lastref;
			time_t expires;
			time_t lastmod;
			size_t swap_file_sz;
			u_short refcount;
			u_short flags;
		    }     *tmp = t->value;
		    assert(sizeof(*tmp) == STORE_HDR_METASIZE_OLD);
		    tmpe.timestamp = tmp->timestamp;
		    tmpe.lastref = tmp->lastref;
		    tmpe.expires = tmp->expires;
		    tmpe.lastmod = tmp->lastmod;
		    tmpe.swap_file_sz = tmp->swap_file_sz;
		    tmpe.refcount = tmp->refcount;
		    tmpe.flags = tmp->flags;
		}
		break;
#endif
	    }
	}
	/* Make sure we have an object; if we don't then it may be an indication of trouble */
	if (l == NULL) {
	    debug(47, 3) ("COSS: %s: stripe %d: Object with no size; end of stripe\n", stripePath(SD), cs->rebuild.curstripe);
	    storeSwapTLVFree(tlv_list);
	    return;
	}
	len = *l;
	/* Finally, make sure there's enough data left in this stripe to satisfy the object
	 * we've just been informed about
	 */
	if ((cs->rebuild.buflen - j) < (len + bl)) {
	    debug(47, 3) ("COSS: %s: stripe %d: Not enough data in this stripe for this object, bye bye.\n", stripePath(SD), cs->rebuild.curstripe);
	    storeSwapTLVFree(tlv_list);
	    return;
	}
	/* Houston, we have an object */
	if (storeKeyNull(key)) {
	    debug(47, 3) ("COSS: %s: stripe %d: null data, next!\n", stripePath(SD), cs->rebuild.curstripe);
	    goto nextobject;
	}
	rb->counts.scancount++;
	tmpe.hash.key = key;
	/* Check sizes */
	if (tmpe.swap_file_sz == 0) {
	    tmpe.swap_file_sz = len + bl;
	}
	if (tmpe.swap_file_sz != (len + bl)) {
	    debug(47, 3) ("COSS: %s: stripe %d: file size mismatch (%" PRINTF_OFF_T " != %" PRINTF_OFF_T ")\n", stripePath(SD), cs->rebuild.curstripe, tmpe.swap_file_sz, len);
	    goto nextobject;
	}
	if (EBIT_TEST(tmpe.flags, KEY_PRIVATE)) {
	    debug(47, 3) ("COSS: %s: stripe %d: private key flag set, ignoring.\n", stripePath(SD), cs->rebuild.curstripe);
	    rb->counts.badflags++;
	    goto nextobject;
	}
	/* Time to consider the object! */
	tmpe.swap_filen = filen;
	tmpe.swap_dirn = SD->index;

	debug(47, 3) ("COSS: %s Considering filneumber %d\n", stripePath(SD), tmpe.swap_filen);
	storeCoss_ConsiderStoreEntry(rb, key, &tmpe);

      nextobject:
	/* Free the TLV data */
	storeSwapTLVFree(tlv_list);
	tlv_list = NULL;

	/* Now, advance to the next block-aligned offset after this object */
	j = j + len + bl;
	/* And now, the blocksize! */
	tmp = j / blocksize;
	tmp = (tmp + 1) * blocksize;
	j = tmp;
    }
}


static void
storeCoss_AddStoreEntry(RebuildState * rb, const cache_key * key, StoreEntry * e)
{
    StoreEntry *ne;
    SwapDir *SD = rb->sd;
    CossInfo *cs = SD->fsdata;
    rb->counts.objcount++;
    /* The Passed-in store entry is temporary; don't bloody use it directly! */
    assert(e->swap_dirn == SD->index);
    ne = new_StoreEntry(STORE_ENTRY_WITHOUT_MEMOBJ, NULL, NULL);
    ne->store_status = STORE_OK;
    storeSetMemStatus(ne, NOT_IN_MEMORY);
    ne->swap_status = SWAPOUT_DONE;
    ne->swap_filen = e->swap_filen;
    ne->swap_dirn = SD->index;
    ne->swap_file_sz = e->swap_file_sz;
    ne->lock_count = 0;
    ne->lastref = e->lastref;
    ne->timestamp = e->timestamp;
    ne->expires = e->expires;
    ne->lastmod = e->lastmod;
    ne->refcount = e->refcount;
    ne->flags = e->flags;
    EBIT_SET(ne->flags, ENTRY_CACHABLE);
    EBIT_CLR(ne->flags, RELEASE_REQUEST);
    EBIT_CLR(ne->flags, KEY_PRIVATE);
    ne->ping_status = PING_NONE;
    EBIT_CLR(ne->flags, ENTRY_VALIDATED);
    storeHashInsert(ne, key);	/* do it after we clear KEY_PRIVATE */
    storeCossAdd(SD, ne, cs->rebuild.curstripe);
    storeEntryDump(ne, 5);
    assert(ne->repl.data != NULL);
    assert(e->repl.data == NULL);
}

static void
storeCoss_DeleteStoreEntry(RebuildState * rb, const cache_key * key, StoreEntry * e)
{
    assert(rb->counts.objcount >= 0);
    /* XXX are these counters even correct, considering e could be a different storedir? */
    rb->counts.objcount--;
    assert(e->swap_dirn >= 0);
    storeRecycle(e);
}

/*
 * Consider inserting the given StoreEntry into the given
 * COSS directory.
 *
 * The rules for doing this is reasonably simple:
 *
 * If the object doesn't exist in the cache then we simply
 * add it to the current stripe list
 *
 * If the object does exist in the cache then we compare
 * "freshness"; if the newer object is fresher then we
 * remove it from its stripe and re-add it to the current
 * stripe.
 */
static void
storeCoss_ConsiderStoreEntry(RebuildState * rb, const cache_key * key, StoreEntry * e)
{
    StoreEntry *oe;

    /* Check for clashes */
    oe = storeGet(key);
    if (oe == NULL) {
	rb->cosscounts.new++;
	debug(47, 3) ("COSS: Adding filen %d\n", e->swap_filen);
	/* no clash! woo, can add and forget */
	storeCoss_AddStoreEntry(rb, key, e);
	return;
    }
    /* This isn't valid - its possible we have a fresher object in another store */
    /* unlike the UFS-based stores we don't "delete" the disk object when we
     * have deleted the object; its one of the annoying things about COSS. */
    //assert(oe->swap_dirn == SD->index);
    /* Dang, its a clash. See if its fresher */

    /* Fresher? Its a new object: deallocate the old one, reallocate the new one */
    if (e->lastref > oe->lastref) {
	debug(47, 3) ("COSS: fresher object for filen %d found (%ld -> %ld)\n", oe->swap_filen, (long int) oe->timestamp, (long int) e->timestamp);
	rb->cosscounts.fresher++;
	storeCoss_DeleteStoreEntry(rb, key, oe);
	oe = NULL;
	storeCoss_AddStoreEntry(rb, key, e);
	return;
    }
    /*
     * Not fresher? Its the same object then we /should/ probably relocate it; I'm
     * not sure what should be done here.
     */
    if (oe->timestamp == e->timestamp && oe->expires == e->expires) {
	debug(47, 3) ("COSS: filen %d -> %d (since they're the same!)\n", oe->swap_filen, e->swap_filen);
	rb->cosscounts.reloc++;
	storeCoss_DeleteStoreEntry(rb, key, oe);
	oe = NULL;
	storeCoss_AddStoreEntry(rb, key, e);
	return;
    }
    debug(47, 3) ("COSS: filen %d: ignoring this one for some reason\n", e->swap_filen);
    rb->cosscounts.unknown++;
}