<commit_before>/* 
 * Copyright (c) 2007-2008, Hazel Ltd. All Rights Reserved.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package com.hazelcast.impl;

import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.impl.BlockingQueueManager.Q.ScheduledOfferAction;
import com.hazelcast.impl.BlockingQueueManager.Q.ScheduledPollAction;
import com.hazelcast.impl.ClusterManager.AbstractRemotelyProcessable;
import com.hazelcast.impl.Config.QConfig;
import static com.hazelcast.impl.Constants.BlockingQueueOperations.*;
import static com.hazelcast.impl.Constants.Objects.OBJECT_NULL;
import com.hazelcast.nio.Address;
import static com.hazelcast.nio.BufferUtil.*;
import com.hazelcast.nio.Data;
import com.hazelcast.nio.DataSerializable;
import com.hazelcast.nio.PacketQueue;
import com.hazelcast.nio.PacketQueue.Packet;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.*;

class BlockingQueueManager extends BaseManager {
    private final Request remoteReq = new Request();
    private final static BlockingQueueManager instance = new BlockingQueueManager();
    private final Map<String, Q> mapQueues = new HashMap<String, Q>(10);
    private final Map<Long, List<Data>> mapTxnPolledElements = new HashMap<Long, List<Data>>(10);
    private static final int BLOCK_SIZE = 1000;
    private int nextIndex = 0;

    private BlockingQueueManager() {
        ClusterService.get().registerPacketProcessor(OP_B_POLL, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_OFFER, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleOffer(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_BACKUP_ADD, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleBackup(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_BACKUP_REMOVE, new PacketProcessor() {
            public void process(Packet packet) {
                handleBackup(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_PUBLISH, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePublish(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_ADD_TOPIC_LISTENER, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleAddTopicListener(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_SIZE, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleSize(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_PEEK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_READ, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRead(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_REMOVE, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRemove(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_TXN_BACKUP_POLL, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleTxnBackupPoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_TXN_COMMIT, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleTxnCommit(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_ADD_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleAddBlock(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_REMOVE_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRemoveBlock(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_FULL_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleFullBlock(packet);
            }
        });

    }

    public static BlockingQueueManager get() {
        return instance;
    }

    class BlockBackupSyncRunner implements Runnable {
        final BlockBackupSync blockSync;

        public BlockBackupSyncRunner(BlockBackupSync blockSync) {
            this.blockSync = blockSync;
        }

        public void run() {
            while (!blockSync.done) {
                try {
                    synchronized (blockSync) {
                        enqueueAndReturn(blockSync);
                        blockSync.wait();
                    }
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class BlockBackupSync implements Processable {
        final Block block;
        final Q q;
        int index = 0;
        final int indexUpto;
        volatile boolean done = false;

        public BlockBackupSync(Q q, Block block, int indexUpto) {
            super();
            this.q = q;
            this.block = block;
            this.indexUpto = indexUpto;
        }

        public void process() {
            Data data = next();
            if (data != null && data.size() != 0) {
                q.sendBackup(true, thisAddress, data, block.blockId, index);
                index++;
                if (index > indexUpto) {
                    done = true;
                }
            } else {
                done = true;
            }
            synchronized (BlockBackupSync.this) {
                BlockBackupSync.this.notify();
            }
        }

        private Data next() {
            while (true) {
                Data data = block.get(index);
                if (data != null) {
                    return data;
                }
                index++;
                if (index > indexUpto)
                    return null;
                if (index >= BLOCK_SIZE)
                    return null;
            }
        }
    }

    void syncForDead(Address addressDead) {
        MemberImpl member = getNextMemberBeforeSync(addressDead, true, 1);
        if (DEBUG) {
            log(addressDead + " is dead and its backup was " + member);
        }
        Address addressNewOwner = (member == null) ? thisAddress : member.getAddress();

        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            List<Block> lsBlocks = q.lsBlocks;
            for (Block block : lsBlocks) {
                if (block.address.equals(addressDead)) {
                    // set the new owner
                    block.address = addressNewOwner;
                    block.resetAddIndex();
                    if (lsMembers.size() > 1) {
                        if (addressNewOwner.equals(thisAddress)) {
                            // I am the new owner so backup to next member
                            int indexUpto = block.size() - 1;
                            if (DEBUG) {
                                log("IndexUpto " + indexUpto);
                            }
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                } else if (block.address.equals(thisAddress)) {
                    // If I am/was the owner of this block
                    // did my backup change..
                    // if so backup to the new next
                    if (lsMembers.size() > 1) {
                        MemberImpl memberBackupWas = getNextMemberBeforeSync(thisAddress, true, 1);
                        if (memberBackupWas == null
                                || memberBackupWas.getAddress().equals(addressDead)) {
                            int indexUpto = block.size() - 1;
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                }
            }
            // packetalidate the dead member's scheduled actions
            List<ScheduledPollAction> scheduledPollActions = q.lsScheduledPollActions;
            for (ScheduledPollAction scheduledAction : scheduledPollActions) {
                if (addressDead.equals(scheduledAction.request.caller)) {
                    scheduledAction.setValid(false);
                    ClusterManager.get().deregisterScheduledAction(scheduledAction);
                }
            }
            List<ScheduledOfferAction> scheduledOfferActions = q.lsScheduledOfferActions;
            for (ScheduledOfferAction scheduledAction : scheduledOfferActions) {
                if (addressDead.equals(scheduledAction.request.caller)) {
                    scheduledAction.setValid(false);
                    ClusterManager.get().deregisterScheduledAction(scheduledAction);
                }
            }
        }
        doResetBlockSizes();
    }

    public void syncForAdd() {
        if (isMaster()) {
            Collection<Q> queues = mapQueues.values();
            for (Q q : queues) {
                List<Block> lsBlocks = q.lsBlocks;
                for (Block block : lsBlocks) {
                    int fullBlockId = -1;
                    if (block.isFull()) {
                        fullBlockId = block.blockId;
                    }
                    sendAddBlockMessageToOthers(block, fullBlockId, null, true);
                }
            }
        }
        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            List<Block> lsBlocks = q.lsBlocks;
            for (Block block : lsBlocks) {
                if (block.address.equals(thisAddress)) {
                    // If I am/was the owner of this block
                    // did my backup change..
                    // if so backup to the new next
                    if (lsMembers.size() > 1) {
                        MemberImpl memberBackupWas = getNextMemberBeforeSync(thisAddress, true, 1);
                        MemberImpl memberBackupIs = getNextMemberAfter(thisAddress, true, 1);
                        if (memberBackupWas == null || !memberBackupWas.equals(memberBackupIs)) {
                            if (DEBUG) {
                                log("Backup changed!!! so backing up to " + memberBackupIs);
                            }
                            int indexUpto = block.size() - 1;
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                }
            }
        }
        doResetBlockSizes();
    }

    void doResetBlockSizes() {
        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            // q.printStack();
            List<Block> lsBlocks = q.lsBlocks;
            int size = 0;
            for (Block block : lsBlocks) {
                if (block.address.equals(thisAddress)) {
                    size += block.size();
                }
            }
            q.size = size;
        }
    }

    final void handleSize(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        packet.longValue = q.size;
        sendResponse(packet);
    }

    @Override
    final void handleListenerRegisterations(boolean add, String name, Data key, Address address,
                                            boolean includeValue) {
        Q q = getQ(name);
        if (add) {
            q.addListener(address, includeValue);
        } else {
            q.removeListener(address);
        }
    }

    final void handleTxnBackupPoll(PacketQueue.Packet packet) {
        doTxnBackupPoll(packet.txnId, doTake(packet.value));
    }

    final void handleTxnCommit(PacketQueue.Packet packet) {
        mapTxnPolledElements.remove(packet.txnId);
    }

    final void doTxnBackupPoll(long txnId, Data value) {
        List<Data> lsTxnPolledElements = mapTxnPolledElements.get(txnId);
        if (lsTxnPolledElements == null) {
            lsTxnPolledElements = new ArrayList<Data>(1);
            mapTxnPolledElements.put(txnId, lsTxnPolledElements);
        }
        lsTxnPolledElements.add(value);
    }

    final void handleBackup(PacketQueue.Packet packet) {
        try {
            String name = packet.name;
            int blockId = packet.blockId;
            Q q = getQ(name);
            if (packet.operation == OP_B_BACKUP_ADD) {
                Data data = doTake(packet.value);
                q.doBackup(true, data, blockId, (int) packet.longValue);
            } else if (packet.operation == OP_B_BACKUP_REMOVE) {
                q.doBackup(false, null, blockId, 0);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    final void handleRemoveBlock(PacketQueue.Packet packet) {
        try {
            BlockUpdate addRemove = (BlockUpdate) ThreadContext.get().toObject(packet.value);
            String name = packet.name;
            Q q = getQ(name);
            doRemoveBlock(q, packet.conn.getEndPoint(), addRemove.removeBlockId);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    public void doRemoveBlock(Q q, Address originalRemover, int blockId) {
        Block blockRemoved = q.removeBlock(blockId);
        if (blockRemoved != null) {
            if (isMaster()) {
                sendAddBlockMessageToOthers(blockRemoved, -1, originalRemover, false);
            }
        }
    }

    final void handleFullBlock(PacketQueue.Packet packet) {
        if (isMaster()) {
            String name = packet.name;
            Q q = getQ(name);
            doFullBlock(q, packet.blockId, packet.conn.getEndPoint());
        }
        packet.returnToContainer();
    }

    final void doFullBlock(Q q, int fullBlockId, Address originalFuller) {
        int blockId = q.getLatestAddedBlock() + 1;
        Address target = nextTarget();
        Block newBlock = setFullAndCreateNewBlock(q, fullBlockId, target, blockId);
        if (newBlock != null) {
            sendAddBlockMessageToOthers(newBlock, fullBlockId, originalFuller, true);
        }
    }

    final void handleAddBlock(PacketQueue.Packet packet) {
        try {
            BlockUpdate addRemove = (BlockUpdate) ThreadContext.get().toObject(packet.value);
            String name = packet.name;
            int blockId = addRemove.addBlockId;
            Address addressOwner = addRemove.addAddress;
            Q q = getQ(name);
            setFullAndCreateNewBlock(q, addRemove.fullBlockId, addressOwner, blockId);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    Block setFullAndCreateNewBlock(Q q, int fullBlockId, Address newBlockOwner, int newBlockId) {
        List<Block> lsBlocks = q.lsBlocks;
        boolean exist = false;
        for (Block block : lsBlocks) {
            if (block.blockId == newBlockId) {
                exist = true;
            }
        }
        Block newBlock = null;
        if (!exist) {
            newBlock = q.createBlock(newBlockOwner, newBlockId);
            q.addBlock(newBlock);
        }
        if (fullBlockId != -1) {
            q.setBlockFull(fullBlockId);
        }
        return newBlock;
    }

    final void handleOffer(PacketQueue.Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doOffer(remoteReq);
            packet.longValue = remoteReq.longValue;
            if (!remoteReq.scheduled) {
                if (remoteReq.response == Boolean.TRUE) {
                    sendResponse(packet);
                } else {
                    sendResponseFailure(packet);
                }
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handlePublish(Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doPublish(remoteReq);
            packet.longValue = remoteReq.longValue;
            if (!remoteReq.scheduled) {
                if (remoteReq.response == Boolean.TRUE) {
                    sendResponse(packet);
                } else {
                    sendResponseFailure(packet);
                }
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handleAddTopicListener(PacketQueue.Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doAddTopicListener(remoteReq);
            packet.longValue = remoteReq.longValue;
            sendResponse(packet);
            remoteReq.reset();
        }
    }

    final boolean rightRemoteOfferTarget(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        if (q.blCurrentPut == null) {
            q.setCurrentPut();
        }
        boolean invalid = false;
        if (packet.blockId != q.blCurrentPut.blockId) {
            if (packet.blockId > q.blCurrentPut.blockId) {
                int size = q.lsBlocks.size();
                for (int i = 0; i < size; i++) {
                    Block block = q.lsBlocks.get(i);
                    if (block.blockId == packet.blockId) {
                        if (thisAddress.equals(block.address)) {
                            q.blCurrentPut = block;
                        }
                    }
                }
            } else {
                invalid = true;
            }
        }
        if (q.blCurrentPut.isFull() || !thisAddress.equals(q.blCurrentPut.address)) {
            invalid = true;
        }
        if (invalid) {
            sendRedoResponse(packet);
        }
        return (!invalid);
    }

    boolean rightRemotePollTarget(Packet packet) {
        Q q = getQ(packet.name);
        if (q.blCurrentTake == null) {
            q.setCurrentTake();
        }
        boolean invalid = false;
        if (packet.blockId != q.blCurrentTake.blockId) {
            if (packet.blockId > q.blCurrentTake.blockId) {
                int size = q.lsBlocks.size();
                for (int i = 0; i < size; i++) {
                    Block block = q.lsBlocks.get(i);
                    if (block.blockId == packet.blockId) {
                        if (thisAddress.equals(block.address)) {
                            q.blCurrentTake = block;
                        }
                    }
                }
            } else {
                invalid = true;
            }
        }
        if ((q.blCurrentTake.size() == 0 && q.blCurrentTake.isFull())
                || !thisAddress.equals(q.blCurrentTake.address)) {
            invalid = true;
        }
        if (invalid) {
            sendRedoResponse(packet);
        }
        return (!invalid);
    }

    final void handlePoll(PacketQueue.Packet packet) {
        if (rightRemotePollTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doPoll(remoteReq);
            if (!remoteReq.scheduled) {
                Data oldValue = (Data) remoteReq.response;
                if (oldValue != null && oldValue.size() > 0) {
                    doSet(oldValue, packet.value);
                }
                sendResponse(packet);
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handleRead(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        q.read(packet);
    }

    final void handleRemove(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        q.remove(packet);
    }

    final Address nextTarget() {
        int size = lsMembers.size();
        int index = nextIndex++ % size;
        if (nextIndex >= size) {
            nextIndex = 0;
        }
        return lsMembers.get(index).getAddress();
    }

    public Q getQ(String name) {
        if (name == null)
            return null;
        Q q = mapQueues.get(name);
        if (q == null) {
            q = new Q(name);
            mapQueues.put(name, q);
        }
        return q;
    }

    public class CommitPoll extends AbstractCall {
        volatile long txnId = -1;
        String name = null;

        public int getOperation() {
            return OP_B_TXN_COMMIT;
        }

        public void commitPoll(String name) {
            this.name = name;
            this.txnId = ThreadContext.get().getTxnId();
            enqueueAndReturn(this);
        }

        public void handleResponse(PacketQueue.Packet packet) {
        }

        public void process() {
            MemberImpl nextMember = getNextMemberAfter(thisAddress, true, 1);
            if (nextMember != null) {
                Address next = nextMember.getAddress();
                PacketQueue.Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.txnId = txnId;
                boolean sent = send(packet, next);
                if (!sent) {
                    packet.returnToContainer();
                }
            }
        }
    }

    class QDestroy {
        public void destroy(String name) {
            sendProcessableToAll(new Destroy(name), true);
        }
    }

    public void destroy(String name) {
        mapQueues.remove(name);
    }

    public class Size extends AbstractCall {
        String name = null;
        int total = 0;
        int numberOfResponses = 0;
        int numberOfExpectedResponses = 1;

        public int getSize(String name) {
            this.name = name;
            synchronized (this) {
                try {
                    enqueueAndReturn(this);
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return total;
        }

        @Override
        public void onDisconnect(Address dead) {
            redo();
        }

        public void reset() {
            total = 0;
            numberOfResponses = 0;
            numberOfExpectedResponses = 0;
        }

        public void handleResponse(PacketQueue.Packet packet) {
            total += (int) packet.longValue;
            packet.returnToContainer();
            numberOfResponses++;
            if (numberOfResponses >= numberOfExpectedResponses) {
                removeCall(getId());
                synchronized (this) {
                    notify();
                }
            }
        }

        public void process() {
            reset();
            Q q = getQ(name);
            if (DEBUG) {
                q.printStack();
            }
            total += q.size;
            numberOfResponses = 1;
            if (lsMembers.size() > 1) {
                addCall(this);
                numberOfExpectedResponses = lsMembers.size();
                for (MemberImpl member : lsMembers) {
                    if (!member.localMember()) {
                        PacketQueue.Packet packet = obtainPacket();
                        packet.name = name;
                        packet.operation = getOperation();
                        packet.callId = getId();
                        packet.timeout = 0;
                        boolean sent = send(packet, member.getAddress());
                        if (!sent) {
                            packet.returnToContainer();
                            redo();
                        }
                    }
                }
            } else {
                synchronized (this) {
                    this.notify();
                }
            }
        }

        public int getOperation() {
            return OP_B_SIZE;
        }
    }

    class QIterator<E> implements Iterator<E>, Processable {

        volatile String name;
        volatile List<Integer> blocks = null;
        int currentBlockId = -1;
        int currentIndex = -1;
        Read read = new Read();
        Object next = null;
        boolean hasNextCalled = false;

        public void set(String name) {
            synchronized (QIterator.this) {
                this.name = name;
                enqueueAndReturn(QIterator.this);
                try {
                    QIterator.this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        public void process() {
            Q q = getQ(name);
            List<Block> lsBlocks = q.lsBlocks;
            blocks = new ArrayList<Integer>(lsBlocks.size());
            for (Block block : lsBlocks) {
                blocks.add(block.blockId);
            }
            synchronized (QIterator.this) {
                QIterator.this.notify();
            }
        }

        public boolean hasNext() {
            next = null;
            while (next == null) {
                boolean canRead = setNextBlock();
                // System.out.println(currentBlockId + " Can read " + canRead);
                if (!canRead)
                    return false;
                next = read.read(name, currentBlockId, currentIndex);
                // System.out.println(currentIndex + " Next " + next);
                if (next == null) {
                    currentIndex = -1;
                } else {
                    currentIndex = read.readIndex;
                    currentIndex++;
                }
            }
            hasNextCalled = true;
            return true;
        }

        boolean setNextBlock() {
            if (currentIndex == -1 || currentIndex >= BLOCK_SIZE) {
                if (blocks.size() == 0) {
                    return false;
                } else {
                    currentBlockId = blocks.remove(0);
                    currentIndex = 0;
                    return true;
                }
            }
            return true;
        }

        public E next() {
            if (!hasNextCalled) {
                boolean hasNext = hasNext();
                if (!hasNext) {
                    return null;
                }
            }
            if (next != null) {
                hasNextCalled = false;
            }
            return (E) next;
        }

        public void remove() {
        }

    }

    public class Remove extends QueueBasedCall {
        String name;
        int blockId;
        int index;

        public Object remove(String name, int blockId, int index) {
            this.name = name;
            this.blockId = blockId;
            this.index = index;
            enqueueAndReturn(this);
            Object result;
            try {
                result = responses.take();
                if (result == OBJECT_NULL)
                    return null;
                else {
                    return ThreadContext.get().toObject((Data) result);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        void setResponse() {
            responses.add(OBJECT_NULL);
        }

        public void handleResponse(PacketQueue.Packet packet) {
            setResponse();
            packet.returnToContainer();
        }

        @Override
        public void onDisconnect(Address dead) {
            process();
        }

        public void process() {
            addCall(this);
            Q q = getQ(name);

            Address target = q.getBlockOwner(blockId);
            if (target == null) {
                responses.add(OBJECT_NULL);
            } else if (target.equals(thisAddress)) {
                q.remove(this);
            } else {
                Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.callId = getId();
                packet.blockId = blockId;
                packet.timeout = 0;
                packet.longValue = index;
                boolean sent = false;
                if (target != null) {
                    sent = send(packet, target);
                }
                if (target == null || !sent) {
                    packet.returnToContainer();
                    responses.add(OBJECT_NULL);
                }
            }
        }

        public int getOperation() {
            return OP_B_REMOVE;
        }

    }

    public class Read extends QueueBasedCall {
        String name;
        int blockId;
        int index;
        int readIndex = 0;

        public Object read(String name, int blockId, int index) {
            this.name = name;
            this.blockId = blockId;
            this.index = index;
            enqueueAndReturn(this);
            Object result;
            try {
                result = responses.take();
                if (result == OBJECT_NULL)
                    return null;
                else {
                    return ThreadContext.get().toObject((Data) result);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        void setResponse(Data value, int index) {
            readIndex = index;
            responses.add((value == null) ? OBJECT_NULL : value);
        }

        public void handleResponse(PacketQueue.Packet packet) {
            Data value = doTake(packet.value);
            int indexRead = (int) packet.longValue;
            setResponse(value, indexRead);
            packet.returnToContainer();
        }

        @Override
        public void onDisconnect(Address dead) {
            enqueueAndReturn(Read.this);
        }

        public void process() {
            addCall(this);
            Q q = getQ(name);

            Address target = q.getBlockOwner(blockId);
            if (target == null) {
                responses.add(OBJECT_NULL);
            } else if (target.equals(thisAddress)) {
                q.read(this);
            } else {
                PacketQueue.Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.callId = getId();
                packet.blockId = blockId;
                packet.timeout = 0;
                packet.longValue = index;
                boolean sent = send(packet, target);
                if (!sent) {
                    packet.returnToContainer();
                    onDisconnect(target);
                }<commit_msg>Q iterator read call wasn't terminated/removed properly. <commit_after>/* 
 * Copyright (c) 2007-2008, Hazel Ltd. All Rights Reserved.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package com.hazelcast.impl;

import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.impl.BlockingQueueManager.Q.ScheduledOfferAction;
import com.hazelcast.impl.BlockingQueueManager.Q.ScheduledPollAction;
import com.hazelcast.impl.ClusterManager.AbstractRemotelyProcessable;
import com.hazelcast.impl.Config.QConfig;
import static com.hazelcast.impl.Constants.BlockingQueueOperations.*;
import static com.hazelcast.impl.Constants.Objects.OBJECT_NULL;
import com.hazelcast.nio.Address;
import static com.hazelcast.nio.BufferUtil.*;
import com.hazelcast.nio.Data;
import com.hazelcast.nio.DataSerializable;
import com.hazelcast.nio.PacketQueue;
import com.hazelcast.nio.PacketQueue.Packet;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.*;

class BlockingQueueManager extends BaseManager {
    private final Request remoteReq = new Request();
    private final static BlockingQueueManager instance = new BlockingQueueManager();
    private final Map<String, Q> mapQueues = new HashMap<String, Q>(10);
    private final Map<Long, List<Data>> mapTxnPolledElements = new HashMap<Long, List<Data>>(10);
    private static final int BLOCK_SIZE = 1000;
    private int nextIndex = 0;

    private BlockingQueueManager() {
        ClusterService.get().registerPacketProcessor(OP_B_POLL, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_OFFER, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleOffer(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_BACKUP_ADD, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleBackup(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_BACKUP_REMOVE, new PacketProcessor() {
            public void process(Packet packet) {
                handleBackup(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_PUBLISH, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePublish(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_ADD_TOPIC_LISTENER, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleAddTopicListener(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_SIZE, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleSize(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_PEEK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handlePoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_READ, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRead(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_REMOVE, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRemove(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_TXN_BACKUP_POLL, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleTxnBackupPoll(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_TXN_COMMIT, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleTxnCommit(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_ADD_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleAddBlock(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_REMOVE_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleRemoveBlock(packet);
            }
        });
        ClusterService.get().registerPacketProcessor(OP_B_FULL_BLOCK, new PacketProcessor() {
            public void process(PacketQueue.Packet packet) {
                handleFullBlock(packet);
            }
        });

    }

    public static BlockingQueueManager get() {
        return instance;
    }

    class BlockBackupSyncRunner implements Runnable {
        final BlockBackupSync blockSync;

        public BlockBackupSyncRunner(BlockBackupSync blockSync) {
            this.blockSync = blockSync;
        }

        public void run() {
            while (!blockSync.done) {
                try {
                    synchronized (blockSync) {
                        enqueueAndReturn(blockSync);
                        blockSync.wait();
                    }
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class BlockBackupSync implements Processable {
        final Block block;
        final Q q;
        int index = 0;
        final int indexUpto;
        volatile boolean done = false;

        public BlockBackupSync(Q q, Block block, int indexUpto) {
            super();
            this.q = q;
            this.block = block;
            this.indexUpto = indexUpto;
        }

        public void process() {
            Data data = next();
            if (data != null && data.size() != 0) {
                q.sendBackup(true, thisAddress, data, block.blockId, index);
                index++;
                if (index > indexUpto) {
                    done = true;
                }
            } else {
                done = true;
            }
            synchronized (BlockBackupSync.this) {
                BlockBackupSync.this.notify();
            }
        }

        private Data next() {
            while (true) {
                Data data = block.get(index);
                if (data != null) {
                    return data;
                }
                index++;
                if (index > indexUpto)
                    return null;
                if (index >= BLOCK_SIZE)
                    return null;
            }
        }
    }

    void syncForDead(Address addressDead) {
        MemberImpl member = getNextMemberBeforeSync(addressDead, true, 1);
        if (DEBUG) {
            log(addressDead + " is dead and its backup was " + member);
        }
        Address addressNewOwner = (member == null) ? thisAddress : member.getAddress();

        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            List<Block> lsBlocks = q.lsBlocks;
            for (Block block : lsBlocks) {
                if (block.address.equals(addressDead)) {
                    // set the new owner
                    block.address = addressNewOwner;
                    block.resetAddIndex();
                    if (lsMembers.size() > 1) {
                        if (addressNewOwner.equals(thisAddress)) {
                            // I am the new owner so backup to next member
                            int indexUpto = block.size() - 1;
                            if (DEBUG) {
                                log("IndexUpto " + indexUpto);
                            }
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                } else if (block.address.equals(thisAddress)) {
                    // If I am/was the owner of this block
                    // did my backup change..
                    // if so backup to the new next
                    if (lsMembers.size() > 1) {
                        MemberImpl memberBackupWas = getNextMemberBeforeSync(thisAddress, true, 1);
                        if (memberBackupWas == null
                                || memberBackupWas.getAddress().equals(addressDead)) {
                            int indexUpto = block.size() - 1;
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                }
            }
            // packetalidate the dead member's scheduled actions
            List<ScheduledPollAction> scheduledPollActions = q.lsScheduledPollActions;
            for (ScheduledPollAction scheduledAction : scheduledPollActions) {
                if (addressDead.equals(scheduledAction.request.caller)) {
                    scheduledAction.setValid(false);
                    ClusterManager.get().deregisterScheduledAction(scheduledAction);
                }
            }
            List<ScheduledOfferAction> scheduledOfferActions = q.lsScheduledOfferActions;
            for (ScheduledOfferAction scheduledAction : scheduledOfferActions) {
                if (addressDead.equals(scheduledAction.request.caller)) {
                    scheduledAction.setValid(false);
                    ClusterManager.get().deregisterScheduledAction(scheduledAction);
                }
            }
        }
        doResetBlockSizes();
    }

    public void syncForAdd() {
        if (isMaster()) {
            Collection<Q> queues = mapQueues.values();
            for (Q q : queues) {
                List<Block> lsBlocks = q.lsBlocks;
                for (Block block : lsBlocks) {
                    int fullBlockId = -1;
                    if (block.isFull()) {
                        fullBlockId = block.blockId;
                    }
                    sendAddBlockMessageToOthers(block, fullBlockId, null, true);
                }
            }
        }
        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            List<Block> lsBlocks = q.lsBlocks;
            for (Block block : lsBlocks) {
                if (block.address.equals(thisAddress)) {
                    // If I am/was the owner of this block
                    // did my backup change..
                    // if so backup to the new next
                    if (lsMembers.size() > 1) {
                        MemberImpl memberBackupWas = getNextMemberBeforeSync(thisAddress, true, 1);
                        MemberImpl memberBackupIs = getNextMemberAfter(thisAddress, true, 1);
                        if (memberBackupWas == null || !memberBackupWas.equals(memberBackupIs)) {
                            if (DEBUG) {
                                log("Backup changed!!! so backing up to " + memberBackupIs);
                            }
                            int indexUpto = block.size() - 1;
                            if (indexUpto > -1) {
                                executeLocally(new BlockBackupSyncRunner(new BlockBackupSync(q, block,
                                        indexUpto)));
                            }
                        }
                    }
                }
            }
        }
        doResetBlockSizes();
    }

    void doResetBlockSizes() {
        Collection<Q> queues = mapQueues.values();
        for (Q q : queues) {
            // q.printStack();
            List<Block> lsBlocks = q.lsBlocks;
            int size = 0;
            for (Block block : lsBlocks) {
                if (block.address.equals(thisAddress)) {
                    size += block.size();
                }
            }
            q.size = size;
        }
    }

    final void handleSize(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        packet.longValue = q.size;
        sendResponse(packet);
    }

    @Override
    final void handleListenerRegisterations(boolean add, String name, Data key, Address address,
                                            boolean includeValue) {
        Q q = getQ(name);
        if (add) {
            q.addListener(address, includeValue);
        } else {
            q.removeListener(address);
        }
    }

    final void handleTxnBackupPoll(PacketQueue.Packet packet) {
        doTxnBackupPoll(packet.txnId, doTake(packet.value));
    }

    final void handleTxnCommit(PacketQueue.Packet packet) {
        mapTxnPolledElements.remove(packet.txnId);
    }

    final void doTxnBackupPoll(long txnId, Data value) {
        List<Data> lsTxnPolledElements = mapTxnPolledElements.get(txnId);
        if (lsTxnPolledElements == null) {
            lsTxnPolledElements = new ArrayList<Data>(1);
            mapTxnPolledElements.put(txnId, lsTxnPolledElements);
        }
        lsTxnPolledElements.add(value);
    }

    final void handleBackup(PacketQueue.Packet packet) {
        try {
            String name = packet.name;
            int blockId = packet.blockId;
            Q q = getQ(name);
            if (packet.operation == OP_B_BACKUP_ADD) {
                Data data = doTake(packet.value);
                q.doBackup(true, data, blockId, (int) packet.longValue);
            } else if (packet.operation == OP_B_BACKUP_REMOVE) {
                q.doBackup(false, null, blockId, 0);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    final void handleRemoveBlock(PacketQueue.Packet packet) {
        try {
            BlockUpdate addRemove = (BlockUpdate) ThreadContext.get().toObject(packet.value);
            String name = packet.name;
            Q q = getQ(name);
            doRemoveBlock(q, packet.conn.getEndPoint(), addRemove.removeBlockId);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    public void doRemoveBlock(Q q, Address originalRemover, int blockId) {
        Block blockRemoved = q.removeBlock(blockId);
        if (blockRemoved != null) {
            if (isMaster()) {
                sendAddBlockMessageToOthers(blockRemoved, -1, originalRemover, false);
            }
        }
    }

    final void handleFullBlock(PacketQueue.Packet packet) {
        if (isMaster()) {
            String name = packet.name;
            Q q = getQ(name);
            doFullBlock(q, packet.blockId, packet.conn.getEndPoint());
        }
        packet.returnToContainer();
    }

    final void doFullBlock(Q q, int fullBlockId, Address originalFuller) {
        int blockId = q.getLatestAddedBlock() + 1;
        Address target = nextTarget();
        Block newBlock = setFullAndCreateNewBlock(q, fullBlockId, target, blockId);
        if (newBlock != null) {
            sendAddBlockMessageToOthers(newBlock, fullBlockId, originalFuller, true);
        }
    }

    final void handleAddBlock(PacketQueue.Packet packet) {
        try {
            BlockUpdate addRemove = (BlockUpdate) ThreadContext.get().toObject(packet.value);
            String name = packet.name;
            int blockId = addRemove.addBlockId;
            Address addressOwner = addRemove.addAddress;
            Q q = getQ(name);
            setFullAndCreateNewBlock(q, addRemove.fullBlockId, addressOwner, blockId);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            packet.returnToContainer();
        }
    }

    Block setFullAndCreateNewBlock(Q q, int fullBlockId, Address newBlockOwner, int newBlockId) {
        List<Block> lsBlocks = q.lsBlocks;
        boolean exist = false;
        for (Block block : lsBlocks) {
            if (block.blockId == newBlockId) {
                exist = true;
            }
        }
        Block newBlock = null;
        if (!exist) {
            newBlock = q.createBlock(newBlockOwner, newBlockId);
            q.addBlock(newBlock);
        }
        if (fullBlockId != -1) {
            q.setBlockFull(fullBlockId);
        }
        return newBlock;
    }

    final void handleOffer(PacketQueue.Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doOffer(remoteReq);
            packet.longValue = remoteReq.longValue;
            if (!remoteReq.scheduled) {
                if (remoteReq.response == Boolean.TRUE) {
                    sendResponse(packet);
                } else {
                    sendResponseFailure(packet);
                }
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handlePublish(Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doPublish(remoteReq);
            packet.longValue = remoteReq.longValue;
            if (!remoteReq.scheduled) {
                if (remoteReq.response == Boolean.TRUE) {
                    sendResponse(packet);
                } else {
                    sendResponseFailure(packet);
                }
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handleAddTopicListener(PacketQueue.Packet packet) {
        if (rightRemoteOfferTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doAddTopicListener(remoteReq);
            packet.longValue = remoteReq.longValue;
            sendResponse(packet);
            remoteReq.reset();
        }
    }

    final boolean rightRemoteOfferTarget(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        if (q.blCurrentPut == null) {
            q.setCurrentPut();
        }
        boolean invalid = false;
        if (packet.blockId != q.blCurrentPut.blockId) {
            if (packet.blockId > q.blCurrentPut.blockId) {
                int size = q.lsBlocks.size();
                for (int i = 0; i < size; i++) {
                    Block block = q.lsBlocks.get(i);
                    if (block.blockId == packet.blockId) {
                        if (thisAddress.equals(block.address)) {
                            q.blCurrentPut = block;
                        }
                    }
                }
            } else {
                invalid = true;
            }
        }
        if (q.blCurrentPut.isFull() || !thisAddress.equals(q.blCurrentPut.address)) {
            invalid = true;
        }
        if (invalid) {
            sendRedoResponse(packet);
        }
        return (!invalid);
    }

    boolean rightRemotePollTarget(Packet packet) {
        Q q = getQ(packet.name);
        if (q.blCurrentTake == null) {
            q.setCurrentTake();
        }
        boolean invalid = false;
        if (packet.blockId != q.blCurrentTake.blockId) {
            if (packet.blockId > q.blCurrentTake.blockId) {
                int size = q.lsBlocks.size();
                for (int i = 0; i < size; i++) {
                    Block block = q.lsBlocks.get(i);
                    if (block.blockId == packet.blockId) {
                        if (thisAddress.equals(block.address)) {
                            q.blCurrentTake = block;
                        }
                    }
                }
            } else {
                invalid = true;
            }
        }
        if ((q.blCurrentTake.size() == 0 && q.blCurrentTake.isFull())
                || !thisAddress.equals(q.blCurrentTake.address)) {
            invalid = true;
        }
        if (invalid) {
            sendRedoResponse(packet);
        }
        return (!invalid);
    }

    final void handlePoll(PacketQueue.Packet packet) {
        if (rightRemotePollTarget(packet)) {
            remoteReq.setFromPacket(packet);
            doPoll(remoteReq);
            if (!remoteReq.scheduled) {
                Data oldValue = (Data) remoteReq.response;
                if (oldValue != null && oldValue.size() > 0) {
                    doSet(oldValue, packet.value);
                }
                sendResponse(packet);
            } else {
                packet.returnToContainer();
            }
            remoteReq.reset();
        }
    }

    final void handleRead(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        q.read(packet);
    }

    final void handleRemove(PacketQueue.Packet packet) {
        Q q = getQ(packet.name);
        q.remove(packet);
    }

    final Address nextTarget() {
        int size = lsMembers.size();
        int index = nextIndex++ % size;
        if (nextIndex >= size) {
            nextIndex = 0;
        }
        return lsMembers.get(index).getAddress();
    }

    public Q getQ(String name) {
        if (name == null)
            return null;
        Q q = mapQueues.get(name);
        if (q == null) {
            q = new Q(name);
            mapQueues.put(name, q);
        }
        return q;
    }

    public class CommitPoll extends AbstractCall {
        volatile long txnId = -1;
        String name = null;

        public int getOperation() {
            return OP_B_TXN_COMMIT;
        }

        public void commitPoll(String name) {
            this.name = name;
            this.txnId = ThreadContext.get().getTxnId();
            enqueueAndReturn(this);
        }

        public void handleResponse(PacketQueue.Packet packet) {
        }

        public void process() {
            MemberImpl nextMember = getNextMemberAfter(thisAddress, true, 1);
            if (nextMember != null) {
                Address next = nextMember.getAddress();
                PacketQueue.Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.txnId = txnId;
                boolean sent = send(packet, next);
                if (!sent) {
                    packet.returnToContainer();
                }
            }
        }
    }

    class QDestroy {
        public void destroy(String name) {
            sendProcessableToAll(new Destroy(name), true);
        }
    }

    public void destroy(String name) {
        mapQueues.remove(name);
    }

    public class Size extends AbstractCall {
        String name = null;
        int total = 0;
        int numberOfResponses = 0;
        int numberOfExpectedResponses = 1;

        public int getSize(String name) {
            this.name = name;
            synchronized (this) {
                try {
                    enqueueAndReturn(this);
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return total;
        }

        @Override
        public void onDisconnect(Address dead) {
            redo();
        }

        public void reset() {
            total = 0;
            numberOfResponses = 0;
            numberOfExpectedResponses = 0;
        }

        public void handleResponse(PacketQueue.Packet packet) {
            total += (int) packet.longValue;
            packet.returnToContainer();
            numberOfResponses++;
            if (numberOfResponses >= numberOfExpectedResponses) {
                removeCall(getId());
                synchronized (this) {
                    notify();
                }
            }
        }

        public void process() {
            reset();
            Q q = getQ(name);
            if (DEBUG) {
                q.printStack();
            }
            total += q.size;
            numberOfResponses = 1;
            if (lsMembers.size() > 1) {
                addCall(this);
                numberOfExpectedResponses = lsMembers.size();
                for (MemberImpl member : lsMembers) {
                    if (!member.localMember()) {
                        PacketQueue.Packet packet = obtainPacket();
                        packet.name = name;
                        packet.operation = getOperation();
                        packet.callId = getId();
                        packet.timeout = 0;
                        boolean sent = send(packet, member.getAddress());
                        if (!sent) {
                            packet.returnToContainer();
                            redo();
                        }
                    }
                }
            } else {
                synchronized (this) {
                    this.notify();
                }
            }
        }

        public int getOperation() {
            return OP_B_SIZE;
        }
    }

    class QIterator<E> implements Iterator<E>, Processable {

        volatile String name;
        volatile List<Integer> blocks = null;
        int currentBlockId = -1;
        int currentIndex = -1;
        Object next = null;
        boolean hasNextCalled = false;

        public void set(String name) {
            synchronized (QIterator.this) {
                this.name = name;
                enqueueAndReturn(QIterator.this);
                try {
                    QIterator.this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        public void process() {
            Q q = getQ(name);
            List<Block> lsBlocks = q.lsBlocks;
            blocks = new ArrayList<Integer>(lsBlocks.size());
            for (Block block : lsBlocks) {
                blocks.add(block.blockId);
            }
            synchronized (QIterator.this) {
                QIterator.this.notify();
            }
        }

        public boolean hasNext() {
            next = null;
            while (next == null) {
                boolean canRead = setNextBlock();
                // System.out.println(currentBlockId + " Can read " + canRead);
                if (!canRead)
                    return false;
                Read read = new Read();                
                next = read.read(name, currentBlockId, currentIndex);
                // System.out.println(currentIndex + " Next " + next);
                if (next == null) {
                    currentIndex = -1;
                } else {
                    currentIndex = read.readIndex;
                    currentIndex++;
                }
            }
            hasNextCalled = true;
            return true;
        }

        boolean setNextBlock() {
            if (currentIndex == -1 || currentIndex >= BLOCK_SIZE) {
                if (blocks.size() == 0) {
                    return false;
                } else {
                    currentBlockId = blocks.remove(0);
                    currentIndex = 0;
                    return true;
                }
            }
            return true;
        }

        public E next() {
            if (!hasNextCalled) {
                boolean hasNext = hasNext();
                if (!hasNext) {
                    return null;
                }
            }
            if (next != null) {
                hasNextCalled = false;
            }
            return (E) next;
        }

        public void remove() {
        }

    }

    public class Remove extends QueueBasedCall {
        String name;
        int blockId;
        int index;

        public Object remove(String name, int blockId, int index) {
            this.name = name;
            this.blockId = blockId;
            this.index = index;
            enqueueAndReturn(this);
            Object result;
            try {
                result = responses.take();
                if (result == OBJECT_NULL)
                    return null;
                else {
                    return ThreadContext.get().toObject((Data) result);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        void setResponse() {
            responses.add(OBJECT_NULL);
        }

        public void handleResponse(PacketQueue.Packet packet) {
            setResponse();
            packet.returnToContainer();
        }

        @Override
        public void onDisconnect(Address dead) {
            process();
        }

        public void process() {
            addCall(this);
            Q q = getQ(name);

            Address target = q.getBlockOwner(blockId);
            if (target == null) {
                responses.add(OBJECT_NULL);
            } else if (target.equals(thisAddress)) {
                q.remove(this);
            } else {
                Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.callId = getId();
                packet.blockId = blockId;
                packet.timeout = 0;
                packet.longValue = index;
                boolean sent = false;
                if (target != null) {
                    sent = send(packet, target);
                }
                if (target == null || !sent) {
                    packet.returnToContainer();
                    responses.add(OBJECT_NULL);
                }
            }
        }

        public int getOperation() {
            return OP_B_REMOVE;
        }

    }

    public class Read extends QueueBasedCall {
        String name;
        int blockId;
        int index;
        int readIndex = 0;

        public Object read(String name, int blockId, int index) {
            this.name = name;
            this.blockId = blockId;
            this.index = index;
            enqueueAndReturn(this);
            Object result;
            try {
                result = responses.take();
                if (result == OBJECT_NULL)
                    return null;
                else {
                    return ThreadContext.get().toObject((Data) result);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        void setResponse(Data value, int index) {
            responses.clear();
            readIndex = index;
            responses.add((value == null) ? OBJECT_NULL : value);
        }

        public void handleResponse(PacketQueue.Packet packet) {
            Data value = doTake(packet.value);
            int indexRead = (int) packet.longValue;
            setResponse(value, indexRead);
            removeCall(getId());
            packet.returnToContainer();
        }

        @Override
        public void onDisconnect(Address dead) {
            removeCall(getId());
            enqueueAndReturn(Read.this);
        }

        public void process() {
            responses.clear();
            Q q = getQ(name);
            Address target = q.getBlockOwner(blockId);
            if (target == null) {
                responses.add(OBJECT_NULL);
            } else if (target.equals(thisAddress)) {
                q.read(this);
            } else {
                addCall(this);
                PacketQueue.Packet packet = obtainPacket();
                packet.name = name;
                packet.operation = getOperation();
                packet.callId = getId();
                packet.blockId = blockId;
                packet.timeout = 0;
                packet.longValue = index;
                boolean sent = send(packet, target);
                if (!sent) {
                    packet.returnToContainer();
                    onDisconnect(target);
                }
{'fraction_non_alphanumeric': 0.09187885534066138, 'fraction_numerical': 0.002164401774809455, 'mean_word_length': 1.3933103933103934, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5318365', 'n_tokens_mistral': 16679, 'n_tokens_neox': 15897, 'n_words': 3551}