@Test public void testConcurrentProposal() { FleaseMessage prep1 = new FleaseMessage(FleaseMessage.MsgType.MSG_PREPARE); prep1.setCellId(TESTCELL); prep1.setProposalNo(new ProposalNumber(1, 1)); prep1.setSendTimestamp(TimeSync.getLocalSystemTime()); prep1.setLeaseTimeout(TimeSync.getLocalSystemTime()+10000); prep1.setLeaseHolder(new ASCIIString("me1")); FleaseMessage prep2 = new FleaseMessage(FleaseMessage.MsgType.MSG_PREPARE); prep2.setCellId(TESTCELL); prep2.setProposalNo(new ProposalNumber(1, 2)); prep2.setSendTimestamp(TimeSync.getLocalSystemTime()); prep2.setLeaseTimeout(TimeSync.getLocalSystemTime()+10000); prep2.setLeaseHolder(new ASCIIString("me2")); FleaseMessage response = acceptor.processMessage(prep1); assertEquals(MsgType.MSG_PREPARE_ACK,response.getMsgType()); response = acceptor.processMessage(prep2); assertEquals(MsgType.MSG_PREPARE_ACK,response.getMsgType()); prep1 = new FleaseMessage(MsgType.MSG_ACCEPT,prep1); prep1.setSendTimestamp(TimeSync.getLocalSystemTime()); response = acceptor.processMessage(prep1); assertEquals(MsgType.MSG_ACCEPT_NACK,response.getMsgType()); }
public FleaseMessage processMessage(FleaseMessage msg) { assert(!quit); final long now = TimeSync.getLocalSystemTime(); if (msg.getSendTimestamp()+config.getMessageTimeout() < TimeSync.getGlobalTime()) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A outdated message discarded: %s",msg.toString()); return null; } if (this.waitUntilTimestamp_ms >= now) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A message discarded, acceptor is still in recovery period"); return null; } assert(msg.getCellId() != null); final FleaseAcceptorCell cc = getCell(msg.getCellId()); if (cc.getViewId() < msg.getViewId()) { viewListener.viewIdChangeEvent(msg.getCellId(), msg.getViewId(), false); } else if (cc.getViewId() > msg.getViewId() || (cc.getViewId() == msg.getViewId() && cc.isViewInvalidated())) { FleaseMessage response = new FleaseMessage(FleaseMessage.MsgType.MSG_WRONG_VIEW, msg); response.setViewId(cc.getViewId()); return response; } FleaseMessage response = null; if (msg.getMsgType() == FleaseMessage.MsgType.MSG_PREPARE) response = handlePREPARE(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_ACCEPT) response = handleACCEPT(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_LEARN) handleLEARN(msg); else Logging.logMessage(Logging.LEVEL_ERROR, Category.flease,this,"A invalid message type received: %s",msg.toString()); return response; }
FleaseAcceptor { public FleaseMessage processMessage(FleaseMessage msg) { assert(!quit); final long now = TimeSync.getLocalSystemTime(); if (msg.getSendTimestamp()+config.getMessageTimeout() < TimeSync.getGlobalTime()) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A outdated message discarded: %s",msg.toString()); return null; } if (this.waitUntilTimestamp_ms >= now) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A message discarded, acceptor is still in recovery period"); return null; } assert(msg.getCellId() != null); final FleaseAcceptorCell cc = getCell(msg.getCellId()); if (cc.getViewId() < msg.getViewId()) { viewListener.viewIdChangeEvent(msg.getCellId(), msg.getViewId(), false); } else if (cc.getViewId() > msg.getViewId() || (cc.getViewId() == msg.getViewId() && cc.isViewInvalidated())) { FleaseMessage response = new FleaseMessage(FleaseMessage.MsgType.MSG_WRONG_VIEW, msg); response.setViewId(cc.getViewId()); return response; } FleaseMessage response = null; if (msg.getMsgType() == FleaseMessage.MsgType.MSG_PREPARE) response = handlePREPARE(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_ACCEPT) response = handleACCEPT(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_LEARN) handleLEARN(msg); else Logging.logMessage(Logging.LEVEL_ERROR, Category.flease,this,"A invalid message type received: %s",msg.toString()); return response; } }
FleaseAcceptor { public FleaseMessage processMessage(FleaseMessage msg) { assert(!quit); final long now = TimeSync.getLocalSystemTime(); if (msg.getSendTimestamp()+config.getMessageTimeout() < TimeSync.getGlobalTime()) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A outdated message discarded: %s",msg.toString()); return null; } if (this.waitUntilTimestamp_ms >= now) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A message discarded, acceptor is still in recovery period"); return null; } assert(msg.getCellId() != null); final FleaseAcceptorCell cc = getCell(msg.getCellId()); if (cc.getViewId() < msg.getViewId()) { viewListener.viewIdChangeEvent(msg.getCellId(), msg.getViewId(), false); } else if (cc.getViewId() > msg.getViewId() || (cc.getViewId() == msg.getViewId() && cc.isViewInvalidated())) { FleaseMessage response = new FleaseMessage(FleaseMessage.MsgType.MSG_WRONG_VIEW, msg); response.setViewId(cc.getViewId()); return response; } FleaseMessage response = null; if (msg.getMsgType() == FleaseMessage.MsgType.MSG_PREPARE) response = handlePREPARE(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_ACCEPT) response = handleACCEPT(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_LEARN) handleLEARN(msg); else Logging.logMessage(Logging.LEVEL_ERROR, Category.flease,this,"A invalid message type received: %s",msg.toString()); return response; } FleaseAcceptor(LearnEventListener evtListener, FleaseConfig localConfig, String lockfileDir, boolean ignoreLockFileForTesting); }
FleaseAcceptor { public FleaseMessage processMessage(FleaseMessage msg) { assert(!quit); final long now = TimeSync.getLocalSystemTime(); if (msg.getSendTimestamp()+config.getMessageTimeout() < TimeSync.getGlobalTime()) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A outdated message discarded: %s",msg.toString()); return null; } if (this.waitUntilTimestamp_ms >= now) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A message discarded, acceptor is still in recovery period"); return null; } assert(msg.getCellId() != null); final FleaseAcceptorCell cc = getCell(msg.getCellId()); if (cc.getViewId() < msg.getViewId()) { viewListener.viewIdChangeEvent(msg.getCellId(), msg.getViewId(), false); } else if (cc.getViewId() > msg.getViewId() || (cc.getViewId() == msg.getViewId() && cc.isViewInvalidated())) { FleaseMessage response = new FleaseMessage(FleaseMessage.MsgType.MSG_WRONG_VIEW, msg); response.setViewId(cc.getViewId()); return response; } FleaseMessage response = null; if (msg.getMsgType() == FleaseMessage.MsgType.MSG_PREPARE) response = handlePREPARE(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_ACCEPT) response = handleACCEPT(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_LEARN) handleLEARN(msg); else Logging.logMessage(Logging.LEVEL_ERROR, Category.flease,this,"A invalid message type received: %s",msg.toString()); return response; } FleaseAcceptor(LearnEventListener evtListener, FleaseConfig localConfig, String lockfileDir, boolean ignoreLockFileForTesting); void setViewChangeListener(FleaseViewChangeListenerInterface listener); void setViewId(ASCIIString cellId, int viewId); FleaseMessage handlePREPARE(FleaseMessage msg); FleaseMessage handleACCEPT(FleaseMessage msg); void handleLEARN(FleaseMessage msg); FleaseMessage getLocalLeaseInformation(ASCIIString cellId); Map<ASCIIString,FleaseMessage> localState(); FleaseMessage processMessage(FleaseMessage msg); String toString(); void shutdown(); }
FleaseAcceptor { public FleaseMessage processMessage(FleaseMessage msg) { assert(!quit); final long now = TimeSync.getLocalSystemTime(); if (msg.getSendTimestamp()+config.getMessageTimeout() < TimeSync.getGlobalTime()) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A outdated message discarded: %s",msg.toString()); return null; } if (this.waitUntilTimestamp_ms >= now) { if (Logging.isDebug() && config.isDebugPrintMessages()) Logging.logMessage(Logging.LEVEL_DEBUG, Category.flease, this,"A message discarded, acceptor is still in recovery period"); return null; } assert(msg.getCellId() != null); final FleaseAcceptorCell cc = getCell(msg.getCellId()); if (cc.getViewId() < msg.getViewId()) { viewListener.viewIdChangeEvent(msg.getCellId(), msg.getViewId(), false); } else if (cc.getViewId() > msg.getViewId() || (cc.getViewId() == msg.getViewId() && cc.isViewInvalidated())) { FleaseMessage response = new FleaseMessage(FleaseMessage.MsgType.MSG_WRONG_VIEW, msg); response.setViewId(cc.getViewId()); return response; } FleaseMessage response = null; if (msg.getMsgType() == FleaseMessage.MsgType.MSG_PREPARE) response = handlePREPARE(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_ACCEPT) response = handleACCEPT(msg); else if (msg.getMsgType() == FleaseMessage.MsgType.MSG_LEARN) handleLEARN(msg); else Logging.logMessage(Logging.LEVEL_ERROR, Category.flease,this,"A invalid message type received: %s",msg.toString()); return response; } FleaseAcceptor(LearnEventListener evtListener, FleaseConfig localConfig, String lockfileDir, boolean ignoreLockFileForTesting); void setViewChangeListener(FleaseViewChangeListenerInterface listener); void setViewId(ASCIIString cellId, int viewId); FleaseMessage handlePREPARE(FleaseMessage msg); FleaseMessage handleACCEPT(FleaseMessage msg); void handleLEARN(FleaseMessage msg); FleaseMessage getLocalLeaseInformation(ASCIIString cellId); Map<ASCIIString,FleaseMessage> localState(); FleaseMessage processMessage(FleaseMessage msg); String toString(); void shutdown(); final Map<ASCIIString, FleaseAcceptorCell> cells; static final String LOCKFILE_NAME; public boolean quit; }