import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * When a COMMIT message is received, eventually this method is called, 
 * which matches up the zxid from the COMMIT with (hopefully) the head of
 * the pendingTxns queue and hands it to the commitProcessor to commit.
 * @param zxid - must correspond to the head of pendingTxns if it exists
 */
public void commit(long zxid) {
    if (pendingTxns.size() == 0) {
        LOG.warn("Committing " + Long.toHexString(zxid)
                + " without seeing txn");
        return;
    }
    long firstElementZxid = pendingTxns.element().zxid;
    if (firstElementZxid != zxid) {
        LOG.error("Committing zxid 0x" + Long.toHexString(zxid)
                + " but next pending txn 0x"
                + Long.toHexString(firstElementZxid));
        System.exit(12);
    }
    Request request = pendingTxns.remove();
    commitProcessor.commit(request);
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void processRequest(Request si) {
    if(si.type != OpCode.sync){
        QuorumPacket qp = new QuorumPacket(Leader.ACK, si.hdr.getZxid(), null,
            null);
        try {
            learner.writePacket(qp, false);
        } catch (IOException e) {
            LOG.warn("Closing connection to leader, exception during packet send", e);
            try {
                if (!learner.sock.isClosed()) {
                    learner.sock.close();
                }
            } catch (IOException e1) {
                // Nothing to do, we are shutting things down, so an exception here is irrelevant
                LOG.debug("Ignoring error closing the connection", e1);
            }
        }
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void processRequest(Request request) {
    if (!finished) {
        // Before sending the request, check if the request requires a
        // global session and what we have is a local session. If so do
        // an upgrade.
        Request upgradeRequest = null;
        try {
            upgradeRequest = zks.checkUpgradeSession(request);
        } catch (KeeperException ke) {
            if (request.getHdr() != null) {
                request.getHdr().setType(OpCode.error);
                request.setTxn(new ErrorTxn(ke.code().intValue()));
            }
            request.setException(ke);
            LOG.info("Error creating upgrade request",  ke);
        } catch (IOException ie) {
            LOG.error("Unexpected error in upgrade", ie);
        }
        if (upgradeRequest != null) {
            queuedRequests.add(upgradeRequest);
        }
        queuedRequests.add(request);
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void processRequest(Request request) throws RequestProcessorException {
    next.processRequest(request);

    // The only requests that should be on toBeApplied are write
    // requests, for which we will have a hdr. We can't simply use
    // request.zxid here because that is set on read requests to equal
    // the zxid of the last write op.
    if (request.getHdr() != null) {
        long zxid = request.getHdr().getZxid();
        Iterator<Proposal> iter = leader.toBeApplied.iterator();
        if (iter.hasNext()) {
            Proposal p = iter.next();
            if (p.request != null && p.request.zxid == zxid) {
                iter.remove();
                return;
            }
        }
        LOG.error("Committed request not found on toBeApplied: "
                  + request);
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * When a COMMIT message is received, eventually this method is called,
 * which matches up the zxid from the COMMIT with (hopefully) the head of
 * the pendingTxns queue and hands it to the commitProcessor to commit.
 * @param zxid - must correspond to the head of pendingTxns if it exists
 */
public void commit(long zxid) {
    if (pendingTxns.size() == 0) {
        LOG.warn("Committing " + Long.toHexString(zxid)
                + " without seeing txn");
        return;
    }
    long firstElementZxid = pendingTxns.element().zxid;
    if (firstElementZxid != zxid) {
        LOG.error("Committing zxid 0x" + Long.toHexString(zxid)
                + " but next pending txn 0x"
                + Long.toHexString(firstElementZxid));
        System.exit(12);
    }
    Request request = pendingTxns.remove();
    commitProcessor.commit(request);
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * Simply queue the request, which will be processed in FIFO order.
 */
public void processRequest(Request request) {
    if (!finished) {
        Request upgradeRequest = null;
        try {
            upgradeRequest = zks.checkUpgradeSession(request);
        } catch (KeeperException ke) {
            if (request.getHdr() != null) {
                request.getHdr().setType(OpCode.error);
                request.setTxn(new ErrorTxn(ke.code().intValue()));
            }
            request.setException(ke);
            LOG.info("Error creating upgrade request",  ke);
        } catch (IOException ie) {
            LOG.error("Unexpected error in upgrade", ie);
        }
        if (upgradeRequest != null) {
            queuedRequests.add(upgradeRequest);
        }
        queuedRequests.add(request);
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * Requests coming from the learner should go directly to
 * PrepRequestProcessor
 *
 * @param request
 */
public void submitLearnerRequest(Request request) {
    /*
     * Requests coming from the learner should have gone through
     * submitRequest() on each server which already perform some request
     * validation, so we don't need to do it again.
     *
     * Additionally, LearnerHandler should start submitting requests into
     * the leader's pipeline only when the leader's server is started, so we
     * can submit the request directly into PrepRequestProcessor.
     *
     * This is done so that requests from learners won't go through
     * LeaderRequestProcessor which perform local session upgrade.
     */
    prepRequestProcessor.processRequest(request);
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
protected boolean needCommit(Request request) {
    switch (request.type) {
        case OpCode.create:
        case OpCode.create2:
        case OpCode.createTTL:
        case OpCode.createContainer:
        case OpCode.delete:
        case OpCode.deleteContainer:
        case OpCode.setData:
        case OpCode.reconfig:
        case OpCode.multi:
        case OpCode.setACL:
            return true;
        case OpCode.sync:
            return matchSyncs;    
        case OpCode.createSession:
        case OpCode.closeSession:
            return !request.isLocalSession();
        default:
            return false;
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
@Override
public void processRequest(Request request)
        throws RequestProcessorException {
    // Check if this is a local session and we are trying to create
    // an ephemeral node, in which case we upgrade the session
    Request upgradeRequest = null;
    try {
        upgradeRequest = lzks.checkUpgradeSession(request);
    } catch (KeeperException ke) {
        if (request.getHdr() != null) {
            LOG.debug("Updating header");
            request.getHdr().setType(OpCode.error);
            request.setTxn(new ErrorTxn(ke.code().intValue()));
        }
        request.setException(ke);
        LOG.info("Error creating upgrade request " + ke.getMessage());
    } catch (IOException ie) {
        LOG.error("Unexpected error in upgrade", ie);
    }
    if (upgradeRequest != null) {
        nextProcessor.processRequest(upgradeRequest);
    }

    nextProcessor.processRequest(request);
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
@Override
protected void setLocalSessionFlag(Request si) {
    // We need to set isLocalSession to tree for these type of request
    // so that the request processor can process them correctly.
    switch (si.type) {
    case OpCode.createSession:
        if (self.areLocalSessionsEnabled()) {
            // All new sessions local by default.
            si.setLocalSession(true);
        }
        break;
    case OpCode.closeSession:
        String reqType = "global";
        if (upgradeableSessionTracker.isLocalSession(si.sessionId)) {
            si.setLocalSession(true);
            reqType = "local";
        }
        LOG.info("Submitting " + reqType + " closeSession request"
                + " for session 0x" + Long.toHexString(si.sessionId));
        break;
    default:
        break;
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
@Override
public void run() {
    Random rand = new Random(Thread.currentThread().getId());
    try {
        while(true) {
            // If it is a read-only test, there will be no proposals..
            if (!proposals.isEmpty()){
                Request request = proposals.take();
                Thread.sleep(5 + rand.nextInt(95));
                commitProcessor.commit(request);
            }
        }
    } catch (InterruptedException e) {
        // ignore
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * Validate that this is the only request in the pipeline
 */
private void validateWriteRequestVariant(Request request) {
    if (stopped)
        return;
    long zxid = request.getHdr().getZxid();
    int readRequests = outstandingReadRequests.get();
    if (readRequests != 0) {
        failTest("There are " + readRequests + " outstanding"
                + " read requests while issuing a write request zxid="
                + zxid);
    }
    int writeRequests = outstandingWriteRequests.get();
    if (writeRequests > 1) {
        failTest("There are " + writeRequests + " outstanding"
                + " write requests while issuing a write request zxid="
                + zxid + " (expected one)");
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
MockCommitProcessor() {
    super(new RequestProcessor() {
        public void processRequest(Request request)
                throws RequestProcessorException {
            processedRequests.offer(request);
        }

        public void shutdown() {
        }
    }, "0", false, new ZooKeeperServerListener() {

        @Override
        public void notifyStopping(String threadName, int errorCode) {
            Assert.fail("Commit processor crashed " + errorCode);
        }
    });
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * In the following test, we verify that committed requests are processed
 * even when queuedRequests never gets empty. We add 10 committed request
 * and use infinite queuedRequests. We verify that the committed request was
 * processed.
 */
@Test(timeout = 1000)
public void noStarvationOfNonLocalCommittedRequestsTest() throws Exception {
    final String path = "/noStarvationOfCommittedRequests";
    processor.queuedRequests = new MockRequestsQueue();
    Set<Request> nonLocalCommits = new HashSet<Request>();
    for (int i = 0; i < 10; i++) {
        Request nonLocalCommitReq = newRequest(
                new CreateRequest(path, new byte[0], Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT_SEQUENTIAL.toFlag()),
                OpCode.create, 51, i + 1);
        processor.committedRequests.add(nonLocalCommitReq);
        nonLocalCommits.add(nonLocalCommitReq);
    }
    for (int i = 0; i < 10; i++) {
        processor.initThreads(defaultSizeOfThreadPool);
        processor.stoppedMainLoop = true;
        processor.run();
    }
    Assert.assertTrue("commit request was not processed",
            processedRequests.containsAll(nonLocalCommits));
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
@Test
public void testGetTxnLogSyncElapsedTime() throws IOException {
    File tmpDir = ClientBase.createEmptyTestDir();
    FileTxnSnapLog fileTxnSnapLog = new FileTxnSnapLog(new File(tmpDir, "data"),
            new File(tmpDir, "data_txnlog"));

    TxnHeader hdr = new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.setData);
    Record txn = new SetDataTxn("/foo", new byte[0], 1);
    Request req = new Request(0, 0, 0, hdr, txn, 0);

    try {
        fileTxnSnapLog.append(req);
        fileTxnSnapLog.commit();
        long syncElapsedTime = fileTxnSnapLog.getTxnLogElapsedSyncTime();
        Assert.assertNotEquals("Did not update syncElapsedTime!", -1L, syncElapsedTime);
    } finally {
        fileTxnSnapLog.close();
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void shutdown() {
    LOG.info("Shutting down");
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
@Override
public void shutdown() {
    finished = true;
    queuedRequests.clear();
    queuedRequests.add(Request.requestOfDeath);
    nextProcessor.shutdown();
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void processRequest(Request request) throws RequestProcessorException {
    // request.addRQRec(">tobe");
    next.processRequest(request);
    Proposal p = toBeApplied.peek();
    if (p != null && p.request != null
            && p.request.zxid == request.zxid) {
        toBeApplied.remove();
    }
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
/**
 * @param port
 * @param dataDir
 * @throws IOException
 */
FollowerZooKeeperServer(FileTxnSnapLog logFactory,QuorumPeer self,
        DataTreeBuilder treeBuilder, ZKDatabase zkDb) throws IOException {
    super(logFactory, self.tickTime, self.minSessionTimeout,
            self.maxSessionTimeout, treeBuilder, zkDb, self);
    this.pendingSyncs = new ConcurrentLinkedQueue<Request>();
}
 

import org.apache.zookeeper.server.Request; //导入依赖的package包/类
public void logRequest(TxnHeader hdr, Record txn) {
    Request request = new Request(null, hdr.getClientId(), hdr.getCxid(),
            hdr.getType(), null, null);
    request.hdr = hdr;
    request.txn = txn;
    request.zxid = hdr.getZxid();
    if ((request.zxid & 0xffffffffL) != 0) {
        pendingTxns.add(request);
    }
    syncProcessor.processRequest(request);
}