Author: toad
Date: 2006-01-25 14:14:13 +0000 (Wed, 25 Jan 2006)
New Revision: 7917
Added:
branches/async-client/src/freenet/client/FetchWaiter.java
branches/async-client/src/freenet/client/PutWaiter.java
branches/async-client/src/freenet/client/async/ClientCallback.java
branches/async-client/src/freenet/client/async/ClientGetter.java
branches/async-client/src/freenet/client/async/ClientPutter.java
branches/async-client/src/freenet/client/async/RequestScheduler.java
branches/async-client/src/freenet/node/fcp/URIGeneratedMessage.java
Removed:
branches/async-client/src/freenet/client/BlockFetcher.java
branches/async-client/src/freenet/client/BlockInserter.java
branches/async-client/src/freenet/client/Fetcher.java
branches/async-client/src/freenet/client/FileInserter.java
branches/async-client/src/freenet/client/InsertSegment.java
branches/async-client/src/freenet/client/MultiFileInserter.java
branches/async-client/src/freenet/client/RetryTracker.java
branches/async-client/src/freenet/client/RetryTrackerCallback.java
branches/async-client/src/freenet/client/Segment.java
branches/async-client/src/freenet/client/SplitFetcher.java
branches/async-client/src/freenet/client/SplitInserter.java
branches/async-client/src/freenet/client/StdSplitfileBlock.java
branches/async-client/src/freenet/client/async/Client.java
branches/async-client/src/freenet/client/async/ClientGet.java
branches/async-client/src/freenet/client/async/ClientPut.java
branches/async-client/src/freenet/node/QueuedDataRequest.java
branches/async-client/src/freenet/node/QueuedInsertRequest.java
branches/async-client/src/freenet/node/QueuedRequest.java
branches/async-client/src/freenet/node/QueueingSimpleLowLevelClient.java
branches/async-client/src/freenet/node/RequestStarterClient.java
branches/async-client/src/freenet/node/SimpleLowLevelClient.java
Modified:
branches/async-client/src/freenet/client/ArchiveHandler.java
branches/async-client/src/freenet/client/ArchiveStoreContext.java
branches/async-client/src/freenet/client/FetcherContext.java
branches/async-client/src/freenet/client/HighLevelSimpleClient.java
branches/async-client/src/freenet/client/HighLevelSimpleClientImpl.java
branches/async-client/src/freenet/client/InserterContext.java
branches/async-client/src/freenet/client/async/ClientGetState.java
branches/async-client/src/freenet/client/async/ClientPutState.java
branches/async-client/src/freenet/client/async/ClientRequest.java
branches/async-client/src/freenet/client/async/ClientRequestScheduler.java
branches/async-client/src/freenet/client/async/MultiPutCompletionCallback.java
branches/async-client/src/freenet/client/async/SendableGet.java
branches/async-client/src/freenet/client/async/SendableRequest.java
branches/async-client/src/freenet/client/async/SingleBlockInserter.java
branches/async-client/src/freenet/client/async/SingleFileFetcher.java
branches/async-client/src/freenet/client/async/SingleFileInserter.java
branches/async-client/src/freenet/client/async/SplitFileFetcher.java
branches/async-client/src/freenet/client/async/SplitFileInserter.java
branches/async-client/src/freenet/node/LowLevelGetException.java
branches/async-client/src/freenet/node/Node.java
branches/async-client/src/freenet/node/RealNodeRequestInsertTest.java
branches/async-client/src/freenet/node/RequestStarter.java
branches/async-client/src/freenet/node/TextModeClientInterface.java
branches/async-client/src/freenet/node/fcp/ClientGet.java
branches/async-client/src/freenet/node/fcp/ClientPut.java
branches/async-client/src/freenet/node/fcp/FCPConnectionHandler.java
branches/async-client/src/freenet/support/UpdatableSortedLinkedList.java
Log:
All implemented.
Compiles.
Added FCP message GeneratedURI.
Now to debug...
Modified: branches/async-client/src/freenet/client/ArchiveHandler.java
===================================================================
--- branches/async-client/src/freenet/client/ArchiveHandler.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/ArchiveHandler.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -14,7 +14,7 @@
* @throws MetadataParseException If there was an error parsing
intermediary metadata.
*/
public abstract Bucket getMetadata(ArchiveContext archiveContext,
- FetcherContext fetchContext, ClientMetadata dm, int
recursionLevel,
+ ClientMetadata dm, int recursionLevel,
boolean dontEnterImplicitArchives)
throws ArchiveFailureException, ArchiveRestartException,
MetadataParseException, FetchException;
@@ -30,7 +30,7 @@
* @throws MetadataParseException
*/
public abstract Bucket get(String internalName,
- ArchiveContext archiveContext, FetcherContext
fetchContext,
+ ArchiveContext archiveContext,
ClientMetadata dm, int recursionLevel,
boolean dontEnterImplicitArchives)
throws ArchiveFailureException, ArchiveRestartException,
Modified: branches/async-client/src/freenet/client/ArchiveStoreContext.java
===================================================================
--- branches/async-client/src/freenet/client/ArchiveStoreContext.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/ArchiveStoreContext.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -37,43 +37,29 @@
* Get the metadata for a given archive.
* @return A Bucket containing the metadata, in binary format, for the
archive.
*/
- public Bucket getMetadata(ArchiveContext archiveContext, FetcherContext
fetchContext, ClientMetadata dm, int recursionLevel,
+ public Bucket getMetadata(ArchiveContext archiveContext, ClientMetadata
dm, int recursionLevel,
boolean dontEnterImplicitArchives) throws
ArchiveFailureException, ArchiveRestartException, MetadataParseException,
FetchException {
- return get(".metadata", archiveContext, fetchContext, dm,
recursionLevel, dontEnterImplicitArchives);
+ return get(".metadata", archiveContext, dm, recursionLevel,
dontEnterImplicitArchives);
}
/**
* Fetch a file in an archive. Will check the cache first, then fetch
the archive if
* necessary.
*/
- public Bucket get(String internalName, ArchiveContext archiveContext,
FetcherContext fetchContext, ClientMetadata dm, int recursionLevel,
+ public Bucket get(String internalName, ArchiveContext archiveContext,
ClientMetadata dm, int recursionLevel,
boolean dontEnterImplicitArchives) throws
ArchiveFailureException, ArchiveRestartException, MetadataParseException,
FetchException {
// Do loop detection on the archive that we are about to fetch.
archiveContext.doLoopDetection(key);
Bucket data;
-
+
// Fetch from cache
if((data = manager.getCached(key, internalName)) != null) {
return data;
}
- synchronized(this) {
- // Fetch from cache
- if((data = manager.getCached(key, internalName)) !=
null) {
- return data;
- }
-
- // Not in cache
-
- if(fetchContext == null) return null;
- fetchContext = new FetcherContext(fetchContext,
FetcherContext.SET_RETURN_ARCHIVES);
- Fetcher fetcher = new Fetcher(key, fetchContext,
archiveContext);
- FetchResult result = fetcher.realRun(dm,
recursionLevel, key, dontEnterImplicitArchives, fetchContext.localRequestOnly);
- manager.extractToCache(key, archiveType, result.data,
archiveContext, this);
- return manager.getCached(key, internalName);
- }
+ return null;
}
// Archive size
Deleted: branches/async-client/src/freenet/client/BlockFetcher.java
===================================================================
--- branches/async-client/src/freenet/client/BlockFetcher.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/BlockFetcher.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,122 +0,0 @@
-/**
- *
- */
-package freenet.client;
-
-import freenet.keys.FreenetURI;
-import freenet.support.Bucket;
-import freenet.support.Logger;
-
-public class BlockFetcher extends StdSplitfileBlock {
-
- private final Segment segment;
- final FreenetURI uri;
- final boolean dontEnterImplicitArchives;
- int completedTries;
- boolean actuallyFetched;
-
- public BlockFetcher(Segment segment, RetryTracker tracker, FreenetURI
freenetURI, int index, boolean dontEnterImplicitArchives) {
- super(tracker, index, null);
- this.segment = segment;
- uri = freenetURI;
- completedTries = 0;
- fetchedData = null;
- actuallyFetched = false;
- this.dontEnterImplicitArchives = dontEnterImplicitArchives;
- }
-
- public String getName() {
- return "BlockFetcher for "+getNumber();
- }
-
- public void run() {
- Logger.minor(this, "Running: "+this);
- // Already added to runningFetches.
- // But need to make sure we are removed when we exit.
- try {
- realRun();
- } catch (Throwable t) {
- fatalError(t, FetchException.INTERNAL_ERROR);
- } finally {
- completedTries++;
- }
- }
-
- public String toString() {
- return super.toString()+" tries="+completedTries+" uri="+uri;
- }
-
- private void realRun() {
- // Do the fetch
- Fetcher f = new Fetcher(uri, this.segment.blockFetchContext);
- try {
- FetchResult fr = f.realRun(new ClientMetadata(),
segment.recursionLevel, uri,
- (!this.segment.nonFullBlocksAllowed) ||
dontEnterImplicitArchives, segment.blockFetchContext.localRequestOnly ||
completedTries == 0);
- actuallyFetched = true;
- fetchedData = fr.data;
- Logger.minor(this, "Fetched "+fetchedData.size()+"
bytes on "+this);
- tracker.success(this);
- } catch (MetadataParseException e) {
- fatalError(e, FetchException.INVALID_METADATA);
- } catch (FetchException e) {
- int code = e.getMode();
- boolean isFatal = e.isFatal();
- if(isFatal)
- fatalError(e, code);
- else
- nonfatalError(e, code);
- } catch (ArchiveFailureException e) {
- fatalError(e, FetchException.ARCHIVE_FAILURE);
- } catch (ArchiveRestartException e) {
- Logger.error(this, "Got an ArchiveRestartException in a
splitfile - WTF?");
- fatalError(e, FetchException.ARCHIVE_FAILURE);
- }
- }
-
- private void fatalError(Throwable e, int code) {
- Logger.error(this, "Giving up on block: "+this+": "+e, e);
- tracker.fatalError(this, code);
- }
-
- private void nonfatalError(Exception e, int code) {
- Logger.minor(this, "Non-fatal error on "+this+": "+e);
- tracker.nonfatalError(this, code);
- }
-
- public boolean succeeded() {
- return fetchedData != null;
- }
-
- /**
- * Queue a healing block for insert.
- * Will be implemented using the download manager.
- * FIXME: implement!
- */
- public void queueHeal() {
- // TODO Auto-generated method stub
-
- }
-
- public void kill() {
- // Do nothing, for now
- }
-
- public FreenetURI getURI() {
- return uri;
- }
-
- public void setData(Bucket data) {
- actuallyFetched = false;
- super.setData(data);
- }
-
- protected void checkStartable() {
- if(fetchedData != null) {
- throw new IllegalStateException("Already have data");
- }
- }
-
- public int getRetryCount() {
- return completedTries;
- }
-}
\ No newline at end of file
Deleted: branches/async-client/src/freenet/client/BlockInserter.java
===================================================================
--- branches/async-client/src/freenet/client/BlockInserter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/BlockInserter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,150 +0,0 @@
-package freenet.client;
-
-import freenet.client.events.BlockInsertErrorEvent;
-import freenet.keys.FreenetURI;
-import freenet.support.Bucket;
-import freenet.support.Logger;
-
-/**
- * Inserts a single splitfile block.
- */
-public class BlockInserter extends StdSplitfileBlock implements Runnable {
-
- private boolean succeeded;
- private int completedTries;
- private final InserterContext ctx;
- private final InsertBlock block;
- private FreenetURI uri;
- private final boolean getCHKOnly;
- /** RNF count. We can count many consecutive RNFs as success. */
- private int rnfs;
-
- /**
- * Create a BlockInserter.
- * @param bucket The data to insert, or null if it will be filled in
later.
- * @param num The block number in the splitfile.
- */
- public BlockInserter(Bucket bucket, int num, RetryTracker tracker,
InserterContext ctx, boolean getCHKOnly) {
- super(tracker, num, bucket);
- succeeded = false;
- this.ctx = ctx;
- block = new InsertBlock(bucket, null, FreenetURI.EMPTY_CHK_URI);
- this.getCHKOnly = getCHKOnly;
- Logger.minor(this, "Created "+this);
- }
-
- public synchronized void setData(Bucket data) {
- if(this.fetchedData != null) throw new
IllegalArgumentException("Cannot set data when already have data");
- block.data = data;
- super.setData(data);
- }
-
- public void kill() {
- // Do nothing, for now.
- }
-
- public String toString() {
- return super.toString()+" succeeded="+succeeded+"
tries="+completedTries+" uri="+uri;
- }
-
- public FreenetURI getURI() {
- return uri;
- }
-
- public String getName() {
- return "BlockInserter for "+this.getNumber();
- }
-
- public void run() {
- try {
- Logger.minor(this, "Running "+this);
- if(fetchedData == null)
- throw new NullPointerException();
- realRun();
- } catch (Throwable t) {
- Logger.error(this, "Caught "+t+" on "+this, t);
- fatalError(t, InserterException.INTERNAL_ERROR);
- } finally {
- completedTries++;
- }
- }
-
- private void realRun() {
- FileInserter inserter = new FileInserter(ctx);
- try {
- if(uri == null && !getCHKOnly)
- uri = inserter.run(block, false, true, true,
null);
- uri = inserter.run(block, false, getCHKOnly, true,
null);
- succeeded = true;
- tracker.success(this);
- } catch (InserterException e) {
- int mode = e.getMode();
- switch(mode) {
- case InserterException.ROUTE_NOT_FOUND:
- // N consecutive RNFs = success
- if(ctx.consecutiveRNFsCountAsSuccess > 0) {
- rnfs++;
- if(rnfs >=
ctx.consecutiveRNFsCountAsSuccess) {
- succeeded = true;
- tracker.success(this);
- return;
- }
- }
- nonfatalError(e, mode);
- return;
- case InserterException.REJECTED_OVERLOAD:
- case InserterException.ROUTE_REALLY_NOT_FOUND:
- rnfs = 0;
- nonfatalError(e, mode);
- return;
- case InserterException.INTERNAL_ERROR:
- case InserterException.BUCKET_ERROR:
- fatalError(e, mode);
- return;
- case InserterException.FATAL_ERRORS_IN_BLOCKS:
- case InserterException.TOO_MANY_RETRIES_IN_BLOCKS:
- // Huh?
- Logger.error(this, "Got error inserting blocks
("+e.getMessage()+") while inserting a block - WTF?");
- fatalError(e, InserterException.INTERNAL_ERROR);
- return;
- case InserterException.INVALID_URI:
- Logger.error(this, "Got invalid URI error but
URI was CHK@ in block insert");
- fatalError(e, InserterException.INTERNAL_ERROR);
- return;
- default:
- rnfs = 0;
- Logger.error(this, "Unknown insert error
"+mode+" while inserting a block");
- fatalError(e, InserterException.INTERNAL_ERROR);
- return;
- }
- // FIXME add more cases as we create them
- }
-
- }
-
- private void fatalError(InserterException e, int code) {
- Logger.normal(this, "Giving up on block: "+this+": "+e);
- tracker.fatalError(this, code);
- ctx.eventProducer.produceEvent(new BlockInsertErrorEvent(e,
uri, completedTries));
- }
-
- private void fatalError(Throwable t, int code) {
- // Don't need to include uri
- fatalError(new InserterException(code, t, null), code);
- }
-
- private void nonfatalError(InserterException e, int code) {
- Logger.minor(this, "Non-fatal error on "+this+": "+e);
- tracker.nonfatalError(this, code);
- ctx.eventProducer.produceEvent(new BlockInsertErrorEvent(e,
uri, completedTries));
- }
-
- protected void checkStartable() {
- if(succeeded)
- throw new IllegalStateException("Already inserted
block");
- }
-
- public int getRetryCount() {
- return completedTries;
- }
-}
Added: branches/async-client/src/freenet/client/FetchWaiter.java
===================================================================
--- branches/async-client/src/freenet/client/FetchWaiter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/FetchWaiter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -0,0 +1,53 @@
+package freenet.client;
+
+import freenet.client.async.ClientCallback;
+import freenet.client.async.ClientGetter;
+import freenet.client.async.ClientPutter;
+import freenet.keys.FreenetURI;
+
+public class FetchWaiter implements ClientCallback {
+
+ private FetchResult result;
+ private FetchException error;
+ private boolean finished;
+
+ public synchronized void onSuccess(FetchResult result, ClientGetter
state) {
+ if(finished) return;
+ this.result = result;
+ finished = true;
+ notifyAll();
+ }
+
+ public synchronized void onFailure(FetchException e, ClientGetter
state) {
+ if(finished) return;
+ this.error = e;
+ finished = true;
+ notifyAll();
+ }
+
+ public void onSuccess(ClientPutter state) {
+ throw new UnsupportedOperationException();
+ }
+
+ public void onFailure(InserterException e, ClientPutter state) {
+ throw new UnsupportedOperationException();
+ }
+
+ public void onGeneratedURI(FreenetURI uri, ClientPutter state) {
+ throw new UnsupportedOperationException();
+ }
+
+ public FetchResult waitForCompletion() throws FetchException {
+ synchronized(this) {
+ while(!finished) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ }
+ }
+ if(error != null) throw error;
+ return result;
+ }
+}
Deleted: branches/async-client/src/freenet/client/Fetcher.java
===================================================================
--- branches/async-client/src/freenet/client/Fetcher.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/Fetcher.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,401 +0,0 @@
-package freenet.client;
-
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.util.LinkedList;
-
-import freenet.client.events.DecodedBlockEvent;
-import freenet.client.events.FetchedMetadataEvent;
-import freenet.client.events.GotBlockEvent;
-import freenet.keys.ClientKey;
-import freenet.keys.ClientKeyBlock;
-import freenet.keys.FreenetURI;
-import freenet.keys.KeyDecodeException;
-import freenet.node.LowLevelGetException;
-import freenet.support.Bucket;
-import freenet.support.BucketTools;
-import freenet.support.Logger;
-import freenet.support.compress.CompressionOutputSizeException;
-import freenet.support.compress.Compressor;
-
-/** Class that does the actual fetching. Does not have to have a user friendly
- * interface!
- */
-public class Fetcher {
-
- /** The original URI to be fetched. */
- final FreenetURI origURI;
- /** The settings for the fetch e.g. max file size */
- final FetcherContext ctx;
- /** The archive context object to be passed down the entire request.
This is
- * recreated if we get an ArchiveRestartException. It does loop
detection, partly
- * in order to prevent rare deadlocks.
- */
- ArchiveContext archiveContext;
-
- /**
- * Local-only constructor, with ArchiveContext, for recursion via e.g.
archives.
- */
- public Fetcher(FreenetURI uri, FetcherContext fctx, ArchiveContext
actx) {
- if(uri == null) throw new NullPointerException();
- origURI = uri;
- ctx = fctx;
- archiveContext = actx;
- }
-
- /**
- * Create a Fetcher. Public constructor, for when starting a new
request chain.
- * @param uri The key to fetch.
- * @param ctx The settings for the fetch.
- */
- public Fetcher(FreenetURI uri, FetcherContext ctx) {
- this(uri, ctx, new ArchiveContext());
- }
-
- /**
- * Fetch the key. Called by clients.
- * @return The key requested's data and client metadata.
- * @throws FetchException If we cannot fetch the key for some reason.
Various
- * other exceptions are used internally; they are converted to a
FetchException
- * by this driver routine.
- */
- public FetchResult run() throws FetchException {
- FailureCodeTracker tracker = new FailureCodeTracker(false);
- FetchException lastThrown = null;
- for(int j=0;j<ctx.maxNonSplitfileRetries+1;j++) {
- try {
- return runOnce();
- } catch (FetchException e) {
- lastThrown = e;
- tracker.merge(e);
- if(e.isFatal()) throw e;
- Logger.normal(this, "Possibly retrying "+this+"
despite "+e, e);
- continue;
- }
- }
- if(tracker.totalCount() == 1)
- throw lastThrown;
- else {
- if(tracker.isOneCodeOnly())
- throw new
FetchException(tracker.getFirstCode());
- throw new
FetchException(FetchException.SPLITFILE_ERROR, tracker);
- }
- }
-
- public FetchResult runOnce() throws FetchException {
- for(int i=0;i<ctx.maxArchiveRestarts;i++) {
- try {
- if(ctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- ClientMetadata dm = new ClientMetadata();
- ClientKey key;
- try {
- key = ClientKey.getBaseKey(origURI);
- } catch (MalformedURLException e2) {
- throw new
FetchException(FetchException.INVALID_URI, "Invalid URI: "+origURI);
- }
- LinkedList metaStrings =
origURI.listMetaStrings();
-
- FetchResult fr = realRun(dm, 0, key,
metaStrings, ctx.dontEnterImplicitArchives, ctx.localRequestOnly);
-
- if(metaStrings.isEmpty()) return fr;
- // Still got some meta-strings
- throw new
FetchException(FetchException.HAS_MORE_METASTRINGS);
-
- } catch (ArchiveRestartException e) {
- archiveContext = new ArchiveContext();
- continue;
- } catch (MetadataParseException e) {
- throw new FetchException(e);
- } catch (ArchiveFailureException e) {
-
if(e.getMessage().equals(ArchiveFailureException.TOO_MANY_LEVELS))
- throw new
FetchException(FetchException.TOO_DEEP_ARCHIVE_RECURSION);
- throw new FetchException(e);
- }
- }
- throw new
FetchException(FetchException.TOO_MANY_ARCHIVE_RESTARTS);
- }
-
- FetchResult realRun(ClientMetadata dm, int recursionLevel, FreenetURI
uri, boolean dontEnterImplicitArchives, boolean localOnly)
- throws FetchException, MetadataParseException, ArchiveFailureException,
ArchiveRestartException {
- ClientKey key;
- try {
- key = ClientKey.getBaseKey(origURI);
- } catch (MalformedURLException e2) {
- throw new FetchException(FetchException.INVALID_URI,
"Invalid URI: "+origURI);
- }
- LinkedList metaStrings = origURI.listMetaStrings();
-
- return realRun(dm, recursionLevel, key, metaStrings,
dontEnterImplicitArchives, localOnly);
- }
-
- /**
- * Fetch a key, within an overall fetch process. Called by self in
recursion, and
- * called by driver function @see run() .
- * @param dm The client metadata object to accumulate client metadata
in.
- * @param recursionLevel The recursion level. Incremented every time we
enter
- * realRun(). If it goes above a certain limit, we throw a
FetchException.
- * @param uri The URI to fetch.
- * @return The data, complete with client metadata.
- * @throws FetchException If we could not fetch the data.
- * @throws MetadataParseException If we could not parse the metadata.
- * @throws ArchiveFailureException If we could not extract data from an
archive.
- * @throws ArchiveRestartException
- */
- FetchResult realRun(ClientMetadata dm, int recursionLevel, ClientKey
key, LinkedList metaStrings, boolean dontEnterImplicitArchives, boolean
localOnly)
- throws FetchException, MetadataParseException, ArchiveFailureException,
ArchiveRestartException {
- Logger.minor(this, "Running fetch for: "+key);
- if(ctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- recursionLevel++;
- if(recursionLevel > ctx.maxRecursionLevel)
- throw new
FetchException(FetchException.TOO_MUCH_RECURSION, ""+recursionLevel+" should be
< "+ctx.maxRecursionLevel);
-
- // Do the fetch
- ClientKeyBlock block;
- try {
- block = ctx.client.getKey(key, localOnly,
ctx.starterClient, ctx.cacheLocalRequests, ctx.ignoreStore);
- if(ctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- } catch (LowLevelGetException e) {
- switch(e.code) {
- case LowLevelGetException.DATA_NOT_FOUND:
- throw new
FetchException(FetchException.DATA_NOT_FOUND);
- case LowLevelGetException.DATA_NOT_FOUND_IN_STORE:
- throw new
FetchException(FetchException.DATA_NOT_FOUND);
- case LowLevelGetException.DECODE_FAILED:
- throw new
FetchException(FetchException.BLOCK_DECODE_ERROR);
- case LowLevelGetException.INTERNAL_ERROR:
- throw new
FetchException(FetchException.INTERNAL_ERROR);
- case LowLevelGetException.REJECTED_OVERLOAD:
- throw new
FetchException(FetchException.REJECTED_OVERLOAD);
- case LowLevelGetException.ROUTE_NOT_FOUND:
- throw new
FetchException(FetchException.ROUTE_NOT_FOUND);
- case LowLevelGetException.TRANSFER_FAILED:
- throw new
FetchException(FetchException.TRANSFER_FAILED);
- case LowLevelGetException.VERIFY_FAILED:
- throw new
FetchException(FetchException.BLOCK_DECODE_ERROR);
- default:
- Logger.error(this, "Unknown
LowLevelGetException code: "+e.code);
- throw new
FetchException(FetchException.INTERNAL_ERROR);
- }
- }
-
- ctx.eventProducer.produceEvent(new GotBlockEvent(key));
-
- Bucket data;
- try {
- data = block.decode(ctx.bucketFactory, (int)
(Math.min(ctx.maxTempLength, Integer.MAX_VALUE)));
- } catch (KeyDecodeException e1) {
- throw new
FetchException(FetchException.BLOCK_DECODE_ERROR, e1.getMessage());
- } catch (IOException e) {
- Logger.error(this, "Could not capture data - disk
full?: "+e, e);
- throw new FetchException(FetchException.BUCKET_ERROR,
e);
- }
-
- ctx.eventProducer.produceEvent(new DecodedBlockEvent(key));
-
- if(!block.isMetadata()) {
- // Just return the data
- return new FetchResult(dm, data);
- }
-
- // Otherwise we need to parse the metadata
-
- if(data.size() > ctx.maxMetadataSize)
- throw new
FetchException(FetchException.TOO_BIG_METADATA);
- Metadata metadata;
- try {
- metadata =
Metadata.construct(BucketTools.toByteArray(data));
- } catch (IOException e) {
- throw new FetchException(FetchException.BUCKET_ERROR,
e);
- }
-
- ctx.eventProducer.produceEvent(new FetchedMetadataEvent());
-
- return runMetadata(dm, recursionLevel, key, metaStrings,
metadata, null, key.getURI(), dontEnterImplicitArchives, localOnly);
- }
-
- /**
- * Fetch data, from metadata.
- * @param recursionLevel The recursion level, from above. Not
incremented here, as we will
- * go through realRun() if the key changes, so the number of passes
here is severely limited.
- * @param key The key being fetched.
- * @param metaStrings List of unused meta strings (to be used by
manifests).
- * @param metadata The parsed metadata to process.
- * @param container The container in which this metadata is found.
- * @throws MetadataParseException If we could not parse metadata from a
sub-document. Will be
- * converted to a FetchException above.
- * @throws ArchiveFailureException If extracting data from an archive
failed.
- * @throws FetchException If the fetch failed for some reason.
- * @throws ArchiveRestartException
- */
- private FetchResult runMetadata(ClientMetadata dm, int recursionLevel,
ClientKey key, LinkedList metaStrings,
- Metadata metadata, ArchiveHandler container, FreenetURI
thisKey, boolean dontEnterImplicitArchives, boolean localOnly)
- throws MetadataParseException, FetchException, ArchiveFailureException,
ArchiveRestartException {
-
- if(ctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- if(metadata.isSimpleManifest()) {
- String name;
- if(metaStrings.isEmpty())
- name = null;
- else
- name = (String) metaStrings.removeFirst();
- // Since metadata is a document, we just replace
metadata here
- if(name == null) {
- metadata = metadata.getDefaultDocument();
- if(metadata == null)
- throw new
FetchException(FetchException.NOT_ENOUGH_METASTRINGS);
- } else {
- metadata = metadata.getDocument(name);
- thisKey = thisKey.pushMetaString(name);
- if(metadata == null)
- throw new
FetchException(FetchException.NOT_IN_ARCHIVE);
- }
- return runMetadata(dm, recursionLevel, key,
metaStrings, metadata, container, thisKey, dontEnterImplicitArchives,
localOnly);
- } else if(metadata.isArchiveManifest()) {
- container = ctx.archiveManager.makeHandler(thisKey,
metadata.getArchiveType(), false);
- Bucket metadataBucket =
container.getMetadata(archiveContext, ctx, dm, recursionLevel, true);
- try {
- metadata = Metadata.construct(metadataBucket);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR);
- }
- return runMetadata(dm, recursionLevel+1, key,
metaStrings, metadata, container, thisKey, dontEnterImplicitArchives,
localOnly);
- } else if(metadata.isArchiveInternalRedirect()) {
- if(container == null)
- throw new
FetchException(FetchException.NOT_IN_ARCHIVE);
- else {
- /* Implicit archive handling:
- * Sooner or later we reach a
SimpleFileRedirect to data, a Splitfile to data,
- * or an ArchiveInternalRedirect to data.
- *
- * In this case, if it is an archive type, if
implicit archive handling is enabled, and if
- * we have more meta-strings, we can try to
enter it.
- */
- if((!dontEnterImplicitArchives) &&
ArchiveManager.isUsableArchiveType(dm.getMIMEType()) &&
(!metaStrings.isEmpty())) {
- // Possible implicit archive inside
archive?
- container =
ctx.archiveManager.makeHandler(thisKey,
ArchiveManager.getArchiveType(dm.getMIMEType()), false);
- Bucket metadataBucket =
container.getMetadata(archiveContext, ctx, dm, recursionLevel, true);
- try {
- metadata =
Metadata.construct(metadataBucket);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR);
- }
- return runMetadata(dm,
recursionLevel+1, key, metaStrings, metadata, container, thisKey,
dontEnterImplicitArchives, localOnly);
- }
- Bucket result =
container.get(metadata.getZIPInternalName(), archiveContext, ctx, dm,
recursionLevel, true);
-
dm.mergeNoOverwrite(metadata.getClientMetadata());
- return new FetchResult(dm, result);
- }
- } else if(metadata.isMultiLevelMetadata()) {
- // Doesn't have to be a splitfile; could be from a ZIP
or a plain file.
- metadata.setSimpleRedirect();
- FetchResult res = runMetadata(dm, recursionLevel, key,
metaStrings, metadata, container, thisKey, true, localOnly);
- try {
- metadata = Metadata.construct(res.data);
- } catch (MetadataParseException e) {
- throw new
FetchException(FetchException.INVALID_METADATA, e);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR, e);
- }
- return runMetadata(dm, recursionLevel, key,
metaStrings, metadata, container, thisKey, dontEnterImplicitArchives,
localOnly);
- } else if(metadata.isSingleFileRedirect()) {
- FreenetURI uri = metadata.getSingleTarget();
- dm.mergeNoOverwrite(metadata.getClientMetadata());
- if((!dontEnterImplicitArchives) &&
ArchiveManager.isUsableArchiveType(dm.getMIMEType()) &&
(!metaStrings.isEmpty())) {
- // Is probably an implicit archive.
- ClientKey target;
- try {
- target = ClientKey.getBaseKey(uri);
- } catch (MalformedURLException e1) {
- throw new
FetchException(FetchException.INVALID_URI, "Invalid URI: "+uri);
- }
- // Probably a usable archive as-is. We may not
have to fetch it.
- container = ctx.archiveManager.makeHandler(uri,
ArchiveManager.getArchiveType(dm.getMIMEType()), true);
- if(container != null) {
- Bucket metadataBucket =
container.getMetadata(archiveContext, ctx, dm, recursionLevel, true);
- try {
- metadata =
Metadata.construct(metadataBucket);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR);
- }
- return runMetadata(dm,
recursionLevel+1, key, metaStrings, metadata, container, thisKey,
dontEnterImplicitArchives, localOnly);
- } // else just fetch it, create context later
- }
-
-
- ClientKey newKey;
- try {
- newKey = ClientKey.getBaseKey(uri);
- } catch (MalformedURLException e2) {
- throw new
FetchException(FetchException.INVALID_URI, "Invalid URI: "+uri);
- }
-
- LinkedList newMetaStrings = uri.listMetaStrings();
-
- // Move any new meta strings to beginning of our list
of remaining meta strings
- while(!newMetaStrings.isEmpty()) {
- Object o = newMetaStrings.removeLast();
- metaStrings.addFirst(o);
- }
-
- FetchResult fr = realRun(dm, recursionLevel, newKey,
metaStrings, dontEnterImplicitArchives, localOnly);
- if(metadata.isCompressed()) {
- Compressor codec =
Compressor.getCompressionAlgorithmByMetadataID(metadata.compressionCodec);
- Bucket data = fr.data;
- Bucket output;
- try {
- long maxLen = ctx.maxTempLength;
- if(maxLen < 0) maxLen = Long.MAX_VALUE;
- output = codec.decompress(data,
ctx.bucketFactory, maxLen);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR, e);
- } catch (CompressionOutputSizeException e) {
- throw new
FetchException(FetchException.TOO_BIG);
- }
- return new FetchResult(fr, output);
- }
- return fr;
- } else if(metadata.isSplitfile()) {
- // Straight data splitfile.
- // Might be used by parents for something else, in
which case they will set dontEnterImplicitArchives.
- dm.mergeNoOverwrite(metadata.getClientMetadata()); //
even splitfiles can have mime types!
- if((!dontEnterImplicitArchives) &&
ArchiveManager.isUsableArchiveType(dm.getMIMEType()) &&
(!metaStrings.isEmpty())) {
- // We know target is not metadata.
- container =
ctx.archiveManager.makeHandler(thisKey,
ArchiveManager.getArchiveType(dm.getMIMEType()), false);
- Bucket metadataBucket =
container.getMetadata(archiveContext, ctx, dm, recursionLevel, true);
- try {
- metadata =
Metadata.construct(metadataBucket);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR, e);
- }
- return runMetadata(dm, recursionLevel+1, key,
metaStrings, metadata, container, thisKey, dontEnterImplicitArchives,
localOnly);
- }
-
- FetcherContext newCtx;
- if(metadata.splitUseLengths)
- newCtx = new FetcherContext(ctx,
FetcherContext.SPLITFILE_USE_LENGTHS_MASK);
- else
- newCtx = new FetcherContext(ctx,
FetcherContext.SPLITFILE_DEFAULT_MASK);
-
- SplitFetcher sf = new SplitFetcher(metadata,
archiveContext, newCtx, recursionLevel);
- Bucket sfResult = sf.fetch(); // will throw in event of
error
- if(metadata.isCompressed()) {
- Logger.minor(this, "Is compressed:
"+metadata.compressionCodec);
- Compressor codec =
Compressor.getCompressionAlgorithmByMetadataID(metadata.compressionCodec);
- try {
- long maxLen = ctx.maxTempLength;
- if(maxLen < 0) maxLen = Long.MAX_VALUE;
- sfResult = codec.decompress(sfResult,
ctx.bucketFactory, maxLen);
- } catch (IOException e) {
- throw new
FetchException(FetchException.BUCKET_ERROR, e);
- } catch (CompressionOutputSizeException e) {
- throw new
FetchException(FetchException.TOO_BIG);
- }
- } else
- Logger.minor(this, "Not compressed
("+metadata.compressionCodec+")");
- return new FetchResult(dm, sfResult);
- } else {
- Logger.error(this, "Don't know what to do with
metadata: "+metadata);
- throw new
FetchException(FetchException.UNKNOWN_METADATA);
- }
- }
-}
Modified: branches/async-client/src/freenet/client/FetcherContext.java
===================================================================
--- branches/async-client/src/freenet/client/FetcherContext.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/FetcherContext.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -2,8 +2,6 @@
import freenet.client.events.ClientEventProducer;
import freenet.crypt.RandomSource;
-import freenet.node.RequestStarterClient;
-import freenet.node.SimpleLowLevelClient;
import freenet.support.BucketFactory;
/** Context for a Fetcher. Contains all the settings a Fetcher needs to know
about. */
@@ -15,7 +13,6 @@
public static final int SPLITFILE_USE_LENGTHS_MASK = 3;
public static final int SET_RETURN_ARCHIVES = 4;
/** Low-level client to send low-level requests to. */
- final SimpleLowLevelClient client;
public long maxOutputLength;
public long maxTempLength;
public final ArchiveManager archiveManager;
@@ -38,7 +35,6 @@
public int maxMetadataSize;
public int maxDataBlocksPerSegment;
public int maxCheckBlocksPerSegment;
- public final RequestStarterClient starterClient;
public boolean cacheLocalRequests;
private boolean cancelled;
/** If true, and we get a ZIP manifest, and we have no meta-strings
left, then
@@ -49,15 +45,14 @@
return cancelled;
}
- public FetcherContext(SimpleLowLevelClient client, long curMaxLength,
+ public FetcherContext(long curMaxLength,
long curMaxTempLength, int maxMetadataSize, int
maxRecursionLevel, int maxArchiveRestarts,
boolean dontEnterImplicitArchives, int
maxSplitfileThreads,
int maxSplitfileBlockRetries, int
maxNonSplitfileRetries,
boolean allowSplitfiles, boolean followRedirects,
boolean localRequestOnly,
int maxDataBlocksPerSegment, int
maxCheckBlocksPerSegment,
RandomSource random, ArchiveManager archiveManager,
BucketFactory bucketFactory,
- ClientEventProducer producer, RequestStarterClient
starter, boolean cacheLocalRequests) {
- this.client = client;
+ ClientEventProducer producer, boolean
cacheLocalRequests) {
this.maxOutputLength = curMaxLength;
this.maxTempLength = curMaxTempLength;
this.maxMetadataSize = maxMetadataSize;
@@ -77,13 +72,11 @@
this.eventProducer = producer;
this.maxDataBlocksPerSegment = maxDataBlocksPerSegment;
this.maxCheckBlocksPerSegment = maxCheckBlocksPerSegment;
- this.starterClient = starter;
this.cacheLocalRequests = cacheLocalRequests;
}
public FetcherContext(FetcherContext ctx, int maskID) {
if(maskID == IDENTICAL_MASK) {
- this.client = ctx.client;
this.maxOutputLength = ctx.maxOutputLength;
this.maxMetadataSize = ctx.maxMetadataSize;
this.maxTempLength = ctx.maxTempLength;
@@ -103,11 +96,9 @@
this.eventProducer = ctx.eventProducer;
this.maxDataBlocksPerSegment =
ctx.maxDataBlocksPerSegment;
this.maxCheckBlocksPerSegment =
ctx.maxCheckBlocksPerSegment;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
this.returnZIPManifests = ctx.returnZIPManifests;
} else if(maskID == SPLITFILE_DEFAULT_BLOCK_MASK) {
- this.client = ctx.client;
this.maxOutputLength = ctx.maxOutputLength;
this.maxMetadataSize = ctx.maxMetadataSize;
this.maxTempLength = ctx.maxTempLength;
@@ -127,11 +118,9 @@
this.eventProducer = ctx.eventProducer;
this.maxDataBlocksPerSegment = 0;
this.maxCheckBlocksPerSegment = 0;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
this.returnZIPManifests = false;
} else if(maskID == SPLITFILE_DEFAULT_MASK) {
- this.client = ctx.client;
this.maxOutputLength = ctx.maxOutputLength;
this.maxTempLength = ctx.maxTempLength;
this.maxMetadataSize = ctx.maxMetadataSize;
@@ -151,11 +140,9 @@
this.eventProducer = ctx.eventProducer;
this.maxDataBlocksPerSegment =
ctx.maxDataBlocksPerSegment;
this.maxCheckBlocksPerSegment =
ctx.maxCheckBlocksPerSegment;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
this.returnZIPManifests = ctx.returnZIPManifests;
} else if(maskID == SPLITFILE_USE_LENGTHS_MASK) {
- this.client = ctx.client;
this.maxOutputLength = ctx.maxOutputLength;
this.maxTempLength = ctx.maxTempLength;
this.maxMetadataSize = ctx.maxMetadataSize;
@@ -175,11 +162,9 @@
this.eventProducer = ctx.eventProducer;
this.maxDataBlocksPerSegment =
ctx.maxDataBlocksPerSegment;
this.maxCheckBlocksPerSegment =
ctx.maxCheckBlocksPerSegment;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
this.returnZIPManifests = ctx.returnZIPManifests;
} else if (maskID == SET_RETURN_ARCHIVES) {
- this.client = ctx.client;
this.maxOutputLength = ctx.maxOutputLength;
this.maxMetadataSize = ctx.maxMetadataSize;
this.maxTempLength = ctx.maxTempLength;
@@ -199,7 +184,6 @@
this.eventProducer = ctx.eventProducer;
this.maxDataBlocksPerSegment =
ctx.maxDataBlocksPerSegment;
this.maxCheckBlocksPerSegment =
ctx.maxCheckBlocksPerSegment;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
this.returnZIPManifests = true;
}
Deleted: branches/async-client/src/freenet/client/FileInserter.java
===================================================================
--- branches/async-client/src/freenet/client/FileInserter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/FileInserter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,325 +0,0 @@
-package freenet.client;
-
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.net.MalformedURLException;
-
-import freenet.client.events.SimpleBlockPutEvent;
-import freenet.keys.CHKBlock;
-import freenet.keys.CHKEncodeException;
-import freenet.keys.ClientCHKBlock;
-import freenet.keys.ClientSSKBlock;
-import freenet.keys.FreenetURI;
-import freenet.keys.InsertableClientSSK;
-import freenet.keys.NodeCHK;
-import freenet.keys.SSKBlock;
-import freenet.keys.SSKEncodeException;
-import freenet.node.LowLevelPutException;
-import freenet.support.Bucket;
-import freenet.support.BucketTools;
-import freenet.support.Logger;
-import freenet.support.compress.CompressionOutputSizeException;
-import freenet.support.compress.Compressor;
-
-/**
- * Class that does high-level inserts.
- */
-public class FileInserter {
-
- InserterContext ctx;
-
- public FileInserter(InserterContext context) {
- this.ctx = context;
- }
-
- /**
- * Do an insert.
- * @param block The data to insert.
- * @param localOnly
- * @param returnMetadata If not null, return the metadata in this
bucket, rather
- * than inserting it; return the *data* CHK only. This is used by e.g.
- * MultiFileInserter, where we will aggregate the metadata elsewhere.
- * Only supported on CHKs.
- * @return The URI of the inserted data.
- * @throws InserterException
- */
- public FreenetURI run(InsertBlock block, boolean metadata, boolean
getCHKOnly, boolean noRetries, Bucket returnMetadata) throws InserterException {
- if(block.data == null)
- throw new NullPointerException();
- String type = block.desiredURI.getKeyType();
- if(type.equalsIgnoreCase("CHK")) {
-
if(!block.desiredURI.toString(false).equalsIgnoreCase("CHK@"))
- throw new
InserterException(InserterException.INVALID_URI, null);
- } else if(!(type.equalsIgnoreCase("SSK") ||
type.equalsIgnoreCase("KSK"))) {
- throw new
InserterException(InserterException.INVALID_URI, null);
- }
-
- // Insert the content.
- // If we have reason to create a metadata document, include the
client metadata.
- // Otherwise only create one (a redirect) with the client
metadata, if there is any.
-
- // First, can it fit into a single block?
-
- Bucket origData = block.data;
- Bucket data = block.data;
- int blockSize;
- int maxSourceDataSize;
- boolean isSSK = false;
- boolean dontCompress = ctx.dontCompress;
-
- long origSize = data.size();
- if(type.equals("SSK") || type.equals("KSK")) {
- blockSize = SSKBlock.DATA_LENGTH;
- isSSK = true;
- maxSourceDataSize =
ClientSSKBlock.MAX_DECOMPRESSED_DATA_LENGTH;
- } else if(block.desiredURI.getKeyType().equals("CHK")) {
- blockSize = CHKBlock.DATA_LENGTH;
- maxSourceDataSize =
ClientCHKBlock.MAX_LENGTH_BEFORE_COMPRESSION;
- } else {
- throw new
InserterException(InserterException.INVALID_URI);
- }
-
- ClientCHKBlock chk;
-
- Compressor bestCodec = null;
- Bucket bestCompressedData = null;
-
- if(origSize > blockSize && (!ctx.dontCompress) &&
(!dontCompress)) {
- // Try to compress the data.
- // Try each algorithm, starting with the fastest and
weakest.
- // Stop when run out of algorithms, or the compressed
data fits in a single block.
- int algos = Compressor.countCompressAlgorithms();
- try {
- for(int i=0;i<algos;i++) {
- Compressor comp =
Compressor.getCompressionAlgorithmByDifficulty(i);
- Bucket result;
- result = comp.compress(origData,
ctx.bf, Long.MAX_VALUE);
- if(result.size() < blockSize) {
- bestCodec = comp;
- data = result;
- if(bestCompressedData != null)
-
ctx.bf.freeBucket(bestCompressedData);
- bestCompressedData = data;
- break;
- }
- if(bestCompressedData != null &&
result.size() < bestCompressedData.size()) {
-
ctx.bf.freeBucket(bestCompressedData);
- bestCompressedData = result;
- data = result;
- bestCodec = comp;
- } else if(bestCompressedData == null &&
result.size() < data.size()) {
- bestCompressedData = result;
- bestCodec = comp;
- data = result;
- }
- }
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, e, null);
- } catch (CompressionOutputSizeException e) {
- // Impossible
- throw new Error(e);
- }
- }
-
- InsertableClientSSK isk = null;
-
- if(isSSK && data.size() <= SSKBlock.DATA_LENGTH &&
block.clientMetadata.isTrivial()) {
- short codec;
- if(bestCodec == null) {
- codec = -1;
- } else {
- codec = bestCodec.codecNumberForMetadata();
- }
- try {
- isk =
InsertableClientSSK.create(block.desiredURI);
- } catch (MalformedURLException e1) {
- throw new
InserterException(InserterException.INVALID_URI, e1, null);
- }
- ClientSSKBlock ssk;
- try {
- ssk = isk.encode(data, metadata, true, codec,
data.size(), ctx.random);
- } catch (SSKEncodeException e) {
- throw new
InserterException(InserterException.INTERNAL_ERROR, e, isk.getURI());
- } catch (IOException e) {
- throw new
InserterException(InserterException.INTERNAL_ERROR, e, isk.getURI());
- }
- return simplePutSSK(ssk, getCHKOnly, noRetries);
- }
-
- if(isSSK) {
- // Insert as CHK
- // Create metadata pointing to it (include the
clientMetadata if there is any).
- FreenetURI uri = run(new InsertBlock(block.data, null,
FreenetURI.EMPTY_CHK_URI), metadata, getCHKOnly, noRetries, null);
- Metadata m = new Metadata(Metadata.SIMPLE_REDIRECT,
uri, block.clientMetadata);
- Bucket bucket;
- try {
- bucket =
BucketTools.makeImmutableBucket(ctx.bf, m.writeToByteArray());
- } catch (IOException e) {
- throw new
InserterException(InserterException.INTERNAL_ERROR, e, isk.getURI());
- }
- return run(new InsertBlock(bucket, null,
block.desiredURI), metadata, getCHKOnly, noRetries, null);
- }
-
- if(data.size() <= NodeCHK.BLOCK_SIZE) {
- try {
- if(bestCodec == null) {
- chk = ClientCHKBlock.encode(data,
metadata, true, (short)-1, 0);
- } else {
- if(origSize >
ClientCHKBlock.MAX_LENGTH_BEFORE_COMPRESSION)
- throw new
IllegalArgumentException("Data too big to compress into single block, but it
does");
- chk = ClientCHKBlock.encode(data,
metadata, false, bestCodec.codecNumberForMetadata(), (int)origSize);
- }
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, e, null);
- } catch (CHKEncodeException e) {
- Logger.error(this, "Unexpected error: "+e, e);
- throw new
InserterException(InserterException.INTERNAL_ERROR, null);
- }
- return simplePutCHK(chk, block.clientMetadata,
getCHKOnly, noRetries, returnMetadata);
- }
-
- // Too big, encode to a splitfile
- SplitInserter splitInsert = new SplitInserter(data,
block.clientMetadata, bestCodec, ctx.splitfileAlgorithm, ctx, this,
NodeCHK.BLOCK_SIZE, getCHKOnly, metadata, returnMetadata);
- return splitInsert.run();
- }
-
- /**
- * Simple insert. Only complication is that it might have some client
metadata.
- * @param chk The data encoded into a single CHK.
- * @param clientMetadata The client metadata. If this is non-trivial,
we will have to
- * create a redirect document just to put the metadata in.
- * @param returnMetadata If not null, return the metadata in this
bucket, rather
- * than inserting it; return the *data* CHK only. This is used by e.g.
- * MultiFileInserter, where we will aggregate the metadata elsewhere.
- * @return The URI of the resulting CHK.
- * @throws InserterException If there was an error inserting the block.
- */
- private FreenetURI simplePutCHK(ClientCHKBlock chk, ClientMetadata
clientMetadata, boolean getCHKOnly, boolean noRetries, Bucket returnMetadata)
throws InserterException {
- LowLevelPutException le = null;
- int rnfs = 0;
- for(int i=0;i<=ctx.maxInsertRetries;i++) {
- try {
- if(!getCHKOnly)
- ctx.eventProducer.produceEvent(new
SimpleBlockPutEvent(chk.getClientKey()));
- if(!getCHKOnly)
- ctx.client.putKey(chk,
ctx.starterClient, ctx.cacheLocalRequests);
- break;
- } catch (LowLevelPutException e) {
- le = e;
- switch(le.code) {
- case
LowLevelPutException.ROUTE_REALLY_NOT_FOUND:
- case LowLevelPutException.REJECTED_OVERLOAD:
- rnfs = 0;
- }
- if(noRetries)
- break;
- if(le.code ==
LowLevelPutException.ROUTE_NOT_FOUND && ctx.consecutiveRNFsCountAsSuccess > 0) {
- rnfs++;
- if(rnfs >=
ctx.consecutiveRNFsCountAsSuccess) {
- le = null;
- break;
- }
- }
- }
- }
-
- FreenetURI uri;
-
- if(clientMetadata == null || clientMetadata.isTrivial())
- // Don't need a redirect for the metadata
- uri = chk.getClientKey().getURI();
- else {
- // Do need a redirect for the metadata
- Metadata metadata = new
Metadata(Metadata.SIMPLE_REDIRECT, chk.getClientKey().getURI(), clientMetadata);
- if(returnMetadata != null) {
- uri = chk.getClientKey().getURI();
- try {
- DataOutputStream dos = new
DataOutputStream(returnMetadata.getOutputStream());
- metadata.writeTo(dos);
- dos.close();
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR);
- }
- } else {
- uri = putMetadataCHK(metadata, getCHKOnly,
noRetries);
- }
- }
-
- if(le != null)
- translateException(le, uri);
-
- return uri;
- }
-
- private FreenetURI simplePutSSK(ClientSSKBlock ssk, boolean getCHKOnly,
boolean noRetries) throws InserterException {
- LowLevelPutException le = null;
- int rnfs = 0;
- for(int i=0;i<=ctx.maxInsertRetries;i++) {
- try {
- if(!getCHKOnly)
- ctx.eventProducer.produceEvent(new
SimpleBlockPutEvent(ssk.getClientKey()));
- if(!getCHKOnly)
- ctx.client.putKey(ssk,
ctx.starterClient, ctx.cacheLocalRequests);
- break;
- } catch (LowLevelPutException e) {
- le = e;
- switch(le.code) {
- case
LowLevelPutException.ROUTE_REALLY_NOT_FOUND:
- case LowLevelPutException.REJECTED_OVERLOAD:
- rnfs = 0;
- }
- if(noRetries)
- break;
- if(le.code ==
LowLevelPutException.ROUTE_NOT_FOUND && ctx.consecutiveRNFsCountAsSuccess > 0) {
- rnfs++;
- if(rnfs >=
ctx.consecutiveRNFsCountAsSuccess) {
- le = null;
- break;
- }
- }
- if(le.code == LowLevelPutException.COLLISION)
- break;
- }
- }
-
- FreenetURI uri = ssk.getClientKey().getURI();
-
- if(le != null)
- translateException(le, uri);
-
- return uri;
- }
-
- private void translateException(LowLevelPutException e, FreenetURI uri)
throws InserterException {
- switch(e.code) {
- case LowLevelPutException.INTERNAL_ERROR:
- throw new
InserterException(InserterException.INTERNAL_ERROR, e, null);
- case LowLevelPutException.REJECTED_OVERLOAD:
- throw new
InserterException(InserterException.REJECTED_OVERLOAD, uri);
- case LowLevelPutException.ROUTE_NOT_FOUND:
- throw new
InserterException(InserterException.ROUTE_NOT_FOUND, uri);
- case LowLevelPutException.ROUTE_REALLY_NOT_FOUND:
- throw new
InserterException(InserterException.ROUTE_REALLY_NOT_FOUND, uri);
- case LowLevelPutException.COLLISION:
- throw new
InserterException(InserterException.COLLISION, uri);
- default:
- Logger.error(this, "Unknown LowLevelPutException code:
"+e.code+" on "+this);
- throw new
InserterException(InserterException.INTERNAL_ERROR, e, null);
- }
- }
-
- /** Put a metadata CHK
- * @throws InserterException If the insert fails.
- */
- private FreenetURI putMetadataCHK(Metadata metadata, boolean
getCHKOnly, boolean noRetries) throws InserterException {
- byte[] data = metadata.writeToByteArray();
- Bucket bucket;
- try {
- bucket = BucketTools.makeImmutableBucket(ctx.bf, data);
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, null);
- }
- InsertBlock block = new InsertBlock(bucket, null,
FreenetURI.EMPTY_CHK_URI);
- return run(block, true, getCHKOnly, noRetries, null);
- }
-}
Modified: branches/async-client/src/freenet/client/HighLevelSimpleClient.java
===================================================================
--- branches/async-client/src/freenet/client/HighLevelSimpleClient.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/HighLevelSimpleClient.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -4,7 +4,6 @@
import freenet.client.events.ClientEventListener;
import freenet.keys.FreenetURI;
-import freenet.node.RequestStarterClient;
public interface HighLevelSimpleClient {
Modified:
branches/async-client/src/freenet/client/HighLevelSimpleClientImpl.java
===================================================================
--- branches/async-client/src/freenet/client/HighLevelSimpleClientImpl.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/HighLevelSimpleClientImpl.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -3,6 +3,8 @@
import java.io.IOException;
import java.util.HashMap;
+import freenet.client.async.ClientGetter;
+import freenet.client.async.ClientPutter;
import freenet.client.events.ClientEventListener;
import freenet.client.events.ClientEventProducer;
import freenet.client.events.EventLogger;
@@ -10,8 +12,7 @@
import freenet.crypt.RandomSource;
import freenet.keys.ClientKey;
import freenet.keys.FreenetURI;
-import freenet.node.RequestStarterClient;
-import freenet.node.SimpleLowLevelClient;
+import freenet.node.Node;
import freenet.support.Bucket;
import freenet.support.BucketFactory;
import freenet.support.BucketTools;
@@ -19,17 +20,16 @@
public class HighLevelSimpleClientImpl implements HighLevelSimpleClient {
- private final SimpleLowLevelClient client;
private final ArchiveManager archiveManager;
+ private final short priorityClass;
private final BucketFactory bucketFactory;
+ private final Node node;
/** One CEP for all requests and inserts */
private final ClientEventProducer globalEventProducer;
private long curMaxLength;
private long curMaxTempLength;
private int curMaxMetadataLength;
private final RandomSource random;
- private final RequestStarterClient requestStarter;
- private final RequestStarterClient insertStarter;
/** See comments in Node */
private final boolean cacheLocalRequests;
static final int MAX_RECURSION = 10;
@@ -64,9 +64,10 @@
static final int SPLITFILE_CHECK_BLOCKS_PER_SEGMENT = 64;
- public HighLevelSimpleClientImpl(SimpleLowLevelClient client,
ArchiveManager mgr, BucketFactory bf, RandomSource r, RequestStarterClient
requestStarterClient, RequestStarterClient insertStarterClient, boolean
cacheLocalRequests) {
- this.client = client;
+ public HighLevelSimpleClientImpl(Node node, ArchiveManager mgr,
BucketFactory bf, RandomSource r, boolean cacheLocalRequests, short
priorityClass) {
+ this.node = node;
archiveManager = mgr;
+ this.priorityClass = priorityClass;
bucketFactory = bf;
random = r;
this.globalEventProducer = new SimpleEventProducer();
@@ -74,8 +75,6 @@
curMaxLength = Long.MAX_VALUE;
curMaxTempLength = Long.MAX_VALUE;
curMaxMetadataLength = 1024 * 1024;
- this.requestStarter = requestStarterClient;
- this.insertStarter = insertStarterClient;
this.cacheLocalRequests = cacheLocalRequests;
}
@@ -93,23 +92,25 @@
public FetchResult fetch(FreenetURI uri) throws FetchException {
if(uri == null) throw new NullPointerException();
FetcherContext context = getFetcherContext();
- Fetcher f = new Fetcher(uri, context);
- return f.run();
+ FetchWaiter fw = new FetchWaiter();
+ ClientGetter get = new ClientGetter(fw, node.fetchScheduler,
uri, context, priorityClass);
+ get.start();
+ return fw.waitForCompletion();
}
public FreenetURI insert(InsertBlock insert, boolean getCHKOnly) throws
InserterException {
+ return insert(insert, getCHKOnly, false);
+ }
+
+ public FreenetURI insert(InsertBlock insert, boolean getCHKOnly,
boolean isMetadata) throws InserterException {
InserterContext context = getInserterContext();
- FileInserter i = new FileInserter(context);
- return i.run(insert, false, getCHKOnly, false, null);
+ PutWaiter pw = new PutWaiter();
+ ClientPutter put = new ClientPutter(pw, insert.data,
insert.desiredURI, insert.clientMetadata,
+ context, node.putScheduler, priorityClass,
getCHKOnly, isMetadata);
+ put.start();
+ return pw.waitForCompletion();
}
- public FreenetURI insert(InsertBlock insert, boolean getCHKOnly,
boolean metadata) throws InserterException {
- InserterContext context = new InserterContext(client,
bucketFactory, random, INSERT_RETRIES, CONSECUTIVE_RNFS_ASSUME_SUCCESS,
- SPLITFILE_INSERT_THREADS,
SPLITFILE_BLOCKS_PER_SEGMENT, SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
globalEventProducer, insertStarter, cacheLocalRequests);
- FileInserter i = new FileInserter(context);
- return i.run(insert, metadata, getCHKOnly, false, null);
- }
-
public FreenetURI insertRedirect(FreenetURI insertURI, FreenetURI
targetURI) throws InserterException {
Metadata m = new Metadata(Metadata.SIMPLE_REDIRECT, targetURI,
new ClientMetadata());
Bucket b;
@@ -121,17 +122,11 @@
}
ClientKey k;
InsertBlock block = new InsertBlock(b, null, insertURI);
- InserterContext context = new InserterContext(client,
bucketFactory, random, INSERT_RETRIES, CONSECUTIVE_RNFS_ASSUME_SUCCESS,
- SPLITFILE_INSERT_THREADS,
SPLITFILE_BLOCKS_PER_SEGMENT, SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
globalEventProducer, insertStarter, cacheLocalRequests);
- FileInserter i = new FileInserter(context);
- return i.run(block, true, false, false, null);
+ return insert(block, false, true);
}
public FreenetURI insertManifest(FreenetURI insertURI, HashMap
bucketsByName, String defaultName) throws InserterException {
- InserterContext context = new InserterContext(client,
bucketFactory, random, INSERT_RETRIES, CONSECUTIVE_RNFS_ASSUME_SUCCESS,
- SPLITFILE_INSERT_THREADS,
SPLITFILE_BLOCKS_PER_SEGMENT, SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
globalEventProducer, insertStarter, cacheLocalRequests);
- MultiFileInserter mfi = new MultiFileInserter(insertURI,
bucketsByName, context, defaultName);
- return mfi.run();
+ throw new UnsupportedOperationException();
}
public void addGlobalHook(ClientEventListener listener) {
@@ -140,16 +135,16 @@
public FetcherContext getFetcherContext() {
return
- new FetcherContext(client, curMaxLength,
curMaxTempLength, curMaxMetadataLength,
+ new FetcherContext(curMaxLength, curMaxTempLength,
curMaxMetadataLength,
MAX_RECURSION, MAX_ARCHIVE_RESTARTS,
DONT_ENTER_IMPLICIT_ARCHIVES,
SPLITFILE_THREADS, SPLITFILE_BLOCK_RETRIES,
NON_SPLITFILE_RETRIES,
FETCH_SPLITFILES, FOLLOW_REDIRECTS,
LOCAL_REQUESTS_ONLY,
MAX_SPLITFILE_BLOCKS_PER_SEGMENT,
MAX_SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
- random, archiveManager, bucketFactory,
globalEventProducer, requestStarter, cacheLocalRequests);
+ random, archiveManager, bucketFactory,
globalEventProducer, cacheLocalRequests);
}
public InserterContext getInserterContext() {
- return new InserterContext(client, bucketFactory, random,
INSERT_RETRIES, CONSECUTIVE_RNFS_ASSUME_SUCCESS,
- SPLITFILE_INSERT_THREADS,
SPLITFILE_BLOCKS_PER_SEGMENT, SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
globalEventProducer, insertStarter, cacheLocalRequests);
+ return new InserterContext(bucketFactory, random,
INSERT_RETRIES, CONSECUTIVE_RNFS_ASSUME_SUCCESS,
+ SPLITFILE_INSERT_THREADS,
SPLITFILE_BLOCKS_PER_SEGMENT, SPLITFILE_CHECK_BLOCKS_PER_SEGMENT,
globalEventProducer, cacheLocalRequests);
}
}
Deleted: branches/async-client/src/freenet/client/InsertSegment.java
===================================================================
--- branches/async-client/src/freenet/client/InsertSegment.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/InsertSegment.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,69 +0,0 @@
-package freenet.client;
-
-import java.io.IOException;
-
-import freenet.keys.FreenetURI;
-import freenet.support.BucketFactory;
-import freenet.support.Logger;
-
-/**
- * Segment of a splitfile, for insertion purposes.
- */
-public class InsertSegment {
-
- final FECCodec codec;
- final StartableSplitfileBlock[] origDataBlocks;
- final int blockLength;
- final BucketFactory bf;
- /** Check blocks. Will be created by encode(...). */
- final StartableSplitfileBlock[] checkBlocks;
- final boolean getCHKOnly;
- // just for debugging
- final int segNo;
-
- public InsertSegment(short splitfileAlgo, StartableSplitfileBlock[]
origDataBlocks, int blockLength, BucketFactory bf, boolean getCHKOnly, int
segNo) {
- this.origDataBlocks = origDataBlocks;
- codec = FECCodec.getCodec(splitfileAlgo, origDataBlocks.length);
- if(codec != null)
- checkBlocks = new
StartableSplitfileBlock[codec.countCheckBlocks()];
- else
- checkBlocks = new StartableSplitfileBlock[0];
- this.blockLength = blockLength;
- this.bf = bf;
- this.getCHKOnly = getCHKOnly;
- this.segNo = segNo;
- // FIXME: remove debugging code
- for(int i=0;i<origDataBlocks.length;i++)
- if(origDataBlocks[i].getData() == null) throw new
NullPointerException("Block "+i+" of "+origDataBlocks.length+" data blocks of
seg "+segNo+" is null");
- }
-
- /**
- * Get the check block URIs.
- * Don't call before encode()! Don't call before all blocks have
inserted either.
- */
- public FreenetURI[] getCheckURIs() {
- FreenetURI[] uris = new FreenetURI[checkBlocks.length];
- for(int i=0;i<uris.length;i++) {
- FreenetURI uri = checkBlocks[i].getURI();
- uris[i] = uri;
- }
- return uris;
- }
-
- /**
- * Encode the data blocks into check blocks.
- * @return The number of check blocks generated.
- * @throws IOException If the encode fails due to a bucket error.
- */
- public int encode(int offset, RetryTracker tracker, InserterContext
ctx) throws IOException {
- Logger.minor(this, "Encoding "+segNo+":
"+origDataBlocks.length+" into "+checkBlocks.length);
- if(codec == null) return 0; // no FEC
- for(int i=0;i<checkBlocks.length;i++)
- checkBlocks[i] = new BlockInserter(null, offset + i,
tracker, ctx, getCHKOnly);
- codec.encode(origDataBlocks, checkBlocks, blockLength, bf);
- for(int i=0;i<checkBlocks.length;i++)
- tracker.addBlock(checkBlocks[i]);
- return checkBlocks.length;
- }
-
-}
Modified: branches/async-client/src/freenet/client/InserterContext.java
===================================================================
--- branches/async-client/src/freenet/client/InserterContext.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/InserterContext.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -2,14 +2,11 @@
import freenet.client.events.ClientEventProducer;
import freenet.crypt.RandomSource;
-import freenet.node.RequestStarterClient;
-import freenet.node.SimpleLowLevelClient;
import freenet.support.BucketFactory;
/** Context object for an insert operation, including both simple and
multi-file inserts */
public class InserterContext {
- final SimpleLowLevelClient client;
public final BucketFactory bf;
/** If true, don't try to compress the data */
public final boolean dontCompress;
@@ -21,15 +18,13 @@
public final int splitfileSegmentDataBlocks;
public final int splitfileSegmentCheckBlocks;
final ClientEventProducer eventProducer;
- final RequestStarterClient starterClient;
/** Interesting tradeoff, see comments at top of Node.java. */
final boolean cacheLocalRequests;
private boolean cancelled;
- public InserterContext(SimpleLowLevelClient client, BucketFactory bf,
RandomSource random,
+ public InserterContext(BucketFactory bf, RandomSource random,
int maxRetries, int rnfsToSuccess, int maxThreads, int
splitfileSegmentDataBlocks, int splitfileSegmentCheckBlocks,
- ClientEventProducer eventProducer, RequestStarterClient
sctx, boolean cacheLocalRequests) {
- this.client = client;
+ ClientEventProducer eventProducer, boolean
cacheLocalRequests) {
this.bf = bf;
this.random = random;
dontCompress = false;
@@ -40,12 +35,10 @@
this.eventProducer = eventProducer;
this.splitfileSegmentDataBlocks = splitfileSegmentDataBlocks;
this.splitfileSegmentCheckBlocks = splitfileSegmentCheckBlocks;
- this.starterClient = sctx;
this.cacheLocalRequests = cacheLocalRequests;
}
public InserterContext(InserterContext ctx) {
- this.client = ctx.client;
this.bf = ctx.bf;
this.random = ctx.random;
this.dontCompress = ctx.dontCompress;
@@ -56,7 +49,6 @@
this.eventProducer = ctx.eventProducer;
this.splitfileSegmentDataBlocks =
ctx.splitfileSegmentDataBlocks;
this.splitfileSegmentCheckBlocks =
ctx.splitfileSegmentCheckBlocks;
- this.starterClient = ctx.starterClient;
this.cacheLocalRequests = ctx.cacheLocalRequests;
}
Deleted: branches/async-client/src/freenet/client/MultiFileInserter.java
===================================================================
--- branches/async-client/src/freenet/client/MultiFileInserter.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/MultiFileInserter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,191 +0,0 @@
-package freenet.client;
-
-import java.io.IOException;
-import java.util.HashMap;
-
-import freenet.keys.FreenetURI;
-import freenet.support.ArrayBucket;
-import freenet.support.Bucket;
-import freenet.support.BucketTools;
-import freenet.support.Logger;
-
-public class MultiFileInserter {
-
- public class MFInserter implements Runnable {
-
- final int num;
-
- MFInserter(int x) {
- num = x;
- }
-
- public void run() {
- try {
- while(true) {
- String name = null;
- Bucket data = null;
- synchronized(bucketsByName) {
- if(bucketsByName.isEmpty()) break;
- name = (String)
bucketsByName.keySet().iterator().next();
- data = (Bucket)
bucketsByName.remove(name);
- }
- String mimeType =
DefaultMIMETypes.guessMIMEType(name);
- Logger.minor(this, "Name: "+name+"\nBucket
size: "+data.size()+"\nGuessed MIME type: "+mimeType);
- byte[] metaByteArray;
- try {
- metaByteArray = getMetadata(name, data,
mimeType);
- } catch (InserterException e) {
- Logger.normal(this, "Error inserting
"+name+": "+e.getMessage());
- errorCodes.inc(e.getMode());
- synchronized(this) {
- errors++;
- }
- continue;
- } catch (Throwable t) {
- Logger.error(this, "Caught "+t);
-
errorCodes.inc(InserterException.INTERNAL_ERROR);
- synchronized(this) {
- errors++;
- }
- continue;
- }
- if(metaByteArray != null) {
- synchronized(namesToMetadataByteArrays)
{
-
namesToMetadataByteArrays.put(name, metaByteArray);
- }
- Logger.minor(this, "Inserted "+name);
- } else {
- Logger.normal(this, "Insert failed:
"+name);
- }
-
- }
- } finally {
- synchronized(MultiFileInserter.this) {
- finished[num] = true;
- MultiFileInserter.this.notifyAll();
- }
- }
- }
-
- }
-
- final FreenetURI targetURI;
- final HashMap bucketsByName;
- final InserterContext ctx;
- final String defaultName;
- final HashMap namesToMetadataByteArrays;
- final FailureCodeTracker errorCodes;
- private int errors;
- private final boolean[] finished;
-
- public MultiFileInserter(FreenetURI insertURI, HashMap bucketsByName,
InserterContext context, String defaultName) {
- this.targetURI = insertURI;
- this.bucketsByName = bucketsByName;
- this.ctx = context;
- this.defaultName = defaultName;
- this.namesToMetadataByteArrays = new HashMap();
- this.errorCodes = new FailureCodeTracker(true);
- if(bucketsByName.get(defaultName) == null)
- // FIXME make this an InserterException.
- throw new IllegalArgumentException();
- finished = new boolean[5];
- }
-
- public FreenetURI run() throws InserterException {
- // For each file, guess MIME type, insert it, get the metadata.
- // Then put all the metadata at once into one manifest.
- // Then return it.
-
- // FIXME scaling issues; have to keep everything in RAM...
-
- for(int j=0;j<finished.length;j++) {
- MFInserter it = new MFInserter(j);
- Thread t = new Thread(it, "Inserter #"+j);
- t.setDaemon(true);
- t.start();
- }
-
- synchronized(this) {
- while(true) {
- boolean stillRunning = false;
- for(int i=0;i<finished.length;i++) {
- if(!finished[i]) stillRunning = true;
- }
- if(!stillRunning) break;
- try {
- wait(10000);
- } catch (InterruptedException e) {
- // Impossible??
- }
- }
- }
-
- if(defaultName != null) {
- synchronized(namesToMetadataByteArrays) {
- byte[] defaultData = (byte[])
namesToMetadataByteArrays.get(defaultName);
- if(defaultData != null)
- namesToMetadataByteArrays.put("",
defaultData);
- else {
- Logger.error(this, "Default name
"+defaultName+" does not exist");
-
if(namesToMetadataByteArrays.containsKey(defaultName))
- Logger.error(this, "Default
name exists but has null bytes!");
- // It existed ... and now it doesn't?!
- throw new
InserterException(InserterException.INTERNAL_ERROR);
- }
- }
- }
-
- Metadata manifestMetadata =
Metadata.mkRedirectionManifestWithMetadata(namesToMetadataByteArrays);
-
- Bucket metadata = new
ArrayBucket(manifestMetadata.writeToByteArray());
-
- FileInserter fi = new FileInserter(ctx);
-
- InsertBlock block = new InsertBlock(metadata, null, targetURI);
-
- FreenetURI uri = fi.run(block, true, false, false, null);
-
- if(errors > 0) {
- throw new
InserterException(InserterException.FATAL_ERRORS_IN_BLOCKS, errorCodes, uri);
- }
-
- return uri;
- }
-
- private byte[] getMetadata(String name, Bucket data, String mimeType)
throws InserterException {
- FileInserter fi = new FileInserter(ctx);
- InsertBlock block = new InsertBlock(data, new
ClientMetadata(mimeType), FreenetURI.EMPTY_CHK_URI);
- ArrayBucket metaBucket = new ArrayBucket();
- FreenetURI uri;
- // FIXME make a client event and switch this to logger.log(...)
- System.out.println("Inserting "+name+" ("+data.size()+" bytes,
"+mimeType+")");
- try {
- uri = fi.run(block, false, false, false, metaBucket);
- } catch (InserterException e1) {
- if(e1.uri != null && e1.getMode() ==
InserterException.COLLISION || e1.getMode() ==
InserterException.ROUTE_NOT_FOUND || e1.getMode() ==
InserterException.ROUTE_REALLY_NOT_FOUND) {
- Logger.minor(this, "Ignoring "+e1);
- uri = e1.uri;
- } else {
- // Clear the uri.
- throw new InserterException(e1.getMode());
- }
- }
- byte[] metaByteArray;
- if(metaBucket.size() == 0) {
- // It didn't give us any metadata
- Logger.minor(this, "Did not return metadata: creating
our own");
- Metadata m = new Metadata(Metadata.SIMPLE_REDIRECT,
uri, null);
- metaByteArray = m.writeToByteArray();
- if(metaByteArray == null) throw new
NullPointerException();
- } else {
- try {
- metaByteArray =
BucketTools.toByteArray(metaBucket);
- if(metaByteArray == null) throw new
NullPointerException();
- } catch (IOException e) {
- throw new Error(e);
- }
- }
- return metaByteArray;
- }
-
-}
Added: branches/async-client/src/freenet/client/PutWaiter.java
===================================================================
--- branches/async-client/src/freenet/client/PutWaiter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/PutWaiter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -0,0 +1,57 @@
+package freenet.client;
+
+import freenet.client.async.ClientCallback;
+import freenet.client.async.ClientGetter;
+import freenet.client.async.ClientPutter;
+import freenet.keys.FreenetURI;
+import freenet.support.Logger;
+
+public class PutWaiter implements ClientCallback {
+
+ private boolean finished;
+ private boolean succeeded;
+ private FreenetURI uri;
+ private InserterException error;
+
+ public void onSuccess(FetchResult result, ClientGetter state) {
+ // Ignore
+ }
+
+ public void onFailure(FetchException e, ClientGetter state) {
+ // Ignore
+ }
+
+ public synchronized void onSuccess(ClientPutter state) {
+ succeeded = true;
+ finished = true;
+ notifyAll();
+ }
+
+ public synchronized void onFailure(InserterException e, ClientPutter
state) {
+ error = e;
+ finished = true;
+ notifyAll();
+ }
+
+ public synchronized void onGeneratedURI(FreenetURI uri, ClientPutter
state) {
+ if(this.uri == null)
+ this.uri = uri;
+ if(uri.equals(this.uri)) return;
+ Logger.error(this, "URI already set: "+this.uri+" but new URI:
"+uri, new Exception("error"));
+ }
+
+ public synchronized FreenetURI waitForCompletion() throws
InserterException {
+ while(!finished) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ }
+ if(error != null) throw error;
+ if(succeeded) return uri;
+ Logger.error(this, "Did not succeed but no error");
+ throw new InserterException(InserterException.INTERNAL_ERROR,
"Did not succeed but no error", uri);
+ }
+
+}
Deleted: branches/async-client/src/freenet/client/RetryTracker.java
===================================================================
--- branches/async-client/src/freenet/client/RetryTracker.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/RetryTracker.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,392 +0,0 @@
-package freenet.client;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Vector;
-
-import freenet.crypt.RandomSource;
-import freenet.support.Logger;
-
-/**
- * Keeps a list of SplitfileBlocks for each retry level.
- */
-public class RetryTracker {
-
- class Level {
- final int level;
- final Vector blocks;
-
- Level(int l) {
- level = l;
- blocks = new Vector();
- }
-
- /**
- * Return a random block.
- * Call synchronized on RetryTracker.
- */
- StartableSplitfileBlock getBlock() {
- int len = blocks.size();
- int x = random.nextInt(len);
- StartableSplitfileBlock block =
(StartableSplitfileBlock) blocks.remove(x);
- if(blocks.isEmpty())
- removeLevel(level);
- return block;
- }
-
- void add(StartableSplitfileBlock block) {
- blocks.add(block);
- }
-
- /**
- * Remove a specific block.
- * Remove self if run out of blocks.
- * Call synchronized on RetryTracker.
- */
- void remove(StartableSplitfileBlock block) {
- blocks.remove(block);
- if(blocks.isEmpty())
- removeLevel(level);
- }
- }
-
- final FailureCodeTracker fatalErrors;
- final FailureCodeTracker nonfatalErrors;
- final HashMap levels;
- final RandomSource random;
- final int maxLevel;
- final HashSet failedBlocksTooManyRetries;
- final HashSet failedBlocksFatalErrors;
- final HashSet runningBlocks;
- final HashSet succeededBlocks;
- private int curMaxLevel;
- private int curMinLevel;
- /** Maximum number of concurrently running threads */
- final int maxThreads;
- /** After we have successes on this many blocks, we should terminate,
- * even if there are threads running and blocks queued. */
- final int targetSuccesses;
- final boolean killOnFatalError;
- private boolean killed;
- private boolean finishOnEmpty;
- private final RetryTrackerCallback callback;
- private boolean callOnProgress = false;
-
- /**
- * Create a RetryTracker.
- * @param maxLevel The maximum number of tries for each block.
- * @param random The random number source.
- * @param maxThreads The maximum number of threads to use.
- * @param killOnFatalError Whether to terminate the tracker when a fatal
- * error occurs on a single block.
- * @param cb The callback to call .finish(...) when we no longer have
- * anything to do *and* the client has set the finish on empty flag.
- */
- public RetryTracker(int maxLevel, int targetSuccesses, RandomSource
random, int maxThreads, boolean killOnFatalError, RetryTrackerCallback cb,
boolean isInsert) {
- levels = new HashMap();
- fatalErrors = new FailureCodeTracker(isInsert);
- nonfatalErrors = new FailureCodeTracker(isInsert);
- this.targetSuccesses = targetSuccesses;
- this.maxLevel = maxLevel;
- this.random = random;
- curMaxLevel = curMinLevel = 0;
- failedBlocksTooManyRetries = new HashSet();
- failedBlocksFatalErrors = new HashSet();
- runningBlocks = new HashSet();
- succeededBlocks = new HashSet();
- this.maxThreads = maxThreads;
- this.killOnFatalError = killOnFatalError;
- this.finishOnEmpty = false;
- this.callback = cb;
- }
-
- /**
- * Set the finish-on-empty flag to true.
- * This means that when there are no longer any blocks to process, and
there
- * are none running, the tracker will call the client's finish(...)
method.
- */
- public synchronized void setFinishOnEmpty() {
- finishOnEmpty = true;
- }
-
- /** Remove a level */
- private synchronized void removeLevel(int level) {
- Logger.minor(this, "Removing level "+level);
- Integer x = new Integer(level);
- levels.remove(x);
- if(curMinLevel == level) {
- for(int i=curMinLevel;i<=curMaxLevel;i++) {
- x = new Integer(i);
- if(levels.get(x) != null) {
- curMinLevel = i;
- return;
- }
- }
- curMinLevel = curMaxLevel = 0;
- return;
- }
- if(curMaxLevel == level) {
- for(int i=curMaxLevel;i>=curMinLevel;i--) {
- x = new Integer(i);
- if(levels.get(x) != null) {
- curMaxLevel = i;
- return;
- }
- }
- curMinLevel = curMaxLevel = 0;
- return;
- }
- }
-
- /** Add a level */
- private synchronized Level addLevel(int level, Integer x) {
- Logger.minor(this, "Adding level "+level);
- if(level < 0) throw new IllegalArgumentException();
- Level l = new Level(level);
- if(levels.isEmpty()) {
- curMaxLevel = curMinLevel = level;
- } else {
- if(level > curMaxLevel) curMaxLevel = level;
- if(level < curMinLevel) curMinLevel = level;
- }
- levels.put(x, l);
- return l;
- }
-
- /** Get an existing level, or add one if necessary */
- private synchronized Level makeLevel(int level) {
- Integer x = new Integer(level);
- Level l = (Level) levels.get(x);
- if(l == null) {
- return addLevel(level, x);
- }
- else return l;
- }
-
- /**
- * Add a block at retry level zero.
- */
- public synchronized void addBlock(StartableSplitfileBlock block) {
- if(killed) return;
- Level l = makeLevel(0);
- l.add(block);
- maybeStart(true);
- }
-
- /**
- * A block got a nonfatal error and should be retried.
- * Move it out of the running list and back into the relevant list,
unless
- * we have run out of retries.
- */
- public void nonfatalError(StartableSplitfileBlock block, int
reasonCode) {
- synchronized(this) {
- nonfatalErrors.inc(reasonCode);
- runningBlocks.remove(block);
- int levelNumber = block.getRetryCount();
- levelNumber++;
- Logger.minor(this, "Non-fatal error on "+block+" ->
"+levelNumber);
- if(levelNumber > maxLevel) {
- failedBlocksTooManyRetries.add(block);
- Logger.minor(this, "Finished with "+block);
- } else {
- Level newLevel = makeLevel(levelNumber);
- newLevel.add(block);
- }
- }
- maybeStart(false);
- if(callOnProgress)
- callback.onProgress();
- }
-
- /**
- * A block got a fatal error and should not be retried.
- * Move it into the fatal error list.
- * @param reasonCode A client-specific code indicating the type of
failure.
- */
- public void fatalError(StartableSplitfileBlock block, int reasonCode) {
- synchronized(this) {
- fatalErrors.inc(reasonCode);
- runningBlocks.remove(block);
- failedBlocksFatalErrors.add(block);
- }
- maybeStart(false);
- if(callOnProgress)
- callback.onProgress();
- }
-
- /**
- * If we can start some blocks, start some blocks.
- * Otherwise if we are finished, call the callback's finish method.
- */
- public void maybeStart(boolean cantCallFinished) {
- boolean callFinished = false;
- synchronized(this) {
- if(killed) return;
- Logger.minor(this, "succeeded: "+succeededBlocks.size()+",
target: "+targetSuccesses+
- ", failed:
"+failedBlocksTooManyRetries.size()+", fatal: "+failedBlocksFatalErrors.size()+
- ", running: "+runningBlocks.size()+", levels:
"+levels.size()+"("+curMinLevel+"-"+curMaxLevel+
- "), finishOnEmpty: "+finishOnEmpty+" for
"+callback);
- if(runningBlocks.size() == 1)
- Logger.minor(this, "Only block running:
"+runningBlocks.toArray()[0]);
- else if(levels.isEmpty()) {
- for(Iterator i=runningBlocks.iterator();i.hasNext();) {
- Logger.minor(this, "Still running: "+i.next());
- }
- }
- if((succeededBlocks.size() >= targetSuccesses)
- || (runningBlocks.isEmpty() && levels.isEmpty()
&& finishOnEmpty)) {
- killed = true;
- Logger.minor(this, "Finishing");
- StartableSplitfileBlock[] running = runningBlocks();
- for(int i=0;i<running.length;i++) {
- running[i].kill();
- }
- runningBlocks.clear();
- if(!cantCallFinished)
- callFinished = true;
- else {
- Runnable r = new Runnable() { public void run()
{ callback.finished(succeededBlocks(), failedBlocks(), fatalErrorBlocks()); } };
- Thread t = new Thread(r);
- t.setDaemon(true);
- t.start();
- }
- } else {
- while(runningBlocks.size() < maxThreads) {
- StartableSplitfileBlock block = getBlock();
- if(block == null) break;
- Logger.minor(this, "Starting: "+block);
- block.start();
- runningBlocks.add(block);
- }
- }
- }
- if(callFinished)
- callback.finished(succeededBlocks(), failedBlocks(),
fatalErrorBlocks());
- }
-
- public void success(StartableSplitfileBlock block) {
- synchronized(this) {
- if(killed) return;
- runningBlocks.remove(block);
- succeededBlocks.add(block);
- }
- maybeStart(false);
- if(callOnProgress)
- callback.onProgress();
- }
-
- public synchronized void callOnProgress() {
- callOnProgress = true;
- }
-
- /**
- * Get the next block to try. This is a randomly selected block from the
- * lowest priority currently available. Move it into the running list.
- */
- public synchronized StartableSplitfileBlock getBlock() {
- if(killed) return null;
- Integer iMin = new Integer(curMinLevel);
- Level l = (Level) levels.get(iMin);
- if(l == null) {
- if(!(curMinLevel == 0 && curMaxLevel == 0))
- Logger.error(this, "min="+curMinLevel+",
max="+curMaxLevel+" but min does not exist!");
- if(!levels.isEmpty()) {
- Integer[] levelNums = (Integer[])
levels.keySet().toArray(new Integer[levels.size()]);
- java.util.Arrays.sort(levelNums);
- Integer x = levelNums[0];
- curMinLevel = x.intValue();
- Integer y = levelNums[levelNums.length-1];
- curMaxLevel = y.intValue();
- Logger.normal(this, "Corrected:
min="+curMinLevel+", max="+curMaxLevel);
- return getBlock();
- }
- else return null;
- }
- return l.getBlock();
- }
-
- /**
- * Get all running blocks.
- */
- public synchronized StartableSplitfileBlock[] runningBlocks() {
- return (StartableSplitfileBlock[])
- runningBlocks.toArray(new
StartableSplitfileBlock[runningBlocks.size()]);
- }
-
- /**
- * Get all blocks with fatal errors.
- * StartableSplitfileBlock's are assumed to remember their errors, so
we don't.
- */
- public synchronized StartableSplitfileBlock[] fatalErrorBlocks() {
- return (StartableSplitfileBlock[])
- failedBlocksFatalErrors.toArray(new
StartableSplitfileBlock[failedBlocksFatalErrors.size()]);
- }
-
- /**
- * Get all blocks which didn't succeed in the maximum number of tries.
- */
- public synchronized StartableSplitfileBlock[] failedBlocks() {
- return (StartableSplitfileBlock[])
- failedBlocksTooManyRetries.toArray(new
StartableSplitfileBlock[failedBlocksTooManyRetries.size()]);
- }
-
- /**
- * Get all successfully downloaded blocks.
- */
- public synchronized StartableSplitfileBlock[] succeededBlocks() {
- return (StartableSplitfileBlock[])
- succeededBlocks.toArray(new
StartableSplitfileBlock[succeededBlocks.size()]);
- }
-
- public synchronized int succeededBlocksLength() {
- return succeededBlocks.size();
- }
-
- /**
- * Count the number of blocks which could not be fetched because we ran
out
- * of retries.
- */
- public synchronized int countFailedBlocks() {
- return failedBlocksTooManyRetries.size();
- }
-
- /**
- * Highest number of completed retries of any block so far.
- */
- public synchronized int highestRetries() {
- return curMaxLevel;
- }
-
- /**
- * Lowest number of completed retries of any block so far.
- */
- public synchronized int lowestRetries() {
- return curMinLevel;
- }
-
- /**
- * Are there more blocks to process?
- */
- public synchronized boolean moreBlocks() {
- return !levels.isEmpty();
- }
-
- public FailureCodeTracker getAccumulatedFatalErrorCodes() {
- return fatalErrors;
- }
-
- public FailureCodeTracker getAccumulatedNonFatalErrorCodes() {
- return nonfatalErrors;
- }
-
- public synchronized void kill() {
- killed = true;
- levels.clear();
- for(Iterator i=runningBlocks.iterator();i.hasNext();) {
- StartableSplitfileBlock sb = (StartableSplitfileBlock)
i.next();
- sb.kill();
- }
- runningBlocks.clear();
- }
-}
Deleted: branches/async-client/src/freenet/client/RetryTrackerCallback.java
===================================================================
--- branches/async-client/src/freenet/client/RetryTrackerCallback.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/RetryTrackerCallback.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,21 +0,0 @@
-package freenet.client;
-
-/**
- * Object passed to RetryTracker. This is called when RetryTracker finishes.
- */
-public interface RetryTrackerCallback {
-
- /**
- * Notify the caller that we have finished.
- * @param succeeded The blocks which succeeded.
- * @param failed The blocks which failed.
- * @param fatalErrors The blocks which got fatal errors.
- */
- void finished(StartableSplitfileBlock[] succeeded,
StartableSplitfileBlock[] failed, StartableSplitfileBlock[] fatalErrors);
-
- /**
- * When a block completes etc.
- */
- void onProgress();
-
-}
Deleted: branches/async-client/src/freenet/client/Segment.java
===================================================================
--- branches/async-client/src/freenet/client/Segment.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/Segment.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,270 +0,0 @@
-package freenet.client;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.LinkedList;
-import java.util.Vector;
-
-import freenet.keys.FreenetURI;
-import freenet.support.Bucket;
-import freenet.support.BucketTools;
-import freenet.support.Logger;
-
-/**
- * A segment, within a splitfile.
- * Self-starting Runnable.
- *
- * Does not require locking, because all locking goes through the parent
Segment.
- */
-public class Segment implements RetryTrackerCallback {
-
- final short splitfileType;
- final FreenetURI[] dataBlocks;
- final FreenetURI[] checkBlocks;
- final BlockFetcher[] dataBlockStatus;
- final BlockFetcher[] checkBlockStatus;
- final int minFetched;
- final SplitFetcher parentFetcher;
- final ArchiveContext archiveContext;
- final FetcherContext fetcherContext;
- final long maxBlockLength;
- final boolean nonFullBlocksAllowed;
- /** Has the segment started to do something? Irreversible. */
- private boolean started;
- /** Has the segment finished processing? Irreversible. */
- private boolean finished;
- /** Bucket to store the data retrieved, after it has been decoded */
- private Bucket decodedData;
- /** Fetch context for block fetches */
- final FetcherContext blockFetchContext;
- /** Recursion level */
- final int recursionLevel;
- /** Retry tracker */
- private final RetryTracker tracker;
- private FetchException failureException;
-
- /**
- * Create a Segment.
- * @param splitfileType The type of the splitfile.
- * @param splitfileDataBlocks The data blocks to fetch.
- * @param splitfileCheckBlocks The check blocks to fetch.
- */
- public Segment(short splitfileType, FreenetURI[] splitfileDataBlocks,
FreenetURI[] splitfileCheckBlocks,
- SplitFetcher fetcher, ArchiveContext actx,
FetcherContext fctx, long maxTempLength, boolean useLengths, int recLevel)
throws MetadataParseException {
- this.splitfileType = splitfileType;
- dataBlocks = splitfileDataBlocks;
- checkBlocks = splitfileCheckBlocks;
- if(splitfileType == Metadata.SPLITFILE_NONREDUNDANT) {
- minFetched = dataBlocks.length;
- } else if(splitfileType == Metadata.SPLITFILE_ONION_STANDARD) {
- minFetched = dataBlocks.length;
- } else throw new MetadataParseException("Unknown splitfile
type"+splitfileType);
- tracker = new RetryTracker(fctx.maxSplitfileBlockRetries,
splitfileDataBlocks.length, fctx.random, fctx.maxSplitfileThreads, false, this,
false);
- // Don't add blocks to tracker yet, because don't want to start
fetch yet.
- parentFetcher = fetcher;
- archiveContext = actx;
- fetcherContext = fctx;
- maxBlockLength = maxTempLength;
- nonFullBlocksAllowed = useLengths;
- started = false;
- finished = false;
- decodedData = null;
- dataBlockStatus = new BlockFetcher[dataBlocks.length];
- checkBlockStatus = new BlockFetcher[checkBlocks.length];
- Vector firstSet = new
Vector(dataBlocks.length+checkBlocks.length);
- for(int i=0;i<dataBlocks.length;i++) {
- dataBlockStatus[i] = new BlockFetcher(this, tracker,
dataBlocks[i], i, fctx.dontEnterImplicitArchives);
- firstSet.add(dataBlockStatus[i]);
- }
- for(int i=0;i<checkBlocks.length;i++) {
- checkBlockStatus[i] = new BlockFetcher(this, tracker,
checkBlocks[i], dataBlockStatus.length + i, fctx.dontEnterImplicitArchives);
- firstSet.add(checkBlockStatus[i]);
- }
- // FIXME be a bit more flexible here depending on flags
- if(useLengths) {
- blockFetchContext = new FetcherContext(fetcherContext,
FetcherContext.SPLITFILE_USE_LENGTHS_MASK);
- this.recursionLevel = recLevel;
- } else {
- blockFetchContext = new FetcherContext(fetcherContext,
FetcherContext.SPLITFILE_DEFAULT_BLOCK_MASK);
- this.recursionLevel = 0;
- }
- Logger.minor(this, "Created segment: data blocks:
"+dataBlocks.length+", check blocks: "+checkBlocks.length+" "+this);
- }
-
- /**
- * Is the segment finished? (Either error or fetched and decoded)?
- */
- public boolean isFinished() {
- return finished;
- }
-
- /**
- * If there was an error, throw it now.
- */
- public void throwError() throws FetchException {
- if(failureException != null)
- throw failureException;
- }
-
- /**
- * Return the length of the data, after decoding.
- */
- public long decodedLength() {
- return decodedData.size();
- }
-
- /**
- * Write the decoded data to the given output stream.
- * Do not write more than the specified number of bytes (unless it is
negative,
- * in which case ignore it).
- * @return The number of bytes written.
- * @throws IOException If there was an error reading from the bucket
the data is
- * stored in, or writing to the stream provided.
- */
- public long writeDecodedDataTo(OutputStream os, long truncateLength)
throws IOException {
- long len = decodedData.size();
- if(truncateLength >= 0 && truncateLength < len)
- len = truncateLength;
- BucketTools.copyTo(decodedData, os, Math.min(truncateLength,
decodedData.size()));
- return len;
- }
-
- /**
- * Return true if the Segment has been started, otherwise false.
- */
- public boolean isStarted() {
- return started;
- }
-
- /**
- * Start the Segment fetching the data. When it has finished fetching,
it will
- * notify the SplitFetcher. Note that we must not start fetching until
this
- * method is called, because of the requirement to not fetch all
segments
- * simultaneously.
- */
- public void start() {
- started = true;
- for(int i=0;i<dataBlockStatus.length;i++) {
- tracker.addBlock(dataBlockStatus[i]);
- }
- Logger.minor(this, "Added data blocks");
- for(int i=0;i<checkBlockStatus.length;i++) {
- tracker.addBlock(checkBlockStatus[i]);
- }
- tracker.callOnProgress();
- tracker.setFinishOnEmpty();
- }
-
- /**
- * Once we have enough data to decode, tell parent, and decode it.
- */
- public void finished(StartableSplitfileBlock[] succeeded,
StartableSplitfileBlock[] failed, StartableSplitfileBlock[] fatalErrors) {
-
- Logger.minor(this, "Finished("+succeeded.length+",
"+failed.length+", "+fatalErrors.length+")");
- parentFetcher.gotBlocks(this);
- if(succeeded.length >= minFetched)
- // Not finished yet, need to decode
- try {
- successfulFetch();
- } catch (Throwable t) {
- Logger.error(this, "Caught "+t+" decoding
"+this);
- finished = true;
- failureException = new
FetchException(FetchException.INTERNAL_ERROR, t);
- parentFetcher.segmentFinished(this);
- }
- else {
- failureException = new
SplitFetchException(failed.length, fatalErrors.length, succeeded.length,
minFetched,
tracker.getAccumulatedNonFatalErrorCodes().merge(tracker.getAccumulatedFatalErrorCodes()));
- finished = true;
- parentFetcher.segmentFinished(this);
- }
- }
-
- /**
- * Successful fetch, do the decode, tell the parent, etc.
- */
- private void successfulFetch() {
-
- // Now decode
- Logger.minor(this, "Decoding "+this);
-
- FECCodec codec = FECCodec.getCodec(splitfileType,
dataBlocks.length, checkBlocks.length);
- try {
- if(splitfileType != Metadata.SPLITFILE_NONREDUNDANT) {
- // FIXME hardcoded block size below.
- codec.decode(dataBlockStatus, checkBlockStatus,
32768, fetcherContext.bucketFactory);
- // Now have all the data blocks (not
necessarily all the check blocks)
- }
-
- decodedData =
fetcherContext.bucketFactory.makeBucket(-1);
- Logger.minor(this, "Copying data from data blocks");
- OutputStream os = decodedData.getOutputStream();
- for(int i=0;i<dataBlockStatus.length;i++) {
- BlockFetcher status = dataBlockStatus[i];
- Bucket data = status.fetchedData;
- BucketTools.copyTo(data, os, Long.MAX_VALUE);
- }
- Logger.minor(this, "Copied data");
- os.close();
- // Must set finished BEFORE calling parentFetcher.
- // Otherwise a race is possible that might result in it
not seeing our finishing.
- finished = true;
- parentFetcher.segmentFinished(this);
- } catch (IOException e) {
- Logger.minor(this, "Caught bucket error?: "+e, e);
- finished = true;
- failureException = new
FetchException(FetchException.BUCKET_ERROR);
- parentFetcher.segmentFinished(this);
- return;
- }
-
- // Now heal
-
- // Encode any check blocks we don't have
- if(codec != null) {
- try {
- codec.encode(dataBlockStatus, checkBlockStatus,
32768, fetcherContext.bucketFactory);
- } catch (IOException e) {
- Logger.error(this, "Bucket error while healing:
"+e, e);
- }
- }
-
- // Now insert *ALL* blocks on which we had at least one
failure, and didn't eventually succeed
- for(int i=0;i<dataBlockStatus.length;i++) {
- BlockFetcher block = dataBlockStatus[i];
- if(block.actuallyFetched) continue;
- if(block.completedTries == 0) {
- // 80% chance of not inserting, if we never
tried it
- if(fetcherContext.random.nextInt(5) == 0)
continue;
- }
- block.queueHeal();
- }
-
- // FIXME heal check blocks too
- }
-
- public void onProgress() {
- if(fetcherContext.isCancelled()) {
- finished = true;
- tracker.kill();
- failureException = new
FetchException(FetchException.CANCELLED);
- parentFetcher.gotBlocks(this);
- }
- parentFetcher.onProgress();
- }
-
- public int fetchedBlocks() {
- return tracker.succeededBlocksLength();
- }
-
- public int failedBlocks() {
- return tracker.failedBlocks().length;
- }
-
- public int fatallyFailedBlocks() {
- return tracker.fatalErrorBlocks().length;
- }
-
- public int runningBlocks() {
- return tracker.runningBlocks().length;
- }
-}
Deleted: branches/async-client/src/freenet/client/SplitFetcher.java
===================================================================
--- branches/async-client/src/freenet/client/SplitFetcher.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/SplitFetcher.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,244 +0,0 @@
-package freenet.client;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Vector;
-
-import freenet.client.events.SplitfileProgressEvent;
-import freenet.keys.FreenetURI;
-import freenet.keys.NodeCHK;
-import freenet.support.Bucket;
-import freenet.support.Fields;
-import freenet.support.Logger;
-
-/**
- * Class to fetch a splitfile.
- */
-public class SplitFetcher {
-
- /** The splitfile type. See the SPLITFILE_ constants on Metadata. */
- final short splitfileType;
- /** The segment length. -1 means not segmented and must get everything
to decode. */
- final int blocksPerSegment;
- /** The segment length in check blocks. */
- final int checkBlocksPerSegment;
- /** Total number of segments */
- final int segmentCount;
- /** The detailed information on each segment */
- final Segment[] segments;
- /** The splitfile data blocks. */
- final FreenetURI[] splitfileDataBlocks;
- /** The splitfile check blocks. */
- final FreenetURI[] splitfileCheckBlocks;
- /** The archive context */
- final ArchiveContext actx;
- /** The fetch context */
- final FetcherContext fctx;
- /** Maximum temporary length */
- final long maxTempLength;
- /** Have all segments finished? Access synchronized. */
- private boolean allSegmentsFinished = false;
- /** Currently fetching segment */
- private Segment fetchingSegment;
- /** Array of unstarted segments. Modify synchronized. */
- private Vector unstartedSegments;
- /** Override length. If this is positive, truncate the splitfile to
this length. */
- private long overrideLength;
- /** Accept non-full splitfile chunks? */
- private boolean splitUseLengths;
-
- public SplitFetcher(Metadata metadata, ArchiveContext archiveContext,
FetcherContext ctx, int recursionLevel) throws MetadataParseException,
FetchException {
- actx = archiveContext;
- fctx = ctx;
- if(fctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- overrideLength = metadata.dataLength;
- this.maxTempLength = ctx.maxTempLength;
- splitfileType = metadata.getSplitfileType();
- splitfileDataBlocks = metadata.getSplitfileDataKeys();
- splitfileCheckBlocks = metadata.getSplitfileCheckKeys();
- splitUseLengths = metadata.splitUseLengths;
- int blockLength = splitUseLengths ? -1 : NodeCHK.BLOCK_SIZE;
- if(splitfileType == Metadata.SPLITFILE_NONREDUNDANT) {
- // Don't need to do much - just fetch everything and
piece it together.
- blocksPerSegment = -1;
- checkBlocksPerSegment = -1;
- segmentCount = 1;
- } else if(splitfileType == Metadata.SPLITFILE_ONION_STANDARD) {
- byte[] params = metadata.splitfileParams;
- if(params == null || params.length < 8)
- throw new MetadataParseException("No splitfile
params");
- blocksPerSegment = Fields.bytesToInt(params, 0);
- checkBlocksPerSegment = Fields.bytesToInt(params, 4);
- if(blocksPerSegment > ctx.maxDataBlocksPerSegment
- || checkBlocksPerSegment >
ctx.maxCheckBlocksPerSegment)
- throw new
FetchException(FetchException.TOO_MANY_BLOCKS_PER_SEGMENT, "Too many blocks per
segment: "+blocksPerSegment+" data, "+checkBlocksPerSegment+" check");
- segmentCount = (splitfileDataBlocks.length /
blocksPerSegment) +
- (splitfileDataBlocks.length % blocksPerSegment
== 0 ? 0 : 1);
- // Onion, 128/192.
- // Will be segmented.
- } else throw new MetadataParseException("Unknown splitfile
format: "+splitfileType);
- Logger.minor(this, "Algorithm: "+splitfileType+", blocks per
segment: "+blocksPerSegment+", check blocks per segment:
"+checkBlocksPerSegment+", segments: "+segmentCount);
- segments = new Segment[segmentCount]; // initially null on all
entries
- if(segmentCount == 1) {
- segments[0] = new Segment(splitfileType,
splitfileDataBlocks, splitfileCheckBlocks, this, archiveContext, ctx,
maxTempLength, splitUseLengths, recursionLevel+1);
- } else {
- int dataBlocksPtr = 0;
- int checkBlocksPtr = 0;
- for(int i=0;i<segments.length;i++) {
- // Create a segment. Give it its keys.
- int copyDataBlocks =
Math.min(splitfileDataBlocks.length - dataBlocksPtr, blocksPerSegment);
- int copyCheckBlocks =
Math.min(splitfileCheckBlocks.length - checkBlocksPtr, checkBlocksPerSegment);
- FreenetURI[] dataBlocks = new
FreenetURI[copyDataBlocks];
- FreenetURI[] checkBlocks = new
FreenetURI[copyCheckBlocks];
- if(copyDataBlocks > 0)
- System.arraycopy(splitfileDataBlocks,
dataBlocksPtr, dataBlocks, 0, copyDataBlocks);
- if(copyCheckBlocks > 0)
- System.arraycopy(splitfileCheckBlocks,
checkBlocksPtr, checkBlocks, 0, copyCheckBlocks);
- dataBlocksPtr += copyDataBlocks;
- checkBlocksPtr += copyCheckBlocks;
- segments[i] = new Segment(splitfileType,
dataBlocks, checkBlocks, this, archiveContext, ctx, maxTempLength,
splitUseLengths, blockLength);
- }
- }
- unstartedSegments = new Vector();
- for(int i=0;i<segments.length;i++)
- unstartedSegments.add(segments[i]);
- Logger.minor(this, "Segments: "+unstartedSegments.size()+",
data keys: "+splitfileDataBlocks.length+", check keys:
"+(splitfileCheckBlocks==null?0:splitfileCheckBlocks.length));
- }
-
- /**
- * Fetch the splitfile.
- * Fetch one segment, while decoding the previous one.
- * Fetch the segments in random order.
- * When everything has been fetched and decoded, return the full data.
- * @throws FetchException
- */
- public Bucket fetch() throws FetchException {
- /*
- * While(true) {
- * Pick a random segment, start it fetching.
- * Wait for a segment to finish fetching, a segment to
finish decoding, or an error.
- * If a segment finishes fetching:
- * Continue to start another one if there are any
left
- * If a segment finishes decoding:
- * If all segments are decoded, assemble all the
segments and return the data.
- *
- * Segments are expected to automatically start decoding when
they finish fetching,
- * but to tell us either way.
- */
- while(true) {
- synchronized(this) {
- if(fctx.isCancelled()) throw new
FetchException(FetchException.CANCELLED);
- if(fetchingSegment == null) {
- // Pick a random segment
- fetchingSegment =
chooseUnstartedSegment();
- if(fetchingSegment == null) {
- // All segments have started
- } else {
- fetchingSegment.start();
- }
- }
- if(allSegmentsFinished) {
- return finalStatus();
- }
- try {
- wait(10*1000); // or wait()?
- } catch (InterruptedException e) {
- // Ignore
- }
- }
- }
- }
-
- private Segment chooseUnstartedSegment() {
- synchronized(unstartedSegments) {
- if(unstartedSegments.isEmpty()) return null;
- int x = fctx.random.nextInt(unstartedSegments.size());
- Logger.minor(this, "Starting segment "+x+" of
"+unstartedSegments.size());
- Segment s = (Segment) unstartedSegments.remove(x);
- return s;
- }
- }
-
- /** Return the final status of the fetch. Throws an exception, or
returns a
- * Bucket containing the fetched data.
- * @throws FetchException If the fetch failed for some reason.
- */
- private Bucket finalStatus() throws FetchException {
- long finalLength = 0;
- for(int i=0;i<segments.length;i++) {
- Segment s = segments[i];
- if(!s.isFinished()) throw new
IllegalStateException("Not all finished");
- s.throwError();
- // If still here, it succeeded
- finalLength += s.decodedLength();
- // Healing is done by Segment
- }
- if(finalLength > overrideLength)
- finalLength = overrideLength;
-
- long bytesWritten = 0;
- OutputStream os = null;
- Bucket output;
- try {
- output = fctx.bucketFactory.makeBucket(finalLength);
- os = output.getOutputStream();
- for(int i=0;i<segments.length;i++) {
- Segment s = segments[i];
- long max = (finalLength < 0 ? 0 : (finalLength
- bytesWritten));
- bytesWritten += s.writeDecodedDataTo(os, max);
- }
- } catch (IOException e) {
- throw new FetchException(FetchException.BUCKET_ERROR,
e);
- } finally {
- if(os != null) {
- try {
- os.close();
- } catch (IOException e) {
- // If it fails to close it may return
corrupt data.
- throw new
FetchException(FetchException.BUCKET_ERROR, e);
- }
- }
- }
- return output;
- }
-
- public void gotBlocks(Segment segment) {
- Logger.minor(this, "Got blocks for segment: "+segment);
- synchronized(this) {
- fetchingSegment = null;
- notifyAll();
- }
- }
-
- public void segmentFinished(Segment segment) {
- Logger.minor(this, "Finished segment: "+segment);
- synchronized(this) {
- boolean allDone = true;
- for(int i=0;i<segments.length;i++)
- if(!segments[i].isFinished()) {
- Logger.minor(this, "Segment
"+segments[i]+" is not finished");
- allDone = false;
- }
- if(allDone) allSegmentsFinished = true;
- notifyAll();
- }
- }
-
- public void onProgress() {
- int totalBlocks = splitfileDataBlocks.length;
- int fetchedBlocks = 0;
- int failedBlocks = 0;
- int fatallyFailedBlocks = 0;
- int runningBlocks = 0;
- for(int i=0;i<segments.length;i++) {
- Logger.minor(this, "Segment: "+segments[i]+":
fetched="+segments[i].fetchedBlocks()+", failedBlocks:
"+segments[i].failedBlocks()+
- ", fatally:
"+segments[i].fatallyFailedBlocks()+", running: "+segments[i].runningBlocks());
- fetchedBlocks += segments[i].fetchedBlocks();
- failedBlocks += segments[i].failedBlocks();
- fatallyFailedBlocks +=
segments[i].fatallyFailedBlocks();
- runningBlocks += segments[i].runningBlocks();
- }
- fctx.eventProducer.produceEvent(new
SplitfileProgressEvent(totalBlocks, fetchedBlocks, failedBlocks,
fatallyFailedBlocks, runningBlocks));
- }
-
-}
Deleted: branches/async-client/src/freenet/client/SplitInserter.java
===================================================================
--- branches/async-client/src/freenet/client/SplitInserter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/SplitInserter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,302 +0,0 @@
-package freenet.client;
-
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.Vector;
-
-import freenet.client.events.GeneratedURIEvent;
-import freenet.client.events.SplitfileProgressEvent;
-import freenet.keys.FreenetURI;
-import freenet.keys.NodeCHK;
-import freenet.support.Bucket;
-import freenet.support.BucketTools;
-import freenet.support.Logger;
-import freenet.support.compress.Compressor;
-
-/**
- * Insert a splitfile.
- */
-public class SplitInserter implements RetryTrackerCallback {
-
- final Bucket origData;
- final long dataLength;
- final ClientMetadata clientMetadata;
- final short compressionCodec;
- final short splitfileAlgorithm;
- final InserterContext ctx;
- final RetryTracker tracker;
- final int segmentSize;
- final int checkSegmentSize;
- final int blockSize;
- final boolean isMetadata;
- final Bucket returnMetadata;
- StartableSplitfileBlock[] origDataBlocks;
- InsertSegment encodingSegment;
- InsertSegment[] segments;
- private boolean finishedInserting = false;
- private boolean getCHKOnly;
- private int succeeded;
- private int failed;
- private int fatalErrors;
- private int countCheckBlocks;
- private StartableSplitfileBlock[] fatalErrorBlocks;
- private FileInserter inserter;
-
- /**
- * @param returnMetadata If not null, then write the metadata to this
bucket,
- * rather than inserting it.
- */
- public SplitInserter(Bucket data, ClientMetadata clientMetadata,
Compressor compressor, short splitfileAlgorithm, InserterContext ctx,
FileInserter inserter, int blockLength, boolean getCHKOnly, boolean isMetadata,
Bucket returnMetadata) throws InserterException {
- this.origData = data;
- this.getCHKOnly = getCHKOnly;
- this.blockSize = blockLength;
- this.clientMetadata = clientMetadata;
- if(compressor == null)
- compressionCodec = -1;
- else
- compressionCodec = compressor.codecNumberForMetadata();
- this.splitfileAlgorithm = splitfileAlgorithm;
- this.ctx = ctx;
- this.dataLength = data.size();
- segmentSize = ctx.splitfileSegmentDataBlocks;
- checkSegmentSize = splitfileAlgorithm ==
Metadata.SPLITFILE_NONREDUNDANT ? 0 : ctx.splitfileSegmentCheckBlocks;
- tracker = new RetryTracker(ctx.maxInsertRetries,
Integer.MAX_VALUE, ctx.random, ctx.maxSplitInsertThreads, true, this, true);
- try {
- splitIntoBlocks();
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, e, null);
- }
- this.inserter = inserter;
- this.isMetadata = isMetadata;
- this.returnMetadata = returnMetadata;
- }
-
- /**
- * Inserts the splitfile.
- * @return The URI of the resulting file.
- * @throws InserterException If we are not able to insert the splitfile.
- */
- public FreenetURI run() throws InserterException {
- try {
- startInsertingDataBlocks();
- splitIntoSegments(segmentSize);
- // Backwards, because the last is the shortest
- try {
- for(int i=segments.length-1;i>=0;i--) {
- encodeSegment(i, origDataBlocks.length
+ checkSegmentSize * i);
- Logger.minor(this, "Encoded segment
"+i+" of "+segments.length);
- }
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, e, null);
- }
- // Wait for the insertion thread to finish
- return waitForCompletion();
- } catch (Throwable t) {
- Logger.error(this, "Caught "+t, t);
- tracker.kill();
- if(t instanceof InserterException) throw
(InserterException)t;
- throw new
InserterException(InserterException.INTERNAL_ERROR, t, null);
- }
- }
-
- private FreenetURI waitForCompletion() throws InserterException {
- tracker.setFinishOnEmpty();
- synchronized(this) {
- while(!finishedInserting) {
- try {
- wait(10*1000);
- } catch (InterruptedException e) {
- // Ignore
- }
- }
- }
-
- // Create the manifest (even if we failed, so that the key is
visible)
-
- FreenetURI[] dataURIs = getDataURIs();
- FreenetURI[] checkURIs = getCheckURIs();
-
- Logger.minor(this, "Data URIs: "+dataURIs.length+", check URIs:
"+checkURIs.length);
-
- boolean missingURIs = anyNulls(dataURIs) || anyNulls(checkURIs);
-
- if(missingURIs && fatalErrors == 0 && failed == 0)
- throw new IllegalStateException();
-
- FreenetURI uri = null;
-
- if(!missingURIs) {
-
- Metadata metadata = new Metadata(splitfileAlgorithm,
dataURIs, checkURIs, segmentSize, checkSegmentSize, clientMetadata, dataLength,
compressionCodec, isMetadata);
-
- if(returnMetadata != null) {
- DataOutputStream dos;
- try {
- dos = new
DataOutputStream(returnMetadata.getOutputStream());
- metadata.writeTo(dos);
- dos.close();
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR);
- }
- } else {
-
- Bucket mbucket;
- try {
- mbucket =
BucketTools.makeImmutableBucket(ctx.bf, metadata.writeToByteArray());
- } catch (IOException e) {
- throw new
InserterException(InserterException.BUCKET_ERROR, null);
- }
-
- if(inserter == null)
- inserter = new FileInserter(ctx);
-
- InsertBlock mblock = new InsertBlock(mbucket,
null, FreenetURI.EMPTY_CHK_URI);
-
- // FIXME probably should uncomment below so it
doesn't get inserted at all?
- // FIXME this is a hack for small network
support... but we will need that IRL... hmmm
- try {
- uri = inserter.run(mblock, true,
getCHKOnly/* || (fatalErrors > 0 || failed > 0)*/, false, null);
- } catch (InserterException e) {
- e.errorCodes =
tracker.getAccumulatedNonFatalErrorCodes().merge(tracker.getAccumulatedFatalErrorCodes());
- throw e;
- }
- }
-
-
- }
- // Did we succeed?
-
- if(uri != null)
- ctx.eventProducer.produceEvent(new
GeneratedURIEvent(uri));
-
- if(fatalErrors > 0) {
- throw new
InserterException(InserterException.FATAL_ERRORS_IN_BLOCKS,
tracker.getAccumulatedFatalErrorCodes(), uri);
- }
-
- if(failed > 0) {
- throw new
InserterException(InserterException.TOO_MANY_RETRIES_IN_BLOCKS,
tracker.getAccumulatedNonFatalErrorCodes(), uri);
- }
-
- return uri;
- }
-
- // FIXME move this to somewhere
- private static boolean anyNulls(Object[] array) {
- for(int i=0;i<array.length;i++)
- if(array[i] == null) return true;
- return false;
- }
-
- private FreenetURI[] getCheckURIs() {
- // Copy check blocks from each segment into a FreenetURI[].
- FreenetURI[] uris = new FreenetURI[countCheckBlocks];
- int x = 0;
- for(int i=0;i<segments.length;i++) {
- FreenetURI[] segURIs = segments[i].getCheckURIs();
- if(x + segURIs.length > countCheckBlocks)
- throw new IllegalStateException("x="+x+",
segURIs="+segURIs.length+", countCheckBlocks="+countCheckBlocks);
- System.arraycopy(segURIs, 0, uris, x, segURIs.length);
- x += segURIs.length;
- }
-
- if(uris.length != x)
- throw new IllegalStateException("Total is wrong");
-
- return uris;
- }
-
- private FreenetURI[] getDataURIs() {
- FreenetURI[] uris = new FreenetURI[origDataBlocks.length];
- for(int i=0;i<uris.length;i++)
- uris[i] = origDataBlocks[i].getURI();
- return uris;
- }
-
- private int encodeSegment(int i, int offset) throws IOException {
- encodingSegment = segments[i];
- return encodingSegment.encode(offset, tracker, ctx);
- }
-
- /**
- * Start the insert, by adding all the data blocks.
- */
- private void startInsertingDataBlocks() {
- for(int i=0;i<origDataBlocks.length;i++)
- tracker.addBlock(origDataBlocks[i]);
- tracker.callOnProgress();
- }
-
- /**
- * Split blocks into segments for encoding.
- * @throws IOException If there is a bucket error encoding the file.
- */
- private void splitIntoBlocks() throws IOException {
- Bucket[] dataBuckets = BucketTools.split(origData,
NodeCHK.BLOCK_SIZE, ctx.bf);
- origDataBlocks = new
StartableSplitfileBlock[dataBuckets.length];
- for(int i=0;i<origDataBlocks.length;i++) {
- origDataBlocks[i] = new BlockInserter(dataBuckets[i],
i, tracker, ctx, getCHKOnly);
- if(origDataBlocks[i].getData() == null)
- throw new NullPointerException("Block "+i+" of
"+dataBuckets.length+" is null");
- }
- }
-
- /**
- * Group the blocks into segments.
- */
- private void splitIntoSegments(int segmentSize) {
- int dataBlocks = origDataBlocks.length;
-
- Vector segs = new Vector();
-
- // First split the data up
- if(dataBlocks < segmentSize || segmentSize == -1) {
- // Single segment
- InsertSegment onlySeg = new
InsertSegment(splitfileAlgorithm, origDataBlocks, blockSize, ctx.bf,
getCHKOnly, 0);
- countCheckBlocks = onlySeg.checkBlocks.length;
- segs.add(onlySeg);
- } else {
- int j = 0;
- int segNo = 0;
- for(int i=segmentSize;;i+=segmentSize) {
- if(i > dataBlocks) i = dataBlocks;
- StartableSplitfileBlock[] seg = new
StartableSplitfileBlock[i-j];
- System.arraycopy(origDataBlocks, j, seg, 0,
i-j);
- j = i;
- for(int x=0;x<seg.length;x++)
- if(seg[x].getData() == null) throw new
NullPointerException("In splitIntoSegs: "+x+" is null of "+seg.length+" of
"+segNo);
- InsertSegment s = new
InsertSegment(splitfileAlgorithm, seg, blockSize, ctx.bf, getCHKOnly, segNo);
- countCheckBlocks += s.checkBlocks.length;
- segs.add(s);
-
- if(i == dataBlocks) break;
- segNo++;
- }
- }
- segments = (InsertSegment[]) segs.toArray(new
InsertSegment[segs.size()]);
- }
-
- public void finished(StartableSplitfileBlock[] succeeded,
StartableSplitfileBlock[] failed, StartableSplitfileBlock[] fatalErrors) {
- synchronized(this) {
- finishedInserting = true;
- this.succeeded = succeeded.length;
- this.failed = failed.length;
- this.fatalErrorBlocks = fatalErrors;
- this.fatalErrors = fatalErrorBlocks.length;
- notify();
- }
- }
-
- public void onProgress() {
- /* What info to report?
- * - Total number of blocks to insert.
- * -
- */
- int totalBlocks = origDataBlocks.length + countCheckBlocks;
- int fetchedBlocks = tracker.succeededBlocks().length;
- int failedBlocks = tracker.countFailedBlocks();
- int fatallyFailedBlocks = tracker.fatalErrorBlocks().length;
- int runningBlocks = tracker.runningBlocks().length;
- ctx.eventProducer.produceEvent(new
SplitfileProgressEvent(totalBlocks, fetchedBlocks, failedBlocks,
fatallyFailedBlocks, runningBlocks));
- }
-
-}
Deleted: branches/async-client/src/freenet/client/StdSplitfileBlock.java
===================================================================
--- branches/async-client/src/freenet/client/StdSplitfileBlock.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/StdSplitfileBlock.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,54 +0,0 @@
-package freenet.client;
-
-import freenet.support.Bucket;
-import freenet.support.Logger;
-
-public abstract class StdSplitfileBlock implements StartableSplitfileBlock ,
Runnable {
-
- Bucket fetchedData;
- protected final RetryTracker tracker;
- /** Splitfile index - [0,k[ is the data blocks, [k,n[ is the check
blocks */
- protected final int index;
-
- public StdSplitfileBlock(RetryTracker tracker2, int index2, Bucket
data) {
- if(tracker2 == null) throw new NullPointerException();
- this.tracker = tracker2;
- this.index = index2;
- this.fetchedData = data;
- }
-
- public int getNumber() {
- return index;
- }
-
- public boolean hasData() {
- return fetchedData != null;
- }
-
- public Bucket getData() {
- return fetchedData;
- }
-
- public void setData(Bucket data) {
- if(data == fetchedData) return;
- fetchedData = data;
- Logger.minor(this, "Set data: "+(data == null ? "(null)" :
(""+data.size())+ " on "+this), new Exception("debug"));
- }
-
- public void start() {
- checkStartable();
- Logger.minor(this, "Starting "+this);
- try {
- Thread t = new Thread(this, getName());
- t.setDaemon(true);
- t.start();
- } catch (Throwable error) {
- tracker.fatalError(this,
InserterException.INTERNAL_ERROR);
- Logger.error(this, "Caught "+error+" creating thread
for "+this);
- }
- }
-
- public abstract String getName();
-
- protected abstract void checkStartable();
-}
Deleted: branches/async-client/src/freenet/client/async/Client.java
===================================================================
--- branches/async-client/src/freenet/client/async/Client.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/Client.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,25 +0,0 @@
-package freenet.client.async;
-
-import freenet.client.FetchException;
-import freenet.client.FetchResult;
-import freenet.client.InserterException;
-import freenet.keys.FreenetURI;
-
-/**
- * A client process. Something that initiates requests, and can cancel
- * them. FCP, Fproxy, and the GlobalPersistentClient, implement this
- * somewhere.
- */
-public interface Client {
-
- public void onSuccess(FetchResult result, ClientGet state);
-
- public void onFailure(FetchException e, ClientGet state);
-
- public void onSuccess(ClientPut state);
-
- public void onFailure(InserterException e, ClientPut state);
-
- public void onGeneratedURI(FreenetURI uri, ClientPut state);
-
-}
Copied: branches/async-client/src/freenet/client/async/ClientCallback.java
(from rev 7903, branches/async-client/src/freenet/client/async/Client.java)
===================================================================
--- branches/async-client/src/freenet/client/async/Client.java 2006-01-23
21:30:18 UTC (rev 7903)
+++ branches/async-client/src/freenet/client/async/ClientCallback.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -0,0 +1,25 @@
+package freenet.client.async;
+
+import freenet.client.FetchException;
+import freenet.client.FetchResult;
+import freenet.client.InserterException;
+import freenet.keys.FreenetURI;
+
+/**
+ * A client process. Something that initiates requests, and can cancel
+ * them. FCP, Fproxy, and the GlobalPersistentClient, implement this
+ * somewhere.
+ */
+public interface ClientCallback {
+
+ public void onSuccess(FetchResult result, ClientGetter state);
+
+ public void onFailure(FetchException e, ClientGetter state);
+
+ public void onSuccess(ClientPutter state);
+
+ public void onFailure(InserterException e, ClientPutter state);
+
+ public void onGeneratedURI(FreenetURI uri, ClientPutter state);
+
+}
Deleted: branches/async-client/src/freenet/client/async/ClientGet.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientGet.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientGet.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,81 +0,0 @@
-package freenet.client.async;
-
-import java.net.MalformedURLException;
-
-import freenet.client.ArchiveContext;
-import freenet.client.ClientMetadata;
-import freenet.client.FetchException;
-import freenet.client.FetchResult;
-import freenet.client.FetcherContext;
-import freenet.keys.FreenetURI;
-
-/**
- * A high level data request.
- */
-public class ClientGet extends ClientRequest implements GetCompletionCallback {
-
- final Client client;
- final FreenetURI uri;
- final FetcherContext ctx;
- final ArchiveContext actx;
- final ClientRequestScheduler scheduler;
- ClientGetState currentState;
- private boolean finished;
- private int archiveRestarts;
-
- public ClientGet(Client client, ClientRequestScheduler sched,
FreenetURI uri, FetcherContext ctx, short priorityClass) {
- super(priorityClass);
- this.client = client;
- this.uri = uri;
- this.ctx = ctx;
- this.scheduler = sched;
- this.finished = false;
- this.actx = new ArchiveContext();
- archiveRestarts = 0;
- start();
- }
-
- private void start() {
- try {
- currentState = new SingleFileFetcher(this, this, new
ClientMetadata(), uri, ctx, actx, getPriorityClass(), 0, false, null);
- currentState.schedule();
- } catch (MalformedURLException e) {
- onFailure(new
FetchException(FetchException.INVALID_URI, e), null);
- } catch (FetchException e) {
- onFailure(e, null);
- }
- }
-
- public void onSuccess(FetchResult result, ClientGetState state) {
- finished = true;
- currentState = null;
- client.onSuccess(result, this);
- }
-
- public void onFailure(FetchException e, ClientGetState state) {
- if(e.mode == FetchException.ARCHIVE_RESTART) {
- archiveRestarts++;
- if(archiveRestarts > ctx.maxArchiveRestarts)
- e = new
FetchException(FetchException.TOO_MANY_ARCHIVE_RESTARTS);
- else {
- start();
- return;
- }
- }
- finished = true;
- client.onFailure(e, this);
- }
-
- public void cancel() {
- synchronized(this) {
- super.cancel();
- if(currentState != null)
- currentState.cancel();
- }
- }
-
- public boolean isFinished() {
- return finished || cancelled;
- }
-
-}
Modified: branches/async-client/src/freenet/client/async/ClientGetState.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientGetState.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientGetState.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -6,7 +6,7 @@
*/
public abstract class ClientGetState {
- public abstract ClientGet getParent();
+ public abstract ClientGetter getParent();
public abstract void schedule();
Copied: branches/async-client/src/freenet/client/async/ClientGetter.java (from
rev 7909, branches/async-client/src/freenet/client/async/ClientGet.java)
===================================================================
--- branches/async-client/src/freenet/client/async/ClientGet.java
2006-01-24 20:54:14 UTC (rev 7909)
+++ branches/async-client/src/freenet/client/async/ClientGetter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -0,0 +1,84 @@
+package freenet.client.async;
+
+import java.net.MalformedURLException;
+
+import freenet.client.ArchiveContext;
+import freenet.client.ClientMetadata;
+import freenet.client.FetchException;
+import freenet.client.FetchResult;
+import freenet.client.FetcherContext;
+import freenet.keys.FreenetURI;
+
+/**
+ * A high level data request.
+ */
+public class ClientGetter extends ClientRequest implements
GetCompletionCallback {
+
+ final ClientCallback client;
+ final FreenetURI uri;
+ final FetcherContext ctx;
+ final ArchiveContext actx;
+ final ClientRequestScheduler scheduler;
+ ClientGetState currentState;
+ private boolean finished;
+ private int archiveRestarts;
+
+ public ClientGetter(ClientCallback client, ClientRequestScheduler
sched, FreenetURI uri, FetcherContext ctx, short priorityClass) {
+ super(priorityClass);
+ this.client = client;
+ this.uri = uri;
+ this.ctx = ctx;
+ this.scheduler = sched;
+ this.finished = false;
+ this.actx = new ArchiveContext();
+ archiveRestarts = 0;
+ }
+
+ public void start() {
+ try {
+ currentState = new SingleFileFetcher(this, this, new
ClientMetadata(), uri, ctx, actx, getPriorityClass(), 0, false, null);
+ currentState.schedule();
+ } catch (MalformedURLException e) {
+ onFailure(new
FetchException(FetchException.INVALID_URI, e), null);
+ } catch (FetchException e) {
+ onFailure(e, null);
+ }
+ }
+
+ public void onSuccess(FetchResult result, ClientGetState state) {
+ finished = true;
+ currentState = null;
+ client.onSuccess(result, this);
+ }
+
+ public void onFailure(FetchException e, ClientGetState state) {
+ if(e.mode == FetchException.ARCHIVE_RESTART) {
+ archiveRestarts++;
+ if(archiveRestarts > ctx.maxArchiveRestarts)
+ e = new
FetchException(FetchException.TOO_MANY_ARCHIVE_RESTARTS);
+ else {
+ start();
+ return;
+ }
+ }
+ finished = true;
+ client.onFailure(e, this);
+ }
+
+ public void cancel() {
+ synchronized(this) {
+ super.cancel();
+ if(currentState != null)
+ currentState.cancel();
+ }
+ }
+
+ public boolean isFinished() {
+ return finished || cancelled;
+ }
+
+ public FreenetURI getURI() {
+ return uri;
+ }
+
+}
Deleted: branches/async-client/src/freenet/client/async/ClientPut.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientPut.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientPut.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,79 +0,0 @@
-package freenet.client.async;
-
-import freenet.client.ClientMetadata;
-import freenet.client.InsertBlock;
-import freenet.client.InserterContext;
-import freenet.client.InserterException;
-import freenet.keys.ClientKey;
-import freenet.keys.FreenetURI;
-import freenet.support.Bucket;
-
-public class ClientPut extends ClientRequest implements PutCompletionCallback {
-
- final Client client;
- final Bucket data;
- final FreenetURI targetURI;
- final ClientMetadata cm;
- final InserterContext ctx;
- final ClientRequestScheduler scheduler;
- private ClientPutState currentState;
- private boolean finished;
- private final boolean getCHKOnly;
-
- public ClientPut(Client client, Bucket data, FreenetURI targetURI,
ClientMetadata cm, InserterContext ctx,
- ClientRequestScheduler scheduler, short priorityClass,
boolean getCHKOnly) {
- super(priorityClass);
- this.cm = cm;
- this.getCHKOnly = getCHKOnly;
- this.client = client;
- this.data = data;
- this.targetURI = targetURI;
- this.ctx = ctx;
- this.scheduler = scheduler;
- this.finished = false;
- this.cancelled = false;
- try {
- start();
- } catch (InserterException e) {
- onFailure(e, null);
- }
- }
-
- private void start() throws InserterException {
- currentState =
- new SingleFileInserter(this, this, new
InsertBlock(data, cm, targetURI), false, ctx, false, false, getCHKOnly);
- }
-
- void setCurrentState(ClientPutState s) {
- currentState = s;
- }
-
- public void onSuccess(ClientPutState state) {
- finished = true;
- currentState = null;
- client.onSuccess(this);
- }
-
- public void onFailure(InserterException e, ClientPutState state) {
- finished = true;
- currentState = null;
- client.onFailure(e, this);
- }
-
- public void onEncode(ClientKey key, ClientPutState state) {
- client.onGeneratedURI(key.getURI(), this);
- }
-
- public void cancel() {
- synchronized(this) {
- super.cancel();
- if(currentState != null)
- currentState.cancel();
- }
- }
-
- public boolean isFinished() {
- return finished || cancelled;
- }
-
-}
Modified: branches/async-client/src/freenet/client/async/ClientPutState.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientPutState.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientPutState.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -7,7 +7,7 @@
*/
public interface ClientPutState {
- public abstract ClientPut getParent();
+ public abstract ClientPutter getParent();
public abstract void cancel();
Copied: branches/async-client/src/freenet/client/async/ClientPutter.java (from
rev 7909, branches/async-client/src/freenet/client/async/ClientPut.java)
===================================================================
--- branches/async-client/src/freenet/client/async/ClientPut.java
2006-01-24 20:54:14 UTC (rev 7909)
+++ branches/async-client/src/freenet/client/async/ClientPutter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -0,0 +1,87 @@
+package freenet.client.async;
+
+import freenet.client.ClientMetadata;
+import freenet.client.InsertBlock;
+import freenet.client.InserterContext;
+import freenet.client.InserterException;
+import freenet.keys.ClientKey;
+import freenet.keys.FreenetURI;
+import freenet.support.Bucket;
+
+public class ClientPutter extends ClientRequest implements
PutCompletionCallback {
+
+ final ClientCallback client;
+ final Bucket data;
+ final FreenetURI targetURI;
+ final ClientMetadata cm;
+ final InserterContext ctx;
+ final ClientRequestScheduler scheduler;
+ private ClientPutState currentState;
+ private boolean finished;
+ private final boolean getCHKOnly;
+ private final boolean isMetadata;
+ private FreenetURI uri;
+
+ public ClientPutter(ClientCallback client, Bucket data, FreenetURI
targetURI, ClientMetadata cm, InserterContext ctx,
+ ClientRequestScheduler scheduler, short priorityClass,
boolean getCHKOnly, boolean isMetadata) {
+ super(priorityClass);
+ this.cm = cm;
+ this.isMetadata = isMetadata;
+ this.getCHKOnly = getCHKOnly;
+ this.client = client;
+ this.data = data;
+ this.targetURI = targetURI;
+ this.ctx = ctx;
+ this.scheduler = scheduler;
+ this.finished = false;
+ this.cancelled = false;
+ }
+
+ public void start() throws InserterException {
+ try {
+ currentState =
+ new SingleFileInserter(this, this, new
InsertBlock(data, cm, targetURI), isMetadata, ctx, false, false, getCHKOnly);
+ } catch (InserterException e) {
+ finished = true;
+ currentState = null;
+ }
+ }
+
+ void setCurrentState(ClientPutState s) {
+ currentState = s;
+ }
+
+ public void onSuccess(ClientPutState state) {
+ finished = true;
+ currentState = null;
+ client.onSuccess(this);
+ }
+
+ public void onFailure(InserterException e, ClientPutState state) {
+ finished = true;
+ currentState = null;
+ client.onFailure(e, this);
+ }
+
+ public void onEncode(ClientKey key, ClientPutState state) {
+ this.uri = key.getURI();
+ client.onGeneratedURI(uri, this);
+ }
+
+ public void cancel() {
+ synchronized(this) {
+ super.cancel();
+ if(currentState != null)
+ currentState.cancel();
+ }
+ }
+
+ public boolean isFinished() {
+ return finished || cancelled;
+ }
+
+ public FreenetURI getURI() {
+ return uri;
+ }
+
+}
Modified: branches/async-client/src/freenet/client/async/ClientRequest.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientRequest.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientRequest.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,5 +1,7 @@
package freenet.client.async;
+import freenet.keys.FreenetURI;
+
/** A high level client request. A request (either fetch or put) started
* by a Client. Has a suitable context and a URI; is fulfilled only when
* we have followed all the redirects etc, or have an error. Can be
@@ -27,5 +29,7 @@
return cancelled;
}
+ public abstract FreenetURI getURI();
+
public abstract boolean isFinished();
}
Modified:
branches/async-client/src/freenet/client/async/ClientRequestScheduler.java
===================================================================
--- branches/async-client/src/freenet/client/async/ClientRequestScheduler.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/ClientRequestScheduler.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -10,7 +10,7 @@
* ask for a request to start. A request is then started, in its own
* thread. It is removed at that point.
*/
-public class ClientRequestScheduler {
+public class ClientRequestScheduler implements RequestScheduler {
/**
* Structure:
@@ -24,8 +24,10 @@
// we have one for inserts and one for requests
final boolean isInsertScheduler;
final RandomSource random;
+ private final RequestStarter starter;
- ClientRequestScheduler(boolean forInserts, RandomSource random) {
+ public ClientRequestScheduler(boolean forInserts, RandomSource random,
RequestStarter starter) {
+ this.starter = starter;
this.random = random;
this.isInsertScheduler = forInserts;
priorities = new
SortedVectorByNumber[RequestStarter.NUMBER_OF_PRIORITY_CLASSES];
@@ -33,12 +35,17 @@
priorities[i] = new SortedVectorByNumber();
}
- public synchronized void register(SendableRequest req) {
- if((!isInsertScheduler) && req instanceof ClientPut)
- throw new IllegalArgumentException("Expected a
ClientPut: "+req);
- RandomGrabArrayWithInt grabber =
- makeGrabArray(req.getPriorityClass(),
req.getRetryCount());
- grabber.add(req);
+ public void register(SendableRequest req) {
+ synchronized(this) {
+ if((!isInsertScheduler) && req instanceof ClientPutter)
+ throw new IllegalArgumentException("Expected a
ClientPut: "+req);
+ RandomGrabArrayWithInt grabber =
+ makeGrabArray(req.getPriorityClass(),
req.getRetryCount());
+ grabber.add(req);
+ }
+ synchronized(starter) {
+ starter.notifyAll();
+ }
}
private synchronized RandomGrabArrayWithInt makeGrabArray(short
priorityClass, int retryCount) {
@@ -78,13 +85,13 @@
}
}
- public synchronized ClientRequest getFirst() {
+ public synchronized SendableRequest removeFirst() {
// Priorities start at 0
for(int i=0;i<RequestStarter.MINIMUM_PRIORITY_CLASS;i++) {
SortedVectorByNumber s = priorities[i];
if(s == null) continue;
RandomGrabArrayWithInt rga = (RandomGrabArrayWithInt)
s.getFirst(); // will discard finished items
- ClientRequest req = (ClientRequest) rga.removeRandom();
+ SendableRequest req = (SendableRequest)
rga.removeRandom();
if(rga.isEmpty()) {
s.remove(rga.getNumber());
if(s.isEmpty()) {
Modified:
branches/async-client/src/freenet/client/async/MultiPutCompletionCallback.java
===================================================================
---
branches/async-client/src/freenet/client/async/MultiPutCompletionCallback.java
2006-01-25 01:47:43 UTC (rev 7916)
+++
branches/async-client/src/freenet/client/async/MultiPutCompletionCallback.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -10,11 +10,11 @@
private final LinkedList waitingFor;
private final PutCompletionCallback cb;
private ClientPutState generator;
- private final ClientPut parent;
+ private final ClientPutter parent;
private boolean finished;
private boolean started;
- public MultiPutCompletionCallback(PutCompletionCallback cb, ClientPut
parent, boolean dontTellParent) {
+ public MultiPutCompletionCallback(PutCompletionCallback cb,
ClientPutter parent, boolean dontTellParent) {
this.cb = cb;
this.waitingFor = new LinkedList();
this.parent = parent;
@@ -61,7 +61,7 @@
started = true;
}
- public ClientPut getParent() {
+ public ClientPutter getParent() {
return parent;
}
Added: branches/async-client/src/freenet/client/async/RequestScheduler.java
===================================================================
--- branches/async-client/src/freenet/client/async/RequestScheduler.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/RequestScheduler.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -0,0 +1,7 @@
+package freenet.client.async;
+
+public interface RequestScheduler {
+
+ public SendableRequest removeFirst();
+
+}
Modified: branches/async-client/src/freenet/client/async/SendableGet.java
===================================================================
--- branches/async-client/src/freenet/client/async/SendableGet.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SendableGet.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -2,7 +2,7 @@
import freenet.keys.ClientKey;
import freenet.keys.ClientKeyBlock;
-import freenet.node.LowLevelPutException;
+import freenet.node.LowLevelGetException;
/**
* A low-level key fetch which can be sent immediately. @see SendableRequest
@@ -15,6 +15,6 @@
public void onSuccess(ClientKeyBlock block);
/** Called when/if the low-level request fails. */
- public void onFailure(LowLevelPutException e);
+ public void onFailure(LowLevelGetException e);
}
Modified: branches/async-client/src/freenet/client/async/SendableRequest.java
===================================================================
--- branches/async-client/src/freenet/client/async/SendableRequest.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SendableRequest.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,5 +1,6 @@
package freenet.client.async;
+import freenet.node.Node;
import freenet.support.RandomGrabArrayItem;
/**
@@ -12,4 +13,7 @@
public int getRetryCount();
+ /** ONLY called by RequestStarter */
+ public void send(Node node);
+
}
Modified:
branches/async-client/src/freenet/client/async/SingleBlockInserter.java
===================================================================
--- branches/async-client/src/freenet/client/async/SingleBlockInserter.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SingleBlockInserter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -15,6 +15,7 @@
import freenet.keys.InsertableClientSSK;
import freenet.keys.SSKEncodeException;
import freenet.node.LowLevelPutException;
+import freenet.node.Node;
import freenet.support.Bucket;
import freenet.support.Logger;
@@ -28,7 +29,7 @@
final FreenetURI uri; // uses essentially no RAM in the common case of
a CHK because we use FreenetURI.EMPTY_CHK_URI
FreenetURI resultingURI;
final PutCompletionCallback cb;
- final ClientPut parent;
+ final ClientPutter parent;
final InserterContext ctx;
private int retries;
private final FailureCodeTracker errors;
@@ -39,7 +40,7 @@
final boolean isMetadata;
final int sourceLength;
- public SingleBlockInserter(ClientPut parent, Bucket data, short
compressionCodec, FreenetURI uri, InserterContext ctx, PutCompletionCallback
cb, boolean isMetadata, int sourceLength, int token, boolean getCHKOnly) throws
InserterException {
+ public SingleBlockInserter(ClientPutter parent, Bucket data, short
compressionCodec, FreenetURI uri, InserterContext ctx, PutCompletionCallback
cb, boolean isMetadata, int sourceLength, int token, boolean getCHKOnly) throws
InserterException {
this.token = token;
this.parent = parent;
this.retries = 0;
@@ -184,7 +185,7 @@
cb.onSuccess(this);
}
- public ClientPut getParent() {
+ public ClientPutter getParent() {
return parent;
}
@@ -200,4 +201,14 @@
return finished;
}
+ public void send(Node node) {
+ try {
+ node.realPut(getBlock(), true);
+ } catch (LowLevelPutException e) {
+ onFailure(e);
+ return;
+ }
+ onSuccess();
+ }
+
}
Modified: branches/async-client/src/freenet/client/async/SingleFileFetcher.java
===================================================================
--- branches/async-client/src/freenet/client/async/SingleFileFetcher.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SingleFileFetcher.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -20,6 +20,7 @@
import freenet.keys.KeyDecodeException;
import freenet.node.LowLevelGetException;
import freenet.node.LowLevelPutException;
+import freenet.node.Node;
import freenet.support.Bucket;
import freenet.support.Logger;
import freenet.support.compress.CompressionOutputSizeException;
@@ -27,7 +28,7 @@
public class SingleFileFetcher extends ClientGetState implements SendableGet {
- final ClientGet parent;
+ final ClientGetter parent;
//final FreenetURI uri;
final ClientKey key;
final LinkedList metaStrings;
@@ -54,7 +55,7 @@
* @param token
* @param dontTellClientGet
*/
- public SingleFileFetcher(ClientGet get, GetCompletionCallback cb,
ClientMetadata metadata, ClientKey key, LinkedList metaStrings, FetcherContext
ctx, ArchiveContext actx, int maxRetries, int recursionLevel, boolean
dontTellClientGet, Object token) throws FetchException {
+ public SingleFileFetcher(ClientGetter get, GetCompletionCallback cb,
ClientMetadata metadata, ClientKey key, LinkedList metaStrings, FetcherContext
ctx, ArchiveContext actx, int maxRetries, int recursionLevel, boolean
dontTellClientGet, Object token) throws FetchException {
this.cancelled = false;
this.dontTellClientGet = dontTellClientGet;
this.token = token;
@@ -78,7 +79,7 @@
}
/** Called by ClientGet. */
- public SingleFileFetcher(ClientGet get, GetCompletionCallback cb,
ClientMetadata metadata, FreenetURI uri, FetcherContext ctx, ArchiveContext
actx, int maxRetries, int recursionLevel, boolean dontTellClientGet, Object
token) throws MalformedURLException, FetchException {
+ public SingleFileFetcher(ClientGetter get, GetCompletionCallback cb,
ClientMetadata metadata, FreenetURI uri, FetcherContext ctx, ArchiveContext
actx, int maxRetries, int recursionLevel, boolean dontTellClientGet, Object
token) throws MalformedURLException, FetchException {
this(get, cb, metadata, ClientKey.getBaseKey(uri),
uri.listMetaStrings(), ctx, actx, maxRetries, recursionLevel,
dontTellClientGet, token);
}
@@ -110,7 +111,7 @@
parent.scheduler.register(this);
}
- public ClientGet getParent() {
+ public ClientGetter getParent() {
return parent;
}
@@ -235,7 +236,7 @@
ah = (ArchiveStoreContext)
ctx.archiveManager.makeHandler(thisKey, metadata.getArchiveType(), false);
// ah is set. This means we are currently
handling an archive.
Bucket metadataBucket;
- metadataBucket = ah.getMetadata(actx, null,
null, recursionLevel+1, true);
+ metadataBucket = ah.getMetadata(actx, null,
recursionLevel+1, true);
if(metadataBucket != null) {
try {
metadata =
Metadata.construct(metadataBucket);
@@ -255,7 +256,7 @@
throw new
FetchException(FetchException.UNKNOWN_METADATA, "Archive redirect not in an
archive");
if(metaStrings.isEmpty())
throw new
FetchException(FetchException.NOT_ENOUGH_METASTRINGS);
- Bucket dataBucket = ah.get((String)
metaStrings.removeFirst(), actx, null, null, recursionLevel+1, true);
+ Bucket dataBucket = ah.get((String)
metaStrings.removeFirst(), actx, null, recursionLevel+1, true);
if(dataBucket != null) {
// Return the data
onSuccess(new
FetchResult(this.clientMetadata, dataBucket));
@@ -432,7 +433,7 @@
}
// Translate it, then call the real onFailure
- public void onFailure(LowLevelPutException e) {
+ public void onFailure(LowLevelGetException e) {
switch(e.code) {
case LowLevelGetException.DATA_NOT_FOUND:
onFailure(new
FetchException(FetchException.DATA_NOT_FOUND));
@@ -468,4 +469,20 @@
return cancelled;
}
+ /** Do the request, blocking. Called by RequestStarter. */
+ public void send(Node node) {
+ // Do we need to support the last 3?
+ ClientKeyBlock block;
+ try {
+ block = node.realGetKey(key, false, false, false);
+ } catch (LowLevelGetException e) {
+ onFailure(e);
+ return;
+ } catch (Throwable t) {
+ onFailure(new
LowLevelGetException(LowLevelGetException.INTERNAL_ERROR));
+ return;
+ }
+ onSuccess(block);
+ }
+
}
Modified: branches/async-client/src/freenet/client/async/SingleFileInserter.java
===================================================================
--- branches/async-client/src/freenet/client/async/SingleFileInserter.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SingleFileInserter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -28,7 +28,7 @@
// Config option???
private static final long COMPRESS_OFF_THREAD_LIMIT = 65536;
- final ClientPut parent;
+ final ClientPutter parent;
final InsertBlock block;
final InserterContext ctx;
final boolean metadata;
@@ -39,7 +39,7 @@
final boolean dontTellParent;
private boolean cancelled = false;
- SingleFileInserter(ClientPut parent, PutCompletionCallback cb,
InsertBlock block,
+ SingleFileInserter(ClientPutter parent, PutCompletionCallback cb,
InsertBlock block,
boolean metadata, InserterContext ctx, boolean
dontCompress,
boolean dontTellParent, boolean getCHKOnly) throws
InserterException {
this.parent = parent;
@@ -244,7 +244,7 @@
cb.onFailure(e, this);
}
- public ClientPut getParent() {
+ public ClientPutter getParent() {
return parent;
}
@@ -262,7 +262,7 @@
}
- public ClientPut getParent() {
+ public ClientPutter getParent() {
return parent;
}
Modified: branches/async-client/src/freenet/client/async/SplitFileFetcher.java
===================================================================
--- branches/async-client/src/freenet/client/async/SplitFileFetcher.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SplitFileFetcher.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -30,7 +30,7 @@
final ArchiveContext archiveContext;
final LinkedList decompressors;
final ClientMetadata clientMetadata;
- final ClientGet parent;
+ final ClientGetter parent;
final GetCompletionCallback cb;
final int recursionLevel;
/** The splitfile type. See the SPLITFILE_ constants on Metadata. */
@@ -57,7 +57,7 @@
private final boolean splitUseLengths;
private boolean finished;
- public SplitFileFetcher(Metadata metadata, GetCompletionCallback rcb,
ClientGet parent,
+ public SplitFileFetcher(Metadata metadata, GetCompletionCallback rcb,
ClientGetter parent,
FetcherContext newCtx, LinkedList decompressors,
ClientMetadata clientMetadata,
ArchiveContext actx, int recursionLevel) throws
FetchException, MetadataParseException {
this.finished = false;
@@ -213,7 +213,7 @@
}
}
- public ClientGet getParent() {
+ public ClientGetter getParent() {
return parent;
}
Modified: branches/async-client/src/freenet/client/async/SplitFileInserter.java
===================================================================
--- branches/async-client/src/freenet/client/async/SplitFileInserter.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/client/async/SplitFileInserter.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -19,7 +19,7 @@
public class SplitFileInserter implements ClientPutState {
- final ClientPut parent;
+ final ClientPutter parent;
final InserterContext ctx;
final SplitPutCompletionCallback cb;
final long dataLength;
@@ -36,7 +36,7 @@
final boolean isMetadata;
private boolean finished;
- public SplitFileInserter(ClientPut put, SplitPutCompletionCallback cb,
Bucket data, Compressor bestCodec, ClientMetadata clientMetadata,
InserterContext ctx, SplitHandler sh, boolean getCHKOnly, boolean isMetadata,
boolean dontTellParent) throws InserterException {
+ public SplitFileInserter(ClientPutter put, SplitPutCompletionCallback
cb, Bucket data, Compressor bestCodec, ClientMetadata clientMetadata,
InserterContext ctx, SplitHandler sh, boolean getCHKOnly, boolean isMetadata,
boolean dontTellParent) throws InserterException {
this.parent = put;
if(!dontTellParent)
parent.setCurrentState(this);
@@ -208,7 +208,7 @@
return uris;
}
- public ClientPut getParent() {
+ public ClientPutter getParent() {
return parent;
}
Modified: branches/async-client/src/freenet/node/LowLevelGetException.java
===================================================================
--- branches/async-client/src/freenet/node/LowLevelGetException.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/LowLevelGetException.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -56,7 +56,7 @@
this.code = code;
}
- LowLevelGetException(int reason) {
+ public LowLevelGetException(int reason) {
super(getMessage(reason));
this.code = reason;
}
Modified: branches/async-client/src/freenet/node/Node.java
===================================================================
--- branches/async-client/src/freenet/node/Node.java 2006-01-25 01:47:43 UTC
(rev 7916)
+++ branches/async-client/src/freenet/node/Node.java 2006-01-25 14:14:13 UTC
(rev 7917)
@@ -27,6 +27,7 @@
import freenet.client.ArchiveManager;
import freenet.client.HighLevelSimpleClient;
import freenet.client.HighLevelSimpleClientImpl;
+import freenet.client.async.ClientRequestScheduler;
import freenet.clients.http.FproxyToadlet;
import freenet.clients.http.SimpleToadletServer;
import freenet.crypt.DSAPublicKey;
@@ -76,7 +77,7 @@
/**
* @author amphibian
*/
-public class Node implements QueueingSimpleLowLevelClient {
+public class Node {
static final long serialVersionUID = -1;
@@ -190,6 +191,8 @@
final File downloadDir;
final TestnetHandler testnetHandler;
final TestnetStatusUploader statusUploader;
+ public final ClientRequestScheduler fetchScheduler;
+ public final ClientRequestScheduler putScheduler;
// Client stuff that needs to be configged - FIXME
static final int MAX_ARCHIVE_HANDLERS = 200; // don't take up much RAM...
FIXME
@@ -333,7 +336,7 @@
t.setPriority(Thread.MAX_PRIORITY);
t.start();
SimpleToadletServer server = new SimpleToadletServer(port+2000);
- FproxyToadlet fproxy = new
FproxyToadlet(n.makeClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS,
(short)0));
+ FproxyToadlet fproxy = new
FproxyToadlet(n.makeClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS));
server.register(fproxy, "/", false);
System.out.println("Starting fproxy on port "+(port+2000));
new FCPServer(port+3000, n);
@@ -450,11 +453,15 @@
tempBucketFactory = new
PaddedEphemerallyEncryptedBucketFactory(new
TempBucketFactory(tempFilenameGenerator), random, 1024);
archiveManager = new ArchiveManager(MAX_ARCHIVE_HANDLERS,
MAX_CACHED_ARCHIVE_DATA, MAX_ARCHIVE_SIZE, MAX_ARCHIVED_FILE_SIZE,
MAX_CACHED_ELEMENTS, random, tempFilenameGenerator);
requestThrottle = new RequestThrottle(5000, 2.0F);
- requestStarter = new RequestStarter(requestThrottle, "Request
starter ("+portNumber+")");
+ requestStarter = new RequestStarter(this, requestThrottle,
"Request starter ("+portNumber+")");
+ fetchScheduler = new ClientRequestScheduler(false, random,
requestStarter);
+ requestStarter.start();
//insertThrottle = new ChainedRequestThrottle(10000, 2.0F,
requestThrottle);
// FIXME reenable the above
insertThrottle = new RequestThrottle(10000, 2.0F);
- insertStarter = new RequestStarter(insertThrottle, "Insert
starter ("+portNumber+")");
+ insertStarter = new RequestStarter(this, insertThrottle,
"Insert starter ("+portNumber+")");
+ putScheduler = new ClientRequestScheduler(true, random,
insertStarter);
+ insertStarter.start();
if(testnetHandler != null)
testnetHandler.start();
if(statusUploader != null)
@@ -469,18 +476,6 @@
usm.start();
}
- public ClientKeyBlock getKey(ClientKey key, boolean localOnly,
RequestStarterClient client, boolean cache, boolean ignoreStore) throws
LowLevelGetException {
- if(key instanceof ClientSSK) {
- ClientSSK k = (ClientSSK) key;
- if(k.getPubKey() != null)
- cacheKey(k.pubKeyHash, k.getPubKey());
- }
- if(localOnly)
- return realGetKey(key, localOnly, cache, ignoreStore);
- else
- return client.getKey(key, localOnly, cache, ignoreStore);
- }
-
public ClientKeyBlock realGetKey(ClientKey key, boolean localOnly, boolean
cache, boolean ignoreStore) throws LowLevelGetException {
if(key instanceof ClientCHK)
return realGetCHK((ClientCHK)key, localOnly, cache,
ignoreStore);
@@ -667,10 +662,6 @@
}
}
- public void putKey(ClientKeyBlock block, RequestStarterClient client,
boolean cache) throws LowLevelPutException {
- client.putKey(block, cache);
- }
-
public void realPut(ClientKeyBlock block, boolean cache) throws
LowLevelPutException {
if(block instanceof ClientCHKBlock)
realPutCHK((ClientCHKBlock)block, cache);
@@ -1363,8 +1354,8 @@
writeNodeFile();
}
- public HighLevelSimpleClient makeClient(short prioClass, short prio) {
- return new HighLevelSimpleClientImpl(this, archiveManager,
tempBucketFactory, random, makeStarterClient(prioClass, prio, false),
makeStarterClient(prioClass, prio, true), !DONT_CACHE_LOCAL_REQUESTS);
+ public HighLevelSimpleClient makeClient(short prioClass) {
+ return new HighLevelSimpleClientImpl(this, archiveManager,
tempBucketFactory, random, !DONT_CACHE_LOCAL_REQUESTS, prioClass);
}
private static class MemoryChecker implements Runnable {
@@ -1390,10 +1381,6 @@
return insertThrottle;
}
- public RequestStarterClient makeStarterClient(short prioClass, short
prio, boolean inserts) {
- return new RequestStarterClient(prioClass, prio, random, this,
inserts ? insertStarter : requestStarter);
- }
-
InetAddress lastIP;
public void redetectAddress() {
Deleted: branches/async-client/src/freenet/node/QueuedDataRequest.java
===================================================================
--- branches/async-client/src/freenet/node/QueuedDataRequest.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/QueuedDataRequest.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,28 +0,0 @@
-package freenet.node;
-
-import freenet.keys.ClientKey;
-import freenet.keys.ClientKeyBlock;
-import freenet.keys.KeyBlock;
-
-public class QueuedDataRequest extends QueuedRequest {
-
- private final ClientKey key;
- private final boolean localOnly;
- private final boolean cache;
- private final boolean ignoreStore;
- private QueueingSimpleLowLevelClient client;
-
- public QueuedDataRequest(ClientKey key, boolean localOnly, boolean
cache, QueueingSimpleLowLevelClient client, boolean ignoreStore) {
- this.key = key;
- this.localOnly = localOnly;
- this.client = client;
- this.cache = cache;
- this.ignoreStore = ignoreStore;
- }
-
- public ClientKeyBlock waitAndFetch() throws LowLevelGetException {
- waitForSendClearance();
- return client.realGetKey(key, localOnly, cache, ignoreStore);
- }
-
-}
Deleted: branches/async-client/src/freenet/node/QueuedInsertRequest.java
===================================================================
--- branches/async-client/src/freenet/node/QueuedInsertRequest.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/QueuedInsertRequest.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,22 +0,0 @@
-package freenet.node;
-
-import freenet.keys.ClientCHKBlock;
-import freenet.keys.ClientKeyBlock;
-
-public class QueuedInsertRequest extends QueuedRequest {
-
- private final ClientKeyBlock block;
- private final boolean cache;
- private QueueingSimpleLowLevelClient client;
-
- public QueuedInsertRequest(ClientKeyBlock block,
QueueingSimpleLowLevelClient client, boolean cache) {
- this.block = block;
- this.client = client;
- this.cache = cache;
- }
-
- public void waitAndPut() throws LowLevelPutException {
- waitForSendClearance();
- client.realPut(block, cache);
- }
-}
Deleted: branches/async-client/src/freenet/node/QueuedRequest.java
===================================================================
--- branches/async-client/src/freenet/node/QueuedRequest.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/QueuedRequest.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,32 +0,0 @@
-package freenet.node;
-
-/**
- * A request (including both DataRequest's and InsertRequest's) which can be
queued
- * by a RequestStarter.
- */
-public abstract class QueuedRequest {
-
- private boolean clearToSend = false;
-
- /**
- * Shell for sending the request.
- */
- public final void clearToSend() {
- synchronized(this) {
- clearToSend = true;
- notifyAll();
- }
- }
-
- protected void waitForSendClearance() {
- synchronized(this) {
- while(!clearToSend) {
- try {
- wait(10*1000);
- } catch (InterruptedException e) {
- // Ignore
- }
- }
- }
- }
-}
Deleted:
branches/async-client/src/freenet/node/QueueingSimpleLowLevelClient.java
===================================================================
--- branches/async-client/src/freenet/node/QueueingSimpleLowLevelClient.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/QueueingSimpleLowLevelClient.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,14 +0,0 @@
-package freenet.node;
-
-import freenet.keys.ClientKey;
-import freenet.keys.ClientKeyBlock;
-
-interface QueueingSimpleLowLevelClient extends SimpleLowLevelClient {
-
- /** Unqueued version. Only call from QueuedDataRequest ! */
- ClientKeyBlock realGetKey(ClientKey key, boolean localOnly, boolean
cache, boolean ignoreStore) throws LowLevelGetException;
-
- /** Ditto */
- void realPut(ClientKeyBlock block, boolean cache) throws
LowLevelPutException;
-
-}
Modified: branches/async-client/src/freenet/node/RealNodeRequestInsertTest.java
===================================================================
--- branches/async-client/src/freenet/node/RealNodeRequestInsertTest.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/RealNodeRequestInsertTest.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -73,10 +73,6 @@
b.peers.connect(a.exportFieldSet());
}
- RequestStarterClient[] starters = new
RequestStarterClient[NUMBER_OF_NODES];
- for(int i=0;i<starters.length;i++)
- starters[i] =
nodes[i].makeStarterClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS, (short)0,
false); // pretend are all requests
-
Logger.normal(RealNodeRoutingTest.class, "Added random links");
SwapRequestInterval sri =
@@ -182,7 +178,7 @@
Logger.error(RealNodeRequestInsertTest.class, "Decoded: "+new
String(newBlock.memoryDecode()));
Logger.error(RealNodeRequestInsertTest.class,"CHK:
"+chk.getURI());
Logger.error(RealNodeRequestInsertTest.class,"Headers:
"+HexUtil.bytesToHex(block.getHeaders()));
- randomNode.putKey(block, starters[node1], true);
+ randomNode.realPut(block, true);
Logger.error(RealNodeRequestInsertTest.class, "Inserted to
"+node1);
Logger.error(RealNodeRequestInsertTest.class, "Data:
"+Fields.hashCode(encData)+", Headers: "+Fields.hashCode(encHeaders));
// Pick random node to request from
@@ -191,7 +187,7 @@
node2 = random.nextInt(NUMBER_OF_NODES);
} while(node2 == node1);
Node fetchNode = nodes[node2];
- block = (ClientCHKBlock) fetchNode.getKey((ClientKey) chk,
false, starters[node2], true, false);
+ block = (ClientCHKBlock) fetchNode.realGetKey((ClientKey) chk,
false, true, false);
if(block == null) {
Logger.error(RealNodeRequestInsertTest.class, "Fetch
FAILED from "+node2);
requestsAvg.report(0.0);
Modified: branches/async-client/src/freenet/node/RequestStarter.java
===================================================================
--- branches/async-client/src/freenet/node/RequestStarter.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/RequestStarter.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -3,6 +3,9 @@
import java.util.LinkedList;
import java.util.Vector;
+import freenet.client.async.ClientRequest;
+import freenet.client.async.RequestScheduler;
+import freenet.client.async.SendableRequest;
import freenet.support.Logger;
import freenet.support.UpdatableSortedLinkedList;
import freenet.support.UpdatableSortedLinkedListKilledException;
@@ -35,155 +38,81 @@
public static final short NUMBER_OF_PRIORITY_CLASSES =
MINIMUM_PRIORITY_CLASS - MAXIMUM_PRIORITY_CLASS;
- // Clients registered
- final Vector clientsByPriority;
final RequestThrottle throttle;
- /*
- * Clients which are ready.
- * How do we do round-robin?
- * Have a list of clients which are ready to go, in priority order, and
- * haven't gone this cycle.
- * Have a list of clients which are ready to go next cycle, in priority
- * order.
- * Have each client track the cycle number in which it was last sent.
- */
- final UpdatableSortedLinkedList clientsReadyThisCycle;
- final UpdatableSortedLinkedList clientsReadyNextCycle;
- /** Increment every time we go through the whole list */
- long cycleNumber;
+ RequestScheduler sched;
+ final Node node;
- public RequestStarter(RequestThrottle throttle, String name) {
- clientsByPriority = new Vector();
- clientsReadyThisCycle = new UpdatableSortedLinkedList();
- clientsReadyNextCycle = new UpdatableSortedLinkedList();
- cycleNumber = 0;
+ public RequestStarter(Node node, RequestThrottle throttle, String name)
{
+ this.node = node;
this.throttle = throttle;
this.name = name;
+ }
+
+ void setScheduler(RequestScheduler sched) {
+ this.sched = sched;
+ }
+
+ void start() {
Thread t = new Thread(this, name);
t.setDaemon(true);
t.start();
}
-
+
final String name;
public String toString() {
return name;
}
- public synchronized void registerClient(RequestStarterClient client) {
- int p = client.priority;
- LinkedList prio = makePriority(p);
- prio.add(client);
- }
-
- public synchronized void notifyReady(RequestStarterClient client) {
- Logger.minor(this, "notifyReady("+client+")");
- try {
- if(client.getCycleLastSent() == cycleNumber) {
- clientsReadyNextCycle.addOrUpdate(client);
- } else {
- // Can send immediately
- clientsReadyThisCycle.addOrUpdate(client);
- }
- } catch (UpdatableSortedLinkedListKilledException e) {
- throw new Error(e);
- }
- notifyAll();
- }
-
- private synchronized LinkedList makePriority(int p) {
- while(p >= clientsByPriority.size()) {
- clientsByPriority.add(new LinkedList());
- }
- return (LinkedList) clientsByPriority.get(p);
- }
-
public void run() {
long sentRequestTime = System.currentTimeMillis();
while(true) {
- RequestStarterClient client;
- client = getNextClient();
- Logger.minor(this, "getNextClient() = "+client);
- if(client != null) {
- boolean success;
- try {
- success = client.send(cycleNumber);
- } catch (Throwable t) {
- Logger.error(this, "Caught "+t);
- continue;
- }
- if(success) {
- sentRequestTime =
System.currentTimeMillis();
- Logger.minor(this, "Sent");
- if(client.isReady()) {
- synchronized(this) {
- try {
-
clientsReadyNextCycle.addOrUpdate(client);
- } catch
(UpdatableSortedLinkedListKilledException e) {
- // Impossible
- throw new
Error(e);
- }
- }
- }
- }
- }
- while(true) {
+ SendableRequest req = sched.removeFirst();
+ if(req != null) {
+ // Create a thread to handle starting the
request, and the resulting feedback
+ Thread t = new Thread(new SenderThread(req));
+ t.setDaemon(true);
+ t.start();
+ // Wait
long delay = throttle.getDelay();
long sleepUntil = sentRequestTime + delay;
- long now = System.currentTimeMillis();
- if(sleepUntil < now) {
- if(waitingClients()) break;
- // Otherwise wait for notification
- try {
- synchronized(this) {
- wait(1000);
- }
- } catch (InterruptedException e) {
- // Ignore
- }
- } else {
- Logger.minor(this,
"delay="+delay+"("+throttle+") sleep for "+(sleepUntil-now)+" for "+this);
- if(sleepUntil - now > 0)
+ long now;
+ do {
+ now = System.currentTimeMillis();
+ if(now < sleepUntil)
try {
- synchronized(this) {
- // At most
sleep 500ms, then recompute.
-
wait(Math.min(sleepUntil - now, 500));
- }
+ Thread.sleep(sleepUntil
- now);
} catch (InterruptedException
e) {
// Ignore
}
+ } while(now < sleepUntil);
+ } else {
+ synchronized(this) {
+ // Always take the lock on
RequestStarter first.
+ req = sched.removeFirst();
+ if(req != null) continue;
+ try {
+ wait(1000);
+ } catch (InterruptedException e) {
+ // Ignore
+ }
}
}
}
}
+
+ private class SenderThread implements Runnable {
- private synchronized boolean waitingClients() {
- return !(clientsReadyThisCycle.isEmpty() &&
clientsReadyNextCycle.isEmpty());
- }
+ private final SendableRequest req;
+
+ public SenderThread(SendableRequest req) {
+ this.req = req;
+ }
- /**
- * Get the next ready client.
- */
- private synchronized RequestStarterClient getNextClient() {
- try {
- while(true) {
- if(clientsReadyThisCycle.isEmpty() &&
clientsReadyNextCycle.isEmpty())
- return null;
- if(clientsReadyThisCycle.isEmpty()) {
- cycleNumber++;
-
clientsReadyNextCycle.moveTo(clientsReadyThisCycle);
- }
- RequestStarterClient c = (RequestStarterClient)
clientsReadyThisCycle.removeLowest();
- if(c.getCycleLastSent() == cycleNumber) {
- clientsReadyNextCycle.add(c);
- continue;
- } else {
- c.setCycleLastSet(cycleNumber);
- return c;
- }
- }
- } catch (UpdatableSortedLinkedListKilledException e) {
- throw new Error(e);
+ public void run() {
+ req.send(node);
}
+
}
+
}
Deleted: branches/async-client/src/freenet/node/RequestStarterClient.java
===================================================================
--- branches/async-client/src/freenet/node/RequestStarterClient.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/RequestStarterClient.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,118 +0,0 @@
-package freenet.node;
-
-import java.util.Vector;
-
-import freenet.crypt.RandomSource;
-import freenet.keys.ClientCHKBlock;
-import freenet.keys.ClientKey;
-import freenet.keys.ClientKeyBlock;
-import freenet.keys.ClientSSKBlock;
-import freenet.keys.KeyBlock;
-import freenet.support.DoublyLinkedList;
-import freenet.support.UpdatableSortedLinkedListItemImpl;
-
-/**
- * Interface to clients for starting a request.
- * Also represents a single client for fairness purposes.
- */
-public class RequestStarterClient extends UpdatableSortedLinkedListItemImpl {
-
- final int priority;
- private int random;
- private long cycleLastSent;
- private final Vector requests;
- private final RandomSource rs;
- private final QueueingSimpleLowLevelClient client;
- private final RequestStarter starter;
-
- public RequestStarterClient(short prioClass, short prio, RandomSource
r, QueueingSimpleLowLevelClient c, RequestStarter starter) {
- this((prioClass << 16) + prio, r, c, starter);
- }
-
- private RequestStarterClient(int prio, RandomSource r,
QueueingSimpleLowLevelClient c, RequestStarter starter) {
- priority = prio;
- this.random = r.nextInt();
- this.starter = starter;
- this.cycleLastSent = -1;
- this.requests = new Vector();
- this.rs = r;
- this.client = c;
- starter.registerClient(this);
- }
-
- /**
- * Blocking fetch of a key.
- * @throws LowLevelGetException If the fetch failed for some reason.
- */
- public ClientKeyBlock getKey(ClientKey key, boolean localOnly, boolean
cache, boolean ignoreStore) throws LowLevelGetException {
- QueuedDataRequest qdr = new QueuedDataRequest(key, localOnly,
cache, client, ignoreStore);
- addRequest(qdr);
- return qdr.waitAndFetch();
- }
-
- /**
- * Blocking insert of a key.
- * @throws LowLevelPutException If the fetch failed for some reason.
- */
- public void putKey(ClientKeyBlock block, boolean cache) throws
LowLevelPutException {
- QueuedInsertRequest qir = new QueuedInsertRequest(block,
client, cache);
- addRequest(qir);
- qir.waitAndPut();
- }
-
- void addRequest(QueuedRequest qr) {
- synchronized(this) {
- requests.add(qr);
- }
- if(starter != null)
- starter.notifyReady(this);
- }
-
- public long getCycleLastSent() {
- return cycleLastSent;
- }
-
- private DoublyLinkedList parentList;
-
- public DoublyLinkedList getParent() {
- return parentList;
- }
-
- public DoublyLinkedList setParent(DoublyLinkedList l) {
- DoublyLinkedList oldList = parentList;
- parentList = l;
- return oldList;
- }
-
- public int compareTo(Object o) {
- if(this == o) return 0;
- RequestStarterClient c = (RequestStarterClient) o;
- if(priority > c.priority) return 1;
- if(priority < c.priority) return -1;
- if(random > c.random) return 1;
- if(random < c.random) return -1;
- return 0;
- }
-
- public synchronized boolean isReady() {
- return !requests.isEmpty();
- }
-
- public boolean send(long cycleNumber) {
- QueuedRequest qr;
- synchronized(this) {
- if(!requests.isEmpty()) {
- int x = rs.nextInt(requests.size());
- qr = (QueuedRequest) requests.remove(x);
- } else qr = null;
- }
- if(qr == null) return false;
- qr.clearToSend();
- return true;
- }
-
- public void setCycleLastSet(long cycleNumber) {
- this.cycleLastSent = cycleNumber;
- }
-
-}
Deleted: branches/async-client/src/freenet/node/SimpleLowLevelClient.java
===================================================================
--- branches/async-client/src/freenet/node/SimpleLowLevelClient.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/SimpleLowLevelClient.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -1,31 +0,0 @@
-package freenet.node;
-
-import freenet.keys.ClientCHKBlock;
-import freenet.keys.ClientKey;
-import freenet.keys.ClientKeyBlock;
-import freenet.keys.KeyBlock;
-
-/**
- * @author amphibian
- *
- * Simple client interface... fetch and push single CHKs. No
- * splitfile decoding, no DBRs, no SSKs, for now.
- *
- * We can build higher layers on top of this.
- */
-public interface SimpleLowLevelClient {
-
- /**
- * Fetch a key. Throws if it cannot fetch it.
- * @param cache If false, don't cache the data. See the comments at the top
- * of Node.java.
- */
- public ClientKeyBlock getKey(ClientKey key, boolean localOnly,
RequestStarterClient client, boolean cache, boolean ignoreStore) throws
LowLevelGetException;
-
- /**
- * Insert a key.
- * @param cache If false, don't cache the data. See the comments at the top
- * of Node.java.
- */
- public void putKey(ClientKeyBlock key, RequestStarterClient sctx, boolean
cache) throws LowLevelPutException;
-}
Modified: branches/async-client/src/freenet/node/TextModeClientInterface.java
===================================================================
--- branches/async-client/src/freenet/node/TextModeClientInterface.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/TextModeClientInterface.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -51,19 +51,15 @@
final Node n;
final HighLevelSimpleClient client;
final Hashtable streams;
- final RequestStarterClient requestStarterClient;
- final RequestStarterClient insertStarterClient;
final File downloadsDir;
TextModeClientInterface(Node n) {
this.n = n;
- client = n.makeClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS,
(short)0);
+ client = n.makeClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS);
client.addGlobalHook(new EventDumper());
this.r = n.random;
streams = new Hashtable();
new Thread(this, "Text mode client interface").start();
- this.requestStarterClient =
n.makeStarterClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS, (short)0, false);
- this.insertStarterClient =
n.makeStarterClient(RequestStarter.INTERACTIVE_PRIORITY_CLASS, (short)0, true);
this.downloadsDir = n.downloadDir;
}
Modified: branches/async-client/src/freenet/node/fcp/ClientGet.java
===================================================================
--- branches/async-client/src/freenet/node/fcp/ClientGet.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/fcp/ClientGet.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -2,8 +2,11 @@
import freenet.client.FetchException;
import freenet.client.FetchResult;
-import freenet.client.Fetcher;
import freenet.client.FetcherContext;
+import freenet.client.InserterException;
+import freenet.client.async.ClientCallback;
+import freenet.client.async.ClientGetter;
+import freenet.client.async.ClientPutter;
import freenet.keys.FreenetURI;
import freenet.support.Logger;
@@ -11,17 +14,20 @@
* A simple client fetch. This can of course fetch arbitrarily large
* files, including splitfiles, redirects, etc.
*/
-public class ClientGet extends ClientRequest implements Runnable {
+public class ClientGet extends ClientRequest implements ClientCallback {
private final FreenetURI uri;
private final FetcherContext fctx;
- private final Fetcher f;
private final String identifier;
private final int verbosity;
private final FCPConnectionHandler handler;
+ private final ClientGetter getter;
+ private final short priorityClass;
public ClientGet(FCPConnectionHandler handler, ClientGetMessage
message) {
uri = message.uri;
+ // FIXME
+ this.priorityClass = 0;
// Create a Fetcher directly in order to get more fine-grained
control,
// since the client may override a few context elements.
this.handler = handler;
@@ -39,32 +45,38 @@
throw new IllegalStateException("Unknown return type:
"+message.returnType);
fctx.maxOutputLength = message.maxSize;
fctx.maxTempLength = message.maxTempSize;
- f = new Fetcher(uri, fctx);
- Thread t = new Thread(this, "FCP fetcher for "+uri+"
("+identifier+") on "+handler);
- t.setDaemon(true);
- t.start();
+ getter = new ClientGetter(this, handler.node.fetchScheduler,
uri, fctx, priorityClass);
}
public void cancel() {
fctx.cancel();
}
- public void run() {
- try {
- FetchResult fr = f.run();
- // Success!!!
- FCPMessage msg = new DataFoundMessage(handler, fr,
identifier);
- handler.outputHandler.queue(msg);
- // Send all the data at once
- // FIXME there should be other options
- msg = new AllDataMessage(handler, fr.asBucket(),
identifier);
- handler.outputHandler.queue(msg);
- } catch (FetchException e) {
- // Error
- Logger.minor(this, "Caught "+e, e);
- FCPMessage msg = new GetFailedMessage(handler, e,
identifier);
- handler.outputHandler.queue(msg);
- }
+ public void onSuccess(FetchResult result, ClientGetter state) {
+ FCPMessage msg = new DataFoundMessage(handler, result,
identifier);
+ handler.outputHandler.queue(msg);
+ // Send all the data at once
+ // FIXME there should be other options
+ msg = new AllDataMessage(handler, result.asBucket(),
identifier);
+ handler.outputHandler.queue(msg);
}
+ public void onFailure(FetchException e, ClientGetter state) {
+ Logger.minor(this, "Caught "+e, e);
+ FCPMessage msg = new GetFailedMessage(handler, e, identifier);
+ handler.outputHandler.queue(msg);
+ }
+
+ public void onSuccess(ClientPutter state) {
+ // Ignore
+ }
+
+ public void onFailure(InserterException e, ClientPutter state) {
+ // Ignore
+ }
+
+ public void onGeneratedURI(FreenetURI uri, ClientPutter state) {
+ // Ignore
+ }
+
}
Modified: branches/async-client/src/freenet/node/fcp/ClientPut.java
===================================================================
--- branches/async-client/src/freenet/node/fcp/ClientPut.java 2006-01-25
01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/fcp/ClientPut.java 2006-01-25
14:14:13 UTC (rev 7917)
@@ -1,51 +1,66 @@
package freenet.node.fcp;
import freenet.client.ClientMetadata;
-import freenet.client.FileInserter;
+import freenet.client.FetchException;
+import freenet.client.FetchResult;
import freenet.client.InsertBlock;
import freenet.client.InserterContext;
import freenet.client.InserterException;
+import freenet.client.async.ClientCallback;
+import freenet.client.async.ClientGetter;
+import freenet.client.async.ClientPutter;
import freenet.keys.FreenetURI;
-public class ClientPut extends ClientRequest implements Runnable {
+public class ClientPut extends ClientRequest implements ClientCallback {
final FreenetURI uri;
- final FileInserter inserter;
+ final ClientPutter inserter;
final InserterContext ctx;
final InsertBlock block;
final FCPConnectionHandler handler;
final String identifier;
final boolean getCHKOnly;
+ final short priorityClass;
public ClientPut(FCPConnectionHandler handler, ClientPutMessage
message) {
this.handler = handler;
this.identifier = message.identifier;
this.getCHKOnly = message.getCHKOnly;
+ this.priorityClass = 0;
ctx = new InserterContext(handler.defaultInsertContext);
+ ctx.maxInsertRetries = message.maxRetries;
// Now go through the fields one at a time
uri = message.uri;
String mimeType = message.contentType;
block = new InsertBlock(message.bucket, new
ClientMetadata(mimeType), uri);
- inserter = new FileInserter(ctx);
- ctx.maxInsertRetries = message.maxRetries;
- Thread t = new Thread(this, "FCP inserter for "+uri+"
("+identifier+") on "+handler);
- t.setDaemon(true);
- t.start();
+ inserter = new ClientPutter(this, message.bucket, uri, new
ClientMetadata(mimeType), ctx, handler.node.putScheduler, priorityClass,
getCHKOnly, false);
}
public void cancel() {
- ctx.cancel();
+ inserter.cancel();
}
- public void run() {
- try {
- FreenetURI uri = inserter.run(block, false, getCHKOnly,
false, null);
- FCPMessage msg = new PutSuccessfulMessage(identifier,
uri);
- handler.outputHandler.queue(msg);
- } catch (InserterException e) {
- FCPMessage msg = new PutFailedMessage(e, identifier);
- handler.outputHandler.queue(msg);
- }
+ public void onSuccess(ClientPutter state) {
+ FCPMessage msg = new PutSuccessfulMessage(identifier,
state.getURI());
+ handler.outputHandler.queue(msg);
}
+ public void onFailure(InserterException e, ClientPutter state) {
+ FCPMessage msg = new PutFailedMessage(e, identifier);
+ handler.outputHandler.queue(msg);
+ }
+
+ public void onGeneratedURI(FreenetURI uri, ClientPutter state) {
+ FCPMessage msg = new URIGeneratedMessage(uri, identifier);
+ handler.outputHandler.queue(msg);
+ }
+
+ public void onSuccess(FetchResult result, ClientGetter state) {
+ // ignore
+ }
+
+ public void onFailure(FetchException e, ClientGetter state) {
+ // ignore
+ }
+
}
Modified: branches/async-client/src/freenet/node/fcp/FCPConnectionHandler.java
===================================================================
--- branches/async-client/src/freenet/node/fcp/FCPConnectionHandler.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/fcp/FCPConnectionHandler.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -34,7 +34,7 @@
isClosed = false;
this.bf = node.tempBucketFactory;
requestsByIdentifier = new HashMap();
- HighLevelSimpleClient client =
node.makeClient((short)0,(short)0);
+ HighLevelSimpleClient client = node.makeClient((short)0);
defaultFetchContext = client.getFetcherContext();
defaultInsertContext = client.getInserterContext();
inputHandler.start();
Added: branches/async-client/src/freenet/node/fcp/URIGeneratedMessage.java
===================================================================
--- branches/async-client/src/freenet/node/fcp/URIGeneratedMessage.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/node/fcp/URIGeneratedMessage.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -0,0 +1,33 @@
+package freenet.node.fcp;
+
+import freenet.keys.FreenetURI;
+import freenet.node.Node;
+import freenet.support.SimpleFieldSet;
+
+public class URIGeneratedMessage extends FCPMessage {
+
+ private final FreenetURI uri;
+ private final String identifier;
+
+ public URIGeneratedMessage(FreenetURI uri, String identifier) {
+ this.uri = uri;
+ this.identifier = identifier;
+ }
+
+ public SimpleFieldSet getFieldSet() {
+ SimpleFieldSet fs = new SimpleFieldSet();
+ fs.put("URI", uri.toString());
+ fs.put("Identifier", identifier);
+ return fs;
+ }
+
+ public String getName() {
+ return "URIGenerated";
+ }
+
+ public void run(FCPConnectionHandler handler, Node node)
+ throws MessageInvalidException {
+ throw new
MessageInvalidException(ProtocolErrorMessage.INVALID_MESSAGE, "URIGenerated
goes from server to client not the other way around");
+ }
+
+}
Modified:
branches/async-client/src/freenet/support/UpdatableSortedLinkedList.java
===================================================================
--- branches/async-client/src/freenet/support/UpdatableSortedLinkedList.java
2006-01-25 01:47:43 UTC (rev 7916)
+++ branches/async-client/src/freenet/support/UpdatableSortedLinkedList.java
2006-01-25 14:14:13 UTC (rev 7917)
@@ -2,8 +2,6 @@
import java.util.Enumeration;
-import freenet.node.RequestStarterClient;
-
/**
* @author amphibian
*
