ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [1/4] incubator-ignite git commit: # IGNITE-625: WIP on file batches rework.
Date Wed, 08 Apr 2015 07:26:37 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-625-1 [created] 542a92d12


# IGNITE-625: WIP on file batches rework.


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

Branch: refs/heads/ignite-625-1
Commit: a5ea4b4263a6377ae6efc95057aa78ab34ba9103
Parents: bed567f
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Tue Apr 7 15:30:11 2015 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Tue Apr 7 15:30:11 2015 +0300

----------------------------------------------------------------------
 .../processors/igfs/IgfsFileWorkerBatch.java    | 196 ++++++-------------
 .../processors/igfs/IgfsFileWorkerTask.java     |  32 ---
 .../internal/processors/igfs/IgfsImpl.java      |   3 +-
 3 files changed, 61 insertions(+), 170 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a5ea4b42/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 27f9b7d..782e1d9 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
@@ -19,29 +19,24 @@ package org.apache.ignite.internal.processors.igfs;
 
 import org.apache.ignite.*;
 import org.apache.ignite.igfs.*;
-import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 
 import java.io.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-import java.util.concurrent.locks.*;
 
 /**
  * Work batch is an abstraction of the logically grouped tasks.
  */
 public class IgfsFileWorkerBatch {
-    /** Completion latch. */
-    private final CountDownLatch completeLatch = new CountDownLatch(1);
-
-    /** Finish guard. */
-    private final AtomicBoolean finishGuard = new AtomicBoolean();
-
-    /** Lock for finish operation. */
-    private final ReadWriteLock finishLock = new ReentrantReadWriteLock();
+    /** Stop marker. */
+    private static final byte[] STOP_MARKER = new byte[0];
 
     /** Tasks queue. */
-    private final BlockingDeque<IgfsFileWorkerTask> queue = new LinkedBlockingDeque<>();
+    private final BlockingDeque<byte[]> queue = new LinkedBlockingDeque<>();
+
+    /** Future which completes when batch is finished. */
+    private final GridFutureAdapter fut = new GridFutureAdapter();
 
     /** Path to the file in the primary file system. */
     private final IgfsPath path;
@@ -49,11 +44,8 @@ public class IgfsFileWorkerBatch {
     /** Output stream to the file. */
     private final OutputStream out;
 
-    /** Caught exception. */
-    private volatile IgniteCheckedException err;
-
-    /** Last task marker. */
-    private boolean lastTask;
+    /** Finishing flag. */
+    private volatile boolean finishing;
 
     /**
      * Constructor.
@@ -70,23 +62,37 @@ public class IgfsFileWorkerBatch {
     }
 
     /**
-     * Perform write.
+     * Perform write if batch is not finishing yet.
      *
      * @param data Data to be written.
-     * @return {@code True} in case operation was enqueued.
+     * @return {@code True} in case write was enqueued.
      */
-    boolean write(final byte[] data) {
-        return addTask(new IgfsFileWorkerTask() {
-            @Override public void execute() throws IgniteCheckedException {
-                try {
-                    out.write(data);
-                }
-                catch (IOException e) {
-                    throw new IgniteCheckedException("Failed to write data to the file due
to secondary file system " +
-                        "exception: " + path, e);
-                }
-            }
-        });
+    synchronized boolean write(final byte[] data) {
+        if (!finishing) {
+            queue.add(data);
+
+            return true;
+        }
+        else
+            return false;
+    }
+
+    /**
+     * Add the last task to that batch which will release all the resources.
+     */
+    synchronized void finish() {
+        if (!finishing) {
+            finishing = true;
+
+            queue.add(STOP_MARKER);
+        }
+    }
+
+    /**
+     * @return {@code True} if finish was called on this batch.
+     */
+    boolean finishing() {
+        return finishing;
     }
 
     /**
@@ -94,64 +100,41 @@ public class IgfsFileWorkerBatch {
      */
     void process() {
         try {
-            boolean cancelled = false;
-
-            while (!cancelled) {
+            while (!fut.isDone()) {
                 try {
-                    IgfsFileWorkerTask task = queue.poll(1000, TimeUnit.MILLISECONDS);
+                    byte[] data = queue.poll(1000, TimeUnit.MILLISECONDS);
 
-                    if (task == null)
+                    if (data == null)
                         continue;
 
-                    task.execute();
+                    if (data == STOP_MARKER) {
+                        assert queue.isEmpty();
 
-                    if (lastTask)
-                        cancelled = true;
-                }
-                catch (IgniteCheckedException e) {
-                    err = e;
+                        fut.onDone();
+                    }
 
-                    cancelled = true;
+                    try {
+                        out.write(data);
+                    }
+                    catch (IOException e) {
+                        throw new IgniteCheckedException("Failed to write data to the file
due to secondary " +
+                            "file system exception: " + path, e);
+                    }
                 }
-                catch (InterruptedException ignore) {
+                catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
 
-                    cancelled = true;
+                    fut.onDone(e);
+                }
+                catch (Exception e) {
+                    fut.onDone(e);
                 }
             }
         }
         finally {
-            try {
-                onComplete();
-            }
-            finally {
-                U.closeQuiet(out);
+            assert fut.isDone();
 
-                completeLatch.countDown();
-            }
-        }
-    }
-
-    /**
-     * Add the last task to that batch which will release all the resources.
-     */
-    @SuppressWarnings("LockAcquiredButNotSafelyReleased")
-    void finish() {
-        if (finishGuard.compareAndSet(false, true)) {
-            finishLock.writeLock().lock();
-
-            try {
-                queue.add(new IgfsFileWorkerTask() {
-                    @Override public void execute() {
-                        assert queue.isEmpty();
-
-                        lastTask = true;
-                    }
-                });
-            }
-            finally {
-                finishLock.writeLock().unlock();
-            }
+            U.closeQuiet(out);
         }
     }
 
@@ -161,29 +144,7 @@ public class IgfsFileWorkerBatch {
      * @throws IgniteCheckedException In case any exception has occurred during batch tasks
processing.
      */
     void await() throws IgniteCheckedException {
-        try {
-            completeLatch.await();
-        }
-        catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-
-            throw new IgniteInterruptedCheckedException(e);
-        }
-
-        IgniteCheckedException err0 = err;
-
-        if (err0 != null)
-            throw err0;
-    }
-
-    /**
-     * Await for that worker batch to complete in case it was marked as finished.
-     *
-     * @throws IgniteCheckedException In case any exception has occurred during batch tasks
processing.
-     */
-    void awaitIfFinished() throws IgniteCheckedException {
-        if (finishGuard.get())
-            await();
+        fut.get();
     }
 
     /**
@@ -194,43 +155,4 @@ public class IgfsFileWorkerBatch {
     IgfsPath path() {
         return path;
     }
-
-    /**
-     * Callback invoked when all the tasks within the batch are completed.
-     */
-    protected void onComplete() {
-        // No-op.
-    }
-
-    /**
-     * Add task to the queue.
-     *
-     * @param task Task to add.
-     * @return {@code True} in case the task was added to the queue.
-     */
-    private boolean addTask(IgfsFileWorkerTask task) {
-        finishLock.readLock().lock();
-
-        try {
-            if (!finishGuard.get()) {
-                try {
-                    queue.put(task);
-
-                    return true;
-                }
-                catch (InterruptedException ignore) {
-                    // Task was not enqueued due to interruption.
-                    Thread.currentThread().interrupt();
-
-                    return false;
-                }
-            }
-            else
-                return false;
-
-        }
-        finally {
-            finishLock.readLock().unlock();
-        }
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a5ea4b42/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerTask.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerTask.java
deleted file mode 100644
index ba788b4..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsFileWorkerTask.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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 org.apache.ignite.internal.processors.igfs;
-
-import org.apache.ignite.*;
-
-/**
- * Generic IGFS worker task which could potentially throw an exception.
- */
-public interface IgfsFileWorkerTask {
-    /**
-     * Execute task logic.
-     *
-     * @throws IgniteCheckedException If failed.
-     */
-    public void execute() throws IgniteCheckedException;
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a5ea4b42/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 824f178..4b4203e 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
@@ -351,7 +351,8 @@ public final class IgfsImpl implements IgfsEx {
 
                 if (batch != null) {
                     try {
-                        batch.awaitIfFinished();
+                        if (batch.finishing())
+                            batch.await();
                     }
                     catch (IgniteCheckedException ignore) {
                         // No-op.


Mime
View raw message