ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yzhda...@apache.org
Subject [06/41] incubator-ignite git commit: futures: api cleanup
Date Mon, 09 Mar 2015 20:49:32 GMT
futures: api cleanup


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/bc18635a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/bc18635a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/bc18635a

Branch: refs/heads/sprint-2
Commit: bc18635a9358df5f155eba69f30d74410ae5d3e7
Parents: c1b4695
Author: Yakov Zhdanov <yzhdanov@gridgain.com>
Authored: Thu Mar 5 11:02:08 2015 +0300
Committer: Yakov Zhdanov <yzhdanov@gridgain.com>
Committed: Thu Mar 5 11:02:08 2015 +0300

----------------------------------------------------------------------
 .../cache/GridCacheMultiTxFuture.java           |  40 +-----
 .../internal/util/future/GridFutureAdapter.java | 122 ++++++++++------
 .../processors/schedule/ScheduleFutureImpl.java | 140 +------------------
 3 files changed, 95 insertions(+), 207 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/bc18635a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMultiTxFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMultiTxFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMultiTxFuture.java
index c6829aa..71ba123 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMultiTxFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMultiTxFuture.java
@@ -25,7 +25,6 @@ import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 
-import java.io.*;
 import java.util.*;
 import java.util.concurrent.atomic.*;
 
@@ -33,15 +32,9 @@ import java.util.concurrent.atomic.*;
  * Future which waits for completion of one or more transactions.
  */
 public final class GridCacheMultiTxFuture<K, V> extends GridFutureAdapter<Boolean>
{
-    /** */
-    private static final long serialVersionUID = 0L;
-
     /** Logger reference. */
     private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
 
-    /** Transactions to wait for. */
-    private final Set<IgniteInternalTx<K, V>> txs = new GridLeanSet<>();
-
     /** */
     private Set<IgniteInternalTx<K, V>> remainingTxs;
 
@@ -58,46 +51,28 @@ public final class GridCacheMultiTxFuture<K, V> extends GridFutureAdapter<Boolea
     }
 
     /**
-     * Empty constructor required for {@link Externalizable}.
-     */
-    public GridCacheMultiTxFuture() {
-        // No-op.
-    }
-
-    /**
-     * @return Transactions to wait for.
-     */
-    public Set<IgniteInternalTx<K, V>> txs() {
-        return txs;
-    }
-
-    /**
-     * @return Remaining transactions.
-     */
-    public Set<IgniteInternalTx<K, V>> remainingTxs() {
-        return remainingTxs;
-    }
-
-    /**
      * @param tx Transaction to add.
      */
     public void addTx(IgniteInternalTx<K, V> tx) {
-        txs.add(tx);
+        if (remainingTxs == null)
+            remainingTxs = new GridConcurrentHashSet<>();
+
+        remainingTxs.add(tx);
     }
 
     /**
      * Initializes this future.
      */
     public void init() {
-        if (F.isEmpty(txs)) {
+        if (remainingTxs == null) {
             remainingTxs = Collections.emptySet();
 
             onDone(true);
         }
         else {
-            remainingTxs = new GridConcurrentHashSet<>(txs);
+            assert !remainingTxs.isEmpty();
 
-            for (final IgniteInternalTx<K, V> tx : txs) {
+            for (final IgniteInternalTx<K, V> tx : remainingTxs) {
                 if (!tx.done()) {
                     tx.finishFuture().listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>()
{
                         @Override public void apply(IgniteInternalFuture<IgniteInternalTx>
t) {
@@ -134,7 +109,6 @@ public final class GridCacheMultiTxFuture<K, V> extends GridFutureAdapter<Boolea
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridCacheMultiTxFuture.class, this,
-            "txs", F.viewReadOnly(txs, CU.<K, V>tx2xidVersion()),
             "remaining", F.viewReadOnly(remainingTxs, CU.<K, V>tx2xidVersion()),
             "super", super.toString()
         );

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/bc18635a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
index b107c15..b35d18a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridFutureAdapter.java
@@ -32,9 +32,6 @@ import java.util.concurrent.locks.*;
  * Future adapter.
  */
 public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer implements IgniteInternalFuture<R>
{
-    /** */
-    private static final long serialVersionUID = 0L;
-
     /** Initial state. */
     private static final int INIT = 0;
 
@@ -44,12 +41,18 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
     /** Done state. */
     private static final int DONE = 2;
 
+    /** */
+    private static final byte ERR = 1;
+
+    /** */
+    private static final byte RES = 2;
+
+    /** */
+    private byte resFlag;
+
     /** Result. */
     @GridToStringInclude
-    private R res;
-
-    /** Error. */
-    private Throwable err;
+    private Object res;
 
     /** Future start time. */
     private final long startTime = U.currentTimeMillis();
@@ -57,11 +60,8 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
     /** Future end time. */
     private volatile long endTime;
 
-    /** Asynchronous listeners. */
-    private Collection<IgniteInClosure<? super IgniteInternalFuture<R>>>
lsnrs;
-
     /** */
-    private final Object mux = new Object();
+    private IgniteInClosure<? super IgniteInternalFuture<R>> lsnr;
 
     /** {@inheritDoc} */
     @Override public long startTime() {
@@ -86,14 +86,14 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
      * @return Value of error.
      */
     protected Throwable error() {
-        return err;
+        return (resFlag == RES) ? null : (Throwable)res;
     }
 
     /**
      * @return Value of result.
      */
     protected R result() {
-        return res;
+        return (R)res;
     }
 
     /** {@inheritDoc} */
@@ -105,10 +105,10 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
             if (getState() == CANCELLED)
                 throw new IgniteFutureCancelledCheckedException("Future was cancelled: "
+ this);
 
-            if (err != null)
-                throw U.cast(err);
+            if (resFlag != ERR)
+                throw U.cast((Throwable)res);
 
-            return res;
+            return (R)res;
         }
         catch (InterruptedException e) {
             Thread.currentThread().interrupt();
@@ -152,33 +152,35 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
         if (getState() == CANCELLED)
             throw new IgniteFutureCancelledCheckedException("Future was cancelled: " + this);
 
-        if (err != null)
-            throw U.cast(err);
+        if (resFlag != ERR)
+            throw U.cast((Throwable)res);
 
-        return res;
+        return (R)res;
     }
 
     /** {@inheritDoc} */
-    @Override public void listenAsync(@Nullable final IgniteInClosure<? super IgniteInternalFuture<R>>
lsnr) {
-        if (lsnr != null) {
+    @Override public void listenAsync(@Nullable final IgniteInClosure<? super IgniteInternalFuture<R>>
lsnr0) {
+        if (lsnr0 != null) {
             boolean done = isDone();
 
             if (!done) {
-                synchronized (mux) {
+                synchronized (this) {
                     done = isDone(); // Double check.
 
                     if (!done) {
-                        if (lsnrs == null)
-                            lsnrs = new ArrayList<>();
-
-                        lsnrs.add(lsnr);
+                        if (lsnr == null)
+                            lsnr = lsnr0;
+                        else if (lsnr instanceof ArrayListener)
+                            ((ArrayListener)lsnr).add(lsnr0);
+                        else {
+                            lsnr = (IgniteInClosure)new ArrayListener<IgniteInternalFuture>(lsnr,
lsnr0);
+                        }
                     }
                 }
             }
 
-            if (done) {
-                notifyListener(lsnr);
-            }
+            if (done)
+                notifyListener(lsnr0);
         }
     }
 
@@ -191,21 +193,20 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
      * Notifies all registered listeners.
      */
     private void notifyListeners() {
-        final Collection<IgniteInClosure<? super IgniteInternalFuture<R>>>
lsnrs0;
+        IgniteInClosure<? super IgniteInternalFuture<R>> lsnrs0;
 
-        synchronized (mux) {
-            lsnrs0 = lsnrs;
+        synchronized (this) {
+            lsnrs0 = lsnr;
 
-            if (lsnrs0 == null || lsnrs0.isEmpty())
+            if (lsnrs0 == null)
                 return;
 
-            lsnrs = null;
+            lsnr = null;
         }
 
-        assert !lsnrs0.isEmpty();
+        assert lsnr == null;
 
-        for (IgniteInClosure<? super IgniteInternalFuture<R>> lsnr : lsnrs0)
-            notifyListener(lsnr);
+        notifyListener(lsnr);
     }
 
     /**
@@ -252,7 +253,7 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
      */
     public boolean isFailed() {
         // Must read endTime first.
-        return endTime != 0 && err != null;
+        return endTime != 0 && resFlag == ERR;
     }
 
     /** {@inheritDoc} */
@@ -315,8 +316,14 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
 
         try {
             if (compareAndSetState(INIT, cancel ? CANCELLED : DONE)) {
-                this.res = res;
-                this.err = err;
+                if (err != null) {
+                    resFlag = ERR;
+                    this.res = err;
+                }
+                else {
+                    resFlag = RES;
+                    this.res = res;
+                }
 
                 notify = true;
 
@@ -372,6 +379,41 @@ public class GridFutureAdapter<R> extends AbstractQueuedSynchronizer
implements
     /**
      *
      */
+    private static class ArrayListener<R> implements IgniteInClosure<IgniteInternalFuture<R>>
{
+        /** */
+        private IgniteInClosure<? super IgniteInternalFuture<R>>[] arr;
+
+        /**
+         * @param lsnrs Listeners.
+         */
+        private ArrayListener(IgniteInClosure... lsnrs) {
+            this.arr = lsnrs;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void apply(IgniteInternalFuture<R> fut) {
+            for (int i = 0; i < arr.length; i++)
+                arr[i].apply(fut);
+        }
+
+        /**
+         * @param lsnr Listener.
+         */
+        void add(IgniteInClosure<? super IgniteInternalFuture<R>> lsnr) {
+            arr = Arrays.copyOf(arr, arr.length + 1);
+
+            arr[arr.length - 1] = lsnr;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(ArrayListener.class, this, "arr", Arrays.toString(arr));
+        }
+    }
+
+    /**
+     *
+     */
     private static class ChainFuture<R, T> extends GridFutureAdapter<T> {
         /** */
         private static final long serialVersionUID = 0L;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/bc18635a/modules/schedule/src/main/java/org/apache/ignite/internal/processors/schedule/ScheduleFutureImpl.java
----------------------------------------------------------------------
diff --git a/modules/schedule/src/main/java/org/apache/ignite/internal/processors/schedule/ScheduleFutureImpl.java
b/modules/schedule/src/main/java/org/apache/ignite/internal/processors/schedule/ScheduleFutureImpl.java
index 7f76873..b677770 100644
--- a/modules/schedule/src/main/java/org/apache/ignite/internal/processors/schedule/ScheduleFutureImpl.java
+++ b/modules/schedule/src/main/java/org/apache/ignite/internal/processors/schedule/ScheduleFutureImpl.java
@@ -41,10 +41,7 @@ import static java.util.concurrent.TimeUnit.*;
 /**
  * Implementation of {@link org.apache.ignite.scheduler.SchedulerFuture} interface.
  */
-class ScheduleFutureImpl<R> implements SchedulerFuture<R>, Externalizable {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+class ScheduleFutureImpl<R> implements SchedulerFuture<R> {
     /** Empty time array. */
     private static final long[] EMPTY_TIMES = new long[] {};
 
@@ -109,12 +106,6 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
     /** Listener call count. */
     private int lastLsnrExecCnt;
 
-    /** Synchronous notification flag. */
-    private volatile boolean syncNotify = IgniteSystemProperties.getBoolean(IGNITE_FUT_SYNC_NOTIFICATION,
true);
-
-    /** Concurrent notification flag. */
-    private volatile boolean concurNotify = IgniteSystemProperties.getBoolean(IGNITE_FUT_CONCURRENT_NOTIFICATION,
false);
-
     /** Mutex. */
     private final Object mux = new Object();
 
@@ -274,26 +265,6 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
         }
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean concurrentNotify() {
-        return concurNotify;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void concurrentNotify(boolean concurNotify) {
-        this.concurNotify = concurNotify;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean syncNotify() {
-        return syncNotify;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void syncNotify(boolean syncNotify) {
-        this.syncNotify = syncNotify;
-    }
-
     /**
      * Sets execution task.
      *
@@ -606,7 +577,7 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
             // Avoid race condition in case if listener was added after
             // first execution completed.
             if (notifyLsnr)
-                notifyListener(lsnr, res, err, syncNotify);
+                notifyListener(lsnr, res, err);
         }
     }
 
@@ -648,30 +619,13 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
      * @param lsnr Listener to notify.
      * @param res Last execution result.
      * @param err Last execution error.
-     * @param syncNotify Synchronous notification flag.
      */
-    private void notifyListener(final IgniteInClosure<? super IgniteFuture<R>>
lsnr, R res, Throwable err,
-        boolean syncNotify) {
+    private void notifyListener(final IgniteInClosure<? super IgniteFuture<R>>
lsnr, R res, Throwable err) {
         assert lsnr != null;
         assert !Thread.holdsLock(mux);
         assert ctx != null;
 
-        final SchedulerFuture<R> snapshot = snapshot(res, err);
-
-        if (syncNotify)
-            lsnr.apply(snapshot);
-        else {
-            try {
-                ctx.closure().runLocalSafe(new Runnable() {
-                    @Override public void run() {
-                        lsnr.apply(snapshot);
-                    }
-                }, true);
-            }
-            catch (Throwable e) {
-                U.error(log, "Failed to notify listener: " + this, e);
-            }
-        }
+        lsnr.apply(snapshot(res, err));
     }
 
     /**
@@ -687,22 +641,8 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
 
         final SchedulerFuture<R> snapshot = snapshot(res, err);
 
-        if (concurNotify) {
-            for (final IgniteInClosure<? super IgniteFuture<R>> lsnr : tmp)
-                ctx.closure().runLocalSafe(new GPR() {
-                    @Override public void run() {
-                        lsnr.apply(snapshot);
-                    }
-                }, true);
-        }
-        else {
-            ctx.closure().runLocalSafe(new GPR() {
-                @Override public void run() {
-                    for (IgniteInClosure<? super IgniteFuture<R>> lsnr : tmp)
-                        lsnr.apply(snapshot);
-                }
-            }, true);
-        }
+        for (IgniteInClosure<? super IgniteFuture<R>> lsnr : tmp)
+            lsnr.apply(snapshot);
     }
 
     /**
@@ -839,26 +779,6 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
         }
 
         /** {@inheritDoc} */
-        @Override public boolean concurrentNotify() {
-            return ref.concurrentNotify();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void concurrentNotify(boolean concurNotify) {
-            ref.concurrentNotify(concurNotify);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void syncNotify(boolean syncNotify) {
-            ref.syncNotify(syncNotify);
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean syncNotify() {
-            return ref.syncNotify();
-        }
-
-        /** {@inheritDoc} */
         @Override public String id() {
             return ref.id();
         }
@@ -965,54 +885,6 @@ class ScheduleFutureImpl<R> implements SchedulerFuture<R>,
Externalizable {
     }
 
     /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        boolean cancelled;
-        R lastRes;
-        Throwable lastErr;
-        GridScheduleStatistics stats;
-
-        synchronized (mux) {
-            cancelled = this.cancelled;
-            lastRes = this.lastRes;
-            lastErr = this.lastErr;
-            stats = this.stats;
-        }
-
-        out.writeBoolean(cancelled);
-        out.writeObject(lastRes);
-        out.writeObject(lastErr);
-        out.writeObject(stats);
-
-        out.writeBoolean(syncNotify);
-        out.writeBoolean(concurNotify);
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"unchecked"})
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-        CountDownLatch latch = new CountDownLatch(0);
-
-        boolean cancelled = in.readBoolean();
-        R lastRes = (R)in.readObject();
-        Throwable lastErr = (Throwable)in.readObject();
-        GridScheduleStatistics stats = (GridScheduleStatistics)in.readObject();
-
-        syncNotify = in.readBoolean();
-        concurNotify = in.readBoolean();
-
-        synchronized (mux) {
-            done = true;
-
-            resLatch = latch;
-
-            this.cancelled = cancelled;
-            this.lastRes = lastRes;
-            this.lastErr = lastErr;
-            this.stats = stats;
-        }
-    }
-
-    /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(ScheduleFutureImpl.class, this);
     }


Mime
View raw message