जावा में वितरित KV संग्रहण के लिए RAFT सर्वसम्मति एल्गोरिथम लागू करना

सामान्य समाधान विवरण

• अनुयायी (अनुयायी)। ग्राहक से अनुरोधों को स्वीकार करता है। नेता से दिल की धड़कन लेता है
• उम्मीदवार (उम्मीदवार)। ग्राहक से अनुरोधों को स्वीकार करता है। अन्य नोड्स को वोट अनुरोध भेजता है
• नेता अनुरोधों को पढ़ना और लिखना स्वीकार करता है। अन्य नोड्स के लिए दिल की धड़कन अनुरोध भेजता है। अन्य नोड्स के लिए अनुरोध डेटा भेजता है।

आधार सामग्री भंडारण

public interface OperationsLog {
   void append(Operation operation);
   Operation get(Integer index);
   List<Operation> all();

   Long getTerm(Integer index);
   Integer getLastIndex();
   Long getLastTerm();

   void removeAllFromIndex(int newOperationIndex);
}

public interface Storage {
   List<Entry> all();
   String get(Long key);
   void insert(Long key, String val);
   void update(Long key, String val);
   void delete(Long key);
}

टाइमर

• मतदान जब मतदान का दौर चलता है
• एपेंड, उर्फ ​​दिल की धड़कन (यदि डेटा के बिना), अनुयायियों को लॉग डेटा को दोहराने और मतदान के एक नए दौर की शुरुआत को रोकने के लिए।

• वोट। मतदान का एक दौर शुरू करने के लिए। प्रत्येक नोड का अपना अंतराल होता है, जिसके बाद वह एक नया वोट शुरू करने का प्रयास करेगा। नेता से दिल की धड़कन प्राप्त करते समय उल्टी गिनती शुरू होती है।
• दिल की धड़कन। परिशिष्ट नेता द्वारा अनुयायियों को अनुरोध भेजने के लिए। यदि नोड को दिल की धड़कन नहीं मिलती है और मतदान टाइमर समाप्त हो गया है, तो यह एक उम्मीदवार बन जाता है और चुनाव शुरू करता है, मतदान के दौर की संख्या बढ़ाता है और अन्य नोड्स को मतदान अनुरोध भेजता है। यदि नोड बहुमत के वोटों को इकट्ठा करता है, तो यह नेता बन जाता है और दिल की धड़कन को बाहर भेजना शुरू कर देता है।

नोड की वर्तमान स्थिति

public interface Context {
   Integer getId(); //    
   State getState();//: , ,  
   Integer getVotedFor(); 
               //          
   Long getCurrentTerm(); //  
   Integer getCommitIndex(); //    
   List<Peer> getPeers(); //      
}

public interface Peer {
   Integer getId(); //  
   Integer getNextIndex(); //  ,    
   Integer getMatchIndex();//   
   Boolean getVoteGranted(); //     
}

मतदान

public interface ElectionService {
   void processElection();
   AnswerVoteDTO vote(RequestVoteDTO requestVoteDTO);
} 

मतदान के लिए अनुरोध भेजना

//1
context.setState(CANDIDATE); 
Long term = context.incCurrentTerm(); 
context.setVotedFor(context.getId()); 

List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());
long voteGrantedCount = 1L;
long voteRevokedCount = 0L;

//2
while (checkCurrentElectionStatus(term)) {
   List<AnswerVoteDTO> answers = getVoteFromAllPeers(term, peersIds);
   peersIds = new ArrayList<>();
   for (AnswerVoteDTO answer : answers) {
       //3
       if (answer.getStatusCode().equals(OK)) {
           //4
           if (answer.getTerm()>context.getCurrentTerm()) {
               context.setTermGreaterThenCurrent(answer.getTerm());
               return;
           }
           if (answer.isVoteGranted()) {
               //5 
               context.getPeer(answer.getId()).setVoteGranted(true);
               voteGrantedCount++;
           } else
               //6 
               voteRevokedCount++;
       } else {
          peersIds.add(answer.getId());
       }
   }
  //7
  if (voteGrantedCount >= context.getQuorum()) {
       winElection(term);
       return;
   } else if (voteRevokedCount >= context.getQuorum()) {
       loseElection(term);
       return;
   } 

1. "उम्मीदवार" की स्थिति निर्धारित करें। राउंड नंबर उठाएं और अपने लिए वोट करें।
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. नोड ने हमारे लिए मतदान किया! हम हमारे लिए वोट डालने वाले नोड्स की संख्या बढ़ाते हैं और तय करते हैं कि इस नोड ने हमारे लिए मतदान किया।
6. हमारे लिए वोट नहीं, हम भी मानते हैं।
7. यदि कोरम एकत्र किया जाता है और नोड चुनाव जीता है, तो हम "लीडर" की स्थिति स्थापित करते हैं। अन्यथा, हम अनुयायी बन जाते हैं और प्रतीक्षा करते हैं।

  private void winElection(Long term) {
       context.setState(LEADER);
       context.getPeers().forEach(peer ->
               peer.setNextIndex(operationsLog.getLastIndex()+1)

       );
   }

मतदान अनुरोध प्रसंस्करण

class RequestVoteDTO {
   private final Long term; //     
   private final Integer candidateId; //  
   private final Integer lastLogIndex; //     
   private final Long lastLogTerm; //       
}

public AnswerVoteDTO vote(RequestVoteDTO dto) {
   
       boolean termCheck;
       //1
       if (dto.getTerm() < context.getCurrentTerm())
           return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),false);
       else //2
       if (dto.getTerm().equals(context.getCurrentTerm())) {
           termCheck = (context.getVotedFor() == null||
                          context.getVotedFor().equals(dto.getCandidateId()));
       }
       else
       {   //3
           termCheck = true;
             context.setTermGreaterThenCurrent(dto.getTerm());
       }

       //4  
       boolean logCheck = !((operationsLog.getLastTerm() > dto.getLastLogTerm()) ||
               ((operationsLog.getLastTerm().equals(dto.getLastLogTerm())) &&
                       (operationsLog.getLastIndex() > dto.getLastLogIndex())));


       boolean voteGranted = termCheck&&logCheck;

       //5
       if (voteGranted) {
           context.setVotedFor(dto.getCandidateId());
       }
       //6   
       return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),voteGranted);
   }

1. यदि उम्मीदवार के दौर की तुलना में गाँठ का वर्तमान दौर बड़ा है, तो हम मना कर देते हैं, क्योंकि यह कुछ पिछड़ने वाली गाँठ का एक अनुरोध है, जो, जाहिर तौर पर, कुछ समय के लिए क्लस्टर के बाहर था और चुनाव प्रक्रिया शुरू कर दी क्योंकि यह वर्तमान नेता को नहीं देखता था।
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. एक सकारात्मक परिणाम के साथ, हम इस तथ्य को ठीक करते हैं कि नोड ने चुनाव में भाग लिया और उम्मीदवार के लिए वोट डाला।
6. परिणाम वापस उम्मीदवार को भेजें

प्रतिकृति

class RequestAppendDTO {
   private final Long term; //   
   private final Integer leaderId; //   

   private final Integer prevLogIndex;//   
   private final Long prevLogTerm;//   
   private final Integer leaderCommit;//      
   private final Operation operation; //
}

public interface ReplicationService {
   void appendRequest();
   AnswerAppendDTO append(RequestAppendDTO requestAppendDTO);
}

डेटा परिवर्तन अनुरोध सबमिट करें

private CompletableFuture<AnswerAppendDTO> sendAppendForOnePeer(Integer id) {
   return CompletableFuture.supplyAsync(() -> {
       try {
           //1
           Peer peer = context.getPeer(id);

           Operation operation;
           Integer prevIndex;
           //2    
           if (peer.getNextIndex() <= operationsLog.getLastIndex()) {
               operation = operationsLog.get(peer.getNextIndex());
               prevIndex = peer.getNextIndex() - 1;
           } else 
           //3  
           {
               operation = null;
               prevIndex = operationsLog.getLastIndex();
           }


           RequestAppendDTO requestAppendDTO = new RequestAppendDTO(
                   context.getCurrentTerm(), //   
                   context.getId(), //  
                   prevIndex,//      
                   operationsLog.getTerm(prevIndex),//  
                   context.getCommitIndex(),
                               //      
                   Operation //
           );

...
/*   http     */
}

1. अनुयायी मेटाडेटा
2. , . ( ), , , , . , , , ,
3. , , ; , , ,

public void appendRequest() {
       List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());

       //1 
       while (peersIds.size() > 0) {
           //2 
           List<AnswerAppendDTO> answers = sendAppendToAllPeers(peersIds);
           peersIds = new ArrayList<>();
           for (AnswerAppendDTO answer : answers) {
               //3
               if (answer.getStatusCode().equals(OK)) {
                   //4
                   if (answer.getTerm() > context.getCurrentTerm()) {
                        context.setTermGreaterThenCurrent(answer.getTerm());
                       return;
                   }
                   Peer peer = context.getPeer(answer.getId());
                   //5     
                   if (answer.getSuccess()) {                      
                       peer.setNextIndex(answer.getMatchIndex() + 1);
                       peer.setMatchIndex(answer.getMatchIndex());
                       if (peer.getNextIndex() <= operationsLog.getLastIndex())
                           peersIds.add(answer.getId());
                   //6      
                   } else {
                       peer.decNextIndex();
                       peersIds.add(answer.getId());
                   }
               }
           }
           //7
           tryToCommit();
       }
}

1. हम अनुरोध को दोहराते हैं जब तक कि हमें सभी अनुयायियों से प्रतिक्रिया नहीं मिल जाती है कि प्रतिकृति सफल रही। चूंकि एक ऑपरेशन प्रति अनुरोध भेजा जाता है, इसलिए अनुयायियों के लॉग को सिंक्रनाइज़ करने के लिए कई पुनरावृत्तियां हो सकती हैं
2. सभी अनुयायियों को अनुरोध भेजें और उत्तर के साथ एक सूची प्राप्त करें
3. हम केवल उपलब्ध अनुयायियों से जवाब पर विचार करते हैं
4. यदि यह पता चला है कि अनुयायियों में से एक का दौर नेता के दौर से अधिक है, तो हम सब कुछ रोक देते हैं और अनुयायी में बदल जाते हैं
5. यदि अनुयायी ने उत्तर दिया कि सब कुछ सफल था, तो हम अनुयायी मेटाडेटा को अपडेट करते हैं: हम अनुयायी के लॉग के अंतिम सूचकांक और अनुयायी द्वारा अपेक्षित अगले ऑपरेशन के सूचकांक को बचाते हैं।
6. , , , , . , , . , . , .
7. , . .

public AnswerAppendDTO append(RequestAppendDTO dto) {
     
       //1     
       if (dto.getTerm() < context.getCurrentTerm()) {
           return new AnswerAppendDTO(context.getId(),context.getCurrentTerm(),false, null);
       } else if (dto.getTerm() > context.getCurrentTerm()) {
           //2 
           context.setCurrentTerm(dto.getTerm());
           context.setVotedFor(null);
       }
       //3  
       applicationEventPublisher.publishEvent(new ResetElectionTimerEvent(this));

       if (!context.getState().equals(FOLLOWER)) {
           context.setState(FOLLOWER);
       }
        
       //4  
       if ((dto.getPrevLogIndex() > operationsLog.getLastIndex()) ||                                                                                        !dto.getPrevLogTerm().equals(operationsLog.getTerm(dto.getPrevLogIndex()))) {
                      return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), false, null);
       }


       Operation newOperation = dto.getOperation();
       if (newOperation != null) {
           int newOperationIndex = dto.getPrevLogIndex() + 1;
           
         synchronized (this) {
               //5
               if ((newOperationIndex <= operationsLog.getLastIndex()) &&
                      (!newOperation.getTerm().equals(operationsLog.getTerm(newOperationIndex)))){
                   operationsLog.removeAllFromIndex(newOperationIndex);
               }
               //6
               if (newOperationIndex <= operationsLog.getLastIndex())
               {
                 return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true,      operationsLog.getLastIndex());
               }
               //7
               operationsLog.append(newOperation);
           }
        }
        //8 
        if (dto.getLeaderCommit() > context.getCommitIndex()) {
           context.setCommitIndex(Math.min(dto.getLeaderCommit(), operationsLog.getLastIndex()));
       }

                 
       return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true, operationsLog.getLastIndex());
   }

1. यदि नेता का दौर अनुयायी के दौर से कम है, तो हम अपने नेता को एक दौर भेजते हैं और संकेत देते हैं कि उनका अनुरोध अस्वीकार कर दिया गया है। जैसे ही नेता प्रतिक्रिया में अपने से बड़ा एक गोल प्राप्त करता है, वह अनुयायी में बदल जाएगा
2. यदि नेता का दौर अनुयायी के दौर से अधिक है, तो इस दौर को अनुयायी पर सेट करें।
3. चूंकि नेता से अनुरोध प्राप्त हुआ था, इस बात की परवाह किए बिना कि वहां डेटा है या नहीं, हम वोट टाइमर को रीसेट करते हैं और, यदि हम अनुयायी नहीं थे, तो हम यह बन जाते हैं
4. , , , , , , . , ,
5. , . . , , - , , , , . , .
6. , . ,
7. ,
8. , , , .

  private void tryToCommit() {
       while (true) {
           //1
           int N = context.getCommitIndex() + 1;
           //2
           Supplier<Long> count = () ->
               context.getPeers().stream().map(Peer::getMatchIndex).
                       filter(matchIndex -> matchIndex >= N).count() + 1;

           //3 
           if (operationsLog.getLastIndex() >= N &&
                   operationsLog.getTerm(N).equals(context.getCurrentTerm())&&
                      count.get()>=context.getQuorum()
           )
           {
               context.setCommitIndex(N);
           } else
               return;
       }
   }

1. हम डेटाबेस पर लागू ऑपरेशन के निम्नलिखित सूचकांक प्राप्त करते हैं
2. हम गिनते हैं कि कितने अनुयायियों के पास उनके लॉग में इस तरह के सूचकांक के साथ एक ऑपरेशन है, और एक नेता को जोड़ने के लिए मत भूलना
3. यदि ऐसे अनुयायियों की संख्या कोरम है और इस तरह के सूचकांक के साथ संचालन नेता के लॉग में है, और इस ऑपरेशन का दौर वर्तमान के बराबर है, तो नेता डेटाबेस पर ऑपरेशन लागू करता है और अंतिम लागू किए गए ऑपरेशन के सूचकांक को बढ़ाता है। पिछले दौर के संचालन को लागू नहीं किया जा सकता है, क्योंकि उनके लिए एक और नेता जिम्मेदार था और संघर्ष पैदा हो सकता है। प्रत्येक नेता केवल अपने वर्तमान दौर के संचालन को लागू करता है।

निष्कर्ष

संदर्भ

• रिपोजिटरी ।
• विशिष्टता
• संक्षिप्त वर्णन