Menerapkan Algoritma Konsensus RAFT untuk Penyimpanan KV Terdistribusi di Jawa

Deskripsi Solusi Umum

• Pengikut (pengikut). Menerima permintaan baca dari klien. Mengambil detak jantung dari pemimpin
• Calon (kandidat). Menerima permintaan baca dari klien. Mengirim permintaan suara ke node lain
• Pemimpin Menerima permintaan baca dan tulis. Mengirim permintaan detak jantung ke node lain. Mengirimkan menambahkan permintaan data ke node lain.

Penyimpanan data

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);
}

Pengatur waktu

• memilih saat melakukan putaran pemungutan suara
• append, alias detak jantung (jika tanpa data), untuk mereplikasi data log ke pengikut dan untuk mencegah dimulainya putaran suara baru.

• Pilih. Untuk memulai putaran pemungutan suara. Setiap node memiliki intervalnya sendiri, setelah itu akan mencoba untuk memulai suara baru. Hitungan mundur dimulai lagi ketika menerima detak jantung dari pemimpin.
• denyut jantung. Untuk mengirim permintaan ke pengikut oleh pemimpin menambahkan. Jika node tidak menerima detak jantung dan timer pemungutan suara telah kedaluwarsa, ia menjadi kandidat dan memulai pemilihan, meningkatkan jumlah putaran pemungutan suara dan mengirimkan permintaan pemungutan suara ke node lain. Jika simpul mengumpulkan mayoritas suara, maka ia menjadi pemimpin dan mulai mengirimkan detak jantung.

Keadaan node saat ini

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(); //     
}

Pemungutan suara

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

Mengirim permintaan untuk pemungutan suara

//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. Tetapkan status "Calon". Naikkan angka bulat dan pilih untuk diri kita sendiri.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. Node memilih kami! Kami meningkatkan jumlah node yang memberikan suara untuk kami dan memperbaiki bahwa simpul ini memilih kami.
6. Dipilih bukan untuk kita, kita juga percaya.
7. Jika kuorum dikumpulkan dan simpul memenangkan pemilihan, kami menetapkan status "Pemimpin". Kalau tidak, kita menjadi pengikut dan menunggu.

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

       );
   }

Voting memproses permintaan

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. Jika putaran simpul saat ini lebih besar daripada putaran kandidat, kami menolak, karena ini adalah permintaan dari beberapa simpul yang tertinggal, yang, tampaknya, berada di luar cluster untuk beberapa waktu dan memulai prosedur pemilihan karena tidak melihat pemimpin saat ini.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. Dengan hasil positif, kami memperbaiki fakta bahwa simpul tersebut ikut serta dalam pemilihan dan memberikan suara untuk kandidat.
6. Kirim hasilnya kembali ke kandidat

Replikasi

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);
}

Kirim permintaan perubahan data

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. Metadata pengikut
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. Kami mengulangi permintaan sampai kami menerima tanggapan dari semua pengikut bahwa replikasi berhasil. Karena satu operasi dikirim per permintaan, mungkin diperlukan beberapa iterasi untuk menyinkronkan log pengikut
2. Kirim permintaan ke semua pengikut dan dapatkan daftar dengan jawaban
3. Kami menganggap jawaban hanya dari pengikut yang tersedia
4. Jika ternyata putaran salah satu pengikut lebih dari putaran pemimpin, kami menghentikan semuanya dan berubah menjadi pengikut
5. Jika pengikut menjawab bahwa semuanya berjalan dengan baik, kami memperbarui metadata pengikut: kami menyimpan indeks terakhir dari log pengikut dan indeks dari operasi berikutnya yang diharapkan oleh pengikut.
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. Jika ronde pemimpin kurang dari ronde pengikut, maka kami mengirim ronde pemimpin kami dan tanda bahwa permintaannya telah ditolak. Begitu pemimpin menerima satu ronde yang lebih besar daripada tanggapannya, dia akan berubah menjadi pengikut
2. Jika ronde pemimpin lebih dari ronde pengikut, atur ronde ini ke pengikut.
3. Karena permintaan diterima dari pemimpin, terlepas dari apakah ada data di sana atau tidak, kami menyetel ulang penghitung waktu suara dan, jika kami bukan pengikut, kami menjadi pengikutnya.
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. Kami mendapatkan indeks operasi berikut yang diterapkan ke database
2. Kami menghitung berapa banyak pengikut yang memiliki operasi dengan indeks sedemikian dalam log mereka, dan jangan lupa untuk menambahkan seorang pemimpin
3. Jika jumlah pengikut seperti itu adalah kuorum dan operasi dengan indeks seperti itu ada di log pemimpin, dan putaran operasi ini setara dengan yang sekarang, maka pemimpin menerapkan operasi ke database dan meningkatkan indeks operasi yang terakhir diterapkan. Operasi dari babak sebelumnya tidak dapat diterapkan, karena pemimpin lain bertanggung jawab atas mereka dan konflik dapat muncul. Setiap pemimpin hanya menerapkan operasi putarannya saat ini.

Kesimpulan

Referensi

• Repositori .
• Spesifikasi
• Deskripsi Singkat