eclipse-sailing-analytics/sailing-analytics
1
0 Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-10-22 03:53:56 +00:00
Code Activity

normalized line endings

Browse Source
This commit is contained in:
Steffen Schaefer committed 2017-05-16 09:42:02 +02:00
1 parent 8cf9053d37
commit 9123771f34
2780 files changed
+301735 -301735
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/ClientServerOperationPair.java
+35 -35
View File
@@ -1,35 +1,35 @@
package com.sap.sse.operationaltransformation;
public class ClientServerOperationPair<S, O extends Operation<S>> {
private O clientOp;
private O serverOp;
public ClientServerOperationPair(O clientOp, O serverOp) {
super();
this.clientOp = clientOp;
this.serverOp = serverOp;
}
public O getClientOp() {
return clientOp;
}
public O getServerOp() {
return serverOp;
}
@SuppressWarnings("unchecked")
public static <O extends Operation<S>, S> O getNoOp() {
return (O) new Operation<S>() {
@Override
public S applyTo(S toState) {
return toState; // don't change anything, return state unchanged
}
@Override
public boolean requiresSynchronousExecution() {
return false;
}
};
}
}
package com.sap.sse.operationaltransformation;
public class ClientServerOperationPair<S, O extends Operation<S>> {
private O clientOp;
private O serverOp;
public ClientServerOperationPair(O clientOp, O serverOp) {
super();
this.clientOp = clientOp;
this.serverOp = serverOp;
}
public O getClientOp() {
return clientOp;
}
public O getServerOp() {
return serverOp;
}
@SuppressWarnings("unchecked")
public static <O extends Operation<S>, S> O getNoOp() {
return (O) new Operation<S>() {
@Override
public S applyTo(S toState) {
return toState; // don't change anything, return state unchanged
}
@Override
public boolean requiresSynchronousExecution() {
return false;
}
};
}
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/Operation.java
+30 -30
View File
@@ -1,30 +1,30 @@
package com.sap.sse.operationaltransformation;
/**
* A change operation can be applied to some start state of some
* universe and transforms that universe into a target state.
*
* @author Axel Uhl D043530
*
* @param <S> the class that models the state, on instances of which operations perform changes
*/
public interface Operation<S> extends Cloneable {
/**
* Applies this operation to the state <tt>toState</tt>, producing a new target state
* which is returned.
*
* @param toState the state to which to apply this operation
* @return the state produced by applying this operation to <tt>toState</tt>
*/
S applyTo(S toState);
/**
* Tells if this operation needs to be executed in order with other operations requesting synchronous execution.
* This default implementation returns <code>true</code> which is safe but may be slower and using less concurrency
* than possible. Implementations may make the assertion that parallel execution will do no harm by overriding this
* method, returning <code>true</code>.
*/
default boolean requiresSynchronousExecution() {
return true;
}
}
package com.sap.sse.operationaltransformation;
/**
* A change operation can be applied to some start state of some
* universe and transforms that universe into a target state.
*
* @author Axel Uhl D043530
*
* @param <S> the class that models the state, on instances of which operations perform changes
*/
public interface Operation<S> extends Cloneable {
/**
* Applies this operation to the state <tt>toState</tt>, producing a new target state
* which is returned.
*
* @param toState the state to which to apply this operation
* @return the state produced by applying this operation to <tt>toState</tt>
*/
S applyTo(S toState);
/**
* Tells if this operation needs to be executed in order with other operations requesting synchronous execution.
* This default implementation returns <code>true</code> which is safe but may be slower and using less concurrency
* than possible. Implementations may make the assertion that parallel execution will do no harm by overriding this
* method, returning <code>true</code>.
*/
default boolean requiresSynchronousExecution() {
return true;
}
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/OperationalTransformer.java
+20 -20
View File
@@ -1,20 +1,20 @@
package com.sap.sse.operationaltransformation;
/**
* A default implementation for the {@link Transformer} interface, assuming operations implement the
* {@link OperationWithTransformationSupport} interface which standardizes how operations are being transformed.
*
* @author Axel Uhl (D043530)
*
* @param <S>
* @param <O>
*/
public class OperationalTransformer<S, O extends OperationWithTransformationSupport<S, O>> implements Transformer<S, O> {
@Override
public ClientServerOperationPair<S, O> transform(O clientOp, O serverOp) {
ClientServerOperationPair<S, O> result = new ClientServerOperationPair<S, O>(
clientOp == null ? null : clientOp.transformClientOp(serverOp),
serverOp == null ? null : serverOp.transformServerOp(clientOp));
return result;
}
}
package com.sap.sse.operationaltransformation;
/**
* A default implementation for the {@link Transformer} interface, assuming operations implement the
* {@link OperationWithTransformationSupport} interface which standardizes how operations are being transformed.
*
* @author Axel Uhl (D043530)
*
* @param <S>
* @param <O>
*/
public class OperationalTransformer<S, O extends OperationWithTransformationSupport<S, O>> implements Transformer<S, O> {
@Override
public ClientServerOperationPair<S, O> transform(O clientOp, O serverOp) {
ClientServerOperationPair<S, O> result = new ClientServerOperationPair<S, O>(
clientOp == null ? null : clientOp.transformClientOp(serverOp),
serverOp == null ? null : serverOp.transformServerOp(clientOp));
return result;
}
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/Peer.java
+100 -100
View File
@@ -1,100 +1,100 @@
package com.sap.sse.operationaltransformation;
/**
* Accepts change operations from clients and maintains a shared state. If new
* clients connect, the full state can be sent to the client for initial
* synchronization. When a change {@link Operation} is received from a client
* that operation will be based on some state. Since that state the
* server may have applied one or more change operations already. In order to
* synchronize the requesting client, the server transforms all in-between
* change operations for the requesting client and sends those transformation
* results back to the requesting client. The client's operation, in turn, is
* disseminated to all other clients for them to merge into their local replica
* of the global state.
*
* @author Axel Uhl D043530
*
* @param <O> the type of operations that can be applied to the states
* @param <S> the type of the states on which the operations perform changes
*/
public interface Peer<O extends Operation<S>, S> {
/**
* Enumeration representing the two valid roles a peer can have. There must
* be exactly one server, multiple clients are allowed.
*
* @author Jan Karstens D046040
*
*/
public enum Role {
SERVER, CLIENT
}
/**
* Registers a peer with this peer for replication. From this point on, this
* peer will remember which operations it has forwarded to that <code>peer</code> for
* later transformation jobs for operations coming in from that <code>peer</code>.<p>
*
* The <code>peer</code> is expected to set the state returned by this operation
* as its initial state.
*
* @return the current server state; the calling client can use this as its
* initial state
*/
S addPeer(Peer<O, S> peer);
/**
* Usually, the framework handles calls to this method and framework users
* should therefore not need to call this method.
* <p>
*
* Synchronizes an <tt>operation</tt> from the <tt>source</tt> peer to this
* peer. The <tt>source</tt> peer specifies how many operations from this
* peer it already merged into its operations sequence before having
* executed <tt>operation</tt>.
* <p>
*
* The operation has to be transformed using this peer's operational
* transformation and using the sequence of operations applied to the
* operation's start state up to this peer's current state. This results in
* the operation that needs to be applied to the current state to produce
* the next state.
* <p>
*
* After applying this possibly transformed operation to this peer's current
* state and thus producing a new current state, all registered peers except
* <tt>source</tt> are notified about the change by sending them the
* transformed operation.
*
*/
void apply(Peer<O, S> source, O operation, int numberOfOperationsSourceHasMergedFromThis);
/**
* Applies a change operation locally to the current peer's state and update all known
* peers by sending them the operation.
*/
void apply(O operation);
/**
* Tells this peer that the <tt>source</tt> peer has successfully merged
* <tt>numberOfMergedOperations</tt> altogether coming from this peer
* into its operation sequence. This peer can take this as an opportunity
* to clean up the queue and remove the transformed copies of all operations
* up to number <tt>numberOfMergedOperations-1</tt>.
*/
void confirm(Peer<O, S> source, int numberOfMergedOperations);
/**
* Tells the current state in which the peer is. This is the state constructed by
* the sequence of calls to {@link #apply(Operation)} and the merged in notifications
* from other peers.
*/
S getCurrentState();
/**
* Blocks until the queue of tasks that works on achieving consistency across peers
* has completely drained. This can be useful for testing purposes when you want to
* wait until consistency has been reached across clients and server.
*/
void waitForNotRunning();
}
package com.sap.sse.operationaltransformation;
/**
* Accepts change operations from clients and maintains a shared state. If new
* clients connect, the full state can be sent to the client for initial
* synchronization. When a change {@link Operation} is received from a client
* that operation will be based on some state. Since that state the
* server may have applied one or more change operations already. In order to
* synchronize the requesting client, the server transforms all in-between
* change operations for the requesting client and sends those transformation
* results back to the requesting client. The client's operation, in turn, is
* disseminated to all other clients for them to merge into their local replica
* of the global state.
*
* @author Axel Uhl D043530
*
* @param <O> the type of operations that can be applied to the states
* @param <S> the type of the states on which the operations perform changes
*/
public interface Peer<O extends Operation<S>, S> {
/**
* Enumeration representing the two valid roles a peer can have. There must
* be exactly one server, multiple clients are allowed.
*
* @author Jan Karstens D046040
*
*/
public enum Role {
SERVER, CLIENT
}
/**
* Registers a peer with this peer for replication. From this point on, this
* peer will remember which operations it has forwarded to that <code>peer</code> for
* later transformation jobs for operations coming in from that <code>peer</code>.<p>
*
* The <code>peer</code> is expected to set the state returned by this operation
* as its initial state.
*
* @return the current server state; the calling client can use this as its
* initial state
*/
S addPeer(Peer<O, S> peer);
/**
* Usually, the framework handles calls to this method and framework users
* should therefore not need to call this method.
* <p>
*
* Synchronizes an <tt>operation</tt> from the <tt>source</tt> peer to this
* peer. The <tt>source</tt> peer specifies how many operations from this
* peer it already merged into its operations sequence before having
* executed <tt>operation</tt>.
* <p>
*
* The operation has to be transformed using this peer's operational
* transformation and using the sequence of operations applied to the
* operation's start state up to this peer's current state. This results in
* the operation that needs to be applied to the current state to produce
* the next state.
* <p>
*
* After applying this possibly transformed operation to this peer's current
* state and thus producing a new current state, all registered peers except
* <tt>source</tt> are notified about the change by sending them the
* transformed operation.
*
*/
void apply(Peer<O, S> source, O operation, int numberOfOperationsSourceHasMergedFromThis);
/**
* Applies a change operation locally to the current peer's state and update all known
* peers by sending them the operation.
*/
void apply(O operation);
/**
* Tells this peer that the <tt>source</tt> peer has successfully merged
* <tt>numberOfMergedOperations</tt> altogether coming from this peer
* into its operation sequence. This peer can take this as an opportunity
* to clean up the queue and remove the transformed copies of all operations
* up to number <tt>numberOfMergedOperations-1</tt>.
*/
void confirm(Peer<O, S> source, int numberOfMergedOperations);
/**
* Tells the current state in which the peer is. This is the state constructed by
* the sequence of calls to {@link #apply(Operation)} and the merged in notifications
* from other peers.
*/
S getCurrentState();
/**
* Blocks until the queue of tasks that works on achieving consistency across peers
* has completely drained. This can be useful for testing purposes when you want to
* wait until consistency has been reached across clients and server.
*/
void waitForNotRunning();
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/PeerImpl.java
+295 -295
View File
@@ -1,295 +1,295 @@
package com.sap.sse.operationaltransformation;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
/**
* A very simple peer implementation for operational transformations that starts
* out with a provided initial state and then receives operations which it applies
* to its state. It manages a set of clients. Using the <tt>server</tt> constructor
* parameter one can configure whether this peer is a client or a server.
*
* @author Axel Uhl D043530
*
* @param <O> type of change operations to be processed by this server
* @param <S> type of state managed by this server to which the operations can be applied
*/
public class PeerImpl<O extends Operation<S>, S> implements Peer<O, S> {
private static final Logger logger = Logger.getLogger(PeerImpl.class.getName());
private String name;
private S currentState;
/**
* Tells if this peer acts in the server or client role. This is used to determine in which
* direction to apply the operational transformation.
*/
private Role role;
/**
* Tells if this peer is currently actively running, with a certain number of pending / running requests
*/
private int scheduledOrRunning = 0;
/**
* Queues operations sent out to a peer in {@link #updatePeers(Operation, Peer)} and whose
* merging has not yet been confirmed by that peer. Those messages are transformed when an
* operation from that peer is applied. When applying an operation to the peer has been
* confirmed, the first operation from the queue is removed because operations sent
* from this peer to the remote peer are processed in order.
*/
private final Map<Peer<O, S>, UnmergedOperationsQueue<O, S>> unmergedOperationsForPeer = new HashMap<Peer<O, S>, UnmergedOperationsQueue<O, S>>();
/**
* Remembers per peer how many of that peer's operations this peer has already
* applied locally.
*/
private final Map<Peer<O, S>, Integer> numberOfMergedOperations = new HashMap<Peer<O, S>, Integer>();
private final Transformer<S, O> transformer;
/**
* The background task handler which sends out the asynchronous updates to the peers.
*/
private final ExecutorService merger;
public PeerImpl(Transformer<S, O> transformer, S initialState, Role role) {
this.transformer = transformer;
currentState = initialState;
this.role = role;
this.merger = Executors.newSingleThreadExecutor(new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
}
public PeerImpl(String name, Transformer<S, O> transformer, S initialState, Role role) {
this(transformer, initialState, role);
this.name = name;
}
/**
* Constructs a client peer and connects it to the <tt>server</tt> peer, establishing
* the two-way link between client and server peers. The initial state is taken from
* the server.
*/
public PeerImpl(Transformer<S, O> transformer, Peer<O, S> server) {
this.transformer = transformer;
S initialState = server.addPeer(this);
currentState = initialState;
this.role = Role.CLIENT;
this.merger = Executors.newSingleThreadExecutor(new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
addPeer(server);
}
/**
* Constructs a client peer with a name and connects it to the
* <tt>server</tt> peer, establishing the two-way link between client and
* server peers. The initial state is taken from the server.
*/
public PeerImpl(String name, Transformer<S, O> transformer, Peer<O, S> server) {
this(transformer, server);
this.name = name;
}
@Override
public void finalize() {
merger.shutdown();
}
private Transformer<S, O> getTransformer() {
return transformer;
}
@Override
public S addPeer(Peer<O, S> peer) {
if (role == Role.CLIENT && getPeers().size() > 0) {
throw new RuntimeException("A client must be connected to at most one server");
}
unmergedOperationsForPeer.put(peer, new UnmergedOperationsQueue<O, S>());
numberOfMergedOperations.put(peer, 0);
return getCurrentState();
}
private Collection<Peer<O, S>> getPeers() {
return unmergedOperationsForPeer.keySet();
}
public S getCurrentState() {
return currentState;
}
@Override
public synchronized void apply(O operation) {
taskScheduled();
try {
currentState = operation.applyTo(currentState);
updatePeers(operation, /* except */null);
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
@Override
public synchronized void apply(final Peer<O, S> source, O operation,
final int numberOfOperationsSourceHasMergedFromThis) {
taskScheduled();
try {
if (!getPeers().contains(source)) {
throw new RuntimeException("Peer " + source + " not registered with peer " + this);
}
// Starting from base up to current, compute transformed operation sequence to send
// to client; this will create a sequence of states for the client which eventually
// leads up to a state that equals the server's current state.
O transformedOp = operation;
UnmergedOperationsQueue<O, S> unmergedOperationsForSource = unmergedOperationsForPeer.get(source);
int localOpNumber = numberOfOperationsSourceHasMergedFromThis;
for (O unconfirmedOperation : unmergedOperationsForSource
.getUnmergedOperations(numberOfOperationsSourceHasMergedFromThis)) {
if (role == Role.SERVER) {
final ClientServerOperationPair<S, O> pair;
if (transformedOp == null || unconfirmedOperation == null) {
pair = new ClientServerOperationPair<>(transformedOp, unconfirmedOperation);
} else {
pair = getTransformer().transform(transformedOp, unconfirmedOperation);
}
transformedOp = pair.getClientOp();
unmergedOperationsForSource.updateWithTransformed(localOpNumber, pair.getServerOp());
} else {
final ClientServerOperationPair<S, O> pair;
if (transformedOp == null || unconfirmedOperation == null) {
pair = new ClientServerOperationPair<>(unconfirmedOperation, transformedOp);
} else {
pair = getTransformer().transform(unconfirmedOperation, transformedOp);
}
transformedOp = pair.getServerOp();
unmergedOperationsForSource.updateWithTransformed(localOpNumber, pair.getClientOp());
}
localOpNumber++;
}
if (transformedOp != null) {
currentState = transformedOp.applyTo(currentState); // produce a new current state
}
final int numberOfMergedOperationsFromSource = numberOfMergedOperations.get(source) + 1;
numberOfMergedOperations.put(source, numberOfMergedOperationsFromSource);
// It's important that the following call to confirm is synchronized with the
// sending of updates to the source peer. Therefore, the confirm call is executed
// in the same serializing background thread:
scheduleTask(new Runnable() {
public void run() {
source.confirm(PeerImpl.this, numberOfMergedOperationsFromSource);
}
});
updatePeers(transformedOp, /* except */source);
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
/**
* Propagates the changes described by <tt>operation</tt> to all registered peers (except the peer identified by
* <tt>except</tt> if not <tt>null</tt>). For each peer, the operation is sent along to that peer's
* {@link #apply(Peer, Operation, int)} method together with the {@link #numberOfMergedOperations number of
* operations currently merged from that peer}.
*/
private synchronized void updatePeers(final O operation, Peer<O, S> except) {
for (final Peer<O, S> peer : getPeers()) {
if (except == null || !except.equals(peer)) {
unmergedOperationsForPeer.get(peer).sentOutOperation(operation);
// remember the number of merged operations while still in synchronized block
final int numberOfMergedOpsForPeer = numberOfMergedOperations.get(peer);
scheduleTask(new Runnable() {
public void run() {
/*
* Thoughts on locking: The peer runs the following apply(...) call synchronized.
* Therefore, it can't take other apply calls (local or from other peers) during
* that time. It may, though, have pending tasks in its updatePeers that can
* continue to run. This may include updates for this peer which would be received
* by this peer's apply(...) method.
*/
peer.apply(PeerImpl.this, operation, numberOfMergedOpsForPeer);
// TODO what to do if apply throws a RuntimeException?
}
});
}
}
}
public String toString() {
if (name != null) {
return name;
} else {
return (role == Role.SERVER?"server":"client")+" "+super.toString();
}
}
@Override
public void confirm(Peer<O, S> source, int numberOfMergedOperations) {
unmergedOperationsForPeer.get(source).confirm(numberOfMergedOperations);
}
/**
* Records that a new task is scheduled by incrementing {@link #scheduledOrRunning} by one, then
* schedules the <tt>runnable</tt> with the {@link #merger} executor service. The
* <tt>runnable</tt> is wrapped such that when its execution has finished,
* {@link #taskFinished()} will be called such that {@link #scheduledOrRunning} is decremented
* properly again and waiting clients in {@link #waitForNotRunning()} can be unblocked.
*/
private void scheduleTask(final Runnable runnable) {
taskScheduled();
merger.execute(new Runnable() {
public void run() {
try {
runnable.run();
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
});
}
private void taskScheduled() {
logger.fine(""+this+" taskScheduled incrementing scheduleOrRunning from "+scheduledOrRunning+" to "+(scheduledOrRunning+1));
scheduledOrRunning++;
}
/**
* Must be called by all tasks submitted to the {@link #merger} when finished.
* Unblocks {@link #waitForNotRunning()} if the merger's queue got empty by this.
*/
private synchronized void taskFinished() {
logger.fine(""+this+" taskFinished decrementing scheduleOrRunning from "+scheduledOrRunning+" to "+(scheduledOrRunning-1));
scheduledOrRunning--;
if (scheduledOrRunning == 0) {
logger.fine(""+this+" taskFinished notifying waiters");
notifyAll();
}
}
@Override
public synchronized void waitForNotRunning() {
while (scheduledOrRunning > 0) {
try {
wait();
logger.fine(""+this+" waitForNotRunning scheduleOrRunning: "+scheduledOrRunning);
} catch (InterruptedException e) {
// ignore interruption, try again
}
}
}
}
package com.sap.sse.operationaltransformation;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.sap.sse.util.impl.ThreadFactoryWithPriority;
/**
* A very simple peer implementation for operational transformations that starts
* out with a provided initial state and then receives operations which it applies
* to its state. It manages a set of clients. Using the <tt>server</tt> constructor
* parameter one can configure whether this peer is a client or a server.
*
* @author Axel Uhl D043530
*
* @param <O> type of change operations to be processed by this server
* @param <S> type of state managed by this server to which the operations can be applied
*/
public class PeerImpl<O extends Operation<S>, S> implements Peer<O, S> {
private static final Logger logger = Logger.getLogger(PeerImpl.class.getName());
private String name;
private S currentState;
/**
* Tells if this peer acts in the server or client role. This is used to determine in which
* direction to apply the operational transformation.
*/
private Role role;
/**
* Tells if this peer is currently actively running, with a certain number of pending / running requests
*/
private int scheduledOrRunning = 0;
/**
* Queues operations sent out to a peer in {@link #updatePeers(Operation, Peer)} and whose
* merging has not yet been confirmed by that peer. Those messages are transformed when an
* operation from that peer is applied. When applying an operation to the peer has been
* confirmed, the first operation from the queue is removed because operations sent
* from this peer to the remote peer are processed in order.
*/
private final Map<Peer<O, S>, UnmergedOperationsQueue<O, S>> unmergedOperationsForPeer = new HashMap<Peer<O, S>, UnmergedOperationsQueue<O, S>>();
/**
* Remembers per peer how many of that peer's operations this peer has already
* applied locally.
*/
private final Map<Peer<O, S>, Integer> numberOfMergedOperations = new HashMap<Peer<O, S>, Integer>();
private final Transformer<S, O> transformer;
/**
* The background task handler which sends out the asynchronous updates to the peers.
*/
private final ExecutorService merger;
public PeerImpl(Transformer<S, O> transformer, S initialState, Role role) {
this.transformer = transformer;
currentState = initialState;
this.role = role;
this.merger = Executors.newSingleThreadExecutor(new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
}
public PeerImpl(String name, Transformer<S, O> transformer, S initialState, Role role) {
this(transformer, initialState, role);
this.name = name;
}
/**
* Constructs a client peer and connects it to the <tt>server</tt> peer, establishing
* the two-way link between client and server peers. The initial state is taken from
* the server.
*/
public PeerImpl(Transformer<S, O> transformer, Peer<O, S> server) {
this.transformer = transformer;
S initialState = server.addPeer(this);
currentState = initialState;
this.role = Role.CLIENT;
this.merger = Executors.newSingleThreadExecutor(new ThreadFactoryWithPriority(Thread.NORM_PRIORITY, /* daemon */ true));
addPeer(server);
}
/**
* Constructs a client peer with a name and connects it to the
* <tt>server</tt> peer, establishing the two-way link between client and
* server peers. The initial state is taken from the server.
*/
public PeerImpl(String name, Transformer<S, O> transformer, Peer<O, S> server) {
this(transformer, server);
this.name = name;
}
@Override
public void finalize() {
merger.shutdown();
}
private Transformer<S, O> getTransformer() {
return transformer;
}
@Override
public S addPeer(Peer<O, S> peer) {
if (role == Role.CLIENT && getPeers().size() > 0) {
throw new RuntimeException("A client must be connected to at most one server");
}
unmergedOperationsForPeer.put(peer, new UnmergedOperationsQueue<O, S>());
numberOfMergedOperations.put(peer, 0);
return getCurrentState();
}
private Collection<Peer<O, S>> getPeers() {
return unmergedOperationsForPeer.keySet();
}
public S getCurrentState() {
return currentState;
}
@Override
public synchronized void apply(O operation) {
taskScheduled();
try {
currentState = operation.applyTo(currentState);
updatePeers(operation, /* except */null);
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
@Override
public synchronized void apply(final Peer<O, S> source, O operation,
final int numberOfOperationsSourceHasMergedFromThis) {
taskScheduled();
try {
if (!getPeers().contains(source)) {
throw new RuntimeException("Peer " + source + " not registered with peer " + this);
}
// Starting from base up to current, compute transformed operation sequence to send
// to client; this will create a sequence of states for the client which eventually
// leads up to a state that equals the server's current state.
O transformedOp = operation;
UnmergedOperationsQueue<O, S> unmergedOperationsForSource = unmergedOperationsForPeer.get(source);
int localOpNumber = numberOfOperationsSourceHasMergedFromThis;
for (O unconfirmedOperation : unmergedOperationsForSource
.getUnmergedOperations(numberOfOperationsSourceHasMergedFromThis)) {
if (role == Role.SERVER) {
final ClientServerOperationPair<S, O> pair;
if (transformedOp == null || unconfirmedOperation == null) {
pair = new ClientServerOperationPair<>(transformedOp, unconfirmedOperation);
} else {
pair = getTransformer().transform(transformedOp, unconfirmedOperation);
}
transformedOp = pair.getClientOp();
unmergedOperationsForSource.updateWithTransformed(localOpNumber, pair.getServerOp());
} else {
final ClientServerOperationPair<S, O> pair;
if (transformedOp == null || unconfirmedOperation == null) {
pair = new ClientServerOperationPair<>(unconfirmedOperation, transformedOp);
} else {
pair = getTransformer().transform(unconfirmedOperation, transformedOp);
}
transformedOp = pair.getServerOp();
unmergedOperationsForSource.updateWithTransformed(localOpNumber, pair.getClientOp());
}
localOpNumber++;
}
if (transformedOp != null) {
currentState = transformedOp.applyTo(currentState); // produce a new current state
}
final int numberOfMergedOperationsFromSource = numberOfMergedOperations.get(source) + 1;
numberOfMergedOperations.put(source, numberOfMergedOperationsFromSource);
// It's important that the following call to confirm is synchronized with the
// sending of updates to the source peer. Therefore, the confirm call is executed
// in the same serializing background thread:
scheduleTask(new Runnable() {
public void run() {
source.confirm(PeerImpl.this, numberOfMergedOperationsFromSource);
}
});
updatePeers(transformedOp, /* except */source);
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
/**
* Propagates the changes described by <tt>operation</tt> to all registered peers (except the peer identified by
* <tt>except</tt> if not <tt>null</tt>). For each peer, the operation is sent along to that peer's
* {@link #apply(Peer, Operation, int)} method together with the {@link #numberOfMergedOperations number of
* operations currently merged from that peer}.
*/
private synchronized void updatePeers(final O operation, Peer<O, S> except) {
for (final Peer<O, S> peer : getPeers()) {
if (except == null || !except.equals(peer)) {
unmergedOperationsForPeer.get(peer).sentOutOperation(operation);
// remember the number of merged operations while still in synchronized block
final int numberOfMergedOpsForPeer = numberOfMergedOperations.get(peer);
scheduleTask(new Runnable() {
public void run() {
/*
* Thoughts on locking: The peer runs the following apply(...) call synchronized.
* Therefore, it can't take other apply calls (local or from other peers) during
* that time. It may, though, have pending tasks in its updatePeers that can
* continue to run. This may include updates for this peer which would be received
* by this peer's apply(...) method.
*/
peer.apply(PeerImpl.this, operation, numberOfMergedOpsForPeer);
// TODO what to do if apply throws a RuntimeException?
}
});
}
}
}
public String toString() {
if (name != null) {
return name;
} else {
return (role == Role.SERVER?"server":"client")+" "+super.toString();
}
}
@Override
public void confirm(Peer<O, S> source, int numberOfMergedOperations) {
unmergedOperationsForPeer.get(source).confirm(numberOfMergedOperations);
}
/**
* Records that a new task is scheduled by incrementing {@link #scheduledOrRunning} by one, then
* schedules the <tt>runnable</tt> with the {@link #merger} executor service. The
* <tt>runnable</tt> is wrapped such that when its execution has finished,
* {@link #taskFinished()} will be called such that {@link #scheduledOrRunning} is decremented
* properly again and waiting clients in {@link #waitForNotRunning()} can be unblocked.
*/
private void scheduleTask(final Runnable runnable) {
taskScheduled();
merger.execute(new Runnable() {
public void run() {
try {
runnable.run();
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception during executing task in peer "+PeerImpl.this, e);
throw e;
} finally {
taskFinished();
}
}
});
}
private void taskScheduled() {
logger.fine(""+this+" taskScheduled incrementing scheduleOrRunning from "+scheduledOrRunning+" to "+(scheduledOrRunning+1));
scheduledOrRunning++;
}
/**
* Must be called by all tasks submitted to the {@link #merger} when finished.
* Unblocks {@link #waitForNotRunning()} if the merger's queue got empty by this.
*/
private synchronized void taskFinished() {
logger.fine(""+this+" taskFinished decrementing scheduleOrRunning from "+scheduledOrRunning+" to "+(scheduledOrRunning-1));
scheduledOrRunning--;
if (scheduledOrRunning == 0) {
logger.fine(""+this+" taskFinished notifying waiters");
notifyAll();
}
}
@Override
public synchronized void waitForNotRunning() {
while (scheduledOrRunning > 0) {
try {
wait();
logger.fine(""+this+" waitForNotRunning scheduleOrRunning: "+scheduledOrRunning);
} catch (InterruptedException e) {
// ignore interruption, try again
}
}
}
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/Transformer.java
+19 -19
View File
@@ -1,19 +1,19 @@
package com.sap.sse.operationaltransformation;
public interface Transformer<S, O extends Operation<S>> {
/**
* Takes an operation from the client and an operation from the server, both based on the same state. The
* complementary pair of operations is computed such that if the client applies the server operation returned, and
* the server applies the client operation returned, both end up in equal states again.
* <p>
*
* If a resulting operation is supposed to not have any effects, return <code>null</code> as the operations in the
* resulting pair.
*
* @param clientOp
* operation that was executed on the CLIENT; never called with <code>null</code>
* @param serverOp
* operation that was executed on the SERVER; never called with <code>null</code>
*/
ClientServerOperationPair<S, O> transform(O clientOp, O serverOp);
}
package com.sap.sse.operationaltransformation;
public interface Transformer<S, O extends Operation<S>> {
/**
* Takes an operation from the client and an operation from the server, both based on the same state. The
* complementary pair of operations is computed such that if the client applies the server operation returned, and
* the server applies the client operation returned, both end up in equal states again.
* <p>
*
* If a resulting operation is supposed to not have any effects, return <code>null</code> as the operations in the
* resulting pair.
*
* @param clientOp
* operation that was executed on the CLIENT; never called with <code>null</code>
* @param serverOp
* operation that was executed on the SERVER; never called with <code>null</code>
*/
ClientServerOperationPair<S, O> transform(O clientOp, O serverOp);
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/UnmergedOperationsQueue.java
+92 -92
View File
@@ -1,92 +1,92 @@
package com.sap.sse.operationaltransformation;
import java.util.ArrayList;
import java.util.List;
public class UnmergedOperationsQueue<O extends Operation<S>, S> {
/**
* As the {@link #unmergedOperationsForPeer} queues are cleaned up whenever merge
* confirmations are received from peers, we need to remember which local operation
* number the first element in the queue resulted from.
*/
private int numberOfFirstInList;
private List<O> unmergedOperations;
public UnmergedOperationsQueue() {
numberOfFirstInList = 0;
unmergedOperations = new ArrayList<O>(5);
}
/**
* After an operation has been locally applied and needs to be propagated
* to the peer for which this is the queue of unmerged operations, the operation
* must be appended to this queue.
*/
public synchronized void sentOutOperation(O operationSentOut) {
unmergedOperations.add(operationSentOut);
}
/**
* Returns a copied list of unmerged and stepwise transformed local
* operations starting at the requested local operation number.
*
* @param numberOfConfirmedLocalOperations
* A value of 0 means that the remote peer has not confirmed the
* merge of any local operation yet.
*/
public synchronized List<O> getUnmergedOperations(int numberOfConfirmedLocalOperations) {
return new ArrayList<O>(unmergedOperations.subList(
numberOfConfirmedLocalOperations-numberOfFirstInList,
unmergedOperations.size()));
}
/**
* Replaces the local operation with number <tt>localOperationNumber</tt> in the queue by
* <tt>transformedOperation</tt>.
*
* @param localOperationNumber starts with 0
*/
public synchronized void updateWithTransformed(int localOperationNumber, O operation) {
unmergedOperations.set(localOperationNumber-numberOfFirstInList, operation);
}
/**
* Cleans up the queue so that only unconfirmed / unmerged operations
* remain. Afterwards, {@link #getUnmergedOperations(int) asking for an
* unmerged operation} with an argument for
* <tt>numberOfConfirmedLocalOperations</tt> smaller than
* <tt>numberOfMergedOperations</tt> passed to this call will result in an
* exception.
*
* @param numberOfMergedOperations
* tells how many operations that emerged from the local peer the
* remote peer has confirmed to have merged.
*/
public synchronized void confirm(int numberOfMergedOperations) {
for (int i=numberOfFirstInList; i<numberOfMergedOperations; i++) {
unmergedOperations.remove(0);
}
numberOfFirstInList = numberOfMergedOperations;
}
public synchronized String toString() {
int i=numberOfFirstInList;
StringBuilder result = new StringBuilder();
result.append('[');
boolean first = true;
for (Operation<S> op : unmergedOperations) {
if (!first) {
result.append(", ");
} else {
first = false;
}
result.append(i);
result.append(": ");
result.append(op);
}
result.append(']');
return result.toString();
}
}
package com.sap.sse.operationaltransformation;
import java.util.ArrayList;
import java.util.List;
public class UnmergedOperationsQueue<O extends Operation<S>, S> {
/**
* As the {@link #unmergedOperationsForPeer} queues are cleaned up whenever merge
* confirmations are received from peers, we need to remember which local operation
* number the first element in the queue resulted from.
*/
private int numberOfFirstInList;
private List<O> unmergedOperations;
public UnmergedOperationsQueue() {
numberOfFirstInList = 0;
unmergedOperations = new ArrayList<O>(5);
}
/**
* After an operation has been locally applied and needs to be propagated
* to the peer for which this is the queue of unmerged operations, the operation
* must be appended to this queue.
*/
public synchronized void sentOutOperation(O operationSentOut) {
unmergedOperations.add(operationSentOut);
}
/**
* Returns a copied list of unmerged and stepwise transformed local
* operations starting at the requested local operation number.
*
* @param numberOfConfirmedLocalOperations
* A value of 0 means that the remote peer has not confirmed the
* merge of any local operation yet.
*/
public synchronized List<O> getUnmergedOperations(int numberOfConfirmedLocalOperations) {
return new ArrayList<O>(unmergedOperations.subList(
numberOfConfirmedLocalOperations-numberOfFirstInList,
unmergedOperations.size()));
}
/**
* Replaces the local operation with number <tt>localOperationNumber</tt> in the queue by
* <tt>transformedOperation</tt>.
*
* @param localOperationNumber starts with 0
*/
public synchronized void updateWithTransformed(int localOperationNumber, O operation) {
unmergedOperations.set(localOperationNumber-numberOfFirstInList, operation);
}
/**
* Cleans up the queue so that only unconfirmed / unmerged operations
* remain. Afterwards, {@link #getUnmergedOperations(int) asking for an
* unmerged operation} with an argument for
* <tt>numberOfConfirmedLocalOperations</tt> smaller than
* <tt>numberOfMergedOperations</tt> passed to this call will result in an
* exception.
*
* @param numberOfMergedOperations
* tells how many operations that emerged from the local peer the
* remote peer has confirmed to have merged.
*/
public synchronized void confirm(int numberOfMergedOperations) {
for (int i=numberOfFirstInList; i<numberOfMergedOperations; i++) {
unmergedOperations.remove(0);
}
numberOfFirstInList = numberOfMergedOperations;
}
public synchronized String toString() {
int i=numberOfFirstInList;
StringBuilder result = new StringBuilder();
result.append('[');
boolean first = true;
for (Operation<S> op : unmergedOperations) {
if (!first) {
result.append(", ");
} else {
first = false;
}
result.append(i);
result.append(": ");
result.append(op);
}
result.append(']');
return result.toString();
}
}
java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/package.html
+62 -62
View File
@@ -1,63 +1,63 @@
<body>
<p>
Use the contents of this package for so-called <em>operational transformations</em>.
This can be used to synchronize the states of a number of clients with a single server.
The synchronization is performed asynchronously. The construction of the transformation
function guarantees to lead to eventual consistency across clients and server.
Read more about the general background of the algorithm on which this is based
<a href="../docs/JupiterCollaborationSystem_Nichols.pdf">here</a>.
</p>
<p>
To establish a server to which clients can connect, first decide how your <em>state</em>
is encoded on which the <em>operations</em> will perform changes to transition from one
state to the next. Then, design your operations class. The trickiest part comes next. You
have to define a transformer that maps a pair of your operations to another pair of
operations of the same type. The transformer has to implement the
{@link com.sap.sailing.operationaltransformation.Transformer} interface, in particular
the {@link com.sap.sailing.operationaltransformation.Transformer#transform} method. The
transformation function assumes that both input operations have been applied to the same
state, one on a client, the other one on the server. The resulting operation for the client
applied on the client must lead to a state equal to the state reached on the server by
applying the resulting operation for the server to the server's state.
</p>
<p>
Put formally, if <em>t</em> is a transformer, and
if we have <em>s_c</em> as the state on the client and <em>s_s</em> as the state
on the server, and <em>c</em> as the operation that was executed in state <em>s_c</em> on the
client, and <em>s</em> as the operation executed on the server in state <em>s_s</em>, then the
following condition must hold:<br>
<pre>
t.transform(c, s).getClientOp().applyTo(c.applyTo(s_c)).equals(
t.transform(c, s).getServerOp().applyTo(s.applyTo(s_s)))
</pre>
</p>
<p>
You can create servers and clients, e.g., as follows:
<pre>
server = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>(
"Server", new MyTransformer(), /* initial state */ new MyStateType(...), /* server */ true);
client1 = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>("Client1", new MyTransformer(), server);
client2 = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>("Client2", new MyTransformer(), server);
</pre>
After that, you can start sending operations to the clients using the
{@link com.sap.runlet.operationaltransformation.Peer#apply} method as follows:
<pre>
client1.{@link com.sap.runlet.operationaltransformation.Peer#apply apply}(new MyOperationType(...));
client2.{@link com.sap.runlet.operationaltransformation.Peer#apply apply}(new MyOperationType(...));
...
</pre>
and those changes will eventually get synchronized across all clients and the server. You can
obtain the current state of a peer using the {@link com.sap.runlet.operationaltransformation.Peer#getCurrentState}
operation.
</p>
Consider using {@link com.sap.sailing.operationaltransformation.OperationalTransformer} as the basis for your transformation
logic. It generically works with {@link com.sap.sailing.operationaltransformation.OperationWithTransformationSupport} if you
can afford to have your operation classes implement the transformation logic. The benefit of doing so is that adding more
operations requires you to consider the transformation logic.
<body>
<p>
Use the contents of this package for so-called <em>operational transformations</em>.
This can be used to synchronize the states of a number of clients with a single server.
The synchronization is performed asynchronously. The construction of the transformation
function guarantees to lead to eventual consistency across clients and server.
Read more about the general background of the algorithm on which this is based
<a href="../docs/JupiterCollaborationSystem_Nichols.pdf">here</a>.
</p>
<p>
To establish a server to which clients can connect, first decide how your <em>state</em>
is encoded on which the <em>operations</em> will perform changes to transition from one
state to the next. Then, design your operations class. The trickiest part comes next. You
have to define a transformer that maps a pair of your operations to another pair of
operations of the same type. The transformer has to implement the
{@link com.sap.sailing.operationaltransformation.Transformer} interface, in particular
the {@link com.sap.sailing.operationaltransformation.Transformer#transform} method. The
transformation function assumes that both input operations have been applied to the same
state, one on a client, the other one on the server. The resulting operation for the client
applied on the client must lead to a state equal to the state reached on the server by
applying the resulting operation for the server to the server's state.
</p>
<p>
Put formally, if <em>t</em> is a transformer, and
if we have <em>s_c</em> as the state on the client and <em>s_s</em> as the state
on the server, and <em>c</em> as the operation that was executed in state <em>s_c</em> on the
client, and <em>s</em> as the operation executed on the server in state <em>s_s</em>, then the
following condition must hold:<br>
<pre>
t.transform(c, s).getClientOp().applyTo(c.applyTo(s_c)).equals(
t.transform(c, s).getServerOp().applyTo(s.applyTo(s_s)))
</pre>
</p>
<p>
You can create servers and clients, e.g., as follows:
<pre>
server = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>(
"Server", new MyTransformer(), /* initial state */ new MyStateType(...), /* server */ true);
client1 = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>("Client1", new MyTransformer(), server);
client2 = new {@link com.sap.runlet.operationaltransformation.PeerImpl PeerImpl}<MyOperationType, MyStateType>("Client2", new MyTransformer(), server);
</pre>
After that, you can start sending operations to the clients using the
{@link com.sap.runlet.operationaltransformation.Peer#apply} method as follows:
<pre>
client1.{@link com.sap.runlet.operationaltransformation.Peer#apply apply}(new MyOperationType(...));
client2.{@link com.sap.runlet.operationaltransformation.Peer#apply apply}(new MyOperationType(...));
...
</pre>
and those changes will eventually get synchronized across all clients and the server. You can
obtain the current state of a peer using the {@link com.sap.runlet.operationaltransformation.Peer#getCurrentState}
operation.
</p>
Consider using {@link com.sap.sailing.operationaltransformation.OperationalTransformer} as the basis for your transformation
logic. It generically works with {@link com.sap.sailing.operationaltransformation.OperationWithTransformationSupport} if you
can afford to have your operation classes implement the transformation logic. The benefit of doing so is that adding more
operations requires you to consider the transformation logic.
</body>