"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "dso-l2/src/main/java/com/tc/objectserver/handler/ReplicatedTransactionHandler.java" between
terracotta-core-5.7.0.tar.gz and terracotta-core-5.7.1.tar.gz

About: Terracotta - Clustering technology for Java ("Network Attached Memory", "Distributed Cache"). Source code of the server (5.x) core components.

ReplicatedTransactionHandler.java  (terracotta-core-5.7.0):ReplicatedTransactionHandler.java  (terracotta-core-5.7.1)
skipping to change at line 101 skipping to change at line 101
private final EntityManager entityManager; private final EntityManager entityManager;
private final Persistor persistor; private final Persistor persistor;
private final GroupManager<AbstractGroupMessage> groupManager; private final GroupManager<AbstractGroupMessage> groupManager;
private final StateManager stateManager; private final StateManager stateManager;
private final ManagedEntity platform; private final ManagedEntity platform;
private final SyncState state = new SyncState(); private final SyncState state = new SyncState();
// This MUST be manipulated under lock - it is the batch of ack messages we ar e accumulating until the network is ready for another message. // This MUST be manipulated under lock - it is the batch of ack messages we ar e accumulating until the network is ready for another message.
private NodeID cachedMessageAckFrom; private ServerID cachedMessageAckFrom;
private GroupMessageBatchContext<ReplicationMessageAck, ReplicationAckTuple> c achedBatchAck; private GroupMessageBatchContext<ReplicationMessageAck, ReplicationAckTuple> c achedBatchAck;
private final Sink<Runnable> sentToActive; private final Sink<Runnable> sentToActive;
private volatile long currentSequence = 0; private volatile long currentSequence = 0;
public long getCurrentSequence() { public long getCurrentSequence() {
return currentSequence; return currentSequence;
} }
public ReplicatedTransactionHandler(StateManager state, Stage<Runnable> sendTo Active, Persistor persistor, public ReplicatedTransactionHandler(StateManager state, Stage<Runnable> sendTo Active, Persistor persistor,
skipping to change at line 622 skipping to change at line 622
prepareAckForSend(activeSender, activity.getActivityID(), code); prepareAckForSend(activeSender, activity.getActivityID(), code);
} }
} }
private ReplicationMessageAck createAckMessage(ReplicationAckTuple initialActi vity) { private ReplicationMessageAck createAckMessage(ReplicationAckTuple initialActi vity) {
ReplicationMessageAck message = ReplicationMessageAck.createBatchAck(); ReplicationMessageAck message = ReplicationMessageAck.createBatchAck();
message.addToBatch(initialActivity); message.addToBatch(initialActivity);
return message; return message;
} }
private synchronized void prepareAckForSend(NodeID sender, SyncReplicationActi vity.ActivityID respondTo, ReplicationResultCode code) { private synchronized void prepareAckForSend(ServerID sender, SyncReplicationAc tivity.ActivityID respondTo, ReplicationResultCode code) {
// The batch context is cached and constructed lazily when the sender change s. // The batch context is cached and constructed lazily when the sender change s.
if (!sender.equals(this.cachedMessageAckFrom)) { if (!sender.equals(this.cachedMessageAckFrom)) {
this.cachedMessageAckFrom = sender; this.cachedMessageAckFrom = sender;
this.cachedBatchAck = new GroupMessageBatchContext<>(this::createAckMessag e, this.groupManager, this.cachedMessageAckFrom, maximumBatchSize, idealMessages InFlight, (node)->sendToActive()); this.cachedBatchAck = new GroupMessageBatchContext<>(this::createAckMessag e, this.groupManager, this.cachedMessageAckFrom, maximumBatchSize, idealMessages InFlight, (node)->sendToActive());
} }
boolean didCreate = this.cachedBatchAck.batchMessage(new ReplicationAckTuple (respondTo, code)); boolean didCreate = this.cachedBatchAck.batchMessage(new ReplicationAckTuple (respondTo, code));
// If we created this message, enqueue the decision to flush it (the other c ase where we may flush is network // If we created this message, enqueue the decision to flush it (the other c ase where we may flush is network
// available). // available).
 End of changes. 2 change blocks. 
2 lines changed or deleted 2 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)