ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [4/4] incubator-ignite git commit: # IGNITE-625: WIP.
Date Wed, 08 Apr 2015 07:26:40 GMT
# IGNITE-625: WIP.


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

Branch: refs/heads/ignite-625-1
Commit: 542a92d122a998c8cb6aa0be982a99ec6b6c23be
Parents: 0defad0
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Wed Apr 8 10:19:21 2015 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Wed Apr 8 10:19:21 2015 +0300

----------------------------------------------------------------------
 .../processors/igfs/IgfsFileWorkerBatch.java        | 12 ++++++++++--
 .../ignite/internal/processors/igfs/IgfsImpl.java   | 16 +++-------------
 .../processors/igfs/IgfsAbstractSelfTest.java       | 14 ++++++--------
 3 files changed, 19 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/542a92d1/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerBatch.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerBatch.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerBatch.java
index 352bf06..4d91432 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerBatch.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerBatch.java
@@ -92,6 +92,13 @@ public abstract class IgfsFileWorkerBatch implements Runnable {
     }
 
     /**
+     * Cancel batch processing.
+     */
+    synchronized void cancel() {
+        queue.addFirst(CANCEL_MARKER);
+    }
+
+    /**
      * @return {@code True} if finish was called on this batch.
      */
     boolean finishing() {
@@ -109,12 +116,13 @@ public abstract class IgfsFileWorkerBatch implements Runnable {
 
                     if (data == null)
                         continue;
-
-                    if (data == STOP_MARKER) {
+                    else if (data == STOP_MARKER) {
                         assert queue.isEmpty();
 
                         fut.onDone();
                     }
+                    else if (data == CANCEL_MARKER)
+                        throw new IgniteCheckedException("Write to file was cancelled due
to node stop.");
 
                     try {
                         out.write(data);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/542a92d1/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
index 3dac09d..6eefde5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
@@ -243,17 +243,7 @@ public final class IgfsImpl implements IgfsEx {
         if (secondaryFs != null) {
             // Force all workers to finish their batches.
             for (IgfsFileWorkerBatch batch : workerMap.values())
-                batch.finish();
-
-            // Wait for all writers to finish their execution.
-            for (IgfsFileWorker w : workerMap.values()) {
-                try {
-                    w.join();
-                }
-                catch (InterruptedException e) {
-                    U.error(log, e.getMessage(), e);
-                }
-            }
+                batch.cancel();
 
             if (secondaryFs instanceof AutoCloseable)
                 U.closeQuiet((AutoCloseable)secondaryFs);
@@ -333,7 +323,7 @@ public final class IgfsImpl implements IgfsEx {
     void await(IgfsPath... paths) {
         assert paths != null;
 
-        for (Map.Entry<IgfsPath, IgfsFileWorker> workerEntry : workerMap.entrySet())
{
+        for (Map.Entry<IgfsPath, IgfsFileWorkerBatch> workerEntry : workerMap.entrySet())
{
             IgfsPath workerPath = workerEntry.getKey();
 
             boolean await = false;
@@ -347,7 +337,7 @@ public final class IgfsImpl implements IgfsEx {
             }
 
             if (await) {
-                IgfsFileWorkerBatch batch = workerEntry.getValue().currentBatch();
+                IgfsFileWorkerBatch batch = workerEntry.getValue();
 
                 if (batch != null) {
                     try {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/542a92d1/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
index 81fbce8..1ebbf9e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
@@ -2589,12 +2589,11 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest
{
         workerMapFld.setAccessible(true);
 
         // Wait for all workers to finish.
-        Map<IgfsPath, IgfsFileWorker> workerMap = (Map<IgfsPath, IgfsFileWorker>)workerMapFld.get(igfs);
+        Map<IgfsPath, IgfsFileWorkerBatch> workerMap = (Map<IgfsPath, IgfsFileWorkerBatch>)workerMapFld.get(igfs);
 
-        for (Map.Entry<IgfsPath, IgfsFileWorker> entry : workerMap.entrySet()) {
+        for (Map.Entry<IgfsPath, IgfsFileWorkerBatch> entry : workerMap.entrySet())
{
             entry.getValue().cancel();
-
-            U.join(entry.getValue());
+            entry.getValue().await();
         }
 
         // Clear igfs.
@@ -2617,12 +2616,11 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest
{
             workerMapFld.setAccessible(true);
 
             // Wait for all workers to finish.
-            Map<IgfsPath, IgfsFileWorker> workerMap = (Map<IgfsPath, IgfsFileWorker>)workerMapFld.get(igfsEx);
+            Map<IgfsPath, IgfsFileWorkerBatch> workerMap = (Map<IgfsPath, IgfsFileWorkerBatch>)workerMapFld.get(igfs);
 
-            for (Map.Entry<IgfsPath, IgfsFileWorker> entry : workerMap.entrySet())
{
+            for (Map.Entry<IgfsPath, IgfsFileWorkerBatch> entry : workerMap.entrySet())
{
                 entry.getValue().cancel();
-
-                U.join(entry.getValue());
+                entry.getValue().await();
             }
         }
 


Mime
View raw message