asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mhub...@apache.org
Subject [3/4] asterixdb git commit: Introduce MessagingNetworkManager for NC2NC AppMessaging
Date Fri, 26 Aug 2016 14:38:31 GMT
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/memory/ConcurrentFramePoolUnitTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/memory/ConcurrentFramePoolUnitTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/memory/ConcurrentFramePoolUnitTest.java
new file mode 100644
index 0000000..10c6b24
--- /dev/null
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/memory/ConcurrentFramePoolUnitTest.java
@@ -0,0 +1,610 @@
+/*
+ * 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.asterix.test.memory;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import org.apache.asterix.common.config.AsterixFeedProperties;
+import org.apache.asterix.common.memory.ConcurrentFramePool;
+import org.apache.asterix.common.memory.FrameAction;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.junit.Assert;
+import org.mockito.Mockito;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+public class ConcurrentFramePoolUnitTest extends TestCase {
+
+    private static final int DEFAULT_FRAME_SIZE = 32768;
+    private static final int NUM_FRAMES = 2048;
+    private static final long FEED_MEM_BUDGET = DEFAULT_FRAME_SIZE * NUM_FRAMES;
+    private static final int NUM_THREADS = 8;
+    private static final int MAX_SIZE = 52;
+    private static final double RELEASE_PROBABILITY = 0.20;
+    private volatile static HyracksDataException cause = null;
+
+    public ConcurrentFramePoolUnitTest(String testName) {
+        super(testName);
+    }
+
+    /**
+     * @return the suite of tests being tested
+     */
+    public static Test suite() {
+        return new TestSuite(ConcurrentFramePoolUnitTest.class);
+    }
+
+    @org.junit.Test
+    public void testMemoryManager() {
+        AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+        Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+        ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                DEFAULT_FRAME_SIZE);
+        int i = 0;
+        while (fmm.get() != null) {
+            i++;
+        }
+        Assert.assertEquals(i, NUM_FRAMES);
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testConcurrentMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            FixedSizeAllocator[] runners = new FixedSizeAllocator[NUM_THREADS];
+            Thread[] threads = new Thread[NUM_THREADS];
+            Arrays.parallelSetAll(runners, (int i) -> new FixedSizeAllocator(fmm));
+            for (int i = 0; i < threads.length; i++) {
+                threads[i] = new Thread(runners[i]);
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].start();
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].join();
+            }
+            int i = 0;
+            for (FixedSizeAllocator allocator : runners) {
+                i += allocator.getAllocated();
+            }
+            Assert.assertEquals(NUM_FRAMES, i);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testVarSizeMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            Random random = new Random();
+            int i = 0;
+            int req;
+            while (true) {
+                req = random.nextInt(MAX_SIZE) + 1;
+                if (req == 1) {
+                    if (fmm.get() != null) {
+                        i += 1;
+                    } else {
+                        break;
+                    }
+                } else if (fmm.get(req * DEFAULT_FRAME_SIZE) != null) {
+                    i += req;
+                } else {
+                    break;
+                }
+            }
+
+            Assert.assertEquals(i <= NUM_FRAMES, true);
+            Assert.assertEquals(i + req > NUM_FRAMES, true);
+            Assert.assertEquals(i + fmm.remaining(), NUM_FRAMES);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testConcurrentVarSizeMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+
+            VarSizeAllocator[] runners = new VarSizeAllocator[NUM_THREADS];
+            Thread[] threads = new Thread[NUM_THREADS];
+            Arrays.parallelSetAll(runners, (int i) -> new VarSizeAllocator(fmm));
+            for (int i = 0; i < threads.length; i++) {
+                threads[i] = new Thread(runners[i]);
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].start();
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].join();
+            }
+            int allocated = 0;
+            for (int i = 0; i < threads.length; i++) {
+                if (runners[i].cause() != null) {
+                    runners[i].cause().printStackTrace();
+                    Assert.fail(runners[i].cause().getMessage());
+                }
+                allocated += runners[i].getAllocated();
+            }
+            Assert.assertEquals(allocated <= NUM_FRAMES, true);
+            for (int i = 0; i < threads.length; i++) {
+                Assert.assertEquals(allocated + runners[i].getLastReq() > NUM_FRAMES, true);
+            }
+            Assert.assertEquals(allocated + fmm.remaining(), NUM_FRAMES);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testAcquireReleaseMemoryManager() throws HyracksDataException {
+        AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+        Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+        ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                DEFAULT_FRAME_SIZE);
+        Random random = new Random();
+        ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
+        while (true) {
+            if (random.nextDouble() < RELEASE_PROBABILITY) {
+                if (!stack.isEmpty()) {
+                    fmm.release(stack.pop());
+                }
+            } else {
+                ByteBuffer buffer = fmm.get();
+                if (buffer == null) {
+                    break;
+                } else {
+                    stack.push(buffer);
+                }
+            }
+        }
+        Assert.assertEquals(stack.size(), NUM_FRAMES);
+        Assert.assertEquals(fmm.remaining(), 0);
+        for (ByteBuffer buffer : stack) {
+            fmm.release(buffer);
+        }
+        stack.clear();
+        Assert.assertEquals(fmm.remaining(), NUM_FRAMES);
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testConcurrentAcquireReleaseMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            FixedSizeGoodAllocator[] runners = new FixedSizeGoodAllocator[NUM_THREADS];
+            Thread[] threads = new Thread[NUM_THREADS];
+            Arrays.parallelSetAll(runners, (int i) -> new FixedSizeGoodAllocator(fmm));
+            for (int i = 0; i < threads.length; i++) {
+                threads[i] = new Thread(runners[i]);
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].start();
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].join();
+            }
+            int i = 0;
+            for (FixedSizeGoodAllocator allocator : runners) {
+                i += allocator.getAllocated();
+            }
+            Assert.assertEquals(NUM_FRAMES, i);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testAcquireReleaseVarSizeMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            Random random = new Random();
+            ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
+            int i = 0;
+            int req;
+            while (true) {
+                // release
+                if (random.nextDouble() < RELEASE_PROBABILITY) {
+                    if (!stack.isEmpty()) {
+                        ByteBuffer buffer = stack.pop();
+                        i -= (buffer.capacity() / DEFAULT_FRAME_SIZE);
+                        fmm.release(buffer);
+                    }
+                } else {
+                    // acquire
+                    req = random.nextInt(MAX_SIZE) + 1;
+                    if (req == 1) {
+                        ByteBuffer buffer = fmm.get();
+                        if (buffer != null) {
+                            stack.push(buffer);
+                            i += 1;
+                        } else {
+                            break;
+                        }
+                    } else {
+                        ByteBuffer buffer = fmm.get(req * DEFAULT_FRAME_SIZE);
+                        if (buffer != null) {
+                            stack.push(buffer);
+                            i += req;
+                        } else {
+                            break;
+                        }
+                    }
+                }
+            }
+
+            Assert.assertEquals(i <= NUM_FRAMES, true);
+            Assert.assertEquals(i + req > NUM_FRAMES, true);
+            Assert.assertEquals(i + fmm.remaining(), NUM_FRAMES);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        } finally {
+            Assert.assertNull(cause);
+        }
+    }
+
+    @org.junit.Test
+    public void testConcurrentAcquireReleaseVarSizeMemoryManager() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            VarSizeGoodAllocator[] runners = new VarSizeGoodAllocator[NUM_THREADS];
+            Thread[] threads = new Thread[NUM_THREADS];
+            Arrays.parallelSetAll(runners, (int i) -> new VarSizeGoodAllocator(fmm));
+            for (int i = 0; i < threads.length; i++) {
+                threads[i] = new Thread(runners[i]);
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].start();
+            }
+            for (int i = 0; i < threads.length; i++) {
+                threads[i].join();
+            }
+            int i = 0;
+            for (VarSizeGoodAllocator allocator : runners) {
+                if (allocator.cause() != null) {
+                    allocator.cause().printStackTrace();
+                    Assert.fail(allocator.cause().getMessage());
+                }
+                i += allocator.getAllocated();
+            }
+            Assert.assertEquals(NUM_FRAMES, i + fmm.remaining());
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        } finally {
+            Assert.assertNull(cause);
+        }
+    }
+
+    @org.junit.Test
+    public void testFixedSizeSubscribtion() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            int i = 0;
+            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE);
+            LinkedBlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
+            FrameAction frameAction = new FrameAction();
+            frameAction.setFrame(buffer);
+            while (!fmm.subscribe(frameAction)) {
+                buffers.put(frameAction.retrieve());
+                i++;
+            }
+            // One subscriber.
+            // Check that all frames have been consumed
+            Assert.assertEquals(i, NUM_FRAMES);
+            // Release a frame (That will be handed out to the subscriber)
+            fmm.release(buffers.take());
+            // Check that all frames have been consumed (since the released frame have been handed to the consumer)
+            Assert.assertEquals(0, fmm.remaining());
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        } finally {
+            Assert.assertNull(cause);
+        }
+    }
+
+    @org.junit.Test
+    public void testLargerThanBudgetRequests() {
+        HyracksDataException hde = null;
+        try {
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", DEFAULT_FRAME_SIZE * 16, DEFAULT_FRAME_SIZE);
+            fmm.get(32 * DEFAULT_FRAME_SIZE);
+        } catch (HyracksDataException e) {
+            hde = e;
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNotNull(hde);
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testLargerThanBudgetSubscribe() {
+        HyracksDataException hde = null;
+        try {
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", DEFAULT_FRAME_SIZE * 16, DEFAULT_FRAME_SIZE);
+            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE * 32);
+            FrameAction frameAction = new FrameAction();
+            frameAction.setFrame(buffer);
+            fmm.subscribe(frameAction);
+        } catch (HyracksDataException e) {
+            hde = e;
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        }
+        Assert.assertNotNull(hde);
+        Assert.assertNull(cause);
+    }
+
+    @org.junit.Test
+    public void testgetWhileSubscribersExist() {
+        try {
+            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
+            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
+            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(),
+                    DEFAULT_FRAME_SIZE);
+            int i = 0;
+            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE);
+            LinkedBlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
+            FrameAction frameAction = new FrameAction();
+            frameAction.setFrame(buffer);
+            while (!fmm.subscribe(frameAction)) {
+                buffers.put(frameAction.retrieve());
+                i++;
+            }
+            // One subscriber.
+            // Check that all frames have been consumed
+            Assert.assertEquals(i, NUM_FRAMES);
+            // Release a frame (That will be handed out to the subscriber)
+            fmm.release(buffers.take());
+            // Check that all frames have been consumed (since the released frame have been handed to the consumer)
+            Assert.assertEquals(fmm.remaining(), 0);
+            buffers.put(frameAction.retrieve());
+            // Create another subscriber that takes frames of double the size
+            ByteBuffer bufferTimes2 = ByteBuffer.allocate(DEFAULT_FRAME_SIZE * 2);
+            LinkedBlockingDeque<ByteBuffer> buffersTimes2 = new LinkedBlockingDeque<>();
+            FrameAction frameActionTimes2 = new FrameAction();
+            frameActionTimes2.setFrame(bufferTimes2);
+            Assert.assertEquals(true, fmm.subscribe(frameActionTimes2));
+            // release a small one
+            fmm.release(buffers.take());
+            Assert.assertEquals(fmm.remaining(), 1);
+            // Check that a small get fails
+            Assert.assertEquals(null, fmm.get());
+            // release another small one
+            fmm.release(buffers.take());
+            // Check that no small frames exists in the pool since subscriber request was satisfied
+            Assert.assertEquals(fmm.remaining(), 0);
+            buffersTimes2.add(frameActionTimes2.retrieve());
+            fmm.release(buffers);
+            fmm.release(bufferTimes2);
+            Assert.assertEquals(fmm.remaining(), NUM_FRAMES);
+        } catch (Throwable th) {
+            th.printStackTrace();
+            Assert.fail(th.getMessage());
+        } finally {
+            Assert.assertNull(cause);
+        }
+    }
+
+    /*
+     * Runnables used for unit tests
+     */
+    private class FixedSizeAllocator implements Runnable {
+        private final ConcurrentFramePool fmm;
+        private int allocated = 0;
+
+        public FixedSizeAllocator(ConcurrentFramePool fmm) {
+            this.fmm = fmm;
+        }
+
+        public int getAllocated() {
+            return allocated;
+        }
+
+        @Override
+        public void run() {
+            while (fmm.get() != null) {
+                allocated++;
+            }
+        }
+    }
+
+    private class FixedSizeGoodAllocator implements Runnable {
+        private final ConcurrentFramePool fmm;
+        private final ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
+        private final Random random = new Random();
+
+        public FixedSizeGoodAllocator(ConcurrentFramePool fmm) {
+            this.fmm = fmm;
+        }
+
+        public int getAllocated() {
+            return stack.size();
+        }
+
+        @Override
+        public void run() {
+            while (true) {
+                if (random.nextDouble() < RELEASE_PROBABILITY) {
+                    if (!stack.isEmpty()) {
+                        try {
+                            fmm.release(stack.pop());
+                        } catch (HyracksDataException e) {
+                            e.printStackTrace();
+                            cause = e;
+                        }
+                    }
+                } else {
+                    ByteBuffer buffer = fmm.get();
+                    if (buffer == null) {
+                        break;
+                    } else {
+                        stack.push(buffer);
+                    }
+                }
+            }
+        }
+    }
+
+    private class VarSizeGoodAllocator implements Runnable {
+        private final ConcurrentFramePool fmm;
+        private int allocated = 0;
+        private int req = 0;
+        private final Random random = new Random();
+        private Throwable cause;
+        private final ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
+
+        public VarSizeGoodAllocator(ConcurrentFramePool fmm) {
+            this.fmm = fmm;
+        }
+
+        public int getAllocated() {
+            return allocated;
+        }
+
+        public Throwable cause() {
+            return cause;
+        }
+
+        @Override
+        public void run() {
+            try {
+                while (true) {
+                    if (random.nextDouble() < RELEASE_PROBABILITY) {
+                        if (!stack.isEmpty()) {
+                            ByteBuffer buffer = stack.pop();
+                            allocated -= (buffer.capacity() / DEFAULT_FRAME_SIZE);
+                            fmm.release(buffer);
+                        }
+                    } else {
+                        req = random.nextInt(MAX_SIZE) + 1;
+                        if (req == 1) {
+                            ByteBuffer buffer = fmm.get();
+                            if (buffer != null) {
+                                stack.push(buffer);
+                                allocated += 1;
+                            } else {
+                                break;
+                            }
+                        } else {
+                            ByteBuffer buffer = fmm.get(req * DEFAULT_FRAME_SIZE);
+                            if (buffer != null) {
+                                stack.push(buffer);
+                                allocated += req;
+                            } else {
+                                break;
+                            }
+                        }
+                    }
+                }
+            } catch (Throwable th) {
+                this.cause = th;
+            }
+        }
+    }
+
+    private class VarSizeAllocator implements Runnable {
+        private final ConcurrentFramePool fmm;
+        private int allocated = 0;
+        private int req = 0;
+        private final Random random = new Random();
+        private Throwable cause;
+
+        public VarSizeAllocator(ConcurrentFramePool fmm) {
+            this.fmm = fmm;
+        }
+
+        public int getAllocated() {
+            return allocated;
+        }
+
+        public int getLastReq() {
+            return req;
+        }
+
+        public Throwable cause() {
+            return cause;
+        }
+
+        @Override
+        public void run() {
+            try {
+                while (true) {
+                    req = random.nextInt(MAX_SIZE) + 1;
+                    if (req == 1) {
+                        if (fmm.get() != null) {
+                            allocated += 1;
+                        } else {
+                            break;
+                        }
+                    } else if (fmm.get(req * DEFAULT_FRAME_SIZE) != null) {
+                        allocated += req;
+                    } else {
+                        break;
+                    }
+                }
+            } catch (Throwable th) {
+                this.cause = th;
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
index d4e3641..cd04515 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
@@ -24,8 +24,8 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.apache.asterix.active.ActiveRuntimeId;
-import org.apache.asterix.active.ConcurrentFramePool;
-import org.apache.asterix.active.FrameAction;
+import org.apache.asterix.common.memory.ConcurrentFramePool;
+import org.apache.asterix.common.memory.FrameAction;
 import org.apache.asterix.external.feed.management.FeedConnectionId;
 import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
 import org.apache.asterix.external.util.FeedUtils.Mode;
@@ -73,11 +73,12 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
             throws HyracksDataException {
         this.writer = writer;
 
-        this.spiller =
-                fpa.spillToDiskOnCongestion() ? new FrameSpiller(ctx,
+        this.spiller = fpa.spillToDiskOnCongestion()
+                ? new FrameSpiller(ctx,
                         connectionId.getFeedId() + "_" + connectionId.getDatasetName() + "_"
                                 + runtimeId.getRuntimeName() + "_" + runtimeId.getPartition(),
-                        fpa.getMaxSpillOnDisk()) : null;
+                        fpa.getMaxSpillOnDisk())
+                : null;
         this.exceptionHandler = new FeedExceptionHandler(ctx, fta);
         this.fpa = fpa;
         this.framePool = framePool;

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/ConcurrentFramePoolUnitTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/ConcurrentFramePoolUnitTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/ConcurrentFramePoolUnitTest.java
deleted file mode 100644
index 0f6a2ea..0000000
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/ConcurrentFramePoolUnitTest.java
+++ /dev/null
@@ -1,610 +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.asterix.external.feed.test;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayDeque;
-import java.util.Arrays;
-import java.util.Random;
-import java.util.concurrent.LinkedBlockingDeque;
-
-import org.apache.asterix.active.ConcurrentFramePool;
-import org.apache.asterix.active.FrameAction;
-import org.apache.asterix.common.config.AsterixFeedProperties;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.junit.Assert;
-import org.mockito.Mockito;
-
-import junit.framework.Test;
-import junit.framework.TestCase;
-import junit.framework.TestSuite;
-
-public class ConcurrentFramePoolUnitTest extends TestCase {
-
-    private static final int DEFAULT_FRAME_SIZE = 32768;
-    private static final int NUM_FRAMES = 2048;
-    private static final long FEED_MEM_BUDGET = DEFAULT_FRAME_SIZE * NUM_FRAMES;
-    private static final int NUM_THREADS = 8;
-    private static final int MAX_SIZE = 52;
-    private static final double RELEASE_PROBABILITY = 0.20;
-    private volatile static HyracksDataException cause = null;
-
-    public ConcurrentFramePoolUnitTest(String testName) {
-        super(testName);
-    }
-
-    /**
-     * @return the suite of tests being tested
-     */
-    public static Test suite() {
-        return new TestSuite(ConcurrentFramePoolUnitTest.class);
-    }
-
-    @org.junit.Test
-    public void testMemoryManager() {
-        AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-        Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-        ConcurrentFramePool fmm =
-                new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-        int i = 0;
-        while (fmm.get() != null) {
-            i++;
-        }
-        Assert.assertEquals(i, NUM_FRAMES);
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testConcurrentMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            FixedSizeAllocator[] runners = new FixedSizeAllocator[NUM_THREADS];
-            Thread[] threads = new Thread[NUM_THREADS];
-            Arrays.parallelSetAll(runners, (int i) -> new FixedSizeAllocator(fmm));
-            for (int i = 0; i < threads.length; i++) {
-                threads[i] = new Thread(runners[i]);
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].start();
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].join();
-            }
-            int i = 0;
-            for (FixedSizeAllocator allocator : runners) {
-                i += allocator.getAllocated();
-            }
-            Assert.assertEquals(NUM_FRAMES, i);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testVarSizeMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            Random random = new Random();
-            int i = 0;
-            int req;
-            while (true) {
-                req = random.nextInt(MAX_SIZE) + 1;
-                if (req == 1) {
-                    if (fmm.get() != null) {
-                        i += 1;
-                    } else {
-                        break;
-                    }
-                } else if (fmm.get(req * DEFAULT_FRAME_SIZE) != null) {
-                    i += req;
-                } else {
-                    break;
-                }
-            }
-
-            Assert.assertEquals(i <= NUM_FRAMES, true);
-            Assert.assertEquals(i + req > NUM_FRAMES, true);
-            Assert.assertEquals(i + fmm.remaining(), NUM_FRAMES);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testConcurrentVarSizeMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-
-            VarSizeAllocator[] runners = new VarSizeAllocator[NUM_THREADS];
-            Thread[] threads = new Thread[NUM_THREADS];
-            Arrays.parallelSetAll(runners, (int i) -> new VarSizeAllocator(fmm));
-            for (int i = 0; i < threads.length; i++) {
-                threads[i] = new Thread(runners[i]);
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].start();
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].join();
-            }
-            int allocated = 0;
-            for (int i = 0; i < threads.length; i++) {
-                if (runners[i].cause() != null) {
-                    runners[i].cause().printStackTrace();
-                    Assert.fail(runners[i].cause().getMessage());
-                }
-                allocated += runners[i].getAllocated();
-            }
-            Assert.assertEquals(allocated <= NUM_FRAMES, true);
-            for (int i = 0; i < threads.length; i++) {
-                Assert.assertEquals(allocated + runners[i].getLastReq() > NUM_FRAMES, true);
-            }
-            Assert.assertEquals(allocated + fmm.remaining(), NUM_FRAMES);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testAcquireReleaseMemoryManager() throws HyracksDataException {
-        AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-        Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-        ConcurrentFramePool fmm =
-                new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-        Random random = new Random();
-        ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
-        while (true) {
-            if (random.nextDouble() < RELEASE_PROBABILITY) {
-                if (!stack.isEmpty()) {
-                    fmm.release(stack.pop());
-                }
-            } else {
-                ByteBuffer buffer = fmm.get();
-                if (buffer == null) {
-                    break;
-                } else {
-                    stack.push(buffer);
-                }
-            }
-        }
-        Assert.assertEquals(stack.size(), NUM_FRAMES);
-        Assert.assertEquals(fmm.remaining(), 0);
-        for (ByteBuffer buffer : stack) {
-            fmm.release(buffer);
-        }
-        stack.clear();
-        Assert.assertEquals(fmm.remaining(), NUM_FRAMES);
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testConcurrentAcquireReleaseMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            FixedSizeGoodAllocator[] runners = new FixedSizeGoodAllocator[NUM_THREADS];
-            Thread[] threads = new Thread[NUM_THREADS];
-            Arrays.parallelSetAll(runners, (int i) -> new FixedSizeGoodAllocator(fmm));
-            for (int i = 0; i < threads.length; i++) {
-                threads[i] = new Thread(runners[i]);
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].start();
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].join();
-            }
-            int i = 0;
-            for (FixedSizeGoodAllocator allocator : runners) {
-                i += allocator.getAllocated();
-            }
-            Assert.assertEquals(NUM_FRAMES, i);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testAcquireReleaseVarSizeMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            Random random = new Random();
-            ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
-            int i = 0;
-            int req;
-            while (true) {
-                // release
-                if (random.nextDouble() < RELEASE_PROBABILITY) {
-                    if (!stack.isEmpty()) {
-                        ByteBuffer buffer = stack.pop();
-                        i -= (buffer.capacity() / DEFAULT_FRAME_SIZE);
-                        fmm.release(buffer);
-                    }
-                } else {
-                    // acquire
-                    req = random.nextInt(MAX_SIZE) + 1;
-                    if (req == 1) {
-                        ByteBuffer buffer = fmm.get();
-                        if (buffer != null) {
-                            stack.push(buffer);
-                            i += 1;
-                        } else {
-                            break;
-                        }
-                    } else {
-                        ByteBuffer buffer = fmm.get(req * DEFAULT_FRAME_SIZE);
-                        if (buffer != null) {
-                            stack.push(buffer);
-                            i += req;
-                        } else {
-                            break;
-                        }
-                    }
-                }
-            }
-
-            Assert.assertEquals(i <= NUM_FRAMES, true);
-            Assert.assertEquals(i + req > NUM_FRAMES, true);
-            Assert.assertEquals(i + fmm.remaining(), NUM_FRAMES);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        } finally {
-            Assert.assertNull(cause);
-        }
-    }
-
-    @org.junit.Test
-    public void testConcurrentAcquireReleaseVarSizeMemoryManager() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            VarSizeGoodAllocator[] runners = new VarSizeGoodAllocator[NUM_THREADS];
-            Thread[] threads = new Thread[NUM_THREADS];
-            Arrays.parallelSetAll(runners, (int i) -> new VarSizeGoodAllocator(fmm));
-            for (int i = 0; i < threads.length; i++) {
-                threads[i] = new Thread(runners[i]);
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].start();
-            }
-            for (int i = 0; i < threads.length; i++) {
-                threads[i].join();
-            }
-            int i = 0;
-            for (VarSizeGoodAllocator allocator : runners) {
-                if (allocator.cause() != null) {
-                    allocator.cause().printStackTrace();
-                    Assert.fail(allocator.cause().getMessage());
-                }
-                i += allocator.getAllocated();
-            }
-            Assert.assertEquals(NUM_FRAMES, i + fmm.remaining());
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        } finally {
-            Assert.assertNull(cause);
-        }
-    }
-
-    @org.junit.Test
-    public void testFixedSizeSubscribtion() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            int i = 0;
-            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE);
-            LinkedBlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
-            FrameAction frameAction = new FrameAction();
-            frameAction.setFrame(buffer);
-            while (!fmm.subscribe(frameAction)) {
-                buffers.put(frameAction.retrieve());
-                i++;
-            }
-            // One subscriber.
-            // Check that all frames have been consumed
-            Assert.assertEquals(i, NUM_FRAMES);
-            // Release a frame (That will be handed out to the subscriber)
-            fmm.release(buffers.take());
-            // Check that all frames have been consumed (since the released frame have been handed to the consumer)
-            Assert.assertEquals(0, fmm.remaining());
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        } finally {
-            Assert.assertNull(cause);
-        }
-    }
-
-    @org.junit.Test
-    public void testLargerThanBudgetRequests() {
-        HyracksDataException hde = null;
-        try {
-            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", DEFAULT_FRAME_SIZE * 16, DEFAULT_FRAME_SIZE);
-            fmm.get(32 * DEFAULT_FRAME_SIZE);
-        } catch (HyracksDataException e) {
-            hde = e;
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNotNull(hde);
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testLargerThanBudgetSubscribe() {
-        HyracksDataException hde = null;
-        try {
-            ConcurrentFramePool fmm = new ConcurrentFramePool("TestNode", DEFAULT_FRAME_SIZE * 16, DEFAULT_FRAME_SIZE);
-            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE * 32);
-            FrameAction frameAction = new FrameAction();
-            frameAction.setFrame(buffer);
-            fmm.subscribe(frameAction);
-        } catch (HyracksDataException e) {
-            hde = e;
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        }
-        Assert.assertNotNull(hde);
-        Assert.assertNull(cause);
-    }
-
-    @org.junit.Test
-    public void testgetWhileSubscribersExist() {
-        try {
-            AsterixFeedProperties afp = Mockito.mock(AsterixFeedProperties.class);
-            Mockito.when(afp.getMemoryComponentGlobalBudget()).thenReturn(FEED_MEM_BUDGET);
-            ConcurrentFramePool fmm =
-                    new ConcurrentFramePool("TestNode", afp.getMemoryComponentGlobalBudget(), DEFAULT_FRAME_SIZE);
-            int i = 0;
-            ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_FRAME_SIZE);
-            LinkedBlockingDeque<ByteBuffer> buffers = new LinkedBlockingDeque<>();
-            FrameAction frameAction = new FrameAction();
-            frameAction.setFrame(buffer);
-            while (!fmm.subscribe(frameAction)) {
-                buffers.put(frameAction.retrieve());
-                i++;
-            }
-            // One subscriber.
-            // Check that all frames have been consumed
-            Assert.assertEquals(i, NUM_FRAMES);
-            // Release a frame (That will be handed out to the subscriber)
-            fmm.release(buffers.take());
-            // Check that all frames have been consumed (since the released frame have been handed to the consumer)
-            Assert.assertEquals(fmm.remaining(), 0);
-            buffers.put(frameAction.retrieve());
-            // Create another subscriber that takes frames of double the size
-            ByteBuffer bufferTimes2 = ByteBuffer.allocate(DEFAULT_FRAME_SIZE * 2);
-            LinkedBlockingDeque<ByteBuffer> buffersTimes2 = new LinkedBlockingDeque<>();
-            FrameAction frameActionTimes2 = new FrameAction();
-            frameActionTimes2.setFrame(bufferTimes2);
-            Assert.assertEquals(true, fmm.subscribe(frameActionTimes2));
-            // release a small one
-            fmm.release(buffers.take());
-            Assert.assertEquals(fmm.remaining(), 1);
-            // Check that a small get fails
-            Assert.assertEquals(null, fmm.get());
-            // release another small one
-            fmm.release(buffers.take());
-            // Check that no small frames exists in the pool since subscriber request was satisfied
-            Assert.assertEquals(fmm.remaining(), 0);
-            buffersTimes2.add(frameActionTimes2.retrieve());
-            fmm.release(buffers);
-            fmm.release(bufferTimes2);
-            Assert.assertEquals(fmm.remaining(), NUM_FRAMES);
-        } catch (Throwable th) {
-            th.printStackTrace();
-            Assert.fail(th.getMessage());
-        } finally {
-            Assert.assertNull(cause);
-        }
-    }
-
-    /*
-     * Runnables used for unit tests
-     */
-    private class FixedSizeAllocator implements Runnable {
-        private final ConcurrentFramePool fmm;
-        private int allocated = 0;
-
-        public FixedSizeAllocator(ConcurrentFramePool fmm) {
-            this.fmm = fmm;
-        }
-
-        public int getAllocated() {
-            return allocated;
-        }
-
-        @Override
-        public void run() {
-            while (fmm.get() != null) {
-                allocated++;
-            }
-        }
-    }
-
-    private class FixedSizeGoodAllocator implements Runnable {
-        private final ConcurrentFramePool fmm;
-        private final ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
-        private final Random random = new Random();
-
-        public FixedSizeGoodAllocator(ConcurrentFramePool fmm) {
-            this.fmm = fmm;
-        }
-
-        public int getAllocated() {
-            return stack.size();
-        }
-
-        @Override
-        public void run() {
-            while (true) {
-                if (random.nextDouble() < RELEASE_PROBABILITY) {
-                    if (!stack.isEmpty()) {
-                        try {
-                            fmm.release(stack.pop());
-                        } catch (HyracksDataException e) {
-                            e.printStackTrace();
-                            cause = e;
-                        }
-                    }
-                } else {
-                    ByteBuffer buffer = fmm.get();
-                    if (buffer == null) {
-                        break;
-                    } else {
-                        stack.push(buffer);
-                    }
-                }
-            }
-        }
-    }
-
-    private class VarSizeGoodAllocator implements Runnable {
-        private final ConcurrentFramePool fmm;
-        private int allocated = 0;
-        private int req = 0;
-        private final Random random = new Random();
-        private Throwable cause;
-        private final ArrayDeque<ByteBuffer> stack = new ArrayDeque<>();
-
-        public VarSizeGoodAllocator(ConcurrentFramePool fmm) {
-            this.fmm = fmm;
-        }
-
-        public int getAllocated() {
-            return allocated;
-        }
-
-        public Throwable cause() {
-            return cause;
-        }
-
-        @Override
-        public void run() {
-            try {
-                while (true) {
-                    if (random.nextDouble() < RELEASE_PROBABILITY) {
-                        if (!stack.isEmpty()) {
-                            ByteBuffer buffer = stack.pop();
-                            allocated -= (buffer.capacity() / DEFAULT_FRAME_SIZE);
-                            fmm.release(buffer);
-                        }
-                    } else {
-                        req = random.nextInt(MAX_SIZE) + 1;
-                        if (req == 1) {
-                            ByteBuffer buffer = fmm.get();
-                            if (buffer != null) {
-                                stack.push(buffer);
-                                allocated += 1;
-                            } else {
-                                break;
-                            }
-                        } else {
-                            ByteBuffer buffer = fmm.get(req * DEFAULT_FRAME_SIZE);
-                            if (buffer != null) {
-                                stack.push(buffer);
-                                allocated += req;
-                            } else {
-                                break;
-                            }
-                        }
-                    }
-                }
-            } catch (Throwable th) {
-                this.cause = th;
-            }
-        }
-    }
-
-    private class VarSizeAllocator implements Runnable {
-        private final ConcurrentFramePool fmm;
-        private int allocated = 0;
-        private int req = 0;
-        private final Random random = new Random();
-        private Throwable cause;
-
-        public VarSizeAllocator(ConcurrentFramePool fmm) {
-            this.fmm = fmm;
-        }
-
-        public int getAllocated() {
-            return allocated;
-        }
-
-        public int getLastReq() {
-            return req;
-        }
-
-        public Throwable cause() {
-            return cause;
-        }
-
-        @Override
-        public void run() {
-            try {
-                while (true) {
-                    req = random.nextInt(MAX_SIZE) + 1;
-                    if (req == 1) {
-                        if (fmm.get() != null) {
-                            allocated += 1;
-                        } else {
-                            break;
-                        }
-                    } else if (fmm.get(req * DEFAULT_FRAME_SIZE) != null) {
-                        allocated += req;
-                    } else {
-                        break;
-                    }
-                }
-            } catch (Throwable th) {
-                this.cause = th;
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
index e643206..171d271 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
@@ -26,8 +26,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import org.apache.asterix.active.ActiveRuntimeId;
-import org.apache.asterix.active.ConcurrentFramePool;
 import org.apache.asterix.active.EntityId;
+import org.apache.asterix.common.memory.ConcurrentFramePool;
 import org.apache.asterix.external.feed.dataflow.FeedRuntimeInputHandler;
 import org.apache.asterix.external.feed.management.FeedConnectionId;
 import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
@@ -114,11 +114,11 @@ public class InputHandlerTest extends TestCase {
             Random random = new Random();
             IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
             // No spill, No discard
-            FeedPolicyAccessor fpa =
-                    createFeedPolicyAccessor(true, false, NUM_FRAMES * DEFAULT_FRAME_SIZE, DISCARD_ALLOWANCE);
+            FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, false, NUM_FRAMES * DEFAULT_FRAME_SIZE,
+                    DISCARD_ALLOWANCE);
             // Non-Active Writer
-            TestFrameWriter writer =
-                    FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(), false);
+            TestFrameWriter writer = FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(),
+                    false);
             // FramePool
             ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, 0, DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
@@ -156,11 +156,11 @@ public class InputHandlerTest extends TestCase {
             int numRounds = 10;
             IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
             // No spill, No discard
-            FeedPolicyAccessor fpa =
-                    createFeedPolicyAccessor(true, false, NUM_FRAMES * DEFAULT_FRAME_SIZE, DISCARD_ALLOWANCE);
+            FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, false, NUM_FRAMES * DEFAULT_FRAME_SIZE,
+                    DISCARD_ALLOWANCE);
             // Non-Active Writer
-            TestFrameWriter writer =
-                    FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(), false);
+            TestFrameWriter writer = FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(),
+                    false);
             // FramePool
             ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, 0, DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
@@ -205,14 +205,14 @@ public class InputHandlerTest extends TestCase {
             int totalMinFrames = 0;
             IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
             // Spill budget = Memory budget, No discard
-            FeedPolicyAccessor fpa =
-                    createFeedPolicyAccessor(true, true, DEFAULT_FRAME_SIZE * numberOfSpillFrames, DISCARD_ALLOWANCE);
+            FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, true, DEFAULT_FRAME_SIZE * numberOfSpillFrames,
+                    DISCARD_ALLOWANCE);
             // Non-Active Writer
             TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
             writer.freeze();
             // FramePool
-            ConcurrentFramePool framePool =
-                    new ConcurrentFramePool(NODE_ID, numberOfMemoryFrames * DEFAULT_FRAME_SIZE, DEFAULT_FRAME_SIZE);
+            ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, numberOfMemoryFrames * DEFAULT_FRAME_SIZE,
+                    DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
             handler.open();
             ByteBuffer buffer1 = ByteBuffer.allocate(DEFAULT_FRAME_SIZE);
@@ -275,8 +275,8 @@ public class InputHandlerTest extends TestCase {
             Assert.assertEquals(0, handler.getNumDiscarded());
             // We can only discard one frame
             double numDiscarded = 0;
-            boolean nextShouldDiscard =
-                    ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa.getMaxFractionDiscard();
+            boolean nextShouldDiscard = ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa
+                    .getMaxFractionDiscard();
             while (nextShouldDiscard) {
                 handler.nextFrame(buffer5);
                 numDiscarded++;
@@ -315,14 +315,14 @@ public class InputHandlerTest extends TestCase {
             int numberOfSpillFrames = 50;
             IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
             // Spill budget = Memory budget, No discard
-            FeedPolicyAccessor fpa =
-                    createFeedPolicyAccessor(true, true, DEFAULT_FRAME_SIZE * numberOfSpillFrames, DISCARD_ALLOWANCE);
+            FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, true, DEFAULT_FRAME_SIZE * numberOfSpillFrames,
+                    DISCARD_ALLOWANCE);
             // Non-Active Writer
             TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
             writer.freeze();
             // FramePool
-            ConcurrentFramePool framePool =
-                    new ConcurrentFramePool(NODE_ID, numberOfMemoryFrames * DEFAULT_FRAME_SIZE, DEFAULT_FRAME_SIZE);
+            ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, numberOfMemoryFrames * DEFAULT_FRAME_SIZE,
+                    DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
             handler.open();
             VSizeFrame frame = new VSizeFrame(ctx);
@@ -345,8 +345,8 @@ public class InputHandlerTest extends TestCase {
             Assert.assertEquals(0, handler.getNumDiscarded());
             // We can only discard one frame
             double numDiscarded = 0;
-            boolean nextShouldDiscard =
-                    ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa.getMaxFractionDiscard();
+            boolean nextShouldDiscard = ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa
+                    .getMaxFractionDiscard();
             while (nextShouldDiscard) {
                 handler.nextFrame(frame.getBuffer());
                 numDiscarded++;
@@ -394,8 +394,8 @@ public class InputHandlerTest extends TestCase {
             TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
             writer.freeze();
             // FramePool
-            ConcurrentFramePool framePool =
-                    new ConcurrentFramePool(NODE_ID, discardTestFrames * DEFAULT_FRAME_SIZE, DEFAULT_FRAME_SIZE);
+            ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, discardTestFrames * DEFAULT_FRAME_SIZE,
+                    DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
             handler.open();
             // add NUM_FRAMES times
@@ -411,8 +411,8 @@ public class InputHandlerTest extends TestCase {
             }
             // Next call should NOT block but should discard.
             double numDiscarded = 0.0;
-            boolean nextShouldDiscard =
-                    ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa.getMaxFractionDiscard();
+            boolean nextShouldDiscard = ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa
+                    .getMaxFractionDiscard();
             while (nextShouldDiscard) {
                 handler.nextFrame(buffer);
                 numDiscarded++;
@@ -456,8 +456,8 @@ public class InputHandlerTest extends TestCase {
             TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
             writer.freeze();
             // FramePool
-            ConcurrentFramePool framePool =
-                    new ConcurrentFramePool(NODE_ID, discardTestFrames * DEFAULT_FRAME_SIZE, DEFAULT_FRAME_SIZE);
+            ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, discardTestFrames * DEFAULT_FRAME_SIZE,
+                    DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
             handler.open();
             VSizeFrame frame = new VSizeFrame(ctx);
@@ -467,8 +467,8 @@ public class InputHandlerTest extends TestCase {
             }
             // Next 5 calls call should NOT block but should discard.
             double numDiscarded = 0.0;
-            boolean nextShouldDiscard =
-                    ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa.getMaxFractionDiscard();
+            boolean nextShouldDiscard = ((numDiscarded + 1.0) / (handler.getTotal() + 1.0)) <= fpa
+                    .getMaxFractionDiscard();
             while (nextShouldDiscard) {
                 handler.nextFrame(frame.getBuffer());
                 numDiscarded++;
@@ -507,8 +507,8 @@ public class InputHandlerTest extends TestCase {
         try {
             IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
             // Spill budget = Memory budget, No discard
-            FeedPolicyAccessor fpa =
-                    createFeedPolicyAccessor(true, false, DEFAULT_FRAME_SIZE * NUM_FRAMES, DISCARD_ALLOWANCE);
+            FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, false, DEFAULT_FRAME_SIZE * NUM_FRAMES,
+                    DISCARD_ALLOWANCE);
             // Non-Active Writer
             TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
             writer.freeze();
@@ -554,8 +554,8 @@ public class InputHandlerTest extends TestCase {
             // No spill, No discard
             FeedPolicyAccessor fpa = createFeedPolicyAccessor(false, false, 0L, DISCARD_ALLOWANCE);
             // Non-Active Writer
-            TestFrameWriter writer =
-                    FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(), false);
+            TestFrameWriter writer = FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(),
+                    false);
             // FramePool
             ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, FEED_MEM_BUDGET, DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
@@ -595,8 +595,8 @@ public class InputHandlerTest extends TestCase {
             // No spill, No discard
             FeedPolicyAccessor fpa = createFeedPolicyAccessor(false, false, 0L, DISCARD_ALLOWANCE);
             // Non-Active Writer
-            TestFrameWriter writer =
-                    FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(), false);
+            TestFrameWriter writer = FrameWriterTestUtils.create(Collections.emptyList(), Collections.emptyList(),
+                    false);
             // FramePool
             ConcurrentFramePool framePool = new ConcurrentFramePool(NODE_ID, FEED_MEM_BUDGET, DEFAULT_FRAME_SIZE);
             FeedRuntimeInputHandler handler = createInputHandler(ctx, writer, fpa, framePool);
@@ -683,8 +683,8 @@ public class InputHandlerTest extends TestCase {
                 Random random = new Random();
                 IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
                 // Spill budget = Memory budget, No discard
-                FeedPolicyAccessor fpa =
-                        createFeedPolicyAccessor(true, false, DEFAULT_FRAME_SIZE * NUM_FRAMES, DISCARD_ALLOWANCE);
+                FeedPolicyAccessor fpa = createFeedPolicyAccessor(true, false, DEFAULT_FRAME_SIZE * NUM_FRAMES,
+                        DISCARD_ALLOWANCE);
                 // Non-Active Writer
                 TestControlledFrameWriter writer = FrameWriterTestUtils.create(DEFAULT_FRAME_SIZE, false);
                 writer.freeze();

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/AsterixAppContextInfo.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/AsterixAppContextInfo.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/AsterixAppContextInfo.java
index bf103fc..e7b15a2 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/AsterixAppContextInfo.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/om/util/AsterixAppContextInfo.java
@@ -33,6 +33,7 @@ import org.apache.asterix.common.config.AsterixReplicationProperties;
 import org.apache.asterix.common.config.AsterixStorageProperties;
 import org.apache.asterix.common.config.AsterixTransactionProperties;
 import org.apache.asterix.common.config.IAsterixPropertiesProvider;
+import org.apache.asterix.common.config.MessagingProperties;
 import org.apache.asterix.common.dataflow.IAsterixApplicationContextInfo;
 import org.apache.asterix.common.exceptions.AsterixException;
 import org.apache.asterix.common.library.ILibraryManager;
@@ -62,6 +63,7 @@ public class AsterixAppContextInfo implements IAsterixApplicationContextInfo, IA
     private AsterixBuildProperties buildProperties;
     private AsterixReplicationProperties replicationProperties;
     private AsterixExtensionProperties extensionProperties;
+    private MessagingProperties messagingProperties;
     private final IGlobalRecoveryMaanger globalRecoveryMaanger;
     private IHyracksClientConnection hcc;
     private final ILibraryManager libraryManager;
@@ -92,10 +94,11 @@ public class AsterixAppContextInfo implements IAsterixApplicationContextInfo, IA
         INSTANCE.txnProperties = new AsterixTransactionProperties(propertiesAccessor);
         INSTANCE.feedProperties = new AsterixFeedProperties(propertiesAccessor);
         INSTANCE.extensionProperties = new AsterixExtensionProperties(propertiesAccessor);
-        INSTANCE.replicationProperties =
-                new AsterixReplicationProperties(propertiesAccessor, AsterixClusterProperties.INSTANCE.getCluster());
+        INSTANCE.replicationProperties = new AsterixReplicationProperties(propertiesAccessor,
+                AsterixClusterProperties.INSTANCE.getCluster());
         INSTANCE.hcc = hcc;
         INSTANCE.buildProperties = new AsterixBuildProperties(propertiesAccessor);
+        INSTANCE.messagingProperties = new MessagingProperties(propertiesAccessor);
         Logger.getLogger("org.apache").setLevel(INSTANCE.externalProperties.getLogLevel());
     }
 
@@ -191,4 +194,9 @@ public class AsterixAppContextInfo implements IAsterixApplicationContextInfo, IA
     public AsterixExtensionProperties getExtensionProperties() {
         return extensionProperties;
     }
+
+    @Override
+    public MessagingProperties getMessagingProperties() {
+        return messagingProperties;
+    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/GlobalResourceIdFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/GlobalResourceIdFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/GlobalResourceIdFactory.java
index 5b29530..ab1ebe1 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/GlobalResourceIdFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/GlobalResourceIdFactory.java
@@ -57,7 +57,7 @@ public class GlobalResourceIdFactory implements IResourceIdFactory, IApplication
             //if no response available or it has an exception, request a new one
             if (reponse == null || reponse.getException() != null) {
                 ResourceIdRequestMessage msg = new ResourceIdRequestMessage();
-                ((INCMessageBroker) appCtx.getMessageBroker()).sendMessage(msg, this);
+                ((INCMessageBroker) appCtx.getMessageBroker()).sendMessageToCC(msg, this);
                 reponse = (ResourceIdRequestResponseMessage) resourceIdResponseQ.take();
                 if (reponse.getException() != null) {
                     throw new HyracksDataException(reponse.getException().getMessage());

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/INCApplicationContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/INCApplicationContext.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/INCApplicationContext.java
index 77934c6..b1aa45f 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/INCApplicationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/INCApplicationContext.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hyracks.api.application;
 
+import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
 import org.apache.hyracks.api.context.IHyracksRootContext;
 import org.apache.hyracks.api.lifecycle.ILifeCycleComponentManager;
 import org.apache.hyracks.api.resources.memory.IMemoryManager;
@@ -77,4 +78,19 @@ public interface INCApplicationContext extends IApplicationContext {
      * @param handler
      */
     public void setStateDumpHandler(IStateDumpHandler handler);
+
+    /**
+     * Set the application MessagingChannelInterfaceFactory
+     *
+     * @param interfaceFactory
+     */
+    public void setMessagingChannelInterfaceFactory(IChannelInterfaceFactory interfaceFactory);
+
+    /**
+     * Get the application MessagingChannelInterfaceFactory previously set by
+     * the {@link #setMessagingChannelInterfaceFactory(IChannelInterfaceFactory)} call.
+     *
+     * @return
+     */
+    public IChannelInterfaceFactory getMessagingChannelInterfaceFactory();
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/client/NodeControllerInfo.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/client/NodeControllerInfo.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/client/NodeControllerInfo.java
index c41dafe..a79b955 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/client/NodeControllerInfo.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/client/NodeControllerInfo.java
@@ -33,12 +33,15 @@ public class NodeControllerInfo implements Serializable {
 
     private final NetworkAddress datasetNetworkAddress;
 
+    private final NetworkAddress messagingNetworkAddress;
+
     public NodeControllerInfo(String nodeId, NodeStatus status, NetworkAddress netAddress,
-            NetworkAddress datasetNetworkAddress) {
+            NetworkAddress datasetNetworkAddress, NetworkAddress messagingNetworkAddress) {
         this.nodeId = nodeId;
         this.status = status;
         this.netAddress = netAddress;
         this.datasetNetworkAddress = datasetNetworkAddress;
+        this.messagingNetworkAddress = messagingNetworkAddress;
     }
 
     public String getNodeId() {
@@ -56,4 +59,8 @@ public class NodeControllerInfo implements Serializable {
     public NetworkAddress getDatasetNetworkAddress() {
         return datasetNetworkAddress;
     }
+
+    public NetworkAddress getMessagingNetworkAddress() {
+        return messagingNetworkAddress;
+    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferAcceptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferAcceptor.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferAcceptor.java
new file mode 100644
index 0000000..585b6bd
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferAcceptor.java
@@ -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 org.apache.hyracks.api.comm;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Accepts buffers.
+ *
+ * @author vinayakb
+ */
+@FunctionalInterface
+public interface IBufferAcceptor {
+    /**
+     * Accept a buffer.
+     *
+     * @param buffer
+     */
+    public void accept(ByteBuffer buffer);
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferFactory.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferFactory.java
new file mode 100644
index 0000000..5b3a233
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IBufferFactory.java
@@ -0,0 +1,30 @@
+/*
+ * 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.hyracks.api.comm;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+@FunctionalInterface
+public interface IBufferFactory {
+
+    public ByteBuffer createBuffer() throws HyracksDataException;
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelControlBlock.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelControlBlock.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelControlBlock.java
new file mode 100644
index 0000000..02de858
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelControlBlock.java
@@ -0,0 +1,83 @@
+/*
+ * 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.hyracks.api.comm;
+
+public interface IChannelControlBlock {
+
+    /**
+     * Get the read interface of this channel.
+     *
+     * @return the read interface.
+     */
+    public IChannelReadInterface getReadInterface();
+
+    /**
+     * Get the write interface of this channel.
+     *
+     * @return the write interface.
+     */
+    public IChannelWriteInterface getWriteInterface();
+
+    /**
+     * Add write credit to this channel.
+     *
+     * @param delta
+     *            number of bytes
+     */
+    public void addWriteCredits(int delta);
+
+    /**
+     * @return The channel's unique id within its ChannelSet.
+     */
+    public int getChannelId();
+
+    /**
+     * Add pending credit.
+     *
+     * @param credit
+     */
+    public void addPendingCredits(int credit);
+
+    /**
+     * Increments the pending write operations of this channel.
+     */
+    public void markPendingWrite();
+
+    /**
+     * Clears the pending write operations of this channel.
+     */
+    public void unmarkPendingWrite();
+
+    /**
+     * Sets a flag indicating this channel was closed locally.
+     */
+    public void reportLocalEOS();
+
+    /**
+     * A flag indicating if the channel was closed on the remote side.
+     *
+     * @return
+     */
+    public boolean isRemotelyClosed();
+
+    /**
+     * Complete the current write operation on this channel.
+     */
+    public void writeComplete();
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelInterfaceFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelInterfaceFactory.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelInterfaceFactory.java
new file mode 100644
index 0000000..d147fc7
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelInterfaceFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.hyracks.api.comm;
+
+public interface IChannelInterfaceFactory {
+
+    /**
+     * Creates {@link IChannelReadInterface} and assigns the passed
+     * {@link IChannelControlBlock} to it.
+     *
+     * @param ccb
+     * @return
+     */
+    public IChannelReadInterface createReadInterface(IChannelControlBlock ccb);
+
+    /**
+     * Creates {@link IChannelWriteInterface} and assigns the passed
+     * {@link IChannelControlBlock} to it.
+     *
+     * @param ccb
+     * @return
+     */
+    public IChannelWriteInterface createWriteInterface(IChannelControlBlock ccb);
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelReadInterface.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelReadInterface.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelReadInterface.java
new file mode 100644
index 0000000..357d761
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelReadInterface.java
@@ -0,0 +1,99 @@
+/*
+ * 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.hyracks.api.comm;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+
+import org.apache.hyracks.api.exceptions.NetException;
+
+/**
+ * Represents the read interface of a {@link IChannelControlBlock}.
+ *
+ * @author vinayakb
+ */
+public interface IChannelReadInterface {
+    /**
+     * Set the callback that will be invoked by the network layer when a buffer has been
+     * filled with data received from the remote side.
+     *
+     * @param fullBufferAcceptor
+     *            - the full buffer acceptor.
+     */
+    public void setFullBufferAcceptor(ICloseableBufferAcceptor fullBufferAcceptor);
+
+    /**
+     * Get the acceptor that collects empty buffers when the client has finished consuming
+     * a previously full buffer.
+     *
+     * @return the empty buffer acceptor.
+     */
+    public IBufferAcceptor getEmptyBufferAcceptor();
+
+    /**
+     * Set the buffer factory which is in charge of creating buffers if the request does not
+     * make the number of allocated buffers goes beyond limit
+     *
+     * @param bufferFactory
+     *            - the buffer factory
+     * @param limit
+     *            - the limit of buffers
+     * @param frameSize
+     *            - the size of each buffer
+     */
+    public void setBufferFactory(IBufferFactory bufferFactory, int limit, int frameSize);
+
+    /**
+     * Try to read as much as {@code size} bytes from {@code sc}
+     *
+     * @param sc
+     * @param size
+     * @return The number of read bytes.
+     * @throws IOException
+     * @throws NetException
+     */
+    public int read(SocketChannel sc, int size) throws IOException, NetException;
+
+    /**
+     * Sets the read credits of this {@link IChannelReadInterface}
+     *
+     * @param credits
+     */
+    public void setReadCredits(int credits);
+
+    /**
+     * @return The current read credits of this {@link IChannelReadInterface}
+     */
+    public int getCredits();
+
+    /**
+     * Forces the current read buffer to be flushed
+     */
+    public void flush();
+
+    /**
+     * @return The current full buffer acceptor of this {@link IChannelReadInterface}
+     */
+    public ICloseableBufferAcceptor getFullBufferAcceptor();
+
+    /**
+     * @return The buffer factory used by this {@link IChannelReadInterface}
+     */
+    public IBufferFactory getBufferFactory();
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/fb7f4d32/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelWriteInterface.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelWriteInterface.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelWriteInterface.java
new file mode 100644
index 0000000..993dd2c
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IChannelWriteInterface.java
@@ -0,0 +1,88 @@
+/*
+ * 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.hyracks.api.comm;
+
+import org.apache.hyracks.api.exceptions.NetException;
+
+/**
+ * Represents the write interface of a {@link IChannelControlBlock}.
+ */
+public interface IChannelWriteInterface {
+    /**
+     * Set the callback interface that must be invoked when a full buffer has been emptied by
+     * writing the data to the remote end.
+     *
+     * @param emptyBufferAcceptor
+     *            - the empty buffer acceptor.
+     */
+    public void setEmptyBufferAcceptor(IBufferAcceptor emptyBufferAcceptor);
+
+    /**
+     * Get the full buffer acceptor that accepts buffers filled with data that need to be written
+     * to the remote end.
+     *
+     * @return the full buffer acceptor.
+     */
+    public ICloseableBufferAcceptor getFullBufferAcceptor();
+
+    /**
+     * Set the buffer factory which is in charge of creating buffers if the request does not
+     * make the number of allocated buffers goes beyond limit
+     *
+     * @param bufferFactory
+     *            - the buffer factory
+     * @param limit
+     *            - the limit of buffers
+     * @param frameSize
+     *            - the size of each buffer
+     */
+    public void setBufferFactory(IBufferFactory bufferFactory, int limit, int frameSize);
+
+    /**
+     * Performs a pending write operation based on the current state of
+     * this {@link IChannelWriteInterface}
+     *
+     * @param writerState
+     * @throws NetException
+     */
+    public void write(IConnectionWriterState writerState) throws NetException;
+
+    /**
+     * Completes the current write operation on this {@link IChannelWriteInterface}
+     */
+    public void writeComplete();
+
+    /**
+     * Add credits to this this {@link IChannelWriteInterface}
+     *
+     * @param credit
+     */
+    public void addCredits(int credit);
+
+    /**
+     * @return The current credits of this {@link IChannelWriteInterface}
+     */
+    public int getCredits();
+
+    /**
+     * Adjusts the {@link IChannelControlBlock} writability based on the current
+     * state of this {@link IChannelWriteInterface}
+     */
+    public void adjustChannelWritability();
+}


Mime
View raw message