I am currently attempting to create a message-passing library, and one of the tenets of message-passing is that mutable state is only modified through messages. The 'messages' that will be passed around are function objects that take a 'sender' (that created the message) and a 'receiver' (the worker/actor/what have you processing the message from it's queue.)
Workers are defined as follows, and the self-referential nature of the interface is used because a worker may have state that it wants to expose to message senders, and this requires a sender to be aware of it's unique type.
public interface Worker<T extends Worker<T>> {
T getWorker(); //convert a Worker<T> to it's <T> since T extends Worker<T>
<S extends Worker<S>> void message(Message<S,T> msg);
}
where <S> is the type of the worker that sent the message.
Messages are defined as follows:
public interface Message<S extends Worker<S>,R extends Worker<R>> {
void process(S sender, R thisWorker);
}
where <S> is the class type of the sender and <R> is the class type of the processing worker. As expected by the interface and the mechanics of message-passing, the function will be able to modify the state of thisWorker. However, if the message were to directly mutate sender, a race condition would result because there's no synchronization across the workers/threads (which is the whole point of using messages to begin with!)
While I could declare <S> to be a generic Worker<?>, this would break the ability for a message to 'reply' to it's sender in a meaningful way, since it can no longer reference it's specific fields.
Therefore, how can I ensure that sender won't be modified except in the context of a message addressed specifically to it?
T extends Worker<T> is indeed used to prevent a class from returning a non-Worker type. I could probably get away with just Worker<T> and trust the user.
What I am specifically trying to achieve is a message-passing system where the messages are code, which would avoid the need for any special handling of different message types in each worker. However, I'm probably realizing there's no elegant way to do this without adhering to some message protocol if I wanted to enforce it, like with execution on the Swing EDT.
I don't quite understand your design. What exactly is the purpose of the recursive generics in the interface definitions such as interface Worker<T extends Worker<T>>? Wouldn't interface Worker<T> suffice, still allowing for the implementation of T getWorker() to get the concrete type of the class that implements the Worker interface? Is it an attempt to restrict the interface implementation from returning something that is not a Worker from getWorker()? In that case, I think interface Worker<T extends Worker<?>> would create less confusion.
Therefore, how can I ensure that
senderwon't be modified except in the context of a message addressed specifically to it?
To the best of my knowledge, the only way to prevent invocation of an object's methods outside a desired context is to confine the execution of the object's methods to a single thread that is inaccessible to client code. That would require the object to check the current thread as the very first thing in every single method. Below is a simple example. It is essentially a standard producer/consumer implementation in which the object that is being modified (not the Mailbox/the consumer implementation!) prevents modifications of itself on threads other than a designated thread (the one that consumes the incoming messages).
class Mailbox {
private final BlockingQueue<Runnable> mQueue = new LinkedBlockingQueue<>();
private final Thread mReceiverThread = new Thread(() -> {
while (true) {
try {
Runnable job = mQueue.take();
job.run();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
public Mailbox() {
mReceiverThread.start();
}
public void submitMsg(Runnable msg) {
try {
mQueue.put(msg);
} catch (InterruptedException e) {
// TODO exception handling; rethrow wrapped in unchecked exception here in order to allow for use of lambdas.
throw new RuntimeException(e);
}
}
public void ensureMailboxThread(Object target) {
if (Thread.currentThread() != mReceiverThread) {
throw new RuntimeException("operations on " + target + " are confined to thread " + mReceiverThread);
}
}
}
class A {
// Example use
public static void main(String[] args) {
Mailbox a1Mailbox = new Mailbox();
Mailbox a2Mailbox = new Mailbox();
A a1 = new A(a1Mailbox);
A a2 = new A(a2Mailbox);
// Let's send something to a1 and have it send something to a2 if a certain condition is met.
// Notice that there is no explicit sender:
// If you wish to reply, you "hardcode" the reply to what you consider the sender in the Runnable's run().
a1Mailbox.submitMsg(() -> {
if (a1.calculateSomething() > 3.0) {
a2Mailbox.submitMsg(() -> a2.doSomething());
} else {
a1.doSomething();
}
});
}
private final Mailbox mAssociatedMailbox;
public A(Mailbox mailbox) {
mAssociatedMailbox = mailbox;
}
public double calculateSomething() {
mAssociatedMailbox.ensureMailboxThread(this);
return 3 + .14;
}
public void doSomething() {
mAssociatedMailbox.ensureMailboxThread(this);
System.out.println("hello");
}
}
Let me reiterate the emphasized part: it is up to each individual class participating in the message-passing to verify the current thread at the start of every single method. There is no way to extract this verification to a general-purpose class/interface as an object's method may be invoked from any thread. For example, the Runnable submitted using submitMsg in the example above could choose to spawn a new thread and attempt to modify the object on that thread:
a1Mailbox.submitMsg(() -> {
Thread t = new Thread(() -> a1.doSomething());
t.start();
});
However, this would be prevented, but only due to the check being part of A.doSomething() itself.