zookeeper原始碼決議--請求處理--ProposalRequestProcessor-有解無憂

ProposalRequestProcessor

SyncRequestProcessor
AckRequestProcessor

// 如果請求是LearnerSyncRequest實體物件
if (request instanceof LearnerSyncRequest) 
{
	// 主節點執行processSync
    zks.getLeader().processSync((LearnerSyncRequest) request);
    	// 沒有等待完成的提議
    	if (outstandingProposals.isEmpty()) 
        {
            sendSync(r);
            	QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
    			r.fh.queuePacket(qp);
        } 
        else 
        {
        	// 尚有等待完成的提議
        	// 最后的提議，阻塞等待的同步請求集合
            pendingSyncs.computeIfAbsent(lastProposed, k -> new ArrayList<>()).add(r);
        }
} 
else 
{
	/*
	// 取決于環境變數
	if (!forwardLearnerRequestsToCommitProcessorDisabled) 
    {
        return true;
    }
    // 請求來自從節點&觀察者
    if (request.getOwner() instanceof LearnerHandler) 
    {
        ServerMetrics.getMetrics()
        	.REQUESTS_NOT_FORWARDED_TO_COMMIT_PROCESSOR.add(1);
        return false;
    }
    // 請求來自直連主節點的外部用戶
    return true;
	*/
    if (shouldForwardToNextProcessor(request)) 
    {
    	// 對于直接用戶，直接交由下一階段處理
        nextProcessor.processRequest(request);
    }
    
    // 請求包含事務頭--修改類請求
    if (request.getHdr() != null) 
    {
        try 
        {
        	// 通過主節點進行提議
            zks.getLeader().propose(request);
        }
        catch (XidRolloverException e) 
 		{
    		throw new RequestProcessorException(e.getMessage(), e);
		}
               	
        // 提議后，進行請求同步
        syncProcessor.processRequest(request);
    }
}

提議

if (request.isThrottled()) 
{
    LOG.error("Throttled request send as proposal: {}. Exiting.", request);
    ServiceUtils.requestSystemExit(ExitCode.UNEXPECTED_ERROR.getValue());
}

if ((request.zxid & 0xffffffffL) == 0xffffffffL) 
{
    String msg = 
    	"zxid lower 32 bits have rolled over, forcing re-election, and therefore new epoch start";
    shutdown(msg);
    throw new XidRolloverException(msg);
}

// 請求物件序列化
byte[] data = SerializeUtils.serializeRequest(request);
// 提議狀態資訊
proposalStats.setLastBufferSize(data.length);
// 資料包類別：Leader.PROPOSAL
// 資料包zxid：
// 資料包資料：
QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
// 提議物件
Proposal p = new Proposal();
p.packet = pp;
p.request = request;
synchronized (this) 
{
	// 提議物件還包含自身集群資訊
    p.addQuorumVerifier(self.getQuorumVerifier());
    if (request.getHdr().getType() == OpCode.reconfig) 
    {
        self.setLastSeenQuorumVerifier(request.qv, true);
    }

    if (self.getQuorumVerifier().getVersion() < self.getLastSeenQuorumVerifier().getVersion()) 
    {
        p.addQuorumVerifier(self.getLastSeenQuorumVerifier());
    }

    LOG.debug("Proposing:: {}", request);
    // 更新最后提議物件
    lastProposed = p.packet.getZxid();
    // 等待完成提議中新增lastProposed--提議物件
    outstandingProposals.put(lastProposed, p);
    // 向所有從節點/觀察者發送提議
    sendPacket(pp);
}

	ServerMetrics.getMetrics().PROPOSAL_COUNT.add(1);
	return p;
}

SyncRequestProcessor

resetSnapshotStats();
	// 隨機值
	randRoll = ThreadLocalRandom.current().nextInt(snapCount / 2);
	// 不超過snapSizeInBytes/2的隨機值
    randSize = Math.abs(ThreadLocalRandom.current().nextLong() % (snapSizeInBytes / 2));
// 記錄時間
lastFlushTime = Time.currentElapsedTime();
while (true) 
{
    ServerMetrics.getMetrics().SYNC_PROCESSOR_QUEUE_SIZE.add(queuedRequests.size());
    // 超時時間
    long pollTime = Math.min(zks.getMaxWriteQueuePollTime(), getRemainingDelay());
    // 取出請求
    Request si = queuedRequests.poll(pollTime, TimeUnit.MILLISECONDS);
    if (si == null) 
    {
        flush();
        si = queuedRequests.take();
    }

    if (si == REQUEST_OF_DEATH) 
    {
        break;
    }
	
	// 當前時間
    long startProcessTime = Time.currentElapsedTime();
    ServerMetrics.
    	getMetrics().SYNC_PROCESSOR_QUEUE_TIME.add(startProcessTime - si.syncQueueStartTime);
    // 資料物體執行append
    if (!si.isThrottled() && zks.getZKDatabase().append(si)) 
    {
    	// 如果事務日志累計到此刻，適合對資料物體執行快照
        if (shouldSnapshot()) 
        {
        	// 重置快照狀態
            resetSnapshotStats();
            // 將當前事務日志檔案進行緩沖重繪到磁盤操作
            zks.getZKDatabase().rollLog();
            if (!snapThreadMutex.tryAcquire()) 
            {
                LOG.warn("Too busy to snap, skipping");
            } 
            else 
            {
                new ZooKeeperThread("Snapshot Thread") 
                {
                    public void run() 
                    {
                        try 
                        {
                        	// 執行快照生產，程序受到snapThreadMutex保護
                            zks.takeSnapshot();
                        } 
                        catch (Exception e) 
                        {
                            LOG.warn("Unexpected exception", e);
                        } 
                        finally 
                        {
                            snapThreadMutex.release();
                        }
                    }
                }.start();
            }
        }
    } 
    // 沒有事務請求等待重繪到磁盤，且本請求遭遇過濾
    else if (toFlush.isEmpty()) 
    {
        if (nextProcessor != null) 
        {
            nextProcessor.processRequest(si);
            if (nextProcessor instanceof Flushable) 
            {
                ((Flushable) nextProcessor).flush();
            }
        }

        continue;
    }
    
    // 這個請求屬于尚未重繪到磁盤的事務請求
    toFlush.add(si);
    // 如果此時應該重繪，重繪事務日志檔案緩沖到磁盤
    if (shouldFlush()) 
    {
        flush();
    }
    
    ServerMetrics.getMetrics().
    	SYNC_PROCESS_TIME.add(Time.currentElapsedTime() - startProcessTime);
}

append

txnCount.incrementAndGet();
return this.snapLog.append(si);
	txnLog.append(si.getHdr(), si.getTxn(), si.getTxnDigest());
		// 只有寫請求有事務日志
		if (hdr == null) 
        {
            return false;
        }
        
        if (hdr.getZxid() <= lastZxidSeen) 
        {
            LOG.warn(
            	"Current zxid {} is <= {} for {}", hdr.getZxid(), lastZxidSeen, Request.op2String(hdr.getType()));
        } 
        else 
        {
            lastZxidSeen = hdr.getZxid();
        }

        if (logStream == null) 
        {
            LOG.info("Creating new log file: {}", Util.makeLogName(hdr.getZxid()));
            // 新的事務日志檔案
            logFileWrite = new File(logDir, Util.makeLogName(hdr.getZxid()));
            fos = new FileOutputStream(logFileWrite);
            logStream = new BufferedOutputStream(fos);
            oa = BinaryOutputArchive.getArchive(logStream);
        	
        	// 新檔案檔案頭
            FileHeader fhdr = new FileHeader(TXNLOG_MAGIC, VERSION, dbId);
            fhdr.serialize(oa, "fileheader");
            logStream.flush();
            // 設定好檔案大小
            filePadding.setCurrentSize(fos.getChannel().position());
            // 當前事務日志檔案，寫入內容不保證立即重繪到磁盤
            streamsToFlush.add(fos);
        }

        filePadding.padFile(fos.getChannel());
        // 序列化
        byte[] buf = Util.marshallTxnEntry(hdr, txn, digest);
        if (buf == null || buf.length == 0) 
        {
            throw new IOException("Faulty serialization for header " + "and txn");
        }
        // 寫入crc
        Checksum crc = makeChecksumAlgorithm();
        crc.update(buf, 0, buf.length);
        oa.writeLong(crc.getValue(), "txnEntryCRC");
        // 寫入事務本身
        Util.writeTxnBytes(oa, buf);
        return true;

commit

this.snapLog.commit();
	txnLog.commit();
		// 如果當前事務日志檔案存在，重繪 檔案緩沖到磁盤
		if (logStream != null) 
		{
		       logStream.flush();
		}
		   
		// 對需要重繪的每個物件
		for (FileOutputStream log : streamsToFlush) 
		{
		   	   // 重繪緩沖到磁盤
		       log.flush();
		       // 取決于環境變數
		       if (forceSync) 
		       {
		       	   // 當前時間
		           long startSyncNS = System.nanoTime();
		           FileChannel channel = log.getChannel();
		           // 強制重繪檔案內容
		           channel.force(false);
		           syncElapsedMS 
		           	= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS);
		           if (syncElapsedMS > fsyncWarningThresholdMS) 
		           {
		               if (serverStats != null) 
		               {
		                   serverStats.incrementFsyncThresholdExceedCount();
		               }
		
		               LOG.warn("fsync-ing the write ahead log in {} took {}ms 
		               	which will adversely effect operation latency." + "File size is {} bytes. 
		               	See the ZooKeeper troubleshooting guide", Thread.currentThread().getName(), 
		               	syncElapsedMS, channel.size());
		           }
		
		           ServerMetrics.getMetrics().FSYNC_TIME.add(syncElapsedMS);
		       }
		}
		
		// 重繪后事務日志檔案進行關閉
		while (streamsToFlush.size() > 1) 
		{
		       streamsToFlush.poll().close();
		}
		
		// 當前事務日志檔案緩沖重繪到磁盤
		if (txnLogSizeLimit > 0) 
		{
		       long logSize = getCurrentLogSize();
		       if (logSize > txnLogSizeLimit) 
		       {
		           LOG.debug("Log size limit reached: {}", logSize);
		           rollLog();
		       }
		}

flush

// 沒有等待要重繪到磁盤的事務請求--不處理
if (this.toFlush.isEmpty()) 
{
    return;
}

ServerMetrics.getMetrics().BATCH_SIZE.add(toFlush.size());
// 當前時間
long flushStartTime = Time.currentElapsedTime();
// 重繪緩沖到磁盤
zks.getZKDatabase().commit();
ServerMetrics.getMetrics().
	SYNC_PROCESSOR_FLUSH_TIME.add(Time.currentElapsedTime() - flushStartTime);
// 一般下一級為Ack
if (this.nextProcessor == null) 
{
    this.toFlush.clear();
} 
else 
{
	// 由于積壓的事務請求被順序寫入了磁盤
	// 所以積壓的事務請求現在可以順序投遞給下一級處理【持久性】
    while (!this.toFlush.isEmpty()) 
    {
        final Request i = this.toFlush.remove();
        long latency = Time.currentElapsedTime() - i.syncQueueStartTime;
        ServerMetrics.getMetrics().SYNC_PROCESSOR_QUEUE_AND_FLUSH_TIME.add(latency);
        // 事務請求被寫入磁盤后，才投遞給下一級繼續處理
        this.nextProcessor.processRequest(i);
    }
    
    // 下一級支持flush的話，也flush
    if (this.nextProcessor instanceof Flushable) 
    {
        ((Flushable) this.nextProcessor).flush();
    }
}

lastFlushTime = Time.currentElapsedTime();

shouldFlush

// 取決于時間&積壓待重繪事務請求數量
long flushDelay = zks.getFlushDelay();
long maxBatchSize = zks.getMaxBatchSize();
if ((flushDelay > 0) && (getRemainingDelay() == 0)) 
{
    return true;
}

return (maxBatchSize > 0) && (toFlush.size() >= maxBatchSize);

AckRequestProcessor

QuorumPeer self = leader.self;
if (self != null) 
 {
     request.logLatency(ServerMetrics.getMetrics().PROPOSAL_ACK_CREATION_LATENCY);
     // leader--受到來自主節點，關于zxid的ack
     leader.processAck(self.getId(), request.zxid, null);
 } 
 else 
 {
     LOG.error("Null QuorumPeer");
 }

processAck

if (!allowedToCommit) 
{
    return; 
}

if (LOG.isTraceEnabled()) 
{
    LOG.trace("Ack zxid: 0x{}", Long.toHexString(zxid));
    for (Proposal p : outstandingProposals.values()) 
    {
        long packetZxid = p.packet.getZxid();
        LOG.trace("outstanding proposal: 0x{}", Long.toHexString(packetZxid));
    }
    
    LOG.trace("outstanding proposals all");
}

// 主節點首個包
if ((zxid & 0xffffffffL) == 0) 
{
    return;
}

if (outstandingProposals.size() == 0) 
{
    LOG.debug("outstanding is 0");
    return;
}

if (lastCommitted >= zxid) 
{
    LOG.debug("proposal has already been committed, pzxid: 0x{} zxid: 0x{}", 
    	Long.toHexString(lastCommitted), Long.toHexString(zxid));
    return;
}
// 取得此zxid關聯的待處理提議物件
Proposal p = outstandingProposals.get(zxid);
if (p == null) 
{
    LOG.warn("Trying to commit future proposal: zxid 0x{} from {}", 
    	Long.toHexString(zxid), followerAddr);
    return;
}

if (ackLoggingFrequency > 0 && (zxid % ackLoggingFrequency == 0)) 
{
    p.request.logLatency(ServerMetrics.getMetrics().ACK_LATENCY, Long.toString(sid));
}

// 提議物件中收集對此提議進行ack的成員
p.addAck(sid);
// 嘗試提交
boolean hasCommitted = tryToCommit(p, zxid, followerAddr);
// 重新配置暫不考慮
if (hasCommitted && p.request != null && p.request.getHdr().getType() == OpCode.reconfig) 
{
    long curZxid = zxid;
    while (allowedToCommit && hasCommitted && p != null) 
    {
        curZxid++;
        p = outstandingProposals.get(curZxid);
        if (p != null) 
        {
            hasCommitted = tryToCommit(p, curZxid, null);
        }
    }
}

tryToCommit

// 上一zxid還未提交，處于順序性要求本zxid也不可提交
if (outstandingProposals.containsKey(zxid - 1)) 
{
    return false;
}

// 提議未收到過半投票成員ack時，不可提交
if (!p.hasAllQuorums()) 
{
    return false;
}

if (zxid != lastCommitted + 1) 
{
    LOG.warn("Commiting zxid 0x{} from {} not first!", Long.toHexString(zxid), followerAddr);
    LOG.warn("First is 0x{}", Long.toHexString(lastCommitted + 1));
}

// 從待處理提議集合中移除zxid關聯的
outstandingProposals.remove(zxid);
if (p.request != null) 
{
    // 將提議放到toBeApplied容器
    toBeApplied.add(p);
}

if (p.request == null) 
{
    LOG.warn("Going to commit null: {}", p);
} 
// 重新配置暫不考慮
else if (p.request.getHdr().getType() == OpCode.reconfig) 
{
    LOG.debug("Committing a reconfiguration! {}", outstandingProposals.size());
    Long designatedLeader = getDesignatedLeader(p, zxid);
    QuorumVerifier newQV = p.qvAcksetPairs.get(p.qvAcksetPairs.size() - 1).getQuorumVerifier();
    self.processReconfig(newQV, designatedLeader, zk.getZxid(), true);
    if (designatedLeader != self.getId()) 
    {
        LOG.info(String.format("Committing a reconfiguration (reconfigEnabled=%s); 
        	this leader is not the designated " + "leader anymore, setting allowedToCommit=false", 
        	self.isReconfigEnabled()));
        allowedToCommit = false;
    }

    commitAndActivate(zxid, designatedLeader);
    informAndActivate(p, designatedLeader);
} 
else 
{
    p.request.logLatency(ServerMetrics.getMetrics().QUORUM_ACK_LATENCY);
    commit(zxid);
    	synchronized (this) 
        {
            lastCommitted = zxid;
        }
        
        QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
        // 給每個從節點，觀察者發送型別為Leader.COMMIT，zxid為zxid的集群包
        sendPacket(qp);
        ServerMetrics.getMetrics().COMMIT_COUNT.add(1);
    inform(p);
    	// 用于給觀察者發送提交通知
    	QuorumPacket qp = 
    		new QuorumPacket(Leader.INFORM, proposal.request.zxid, proposal.packet.getData(), null);
    	sendObserverPacket(qp);
}

// 主節點對每個寫入請求都需要向自己&各個參與者發送提議
// 收集過半自己&參與者的ack后，
// 自己&通知參與者進入請求提交
// 向commitProcessor加入請求
// commitProcessor負責對加入其中的請求逐個執行請求提交操作
zk.commitProcessor.commit(p.request);
// 如果有人之前發送了針對此zxid的同步請求
if (pendingSyncs.containsKey(zxid)) 
{
    for (LearnerSyncRequest r : pendingSyncs.remove(zxid)) 
    {
        // 因為此zxid此刻進入提交處理了，所以向等待在此zxid的同步者發送SYNC
        sendSync(r);
        	QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
    		r.fh.queuePacket(qp);
    }
}

return true;

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/420010.html

標籤：其他

上一篇：ModelArts + Gallery = 0基礎玩轉AI

下一篇：zookeeper原始碼決議--請求處理--Leader.ToBeAppliedRequestProcessor