You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@river.apache.org by ni...@apache.org on 2008/12/10 06:15:14 UTC
svn commit: r724982 [1/7] - in
/incubator/river/jtsk/skunk/niclas1/services/mahalo: ./ src/ src/main/
src/main/java/ src/main/java/com/ src/main/java/com/sun/
src/main/java/com/sun/jini/ src/main/java/com/sun/jini/mahalo/
src/main/java/com/sun/jini/mah...
Author: niclas
Date: Tue Dec 9 21:15:12 2008
New Revision: 724982
URL: http://svn.apache.org/viewvc?rev=724982&view=rev
Log:
A first cut at the modularization of River Experiment. More refactorings to follow. Seems I need a multi-part commit, as SVN is complaining about individual files.
Added:
incubator/river/jtsk/skunk/niclas1/services/mahalo/
incubator/river/jtsk/skunk/niclas1/services/mahalo/pom.xml
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortJob.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ActivatableMahaloImpl.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitJob.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/InternalManagerException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/Job.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobNotStartedException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JoinStateManager.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/LeaseExpirationMgr.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/LeaseManager.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/MahaloPermission.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/NonActivatableMahaloImpl.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/PartialResultException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantAbortRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantCommitRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantHandle.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantListException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantModRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ParticipantTask.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/PrepareAndCommitJob.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/PrepareAndCommitRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/PrepareJob.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/PrepareRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ProxyVerifier.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ResultNotReadyException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/SettlerTask.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/StorableObject.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TransientMahaloImpl.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnLogRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnManager.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnManagerImpl.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnManagerInternalException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnManagerTransaction.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnMgrAdminProxy.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnMgrProxy.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/TxnSettler.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/UnknownTaskException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/build.xml
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/CannotRecoverException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/ClientLog.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/FileModes.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/InvalidatedLogException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/Log.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/LogException.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/LogManager.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/LogRecord.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/LogRecovery.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/MultiLogManager.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/MultiLogManagerAdmin.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/SimpleLogFile.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/TransientLogFile.java
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/log/package.html
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/package.html
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/configentry/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/configentry/mahalo-activatable
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/configentry/mahalo-persistent
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/configentry/mahalo-transient
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/resources/mahalo.mf
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/test/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/test/java/
incubator/river/jtsk/skunk/niclas1/services/mahalo/src/test/resources/
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/pom.xml
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/pom.xml?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/pom.xml (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/pom.xml Tue Dec 9 21:15:12 2008
@@ -0,0 +1,49 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.river.service</groupId>
+ <artifactId>master-services</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </parent>
+ <groupId>org.apache.river.service</groupId>
+ <artifactId>mahalo</artifactId>
+ <name>Apache River - Service - Mahalo</name>
+ <packaging>jar</packaging>
+ <version>10.0-SNAPSHOT</version>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.river.core</groupId>
+ <artifactId>river-core-api</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.river.api</groupId>
+ <artifactId>river-admin</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.river.api</groupId>
+ <artifactId>river-lease</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.river.api</groupId>
+ <artifactId>river-landlord</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.river.api</groupId>
+ <artifactId>river-lookup</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.river.service</groupId>
+ <artifactId>river-start</artifactId>
+ <version>10.0-SNAPSHOT</version>
+ </dependency>
+
+ </dependencies>
+</project>
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortJob.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortJob.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortJob.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortJob.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,337 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import com.sun.jini.mahalo.log.ClientLog;
+import com.sun.jini.thread.TaskManager;
+import com.sun.jini.thread.WakeupManager;
+import java.rmi.AccessException;
+import java.rmi.ConnectException;
+import java.rmi.ConnectIOException;
+import java.rmi.MarshalException;
+import java.rmi.NoSuchObjectException;
+import java.rmi.RemoteException;
+import java.rmi.UnknownHostException;
+import java.rmi.activation.ActivateFailedException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import net.jini.core.transaction.Transaction;
+import net.jini.core.transaction.TransactionException;
+import net.jini.core.transaction.server.ServerTransaction;
+import net.jini.core.transaction.server.TransactionConstants;
+import net.jini.core.transaction.server.TransactionParticipant;
+
+/**
+ * An implementation of a <code>com.sun.jini.mahalo.Job</code> which
+ * interacts with a set of
+ * <code>net.jini.core.transaction.server.TransactionParticipant</code>s
+ * to inform them to roll back changes associated with a given
+ * <code>net.jini.core.transaction.Transaction</code>.
+ *
+ * @author Sun Microsystems, Inc.
+ * @see com.sun.jini.mahalo.Job
+ * @see com.sun.jini.mahalo.ParticipantTask
+ * @see net.jini.core.transaction.Transaction
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ */
+public class AbortJob extends Job implements TransactionConstants
+{
+ ServerTransaction tr;
+ ClientLog log;
+ ParticipantHandle[] handles;
+ int maxtries = 5;
+ static final Logger logger = TxnManagerImpl.participantLogger;
+
+ /**
+ * Constructs an <code>AbortJob</code>
+ *
+ * @param tr The <code>Transaction</code> whose participants
+ * will be instructed to roll-back.
+ * @param pool The <code>TaskManager</code> which provides the
+ * threads used for interacting with participants.
+ * @param log The <code>ClientLog</code> used for recording
+ * recovery data.
+ * @param handles The array of participants which will be contacted
+ * and informed to roll-back.
+ * @see com.sun.jini.thread.TaskManager
+ * @see com.sun.jini.mahalo.log.ClientLog
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ */
+ public AbortJob( Transaction tr, TaskManager pool,
+ WakeupManager wm, ClientLog log,
+ ParticipantHandle[] handles )
+ {
+ super( pool, wm );
+
+ if( log == null )
+ {
+ throw new IllegalArgumentException( "AbortJob: AbortJob: " +
+ "log is null" );
+ }
+
+ this.log = log;
+
+ if( !( tr instanceof ServerTransaction ) )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must be a ServerTransaction" );
+ }
+
+ this.tr = (ServerTransaction) tr;
+
+ if( handles == null )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must have participants" );
+ }
+
+ if( handles.length == 0 )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must have participants" );
+ }
+
+ this.handles = handles;
+ }
+
+ /**
+ * Creates the <code>TaskManager.Task</code>s necessary to
+ * inform participants to roll-back.
+ */
+ TaskManager.Task[] createTasks()
+ {
+ TaskManager.Task[] tmp = new TaskManager.Task[handles.length];
+
+ for( int i = 0; i < handles.length; i++ )
+ {
+ tmp[ i ] =
+ new ParticipantTask( getPool(), getMgr(), this, handles[ i ] );
+ }
+
+ return tmp;
+ }
+
+ /**
+ * The work to be performed by each <code>TaskManager.Task</code>
+ * is provided by the <code>Job</code> that creates it.
+ * The work performed by a task belonging to the AbortJob
+ * contacts a participant, instructs it to roll-back and
+ * log appropriately.
+ *
+ * @param who The task performing the work
+ * @param param A parameter, of the task's choosing, useful
+ * in performing work.
+ * @see com.sun.jini.mahalo.Job
+ * @see com.sun.jini.thread.TaskManager.Task
+ */
+ Object doWork( TaskManager.Task who, Object param )
+ {
+ ParticipantHandle handle = (ParticipantHandle) param;
+ TransactionParticipant par = null;
+
+ //check if a vote already exists because it was
+ //recovered from the log. In this situation,
+ //we do not need to log this info since it
+ //exists in the log which was used for recovery...
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "AbortJob:doWork aborting handle: {0}", handle );
+ }
+ int vote = 0;
+
+ vote = handle.getPrepState();
+
+ switch( vote )
+ {
+ case NOTCHANGED:
+ case ABORTED:
+ return new Integer( ABORTED );
+ }
+
+ //Instruct the TransactionParticipant to roll back
+ //after unpacking it and checking against the
+ //max retry threshold.
+
+ if( par == null )
+ {
+ par = handle.getPreParedParticipant();
+ }
+
+ //If you have exhausted the max retry threshold
+ //stop, so that no further attempts are made.
+
+ try
+ {
+ if( attempt( who ) > maxtries )
+ {
+ return new Integer( ABORTED );
+ }
+ }
+ catch( JobException je )
+ {
+ return null;
+ }
+
+
+ //At this point, if participant is null, there
+ //must be an error unpacking, so retry later
+ if( par == null )
+ {
+ return null;
+ }
+
+
+ //Here we actually need to instruct the participant to
+ //roll back. Note the RemoteException causes a
+ //retry. Here we only log info for the cases
+ //where a final outcome is available.
+ //
+ //Note: handle UnknownHostException and
+ // NoSuchObjectException (activation problem)
+
+ Object response = null;
+
+ try
+ {
+ par.abort( tr.mgr, tr.id );
+ response = new Integer( ABORTED );
+ }
+ catch( TransactionException bte )
+ {
+ //The participant doesn't have record of the
+ //transaction, so it must have already rolled
+ //back.
+ response = new Integer( ABORTED );
+ }
+ catch( NoSuchObjectException nsoe )
+ {
+ //No definition for object in VM, so stop
+ //and consider committed.
+ response = new Integer( ABORTED );
+ }
+ catch( ConnectException ce )
+ {
+ //failure setting up connection, so give
+ //participant more time by retrying
+ }
+ catch( UnknownHostException uhe )
+ {
+ //could not resolve host for participant, so
+ //stop and consider committed
+ response = new Integer( ABORTED );
+ }
+ catch( ConnectIOException cioe )
+ {
+ //multiplexed connection or cached
+ //connection problem, give participant more time
+ }
+ catch( MarshalException me )
+ {
+ //cannot send parameters, so stop and consider done
+ response = new Integer( ABORTED );
+ }
+ catch( AccessException ae )
+ {
+ //Access error on registry or rmid consider done
+ response = new Integer( ABORTED );
+ }
+ catch( ActivateFailedException afe )
+ {
+ //Activatable Ref Stub couldn't activate
+ //participant, so stop and consider done
+ response = new Integer( ABORTED );
+ }
+ catch( RemoteException re )
+ {
+ //Something happened with the network, so
+ //retry at a later time.
+ }
+ catch( RuntimeException rte )
+ {
+ //Something happened with the participant, so
+ //stop retrying
+ response = new Integer( ABORTED );
+ }
+
+ if( response != null )
+ {
+ handle.setPrepState( ABORTED );
+ try
+ {
+ log.write( new ParticipantAbortRecord( handle ) );
+ }
+ catch( com.sun.jini.mahalo.log.LogException le )
+ {
+ //the full package name used to disambiguate
+ //the LogException
+ }
+
+ return response;
+ }
+
+ return null;
+ }
+
+ /**
+ * Gathers partial results submitted by tasks and produces
+ * a single outcome.
+ *
+ * @see com.sun.jini.mahalo.Job
+ */
+ Object computeResult() throws JobException
+ {
+ try
+ {
+ if( !isCompleted( 0 ) )
+ {
+ throw new ResultNotReadyException( "Cannot compute result " +
+ "since there are jobs pending" );
+ }
+ }
+ catch( JobNotStartedException jnse )
+ {
+ throw new ResultNotReadyException( "Cannot compute result since" +
+ " jobs were not created" );
+ }
+
+ int tmp = 0;
+ int count = 0;
+
+ checkresults:
+ for( int i = 0; i < results.length; i++ )
+ {
+ tmp = ( (Integer) results[ i ] ).intValue();
+
+ if( tmp == ABORTED )
+ {
+ count++;
+ }
+ }
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "AbortJob:computeResult {0} participants ABORTED",
+ new Integer( count ) );
+ }
+
+ return new Integer( ABORTED );
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortRecord.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortRecord.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortRecord.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/AbortRecord.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import com.sun.jini.mahalo.log.CannotRecoverException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * An <code>AbortRecord</code> represents the logged state of
+ * a <code>Transaction</code> which has changed to the ABORTED
+ * state.
+ *
+ * @author Sun Microsystems, Inc.
+ */
+class AbortRecord implements TxnLogRecord
+{
+ /**
+ * @serial
+ */
+ private ParticipantHandle[] parts;
+
+ static final long serialVersionUID = -8121722031382234695L;
+
+ static final Logger logger = TxnManagerImpl.participantLogger;
+
+ /**
+ * Constructs an <code>AbortRecord</code> which represents a
+ * <code>Transaction</code> which has moved to the ABORTED state.
+ *
+ * @param parts The array of participants joined in the transaction
+ * @see net.jini.core.transaction.Transaction
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ * @see net.jini.core.transaction.server.TransactionConstants
+ */
+ AbortRecord( ParticipantHandle[] parts )
+ {
+ if( parts == null )
+ {
+ throw new IllegalArgumentException( "AbortRecord: must specify " +
+ "a non-null parts array" );
+ }
+ this.parts = parts;
+ }
+
+ /**
+ * Recovers the state encapsulated the <code>AbortRecord</code> to
+ * the caller.
+ *
+ * @param tmt The <code>TxnManagerTransaction</code> to which
+ * state is recovered.
+ * @see com.sun.jini.mahalo.TxnManagerTransaction
+ */
+ public void recover( TxnManagerTransaction tmt )
+ throws CannotRecoverException
+ {
+ try
+ {
+ for( int i = 0; i < parts.length; i++ )
+ {
+ tmt.add( parts[ i ] );
+ }
+ tmt.modifyTxnState( ABORTED );
+ }
+ catch( InternalManagerException ime )
+ {
+ throw new CannotRecoverException( "AbortRecord: recover: " +
+ ime.getMessage() );
+ }
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST, "AbortJob:recover recovered" );
+ }
+ }
+
+ /**
+ * Retrieves the set of <code>TransactionParticipant</code>s associated
+ * with the recovered <code>Transaction</code>.
+ */
+ ParticipantHandle[] getParts()
+ {
+ return parts;
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ActivatableMahaloImpl.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ActivatableMahaloImpl.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ActivatableMahaloImpl.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/ActivatableMahaloImpl.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import java.rmi.MarshalledObject;
+import java.rmi.activation.ActivationID;
+
+/**
+ * Convenience class intended for use with the
+ * {@link com.sun.jini.start.ServiceStarter} framework to start
+ * an implementation of Mahalo that is activatable, and which will
+ * log its state information to persistent storage.
+ *
+ * @author Sun Microsystems, Inc.
+ * @since 2.0
+ */
+class ActivatableMahaloImpl extends TxnManagerImpl
+{
+
+ /**
+ * Constructs a new instance of <code>TxnManagerImpl</code> that is
+ * activatable, and which will persist its state.
+ * <p>
+ * A constructor having this signature is required for the class to
+ * be activatable. This constructor is automatically called by the
+ * activation group when the service is activated.
+ *
+ * @param activationID the activation ID generated by the activation
+ * system and assigned to the instance of the server
+ * being activated
+ * @param data state data (represented as a
+ * <code>MarshalledObject</code>) which is needed to
+ * re-activate this server
+ * @throws Exception If there was a problem initializing the service.
+ */
+ ActivatableMahaloImpl( ActivationID activationID, MarshalledObject data )
+ throws Exception
+ {
+ super( activationID, data );
+ }//end constructor
+
+}//end class ActivatableMahaloImpl
+
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitJob.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitJob.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitJob.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitJob.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,333 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import com.sun.jini.mahalo.log.ClientLog;
+import com.sun.jini.thread.TaskManager;
+import com.sun.jini.thread.WakeupManager;
+import java.rmi.AccessException;
+import java.rmi.ConnectException;
+import java.rmi.ConnectIOException;
+import java.rmi.MarshalException;
+import java.rmi.NoSuchObjectException;
+import java.rmi.RemoteException;
+import java.rmi.UnknownHostException;
+import java.rmi.activation.ActivateFailedException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import net.jini.core.transaction.Transaction;
+import net.jini.core.transaction.TransactionException;
+import net.jini.core.transaction.server.ServerTransaction;
+import net.jini.core.transaction.server.TransactionConstants;
+import net.jini.core.transaction.server.TransactionParticipant;
+
+/**
+ * An implementation of a <code>Job</code> which interacts with
+ * a set of <code>TransactionParticipant</code>s to inform them
+ * to roll forward changes associated with a given <code>Transaction</code>.
+ *
+ * @author Sun Microsystems, Inc.
+ * @see com.sun.jini.mahalo.Job
+ * @see com.sun.jini.mahalo.ParticipantTask
+ * @see net.jini.core.transaction.Transaction
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ */
+public class CommitJob extends Job implements TransactionConstants
+{
+ ServerTransaction tr;
+ ClientLog log;
+ ParticipantHandle[] handles;
+ int maxtries = Integer.MAX_VALUE;
+ static final Logger logger = TxnManagerImpl.participantLogger;
+
+ /**
+ * Constructs an <code>CommitJob</code>
+ *
+ * @param tr The <code>Transaction</code> whose participants
+ * will be instructed to roll-forward.
+ * @param pool The <code>TaskManager</code> which provides the
+ * threads used for interacting with participants.
+ * @param log The <code>ClientLog</code> used for recording
+ * recovery data.
+ * @param handles The array of participants which will be contacted
+ * and informed to roll-forward.
+ * @see com.sun.jini.thread.TaskManager
+ * @see com.sun.jini.mahalo.log.ClientLog
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ */
+ public CommitJob( Transaction tr, TaskManager pool,
+ WakeupManager wm, ClientLog log,
+ ParticipantHandle[] handles )
+ {
+ super( pool, wm );
+
+ if( log == null )
+ {
+ throw new IllegalArgumentException( "CommitJob: CommitJob: " +
+ "log is null" );
+ }
+
+ this.log = log;
+
+ if( !( tr instanceof ServerTransaction ) )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must be a ServerTransaction" );
+ }
+
+ this.tr = (ServerTransaction) tr;
+
+ if( handles == null )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must have participants" );
+ }
+
+ if( handles.length == 0 )
+ {
+ throw new IllegalArgumentException( "PrepareJob: PrepareJob: " +
+ "must have participants" );
+ }
+
+ this.handles = handles;
+ }
+
+
+ /**
+ * The work to be performed by each <code>TaskManager.Task</code>
+ * is provided by the <code>Job</code> that creates it.
+ * The work performed by a task belonging to the CommitJob
+ * contacts a participant, instructs it to roll-forward and
+ * log appropriately.
+ *
+ * @param who The task performing the work
+ * @param param A parameter, of the task's choosing, useful
+ * in performing work.
+ * @see com.sun.jini.mahalo.Job
+ * @see com.sun.jini.thread.TaskManager.Task
+ */
+ Object doWork( TaskManager.Task who, Object param )
+ {
+ ParticipantHandle handle = (ParticipantHandle) param;
+ TransactionParticipant par = null;
+
+ //Check to see if participant has state associated
+ //with it on account of being recovered from a log
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "CommitJob:doWork committing handle: {0}", handle );
+ }
+
+ int vote = handle.getPrepState();
+
+ switch( vote )
+ {
+ case COMMITTED:
+ case NOTCHANGED:
+ case ABORTED:
+ return new Integer( vote );
+ }
+
+ //Instruct the TransactionParticipant to roll forward
+ //after unpacking it and checking against the
+ //max retry threshold.
+
+ if( par == null )
+ {
+ par = handle.getPreParedParticipant();
+ }
+
+ //If you have exhausted the max retry threshold
+ //stop, so that no further attempts are made.
+
+ try
+ {
+ if( attempt( who ) > maxtries )
+ {
+ return new Integer( COMMITTED );
+ }
+ }
+ catch( JobException je )
+ {
+ return null;
+ }
+
+ //At this point, if participant is null, there
+ //must be an error unpacking, so retry later
+ if( par == null )
+ {
+ return null;
+ }
+
+
+ //Here we actually need to instruct the participant to
+ //roll forward. Note the RemoteException causes a
+ //retry. Here we only log info for the cases
+ //where a final outcome is available.
+
+ Object response = null;
+
+ try
+ {
+ par.commit( tr.mgr, tr.id );
+ response = new Integer( COMMITTED );
+ }
+ catch( TransactionException bte )
+ {
+ //The participant doesn't have record of the
+ //transaction, so it must have already rolled
+ //forward.
+ response = new Integer( COMMITTED );
+ }
+ catch( NoSuchObjectException nsoe )
+ {
+ //No definition for object in VM, so stop
+ //and consider committed.
+ response = new Integer( COMMITTED );
+ }
+ catch( ConnectException ce )
+ {
+ //failure setting up connection, so give
+ //participant more time by retrying
+ }
+ catch( UnknownHostException uhe )
+ {
+ //could not resolve host for participant, so
+ //stop and consider committed
+ response = new Integer( COMMITTED );
+ }
+ catch( ConnectIOException cioe )
+ {
+ //multiplexed connection or cached
+ //connection problem, give participant more time
+ }
+ catch( MarshalException me )
+ {
+ //cannot send parameters, so stop and consider done
+ response = new Integer( COMMITTED );
+ }
+ catch( AccessException ae )
+ {
+ //Access error on registry or rmid consider done
+ response = new Integer( COMMITTED );
+ }
+ catch( ActivateFailedException afe )
+ {
+ //Activatable Ref Stub couldn't activate
+ //participant, so stop and consider done
+ response = new Integer( COMMITTED );
+ }
+ catch( RemoteException re )
+ {
+ //Something happened with the network, so
+ //return null to retry at a later time.
+ }
+ catch( RuntimeException rte )
+ {
+ //Something happened with the participant, so
+ //stop and consider done
+ response = new Integer( COMMITTED );
+ }
+
+
+ if( response != null )
+ {
+ handle.setPrepState( COMMITTED );
+ try
+ {
+ log.write( new ParticipantCommitRecord( handle ) );
+ }
+ catch( com.sun.jini.mahalo.log.LogException le )
+ {
+ //the full package name used to disambiguate
+ //the LogException
+ }
+
+ return response;
+ }
+
+ return null;
+ }
+
+
+ /**
+ * Creates the <code>TaskManager.Task</code>s necessary to
+ * inform participants to roll-back.
+ */
+ TaskManager.Task[] createTasks()
+ {
+ TaskManager.Task[] tmp = new TaskManager.Task[handles.length];
+
+ for( int i = 0; i < handles.length; i++ )
+ {
+ tmp[ i ] =
+ new ParticipantTask( getPool(), getMgr(), this, handles[ i ] );
+ }
+
+ return tmp;
+ }
+
+
+ /**
+ * Gathers partial results submitted by tasks and produces
+ * a single outcome.
+ *
+ * @see com.sun.jini.mahalo.Job
+ */
+ Object computeResult() throws JobException
+ {
+ try
+ {
+ if( !isCompleted( 0 ) )
+ {
+ throw new ResultNotReadyException( "Cannot compute result " +
+ "since there are jobs pending" );
+ }
+ }
+ catch( JobNotStartedException jnse )
+ {
+ throw new ResultNotReadyException( "Cannot compute result since" +
+ " jobs were not created" );
+ }
+
+ int tmp = 0;
+ int count = 0;
+
+ checkresults:
+ for( int i = 0; i < results.length; i++ )
+ {
+ tmp = ( (Integer) results[ i ] ).intValue();
+
+ if( tmp == COMMITTED )
+ {
+ count++;
+ }
+ }
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "CommitJob:computeResult {0} participants COMMITTED",
+ new Integer( count ) );
+ }
+
+ return new Integer( COMMITTED );
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitRecord.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitRecord.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitRecord.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/CommitRecord.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import com.sun.jini.mahalo.log.CannotRecoverException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A <code>CommitRecord</code> represents the logged state of
+ * a <code>Transaction</code> which has moved to the COMMITTED
+ * state.
+ *
+ * @author Sun Microsystems, Inc.
+ */
+class CommitRecord implements TxnLogRecord
+{
+ static final long serialVersionUID = 5706802011705258126L;
+
+ static final Logger logger = TxnManagerImpl.participantLogger;
+
+ /**
+ * @serial
+ */
+ ParticipantHandle[] parts; //Note: Use an array of ParticipantHandles;
+ // We want a list of things. By using
+ // an array, we can use the type system
+ // to guarantee that each thing is a
+ // ParticipantHandle rather than checking
+ // explicitly.
+
+ /**
+ * Constructs an <code>CommitRecord</code> which represents a
+ * <code>Transaction</code> which has moved to the COMMITTED state.
+ *
+ * @param parts The array of participants joined in the transaction
+ * @see net.jini.core.transaction.Transaction
+ * @see net.jini.core.transaction.server.TransactionParticipant
+ * @see net.jini.core.transaction.server.TransactionConstants
+ */
+ CommitRecord( ParticipantHandle parts[] )
+ {
+ //Note: the state is implied in the
+ // class name
+
+ if( parts == null )
+ {
+ throw new IllegalArgumentException( "CommitRecord: must specify " +
+ "a non-null parts array" );
+ }
+
+ this.parts = parts;
+ }
+
+
+ /**
+ * Retrieves the set of <code>TransactionParticipant</code>s associated
+ * with the recovered <code>Transaction</code>.
+ */
+ ParticipantHandle[] getParts()
+ {
+ return parts;
+ }
+
+
+ /**
+ * Recovers the state encapsulated the <code>CommitRecord</code> to
+ * the caller.
+ *
+ * @param tmt The <code>TxnManagerTransaction</code> to which
+ * state is recovered.
+ * @see com.sun.jini.mahalo.TxnManagerTransaction
+ */
+ public void recover( TxnManagerTransaction tmt )
+ throws CannotRecoverException
+ {
+ try
+ {
+ for( int i = 0; i < parts.length; i++ )
+ {
+ tmt.add( parts[ i ] );
+ }
+ tmt.modifyTxnState( VOTING );
+ }
+ catch( InternalManagerException ime )
+ {
+ throw new CannotRecoverException( "CommitRecord: recover: " +
+ ime.getMessage() );
+ }
+
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST, "CommitRecord:recover recovered" );
+ }
+ }
+
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/InternalManagerException.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/InternalManagerException.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/InternalManagerException.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/InternalManagerException.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+/**
+ * This exception denotes a problem with the transaction manager
+ * implementation.
+ *
+ * @author Sun Microsystems, Inc.
+ */
+class InternalManagerException extends RuntimeException
+{
+ static final long serialVersionUID = -8249846562594717389L;
+
+ InternalManagerException()
+ {
+ super();
+ }
+
+ InternalManagerException( String msg )
+ {
+ super( msg );
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/Job.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/Job.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/Job.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/Job.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,433 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+import com.sun.jini.thread.TaskManager;
+import com.sun.jini.thread.WakeupManager;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A <code>Job</code> manages the division of work for a problem
+ * whose solution is obtained by assembling partial results to
+ * original problem.
+ *
+ * @author Sun Microsystems, Inc.
+ */
+public abstract class Job
+{
+ private TaskManager pool;
+ private WakeupManager wm;
+ private int pending = -1;
+ Object[] results;
+ int[] attempts;
+ private Map tasks = new HashMap(); //used to maintain account
+ //of the tasks for which
+ //the job is responsible
+
+ static final Logger logger = TxnManagerImpl.participantLogger;
+
+ /**
+ * Create the <code>Job</code> object giving it the
+ * <code>TaskManager</code> responsible for the pool of
+ * threads which perform the necessary work.
+ *
+ * @param pool the <code>TaskManager</code> which provides the threads
+ */
+ public Job( TaskManager pool, WakeupManager wm )
+ {
+ this.wm = wm;
+ this.pool = pool;
+ }
+
+
+ /**
+ * Used by a task to do a piece of work and record the
+ * number of attempts.
+ *
+ * @param who The task which is performing the work
+ * @param param A parameter used in performing the work
+ */
+ boolean performWork( TaskManager.Task who, Object param )
+ throws JobException
+ {
+ Integer tmp = null;
+
+ synchronized( tasks )
+ {
+ tmp = (Integer) tasks.get( who );
+ }
+
+ if( tmp == null )
+ {
+ throw new UnknownTaskException();
+ }
+
+ int rank = tmp.intValue();
+
+ synchronized( attempts )
+ {
+ attempts[ rank ]++;
+ }
+
+ Object result = doWork( who, param );
+ if( result == null )
+ {
+ return false;
+ }
+
+ try
+ {
+ reportDone( who, result );
+ }
+ catch( UnknownTaskException e )
+ {
+ }
+ catch( PartialResultException e )
+ {
+ }
+ catch( JobException e )
+ {
+ }
+
+ return true;
+ }
+
+
+ /**
+ * Given a <code>TaskManager.Task</code>, this method
+ * returns the current number of attempts it has made.
+ *
+ * @param who The task for which the number of attempts
+ * is inquired
+ */
+ int attempt( TaskManager.Task who ) throws JobException
+ {
+ Integer tmp = null;
+
+ synchronized( tasks )
+ {
+ tmp = (Integer) tasks.get( who );
+ }
+
+ if( tmp == null )
+ {
+ throw new UnknownTaskException();
+ }
+
+ int rank = tmp.intValue();
+
+ synchronized( attempts )
+ {
+ return attempts[ rank ];
+ }
+ }
+
+
+ /**
+ * The work performed is implemented here.
+ * A null return value indicates failure
+ * while a non-null return value indicates
+ * success and contains the result.
+ *
+ * @param who The task performing the work
+ * @param param A parameter used to do the work
+ */
+ abstract Object doWork( TaskManager.Task who, Object param )
+ throws JobException;
+
+ /**
+ * Create the tasks required to compute all of the
+ * <code>PartialResult</code> objects necessary for the
+ * solution to the original problem.
+ */
+ abstract TaskManager.Task[] createTasks();
+
+
+ /**
+ * Schedules tasks for execution
+ */
+ public void scheduleTasks()
+ {
+ TaskManager.Task[] tmp = createTasks();
+
+ if( tmp != null )
+ {
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:scheduleTasks with {0} tasks",
+ new Integer( tmp.length ) );
+ }
+
+ results = new Object[tmp.length];
+ attempts = new int[tmp.length];
+ setPending( tmp.length );
+
+ for( int i = 0; i < tmp.length; i++ )
+ {
+
+ //Record the position if each
+ //task for later use when assembling
+ //the partial results
+
+ synchronized( tasks )
+ {
+ tasks.put( tmp[ i ], new Integer( i ) );
+ pool.add( tmp[ i ] );
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:scheduleTasks added {0} to thread pool",
+ tmp[ i ] );
+ }
+ attempts[ i ] = 0;
+ }
+ }
+ }
+ }
+
+
+ private synchronized void awaitPending( long waitFor )
+ {
+ if( pending < 0 )
+ {
+ return;
+ }
+
+ if( pending == 0 )
+ {
+ return;
+ }
+
+ try
+ {
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:awaitPending waiting for {0} items",
+ new Integer( pending ) );
+ }
+
+ if( waitFor == Long.MAX_VALUE )
+ {
+ while( pending > 0 )
+ {
+ wait();
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:awaitPending awoken" );
+ }
+ }
+ }
+ else
+ {
+ //When waiting for a given amount of time,
+ //if notified, make sure that the desired
+ //wait time has actually transpired.
+
+ long start = System.currentTimeMillis();
+ long curr = start;
+
+ while( ( pending > 0 ) && ( ( curr - start ) < waitFor ) )
+ {
+ wait( waitFor - ( curr - start ) );
+ curr = System.currentTimeMillis();
+ }
+ }
+ }
+ catch( InterruptedException ie )
+ {
+ }
+ }
+
+ private synchronized void setPending( int num )
+ {
+ pending = num;
+
+ if( pending <= 0 )
+ {
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:setPending notifying, pending = {0}",
+ new Integer( pending ) );
+ }
+ notifyAll();
+ }
+ }
+
+ private synchronized void decrementPending()
+ {
+ pending--;
+
+ if( pending <= 0 )
+ {
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:decrementPending notifying, pending = {0}",
+ new Integer( pending ) );
+ }
+ notifyAll();
+ }
+ }
+
+
+ /**
+ * Returns a reference to the <code>TaskManager</code> which
+ * supplies the threads used to executed tasks created by
+ * this <code>Job</code>
+ */
+ protected TaskManager getPool()
+ {
+ return pool;
+ }
+
+ /**
+ * Returns a reference to the <code>WakeupManager</code> which
+ * provides the scheduling of tasks created by
+ * this <code>Job</code>
+ */
+ protected WakeupManager getMgr()
+ {
+ return wm;
+ }
+
+ /*
+ * Tasks which perform work on behalf of the <code>Job</code>
+ * report in that they are done using this method.
+ */
+ private void reportDone( TaskManager.Task who, Object param )
+ throws JobException
+ {
+ if( param == null )
+ {
+ throw new NullPointerException( "param must be non-null" );
+ }
+
+ if( who == null )
+ {
+ throw new NullPointerException( "task must be non-null" );
+ }
+
+ Integer position = null;
+
+ synchronized( tasks )
+ {
+ position = (Integer) tasks.get( who );
+ }
+
+ if( position == null )
+ {
+ throw new UnknownTaskException();
+ }
+
+ synchronized( results )
+ {
+ if( results[ position.intValue() ] == null )
+ {
+ if( logger.isLoggable( Level.FINEST ) )
+ {
+ logger.log( Level.FINEST,
+ "Job:reportDone who = {0}, param = {1}",
+ new Object[]{ who, param } );
+ }
+ results[ position.intValue() ] = param;
+ decrementPending();
+ }
+ else
+ {
+ throw new PartialResultException( "result already set" );
+ }
+ }
+ }
+
+
+ /**
+ * Check to see if the <code>Job</code> execution has
+ * completed.
+ *
+ * @param waitFor The amount of time the caller is willing
+ * to wait for the completion status to arrive.
+ */
+ public boolean isCompleted( long waitFor ) throws JobException
+ {
+ //If nothing has started, the
+ //task could not have completed.
+ //Less than zero means initial value
+ //and greater than zero means there
+ //are outstanding tasks. In each of
+ //these cases, the Job is not done.
+
+ awaitPending( waitFor );
+
+ synchronized( this )
+ {
+ if( pending == 0 )
+ {
+ return true;
+ }
+
+ if( pending < 0 )
+ {
+ throw new JobNotStartedException( "No jobs started" );
+ }
+
+ return false;
+ }
+ }
+
+
+ /**
+ * Generate the solution to the original problem.
+ * The subclass decides how it computes the final
+ * outcome.
+ */
+ abstract Object computeResult() throws JobException;
+
+
+ /**
+ * Halt all of the work being performed by
+ * the <code>Job</code>
+ */
+ public void stop()
+ {
+ Set s = tasks.keySet();
+ Object[] vals = s.toArray();
+
+ //Remove and interrupt all tasks
+
+ for( int i = 0; i < vals.length; i++ )
+ {
+ TaskManager.Task t = (TaskManager.Task) vals[ i ];
+ pool.remove( t );
+ }
+
+ //Erase record of tasks, results and the
+ //counting mechanism
+
+ tasks = new HashMap();
+ setPending( -1 );
+ results = null;
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobException.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobException.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobException.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobException.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+/**
+ * @author Sun Microsystems, Inc.
+ */
+public class JobException extends Exception
+{
+ static final long serialVersionUID = 4182966504752382657L;
+
+ public JobException()
+ {
+ super();
+ }
+
+ public JobException( String msg )
+ {
+ super( msg );
+ }
+}
Added: incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobNotStartedException.java
URL: http://svn.apache.org/viewvc/incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobNotStartedException.java?rev=724982&view=auto
==============================================================================
--- incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobNotStartedException.java (added)
+++ incubator/river/jtsk/skunk/niclas1/services/mahalo/src/main/java/com/sun/jini/mahalo/JobNotStartedException.java Tue Dec 9 21:15:12 2008
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.sun.jini.mahalo;
+
+/**
+ * @author Sun Microsystems, Inc.
+ */
+public class JobNotStartedException extends JobException
+{
+ static final long serialVersionUID = -454913962920091805L;
+
+ public JobNotStartedException()
+ {
+ super();
+ }
+
+ public JobNotStartedException( String msg )
+ {
+ super( msg );
+ }
+}